Node.js の Streams は大規模 I/O 処理のコアであり続けているが、Node.js Streams と Web Streams の 2 つの API が並存していることは開発者に選択の困難をもたらしていた。Node.js 26 は、下層のリファクタリングにより、後方互換性を保ちながら 2 つの API の相互運用性をよりスムーズにした。

二重 Streams 体制の現状

Node.js 26 以前、Streams には 2 つの並列システムがあった：

// Node.js Streams（従来）
const { Readable, Writable, Transform } = require('node:stream')
const fs = require('node:fs')

const readStream = fs.createReadStream('file.txt')
const writeStream = fs.createWriteStream('output.txt')
readStream.pipe(transformStream).pipe(writeStream)

// Web Streams（モダン標準）
const response = await fetch('https://api.example.com/data')
const reader = response.body.getReader()
const writable = writableStream.getWriter()

while (true) {
  const { done, value } = await reader.read()
  if (done) break
  await writable.write(value)
}

Node.js 26 の統一戦略

Node.js 26 は「段階的統一」戦略を採用し、下層で 2 つの API のシームレスな相互運用を実現した：

ReadableStream の強化

import { ReadableStream } from 'node:stream/web'
import { createReadStream } from 'node:fs'

// Node.js 26：fs.createReadStream から直接 Web ReadableStream を作成
const nodeStream = createReadStream('large-file.txt')
const webStream = ReadableStream.from(nodeStream)

// Web ReadableStream を fetch に直接使用
const response = new Response(webStream, {
  headers: { 'Content-Type': 'text/plain' }
})

// 逆変換：Web Stream → Node Stream
const { Readable } = await import('node:stream')
const nodeReadable = Readable.fromWeb(webReadable)

TransformStream の相互運用

import { TransformStream } from 'node:stream/web'

const upperCaseTransform = new TransformStream({
  transform(chunk, controller) {
    const text = new TextDecoder().decode(chunk)
    controller.enqueue(new TextEncoder().encode(text.toUpperCase()))
  }
})

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'

await pipeline(
  createReadStream('input.txt'),
  upperCaseTransform,  // Web TransformStream を直接使用
  createWriteStream('output.txt')
)

非同期イテレーター：モダン Streams のベストプラクティス

Node.js 26 は非同期イテレーターのサポートを強化し、Streams を処理する最も直感的な方法となっている：

import { createReadStream } from 'node:fs'

async function* processLines(readable) {
  let buffer = ''
  
  for await (const chunk of readable) {
    buffer += chunk
    const lines = buffer.split('\n')
    buffer = lines.pop() || ''
    
    for (const line of lines) {
      if (line.trim()) {
        yield JSON.parse(line)
      }
    }
  }
  
  if (buffer.trim()) {
    yield JSON.parse(buffer)
  }
}

const fileStream = createReadStream('events.jsonl')
for await (const event of processLines(fileStream)) {
  console.log('Event:', event.type)
}

バックプレッシャー処理の簡略化

バックプレッシャーは Streams プログラミングで最もエラーが発生しやすい部分の一つだ。Node.js 26 はバックプレッシャー処理を簡略化した：

import { Writable } from 'node:stream'

// Node.js 26：自動バックプレッシャー処理
const writable = new Writable({
  write(chunk, encoding, callback) {
    setTimeout(() => {
      console.log('Processed:', chunk.toString())
      callback()  // コールバック完了後に次の chunk が書き込まれる
    }, 100)
  },
  highWaterMark: 16 * 1024  // 16KB
})

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

await pipeline(
  createReadStream('large-input.bin'),
  transformStream,
  createWriteStream('output.bin')
)

エラーハンドリングのベストプラクティス

import { pipeline } from 'node:stream/promises'
import { createReadStream, createWriteStream } from 'node:fs'

async function processFile(inputPath, outputPath) {
  try {
    await pipeline(
      createReadStream(inputPath),
      transformStream,
      createWriteStream(outputPath)
    )
    console.log('File processed successfully')
  } catch (err) {
    if (err.code === 'ENOENT') {
      console.error('Input file not found')
    } else if (err.code === 'EACCES') {
      console.error('Permission denied')
    } else {
      console.error('Stream processing failed:', err.message)
    }
    throw err
  }
}

まとめ

Node.js 26 の Streams リファクタリングにより、Node.js Streams と Web Streams API の間の真の相互運用が実現された。強化された非同期イテレーター、簡略化されたバックプレッシャー処理、Fetch API との深い統合により、Streams プログラミングがより直感的で効率的になった。実際のプロジェクトでは、適切な highWaterMark の選択、エラーとリソースの適切な処理、再利用可能なデータパイプラインの構築が Streams のパフォーマンス利点を活かす鍵となる。