summaryrefslogtreecommitdiff
path: root/ext
diff options
context:
space:
mode:
Diffstat (limited to 'ext')
-rw-r--r--ext/http/01_http.js63
-rw-r--r--ext/http/lib.rs51
-rw-r--r--ext/net/01_net.js30
-rw-r--r--ext/web/06_streams.js42
4 files changed, 133 insertions, 53 deletions
diff --git a/ext/http/01_http.js b/ext/http/01_http.js
index 217bfc061..ff4b6f41f 100644
--- a/ext/http/01_http.js
+++ b/ext/http/01_http.js
@@ -32,7 +32,8 @@
} = window.__bootstrap.webSocket;
const { TcpConn, UnixConn } = window.__bootstrap.net;
const { TlsConn } = window.__bootstrap.tls;
- const { Deferred } = window.__bootstrap.streams;
+ const { Deferred, getReadableStreamRid, readableStreamClose } =
+ window.__bootstrap.streams;
const {
ArrayPrototypeIncludes,
ArrayPrototypePush,
@@ -235,7 +236,6 @@
typeof respBody === "string" ||
ObjectPrototypeIsPrototypeOf(Uint8ArrayPrototype, respBody)
);
-
try {
await core.opAsync(
"op_http_write_headers",
@@ -269,35 +269,50 @@
) {
throw new TypeError("Unreachable");
}
- const reader = respBody.getReader();
- while (true) {
- const { value, done } = await reader.read();
- if (done) break;
- if (!ObjectPrototypeIsPrototypeOf(Uint8ArrayPrototype, value)) {
- await reader.cancel(new TypeError("Value not a Uint8Array"));
- break;
+ const resourceRid = getReadableStreamRid(respBody);
+ if (resourceRid) {
+ if (respBody.locked) {
+ throw new TypeError("ReadableStream is locked.");
}
+ const _reader = respBody.getReader(); // Aquire JS lock.
+ await core.opAsync(
+ "op_http_write_resource",
+ streamRid,
+ resourceRid,
+ );
+ readableStreamClose(respBody); // Release JS lock.
+ } else {
+ const reader = respBody.getReader();
+ while (true) {
+ const { value, done } = await reader.read();
+ if (done) break;
+ if (!ObjectPrototypeIsPrototypeOf(Uint8ArrayPrototype, value)) {
+ await reader.cancel(new TypeError("Value not a Uint8Array"));
+ break;
+ }
+ try {
+ await core.opAsync("op_http_write", streamRid, value);
+ } catch (error) {
+ const connError = httpConn[connErrorSymbol];
+ if (
+ ObjectPrototypeIsPrototypeOf(BadResourcePrototype, error) &&
+ connError != null
+ ) {
+ // deno-lint-ignore no-ex-assign
+ error = new connError.constructor(connError.message);
+ }
+ await reader.cancel(error);
+ throw error;
+ }
+ }
+
try {
- await core.opAsync("op_http_write", streamRid, value);
+ await core.opAsync("op_http_shutdown", streamRid);
} catch (error) {
- const connError = httpConn[connErrorSymbol];
- if (
- ObjectPrototypeIsPrototypeOf(BadResourcePrototype, error) &&
- connError != null
- ) {
- // deno-lint-ignore no-ex-assign
- error = new connError.constructor(connError.message);
- }
await reader.cancel(error);
throw error;
}
}
- try {
- await core.opAsync("op_http_shutdown", streamRid);
- } catch (error) {
- await reader.cancel(error);
- throw error;
- }
}
const deferred = request[_deferred];
diff --git a/ext/http/lib.rs b/ext/http/lib.rs
index 9c0109937..dff5c14cb 100644
--- a/ext/http/lib.rs
+++ b/ext/http/lib.rs
@@ -75,6 +75,7 @@ pub fn init() -> Extension {
op_http_read::decl(),
op_http_write_headers::decl(),
op_http_write::decl(),
+ op_http_write_resource::decl(),
op_http_shutdown::decl(),
op_http_websocket_accept_header::decl(),
op_http_upgrade_websocket::decl(),
@@ -665,6 +666,56 @@ async fn op_http_write_headers(
}
#[op]
+async fn op_http_write_resource(
+ state: Rc<RefCell<OpState>>,
+ rid: ResourceId,
+ stream: ResourceId,
+) -> Result<(), AnyError> {
+ let http_stream = state
+ .borrow()
+ .resource_table
+ .get::<HttpStreamResource>(rid)?;
+ let mut wr = RcRef::map(&http_stream, |r| &r.wr).borrow_mut().await;
+ let resource = state.borrow().resource_table.get_any(stream)?;
+ loop {
+ let body_tx = match &mut *wr {
+ HttpResponseWriter::Body(body_tx) => body_tx,
+ HttpResponseWriter::Headers(_) => {
+ return Err(http_error("no response headers"))
+ }
+ HttpResponseWriter::Closed => {
+ return Err(http_error("response already completed"))
+ }
+ };
+
+ let mut vec = vec![0u8; 64 * 1024];
+ let vec_ptr = vec.as_mut_ptr();
+ let buf = ZeroCopyBuf::new_temp(vec);
+ let nread = resource.clone().read(buf).await?;
+ if nread == 0 {
+ break;
+ }
+ // SAFETY: ZeroCopyBuf keeps the Vec<u8> alive.
+ let bytes =
+ Bytes::from_static(unsafe { std::slice::from_raw_parts(vec_ptr, nread) });
+ match body_tx.send_data(bytes).await {
+ Ok(_) => {}
+ Err(err) => {
+ // Don't return "channel closed", that's an implementation detail.
+ // Pull up the failure associated with the transport connection instead.
+ assert!(err.is_closed());
+ http_stream.conn.closed().await?;
+ // If there was no connection error, drop body_tx.
+ *wr = HttpResponseWriter::Closed;
+ }
+ }
+ }
+
+ take(&mut *wr);
+ Ok(())
+}
+
+#[op]
async fn op_http_write(
state: Rc<RefCell<OpState>>,
rid: ResourceId,
diff --git a/ext/net/01_net.js b/ext/net/01_net.js
index 48cbfaaab..fde75fe56 100644
--- a/ext/net/01_net.js
+++ b/ext/net/01_net.js
@@ -4,7 +4,7 @@
((window) => {
const core = window.Deno.core;
const { BadResourcePrototype, InterruptedPrototype } = core;
- const { ReadableStream, WritableStream } = window.__bootstrap.streams;
+ const { WritableStream, readableStreamForRid } = window.__bootstrap.streams;
const {
Error,
ObjectPrototypeIsPrototypeOf,
@@ -65,8 +65,6 @@
return core.opAsync("op_dns_resolve", { query, recordType, options });
}
- const DEFAULT_CHUNK_SIZE = 64 * 1024;
-
function tryClose(rid) {
try {
core.close(rid);
@@ -75,32 +73,6 @@
}
}
- function readableStreamForRid(rid) {
- return new ReadableStream({
- type: "bytes",
- async pull(controller) {
- const v = controller.byobRequest.view;
- try {
- const bytesRead = await read(rid, v);
- if (bytesRead === null) {
- tryClose(rid);
- controller.close();
- controller.byobRequest.respond(0);
- } else {
- controller.byobRequest.respond(bytesRead);
- }
- } catch (e) {
- controller.error(e);
- tryClose(rid);
- }
- },
- cancel() {
- tryClose(rid);
- },
- autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
- });
- }
-
function writableStreamForRid(rid) {
return new WritableStream({
async write(chunk, controller) {
diff --git a/ext/web/06_streams.js b/ext/web/06_streams.js
index 6daea0898..492694563 100644
--- a/ext/web/06_streams.js
+++ b/ext/web/06_streams.js
@@ -8,6 +8,7 @@
"use strict";
((window) => {
+ const core = window.Deno.core;
const webidl = window.__bootstrap.webidl;
const { add, remove, signalAbort, newSignal, AbortSignalPrototype } =
window.__bootstrap.abortSignal;
@@ -640,6 +641,41 @@
return stream[_disturbed];
}
+ const DEFAULT_CHUNK_SIZE = 64 * 1024; // 64 KiB
+
+ function readableStreamForRid(rid) {
+ const stream = new ReadableStream({
+ type: "bytes",
+ async pull(controller) {
+ const v = controller.byobRequest.view;
+ try {
+ const bytesRead = await core.read(rid, v);
+ if (bytesRead === 0) {
+ core.tryClose(rid);
+ controller.close();
+ controller.byobRequest.respond(0);
+ } else {
+ controller.byobRequest.respond(bytesRead);
+ }
+ } catch (e) {
+ controller.error(e);
+ core.tryClose(rid);
+ }
+ },
+ cancel() {
+ core.tryClose(rid);
+ },
+ autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
+ });
+
+ stream[_maybeRid] = rid;
+ return stream;
+ }
+
+ function getReadableStreamRid(stream) {
+ return stream[_maybeRid];
+ }
+
/**
* @param {unknown} value
* @returns {value is WritableStream}
@@ -4288,6 +4324,7 @@
WeakMapPrototypeSet(countSizeFunctionWeakMap, globalObject, size);
}
+ const _maybeRid = Symbol("[[maybeRid]]");
/** @template R */
class ReadableStream {
/** @type {ReadableStreamDefaultController | ReadableByteStreamController} */
@@ -4302,6 +4339,8 @@
[_state];
/** @type {any} */
[_storedError];
+ /** @type {number | null} */
+ [_maybeRid] = null;
/**
* @param {UnderlyingSource<R>=} underlyingSource
@@ -5840,6 +5879,9 @@
errorReadableStream,
createProxy,
writableStreamClose,
+ readableStreamClose,
+ readableStreamForRid,
+ getReadableStreamRid,
Deferred,
// Exposed in global runtime scope
ByteLengthQueuingStrategy,
generated by cgit v1.2.3 (git 2.25.1) at 2025-08-04 13:35:26 +0000