API · Async

`asyncReduce()`

Combine all successful values into one accumulator while awaiting each reduction step.

Signature

asyncReduce<A>(
  fn: (accumulator: A, value: T, context: C) => A | PromiseLike<A>,
  initialValue: A,
): Exstream<A, AggregateOutputContext<C, A>>

Example

const total = await exstream(ids)
  .asyncReduce(async (sum, id) => sum + (await fetchPrice(id)), 0)
  .value()

Parameters

fn

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

Invoked once per successful input. Each result is awaited before requesting the next value.

initialValue

Type ARequired

Initial accumulator and empty-source result.

Execution

Concurrency is always 1 and output order follows input order. The operator itself retains only the accumulator, though the aggregate context can retain input contexts. One result is emitted after upstream ends; infinite input never finishes. Backpressure is inherent because next() is called only after the awaited reducer settles.

There is no built-in timeout or retry policy. Use an abort-aware callback and context.signal for cancellation-sensitive work, or compose mapAsync() before a synchronous reduction when concurrency and retry controls are needed.

Errors

A thrown or rejected reducer failure becomes a record error for the current input and terminates the reduction branch without an accumulator result. Existing record errors pass through and are excluded.

Forms

stream.asyncReduce(reducer, initialValue)
exstream.pipeline().asyncReduce(reducer, initialValue)
exstream.asyncReduce(reducer, initialValue, stream)
stream.through(exstream.asyncReduce(reducer, initialValue))

reduce(), mapAsync(), collect()

asyncReduce()