fetchのprogressを取るワンライナーをasyncIteratorで書く

動機

fetchでおおきめデータをダウンロードするとき、ユーザが中断できるようにしたかった。
fetch自体は最初の応答の時点で解決し、実際に"ダウンロード"を行うのはResponse.prototype.blob()である。
つまり、ダウンロードを中断するというのは、Response.prototype.blob()を中断することであり、fetchのAbortSignalでは実現できず、自前でResponse.prototype.blob()に相当する処理を実装する必要がある。
AbortSignalでいいっぽい

進捗表示用のprogress関数は既にその実装がなされているので、これに組込む形で実装することにした。

一般形

まず、よくあるprogressとその使い方が以下

const
progress=(w,f)=>new Response(new ReadableStream({start:async(
    c,
    x,
    s=[0,+w.headers.get('content-length')],
    r=w.body.getReader()
)=>{
    f(s);
    while(x=(await r.read()).value){
        c.enqueue(x);
        s[0]+=x.length;
        f(s);
    }
    c.close();
}}));

await progress(await fetch(),console.log).blob();

セミコロンがいっぱいあって美しくないので、asyncItertorで書きなおす。

asyncIterator?

まずiteratorを理解する必要があるが、これは前の記事を参照してほしい。

で、asyncIteratorは名前の通り、asyncなiterator。
具体的にはnext()がイテレータリザルトではなく、イテレータリザルトを結果に取るPromiseを返す。

お目にかかる機会はReadableStream程度であり、まずない。

配列に直す

iteratorは配列に直して使うことが多い。
普通のiteratorはArray.from(iterator)や[...iterator]でよいが、asyncIteratorは長らくfor await...ofでしか扱えなかった(自前でnextするもそう)。
ところが最近はArray.fromAsync()なるものを使えば、Array.from()と同様に簡単に配列に直せる。

これでようやくasyncIteratorを式だけで扱えるようになった。

実装

const
progress=(w,f)=>((
    r=w.body.getReader(),
    s=[0,+w.headers.get('content-length')]
)=>Array.fromAsync({
    [Symbol.asyncIterator]:_=>(
        f(s),
        {next:async x=>(
            x=(await r.read()).value,
            x&&(s[0]+=x.length,f(s)),
            {done:!x,value:x}
        )}
    )
}))();

new Blob(await progress(await fetch(),console.log));

progress().blob()の二度手間を楽に回避できて気分が良い。
Uint8Arrayで欲しい場合も

new Uint8Array((await progress(await fetch(),console.log)).flatMap(x=>[...x]))

でok。

ダウンロードの中断

ダウンロードの中断自体はリーダを.cancel()すれば良い。
ただし、そのままでは中断した時点でダウンロード済みのデータで解決してしまうので、rejectする必要がある。
rejectするにはasyncの内側でthrowすれば良いが、throwは文なので宗教上の理由で使えない。
asyncは、中でawaitしているPrmiseがrejectされると自身もrejectする。
つまり、throwの代わりにawait Promise.reject()が使える。

progressの引数として外から中断を指示できる手段を用意する必要がある。
fetchに倣ってAbortController、無難にEventTarget等が候補に挙がるが、addEventListenerはちょっと長すぎるので、実行すると中断する関数を引数に取る関数を引数にする。

progress=(w,f,c)=>((
    r=w.body.getReader(),
    s=[0,+w.headers.get('content-length')],
    b
)=>Array.fromAsync({
    [Symbol.asyncIterator]:_=>(
        c?.(_=>(b=1,r.cancel())),
        f(s),
        {next:async x=>(
            x=(await r.read()).value,
            b&&await Promise.reject('canceled'),
            x&&(s[0]+=x.length,f(s)),
            {done:!x,value:x}
        )}
    )
}))();

// 10msで中断
new Blob(await progress(await fetch(),console.log,c=>setTimeout(r,10)));

b=Promise.reject()とawait bではなくb=1とb&&await Promise.rejectにしているのは、Promise.reject()が呼ばれた時点でUncaught Rejectが出る、つまりnextは愚かprogressの外でrejectされてしまうからで、ならawaitが要らないのでは、というのはawaitがないとrejectされずにただのエラーになってしまうから。

ややこしい。

追記
やっぱりキャンセルの扱いが面倒な気がしたのでAbortSignal版も作った。

progress=(w,f,s)=>((
	r=w.body.getReader(),
    p=[0,+w.headers.get('content-length')],
    b,
    c=_=>(b=1,r.cancel()),
    d=_=>Promise.reject({message:'Load aborted.'})
)=>s?.aborted?
    d():
    Array.fromAsync({[Symbol.asyncIterator]:_=>(
        s?.addEventListener('abort',c,{once:1}),
        f(p),
        {next:async x=>(
        	x=(await r.read()).value,
            b&&await d(),
            x?(p[0]+=x.length,f(p)):s?.removeEventListener('abort',c),
            {done:!x,value:x}
        )}
    )})
)();