refactor: further work on node http client (#19327)

Closes https://github.com/denoland/deno/issues/18300

cli/tests/unit_node/http_test.ts

@@ -531,3 +531,68 @@ Deno.test("[node/http] ClientRequest uses HTTP/1.1", async () => {
   await def;
   assertEquals(body, "HTTP/1.1");
 });
+
+Deno.test("[node/http] ClientRequest setTimeout", async () => {
+  let body = "";
+  const def = deferred();
+  const timer = setTimeout(() => def.reject("timed out"), 50000);
+  const req = http.request("http://localhost:4545/http_version", (resp) => {
+    resp.on("data", (chunk) => {
+      body += chunk;
+    });
+
+    resp.on("end", () => {
+      def.resolve();
+    });
+  });
+  req.setTimeout(120000);
+  req.once("error", (e) => def.reject(e));
+  req.end();
+  await def;
+  clearTimeout(timer);
+  assertEquals(body, "HTTP/1.1");
+});
+
+Deno.test("[node/http] ClientRequest PATCH", async () => {
+  let body = "";
+  const def = deferred();
+  const req = http.request("http://localhost:4545/echo_server", {
+    method: "PATCH",
+  }, (resp) => {
+    resp.on("data", (chunk) => {
+      body += chunk;
+    });
+
+    resp.on("end", () => {
+      def.resolve();
+    });
+  });
+  req.write("hello ");
+  req.write("world");
+  req.once("error", (e) => def.reject(e));
+  req.end();
+  await def;
+  assertEquals(body, "hello world");
+});
+
+Deno.test("[node/http] ClientRequest PUT", async () => {
+  let body = "";
+  const def = deferred();
+  const req = http.request("http://localhost:4545/echo_server", {
+    method: "PUT",
+  }, (resp) => {
+    resp.on("data", (chunk) => {
+      body += chunk;
+    });
+
+    resp.on("end", () => {
+      def.resolve();
+    });
+  });
+  req.write("hello ");
+  req.write("world");
+  req.once("error", (e) => def.reject(e));
+  req.end();
+  await def;
+  assertEquals(body, "hello world");
+});

ext/node/polyfills/http.ts

@@ -50,6 +50,8 @@ import {
 import { getTimerDuration } from "ext:deno_node/internal/timers.mjs";
 import { serve, upgradeHttpRaw } from "ext:deno_http/00_serve.js";
 import { createHttpClient } from "ext:deno_fetch/22_http_client.js";
+import { timerId } from "ext:deno_web/03_abort_signal.js";
+import { clearTimeout as webClearTimeout } from "ext:deno_web/02_timers.js";
 
 enum STATUS_CODES {
   /** RFC 7231, 6.2.1 */
@@ -350,10 +352,7 @@ class ClientRequest extends OutgoingMessage {
     this.socketPath = options!.socketPath;
 
     if (options!.timeout !== undefined) {
-      const msecs = getTimerDuration(options.timeout, "timeout");
-      const timeout = AbortSignal.timeout(msecs);
-      timeout.onabort = () => this.emit("timeout");
-      this._timeout = timeout;
+      this.setTimeout(options.timeout);
     }
 
     const signal = options!.signal;
@@ -561,7 +560,8 @@ class ClientRequest extends OutgoingMessage {
       url,
       headers,
       client.rid,
-      this.method === "POST" || this.method === "PATCH",
+      this.method === "POST" || this.method === "PATCH" ||
+        this.method === "PUT",
     );
     this._bodyWriteRid = this._req.requestBodyRid;
   }
@@ -637,7 +637,8 @@ class ClientRequest extends OutgoingMessage {
           })(),
         ]);
         if (this._timeout) {
-          this._timeout.onabort = null;
+          this._timeout.removeEventListener("abort", this._timeoutCb);
+          webClearTimeout(this._timeout[timerId]);
         }
         this._client.close();
         const incoming = new IncomingMessageForClient(this.socket);
@@ -752,7 +753,7 @@ class ClientRequest extends OutgoingMessage {
     if (msecs === 0) {
       if (this._timeout) {
         this.removeAllListeners("timeout");
-        this._timeout.onabort = () => {};
+        this._timeout.removeEventListener("abort", this._timeoutCb);
         this._timeout = undefined;
       }
 
@@ -766,7 +767,8 @@ class ClientRequest extends OutgoingMessage {
     if (callback) this.once("timeout", callback);
 
     const timeout = AbortSignal.timeout(msecs);
-    timeout.onabort = () => this.emit("timeout");
+    this._timeoutCb = () => this.emit("timeout");
+    timeout.addEventListener("abort", this._timeoutCb);
     this._timeout = timeout;
 
     return this;

ext/web/03_abort_signal.js

@@ -203,4 +203,5 @@ export {
   newSignal,
   remove,
   signalAbort,
+  timerId,
 };

test_util/src/lib.rs

@@ -727,7 +727,10 @@ async fn main_server(
   req: Request<Body>,
 ) -> Result<Response<Body>, hyper::http::Error> {
   return match (req.method(), req.uri().path()) {
-    (&hyper::Method::POST, "/echo_server") => {
+    (
+      &hyper::Method::POST | &hyper::Method::PATCH | &hyper::Method::PUT,
+      "/echo_server",
+    ) => {
       let (parts, body) = req.into_parts();
       let mut response = Response::new(body);