Choosing operators

Use the catalog for individual examples and API links. Choose how values should flow before setting concurrency or timing limits.

Goal	Operators	Behavior
Transform or select items	map, filter, filterMap	map supports concurrency; the others run sequentially.
Limit output or group items	take, chunks, bufferTime	Count at the stage's position; see processing order.
Join known sources in sequence	chain(a, b)	Closes each source before opening the next.
Interleave known sources	merge(a, b)	Reads sources concurrently.
Flatten a stream of sources	concat, flatten	concat preserves child order; flatten interleaves ready children.
Create a source per item	concatMap, flatMap	Ordered or interleaved flattening after a synchronous mapper. Use defer for async setup.
Pair corresponding values	zip(a, b)	Stops when the shortest source ends.
Combine current values	zipLatest([a, b])	Waits for every source's first value; partial: true fills missing values with undefined.

concat, flatten, concatMap, and flatMap default to concurrency: 1. Increase it to overlap children; ordered variants still deliver each child's output in sequence. See concurrency bounds.

For pacing and deadlines, use the timing guide. For retries and fallback sources, use error handling. Choose a terminal consumer to start reading.

Accumulated state

scan emits each updated accumulator; reduce returns only the final one. Both await the initial value and reducer results. An empty input emits nothing from scan, while reduce returns the initial value.

Objects are passed by reference, not copied. Use immutable updates for snapshots. If you mutate state, create it inside defer for each execution and capture each result before the next mutation:

import { defer, from } from "mogami-stream";

export const totals = defer(() =>
  from([1, 2])
    .scan(
      (state, value) => {
        state.total += value;
        return state;
      },
      { total: 0 },
    )
    .map((state) => state.total),
);
console.log(await totals.toArray()); // [1, 3]
console.log(await totals.toArray()); // [1, 3]

Promise-valued items

from and selection/combination operators preserve Promise objects as items. Unlike native for await over a synchronous iterable, they do not unwrap those items. Use map to await them:

export const pending = Promise.resolve(2);
export const preserved = await from([pending]).toArray(); // [Promise<2>]
export const resolved = await from([pending])
  .map((value) => value)
  .toArray(); // [2]

find receives the original item in its predicate, but awaits a matched Promise when returning it.