Terminal consumers

A pipeline built with .pipe() starts executing when a consumer begins reading it. Choose a consumer to collect output, compute one result, or process items individually.

Choose a consumer

Goal	API	Result
Collect every item into an array	toArray	An array of all items
Compute a sum or another aggregate	reduce	The final accumulator
Process each item without collecting output	forEach	Promise<void>
Find the first matching value	find	The first match, or undefined
Check whether any item matches	some	true if any item matches
Check whether every item matches	every	true if all items match

These consumers return promises. Each is available as a method, such as stream.toArray({ signal }), and as a standalone function, such as toArray(source, { signal }). Standalone functions also accept ordinary synchronous and asynchronous iterables.

Collect or aggregate

import { from } from "mogami-stream";

export async function collectValues() {
  const numbers = from([1, 2, 3]);
  const values = await numbers.toArray(); // [1, 2, 3]
  const total = await numbers.reduce((sum, value) => sum + value, 0); // 6
  return { values, total };
}

toArray retains every output in memory, so use it for finite inputs. reduce folds items sequentially from an initial value and returns that value for an empty input. The example uses an array as its source, so the same stream can be read again.

Process items individually

export async function processValues<T>(
  input: Iterable<T> | AsyncIterable<T>,
  save: (value: T, signal: AbortSignal) => Promise<void>,
  signal: AbortSignal,
) {
  await from(input).forEach((value, { signal }) => save(value, signal), {
    signal,
  });
}

forEach waits for each callback before reading the next value and does not build an output array. Both forEach and reduce run callbacks sequentially. To process items concurrently, use an upstream map(..., { concurrency }) stage.

Stop when the answer is known

export async function checkValues() {
  const numbers = from([1, 2, 3]);
  const first = await numbers.find((value) => value > 1); // 2
  const any = await numbers.some((value) => value > 2); // true
  const all = await numbers.every((value) => value > 0); // true
  return { first, any, all };
}

find and some close upstream on the first match; every closes it on the first non-match. For empty inputs, find returns undefined, some returns false, and every returns true.

Cancellation and cleanup

Pass { signal } to cancel the execution. Forward the callback's signal into I/O to propagate the stop request to in-flight work.

Consumers settle after cleanup finishes. If processing and cleanup both fail, the primary processing error takes precedence. I/O that does not support cancellation must still finish before the consumer settles.

For manual consumption, use for await...of or iterate. Acquiring an iterator does not start execution; its first next() does. See Cancellation and cleanup for cancel(), return(), and closed.

For secondary errors, use the collection report recipe. It also explains how to observe a timeout before cleanup completes.