TransformStreamで行単位のテキスト分割を実装: LineSplitTransform

LineSplitTransformは、ストリームから受け取ったバイナリデータ(Uint8Array)をテキストに変換し、改行文字で分割するためのTransformStream実装だ。

連続したデータチャンクを処理し、各行ごとに出力することで、ログファイルや大きなテキストファイルの逐次処理に適している。

実装

TextDecoderを使ってバイト列を文字列に変換し、正規表現/\r?\n/で改行ごとに分割する。

バッファに保持された不完全な行は、次のチャンクで補完され、ストリームの終端時に残った内容も出力される仕組みだ。

export class LineSplitTransform extends TransformStream<Uint8Array, string> {
  constructor() {
    let buffer = "";
    const decoder = new TextDecoder();
    const delimiterRegex = /\r?\n/;

    super({
      transform(chunk, ctrl) {
        buffer += decoder.decode(chunk, { stream: true });
        const lines = buffer.split(delimiterRegex);
        buffer = lines.pop() || "";

        for (const line of lines) {
          ctrl.enqueue(line);
        }
      },
      flush(ctrl) {
        buffer += decoder.decode();
        if (buffer.length > 0) {
          ctrl.enqueue(buffer);
        }
      },
    });
  }
}

使用例

標準入力からのデータをLineSplitTransformを用いて行単位に分割し、各行をコンソールに出力する例。

import { LineSplitTransform } from './LineSplitTransform';

// 標準入力を LineSplitTransform にパイプする
process.stdin.pipeThrough(new LineSplitTransform())
  .on('data', (line: string) => {
    console.log(`受信した行: ${line}`);
  });

活用シーン

ログファイルやテキストファイルのリアルタイム解析
ネットワークから受信したデータの逐次行処理
改行区切りテキストプロトコルの実装

まとめ

TransformStreamを使えば、ストリームデータを行単位で処理する機構を簡単に実装できる。

大きなファイルを一度にメモリに読み込まず、逐次処理できるため、メモリ効率の良いテキスト処理が実現できる。