パイプラインの組み立て方

タスクの一覧から、未完了のタスクを先頭から2件選び、タイトルの前後の空白を取り除きます。「選ぶ」「変換する」「件数を決める」をそれぞれオペレータに対応させて組み立てます。

オペレータの適用順

import { from, type Operator } from "mogami-stream";

type Task = { title: string; done: boolean };

const tasks: Task[] = [
  { title: " Review draft ", done: false },
  { title: "Send invoice", done: true },
  { title: " Book room ", done: false },
  { title: "Update calendar", done: false },
];

const titles = from(tasks)
  .filter((task) => !task.done)
  .map((task) => task.title.trim())
  .take(2);

const result = await titles.toArray();
console.log(result); // ["Review draft", "Book room"]

filter で未完了のタスクを選び、map でタイトルを整え、take(2) で結果を2件に制限します。完了済みのタスクは map に届かないため、そのタイトルを整える処理は行いません。

オペレータはメソッドの記述順に適用されます。この例の map はタスクを文字列に変えるので、その後では task.done を参照できません。タスクの状態による絞り込みは変換前に置きます。

各ステージの値と型

Stream<T> の T はその段階で流れる値の型です。各段階から出る値を追うと、次のコールバックが何を受け取るか分かります。

段階	値	型
from(tasks)	タスクのオブジェクト	Stream<Task>
filter(...)	未完了のタスクのオブジェクト	Stream<Task>
map(...)	前後の空白を除いたタイトル	Stream<string>
take(2)	最大2件のタイトル	Stream<string>
.toArray()	タイトルを集めた配列	Promise<string[]>

この例の filter は対象を選び、map は値と型を変えます。take は流れる値の型を変えません。await した result は string[] です。

take の位置と取得件数

take(2) が数えるのはその位置に届いた値です。未完了のタスクを2件選ぶなら、最初の例のように filter より後ろに置きます。未完了が2件に満たなければ、得られる結果も2件未満です。

先に入力を2件に制限すると、対象は元の配列の先頭2件だけになります。

const firstTwoInputs = from(tasks)
  .take(2)
  .filter((task) => !task.done)
  .map((task) => task.title.trim());

console.log(await firstTwoInputs.toArray()); // ["Review draft"]

2件目は完了済みなので、結果は "Review draft" の1件です。後にある "Book room" は読み取りません。制限したいのが入力件数か、条件に合う件数かで、take の位置を選びます。

配列への収集と1件ずつの処理

最初の例の .toArray() は結果を配列にまとめて使う場合に向いています。このパイプラインでは take(2) があるので、配列に入るのは最大2件です。件数を制限せずに使う場合は、全結果をメモリに保持できるか確認してください。

配列が不要なら、同じ titles を forEach で1件ずつ処理できます。

await titles.forEach((title) => console.log(title));
// Review draft
// Book room

この呼び出しは配列 tasks から新たに読み取る別の実行です。forEach は処理結果の配列を作らず、非同期のコールバックも1件ずつ完了を待ちます。

集約や条件による早期終了は終端処理、非同期処理の同時実行は並行数・容量・需要を参照してください。外部 I/O を扱う場合の停止と解放はキャンセルと後処理で説明しています。

.pipe() と独自オペレータ

複数の独自オペレータは .pipe(a, b) で渡せます。各オペレータは直前のソースを受け取って新しいストリームを返します。返されたストリームにも変換メソッドを適用できます。

function trimTitles(): Operator<string, string> {
  return (source) => from(source).map((title) => title.trim());
}

function titleLengths(): Operator<string, number> {
  return (source) => from(source).map((title) => title.length);
}

const lengths = from(tasks)
  .filter((task) => !task.done)
  .map((task) => task.title)
  .pipe(trimTitles(), titleLengths())
  .take(2);

console.log(await lengths.toArray()); // [12, 9]

エントリポイントと型推論

mogami-stream/core は関数形式のオペレータ、.pipe()、終端メソッドを提供します。.map() などの変換メソッドを含まないため、バンドラーが未使用オペレータを除去しやすくなります。

import { from, map, take } from "mogami-stream/core";

const values = await from([1, 2, 3])
  .pipe(
    map((value) => value * 2),
    take(2),
  )
  .toArray();

console.log(values); // [2, 4]

実行と後処理の仕組みはどちらのエントリでも共通です。core のストリームで変換メソッドを使うには、mogami-stream から読み込んだ from() に渡します。import するだけでは、既存のストリームにメソッドは追加されません。

メソッドチェーンの段数に上限はありません。.pipe() と関数形式の pipe は1回につき最大8個のオペレータまで型推論するため、長いパイプラインは .pipe() を分けます。const first = take(1) のような値の型に依存しないオペレータは、異なる要素型でも再利用できます。