1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2025-01-21 04:52:26 -05:00

Compare commits

...

2 commits

Author SHA1 Message Date
snek
6e8be9838e
Merge 74d705b35a into 9aa02769c8 2025-01-19 23:35:16 +04:00
snek
74d705b35a
feat(ext/node): support http information 2025-01-14 22:24:17 +01:00
7 changed files with 211 additions and 18 deletions

33
Cargo.lock generated
View file

@ -856,7 +856,7 @@ dependencies = [
"hickory-server",
"http 1.1.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"nix",
"once_cell",
@ -1702,7 +1702,7 @@ dependencies = [
"hickory-resolver",
"http 1.1.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-rustls",
"hyper-util",
"ipnet",
@ -1820,7 +1820,7 @@ dependencies = [
"http-body-util",
"httparse",
"hyper 0.14.28",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"itertools 0.10.5",
"memmem",
@ -2057,7 +2057,7 @@ dependencies = [
"hkdf",
"http 1.1.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"idna",
"indexmap 2.3.0",
@ -2345,7 +2345,7 @@ dependencies = [
"http 1.1.0",
"http-body-util",
"hyper 0.14.28",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"libc",
"log",
@ -2422,7 +2422,7 @@ dependencies = [
"deno_core",
"deno_error",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"log",
"once_cell",
@ -2580,7 +2580,7 @@ dependencies = [
"h2 0.4.4",
"http 1.1.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"once_cell",
"rustls-tokio-stream",
@ -3325,7 +3325,7 @@ dependencies = [
"base64 0.21.7",
"bytes",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"pin-project",
"rand",
@ -4210,9 +4210,8 @@ dependencies = [
[[package]]
name = "hyper"
version = "1.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
version = "1.5.2"
source = "git+https://github.com/hyperium/hyper.git?rev=15227a3#15227a3006ae8b402394904ed4e1b86233f1bd65"
dependencies = [
"bytes",
"futures-channel",
@ -4237,7 +4236,7 @@ checksum = "5ee4be2c948921a1a5320b629c4193916ed787a7f7f293fd3f7f5a6c9de74155"
dependencies = [
"futures-util",
"http 1.1.0",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"rustls",
"rustls-pki-types",
@ -4253,7 +4252,7 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
dependencies = [
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"pin-project-lite",
"tokio",
@ -4271,7 +4270,7 @@ dependencies = [
"futures-util",
"http 1.1.0",
"http-body 1.0.0",
"hyper 1.4.1",
"hyper 1.5.2",
"pin-project-lite",
"socket2",
"tokio",
@ -6433,7 +6432,7 @@ dependencies = [
"http 1.1.0",
"http-body 1.0.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-rustls",
"hyper-util",
"ipnet",
@ -7973,7 +7972,7 @@ dependencies = [
"h2 0.4.4",
"http 1.1.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-util",
"jsonc-parser",
"lazy-regex",
@ -8268,7 +8267,7 @@ dependencies = [
"http 1.1.0",
"http-body 1.0.0",
"http-body-util",
"hyper 1.4.1",
"hyper 1.5.2",
"hyper-timeout",
"hyper-util",
"percent-encoding",

View file

@ -350,3 +350,6 @@ opt-level = 3
opt-level = 3
[profile.release.package.zstd-sys]
opt-level = 3
[patch.crates-io]
hyper = { git = "https://github.com/hyperium/hyper.git", rev = "15227a3" }

View file

@ -373,6 +373,7 @@ deno_core::extension!(deno_node,
ops::zlib::brotli::op_brotli_decompress_stream_end,
ops::http::op_node_http_fetch_response_upgrade,
ops::http::op_node_http_request_with_conn<P>,
ops::http::op_node_http_await_information,
ops::http::op_node_http_await_response,
ops::http2::op_http2_connect,
ops::http2::op_http2_poll_client_connection,

View file

@ -11,6 +11,7 @@ use std::task::Poll;
use bytes::Bytes;
use deno_core::error::ResourceError;
use deno_core::futures::channel::mpsc;
use deno_core::futures::stream::Peekable;
use deno_core::futures::Future;
use deno_core::futures::FutureExt;
@ -70,9 +71,20 @@ pub struct NodeHttpResponse {
type CancelableResponseResult =
Result<Result<http::Response<Incoming>, hyper::Error>, Canceled>;
#[derive(Serialize, Debug)]
#[serde(rename_all = "camelCase")]
struct InformationalResponse {
status: u16,
status_text: String,
headers: Vec<(ByteString, ByteString)>,
version_major: u16,
version_minor: u16,
}
pub struct NodeHttpClientResponse {
response: Pin<Box<dyn Future<Output = CancelableResponseResult>>>,
url: String,
informational_rx: RefCell<Option<mpsc::Receiver<InformationalResponse>>>,
}
impl Debug for NodeHttpClientResponse {
@ -252,6 +264,36 @@ where
request.headers_mut().insert(CONTENT_LENGTH, len.into());
}
let (tx, informational_rx) = mpsc::channel(1);
hyper::ext::on_informational(&mut request, move |res| {
let mut tx = tx.clone();
let _ = tx.try_send(InformationalResponse {
status: res.status().as_u16(),
status_text: res.status().canonical_reason().unwrap_or("").to_string(),
headers: res
.headers()
.iter()
.map(|(k, v)| (k.as_str().into(), v.as_bytes().into()))
.collect(),
version_major: match res.version() {
hyper::Version::HTTP_09 => 0,
hyper::Version::HTTP_10 => 1,
hyper::Version::HTTP_11 => 1,
hyper::Version::HTTP_2 => 2,
hyper::Version::HTTP_3 => 3,
_ => unreachable!(),
},
version_minor: match res.version() {
hyper::Version::HTTP_09 => 9,
hyper::Version::HTTP_10 => 0,
hyper::Version::HTTP_11 => 1,
hyper::Version::HTTP_2 => 0,
hyper::Version::HTTP_3 => 0,
_ => unreachable!(),
},
});
});
let cancel_handle = CancelHandle::new_rc();
let cancel_handle_ = cancel_handle.clone();
@ -264,6 +306,7 @@ where
.add(NodeHttpClientResponse {
response: Box::pin(fut),
url: url.clone(),
informational_rx: RefCell::new(Some(informational_rx)),
});
let cancel_handle_rid = state
@ -277,6 +320,27 @@ where
})
}
#[op2(async)]
#[serde]
pub async fn op_node_http_await_information(
state: Rc<RefCell<OpState>>,
#[smi] rid: ResourceId,
) -> Option<InformationalResponse> {
let Ok(resource) = state
.borrow_mut()
.resource_table
.get::<NodeHttpClientResponse>(rid)
else {
return None;
};
let mut rx = resource.informational_rx.borrow_mut().take()?;
drop(resource);
rx.next().await
}
#[op2(async)]
#[serde]
pub async fn op_node_http_await_response(

View file

@ -5,6 +5,7 @@
import { core, primordials } from "ext:core/mod.js";
import {
op_node_http_await_information,
op_node_http_await_response,
op_node_http_fetch_response_upgrade,
op_node_http_request_with_conn,
@ -479,6 +480,44 @@ class ClientRequest extends OutgoingMessage {
this._encrypted,
);
this._flushBuffer();
const infoPromise = op_node_http_await_information(
this._req!.requestRid,
);
core.unrefOpPromise(infoPromise);
infoPromise.then((info) => {
if (!info) return;
if (info.status === 100) this.emit("continue");
let headers;
let rawHeaders;
this.emit("information", {
statusCode: info.status,
statusMessage: info.statusText,
httpVersionMajor: info.versionMajor,
httpVersionMinor: info.versionMinor,
httpVersion: `${info.versionMajor}.${info.versionMinor}`,
get headers() {
if (!headers) {
headers = {};
for (let i = 0; i < info.headers.length; i++) {
const entry = info.headers[i];
headers[entry[0]] = entry[1];
}
}
return headers;
},
get rawHeaders() {
if (!rawHeaders) {
rawHeaders = info.headers.flat();
}
return rawHeaders;
},
});
});
const res = await op_node_http_await_response(this._req!.requestRid);
if (this._req.cancelHandleRid !== null) {
core.tryClose(this._req.cancelHandleRid);
@ -1621,6 +1660,12 @@ ServerResponse.prototype.detachSocket = function (
this._socketOverride = null;
};
ServerResponse.prototype.writeContinue = function writeContinue(cb) {
if (cb) {
nextTick(cb);
}
};
Object.defineProperty(ServerResponse.prototype, "connection", {
get: deprecate(
function (this: ServerResponse) {
@ -1826,7 +1871,24 @@ export class ServerImpl extends EventEmitter {
} else {
return new Promise<Response>((resolve): void => {
const res = new ServerResponse(resolve, socket);
this.emit("request", req, res);
if (request.headers.has("expect")) {
if (/(?:^|\W)100-continue(?:$|\W)/i.test(req.headers.expect)) {
if (this.listenerCount("checkContinue") > 0) {
this.emit("checkContinue", req, res);
} else {
res.writeContinue();
this.emit("request", req, res);
}
} else if (this.listenerCount("checkExpectation") > 0) {
this.emit("checkExpectation", req, res);
} else {
res.writeHead(417);
res.end();
}
} else {
this.emit("request", req, res);
}
});
}
};

View file

@ -0,0 +1,4 @@
{
"args": "run -A main.cjs",
"output": "ok\n"
}

View file

@ -0,0 +1,60 @@
"use strict";
const assert = require("assert");
const http = require("http");
const test_req_body = "some stuff...\n";
const test_res_body = "other stuff!\n";
let sent_continue = false;
let got_continue = false;
const server = http.createServer();
server.on("checkContinue", (req, res) => {
res.writeContinue();
sent_continue = true;
req.on("data", () => {});
req.on("end", () => {
res.writeHead(200, {
"Content-Type": "text/plain",
"ABCD": "1",
});
res.end(test_res_body);
});
});
server.listen(0);
server.on("listening", () => {
const req = http.request({
port: server.address().port,
method: "POST",
path: "/world",
headers: {
"Expect": "100-continue",
"Content-Length": test_req_body.length,
},
});
let body = "";
req.on("continue", () => {
assert.ok(sent_continue);
got_continue = true;
req.end(test_req_body);
});
req.on("response", (res) => {
assert.ok(got_continue, "Full response received before 100 Continue");
assert.strictEqual(
res.statusCode,
200,
`Final status code was ${res.statusCode}, not 200.`,
);
res.setEncoding("utf8");
res.on("data", function (chunk) {
body += chunk;
});
res.on("end", () => {
assert.strictEqual(body, test_res_body);
assert.ok("abcd" in res.headers, "Response headers missing.");
console.log("ok");
server.close();
});
});
});