diff options
Diffstat (limited to 'ext')
| -rw-r--r-- | ext/cache/01_cache.js | 2 | ||||
| -rw-r--r-- | ext/fetch/26_fetch.js | 2 | ||||
| -rw-r--r-- | ext/net/01_net.js | 39 | ||||
| -rw-r--r-- | ext/web/06_streams.js | 58 |
4 files changed, 62 insertions, 39 deletions
diff --git a/ext/cache/01_cache.js b/ext/cache/01_cache.js index fa0b68037..afde7f789 100644 --- a/ext/cache/01_cache.js +++ b/ext/cache/01_cache.js @@ -148,7 +148,7 @@ await core.shutdown(rid); break; } - await core.write(rid, value); + await core.writeAll(rid, value); } } finally { core.close(rid); diff --git a/ext/fetch/26_fetch.js b/ext/fetch/26_fetch.js index 169db2bbf..5c824898d 100644 --- a/ext/fetch/26_fetch.js +++ b/ext/fetch/26_fetch.js @@ -225,7 +225,7 @@ } try { await PromisePrototypeCatch( - core.write(requestBodyRid, value), + core.writeAll(requestBodyRid, value), (err) => { if (terminator.aborted) return; throw err; diff --git a/ext/net/01_net.js b/ext/net/01_net.js index 04360116b..d7a093ba6 100644 --- a/ext/net/01_net.js +++ b/ext/net/01_net.js @@ -4,7 +4,8 @@ ((window) => { const core = window.Deno.core; const { BadResourcePrototype, InterruptedPrototype, ops } = core; - const { WritableStream, readableStreamForRid } = window.__bootstrap.streams; + const { readableStreamForRid, writableStreamForRid } = + window.__bootstrap.streams; const { Error, ObjectPrototypeIsPrototypeOf, @@ -65,39 +66,6 @@ return core.opAsync("op_dns_resolve", { query, recordType, options }); } - function tryClose(rid) { - try { - core.close(rid); - } catch { - // Ignore errors - } - } - - function writableStreamForRid(rid) { - return new WritableStream({ - async write(chunk, controller) { - try { - let nwritten = 0; - while (nwritten < chunk.length) { - nwritten += await write( - rid, - TypedArrayPrototypeSubarray(chunk, nwritten), - ); - } - } catch (e) { - controller.error(e); - tryClose(rid); - } - }, - close() { - tryClose(rid); - }, - abort() { - tryClose(rid); - }, - }); - } - class Conn { #rid = 0; #remoteAddr = null; @@ -353,7 +321,4 @@ Datagram, resolveDns, }; - window.__bootstrap.streamUtils = { - writableStreamForRid, - }; })(this); diff --git a/ext/web/06_streams.js b/ext/web/06_streams.js index 76e31503f..09e5b7414 100644 --- a/ext/web/06_streams.js +++ b/ext/web/06_streams.js @@ -826,6 +826,62 @@ return finalBuffer; } + /** + * Create a new Writable object that is backed by a Resource that implements + * `Resource::write` / `Resource::write_all`. This object contains enough + * metadata to allow callers to bypass the JavaScript WritableStream + * implementation and write directly to the underlying resource if they so + * choose (FastStream). + * + * @param {number} rid The resource ID to write to. + * @param {boolean=} autoClose If the resource should be auto-closed when the stream closes. Defaults to true. + * @returns {ReadableStream<Uint8Array>} + */ + function writableStreamForRid(rid, autoClose = true) { + const stream = webidl.createBranded(WritableStream); + stream[_resourceBacking] = { rid, autoClose }; + + const tryClose = () => { + if (!autoClose) return; + RESOURCE_REGISTRY.unregister(stream); + core.tryClose(rid); + }; + + if (autoClose) { + RESOURCE_REGISTRY.register(stream, rid, stream); + } + + const underlyingSink = { + async write(chunk, controller) { + try { + await core.writeAll(rid, chunk); + } catch (e) { + controller.error(e); + tryClose(); + } + }, + close() { + tryClose(); + }, + abort() { + tryClose(); + }, + }; + initializeWritableStream(stream); + setUpWritableStreamDefaultControllerFromUnderlyingSink( + stream, + underlyingSink, + underlyingSink, + 1, + () => 1, + ); + return stream; + } + + function getWritableStreamResourceBacking(stream) { + return stream[_resourceBacking]; + } + /* * @param {ReadableStream} stream */ @@ -6059,6 +6115,8 @@ readableStreamForRidUnrefableUnref, readableStreamThrowIfErrored, getReadableStreamResourceBacking, + writableStreamForRid, + getWritableStreamResourceBacking, Deferred, // Exposed in global runtime scope ByteLengthQueuingStrategy, |