Node.js Stream を使いこなす

# Node.js Stream を使いこなす
Node.js には [Stream](https://nodejs.org/api/stream.html) というオブジェクトがあります。名前の通り、データストリームを扱うもので、Java や .NET にも同じようなクラスがあります。

`Stream` オブジェクトはデータをストリームとして扱いたいということに重宝します。


## Stream の便利さ
例えば、ファイルの内容を別のファイルにコピーする場合、以下のようにファイルを全て読み込んでから全て書き込む方法があります。後述するように問題があるのですが、コードが非常に読みやすいというメリットがあります。

```javascript
const text = fs.readFileSync('src.txt', 'utf8');
fs.writeFileSync('dest.txt', text);
```

この方法はファイルを読み込んだり書き込んだりしている間は処理がブロッキングされてしまいます。特にウェブアプリの場合、ブロッキングされている間は他のブラウザーからの応答をさばくことができません。

非ブロッキング I/O を使ってこのように書き換えます。

```javascript
fs.readFile('src.txt', 'utf8', (err, data) => {
  fs.writeFile('dest.txt', data);
});
```

一気に読みにくくなりましたね! しかも、全てのデータを読み込んでから全てのデータを書き込むことには変わりはありませんので、2GB のテキストファイルの場合、メモリを 2GB 消費します。

そこでストリームの出番です。

```javascript
const src = fs.createReadStream('src.txt', 'utf8');
const dest = fs.createWriteStream('dest.txt', 'utf8');
src.on('data', chunk => dest.write(chunk));
src.on('end', () => dest.end());
```

この例ではを一定量だけ読み込んでイベントを発生させています。大きなファイルでも全てを読み込むわけではないのでメモリに非常に優しいです。

ただ、ちょっと見にくいですね。`pipe()` を使うと一気に読みやすくなります。

```js
const src = fs.createReadStream('src.txt', 'utf8');
const dest = fs.createWriteStream('dest.txt', 'utf8');
src.pipe(dest);
```

`pipe()` メソッドは読み取り可能なストリームと書き込み可能なストリームを関連付けるもので、後はよしなにデータを渡してくれます。

Node.js では、ここで例にあげたファイルだけでなく、いくつかのオブジェクトはこの Stream となっています。`process.stdout` もそうです。ファイルの内容を標準出力に出力するのもとても簡単です。

```js
const src = fs.createReadStream('src.txt', 'utf8');
src.pipe(process.stdout);
```

`Stream` を使うとデータを流れるように扱えるのが分かります。

## Stream の種類
一口に `Stream` と言っても四種類あります。

* [`stream.Readable`](https://nodejs.org/api/stream.html#stream_readable_streams) - 読み取りだけができるストリーム
* [`stream.Writable`](https://nodejs.org/api/stream.html#stream_writable_streams) - 書き込みだけができるストリーム
* [`stream.Duplex`](https://nodejs.org/api/stream.html#stream_class_stream_duplex) - 読み取りも書き込みもできるストリーム
* [`stream.Transform`](https://nodejs.org/api/stream.html#stream_class_stream_transform) - 読み取ったデータを変換して出力するストリーム


### stream.Readable
読み取りだけができるストリームです。[`data` イベント](https://nodejs.org/api/stream.html#stream_event_data)を利用してデータを読み込みます。

[`fs.createReadStream()`](https://nodejs.org/api/fs.html#fs_fs_createreadstream_path_options) や [`child.stdout`](https://nodejs.org/api/child_process.html#child_process_child_stdout) がこれにあたります。

```js
const src = fs.createReadStream('src.txt' 'utf8');
src.on('data', chunk => process.stdout.write(chunk));
```


### stream.Writable
書き込みだけができるストリームです。[`write()` メソッド](https://nodejs.org/api/stream.html#stream_writable_write_chunk_encoding_callback) を使ってデータを書き込みます。

[`fs.createWriteStream()`](https://nodejs.org/api/fs.html#fs_fs_createwritestream_path_options) や [`child.stdin`](https://nodejs.org/api/child_process.html#child_process_child_stdin) がこれにあたります。

```js
const dest = fs.createWriteStream('dest.txt');
dest.write('Hello!');
```


### stream.Duplex
読み取りと書き込みの両方ができるストリームです。読み取るときは `stream.Readable` と同じで、書き込むときは `stream.Writable` と同じです。

[TCP Socket](https://nodejs.org/api/net.html) がこれにあたります。

```js
const client = net.connect(80, 'www.yahoo.co.jp');
client.pipe(process.stdout);
client.once('connect', () => client.write('GET / HTTP/1.0\r\n\r\n'));
```


### stream.Transform
変換をするためのストリームです。書き込まれたデータを変換して出力します。

[`Crypt`](https://nodejs.org/api/crypto.html) や ['zlib'](https://nodejs.org/api/zlib.html) がこれにあたります。

```js
const gzip = zlib.createGzip();
const src = fs.createReadStream('src.txt');
const dest = fs.createWriteStream('src.txt.gz');
src.pipe(gzip).pipe(dest);
```

サンプルコードではひとつしか使ってませんが、変換をつなげて書くこともできます。

```
src
  .pipe(transformA)
  .pipe(transformB)
  .pipe(transformC)
  .pipe(dest);
```


### ストリームの種類の判定
読み取り可能なストリームが全て `stream.Readable` を継承しているというわけではないようです。`instanceof` で読み取り可能なストリームかどうかを判定するのは難しいです。また、`stream.Duplex` でも `process.stdout` のように読み取りができないものもあります。

自分は `readable` もしくは `writable` プロパティの値が `true` かどうかで判断していますが、ドキュメントにその記述が見つからないので使っていいか微妙です。


## さまざまな Stream
Stream をより便利に使うためにいろいろと紹介します。


### fs.createReadStream/createWriteStream
ファイルの入出力を扱う `fs` ライブラリにはストリームでファイルをオープンするメソッドが用意されています。

* [fs.createReadStream(path[, options])](https://nodejs.org/api/fs.html#fs_fs_createreadstream_path_options)
* [fs.createWriteStream(path[, options])](https://nodejs.org/api/fs.html#fs_fs_createwritestream_path_options)

細かい説明は不要ですよね!


### TCP Socket
TCP Socket を扱う [`net`](https://nodejs.org/api/net.html) モジュールもストリームが使えます。入出力の両方に対応しています。

以下のプログラムは Yahoo のサーバーに接続して、トップページの HTML を標準出力に出力しています。

```js
const net = require('net');

const client = net.connect(80, 'www.yahoo.co.jp');
client.pipe(process.stdout);
client.once('connect', () => client.write('GET / HTTP/1.0\r\n\r\n'));
```

* [net.Server Event: `connection`](https://nodejs.org/api/net.html#net_event_connection)
* [net.connect()](https://nodejs.org/api/net.html#net_net_connect_options_connectlistener)

[http](https://nodejs.org/api/http.html) や [https](https://nodejs.org/api/https.html) モジュールもストリームが使えます。


### child_process の stdin/stdout/stderr
`child_process` は外部プロセスを扱うことができます。その標準入出力をストリームとして扱うことができます。

* [child.stdin](https://nodejs.org/api/child_process.html#child_process_child_stdin) - 外部プロセスの標準入力へ書き込むストリーム
* [child.stdout](https://nodejs.org/api/child_process.html#child_process_child_stdout) - 外部プロセスの標準出力から読み取るストリーム
* [child.stderr](https://nodejs.org/api/child_process.html#child_process_child_stderr) - 外部プロセスの標準エラー出力から読み取るストリーム

サンプルコードでは `ls` コマンドの結果をファイルに記録しています。

```js
const ls = exec('ls');
const log = fs.createWriteStream('ls.txt', 'utf8');
ls.stdout.pipe(log);
```


### process の stdin/stdout/stderr
[process](https://nodejs.org/api/process.html#process_process) は自身のプロセスを表す Node.js のグローバルオブジェクトです。こちらの標準入出力もストリームとして扱うことができます。

* [process.stdin](https://nodejs.org/api/process.html#process_process_stdin) - 標準入力から読み取るストリーム
* [process.stdout](https://nodejs.org/api/process.html#process_process_stdout) - 標準出力へ書き込むストリーム
* [process.stderr](https://nodejs.org/api/process.html#process_process_stderr) - 標準エラー出力へ書き込むストリーム

`ls` コマンドを実行した結果をファイルではなく標準出力に出力します。

```js
const ls = exec('ls');
ls.stdout.pipe(process.stdout);
```

`child_process` と違い、書き込みと読み取りが反対なことに注意してください。


### Crypt/Zlib
暗号化を行う [Crypto](https://nodejs.org/api/crypto.html#crypto_crypto) や圧縮を行う [Zlib](https://nodejs.org/api/zlib.html#zlib_zlib) は変換用の `stream.Transfrom` として利用できます。

```js
const gzip = zlib.createGzip();
const src = fs.createReadStream('src.txt');
const dest = fs.createWriteStream('src.txt.gz');
src.pipe(gzip).pipe(dest);
```

UNIX でパイプをよく使っている人は解ると思いますが、変換用のストリームはとても強力で、他の書き方に比べると非常に読みやすいです。

```js
fs.createReadStream('src.txt')
  .pipe(zlib.createGzip())
  .pipe(crypto.createCipher('aes192', 'pass'))
  .pipe(fs.createWriteStream('src.txt.gz.aes'));
```


### memory-streams
[memory-streams](https://www.npmjs.com/package/memory-streams) はメモリをストリームとして扱います。Node.js の標準モジュールには、ストリームの内容をすべてメモリ上に読み取るようなオブジェクトが用意されておりません。

ファイルに `Hello!` と書き込むサンプルコードです。

```js
const streams = require('memory-streams');

const src = new streams.ReadableStream('Hello!');
const dest = fs.createWriteStream('hello.txt', 'utf8');
src.pipe(dest);
```

もちろん、ストリームから読み取ることもできます。

```js
const streams = require('memory-streams');

const src = fs.createReadStream('hello.txt', 'utf8');
const dest = new streams.WritableStream();
src.pipe(dest);

// console.log(dest.toString()); <- ここに書いても何も出力されない
src.on('end', () => console.log(dest.toString()));
```

処理は非同期で行われるため、`src.pipe(dest)` の次の行ではまだストリーム処理が開始されていないことに注意してください。


### duplex-child-process
標準モジュールの `child_process` に非常によく似ています。(残念ながら `spawn` しか使えませんが) [duplex-child-process]` は戻り値をそのままストリームとして扱うことができます。

`child_process` を使った場合はこのように書いていました。

```js
const ls = spawn('ls');
const log = fs.createWriteStream('ls.txt', 'utf8');
ls.stdout.pipe(log);
```

`duplex-child-process` ではこう書くことができます。

```js
const spawn = require('duplex-child-process');

const ls = spawn('ls');
const log = fs.createWriteStream('ls.txt', 'utf8');
// ls.stdout.pipe(log);
ls.pipe(log); // <- stdout が要らない
```

`stdin` と `stdout` を結合したストリームとして扱えるため、場合によっては非常に便利です。ですが、残念ながら、`stdout` と `stderr` を結合したりはしてくれないようです。


### pty.js
[pty.js](https://www.npmjs.com/package/pty.js/) は表示を持たない仮想ターミナルです。詳しい使い方はドキュメントを読んでもらうとしますが、このターミナルはそのまま読み書き可能なストリームとして扱うことができます。

以下のサンプルコードでは、`bash` を起動し、`ls` コマンドと `exit` コマンドを実行しています。

```js
const pty = require('pty.js');
const streams = require('memory-streams');

const term = pty('bash');
const ls = new streams.ReadableStream('ls\nexit\n');
term.pipe(process.stdout);
ls.pipe(term);
```

### wsstream
Node.js では比較的簡単に WebSocket を扱うことができます。[ws](https://www.npmjs.com/package/ws) あたりが有名だと思います。残念ながら `ws` はストリームではないようです。

[wsstream](https://www.npmjs.com/package/wsstream) は `ws` を読み書き可能なストリームに変換します。

サンプルコードではクライアントからの WebSocket で送信されたデータを標準出力に出力しています。

```js
const WsStream = require('wsstream');

// Express.js の初期化
// ...

app.ws('/', (ws, req) => {
  const stream = WsStream(ws);
  stream.pipe(process.stdout);
});
```


## おまけ
[pty.js](https://www.npmjs.com/package/pty.js) とブラウザー向けの仮想ターミナル [xterm.js](https://xtermjs.org/) を組み合わせると、簡単にブラウザーで動くターミナルが作れます。

サーバー側はこんな感じ。

```js
const pty.js = require('pty.js');
const WsStream = require('wsstream');

// Express.js の初期化
// ...

app.ws('/', (ws, req) => {
  const stream = WsStream(ws);
  const bash = pty.spawn('bash');

  bash.pipe(stream);
  stream.pipe(bash);
});
```

クライアント側はこんな感じ。

```js
const term = new Terminal();
term.open(document.querySelector('#terminal'));
term.attach(new WebSocket('http://.../'));
```

ストリームって素晴らしい!