-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix bugs in handling multiple incoming bodies piped through a `TransF…
…ormStream` to an outgoing body (#171) This was pretty gnarly to work through, but I think came out reasonably clean. The two key pieces are: - delaying the first read from an incoming stream that's been piped to a `TransformStream` until the latter is actually read from, and hence doesn't have backpressure applied anymore. This part guarantees that when the read would happen, we have the full pipeline in place and can hand things off to the host API instead of handling anything in the JS `fetch` API implementation. - properly integrating these two pieces, and only closing the incoming body's `ReadableStream` once the host API reports that it's been dealt with to completion. Additionally, the following two changes are folded in: * Cleanup: improve some functions dealing with `Request` `Response` object handles It was pretty hard to follow when a caller would expect to have a valid handle, or when a `nullptr` would be acceptable. * Handle writing chunks to outgoing bodies fully asynchronously Before, any chunk would be written synchronously, with write operations to the outgoing body in a loop. That both meant that large chunks would block all script execution until they're done, and that things wouldn't necessarily always work. To wit, the test added here doesn't pass without this change. It streams a sentence in a very convoluted way, sending nested requests for each word, which then respond by sending the word back, letter for letter. Without this patch, that part breaks after the second letter in a word has been sent back. Signed-off-by: Till Schneidereit <[email protected]>
- Loading branch information
1 parent
3d85982
commit aa18177
Showing
8 changed files
with
255 additions
and
127 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| This sentence will be streamed in chunks. |
58 changes: 58 additions & 0 deletions
58
tests/e2e/multi-stream-forwarding/multi-stream-forwarding.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| addEventListener('fetch', async (event) => { | ||
| try { | ||
| if (!event.request.url.includes('/nested')) { | ||
| event.respondWith(main(event)); | ||
| return; | ||
| } | ||
|
|
||
| let encoder = new TextEncoder(); | ||
| let body = new TransformStream({ | ||
| start(controller) { | ||
| }, | ||
| transform(chunk, controller) { | ||
| controller.enqueue(encoder.encode(chunk)); | ||
| }, | ||
| flush(controller) { | ||
| } | ||
| }); | ||
| let writer = body.writable.getWriter(); | ||
| event.respondWith(new Response(body.readable)); | ||
| let word = new URL(event.request.url).searchParams.get('word'); | ||
| console.log(`streaming word: ${word}`); | ||
| for (let letter of word) { | ||
| console.log(`Writing letter ${letter}`); | ||
| await writer.write(letter); | ||
| } | ||
| if (word.endsWith(".")) { | ||
| await writer.write("\n"); | ||
| } | ||
| await writer.close(); | ||
| } catch (e) { | ||
| console.error(e); | ||
| } | ||
| }); | ||
| async function main(event) { | ||
| let fullBody = "This sentence will be streamed in chunks."; | ||
| let responses = []; | ||
| for (let word of fullBody.split(" ").join("+ ").split(" ")) { | ||
| responses.push((await fetch(`${event.request.url}/nested?word=${word}`)).body); | ||
| } | ||
| return new Response(concatStreams(responses)); | ||
| } | ||
|
|
||
| function concatStreams(streams) { | ||
| let { readable, writable } = new TransformStream(); | ||
| async function iter() { | ||
| for (let stream of streams) { | ||
| try { | ||
| await stream.pipeTo(writable, {preventClose: true}); | ||
| } catch (e) { | ||
| console.error(`error during pipeline execution: ${e}`); | ||
| } | ||
| } | ||
| console.log("closing writable"); | ||
| await writable.close(); | ||
| } | ||
| iter(); | ||
| return readable; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters