API · Aggregate

`reduce()`

Combine all successful values into one accumulator, starting from an explicit initial value.

Signature

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

Example

const total = await exstream(orders)
  .reduce((sum, order) => sum + order.total, 0)
  .value()

Parameters

fn

Type (accumulator, value, context) => ARequired

Synchronous reducer called in input order for every successful value. Its return value becomes the next accumulator.

initialValue

Type ARequired

The first accumulator and the result emitted for an empty source.

Behavior

One value is emitted only after upstream ends. The accumulator is retained in memory; reduce() does not retain every input unless your accumulator does. It consumes sequentially, preserves input order, and propagates pressure. Infinite input does not finish.

The result receives an aggregate context containing materialized input contexts in order when they exist. Existing record errors pass through and are excluded from the reduction; handling them allows reduction to continue.

Errors

If fn throws, Exstream emits a record error associated with the current input and destroys this reduction branch; no accumulator result follows. Promise results are not awaited. Use asyncReduce() for asynchronous accumulation.

Forms

The method order is fn, initialValue, matching the standalone operator:

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

reduce1(), asyncReduce(), collect()

reduce()