streams

Node.js ストリームスキル

大容量ファイルのメモリ効率に優れた処理、リアルタイムデータ処理、そして組み合わせ可能なデータパイプラインの構築に役立つストリームを習得します。

クイックスタート

ストリームは4種類あります。

1. Readable - データのソース（ファイル、HTTP リクエスト）
2. Writable - データの出力先（ファイル、HTTP レスポンス）
3. Transform - データの転送中に変換します
4. Duplex - 読み取りと書き込みの両方が可能です

コアコンセプト

Readable ストリーム

const fs = require('fs');

// Create readable stream
const readStream = fs.createReadStream('large-file.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024 // 64KB chunks
});

// Event-based consumption
readStream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes`);
});

readStream.on('end', () => {
  console.log('Finished reading');
});

readStream.on('error', (err) => {
  console.error('Read error:', err);
});

Writable ストリーム

const writeStream = fs.createWriteStream('output.txt');

// Write data
writeStream.write('Hello, ');
writeStream.write('World!\n');
writeStream.end(); // Signal end

// Handle backpressure
const ok = writeStream.write(data);
if (!ok) {
  // Wait for drain event before writing more
  writeStream.once('drain', () => {
    continueWriting();
  });
}

Transform ストリーム

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

// Custom transform: uppercase text
const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

// Usage
fs.createReadStream('input.txt')
  .pipe(upperCase)
  .pipe(fs.createWriteStream('output.txt'));

学習パス

初級（1〜2週間）

• ✅ ストリームの種類を理解する
• ✅ ファイルストリームの読み書き
• ✅ 基本的なパイプ操作
• ✅ ストリームイベントの処理

中級（3〜4週間）

• ✅ Transform ストリーム
• ✅ バックプレッシャーの処理
• ✅ オブジェクトモードのストリーム
• ✅ パイプラインユーティリティ

上級（5〜6週間）

• ✅ カスタムストリームの実装
• ✅ 非同期イテレーター
• ✅ Web Streams API
• ✅ パフォーマンスの最適化

パイプライン（推奨）

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

// Compose streams with error handling
async function compressFile(input, output) {
  await pipeline(
    fs.createReadStream(input),
    zlib.createGzip(),
    fs.createWriteStream(output)
  );
  console.log('Compression complete');
}

// With transform
await pipeline(
  fs.createReadStream('data.csv'),
  csvParser(),
  transformRow(),
  jsonStringify(),
  fs.createWriteStream('data.json')
);

エラー処理を含むパイプライン

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

pipeline(
  source,
  transform1,
  transform2,
  destination,
  (err) => {
    if (err) {
      console.error('Pipeline failed:', err);
    } else {
      console.log('Pipeline succeeded');
    }
  }
);

HTTP ストリーミング

const http = require('http');
const fs = require('fs');

// Stream file as HTTP response
http.createServer((req, res) => {
  const filePath = './video.mp4';
  const stat = fs.statSync(filePath);

  res.writeHead(200, {
    'Content-Type': 'video/mp4',
    'Content-Length': stat.size
  });

  // Stream instead of loading entire file
  fs.createReadStream(filePath).pipe(res);
}).listen(3000);

// Stream HTTP request body
http.createServer((req, res) => {
  const writeStream = fs.createWriteStream('./upload.bin');
  req.pipe(writeStream);

  req.on('end', () => {
    res.end('Upload complete');
  });
}).listen(3001);

オブジェクトモードのストリーム

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

const jsonParser = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    try {
      const obj = JSON.parse(chunk);
      this.push(obj);
      callback();
    } catch (err) {
      callback(err);
    }
  }
});

// Process objects instead of buffers
const processRecords = new Transform({
  objectMode: true,
  transform(record, encoding, callback) {
    record.processed = true;
    record.timestamp = Date.now();
    this.push(record);
    callback();
  }
});

非同期イテレーター

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

// Create from async iterator
async function* generateData() {
  for (let i = 0; i < 100; i++) {
    yield { id: i, data: `item-${i}` };
  }
}

const stream = Readable.from(generateData(), { objectMode: true });

// Consume with for-await
async function processStream(readable) {
  for await (const chunk of readable) {
    console.log('Processing:', chunk);
  }
}

バックプレッシャーの処理

const readable = fs.createReadStream('huge-file.txt');
const writable = fs.createWriteStream('output.txt');

readable.on('data', (chunk) => {
  // Check if writable can accept more data
  const canContinue = writable.write(chunk);

  if (!canContinue) {
    // Pause reading until writable is ready
    readable.pause();
    writable.once('drain', () => {
      readable.resume();
    });
  }
});

// Or use pipeline (handles automatically)
pipeline(readable, writable, (err) => {
  if (err) console.error('Error:', err);
});

カスタム Readable ストリーム

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

class DatabaseStream extends Readable {
  constructor(query, options) {
    super({ ...options, objectMode: true });
    this.query = query;
    this.cursor = null;
  }

  async _read() {
    if (!this.cursor) {
      this.cursor = await db.collection('items').find(this.query).cursor();
    }

    const doc = await this.cursor.next();
    if (doc) {
      this.push(doc);
    } else {
      this.push(null); // Signal end
    }
  }
}

// Usage
const dbStream = new DatabaseStream({ status: 'active' });
for await (const item of dbStream) {
  console.log(item);
}

ユニットテストのテンプレート

const { Readable, Transform } = require('stream');
const { pipeline } = require('stream/promises');

describe('Stream Processing', () => {
  it('should transform data correctly', async () => {
    const input = Readable.from(['hello', 'world']);
    const chunks = [];

    const upperCase = new Transform({
      transform(chunk, enc, cb) {
        this.push(chunk.toString().toUpperCase());
        cb();
      }
    });

    await pipeline(
      input,
      upperCase,
      async function* (source) {
        for await (const chunk of source) {
          chunks.push(chunk.toString());
        }
      }
    );

    expect(chunks).toEqual(['HELLO', 'WORLD']);
  });
});

トラブルシューティング

問題	原因	解決策
メモリ使用量が増え続ける	バックプレッシャーがない	pipeline を使うか、drain を処理する
データが失われる	エラーが捕捉されていない	エラーコールバック付きの pipeline を使う
処理が遅い	チャンクサイズが小さい	highWaterMark を大きくする
ストリームが停止する	end() の呼び出しがない	writable.end() を呼び出す

使用する場面

次の場合にストリームを使用します。

• 大容量ファイル（GB 以上）の処理
• リアルタイムデータ処理
• メモリに制約のある環境
• データパイプラインの構築
• HTTP リクエスト／レスポンスの処理

関連スキル

• 非同期プログラミング（非同期パターン）
• パフォーマンス最適化（メモリ効率）
• Express REST API（ストリーミングレスポンス）

リソース

• Node.js ストリームのドキュメント
• ストリームハンドブック
• Node.js ストリームガイド