2つのTransformStreamを合成する

Node.jsの最新リリースではWeb Streamがグローバルに追加されました。また、Denoでも最近のリリースで各種APIがWeb Streamに対応しました。

Web Streamには、

• 読み取り用のReadableStream
• 書き込み用のWritableStream
• 変換用のTransformStream

の3種類のAPIがあります。

この記事では、ストリームの変換に使うTransformStreamを合成して、1つのTransformStreamを作成する方法について解説します。

TransformStreamの使い方

まずTransformStreamの使い方を、TypeScriptの型定義と共に解説したいと思います。

ReadableStream<A>型は、細切れになったA型のchunkが流れてくるストリームです。ReadableStream<A>型はWritableStream<A>型へパイプすることができます。

function pipe<A>(
  readable: ReadableStream<A>,
  writable: WritableStream<A>,
) {
  readable.pipeTo(writable);
}

ReadableStreamをWritableStreamに書き込む前に変換したい時はどうすればよいでしょうか。ここでTransformStreamの出番です。
ReadableStream<A>をReadableStream<B>型に変換するには、TransformStream<A, B>型のTransformStreamを使います。
こうすることでWritableStream<B>に書きこむことが可能になります。

function pipe<A, B>(
  readable: ReadableStream<A>,
  transform: TransformStream<A, B>,
  writable: WritableStream<B>,
) {
  readable.pipeThrough(transform).pipeTo(writable);
}

TransformStreamは複数個繋げることができます。

function pipe<A, B, C, D>(
  readable: ReadableStream<A>,
  transform1: TransformStream<A, B>,
  transform2: TransformStream<B, C>,
  transform3: TransformStream<C, D>,
  writable: WritableStream<B>,
) {
  readable
    .pipeThrough(transform1)
    .pipeThrough(transform2)
    .pipeThrough(transform3)
    .pipeTo(writable);
}

普通に使う際は上のように繋げればいいのですが、使うTransformStreamの組み合わせが決まっている場合は、合成して1つのTransformStreamにしたくなります。今回はその方法についての記事です。（ここまで前置き）

2つのTransformStreamを合成する

TransformStream1とTransformStream2を合成して、新しいTransformStreamクラスを作るには、以下のようにします。

class MyTransformStream extends TransformStream<number, number> {
  readable: ReadableStream<number>;
  writable: WritableStream<number>;
  constructor() {
    super();

    // 合成したいTransformStreamを生成
    const transform1 = new TransformStream1<number, number>();
    const transform2 = new TransformStream2<number, number>();

    // 新しいTransformStreamにwritableとreadableを設定
    this.writable = transform1.writable;
    this.readable = transform2.readable;

    // transform1からtransform2へパイプ
    transform1.readable.pipeTo(transform2.writable);
  }
}

// 使い方
// readable
//   .pipeThrough(new MyTransformStream())
//   .pipeTo(writable);

ポイントは、TransformStreamのreadbleプロパティとwritableプロパティを適切に繋げることです。

実は、TransformStreamのreadbleプロパティはReadableStreamに、writableプロパティはWritableStreamになっており、ここを通して読み書きすることができます。
まず、1つ目のTransformStreamのwritableを新しいTransformStreamのwritableに、2つ目のTransformStreamのreadableを新しいTransformStreamのreadableに代入します。
次に1つ目のTransformStreamのreadableを2つ目のTransformStreamのwritableにパイプすることで、2つのTransformStreamを繋げることができます。

文字で書くと分かりにくいので図にします。

3つ以上のTransformStreamを合成する

3つ以上のTransformStreamも、同様の方法で合成できます。

class MyTransformStream extends TransformStream<number, number> {
  readable: ReadableStream<number>;
  writable: WritableStream<number>;
  constructor() {
    super();

    // 合成したいTransformStreamを生成
    const transform1 = new TransformStream1<number, number>();
    const transform2 = new TransformStream2<number, number>();
    const transform3 = new TransformStream3<number, number>();
    const transform4 = new TransformStream4<number, number>();

    // 新しいTransformStreamにwritableとreadableを設定
    this.writable = transform1.writable; // 最初のTransformStreamのwritable
    this.readable = transform4.readable; // 最後のTransformStreamのreadable

    // transform1をtransform2に、2を3に、3を4にパイプ
    transform1.readable.pipeTo(transform2.writable);
    transform2.readable.pipeTo(transform3.writable);
    transform3.readable.pipeTo(transform4.writable);
  }
}

// 使い方
// readable
//   .pipeThrough(new MyTransformStream())
//   .pipeTo(writable);

図にするとこんな感じです。

ちなみにMDNやDeno標準ライブラリを見る限り、カスタムしたTransformStreamを自作する際は、上記のようにTransformStreamを継承（extends）したクラスを作るのが"お作法"のようです。理由はよく分かりません。

ここで紹介した方法を使うと、例えばTextDecoderStreamとCSVStreamを組み合わせて、ファイルから読み取ったUint8ArrayをCSVとして2次元配列に変換するTransformStreamを作ったりできます。