コンテンツへスキップ

ストリーム

Markdownとして表示 GitHubで編集 ドキュメント: 5b53f69

PhasisはWHATWG Streams Standardの全機能を実装しています。すべてのクラス、すべてのリーダーモード、すべてのヘルパーが揃っています。Nodeや最新のブラウザーで動作するコードは、Phasisでも動作します。

この実装はfetchが内部で使用しているものです。Response.bodyは実際のReadableStreamであり、Requestはストリーミングボディを受け入れ、Responseを複製するのに必要なtee()も動作します。

提供機能

クラス 状態
ReadableStream 完全
ReadableStreamDefaultController 完全
ReadableByteStreamController 完全
ReadableStreamDefaultReader 完全
ReadableStreamBYOBReader 完全
ReadableStreamBYOBRequest 完全
WritableStream 完全
WritableStreamDefaultController 完全
WritableStreamDefaultWriter 完全
TransformStream 完全
TransformStreamDefaultController 完全
ByteLengthQueuingStrategy 完全
CountQueuingStrategy 完全

よく使うメソッドも揃っています。pipeTo、pipeThrough、tee、getReader、非同期反復、Symbol.asyncIterator、{ min }を使ったBYOB読み取り。

生成

JavaScript
// Pull-based — only generates values when read
const counter = new ReadableStream({
  start(controller) { this.n = 0; },
  pull(controller) {
    if (this.n >= 5) { controller.close(); return; }
    controller.enqueue(this.n++);
  },
});

for await (const v of counter) console.log(v);  // 0,1,2,3,4
JavaScript
// Push-based — start enqueues everything eagerly
const small = new ReadableStream({
  start(c) {
    [10, 20, 30].forEach(n => c.enqueue(n));
    c.close();
  },
});

消費

読み取り方法は3つあります。

JavaScript
// 1. Reader
const reader = stream.getReader();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}
reader.releaseLock();

// 2. Async iteration (preferred)
for await (const chunk of stream) {
  console.log(chunk);
}

// 3. tee + parallel pipes
const [a, b] = stream.tee();
await Promise.all([a.pipeTo(sinkA), b.pipeTo(sinkB)]);

変換

JavaScript
const upper = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const result = new ReadableStream({
  start(c) { ['hi', 'there'].forEach(s => c.enqueue(s)); c.close(); }
}).pipeThrough(upper);

for await (const v of result) console.log(v);  // HI, THERE

バイトストリーム（BYOB）

JavaScript
const bytes = new ReadableStream({
  type: 'bytes',
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3, 4, 5, 6]));
    controller.close();
  },
});

const reader = bytes.getReader({ mode: 'byob' });
const buf = new Uint8Array(4);
const { value, done } = await reader.read(buf);
// value is a Uint8Array of length 4, sharing buffer with `buf`

{ min: N }がサポートされています — read(view, { min: 2 })は、要素が2つ以上ある場合にのみ解決されます。

バックプレッシャー

desiredSizeとreadyは仕様どおりに動作します。キューイング戦略を使って、プロデューサーにフィードバック経路を設けます:

JavaScript
const stream = new ReadableStream({
  pull(controller) {
    if (controller.desiredSize <= 0) return;  // back off
    controller.enqueue(nextChunk());
  },
}, new CountQueuingStrategy({ highWaterMark: 4 }));

fetch を使う

JavaScript
// Response body is a ReadableStream
const res = await fetch('https://example.com/big.txt');
let total = 0;
for await (const chunk of res.body) {
  total += chunk.byteLength;
}

// Request bodies can stream too
const upload = new ReadableStream({
  start(c) { c.enqueue(new TextEncoder().encode('hello')); c.close(); }
});
await fetch('https://example.com/upload', { method: 'POST', body: upload });

キャッシュ用の Tee

よくあるパターン: 本文を2回読み取ります。

JavaScript
const res = await fetch(url);
const [forCache, forUse] = res.body.tee();
await Promise.all([
  forCache.pipeTo(cacheWritable),
  parseJson(forUse),
]);

Response.clone()がこれを行い、新しいResponseを返します。

注意点

  • 同期プロデューサーのみ。 ブロッキングPHPI/Oを行うpull()はスクリプトをブロックします。HTTPにはfetchトランスポートフックを使い、ファイルの場合は事前にメモリへ読み込む方法を推奨します。setTimeout / setIntervalも利用できます（イベントループを参照）が、エンジンはデフォルトで仮想時間を使います。pull()内のsetTimeout(cb, 1000)はPHPの間スリープするのではなく、ループのクロックを進めます。リアルタイムでペースを調整するには、ホストが壁時計に合わせてtickEventLoop()を動かす必要があります。

関連項目