オペレータの使い分け

各オペレータの使用例と API リファレンスへのリンクは一覧にあります。並行数や時間制限を設定する前に、値の流し方を選びます。

目的	オペレータ	動作
変換・絞り込み	map、filter、filterMap	map は並行実行に対応。他の2つは逐次実行。
件数制限・バッチ化	take、chunks、bufferTime	配置した位置で件数を数える。処理順を参照。
指定したソースを順に連結	chain(a, b)	前のソースを閉じてから次を開く。
指定したソースの値を混ぜて出力	merge(a, b)	ソースを並行して読み取る。
子ソースのストリームを平坦化	concat、flatten	concat は子の順序を維持。flatten は準備できた子から出力。
項目ごとに子ソースを生成	concatMap、flatMap	同期コールバックで生成し、子の順序どおりに、または準備できた子から出力。非同期の準備には defer を使う。
同じ位置の値を組にする	zip(a, b)	最も要素数の少ないソースが終了すると停止。
最新値を組にする	zipLatest([a, b])	全ソースの初回出力を待つ。partial: true なら未出力を undefined で埋める。

concat・flatten・concatMap・flatMap の既定値は concurrency: 1 です。並行数を増やしても、concat と concatMap は子ソースの順序どおりに出力します。並行数の上限も参照してください。

読み取りペースや期限は時間制御、再試行や代替ソースはエラー処理で説明しています。読み取りを始めるには終端処理を選びます。

集約する状態

scan は更新のたびに累積値を出力し、reduce は最後の値だけを返します。どちらも初期値と集約コールバックの結果を await し、入力が空なら scan は何も出力せず、reduce は初期値を返します。

オブジェクトは複製されず、参照が渡されます。各時点の状態を残すなら、新しいオブジェクトを返してください。状態を直接変更する場合は、defer 内で実行ごとに作り、次の変更前に必要な値を取り出します。

import { defer, from } from "mogami-stream";

export const totals = defer(() =>
  from([1, 2])
    .scan(
      (state, value) => {
        state.total += value;
        return state;
      },
      { total: 0 },
    )
    .map((state) => state.total),
);
console.log(await totals.toArray()); // [1, 3]
console.log(await totals.toArray()); // [1, 3]

Promise を要素として扱う

from と選択・合成オペレータは Promise オブジェクトを要素として保持します。同期 Iterable に対する標準の for await と異なり、要素の Promise を解決値に置き換えません。解決値を得るには map を使います。

export const pending = Promise.resolve(2);
export const preserved = await from([pending]).toArray(); // [Promise<2>]
export const resolved = await from([pending])
  .map((value) => value)
  .toArray(); // [2]

find の条件関数には元の要素が渡りますが、一致した Promise を返す際にはその解決を待ちます。