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>`型へパイプすることができます。

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/678625/9eab9bb2-36e4-a473-faf8-4ccad0a10868.png)

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/678625/f48f7321-892e-8b35-fc43-b14daabd9ac6.png)

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

```ts
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`クラスを作るには、以下のようにします。

```ts
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を繋げることができます。

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/678625/1446f086-fdc9-c2d7-ddc7-04f5e3e6403f.png)


## 3つ以上のTransformStreamを合成する

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

```ts
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);
```

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/678625/c13d4b82-874a-9a2c-6473-3e090be1c886.png)


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

ここで紹介した方法を使うと、例えば[TextDecoderStream](https://developer.mozilla.org/en-US/docs/Web/API/TextDecoderStream)と[CSVStream](https://doc.deno.land/https://deno.land/std@0.136.0/encoding/csv/stream.ts/~/CSVStream)を組み合わせて、ファイルから読み取ったUint8ArrayをCSVとして2次元配列に変換するTransformStreamを作ったりできます。