Operators

Browse names and descriptions by category. Expand a row for a copyable example and its output. Follow the name to the API reference for exact types and detailed behavior.

See the Timing guide for help choosing and combining timing operators.

Sources

fromWrap a synchronous or asynchronous iterable into a Stream, preserving its replayability and Promise values.

import { from } from "mogami-stream";
const result = await from([1, 2, 3]).toArray();

console.log(result);

Output

[1, 2, 3]

deferCreate a fresh source on the first read of each execution; supports async factory setup and AbortSignal.

import { defer } from "mogami-stream";
let run = 0;
const source = defer(() => [++run]);
const result = [await source.toArray(), await source.toArray()];

console.log(result);

Output

[[1], [2]]

ofCreate a replayable stream that emits the provided values in order.

import { of } from "mogami-stream";
const result = await of("A", "B").toArray();

console.log(result);

Output

["A", "B"]

emptyCreate a stream that completes immediately on the first read without emitting values.

import { empty } from "mogami-stream";
const result = await empty<number>().toArray();

console.log(result);

Output

[]

rangeEmit safe integers from start (inclusive) to end (exclusive) with a configurable step.

import { range } from "mogami-stream";
const result = await range(1, 6, 2).toArray();

console.log(result);

Output

[1, 3, 5]

fromCallbackBridge push-based callbacks into a pull stream using a bounded queue and configurable overflow policy.

import { fromCallback } from "mogami-stream";
const result = await fromCallback<string>(
  (observer) => {
    observer.next("A");
    observer.next("B");
    observer.complete();
  },
  { capacity: 2 },
).toArray();

console.log(result);

Output

["A", "B"]

fromEventListen to EventTarget events through a bounded queue with explicit overflow handling.

import { fromEvent, map, take } from "mogami-stream";
const target = new EventTarget();
const pending = fromEvent(target, "ready", { capacity: 1 })
  .pipe(
    map((event) => event.type),
    take(1),
  )
  .toArray();
target.dispatchEvent(new Event("ready"));
const result = await pending;

console.log(result);

Output

["ready"]

fromTimelineReplay a scripted schedule of values, completion, or errors at millisecond offsets for testing and mocking.

import { fromTimeline } from "mogami-stream";
const result = await fromTimeline([
  { at: 0, kind: "value", value: "A" },
  { at: 10, kind: "value", value: "B" },
  { at: 15, kind: "complete" },
]).toArray();

console.log(result);

Output

["A", "B"]

Transform

mapTransform items with bounded concurrency, capacity limits, output ordering, and optional per-item retry policies.

import { from, map } from "mogami-stream";
const result = await from([1, 2, 3])
  .pipe(map(async (value) => value * 2, { concurrency: 2 }))
  .toArray();

console.log(result);

Output

[2, 4, 6]

mapSettledTransform items like map, emitting { status: 'fulfilled', value } or { status: 'rejected', reason } without halting on mapper failures.

import { from, mapSettled } from "mogami-stream";
const result = await from([1, 2, 3])
  .pipe(
    mapSettled((value) =>
      value === 2 ? Promise.reject("unavailable") : Promise.resolve(value * 10),
    ),
  )
  .toArray();

console.log(result);

Output

[
  { "status": "fulfilled", "value": 10 },
  { "status": "rejected", "reason": "unavailable" },
  { "status": "fulfilled", "value": 30 }
]

filterMapMap items sequentially and drop only null and undefined results (preserving 0, false, and empty strings).

import { filterMap, from } from "mogami-stream";
const result = await from(["12", "?", "0"])
  .pipe(
    filterMap((text) => {
      const value = Number(text);
      return Number.isNaN(value) ? undefined : value;
    }),
  )
  .toArray();

console.log(result);

Output

[12, 0]

scanFold items sequentially from an initial value and emit the updated accumulator after every step.

import { from, scan } from "mogami-stream";
const result = await from([1, 2, 3])
  .pipe(scan((sum, value) => sum + value, 0))
  .toArray();

console.log(result);

Output

[1, 3, 6]

Selection & batching

filterKeep items satisfying a sequential sync or async predicate; synchronous type guards narrow the element type.

import { filter, from } from "mogami-stream";
const result = await from([1, 2, 3, 4])
  .pipe(filter((value) => value % 2 === 0))
  .toArray();

console.log(result);

Output

[2, 4]

takeEmit at most the first n items and begin closing upstream before delivering the nth item.

import { from, take } from "mogami-stream";
const result = await from([1, 2, 3]).pipe(take(2)).toArray();

console.log(result);

Output

[1, 2]

skipDiscard the first n items and emit all remaining items.

import { from, skip } from "mogami-stream";
const result = await from([1, 2, 3]).pipe(skip(1)).toArray();

console.log(result);

Output

[2, 3]

takeWhileEmit leading items while the predicate holds, closing upstream as soon as it returns false.

import { from, takeWhile } from "mogami-stream";
const result = await from([1, 2, 3, 1])
  .pipe(takeWhile((value) => value < 3))
  .toArray();

console.log(result);

Output

[1, 2]

skipWhileDiscard leading items while the predicate holds, then emit the first non-matching item and all subsequent items.

import { from, skipWhile } from "mogami-stream";
const result = await from([1, 2, 3, 1])
  .pipe(skipWhile((value) => value < 3))
  .toArray();

console.log(result);

Output

[3, 1]

chunksGroup items into arrays of up to size elements, flushing any remaining partial batch on normal completion.

import { chunks, from } from "mogami-stream";
const result = await from([1, 2, 3, 4, 5]).pipe(chunks(2)).toArray();

console.log(result);

Output

[[1, 2], [3, 4], [5]]

dedupeDrop consecutive items whose keys match the immediately preceding emitted key via Object.is.

import { dedupe, from } from "mogami-stream";
const result = await from([1, 1, 2, 2, 1]).pipe(dedupe()).toArray();

console.log(result);

Output

[1, 2, 1]

Combination

chainConcatenate multiple sources sequentially, waiting for each source to close before opening the next.

import { chain } from "mogami-stream";
const result = await chain([1, 2], [3, 4]).toArray();

console.log(result);

Output

[1, 2, 3, 4]

mergeInterleave multiple sources concurrently with one read slot per source and round-robin selection among ready sources.

import { fromTimeline, merge, of } from "mogami-stream";
const slow = fromTimeline([{ at: 20, kind: "value", value: "slow" }]);
const result = await merge(slow, of("fast")).toArray();

console.log(result);

Output

["fast", "slow"]

zipPair corresponding items into tuples, completing and closing remaining sources as soon as the shortest source ends.

import { zip } from "mogami-stream";
const result = await zip([1, 2, 3], ["A", "B"]).toArray();

console.log(result);

Output

[
  [1, "A"],
  [2, "B"]
]

zipLatestEmit tuples of the latest values across sources; waits for all sources to emit once unless partial: true is set.

import { fromTimeline, of, zipLatest } from "mogami-stream";
const numbers = fromTimeline([
  { at: 0, kind: "value", value: 1 },
  { at: 20, kind: "value", value: 2 },
]);
const result = await zipLatest([numbers, of("A")]).toArray();

console.log(result);

Output

[
  [1, "A"],
  [2, "A"]
]

flattenFlatten a stream of child sources concurrently up to the configured concurrency limit.

import { flatten, from, fromTimeline, of } from "mogami-stream";
const children = [
  fromTimeline([{ at: 20, kind: "value", value: "slow" }]),
  of("fast"),
];
const result = await from(children)
  .pipe(flatten({ concurrency: 2 }))
  .toArray();

console.log(result);

Output

["fast", "slow"]

concatFlatten child sources strictly in child order while bounding active and waiting child slots.

import { concat, from, fromTimeline, of } from "mogami-stream";
const children = [
  fromTimeline([{ at: 20, kind: "value", value: "slow" }]),
  of("fast"),
];
const result = await from(children)
  .pipe(concat({ concurrency: 2 }))
  .toArray();

console.log(result);

Output

["slow", "fast"]

flatMapMap items synchronously to child sources and flatten them concurrently.

import { flatMap, from, fromTimeline } from "mogami-stream";
const result = await from([1, 2])
  .pipe(
    flatMap(
      (value) =>
        fromTimeline([{ at: value === 1 ? 20 : 0, kind: "value", value }]),
      { concurrency: 2 },
    ),
  )
  .toArray();

console.log(result);

Output

[2, 1]

concatMapMap items synchronously to child sources and flatten them in input order.

import { concatMap, from, fromTimeline } from "mogami-stream";
const result = await from([1, 2])
  .pipe(
    concatMap(
      (value) =>
        fromTimeline([{ at: value === 1 ? 20 : 0, kind: "value", value }]),
      { concurrency: 2 },
    ),
  )
  .toArray();

console.log(result);

Output

[1, 2]

pipeStandalone helper to wrap a source in a Stream and compose up to eight typed operators from left to right.

import { filter, map, pipe } from "mogami-stream";
const result = await pipe(
  [1, 2, 3],
  filter((value) => value > 1),
  map((value) => value * 2),
).toArray();

console.log(result);

Output

[4, 6]

Timing

rateLimitAllow at most count upstream read starts per perMs sliding window within a single execution.

import { from, rateLimit } from "mogami-stream";
const result = await from([1, 2, 3])
  .pipe(rateLimit({ count: 2, perMs: 100 }))
  .toArray();

console.log(result);

Output

[1, 2, 3]

spaceoutEmit the first item without added delay, then enforce at least intervalMs between subsequent emissions.

import { from, spaceout } from "mogami-stream";
const result = await from([1, 2, 3]).pipe(spaceout(20)).toArray();

console.log(result);

Output

[1, 2, 3]

delayStartWait ms milliseconds on the first downstream read before opening upstream.

import { delayStart, of } from "mogami-stream";
const result = await of("ready").pipe(delayStart(20)).toArray();

console.log(result);

Output

["ready"]

bufferTimeCollect items into non-empty batches, flushing when the first item's windowMs timer expires or maxItems is reached.

import { bufferTime, from } from "mogami-stream";
const result = await from([1, 2, 3])
  .pipe(bufferTime(1_000, { maxItems: 2 }))
  .toArray();

console.log(result);

Output

[[1, 2], [3]]

timeoutEachFail with TimeoutError if a single upstream read takes longer than ms milliseconds (excluding downstream idle time).

import { catchError, fromTimeline, of, timeoutEach } from "mogami-stream";
const result = await fromTimeline([{ at: 100, kind: "value", value: "late" }])
  .pipe(
    timeoutEach(10),
    catchError(() => of("timeout")),
  )
  .toArray();

console.log(result);

Output

["timeout"]

deadlineFail with DeadlineExceededError once total elapsed stage time reaches durationMs (including downstream idle time).

import { catchError, deadline, fromTimeline, of } from "mogami-stream";
const result = await fromTimeline([{ at: 100, kind: "value", value: "late" }])
  .pipe(
    deadline(10),
    catchError(() => of("deadline")),
  )
  .toArray();

console.log(result);

Output

["deadline"]

Error recovery

catchErrorWait for a failed upstream stage to finish cleanup, then switch once to the fallback source returned by handler.

import { catchError, fromTimeline, of } from "mogami-stream";
const result = await fromTimeline([
  { at: 0, kind: "value", value: "cached" },
  { at: 10, kind: "error", error: new Error("offline") },
])
  .pipe(catchError(() => of("fallback")))
  .toArray();

console.log(result);

Output

["cached", "fallback"]

Side effects & cleanup

tapRun a sequential sync or async side effect for each item before passing the item downstream.

import { from, tap } from "mogami-stream";
const seen: number[] = [];
const values = await from([1, 2])
  .pipe(
    tap((value) => {
      seen.push(value);
    }),
  )
  .toArray();
const result = { values, seen };

console.log(result);

Output

{ "values": [1, 2], "seen": [1, 2] }

finalizeRun a sync or async teardown callback once when a started stage closes, after child stages and in-flight work finish.

import { finalize, of } from "mogami-stream";
const result: string[] = [];
await of("A")
  .pipe(
    finalize(() => {
      result.push("closed");
    }),
  )
  .forEach((value) => {
    result.push(value);
  });

console.log(result);

Output

["A", "closed"]

Consumers

iterateAcquire a controlled StreamIterator exposing cancel(reason?), return(), throw(error?), and the closed report promise.

import { from } from "mogami-stream";
const iterator = from([1, 2, 3]).iterate();
const first = await iterator.next();
await iterator.return();
const result = { value: first.value, kind: (await iterator.closed).kind };

console.log(result);

Output

{ "value": 1, "kind": "completed" }

toArrayConsume the entire stream into an array after waiting for cleanup to complete.

import { toArray } from "mogami-stream";
const result = await toArray([1, 2, 3]);

console.log(result);

Output

[1, 2, 3]

reduceFold all values sequentially from an awaited initial value into a single final result after cleanup.

import { reduce } from "mogami-stream";
const result = await reduce([1, 2, 3], (sum, value) => sum + value, 0);

console.log(result);

Output

forEachRun a sequential sync or async effect for every item without collecting results in memory.

import { forEach } from "mogami-stream";
const result: number[] = [];
await forEach([1, 2, 3], async (value) => {
  result.push(value * 2);
});

console.log(result);

Output

[2, 4, 6]

findReturn the first matching item (or undefined) and close upstream immediately; supports synchronous type guards.

import { find } from "mogami-stream";
const result = await find([1, 2, 3], (value) => value > 1);

console.log(result);

Output

someShort-circuit to true on the first matching item and close upstream early, or return false for empty/non-matching streams.

import { some } from "mogami-stream";
const result = await some([1, 2, 3], (value) => value > 2);

console.log(result);

Output

true

everyShort-circuit to false on the first non-matching item and close upstream early, or return true if all items match.

import { every } from "mogami-stream";
const result = await every([1, 2, 3], (value) => value > 0);

console.log(result);

Output

true

See Choosing operators to select by task.

Operators ​

Sources

Transform

Selection & batching

Combination

Timing

Error recovery

Side effects & cleanup

Consumers

Operators