feat(node:http): add http information support (#27381)

Implements some client and server events to improve compat.

Fixes: https://github.com/denoland/deno/issues/27239

Cargo.lock

@@ -868,7 +868,7 @@ dependencies = [
  "hickory-server",
  "http 1.1.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "nix",
  "once_cell",
@@ -1717,7 +1717,7 @@ dependencies = [
  "hickory-resolver",
  "http 1.1.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-rustls",
  "hyper-util",
  "ipnet",
@@ -1835,7 +1835,7 @@ dependencies = [
  "http-body-util",
  "httparse",
  "hyper 0.14.28",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "itertools 0.10.5",
  "memmem",
@@ -2073,7 +2073,7 @@ dependencies = [
  "hkdf",
  "http 1.1.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "idna",
  "indexmap 2.3.0",
@@ -2367,7 +2367,7 @@ dependencies = [
  "http 1.1.0",
  "http-body-util",
  "hyper 0.14.28",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "libc",
  "log",
@@ -2445,7 +2445,7 @@ dependencies = [
  "deno_error",
  "deno_tls",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-rustls",
  "hyper-util",
  "log",
@@ -2604,7 +2604,7 @@ dependencies = [
  "h2 0.4.7",
  "http 1.1.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "once_cell",
  "rustls-tokio-stream",
@@ -3349,7 +3349,7 @@ dependencies = [
  "base64 0.21.7",
  "bytes",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "pin-project",
  "rand",
@@ -4192,9 +4192,9 @@ dependencies = [
 
 [[package]]
 name = "httparse"
-version = "1.8.0"
+version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
+checksum = "f2d708df4e7140240a16cd6ab0ab65c972d7433ab77819ea693fde9c43811e2a"
 
 [[package]]
 name = "httpdate"
@@ -4234,9 +4234,9 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "1.4.1"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
+checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80"
 dependencies = [
  "bytes",
  "futures-channel",
@@ -4261,7 +4261,7 @@ checksum = "5ee4be2c948921a1a5320b629c4193916ed787a7f7f293fd3f7f5a6c9de74155"
 dependencies = [
  "futures-util",
  "http 1.1.0",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "rustls",
  "rustls-pki-types",
@@ -4277,7 +4277,7 @@ version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
 dependencies = [
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "pin-project-lite",
  "tokio",
@@ -4295,7 +4295,7 @@ dependencies = [
  "futures-util",
  "http 1.1.0",
  "http-body 1.0.0",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "pin-project-lite",
  "socket2",
  "tokio",
@@ -6457,7 +6457,7 @@ dependencies = [
  "http 1.1.0",
  "http-body 1.0.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-rustls",
  "hyper-util",
  "ipnet",
@@ -7997,7 +7997,7 @@ dependencies = [
  "h2 0.4.7",
  "http 1.1.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-util",
  "jsonc-parser",
  "lazy-regex",
@@ -8292,7 +8292,7 @@ dependencies = [
  "http 1.1.0",
  "http-body 1.0.0",
  "http-body-util",
- "hyper 1.4.1",
+ "hyper 1.6.0",
  "hyper-timeout",
  "hyper-util",
  "percent-encoding",

Cargo.toml

@@ -150,7 +150,7 @@ http-body = "1.0"
 http-body-util = "0.1.2"
 http_v02 = { package = "http", version = "0.2.9" }
 httparse = "1.8.0"
-hyper = { version = "1.4.1", features = ["full"] }
+hyper = { version = "1.6.0", features = ["full"] }
 hyper-rustls = { version = "0.27.2", default-features = false, features = ["http1", "http2", "tls12", "ring"] }
 hyper-util = { version = "0.1.10", features = ["tokio", "client", "client-legacy", "server", "server-auto"] }
 hyper_v014 = { package = "hyper", version = "0.14.26", features = ["runtime", "http1"] }

ext/node/lib.rs

@@ -375,6 +375,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,

ext/node/ops/http.rs

@@ -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(

ext/node/polyfills/http.ts

@@ -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,
@@ -484,6 +485,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);
@@ -1626,6 +1665,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) {
@@ -1831,7 +1876,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);
+          }
         });
       }
     };

tests/specs/run/expect_100_continue/__test__.jsonc

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

tests/specs/run/expect_100_continue/main.cjs

@@ -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();
+    });
+  });
+});