時間制御

読み取りのペース、出力間隔、待機時間の上限をオペレータで指定できます。時間の単位はすべてミリ秒です。いずれも読み取りが始まるまで、待機や時間の計測は行いません。

用途から選ぶ

オペレータ	用途
rateLimit	一定時間内の読み取り回数を制限する
spaceout	値を一定の間隔で出力する
delayStart	開始を遅らせる
timeoutEach	1回の読み取りの待ち時間を制限する
deadline	ステージ全体の経過時間を制限する
bufferTime	時間または件数でまとめる

読み取り回数を制限する

rateLimit({ count: 2, perMs: 1_000 }) はどの1,000 ms の範囲でも上流の読み取り開始を最大2回に制限します。枠が空いていれば続けて読み取れるため、出力を等間隔にする用途とは異なります。

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

const values = await from([1, 2, 3, 4, 5])
  .pipe(rateLimit({ count: 2, perMs: 1_000 }))
  .toArray();

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

map への上流読み取りのペースを抑える場合は、その前に rateLimit を置きます。再試行（retry）や複数実行をまたいでリクエスト数を制限する場合は共有リミッタのレシピを使ってください。同時実行数は別途 map の concurrency で指定します。

制限は実行ごとに独立し、失敗した読み取りや完了を確認する読み取りも1回に数えます。map 内部の再試行は数えません。再試行も含めた制御はリトライを参照してください。

出力の間隔を空ける

spaceout(500) は最初の値に待ち時間を加えず、2件目以降は直前の出力から少なくとも500 ms 空けます。上流の読み取りや下流の処理に使った時間も間隔に含まれます。すでに500 ms 経過していれば、追加では待ちません。

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

const values = await from([1, 2, 3, 4, 5]).pipe(spaceout(500)).toArray();

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

入力が即座に用意され、下流が続けて値を要求する場合の目安は次のとおりです。

値	rateLimit の読み取り開始	spaceout の出力
1	0 ms	0 ms
2	0 ms	500 ms
3	1,000 ms	1,000 ms
4	1,000 ms	1,500 ms
5	2,000 ms	2,000 ms

spaceout は上流から値を読み取った後で出力を待ちます。上流の処理開始回数を制限したい場合は rateLimit を使います。

開始だけを遅らせる

delayStart(500) は最初に読み取りを要求されてから500 ms 待ち、その後で上流を開きます。

import { delayStart, of } from "mogami-stream";

const stream = of("ready").pipe(delayStart(500));
console.log(await stream.toArray()); // ["ready"]

読み取りのタイムアウトと全体の期限

timeoutEach は1回の上流読み取りを待っている間だけ時間を測り、期限に達すると TimeoutError になります。下流が次の値を要求するまでの時間は含みません。値やストリームの終了を受け取った時点でも単調時計で経過時間を確認するため、タイマーの実行が遅れても、期限ちょうど・期限後の結果は受け入れません。timeoutEach(0) は最初の読み取り時に、上流を開かずに失敗します。

deadline はそのステージの最初の読み取りから時間を測り、期限に達すると DeadlineExceededError になります。下流が値を処理している間も時間が進みます。

import {
  DeadlineExceededError,
  deadline,
  fromTimeline,
  timeoutEach,
} from "mogami-stream";

const source = fromTimeline([
  { at: 0, kind: "value", value: "A" },
  { at: 800, kind: "value", value: "B" },
  { at: 1_600, kind: "value", value: "C" },
]);

console.log(await source.pipe(timeoutEach(1_000)).toArray()); // ["A", "B", "C"]

try {
  await source.pipe(deadline(1_500)).toArray();
} catch (error) {
  if (!(error instanceof DeadlineExceededError)) throw error;
  console.log(error.name); // "DeadlineExceededError"
}

この例では値が800 ms 間隔で到着するため各読み取りの1,000 ms 制限内に収まりますが、一方で全体の1,500 ms の期限は3件目が到着する前に切れます。ソースは再実行可能なので、2つの処理はそれぞれ最初から始まります。

タイムアウトの対象はそのオペレータから見た上流です。map の後に timeoutEach を置くと、map から次の結果を得るまでの待ち時間を制限します。並行実行中の個々のリクエストに同じ制限を設けるには、コールバック内の I/O にもタイムアウトを指定します。

エラーが起きると上流へ停止が伝わります。fetch(url, { signal }) のようにコールバックの signal を I/O に渡してください。終端処理は後処理の完了を待つため、停止に対応しない I/O があると、期限に達しても直ちには完了しません。キャンセルと後処理も参照してください。

時間または件数でまとめる

bufferTime はバッチの最初の値が到着してから windowMs が経過するか、maxItems 件に達すると配列を出力します。

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

const batches = await from([1, 2, 3, 4, 5])
  .pipe(bufferTime(1_000, { maxItems: 2 }))
  .toArray();

console.log(batches); // [[1, 2], [3, 4], [5]]

この例では件数上限で2件ずつ出力し、正常終了時に残りの1件を出力します。空の配列は出力せず、エラーやキャンセル時には未出力のバッチを破棄します。

時間窓が切れたときに上流の読み取りが待機中なら、その1回だけを保持します。下流が次のバッチを要求するまで、新しい読み取りやバッチの蓄積を続けることはありません。件数だけで区切る場合は chunks を使います。

時間は実行環境のタイマー精度に従うため、例の時刻は目安です。rateLimit の perMs と bufferTime の windowMs は正の値、それ以外の時間は0以上の有限値を指定します。count と maxItems は正の安全な整数です。