API · Transform

`map()`

Run a synchronous callback for every successful input and emit exactly one result.

Example

const totals = exstream(rows).map((row) => ({
  ...row,
  total: Number(row.total),
}))

Parameters

fn

Type (value, context) => URequired

Called once for every successful value. The callback is synchronous. Declare the second parameter only when record metadata or its cancellation signal is needed; Exstream materializes a context lazily.

Behavior

map() preserves input order and adds no independent queue. It asks upstream for work only while downstream can accept it. Existing record errors pass through without calling fn.

When requested, context.input is the value that created the current context, context.signal aborts when this branch should stop, and any custom fields added upstream remain available. The same materialized context continues with the mapped output unless a later branch boundary copies it.

The callback result is emitted as-is, including undefined, arrays, streams, and asynchronous values. A native promise, or an object exposing both callable then and catch properties, is recognized so rejection retains the original input. A minimal thenable with only then is emitted as an ordinary value. Returning any promise does not make map() await it or limit concurrent work:

const pending = exstream(ids).map((id) => fetch(`/items/${id}`))
// Exstream<Promise<Response>, C>

Use mapAsync() for awaited output, bounded concurrency, ordering controls, retries, or timeouts.

To retain both input and output, return that shape explicitly:

const compared = exstream(records).map((input) => ({
  input,
  output: calculateScore(input),
}))

Errors

If fn throws, Exstream emits a contextual record error for that input and continues when an error policy handles it. Use mapAsync() when fn returns a promise so Exstream can await it and apply the configured concurrency. Fatal graph failures bypass map() and abort the branch.

Forms

map() is available on streams and reusable pipelines:

stream.map(fn)
exstream.pipeline().map(fn)

Signature

map<U>(
  fn: (value: T, context: CallbackContext<T, C>) => U,
): Exstream<U, NextContext<C, U>>

filter(), flatMap(), tap(), mapAsync()

map()