#whatwgstreams

[bellbind]

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が返すオブジェクトがこの形式になっている