Node.jsで独自の変換ストリームを作成する

独自の変換ストリームを作成するにはstream.Transformクラスのコンストラクタかthrough2モジュールのthrough2関数を使用します。

stream.Transformクラスのコンストラクタ

var stream = require('stream');

var transform = new stream.Transform({
    objectMode: false, 
    readableObjectMode: false, 
    writableObjectMode: false, 
    highWaterMark: 16384, 
    decodeStrings: true, 
    encoding: null, 
    allowHalfOpen: true, 
    transform: function (chunk, encoding, callback) {
        callback(null, 'xxx');
        //or
        callback(new Error('yyy'), null);
    }, 
    flush: function (callback) {
        callback(null);
        //or
        callback(new Error('zzz'));
    }
});

第1引数にオプションをオブジェクトとして指定します。

オプションには下のようなものがあります。

• transform・・・データの変換処理を関数として指定します。この関数の第1引数は変換前のデータであり、第2引数は文字コードであり、第3引数はコールバック関数です。変換処理が完了した場合や変換処理の実行中にエラーが発生した場合には必ずこのコールバック関数を呼び出さなければなりません。このコールバック関数の第1引数にはエラーオブジェクト（エラーが発生しなかった場合にはnull）を指定し、第2引数には変換後のデータ（エラーが発生した場合にはnull）を指定します。
• flush・・・データの変換処理が完了した場合に呼び出される処理を関数として指定します。この関数の第1引数はコールバック関数です。処理が完了した場合や処理の実行中にエラーが発生した場合には必ずこのコールバック関数を呼び出さなければなりません。このコールバック関数の第1引数にはエラーオブジェクト（エラーが発生しなかった場合にはnull）を指定します。この関数では変換後のデータの末尾に付加するデータをthis.push関数を使用してバッファに追加することができます。この関数の第1引数にはデータを指定し、第2引数には文字コードを指定します。第2引数は指定しなくても構いません。このオプションは指定しなくても構いません。

また、stream.Duplexクラスのコンストラクタの第1引数のオプションを指定することもできます。

through2関数

var through2 = require('through2');

var transform = through2({
    objectMode: false, 
    readableObjectMode: false, 
    writableObjectMode: false, 
    highWaterMark: 16384, 
    decodeStrings: true, 
    encoding: null, 
    allowHalfOpen: true
}, function (chunk, encoding, callback) {
    callback(null, 'xxx');
    //or
    callback(new Error('yyy'), null);
}, function (callback) {
    callback(null);
    //or
    callback(new Error('zzz'));
});

第1引数にオプションをオブジェクトとして指定します。このオプションはstream.Transformクラスのコンストラクタの第1引数のオプションと同じです。この引数は指定しなくても構いません。

第2引数にデータの変換処理を関数として指定します。この関数はstream.Transformクラスのコンストラクタの第1引数のtransformオプションと同じです。

第3引数にデータの変換処理が完了した場合に呼び出される処理を関数として指定します。この引数は指定しなくても構いません。この関数はstream.Transformクラスのコンストラクタの第1引数のflushオプションと同じです。

サンプルコード1

stream.Transformクラスのコンストラクタの使用例です。

読み込まれたデータの文字を二重化する変換ストリームを作成します。

transform.js

var stream = require('stream');

var transform = new stream.Transform({
    decodeStrings: false, 
    encoding: 'utf-8', 
    transform: function (chunk, encoding, callback) {
        var str = '';
        for (var i = 0; i < chunk.length; i++) {
            str += chunk[i];
            str += chunk[i];
        }
        callback(null, str);
    }, 
    flush: function (callback) {
        this.push('end.');
        callback(null);
    }
});
transform.on('error', function (err) {
    console.error(err);
    process.exit(1);
});

transform.write('foo');
transform.write('foofoo');
transform.write('foofoofoo');
transform.write('foofoofoofoo');
transform.write('foofoofoofoofoo');
transform.end();

transform.on('data', function (r) {
    console.log(r);
});

実行結果

C:\work\node>node transform.js
ffoooo
ffooooffoooo
ffooooffooooffoooo
ffooooffooooffooooffoooo
ffooooffooooffooooffooooffoooo
end.

サンプルコード2

through2関数の使用例です。

読み込まれたデータの文字を二重化する変換ストリームを作成します。

through2.js

var through2 = require('through2');

var transform = through2({
    decodeStrings: false, 
    encoding: 'utf-8'
}, function (chunk, encoding, callback) {
    var str = '';
    for (var i = 0; i < chunk.length; i++) {
        str += chunk[i];
        str += chunk[i];
    }
    callback(null, str);
}, function (callback) {
    this.push('end.');
    callback(null);
});
transform.on('error', function (err) {
    console.error(err);
    process.exit(1);
});

transform.write('foo');
transform.write('foofoo');
transform.write('foofoofoo');
transform.write('foofoofoofoo');
transform.write('foofoofoofoofoo');
transform.end();

transform.on('data', function (r) {
    console.log(r);
});

使用パッケージ

• through2
  npm install through2でインストールします。

実行結果

C:\work\node>node through2.js
ffoooo
ffooooffoooo
ffooooffooooffoooo
ffooooffooooffooooffoooo
ffooooffooooffooooffooooffoooo
end.

関連

• Node.js入門

-Node.js