summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--cli/tests/unit/http_test.ts39
-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
-rw-r--r--runtime/js/40_files.js4
-rw-r--r--serde_v8/magic/buffer.rs12
7 files changed, 186 insertions, 55 deletions
diff --git a/cli/tests/unit/http_test.ts b/cli/tests/unit/http_test.ts
index 37c827b9b..f48f314db 100644
--- a/cli/tests/unit/http_test.ts
+++ b/cli/tests/unit/http_test.ts
@@ -854,6 +854,45 @@ Deno.test({ permissions: { net: true } }, async function httpServerPanic() {
listener.close();
});
+Deno.test(
+ { permissions: { net: true, write: true, read: true } },
+ async function httpServerClosedStream() {
+ const listener = Deno.listen({ port: 4502 });
+
+ const client = await Deno.connect({ port: 4502 });
+ await client.write(new TextEncoder().encode(
+ `GET / HTTP/1.0\r\n\r\n`,
+ ));
+
+ const conn = await listener.accept();
+ const httpConn = Deno.serveHttp(conn);
+ const ev = await httpConn.nextRequest();
+ const { respondWith } = ev!;
+
+ const tmpFile = await Deno.makeTempFile();
+ const file = await Deno.open(tmpFile, { write: true, read: true });
+ await file.write(new TextEncoder().encode("hello"));
+
+ const reader = await file.readable.getReader();
+ while (true) {
+ const { done, value } = await reader.read();
+ if (done) break;
+ assert(value);
+ }
+
+ try {
+ await respondWith(new Response(file.readable));
+ fail("The stream should've been locked");
+ } catch {
+ // pass
+ }
+
+ httpConn.close();
+ listener.close();
+ client.close();
+ },
+);
+
// https://github.com/denoland/deno/issues/11595
Deno.test(
{ permissions: { net: true } },
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,
diff --git a/runtime/js/40_files.js b/runtime/js/40_files.js
index 8aa0a4972..d2148be2f 100644
--- a/runtime/js/40_files.js
+++ b/runtime/js/40_files.js
@@ -6,8 +6,8 @@
const { read, readSync, write, writeSync } = window.__bootstrap.io;
const { ftruncate, ftruncateSync, fstat, fstatSync } = window.__bootstrap.fs;
const { pathFromURL } = window.__bootstrap.util;
- const { readableStreamForRid, writableStreamForRid } =
- window.__bootstrap.streamUtils;
+ const { writableStreamForRid } = window.__bootstrap.streamUtils;
+ const { readableStreamForRid } = window.__bootstrap.streams;
const {
ArrayPrototypeFilter,
Error,
diff --git a/serde_v8/magic/buffer.rs b/serde_v8/magic/buffer.rs
index 484984ac5..a0a1c974b 100644
--- a/serde_v8/magic/buffer.rs
+++ b/serde_v8/magic/buffer.rs
@@ -14,19 +14,28 @@ use crate::magic::transl8::impl_magic;
pub enum MagicBuffer {
FromV8(ZeroCopyBuf),
ToV8(Mutex<Option<Box<[u8]>>>),
+ // Variant of the MagicBuffer than is never exposed to the JS.
+ // Generally used to pass Vec<u8> backed buffers to resource methods.
+ Temp(Vec<u8>),
}
+
impl_magic!(MagicBuffer);
impl MagicBuffer {
pub fn empty() -> Self {
MagicBuffer::ToV8(Mutex::new(Some(vec![0_u8; 0].into_boxed_slice())))
}
+
+ pub fn new_temp(vec: Vec<u8>) -> Self {
+ MagicBuffer::Temp(vec)
+ }
}
impl Clone for MagicBuffer {
fn clone(&self) -> Self {
match self {
Self::FromV8(zbuf) => Self::FromV8(zbuf.clone()),
+ Self::Temp(vec) => Self::Temp(vec.clone()),
Self::ToV8(_) => panic!("Don't Clone a MagicBuffer sent to v8"),
}
}
@@ -49,6 +58,7 @@ impl Deref for MagicBuffer {
fn deref(&self) -> &[u8] {
match self {
Self::FromV8(buf) => &*buf,
+ Self::Temp(vec) => &*vec,
Self::ToV8(_) => panic!("Don't Deref a MagicBuffer sent to v8"),
}
}
@@ -58,6 +68,7 @@ impl DerefMut for MagicBuffer {
fn deref_mut(&mut self) -> &mut [u8] {
match self {
Self::FromV8(buf) => &mut *buf,
+ Self::Temp(vec) => &mut *vec,
Self::ToV8(_) => panic!("Don't Deref a MagicBuffer sent to v8"),
}
}
@@ -85,6 +96,7 @@ impl ToV8 for MagicBuffer {
let value: &[u8] = buf;
value.into()
}
+ Self::Temp(_) => unreachable!(),
Self::ToV8(x) => x.lock().unwrap().take().expect("MagicBuffer was empty"),
};
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-27 08:50:13 +0000