はじめに

1. インストール

使っているパッケージマネージャで mogami-stream をインストールします。

npm

npm install mogami-stream

pnpm

pnpm add mogami-stream

Bun

bun add mogami-stream

mogami-stream は実行時の外部依存を持たない ESM パッケージ（sideEffects: false）で、TypeScript の型定義を同梱しています。Node.js 22 以降、Bun、ブラウザのメインスレッド、Dedicated Worker、Service Worker、Cloudflare Workers で動作します。

2. パイプラインの構築と消費

まず、配列の各値を2倍にします。

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 と .map() などの変換メソッドは処理を組み立てるだけです。.toArray() を呼ぶと読み取りが始まり、返された Promise は処理と後処理が終わると [2, 4, 6] に解決します。

用語	この例でのコード	役割
ソース	from([1, 2, 3])	値の供給元となる Stream<number> を作る。
変換	.map((value) => value * 2)	読み取り時に各値を2倍にする新しいストリームを返す。
終端処理（コンシューマ）	.toArray()	ストリームを読み取り、値を配列に集める。

Stream<T> は JavaScript 標準の AsyncIterable<T> プロトコルを実装しています。これは、消費側が next() で次の値を要求し、その到着を待つ仕組みです。for await...of でも読み取れます。

値はソースから出て、左から右へオペレータを通り、終端処理に届きます。ソースに近い側を上流、終端処理に近い側を下流と呼びます。この例では from(...) が map の上流、.toArray() が下流です。

並行制御やリソース管理が役立つ用途は、採用する場面を参照してください。

3. 非同期処理と停止

コールバックを非同期にし、concurrency で同時実行数を指定します。

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);

最大5件を並行取得し、完了順に出力します。rateLimit はこの実行の上流読み取りを毎秒5回までに制限します。take(20) は出力件数の上限です。開始するリクエストも20件に制限したい場合は、take(20) を map より前に置きます。

controller.abort() で停止でき、コールバックの signal を fetch に渡すと実行中のリクエストにも中断を伝えます。toArray は後処理が終わってから解決・拒否されます。

配列は繰り返し読み取れますが、生成済みのジェネレータは1回読み終えると空になります。毎回新しいソースが必要なら defer を使います。

次はパイプラインの組み立て方へ進んでください。機能の選択はオペレータの使い分け、リソースの解放はキャンセルと後処理で説明しています。