feat(ext/http): upgradeHttpRaw

cli/tests/unit/serve_test.ts

@@ -803,6 +803,85 @@ Deno.test({ permissions: { net: true } }, async function httpServerWebSocket() {
   await server;
 });
 
+Deno.test(
+  { permissions: { net: true } },
+  async function httpServerWebSocketRaw() {
+    const ac = new AbortController();
+    const listeningPromise = deferred();
+    const server = Deno.serve({
+      handler: async (request) => {
+        const { conn, response } = Deno.upgradeHttpRaw(request);
+        const buf = new Uint8Array(1024);
+        let read;
+
+        // Write our fake HTTP upgrade
+        await conn.write(
+          new TextEncoder().encode(
+            "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgraded\r\n\r\nExtra",
+          ),
+        );
+
+        // Upgrade data
+        read = await conn.read(buf);
+        assertEquals(
+          new TextDecoder().decode(buf.subarray(0, read!)),
+          "Upgrade data",
+        );
+        // Read the packet to echo
+        read = await conn.read(buf);
+        // Echo
+        await conn.write(buf.subarray(0, read!));
+
+        conn.close();
+        return response;
+      },
+      port: 4501,
+      signal: ac.signal,
+      onListen: onListen(listeningPromise),
+      onError: createOnErrorCb(ac),
+    });
+
+    await listeningPromise;
+
+    const conn = await Deno.connect({ port: 4501 });
+    await conn.write(
+      new TextEncoder().encode(
+        "GET / HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nUpgrade data",
+      ),
+    );
+    const buf = new Uint8Array(1024);
+    let len;
+
+    // Headers
+    let headers = "";
+    for (let i = 0; i < 2; i++) {
+      len = await conn.read(buf);
+      headers += new TextDecoder().decode(buf.subarray(0, len!));
+      if (headers.endsWith("Extra")) {
+        break;
+      }
+    }
+    assertMatch(
+      headers,
+      /HTTP\/1\.1 101 Switching Protocols[ ,.A-Za-z:0-9\r\n]*Extra/im,
+    );
+
+    // Data to echo
+    await conn.write(new TextEncoder().encode("buffer data"));
+
+    // Echo
+    len = await conn.read(buf);
+    assertEquals(
+      new TextDecoder().decode(buf.subarray(0, len!)),
+      "buffer data",
+    );
+
+    conn.close();
+    ac.abort();
+    await server;
+  },
+);
+
 Deno.test(
   { permissions: { net: true } },
   async function httpServerWebSocketUpgradeTwice() {

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

@@ -1514,26 +1514,34 @@ declare namespace Deno {
    */
   export function upgradeHttp(
     request: Request,
-  ): Promise<[Deno.Conn, Uint8Array]>;
+    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}).
    *
-   * Unlike {@linkcode Deno.upgradeHttp} this function does not require that you
-   * respond to the request with a {@linkcode Response} object. Instead this
-   * function returns the underlying connection and first packet received
-   * immediately, and then the caller is responsible for writing the response to
-   * the connection.
-   *
    * This method can only be called on requests originating the
    * {@linkcode Deno.serve} server.
    *
    * @category HTTP Server
    */
-  export function upgradeHttpRaw(request: Request): [Deno.Conn, Uint8Array];
+  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.
    *

ext/http/00_serve.js

@@ -32,6 +32,7 @@ import {
   readableStreamForRid,
   ReadableStreamPrototype,
 } from "ext:deno_web/06_streams.js";
+import { TcpConn } from "ext:deno_net/01_net.js";
 const {
   ObjectPrototypeIsPrototypeOf,
   SafeSet,
@@ -122,10 +123,22 @@ class InnerRequest {
       throw "upgradeHttp is unavailable in Deno.serve at this time";
     }
 
-    // upgradeHttpRaw is async
-    // TODO(mmastrac)
+    // upgradeHttpRaw is sync
     if (upgradeType == "upgradeHttpRaw") {
-      throw "upgradeHttp is unavailable in Deno.serve at this time";
+      const slabId = this.#slabId;
+
+      this.url();
+      this.headerList;
+      this.close();
+
+      this.#upgraded = () => {};
+
+      const upgradeRid = core.ops.op_upgrade_raw(slabId);
+
+      // TODO(mmastrac): remoteAddr
+      const conn = new TcpConn(upgradeRid, null, null);
+
+      return { response: UPGRADE_RESPONSE_SENTINEL, conn };
     }
 
     // upgradeWebSocket is sync

ext/http/01_http.js

@@ -482,14 +482,11 @@ function upgradeHttp(req) {
   return req[_deferred].promise;
 }
 
-async function upgradeHttpRaw(req, tcpConn) {
+function upgradeHttpRaw(req, tcpConn) {
   const inner = toInnerRequest(req);
   if (inner._wantsUpgrade) {
     return inner._wantsUpgrade("upgradeHttpRaw", arguments);
   }
-
-  const res = await core.opAsync("op_http_upgrade_early", inner[streamRid]);
-  return new TcpConn(res, tcpConn.remoteAddr, tcpConn.localAddr);
 }
 
 const spaceCharCode = StringPrototypeCharCodeAt(" ", 0);

ext/http/http_next.rs

@@ -10,11 +10,13 @@ use crate::response_body::CompletionHandle;
 use crate::response_body::ResponseBytes;
 use crate::response_body::ResponseBytesInner;
 use crate::response_body::V8StreamHttpResponseBody;
+use crate::websocket_upgrade::WebSocketUpgrade;
 use crate::LocalExecutor;
 use deno_core::error::AnyError;
 use deno_core::futures::TryFutureExt;
 use deno_core::op;
 use deno_core::AsyncRefCell;
+use deno_core::AsyncResult;
 use deno_core::BufView;
 use deno_core::ByteString;
 use deno_core::CancelFuture;
@@ -39,6 +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;
@@ -52,6 +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;
 
@@ -228,7 +235,76 @@ fn slab_insert(
 }
 
 #[op]
-pub fn op_upgrade_raw(_index: usize) {}
+pub fn op_upgrade_raw(
+  state: &mut OpState,
+  index: u32,
+) -> Result<ResourceId, AnyError> {
+  // Stage 1: extract the upgrade future
+  let upgrade = with_http_mut(index, |http| {
+    // Manually perform the upgrade. We're peeking into hyper's underlying machinery here a bit
+    http
+      .request_parts
+      .extensions
+      .remove::<OnUpgrade>()
+      .ok_or_else(|| AnyError::msg("upgrade unavailable"))
+  })?;
+
+  let (read, write) = tokio::io::duplex(1024);
+  let (mut read_rx, mut 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();
+    let mut buf = [0; 1024];
+    let upgraded = loop {
+      let read = Pin::new(&mut write_rx).read(&mut buf).await?;
+      match upgrade_stream.write(&buf[..read]) {
+        Ok(None) => continue,
+        Ok(Some((response, bytes))) => {
+          with_resp_mut(index, |resp| *resp = Some(response));
+          with_promise_mut(index, |promise| promise.complete(true));
+          let mut upgraded = upgrade.await?;
+          upgraded.write_all(&bytes).await?;
+          break upgraded;
+        }
+        Err(err) => return Err(err),
+      }
+    };
+
+    let (mut upgraded_rx, mut upgraded_tx) = tokio::io::split(upgraded);
+
+    spawn_local(async move {
+      let mut buf = [0; 1024];
+      loop {
+        let read = upgraded_rx.read(&mut buf).await?;
+        if read == 0 {
+          break;
+        }
+        read_tx.write_all(&buf[..read]).await?;
+      }
+      Ok::<_, AnyError>(())
+    });
+    spawn_local(async move {
+      let mut buf = [0; 1024];
+      loop {
+        let read = write_rx.read(&mut buf).await?;
+        if read == 0 {
+          break;
+        }
+        upgraded_tx.write_all(&buf[..read]).await?;
+      }
+      Ok::<_, AnyError>(())
+    });
+
+    Ok(())
+  });
+
+  Ok(
+    state
+      .resource_table
+      .add(UpgradeStream::new(read_rx, write_tx)),
+  )
+}
 
 #[op]
 pub async fn op_upgrade(
@@ -825,3 +901,41 @@ pub async fn op_http_wait(
 
   Ok(u32::MAX)
 }
+
+struct UpgradeStream {
+  read: AsyncRefCell<tokio::io::ReadHalf<tokio::io::DuplexStream>>,
+  write: AsyncRefCell<tokio::io::WriteHalf<tokio::io::DuplexStream>>,
+}
+
+impl UpgradeStream {
+  pub fn new(
+    read: tokio::io::ReadHalf<tokio::io::DuplexStream>,
+    write: tokio::io::WriteHalf<tokio::io::DuplexStream>,
+  ) -> Self {
+    Self {
+      read: AsyncRefCell::new(read),
+      write: AsyncRefCell::new(write),
+    }
+  }
+
+  async fn read(self: Rc<Self>, buf: &mut [u8]) -> Result<usize, AnyError> {
+    let read = RcRef::map(self, |this| &this.read);
+    let mut read = read.borrow_mut().await;
+    Ok(Pin::new(&mut *read).read(buf).await?)
+  }
+
+  async fn write(self: Rc<Self>, buf: &[u8]) -> Result<usize, AnyError> {
+    let write = RcRef::map(self, |this| &this.write);
+    let mut write = write.borrow_mut().await;
+    Ok(Pin::new(&mut *write).write(buf).await?)
+  }
+}
+
+impl Resource for UpgradeStream {
+  fn name(&self) -> Cow<str> {
+    "httpRawUpgradeStream".into()
+  }
+
+  deno_core::impl_readable_byob!();
+  deno_core::impl_writable!();
+}

ext/http/lib.rs

@@ -972,7 +972,7 @@ struct EarlyUpgradeSocket(AsyncRefCell<EarlyUpgradeSocketInner>, CancelHandle);
 enum EarlyUpgradeSocketInner {
   PreResponse(
     Rc<HttpStreamResource>,
-    WebSocketUpgrade,
+    WebSocketUpgrade<Body>,
     // Readers need to block in this state, so they can wait here for the broadcast.
     tokio::sync::broadcast::Sender<
       Rc<AsyncRefCell<tokio::io::ReadHalf<hyper::upgrade::Upgraded>>>,