[JavaScript][WHATWG-Streams] Basic examples for WHATWG Streams API (on nodejs)
WHATWG Streams API](https://streams.spec.whatwg.org/)を使うための、簡単なコード例11個。 Promiseの非同期処理記述はthen()ではなく、asyncとawaitを用いるようにしている。
01-object-readable.js: ReadableStreamを作り、読み込むコード例:
ReadableStreamは、任意のオブジェクトや値を次々非同期でread()できるストリームである
ReadableStreamのコンストラクタで渡すsourceオブジェクトは、完了を表すPromiseを返すメソッド群で機能を実装する (簡単に定義したいならasyncにすればいい)
引数のcontrollerのメソッドenqueue()に渡したオブジェクトがReaderのread()メソッドから順に受け取れる (ReaderはReadableStreamのgetReader()メソッドで取り出せる)
controllerのキューしてる量が不足状態になると、pull()メソッドが呼び出される (コンストラクタの第二引数のオプションのhighWaterMarkプロパティで不足数を指定できる)
注: getReader()を呼ぶたびに別々のReaderオブジェクトが得られるが、read()できるのは最初の一つだけ (ReaderのreleaseLock()メソッドを呼べば、そのオブジェクトが開放され、次のReaderオブジェクトがread()できるようになる)
Readerのread()でかえるPromiseの値は、iteratorと同じvalueとdoneを持ったオブジェクトである
現状の仕様では、コメントアウトしたコードのように、基本forループの中でawait呼び出しするのが楽か
02-bytes-readable.js: バイト列に特化したReadableStreamを作り、読み込む例:
バイト列のストリームの場合、sourceのtypeプロパティで"bytes"を入れると、TypedArrayに特化したReadableStreamを使うことができる
バイト列のcontrollerのenqueue()メソッドにはTypedArrayのArrayBufferViewのみ渡せる(オブジェクトや数値を渡すとTypeError)
どのArrayBufferViewをenqueue()したとしても、Readerのread()で得られるPromise値はUint8Arrayである
03-writable.js: WritableStreamを作り、書き込む例
WritableStreamは、任意のオブジェクトや値を次々write()されたものを非同期に処理できるストリームである (WritableStreamには、ReadableStreamのようなバイト列特化実装はない)
WritableStreamのコンストラクタで渡すsinkオブジェクトは、完了を表すPromiseを返すメソッド群で機能を実装する
write()を呼び出せるWriterは、getWriter()で取り出せる(Reader同様にロックによって一つだけwrite()が呼び出せる)
Writerのwrite()の返すPromiseは、sinkでのwrite()が完了した時にresolveされる (awaitすれば完了待ちするし、awaitしないで連続でwrite()すると、controller内部のキューに貯めさせることになる)
04-transform.js: (未策定の)TransformStreamの作り方と使い方の例:
WHATWGのリファレンス実装にはあるが、現状の仕様内にはTransformStreamクラスは含まれていない
ただし、{readable: ReadableStream, writable: WritableStream}なオブジェクトがReadableStreamのpipeThrough()メソッドに渡す TransformStream扱いされることは仕様内にある
TransformStreamとは、writable側のwrite()で渡されたそれぞれの値を、変換したりフィルタしたりした結果を、 readable側のread()から取り出せる仕組みである(つまり、writable => readableな関係)
writableのWriterでwrite()したときに、コンストラクタに渡したオブジェクトのtransform()メソッドが呼ばれ、 そこでcontrollerのenqueue()に変換したオブジェクトを渡せば、readableのReaderのread()で得ることができるようになる
writableでclose()すると、コンストラクタに渡したオブジェクトのflush()メソッドが呼ばれる
現実装では、TransformStreamのwritableのWriterのwrite()が完了するのは、readableのReaderのread()されたときである
05-transform-pair.js: ReadableStreamとWritableStreamを使って、簡単なTransformStreamを実装する例:
readableのsourceのstart()で得たcontrollerと、wruitableのsinkのstart()で得たcontrollerを保持し、 sourceのメソッドではsinkのcontrollerを使い、逆にsinkのメソッドではsourceのcontrollerを使うことで、 TransformStreamっぽいものが作れる(この実装の通り、この２つのペアでメソッドの対応関係がある)
06-tee.js: ReadableStreamのtee()メソッドで、ReadableStreamを複製するコード例
getReader()で得るReaderは排他制御されるので、二箇所から同一内容を購読する場合は、ReadableStreamをtee()メソッドで 複製して、その２つのReadableStreamを使う必要がある
注: tee()の２つのReadableStreamで両方cancel()されたとき、tee()元がcancel()される
07-pipeto.js: ReadableStreamのpipeTo()メソッドで、WritableStreamに流すコード例
ReadableStreamのpipeTo()メソッドにWritableStreamを渡せば、非同期で、自動で読み書きが流される (つまりreadable => writableな関係)
pipeTo()の戻り値のPromiseで、読み書き完了を待つことができる
デフォルトでは、ReadableStreamのcontrollerでclose()呼び出しをすると、WritableStreamもclose()される (同様に、ReadableStreamのcontrollerのerror()呼び出しはWritableStreamのabort()を呼び出す、 WritableStreamのcontrollerのerror()呼び出しはReadableStreamのcancel()を呼び出す、つながりがある)
pipeTo()の第二引数のオプションオブジェクトで、これらのclose()/cancel()/abort()のつながりをを抑制できる (たとえば、ReadableStreamを消費しきったあと、 同じWritableStreamに追加で別のReadableStreamを加えるときは最初のpipeTo()でのclose()を抑制する必要がある)
08-pipethrough.js: ReadableStreamのpipeThrough()メソッドにTransformStreamを渡せば、変換された値が受け取れるReadbleStreamになる
pipeThrough()の戻り値は、引数に渡したTransformStreamのreadableそのものである
注: pipeTo(ts.writable)と一緒だが、戻り値に対してpipeThrough()チェーン記述できる (自動で読み込ませるには、チェーンの最後でpipeTo(ws)するとよい)
pipeThrough()の第二引数には、pipeTo()と同じオプションを渡せる
09-await-write-ready.js: WritableStreamの内部キューの消費待ちをするコード例
WritableStreamコンストラクタ台に引数のオプションのhighWaterMarkプロパティで、キューの空き待ちの目安を設定できる
注: キュー自体の保持数のリミットではなく、sinkのwrite()消費待ち状態に入る最小のキューされた数で設定する (つまり、たとえhighWaterMarkを超えても、write()できキューに積める)
キューに積まれたsinkで未消費のオブジェクトの数が、highWaterMarkを超えると、Writerのreadyで得られるPromiseが 待ち状態になり、sinkで諸費されhighWaterMark未満になったときresolveされる
注: write()の戻り値のPromiseは処理完了のための(逐次的な)待ちであり、readyは(並列的な)キューの空き待ちに使う
10-cancel.js: ReadableStreamのcacnel()の使い方
ReadableStreamのcancel()とReaderのcancel()がある。read()同様、getReader()した場合はそのReaderのcancel()だけ機能する (ReadableStreamのcancel()はロック保持してるReaderが一つもない場合だけ使える)
sourceのpull()でresolve/rejectされないPromiseを渡しても、cancel()すればread()が完了(done: true)する
11-read-with-async-iteration.js: ECMAScriptのasync iteration提案でReadableStreamを読み込むコード例
Readerのread()呼び出しをasync iteratorでラップして呼び出す
Symbol.asyncIteratorメソッドを持ったオブジェクトがfor await ofループ構文で使える仕様がasync iteration
このループでは、ofの右のSymbol.asyncIteratorメソッドが返すオブジェクトのnext()メソッドが呼び出される
next()メソッドの返り値はイテレータと同じ意味の{value, done}オブジェクトのPromiseである
async function*なAsyncGeneratorFunctionが返すオブジェクトがこの形式になっている