feat(unstable): refactor and expose http upgrades

cli/tsc/dts/lib.deno.ns.d.ts

@@ -6336,6 +6336,12 @@ declare namespace Deno {
      * @default {false}
      */
     allowHost?: boolean;
+    /** Whether HTTP upgrades are allowed or not.
+     *
+     * @default {false}
+     * @experimental **UNSTABLE**: New API, yet to be vetted.
+     */
+    allowUpgrades?: boolean;
     /** Sets the local address where the socket will connect from. */
     localAddress?: string;
   }

cli/tsc/dts/lib.deno.unstable.d.ts

@@ -367,6 +367,59 @@ declare namespace Deno {
     options: UnixListenOptions & { transport: "unixpacket" },
   ): DatagramConn;
 
+  /** **UNSTABLE**: New API, yet to be vetted.
+   * A connection from an upgraded HTTP request or response.
+   * @category Network
+   * @experimental
+   */
+  export interface UpgradedConn extends Conn {}
+
+  /** **UNSTABLE**: New API, yet to be vetted.
+   * Upgrade a Deno.serve Request into an UpgradedConn.
+   *
+   * ```ts
+   * Deno.serve((req) => {
+   *   const { response, conn } = Deno.upgradeRequest(req, "some-protocol", {
+   *     headers: {
+   *       "x-protocol-header": "meow",
+   *     },
+   *   });
+   *
+   *   return response;
+   * });
+   * ```
+   *
+   * @tags allow-net
+   * @category Network
+   * @experimental
+   */
+  export function upgradeRequest(
+    obj: Request,
+    protocol: string,
+    responseInit?: Omit<ResponseInit, "body" | "status" | "statusText">,
+  ): { response: Response; conn: Promise<UpgradedConn> };
+
+  /** **UNSTABLE**: New API, yet to be vetted.
+   * Upgrade a fetch Response into an UpgradedConn.
+   *
+   * ```ts
+   * const client = Deno.createHttpClient({ allowUpgrades: true });
+   *
+   * const res = await fetch('http://localhost:8000', {
+   *   client,
+   *   headers: { Upgrade: "some-protocol" },
+   * });
+   * const conn = await Deno.upgradeResponse(res);
+   * ```
+   *
+   * @tags allow-net
+   * @category Network
+   * @experimental
+   */
+  export function upgradeResponse(
+    obj: Response,
+  ): Promise<UpgradedConn>;
+
   /** **UNSTABLE**: New API, yet to be vetted.
    *
    * Open a new {@linkcode Deno.Kv} connection to persist data.

ext/fetch/22_body.js

@@ -80,10 +80,10 @@ function chunkToString(chunk) {
 
 class InnerBody {
   /**
-   * @param {ReadableStream<Uint8Array> | { body: Uint8Array | string, consumed: boolean }} stream
+   * @param {ReadableStream<Uint8Array> | { body: Uint8Array | number | string, consumed: boolean }} stream
    */
   constructor(stream) {
-    /** @type {ReadableStream<Uint8Array> | { body: Uint8Array | string, consumed: boolean }} */
+    /** @type {ReadableStream<Uint8Array> | { body: Uint8Array | number | string, consumed: boolean }} */
     this.streamOrStatic = stream ??
       { body: new Uint8Array(), consumed: false };
     /** @type {null | Uint8Array | string | Blob | FormData} */
@@ -100,6 +100,9 @@ class InnerBody {
       )
     ) {
       const { body, consumed } = this.streamOrStatic;
+      if (typeof body === "number") {
+        return null;
+      }
       if (consumed) {
         this.streamOrStatic = new ReadableStream();
         this.streamOrStatic.getReader();
@@ -131,6 +134,7 @@ class InnerBody {
       return this.streamOrStatic.locked ||
         isReadableStreamDisturbed(this.streamOrStatic);
     }
+    if (typeof this.streamOrStatic.body === "number") return true;
     return this.streamOrStatic.consumed;
   }
 

ext/fetch/22_http_client.js

@@ -98,22 +98,29 @@ function createHttpClient(options) {
       options,
       keyPair,
     ),
+    options,
   );
 }
 
 class HttpClient {
   #rid;
+  #allowUpgrades;
 
   /**
    * @param {number} rid
    */
-  constructor(rid) {
+  constructor(rid, options) {
     ObjectDefineProperty(this, internalRidSymbol, {
       __proto__: null,
       enumerable: false,
       value: rid,
     });
     this.#rid = rid;
+    this.#allowUpgrades = options.allowUpgrades ?? false;
+  }
+
+  get allowUpgrades() {
+    return this.#allowUpgrades;
   }
 
   close() {

ext/fetch/23_request.js

@@ -9,7 +9,7 @@
 /// <reference path="../../cli/tsc/dts/lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 
-import { core, internals, primordials } from "ext:core/mod.js";
+import { internals, primordials } from "ext:core/mod.js";
 const {
   ArrayPrototypeMap,
   ArrayPrototypeSlice,
@@ -46,7 +46,6 @@ import {
   signalAbort,
 } from "ext:deno_web/03_abort_signal.js";
 import { DOMException } from "ext:deno_web/01_dom_exception.js";
-const { internalRidSymbol } = core;
 
 const _request = Symbol("request");
 const _headers = Symbol("headers");
@@ -84,7 +83,7 @@ function processUrlList(urlList, urlListProcessed) {
  * @property {number} redirectCount
  * @property {(() => string)[]} urlList
  * @property {string[]} urlListProcessed
- * @property {number | null} clientRid NOTE: non standard extension for `Deno.HttpClient`.
+ * @property {HttpClient | null} client NOTE: non standard extension for `Deno.HttpClient`.
  * @property {Blob | null} blobUrlEntry
  */
 
@@ -132,7 +131,7 @@ function newInnerRequest(method, url, headerList, body, maybeBlob) {
     redirectCount: 0,
     urlList: [typeof url === "string" ? () => url : url],
     urlListProcessed: [],
-    clientRid: null,
+    client: null,
     blobUrlEntry,
     url() {
       if (this.urlListProcessed[0] === undefined) {
@@ -183,7 +182,7 @@ function cloneInnerRequest(request, skipBody = false) {
     redirectCount: request.redirectCount,
     urlList: [() => request.url()],
     urlListProcessed: [request.url()],
-    clientRid: request.clientRid,
+    client: request.client,
     blobUrlEntry: request.blobUrlEntry,
     url() {
       if (this.urlListProcessed[0] === undefined) {
@@ -393,7 +392,7 @@ class Request {
           "Argument 2",
         );
       }
-      request.clientRid = init.client?.[internalRidSymbol] ?? null;
+      request.client = init.client;
     }
 
     // 28.

ext/fetch/23_response.js

@@ -11,6 +11,7 @@
 /// <reference lib="esnext" />
 
 import { core, primordials } from "ext:core/mod.js";
+import { op_fetch_upgrade_raw } from "ext:core/ops";
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 import { createFilteredInspectProxy } from "ext:deno_console/01_console.js";
 import {
@@ -30,6 +31,8 @@ import {
   headerListFromHeaders,
   headersFromHeaderList,
 } from "ext:deno_fetch/20_headers.js";
+import { UpgradedConn } from "ext:deno_net/01_net.js";
+
 const {
   ArrayPrototypeMap,
   ArrayPrototypePush,
@@ -516,6 +519,33 @@ function fromInnerResponse(inner, guard) {
   return response;
 }
 
+async function upgradeResponse(response) {
+  const inner = toInnerResponse(response);
+  if (
+    !inner.body || typeof inner.body.streamOrStatic.body !== "number" ||
+    inner.body.streamOrStatic.consumed
+  ) {
+    throw new TypeError("Response cannot be upgraded");
+  }
+  inner.body.streamOrStatic.consumed = true;
+  const { 0: upgradeRid, 1: info } = await op_fetch_upgrade_raw(
+    inner.body.streamOrStatic.body,
+  );
+  return new UpgradedConn(
+    upgradeRid,
+    {
+      transport: "tcp",
+      hostname: info[0],
+      port: info[1],
+    },
+    {
+      transport: "tcp",
+      hostname: info[2],
+      port: info[3],
+    },
+  );
+}
+
 export {
   abortedNetworkError,
   fromInnerResponse,
@@ -526,4 +556,5 @@ export {
   Response,
   ResponsePrototype,
   toInnerResponse,
+  upgradeResponse,
 };

ext/fetch/26_fetch.js

@@ -18,6 +18,7 @@ import {
   op_wasm_streaming_feed,
   op_wasm_streaming_set_url,
 } from "ext:core/ops";
+const { internalRidSymbol } = core;
 const {
   ArrayPrototypePush,
   ArrayPrototypeSplice,
@@ -175,7 +176,7 @@ async function mainFetch(req, recursive, terminator) {
     req.method,
     req.currentUrl(),
     req.headerList,
-    req.clientRid,
+    req.client?.[internalRidSymbol],
     reqBody !== null || reqRid !== null,
     reqBody,
     reqRid,
@@ -219,6 +220,7 @@ async function mainFetch(req, recursive, terminator) {
       return this.urlList[this.urlList.length - 1];
     },
     urlList: req.urlListProcessed,
+    info: resp.info,
   };
   if (redirectStatus(resp.status)) {
     switch (req.redirectMode) {
@@ -235,7 +237,11 @@ async function mainFetch(req, recursive, terminator) {
     }
   }
 
-  if (nullBodyStatus(response.status)) {
+  if (req.client?.allowUpgrades && response.status === 101) {
+    response.body = new InnerBody({ body: resp.responseRid, consumed: false });
+    const { responseRid } = resp;
+    terminator[abortSignal.add](() => core.tryClose(responseRid));
+  } else if (nullBodyStatus(response.status)) {
     core.close(resp.responseRid);
   } else {
     if (req.method === "HEAD" || req.method === "CONNECT") {

ext/fetch/Cargo.toml

@@ -20,6 +20,7 @@ data-url.workspace = true
 deno_core.workspace = true
 deno_error.workspace = true
 deno_fs.workspace = true
+deno_net.workspace = true
 deno_path_util.workspace = true
 deno_permissions.workspace = true
 deno_tls.workspace = true
@@ -34,6 +35,8 @@ hyper-rustls.workspace = true
 hyper-util.workspace = true
 ipnet.workspace = true
 percent-encoding.workspace = true
+pin-project.workspace = true
+rustls-tokio-stream.workspace = true
 rustls-webpki.workspace = true
 serde.workspace = true
 serde_json.workspace = true