feat(ext/http): upgradeHttpRaw

cli/tests/unit/serve_test.ts

@@ -17,6 +17,11 @@ import {
 } from "./test_util.ts";
 import { consoleSize } from "../../../runtime/js/40_tty.js";
 
+const {
+  upgradeHttpRaw,
+  // @ts-expect-error TypeScript (as of 3.7) does not support indexing namespaces by symbol
+} = Deno[Deno.internal];
+
 function createOnErrorCb(ac: AbortController): (err: unknown) => Response {
   return (err) => {
     console.error(err);
@@ -810,7 +815,7 @@ Deno.test(
     const listeningPromise = deferred();
     const server = Deno.serve({
       handler: async (request) => {
-        const { conn, response } = Deno.upgradeHttpRaw(request);
+        const { conn, response } = upgradeHttpRaw(request);
         const buf = new Uint8Array(1024);
         let read;
 

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

@@ -1517,32 +1517,6 @@ declare namespace Deno {
     headers?: HeadersInit,
   ): HttpUpgrade;
 
-  /** **UNSTABLE**: New API, yet to be vetted.
-   *
-   * Allows "hijacking" the connection that the request is associated with.
-   * This can be used to implement protocols that build on top of HTTP (eg.
-   * {@linkcode WebSocket}).
-   *
-   * This method can only be called on requests originating the
-   * {@linkcode Deno.serve} server.
-   *
-   * @category HTTP Server
-   */
-  export function upgradeHttpRaw(request: Request): HttpUpgrade;
-
-  /** The object that is returned from a {@linkcode Deno.upgradeHttp} or {@linkcode Deno.upgradeHttpRaw}
-   * request.
-   *
-   * @category Web Sockets */
-  export interface HttpUpgrade {
-    /** The response object that represents the HTTP response to the client,
-     * which should be used to the {@linkcode RequestEvent} `.respondWith()` for
-     * the upgrade to be successful. */
-    response: Response;
-    /** The socket interface to communicate to the client via. */
-    conn: Deno.Conn;
-  }
-
   /** **UNSTABLE**: New API, yet to be vetted.
    *
    * Open a new {@linkcode Deno.Kv} connection to persist data.

ext/http/00_serve.js

@@ -1,6 +1,7 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 const core = globalThis.Deno.core;
 const primordials = globalThis.__bootstrap.primordials;
+const internals = globalThis.__bootstrap.internals;
 
 const { BadResourcePrototype } = core;
 import { InnerBody } from "ext:deno_fetch/22_body.js";
@@ -10,7 +11,7 @@ import {
   newInnerResponse,
   toInnerResponse,
 } from "ext:deno_fetch/23_response.js";
-import { fromInnerRequest } from "ext:deno_fetch/23_request.js";
+import { fromInnerRequest, toInnerRequest } from "ext:deno_fetch/23_request.js";
 import { AbortController } from "ext:deno_web/03_abort_signal.js";
 import {
   _eventLoop,
@@ -83,6 +84,14 @@ const UPGRADE_RESPONSE_SENTINEL = fromInnerResponse(
   "immutable",
 );
 
+function upgradeHttpRaw(req, conn) {
+  const inner = toInnerRequest(req);
+  if (inner._wantsUpgrade) {
+    return inner._wantsUpgrade("upgradeHttpRaw", conn);
+  }
+  throw new TypeError("upgradeHttpRaw may only be used with Deno.serve");
+}
+
 class InnerRequest {
   #slabId;
   #context;
@@ -126,6 +135,7 @@ class InnerRequest {
     // upgradeHttpRaw is sync
     if (upgradeType == "upgradeHttpRaw") {
       const slabId = this.#slabId;
+      const underlyingConn = originalArgs[0];
 
       this.url();
       this.headerList;
@@ -135,8 +145,11 @@ class InnerRequest {
 
       const upgradeRid = core.ops.op_upgrade_raw(slabId);
 
-      // TODO(mmastrac): remoteAddr
-      const conn = new TcpConn(upgradeRid, null, null);
+      const conn = new TcpConn(
+        upgradeRid,
+        underlyingConn?.remoteAddr,
+        underlyingConn?.localAddr,
+      );
 
       return { response: UPGRADE_RESPONSE_SENTINEL, conn };
     }
@@ -626,4 +639,6 @@ async function serve(arg1, arg2) {
   }
 }
 
+internals.upgradeHttpRaw = upgradeHttpRaw;
+
 export { serve };

ext/http/01_http.js

@@ -482,13 +482,6 @@ function upgradeHttp(req) {
   return req[_deferred].promise;
 }
 
-function upgradeHttpRaw(req, tcpConn) {
-  const inner = toInnerRequest(req);
-  if (inner._wantsUpgrade) {
-    return inner._wantsUpgrade("upgradeHttpRaw", arguments);
-  }
-}
-
 const spaceCharCode = StringPrototypeCharCodeAt(" ", 0);
 const tabCharCode = StringPrototypeCharCodeAt("\t", 0);
 const commaCharCode = StringPrototypeCharCodeAt(",", 0);
@@ -563,4 +556,4 @@ function buildCaseInsensitiveCommaValueFinder(checkText) {
 internals.buildCaseInsensitiveCommaValueFinder =
   buildCaseInsensitiveCommaValueFinder;
 
-export { _ws, HttpConn, serve, upgradeHttp, upgradeHttpRaw, upgradeWebSocket };
+export { _ws, HttpConn, serve, upgradeHttp, upgradeWebSocket };

ext/http/http_next.rs

@@ -41,7 +41,7 @@ use hyper1::server::conn::http2;
 use hyper1::service::service_fn;
 use hyper1::service::HttpService;
 use hyper1::upgrade::OnUpgrade;
-use hyper1::upgrade::Upgraded;
+
 use hyper1::StatusCode;
 use pin_project::pin_project;
 use pin_project::pinned_drop;
@@ -55,10 +55,10 @@ use std::net::SocketAddr;
 use std::net::SocketAddrV4;
 use std::pin::Pin;
 use std::rc::Rc;
-use tokio::io::AsyncRead;
+
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
-use tokio::io::ReadBuf;
+
 use tokio::task::spawn_local;
 use tokio::task::JoinHandle;
 
@@ -250,11 +250,13 @@ pub fn op_upgrade_raw(
   })?;
 
   let (read, write) = tokio::io::duplex(1024);
-  let (mut read_rx, mut write_tx) = tokio::io::split(read);
+  let (read_rx, write_tx) = tokio::io::split(read);
   let (mut write_rx, mut read_tx) = tokio::io::split(write);
 
   spawn_local(async move {
     let mut upgrade_stream = WebSocketUpgrade::<ResponseBytes>::default();
+
+    // Stage 2: Extract the Upgraded connection
     let mut buf = [0; 1024];
     let upgraded = loop {
       let read = Pin::new(&mut write_rx).read(&mut buf).await?;
@@ -271,6 +273,7 @@ pub fn op_upgrade_raw(
       }
     };
 
+    // Stage 3: Pump the data
     let (mut upgraded_rx, mut upgraded_tx) = tokio::io::split(upgraded);
 
     spawn_local(async move {

runtime/js/90_deno_ns.js

@@ -171,7 +171,6 @@ const denoNsUnstable = {
   funlock: fs.funlock,
   funlockSync: fs.funlockSync,
   upgradeHttp: http.upgradeHttp,
-  upgradeHttpRaw: http.upgradeHttpRaw,
   serve: http.serve,
   openKv: kv.openKv,
   Kv: kv.Kv,