mogami-stream

TypeScript operators for safe asynchronous stream processing

Usage

1. Install

npm

npm install mogami-stream

pnpm

pnpm add mogami-stream

Bun

bun add mogami-stream

2. Run a pipeline

import { from } from "mogami-stream";

const doubled = from([1, 2, 3]).map((value) => value * 2);

const result = await doubled.toArray();
console.log(result); // [2, 4, 6]

In this pipeline:

• from creates a source from the array.
• .map(...) describes how to transform each value; it does not start processing yet.
• .toArray() consumes the stream and resolves to [2, 4, 6] after cleanup.

Continue with Getting started for the basic terms and a concurrent API request example.

Get started · Operators · API reference · 日本語ドキュメント

When to use mogami-stream

• Fetching many URLs with limits on concurrent requests and buffered results.
• Transforming database rows or async generator values as they arrive, without collecting the entire input in memory.
• Processing WebSocket or SSE events with a bounded queue and an explicit overflow policy.
• Retrying individual failed items, or collecting each item's success or failure while continuing to process the rest.
• Stopping once enough results are available or the user cancels, propagating cancellation to I/O and waiting for resource cleanup.

Design

• Bounded work and buffering. map and mapSettled bound active work with concurrency and cap the sum of pending reads, running callbacks, and buffered outputs with capacity. Higher-order operators (flatten, concat, flatMap, concatMap) similarly bound active child streams and prevent unbounded queues behind slow children.
• Demand-driven execution. Pipelines are lazy and do no work until the first next() call. Concurrent stages initiate new upstream reads only while a downstream read is pending; once a value is delivered, no further upstream reads start until the next pull.
• Separate cancellation and teardown. iterator.cancel(reason) rejects a pending next() immediately and propagates cancellation upstream via AbortSignal, while iterator.closed resolves with a CloseReport once all in-flight operations and finalizers finish.
• Awaited cleanup and error preservation. Asynchronous finalize callbacks run from inner stages outward after upstream stages and in-flight work settle. Terminal consumers wait for cleanup before resolving or rejecting, prioritizing the primary processing error and recording secondary errors in suppressedErrors.
• Standard iterable interoperability. Wrap any synchronous Iterable or AsyncIterable, chain typed methods such as .map() and .filter(), and consume streams with native for await...of loops or built-in terminal consumers.

Documentation map

• Getting started — Installation, a minimal pipeline, concurrent requests, and cancellation.
• Building a pipeline — Processing order, values and types, count limits, and consumption through one example.
• Choosing operators — Combination choices, accumulated state, and Promise-valued items.
• Operators — Descriptions, examples, and output grouped by category.
• Terminal consumers — Collecting, aggregating, processing each item, and stopping early.
• Concurrency, capacity, and demand — Interactive slot state machine, ordered: true vs. false, backpressure, and higher-order stream bounds.
• Timing — Read rates, output spacing, timeouts, and time-based batching.
• Error handling — Stopping on errors, per-item results, retries, fallback sources, and cleanup errors.
• Cancellation and cleanup — Stopping an execution, finalizer ordering, and resource ownership.
• Recipes — API requests, WebSocket & SSE, Database batches, Timeline testing, Event loop yielding, Shared request limits, and Results and cleanup errors.
• Benchmarks — Node.js and Bun performance comparisons and reproduction commands.