feat(unstable): Add Deno.upgradeHttp API (#13618)

This commit adds "Deno.upgradeHttp" API, which
allows to "hijack" connection and switch protocols, to eg.
implement WebSocket required for Node compat.

Co-authored-by: crowlkats <crowlkats@toaxl.com>
Co-authored-by: Ryan Dahl <ry@tinyclouds.org>
Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

1 files changed, 79 insertions, 1 deletions

diff --git a/runtime/ops/http.rs b/runtime/ops/http.rs
index 47ec31751..1e2bd66ec 100644
--- a/runtime/ops/http.rs
+++ b/runtime/ops/http.rs
@@ -1,18 +1,28 @@
+use std::cell::RefCell;
 use std::rc::Rc;
 
 use deno_core::error::bad_resource_id;
+use deno_core::error::custom_error;
 use deno_core::error::AnyError;
 use deno_core::op;
 use deno_core::Extension;
 use deno_core::OpState;
+use deno_core::RcRef;
 use deno_core::ResourceId;
+use deno_core::ZeroCopyBuf;
 use deno_http::http_create_conn_resource;
+use deno_http::HttpRequestReader;
+use deno_http::HttpStreamResource;
 use deno_net::io::TcpStreamResource;
+use deno_net::ops_tls::TlsStream;
 use deno_net::ops_tls::TlsStreamResource;
+use hyper::upgrade::Parts;
+use serde::Serialize;
+use tokio::net::TcpStream;
 
 pub fn init() -> Extension {
   Extension::builder()
-    .ops(vec![op_http_start::decl()])
+    .ops(vec![op_http_start::decl(), op_http_upgrade::decl()])
     .build()
 }
 
@@ -62,3 +72,71 @@ fn op_http_start(
 
   Err(bad_resource_id())
 }
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HttpUpgradeResult {
+  conn_rid: ResourceId,
+  conn_type: &'static str,
+  read_buf: ZeroCopyBuf,
+}
+
+#[op]
+async fn op_http_upgrade(
+  state: Rc<RefCell<OpState>>,
+  rid: ResourceId,
+  _: (),
+) -> Result<HttpUpgradeResult, AnyError> {
+  let stream = state
+    .borrow_mut()
+    .resource_table
+    .get::<HttpStreamResource>(rid)?;
+  let mut rd = RcRef::map(&stream, |r| &r.rd).borrow_mut().await;
+
+  let request = match &mut *rd {
+    HttpRequestReader::Headers(request) => request,
+    _ => {
+      return Err(custom_error(
+        "Http",
+        "cannot upgrade because request body was used",
+      ))
+    }
+  };
+
+  let transport = hyper::upgrade::on(request).await?;
+  let transport = match transport.downcast::<TcpStream>() {
+    Ok(Parts {
+      io: tcp_stream,
+      read_buf,
+      ..
+    }) => {
+      return Ok(HttpUpgradeResult {
+        conn_type: "tcp",
+        conn_rid: state
+          .borrow_mut()
+          .resource_table
+          .add(TcpStreamResource::new(tcp_stream.into_split())),
+        read_buf: read_buf.to_vec().into(),
+      });
+    }
+    Err(transport) => transport,
+  };
+  match transport.downcast::<TlsStream>() {
+    Ok(Parts {
+      io: tls_stream,
+      read_buf,
+      ..
+    }) => Ok(HttpUpgradeResult {
+      conn_type: "tls",
+      conn_rid: state
+        .borrow_mut()
+        .resource_table
+        .add(TlsStreamResource::new(tls_stream.into_split())),
+      read_buf: read_buf.to_vec().into(),
+    }),
+    Err(_) => Err(custom_error(
+      "Http",
+      "encountered unsupported transport while upgrading",
+    )),
+  }
+}