feat(runtime/http): websocket support

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

@@ -1153,9 +1153,42 @@ declare namespace Deno {
 
   export function memoryUsage(): MemoryUsage;
 
+  export interface WebSocketConnEventString {
+    kind: "text";
+    value: string;
+  }
+
+  export interface WebSocketConnEventBinary {
+    kind: "binary";
+    value: Uint8Array;
+  }
+
+  export interface WebSocketConnEventClose {
+    kind: "close";
+    value: { code: number; reason: string };
+  }
+
+  export type WebSocketConnEvent =
+    | WebSocketConnEventString
+    | WebSocketConnEventBinary
+    | WebSocketConnEventClose;
+
+  export interface WebSocketConn extends AsyncIterable<WebSocketConnEvent> {
+    send(kind: "text", data: string): Promise<void>;
+    send(kind: "binary", data: Uint8Array): Promise<void>;
+
+    close(code?: number, reason?: string): Promise<void>;
+  }
+
+  export interface WebSocketUpgrade {
+    response: Response;
+    ws: WebSocketConn;
+  }
+
   export interface RequestEvent {
     readonly request: Request;
     respondWith(r: Response | Promise<Response>): Promise<void>;
+    upgradeWebSocket(): Promise<WebSocketUpgrade>;
   }
 
   export interface HttpConn extends AsyncIterable<RequestEvent> {

extensions/websocket/lib.rs

@@ -33,6 +33,8 @@ use std::io::Cursor;
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::Arc;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
 use tokio::net::TcpStream;
 use tokio_rustls::{rustls::ClientConfig, TlsConnector};
 use tokio_tungstenite::tungstenite::{
@@ -63,24 +65,22 @@ impl WebSocketPermissions for NoWebSocketPermissions {
   }
 }
 
-type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
-struct WsStreamResource {
-  tx: AsyncRefCell<SplitSink<WsStream, Message>>,
-  rx: AsyncRefCell<SplitStream<WsStream>>,
+pub type WsStream = MaybeTlsStream<TcpStream>;
+pub struct WsStreamResource<T: AsyncRead + AsyncWrite> {
+  pub tx: AsyncRefCell<SplitSink<WebSocketStream<T>, Message>>,
+  pub rx: AsyncRefCell<SplitStream<WebSocketStream<T>>>,
   // When a `WsStreamResource` resource is closed, all pending 'read' ops are
   // canceled, while 'write' ops are allowed to complete. Therefore only
   // 'read' futures are attached to this cancel handle.
-  cancel: CancelHandle,
+  pub cancel: CancelHandle,
 }
 
-impl Resource for WsStreamResource {
+impl<T: AsyncRead + AsyncWrite> Resource for WsStreamResource<T> {
   fn name(&self) -> Cow<str> {
     "webSocketStream".into()
   }
 }
 
-impl WsStreamResource {}
-
 // This op is needed because creating a WS instance in JavaScript is a sync
 // operation and should throw error when permissions are not fulfilled,
 // but actual op that connects WS is async.
@@ -174,7 +174,7 @@ where
     _ => unreachable!(),
   };
 
-  let (stream, response): (WsStream, Response) =
+  let (stream, response): (WebSocketStream<WsStream>, Response) =
     client_async(request, socket).await.map_err(|err| {
       type_error(format!(
         "failed to connect to WebSocket: {}",
@@ -216,7 +216,7 @@ pub struct SendArgs {
   text: Option<String>,
 }
 
-pub async fn op_ws_send(
+pub async fn op_ws_send<T: AsyncRead + AsyncWrite>(
   state: Rc<RefCell<OpState>>,
   args: SendArgs,
   buf: Option<ZeroCopyBuf>,
@@ -232,7 +232,7 @@ pub async fn op_ws_send(
   let resource = state
     .borrow_mut()
     .resource_table
-    .get::<WsStreamResource>(rid)
+    .get::<WsStreamResource<T>>(rid)
     .ok_or_else(bad_resource_id)?;
   let mut tx = RcRef::map(&resource, |r| &r.tx).borrow_mut().await;
   tx.send(msg).await?;
@@ -247,7 +247,7 @@ pub struct CloseArgs {
   reason: Option<String>,
 }
 
-pub async fn op_ws_close(
+pub async fn op_ws_close<T: AsyncRead + AsyncWrite>(
   state: Rc<RefCell<OpState>>,
   args: CloseArgs,
   _bufs: Option<ZeroCopyBuf>,
@@ -264,7 +264,7 @@ pub async fn op_ws_close(
   let resource = state
     .borrow_mut()
     .resource_table
-    .get::<WsStreamResource>(rid)
+    .get::<WsStreamResource<T>>(rid)
     .ok_or_else(bad_resource_id)?;
   let mut tx = RcRef::map(&resource, |r| &r.tx).borrow_mut().await;
   tx.send(msg).await?;
@@ -283,15 +283,15 @@ pub enum NextEventResponse {
   Closed,
 }
 
-pub async fn op_ws_next_event(
+pub async fn op_ws_next_event<T: AsyncRead + AsyncWrite>(
   state: Rc<RefCell<OpState>>,
   rid: ResourceId,
   _bufs: Option<ZeroCopyBuf>,
 ) -> Result<NextEventResponse, AnyError> {
   let resource = state
     .borrow_mut()
     .resource_table
-    .get::<WsStreamResource>(rid)
+    .get::<WsStreamResource<T>>(rid)
     .ok_or_else(bad_resource_id)?;
 
   let mut rx = RcRef::map(&resource, |r| &r.rx).borrow_mut().await;
@@ -334,9 +334,9 @@ pub fn init<P: WebSocketPermissions + 'static>(
         op_sync(op_ws_check_permission::<P>),
       ),
       ("op_ws_create", op_async(op_ws_create::<P>)),
-      ("op_ws_send", op_async(op_ws_send)),
-      ("op_ws_close", op_async(op_ws_close)),
-      ("op_ws_next_event", op_async(op_ws_next_event)),
+      ("op_ws_send", op_async(op_ws_send::<WsStream>)),
+      ("op_ws_close", op_async(op_ws_close::<WsStream>)),
+      ("op_ws_next_event", op_async(op_ws_next_event::<WsStream>)),
     ])
     .state(move |state| {
       state.put::<WsUserAgent>(WsUserAgent(user_agent.clone()));

runtime/Cargo.toml

@@ -48,6 +48,7 @@ deno_webidl = { path = "../extensions/webidl", version = "0.5.0" }
 deno_websocket = { path = "../extensions/websocket", version = "0.10.0" }
 
 atty = "0.2.14"
+base64 = "0.13.0"
 bytes = "1"
 dlopen = "0.1.8"
 encoding_rs = "0.8.28"

runtime/js/40_http.js

@@ -67,8 +67,9 @@
       const request = fromInnerRequest(innerRequest, "immutable");
 
       const respondWith = createRespondWith(responseSenderRid, this.#rid);
+      const upgradeWebSocket = createUpgradeWebSocket(requestBodyRid);
 
-      return { request, respondWith };
+      return { request, respondWith, upgradeWebSocket };
     }
 
     /** @returns {void} */
@@ -219,6 +220,66 @@
     });
   }
 
+  function createUpgradeWebSocket(rid) {
+    return async function upgradeWebSocket() {
+      const { key, rid } = await core.opAsync(
+        "op_http_upgrade_websocket",
+        rid,
+      );
+
+      const response = new Response(undefined, {
+        status: 101,
+        headers: {
+          "Upgrade": "websocket",
+          "Connection": "Upgrade",
+          "Sec-WebSocket-Accept": key,
+        },
+      });
+
+      const ws = {
+        async *[Symbol.asyncIterator]() {
+          let data = await core.opAsync(
+            "op_http_ws_next_event",
+            rid,
+          );
+
+          if (data.kind === "error") {
+            throw new Error(data.value);
+          } else if (data.kind === "close") {
+            return data;
+          } else {
+            yield data;
+          }
+        },
+        async send(data) {
+          if (typeof data === "string") {
+            await core.opAsync("op_http_ws_close", {
+              rid,
+              kind: "text",
+              text: data,
+            });
+          } else if (data instanceof Uint8Array) {
+            await core.opAsync("op_http_ws_close", {
+              rid,
+              kind: "binary",
+            }, data);
+          } else {
+            throw new TypeError("Only string or Uint8Array can be sent");
+          }
+        },
+        async close(code, reason) {
+          await core.opAsync("op_http_ws_close", {
+            rid,
+            code,
+            reason,
+          });
+        },
+      };
+
+      return { response, ws };
+    };
+  }
+
   window.__bootstrap.http = {
     serveHttp,
   };

runtime/ops/http.rs

@@ -7,6 +7,7 @@ use deno_core::error::null_opbuf;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::futures::future::poll_fn;
+use deno_core::futures::stream::SplitStream;
 use deno_core::futures::FutureExt;
 use deno_core::futures::Stream;
 use deno_core::futures::StreamExt;
@@ -22,11 +23,15 @@ use deno_core::RcRef;
 use deno_core::Resource;
 use deno_core::ResourceId;
 use deno_core::ZeroCopyBuf;
+use deno_websocket::tokio_tungstenite::tungstenite::Message;
+use deno_websocket::tokio_tungstenite::WebSocketStream;
+use deno_websocket::WsStreamResource;
 use hyper::body::HttpBody;
 use hyper::http;
 use hyper::server::conn::Connection;
 use hyper::server::conn::Http;
 use hyper::service::Service as HyperService;
+use hyper::upgrade::Upgraded;
 use hyper::Body;
 use hyper::Request;
 use hyper::Response;
@@ -55,6 +60,22 @@ pub fn init() -> Extension {
       ("op_http_response", op_async(op_http_response)),
       ("op_http_response_write", op_async(op_http_response_write)),
       ("op_http_response_close", op_async(op_http_response_close)),
+      (
+        "op_http_upgrade_websocket",
+        op_async(op_http_upgrade_websocket),
+      ),
+      (
+        "op_http_ws_send",
+        op_async(deno_websocket::op_ws_send::<WebSocketStream<Upgraded>>),
+      ),
+      (
+        "op_http_ws_close",
+        op_async(deno_websocket::op_ws_close::<WebSocketStream<Upgraded>>),
+      ),
+      (
+        "op_http_ws_next_event",
+        op_async(deno_websocket::op_ws_next_event::<WebSocketStream<Upgraded>>),
+      ),
     ])
     .build()
 }
@@ -507,6 +528,60 @@ async fn op_http_response_write(
   Ok(())
 }
 
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HttpUpgradeWebsocketResponse {
+  rid: ResourceId,
+  key: String,
+}
+
+async fn op_http_upgrade_websocket(
+  state: Rc<RefCell<OpState>>,
+  rid: ResourceId,
+  _data: Option<ZeroCopyBuf>,
+) -> Result<HttpUpgradeWebsocketResponse, AnyError> {
+  let conn_resource = state
+    .borrow()
+    .resource_table
+    .get::<ConnResource>(rid)
+    .ok_or_else(bad_resource_id)?;
+
+  let service = conn_resource.deno_service.inner.borrow_mut().ok_or("")?; // TODO
+
+  let key = service
+    .request
+    .headers()
+    .get(http::header::SEC_WEBSOCKET_KEY)
+    .ok_or("failed to read ws key from headers")?; // TODO
+  let digest = ring::digest::digest(
+    &ring::digest::SHA1_FOR_LEGACY_USE_ONLY,
+    format!("{}258EAFA5-E914-47DA-95CA-C5AB0DC85B11", key.to_str()?).as_bytes(),
+  );
+  let return_key = base64::encode(digest);
+
+  let upgraded = service.request.on_upgrade().await?;
+  let stream =
+    deno_websocket::tokio_tungstenite::WebSocketStream::from_raw_socket(
+      upgraded,
+      deno_websocket::tokio_tungstenite::tungstenite::protocol::Role::Server,
+      None,
+    )
+    .await;
+
+  let (ws_tx, ws_rx) = stream.split();
+
+  let rid = state.borrow_mut().resource_table.add(WsStreamResource {
+    rx: AsyncRefCell::new(ws_rx),
+    tx: AsyncRefCell::new(ws_tx),
+    cancel: Default::default(),
+  });
+
+  Ok(HttpUpgradeWebsocketResponse {
+    key: return_key,
+    rid,
+  })
+}
+
 type BytesStream =
   Pin<Box<dyn Stream<Item = std::io::Result<bytes::Bytes>> + Unpin>>;