この投稿では、Node.jsのストリームのデータをパイプ(pipe)とTransformクラスを使って変形する方法を説明する。なお、サンプルコードはすべてTypeScriptで示す。
Readableストリームは源流
パイプとTransformの使い方を見る前に、ストリームの源流となるReadableストリームの最もシンプルな作り方をおさらいしておく。次のコードがその例だ:
import {Readable} from 'stream'
const stream = Readable.from(['a', 'b', 'c'])
stream.on('data', data => process.stdout.write(data)) // ストリームのデータを標準出力に書き出す
このコードを実行すると、標準出力にabcと表示される。
パイプとはストリームを橋渡しする土管のこと
次に、パイプの使い方を見てみよう。ストリーム処理において、パイプとはその名が示す通り、あるストリームの内容を別のストリームに流し込む土管のような役割をする。Nodeのstreamでは、Readableストリームオブジェクトのpipeメソッドとして実装されている。pipeメソッドは第一引数にWritableストリームオブジェクトを受け付ける。
readableStream.pipe(writableStream)
// readableStreamからwritableStreamにデータが流れるようになる
最初のサンプルコードで使ったprocess.stdoutオブジェクトは、Writableストリームオブジェクトとして振る舞える。なので、'data'イベントでわざわざprocess.stdout.writeを呼び出していた箇所は、pipeメソッドでシンプルに書き直すことができる:
import {Readable} from 'stream'
const stream = Readable.from(['a', 'b', 'c'])
// stream.on('data', data => process.stdout.write(data))
stream.pipe(process.stdout)
Transformクラス
Transformはストリームデータを逐次加工できる
ここまででReadableストリームの作り方と、パイプの使い方を見てきたので、次にTransformクラスの使い方について説明する。
TransformクラスはReadableストリームであると同時にWritableストリームでもある。つまり、ストリームパイプの途中に入って、上流から流れてきたデータを加工して、下流に流すことができる。
// 上流 ---------データの流れ---------> 下流
stream.pipe(transform).pipe(process.stdout)
// ↑加工
Transformオブジェクトの生成方法
Transformクラスは抽象クラスのようなもので、インスタンス化するときはtransformメソッドの実装をコンストラクタに渡す:
new Transform({
// transformメソッドの実装を渡す
transform(chunk, encoding, done) {
// ...
},
})
このメソッドでは、上流から流れてきたデータの断片(チャンク)を逐次変形する処理を実装することになる。
“何も変形しない”Transformを実装する
Transformの説明をシンプルにするために、何も変形をしない、つまり、データを受け流すだけのTransformオブジェクトを実装してみよう:
import {Transform, TransformCallback} from 'stream'
// 何も変形しないTransform
const noop = new Transform({
transform(
chunk: string | Buffer,
encoding: string,
done: TransformCallback
): void {
this.push(chunk) // データを下流のパイプに渡す処理
done() // 変形処理終了を伝えるために呼び出す
},
})
noopトランスフォームをパイプラインに組み込むと次のようになる:
import {Readable} from 'stream'
const stream = Readable.from(['a', 'b', 'c'])
stream.pipe(noop).pipe(process.stdout)
noopは何も変形しないので、出力結果は当然abcのままになる。
アルファベットを大文字する変形するTransformを実装する
今度は、noopをベースに、流れてきた文字列データを加工するTransformを実装してみよう。次のサンプルコードは、アルファベットを大文字するものだ:
// アルファベットを大文字にするTransform (変形する)
const uppercase = new Transform({
transform(
chunk: string | Buffer,
encoding: string,
done: TransformCallback
): void {
this.push(chunk.toString().toUpperCase()) // 加工処理
done()
},
})
これをパイプラインに組み込む:
stream.pipe(uppercase).pipe(process.stdout)
このパイプラインを実行すると、標準出力の結果はABCになる。