Node.js Streams チートシート

Streams はデータを小さなチャンク単位で順次処理する仕組みであり、巨大なファイルやネットワーク通信をメモリを浪費せずに高速処理できます。

ストリームの基本種別

---

Pipeline と最新の stream/promises

最新の Node.js では pipeline を使用して、自動エラーハンドリングとリソース解放を伴い安全にストリームを連結できます：

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');

async function compressFile() {
  await pipeline(
    fs.createReadStream('source.tar'),
    zlib.createGzip(),
    fs.createWriteStream('source.tar.gz')
  );
  console.log('圧縮が完了しました');
}

---

カスタムストリームの作成

1. カスタム Readable ストリーム

const { Readable } = require('stream');

function createTimestampStream() {
  let count = 0;
  return new Readable({
    objectMode: true,
    read() {
      if (count >= 5) {
        this.push(null); // ストリームの終端
      } else {
        this.push({ id: ++count, time: new Date() });
      }
    }
  });
}

2. カスタム Transform ストリーム

const { Transform } = require('stream');

const uppercaseTransformer = new Transform({
  transform(chunk, encoding, callback) {
    const transformed = chunk.toString().toUpperCase();
    callback(null, transformed);
  }
});

3. カスタム Writable ストリーム

const { Writable } = require('stream');

const consoleWriter = new Writable({
  objectMode: true,
  write(chunk, encoding, callback) {
    console.log('チャンクを書き込み中:', chunk);
    callback();
  }
});

---

Flowing モードと Paused モード

Readable ストリームは一時停止 (Paused) または自動流出 (Flowing) モードで動作します：

const stream = fs.createReadStream('data.csv');

// 'data' イベントリスナーを登録すると自動的に Flowing モードへ移行
stream.on('data', (chunk) => {
  console.log(`受信バイト数: ${chunk.length} バイト`);
});

// 明示的なフロー制御
stream.pause();  // イベント送信を一時停止
stream.resume(); // ストリーミングイベントを再開

---

ストリームの主要イベント

---

関連リファレンス

• Node.js Streams 公式ドキュメント
• Substack による Stream Handbook
• Node.js fs モジュール チートシート
• Node.js コア API チートシート