API · Async

mapAsync()

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

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 inputs owned by the operator: callbacks still running plus completed results waiting for downstream demand. Positive integers and Infinity are accepted; zero, negative values, fractions, and non-numeric values are rejected. A retry, its delay, and an ordered result waiting behind an earlier input retain their slots. 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.

onFail

Type (error, input, push, attempt, retry, context) => void | PromiseLike<void>Default null

Handles one failed attempt inside the current record's concurrency slot. Call retry() to run fn again with the current input, retry(nextInput) to run it with a replacement input, push(null, output) to emit a successful fallback, or push(error, input) to propagate a record error. The failed attempt number is one-based. The handler may be asynchronous and cannot be combined with retry.

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

concurrency bounds the complete operator window, not only unresolved promises. The window contains callbacks still running and completed results waiting to be accepted downstream. As soon as downstream accepts one result, that slot is released and exactly one new input may start. With a slow writer and fast callbacks, active work can therefore fall below concurrency while ready results occupy the rest of the window; the operator does not drain the whole window before refilling it.

Upstream demand pauses while the window is full. 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. Both modes use the same sliding-window refill rule.

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, a dynamic retry.delay declares its fourth, or onFail declares its sixth. 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 and local recovery

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.

onFail is the programmable alternative to automatic retry. It is useful when recovery must await more I/O, when the next attempt needs enriched input, or when a failure should produce a fallback output:

const processed = exstream(orders).mapAsync(processOrder, {
  concurrency: 8,
  onFail: async (error, input, push, attempt, retry, context) => {
    if (error.code === 'MISSING_CUSTOMER' && attempt < 3) {
      const customer = await recoverCustomer(input.customerId, context.signal)
      retry({ ...input, customer })
      return
    }

    push(error, input)
  },
})

The handler must settle at most once. Returning without calling retry or push propagates the original failure. Throwing or rejecting replaces it with the handler failure. retry repeats the complete fn callback, so any side effects inside that callback must be safe to repeat.

Errors

A thrown callback error, rejected promise, exhausted retry policy, propagated onFail decision, invalid dynamic delay, or timeout becomes a contextual record error. If handled downstream, later tasks continue. If it reaches a terminal, that terminal rejects. Fatal graph failures and external cancellation abort the operator immediately and bypass retry and onFail 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:

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

Signature

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

interface MapAsyncOptions<T, C extends object, Output = unknown> {
  concurrency?: number
  ordered?: boolean
  retry?: number | MapAsyncRetry<T, C> | null
  onFail?: (
    error: ExstreamError<T>,
    input: T,
    push: MapAsyncFailurePush<T, Output>,
    attempt: number,
    retry: MapAsyncRetryAttempt<T>,
    context: C,
  ) => void | PromiseLike<void>
  timeout?: number | null
  signal?: AbortSignal
}

interface MapAsyncFailurePush<Input, Output> {
  (error: null | undefined, value: Output): void
  (error: unknown, input?: Input): void
}

interface MapAsyncRetryAttempt<Input> {
  (): void
  (input: Input): void
}

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>
}

Related

map(), errors(), drain()