ストリーム

注意

To use Node.js APIs in your Worker, add the nodejs_compat_v2 compatibility flag to your wrangler.toml file.

Node.js ストリーム API ↗ は、JavaScript におけるストリーミングデータを扱うための元々の API であり、WHATWG ReadableStream 標準 ↗よりも前に存在します。ストリームは、Node.js におけるストリーミングデータを扱うための抽象インターフェースです。ストリームは読み取り可能、書き込み可能、またはその両方であることができます。すべてのストリームは EventEmitter のインスタンスです。

可能な限り、WHATWG 標準の “Web Streams” API ↗ を使用するべきであり、これは Workers でサポートされています ↗。

import {
  Readable,
  Transform,
} from 'node:stream';

import {
  text,
} from 'node:stream/consumers';

import {
  pipeline,
} from 'node:stream/promises';

// データを大文字に変換し、出力の最後に改行を追加する Node.js スタイルの Transform
class MyTransform extends Transform {
  constructor() {
    super({ encoding: 'utf8' });
  }
  _transform(chunk, _, cb) {
    this.push(chunk.toString().toUpperCase());
    cb();
  }
  _flush(cb) {
    this.push('\n');
    cb();
  }
}

export default {
  async fetch() {
    const chunks = [
      "hello ",
      "from ",
      "the ",
      "wonderful ",
      "world ",
      "of ",
      "node.js ",
      "streams!"
    ];

    function nextChunk(readable) {
      readable.push(chunks.shift());
      if (chunks.length === 0) readable.push(null);
      else queueMicrotask(() => nextChunk(readable));
    }

    // 配列からチャンクを発行する Node.js スタイルの Readable...
    const readable = new Readable({
      encoding: 'utf8',
      read() { nextChunk(readable); }
    });

    const transform = new MyTransform();
    await pipeline(readable, transform);
    return new Response(await text(transform));
  }
};

詳細については、Node.js の stream ドキュメント ↗を参照してください。