API · Async

`mapAsync()`

Run promise-returning work with bounded concurrency, optional completion-order output, retries, timeouts, and cancellation.

Signature

mapAsync<U>(
  fn: (value: T, context: C) => U | PromiseLike<U>,
  options?: MapAsyncOptions<T, C> | null,
): Exstream<Awaited<U>, C>

interface MapAsyncOptions<T, C extends object> {
  concurrency?: number
  ordered?: boolean
  retry?: number | MapAsyncRetry<T, C> | null
  timeout?: number | null
  signal?: AbortSignal
}

interface MapAsyncRetry<T, C extends object> {
  retries?: number
  delay?: number | ((attempt: number, error: ExstreamError<T>, value: T, context: C) => number | PromiseLike<number>)
  when?: (error: ExstreamError<T>, value: T, context: C, attempt: number) => boolean | PromiseLike<boolean>
}

Example

const profiles = exstream(userIds).mapAsync(fetchProfile, {
  concurrency: 8,
  ordered: false,
  retry: {
    retries: 3,
    delay: (attempt) => 250 * 2 ** (attempt - 1),
    when: (error) => error.status === 429,
  },
  timeout: 10_000,
})

Parameters

fn

Type (value, context) => U | PromiseLike<U>Required

Called for each successful input when a concurrency slot is available. It may return immediately or return a promise-like value. The resolved value is emitted. The same input and context are reused across retry attempts.

concurrency

Type positive integer | InfinityDefault 1

Maximum active tasks. Positive integers and Infinity are accepted; zero, negative values, fractions, and non-numeric values are rejected. A retry and its delay retain the task's slot. Use a finite value at external I/O boundaries.

ordered

Type booleanDefault true

With true, results are emitted in input order even when later tasks finish first. With false, each result is emitted as soon as its task completes. Only literal booleans are accepted.

retry

Type non-negative integer | MapAsyncRetry | nullDefault null

A number is the count of additional attempts after the first failure. An object configures retries, delay, and when. Zero and null disable retries. Fatal failures are never retried.

retry.retries

Type non-negative integerDefault 0

Number of additional attempts. For example, retries: 2 allows at most three callback calls for one input.

retry.delay

Type non-negative milliseconds | functionDefault 0

A fixed finite delay or a function returning one, synchronously or asynchronously. The function receives the one-based failed attempt, contextual error, input, and context. Invalid returned delays become record failures.

retry.when

Type (error, value, context, attempt) => boolean | PromiseLike<boolean>Default retry every record failure

Return or resolve to false to stop retrying that failure. The attempt number is one-based. A thrown or rejected policy function becomes the final record failure.

timeout

Type non-negative finite number | nullDefault null

Maximum milliseconds for each individual attempt, including 0. A timeout is eligible for retry and produces MapAsyncTimeoutError with code EXSTREAM_MAP_ASYNC_TIMEOUT, plus timeout and attempt fields.

signal

Type AbortSignalDefault undefined

Aborts this operator. A pre-aborted signal prevents work from starting. Pass context.signal into cancellable I/O so cancellation and per-attempt timeouts can stop the underlying operation rather than only ignore its late result.

Passing null or undefined for options applies every default. Other non-object values and arrays are rejected when the operator is created.

The JavaScript runtime normalizes numeric policy fields with Number(), so any coercible value is accepted when it produces the required integer or finite value. The TypeScript API deliberately exposes these fields as numbers; use actual numbers for portable, explicit configuration. ordered is strictly boolean.

Order and pressure

At most concurrency callbacks are active. Upstream demand pauses when all slots are occupied. In ordered mode, completed results may wait in memory behind a slower earlier input; unordered mode avoids that head-of-line delay and emits completion order.

The callback context exposes context.input, context.signal, and custom upstream fields. It is created lazily when fn declares its second parameter, retry.when declares its third, or a dynamic retry.delay declares its fourth. Declare those positional parameters rather than retrieving them through rest arguments when a materialized context is required. The context remains the same object across attempts and continues with the emitted result. During a timed attempt, only its signal is temporarily replaced with an attempt-specific signal and restored afterward.

Retries

Retry policy applies per input and per failed attempt. The callback receives the same value and record context each time. Delay and policy evaluation happen inside the same concurrency slot, so a large retry delay reduces available throughput.

Timeouts are also per attempt. Exstream aborts the attempt context signal when the deadline expires. JavaScript promises themselves are not cancellable, so the callback must forward that signal to fetch, database clients, or other cancellable APIs.

Errors

A thrown callback error, rejected promise, exhausted retry policy, invalid dynamic delay, or timeout becomes a contextual record error. If handled downstream, later tasks continue. Fatal graph failures and external cancellation abort the operator immediately and bypass retry policy.

Cancelling the branch stops new scheduling and ignores late completions from work that could not be cancelled.

Forms

mapAsync() is available on streams and reusable pipelines. The direct standalone form requires an explicit options argument; pass null for defaults:

stream.mapAsync(fn, options)
exstream.pipeline().mapAsync(fn, options)
exstream.mapAsync(fn, null, stream)
stream.through(exstream.mapAsync(fn, options))

map(), resolve(), errors(), drain()

mapAsync()