Getting started

1. Installation

Install mogami-stream with your preferred package manager:

npm

npm install mogami-stream

pnpm

pnpm add mogami-stream

Bun

bun add mogami-stream

mogami-stream ships pure ESM JavaScript and TypeScript declarations (sideEffects: false) with zero runtime dependencies. Supported environments include Node.js 22+, Bun, browser main threads, Dedicated Workers, Service Workers, and Cloudflare Workers.

2. Building and consuming a pipeline

Start with a finite array and double each value:

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]

from and transformation methods such as .map() only describe the work. Calling .toArray() starts reading and returns a Promise for [2, 4, 6] after processing and cleanup finish.

Term	In this example	Role
Source	from([1, 2, 3])	Supplies values as a Stream<number>.
Transformation	.map((value) => value * 2)	Returns a new stream that doubles each input value when read.
Terminal consumer	.toArray()	Reads the stream and collects its values.

Stream<T> implements JavaScript's standard AsyncIterable<T> protocol: a consumer asks for the next value with next() and awaits its arrival. You can also consume a stream with for await...of.

Values flow from the source, through operators from left to right, to the consumer. Upstream means toward the source; downstream means toward the consumer. Here, from(...) is upstream of map, and .toArray() is downstream of it.

See When to use mogami-stream for examples of tasks that benefit from its concurrency and lifecycle controls.

3. Asynchronous work and stopping

Make the callback asynchronous and set concurrency to bound simultaneous work.

import { from } from "mogami-stream";

const urls = ["https://example.com/a", "https://example.com/b"];
const controller = new AbortController();

const texts = from(urls)
  .rateLimit({ count: 5, perMs: 1_000 })
  .map(
    async (url, { signal }) => {
      const response = await fetch(url, { signal });
      if (!response.ok) {
        await response.body?.cancel();
        throw new Error(`HTTP ${response.status}`);
      }
      return response.text();
    },
    { concurrency: 5, capacity: 5, ordered: false },
  )
  .take(20);

const results = await texts.toArray({ signal: controller.signal });
console.log(results);

This fetches up to five URLs concurrently and emits results in completion order. rateLimit limits this execution's upstream reads to five per second. take(20) limits output count; place it before map if you also need to limit started requests to 20.

Call controller.abort() to stop. Passing the callback's signal to fetch propagates cancellation to in-flight requests. toArray settles after cleanup finishes.

Arrays can be read again, but an instantiated generator is exhausted after one consumption. Use defer when each execution needs a fresh source.

Continue with Building a pipeline. See Choosing operators for selection guidance and Cancellation and cleanup for releasing resources.