オペレータ

カテゴリ別に名前と説明をまとめています。行を開くと使用例と出力結果を確認でき、サンプルをコピーできます。正確な型や詳しい仕様は、名前から API リファレンスへ進んで確認してください。

時間制御の使い分けは時間制御ガイドを参照してください。

ソース

from同期・非同期の Iterable を Stream にラップする。再実行可能かどうかや Promise 要素はそのまま維持される。
ts
import { from } from "mogami-stream"; const result = await from([1, 2, 3]).toArray(); console.log(result);
出力
```
[1, 2, 3]
```

defer実行ごとの初回読み取り時に新しいソースを生成する。非同期の初期化と AbortSignal による中断に対応。

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

console.log(result);

出力

[[1], [2]]

of指定した値を順番に出力する再実行可能なストリームを生成する。

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

console.log(result);

出力

["A", "B"]

empty値を一切出力せず、初回読み取り時に即座に完了するストリームを生成する。

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

console.log(result);

出力

[]

rangestart から end（終端を含まない）までの安全な整数列を指定ステップで出力する。

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

console.log(result);

出力

[1, 3, 5]

fromCallbackコールバック型のイベントソースを上限付きキューとオーバーフロー方針（error / drop-oldest / drop-newest）でストリームに変換する。

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

出力

["A", "B"]

fromEventEventTarget のイベントを上限付きキューで受信し、ストリームとして扱う。

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

出力

["ready"]

fromTimeline初回読み取りからのミリ秒オフセットで値・完了・エラーをスケジュール再生するテスト・モック用ソース。

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

出力

["A", "B"]

変換

map並行実行数（concurrency）・容量上限（capacity）・出力順序・項目単位のリトライを指定して要素を非同期変換する。

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

出力

[2, 4, 6]

mapSettledmap と同じ並行制御で変換を行い、各要素の成功・失敗を SettledResult オブジェクトとしてストリームを止めずに出力する。

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

出力

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

filterMap要素を逐次変換し、結果が null または undefined の場合のみ除外する（0、false、空文字は保持）。

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

出力

[12, 0]

scan初期値から要素を逐次集約し、ステップごとの中間アキュムレータを順次出力する。

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

console.log(result);

出力

[1, 3, 6]

選択・分割

filter同期・非同期の条件関数を逐次評価して一致する要素のみを残す。同期の型ガード関数による型の絞り込みに対応。

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

console.log(result);

出力

[2, 4]

take先頭から最大 n 件を取得し、n 件目を下流に渡す前に上流のクローズ処理を開始する。

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

console.log(result);

出力

[1, 2]

skip先頭の n 件を読み飛ばし、残りの要素をすべて出力する。

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

console.log(result);

出力

[2, 3]

takeWhile条件関数が true を返す間だけ要素を出力し、false になった時点で上流を閉じて完了する。

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

console.log(result);

出力

[1, 2]

skipWhile条件関数が true を返す先頭区間を読み飛ばし、最初に false となった要素以降は条件評価なしでそのまま出力する。

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

console.log(result);

出力

[3, 1]

chunks要素を最大 size 件ずつの配列にバッチ化する。正常完了時には端数のバッチも出力される。

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

console.log(result);

出力

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

dedupe直前に出力した要素と Object.is でキーが一致する連続重複を除外する（ストリーム全体の重複排除ではない）。
ts
import { dedupe, from } from "mogami-stream"; const result = await from([1, 1, 2, 2, 1]).pipe(dedupe()).toArray(); console.log(result);
出力
```
[1, 2, 1]
```

合成

chain複数のソースを順番に連結し、前のソースのクリーンアップ完了を待ってから次のソースを開く。
ts
import { chain } from "mogami-stream"; const result = await chain([1, 2], [3, 4]).toArray(); console.log(result);
出力
```
[1, 2, 3, 4]
```

merge複数のソースをソースあたり1スロットで並行して読み取り、準備できたソースからラウンドロビン順に出力する。

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

出力

["fast", "slow"]

zip各ソースの同じ位置の要素をタプルにまとめ、最短のソースが終了した時点で残りのソースも閉じて完了する。
ts
import { zip } from "mogami-stream"; const result = await zip([1, 2, 3], ["A", "B"]).toArray(); console.log(result);
出力
```
[
  [1, "A"],
  [2, "B"]
]
```

zipLatest各ソースの最新値をタプルとして出力する。既定では全ソースの初回出力を待ち、partial: true なら未出力スロットを undefined で埋める。

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

出力

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

flattenストリームの要素として渡された子ストリームを指定の concurrency 上限で並行して読み取り、1本にまとめる。

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

出力

["fast", "slow"]

concat子ストリームの順序を厳密に保って平坦化する。先行する子が完了するまで後続の子のスロット上限も維持される。

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

出力

["slow", "fast"]

flatMap各要素を同期コールバックで子ストリームに変換し、並行して平坦化する（非同期セットアップには defer を併用）。

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

出力

[2, 1]

concatMap各要素を同期コールバックで子ストリームに変換し、入力順を保って平坦化する。

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

出力

[1, 2]

pipe任意のソースを Stream に包み、最大8個の型付きオペレータを左から右へ合成するスタンドアロン関数。

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

出力

[4, 6]

時間制御

rateLimit1つの実行内でスライディングウィンドウ（perMs ミリ秒）あたりの上流読み取り開始回数を最大 count 回に制限する。

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

console.log(result);

出力

[1, 2, 3]

spaceout最初の要素は遅延なしで通し、2件目以降の出力間隔を最低 intervalMs ミリ秒以上空ける。

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

console.log(result);

出力

[1, 2, 3]

delayStart最初の読み取り時に ms ミリ秒待機してから上流ソースを開く。

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

console.log(result);

出力

["ready"]

bufferTimeバッチの最初の要素から windowMs ミリ秒経過するか maxItems 件に達した時点で配列として出力する（空配列は出力しない）。

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

console.log(result);

出力

[[1, 2], [3]]

timeoutEach上流の1回の読み取りが ms ミリ秒を超えた場合に TimeoutError で失敗させる（下流のアイドル時間は含まない）。

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

出力

["timeout"]

deadline下流の待機時間も含め、ステージ開始からの総経過時間が durationMs ミリ秒に達した時点で DeadlineExceededError にする。

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

出力

["deadline"]

エラー回復

catchError上流で発生したエラーを捕捉し、上流のクリーンアップ完了を待ってからハンドラが返す代替ソースへ1回だけ切り替える。

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

出力

["cached", "fallback"]

副作用・後処理

tap各要素を下流へそのまま渡す前に、同期・非同期の副作用コールバックを逐次実行して完了を待つ。

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

出力

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

finalize開始されたステージが閉じる際、子ステージと実行中タスクの完了後に一度だけクリーンアップ処理を実行し完了を待つ。

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

出力

["A", "closed"]

終端処理

iterate即時中断用の cancel() や終了レポート Promise の closed を備えた制御用 StreamIterator を取得する。

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

出力

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

toArrayストリームの全要素を配列に収集し、クリーンアップ完了を待ってから解決する。
ts
import { toArray } from "mogami-stream"; const result = await toArray([1, 2, 3]); console.log(result);
出力
```
[1, 2, 3]
```

reduce初期値から全要素を逐次集約し、クリーンアップ完了後に最終結果を返す。

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

console.log(result);

出力

forEach要素をメモリに溜め込まず、各要素に対して逐次処理を実行し、クリーンアップ完了まで待つ。

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

console.log(result);

出力

[2, 4, 6]

find条件に一致する最初の要素が見つかり次第、上流を早期クローズしてその値を返す（一致なしは undefined）。型ガードに対応。
ts
import { find } from "mogami-stream"; const result = await find([1, 2, 3], (value) => value > 1); console.log(result);
出力
```
2
```
some条件を満たす要素が1つでも見つかれば即座に上流を閉じて true を返す（空または不一致なら false）。
ts
import { some } from "mogami-stream"; const result = await some([1, 2, 3], (value) => value > 2); console.log(result);
出力
```
true
```
every条件を満たさない要素が見つかり次第、即座に上流を閉じて false を返す（全要素が満たすか空なら true）。
ts
import { every } from "mogami-stream"; const result = await every([1, 2, 3], (value) => value > 0); console.log(result);
出力
```
true
```

用途から選ぶ場合はオペレータの使い分けを参照してください。

オペレータ ​

ソース

変換