feat(runtime/http): websocket support

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

@@ -1210,6 +1210,13 @@ declare namespace Deno {
    */
   export function serveHttp(conn: Conn): HttpConn;
 
+  export interface WebSocketUpgrade {
+    response: Response;
+    websocket: WebSocket;
+  }
+
+  export function upgradeWebSocket(request: Request): Promise<WebSocketUpgrade>;
+
   /** **UNSTABLE**: New option, yet to be vetted. */
   export interface TestDefinition {
     /** Specifies the permissions that should be used to run the test.

extensions/fetch/23_response.js

@@ -418,6 +418,7 @@
 
   window.__bootstrap.fetch ??= {};
   window.__bootstrap.fetch.Response = Response;
+  window.__bootstrap.fetch.newInnerResponse = newInnerResponse;
   window.__bootstrap.fetch.toInnerResponse = toInnerResponse;
   window.__bootstrap.fetch.fromInnerResponse = fromInnerResponse;
   window.__bootstrap.fetch.redirectStatus = redirectStatus;

extensions/websocket/01_websocket.js

@@ -94,6 +94,7 @@
   const _protocol = Symbol("[[protocol]]");
   const _binaryType = Symbol("[[binaryType]]");
   const _bufferedAmount = Symbol("[[bufferedAmount]]");
+  const _eventLoop = Symbol("[[eventLoop]]");
   class WebSocket extends EventTarget {
     [_rid];
 
@@ -255,7 +256,7 @@
           event.target = this;
           this.dispatchEvent(event);
 
-          this.#eventLoop();
+          this[_eventLoop]();
         }
       }).catch((err) => {
         this[_readyState] = CLOSED;
@@ -379,7 +380,7 @@
       }
     }
 
-    async #eventLoop() {
+    async [_eventLoop]() {
       while (this[_readyState] === OPEN) {
         const { kind, value } = await core.opAsync(
           "op_ws_next_event",

extensions/websocket/Cargo.toml

@@ -20,5 +20,6 @@ serde = { version = "1.0.125", features = ["derive"] }
 tokio = { version = "1.7.1", features = ["full"] }
 tokio-rustls = "0.22.0"
 tokio-tungstenite = { version = "0.14.0", features = ["rustls-tls"] }
+hyper = { version = "0.14.9" }
 webpki = "0.21.4"
 webpki-roots = "0.21.1"

extensions/websocket/lib.rs

@@ -64,13 +64,67 @@ impl WebSocketPermissions for NoWebSocketPermissions {
 }
 
 type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
-struct WsStreamResource {
-  tx: AsyncRefCell<SplitSink<WsStream, Message>>,
-  rx: AsyncRefCell<SplitStream<WsStream>>,
+pub enum WebSocketStreamType {
+  Client {
+    tx: AsyncRefCell<SplitSink<WsStream, Message>>,
+    rx: AsyncRefCell<SplitStream<WsStream>>,
+  },
+  Server {
+    tx: AsyncRefCell<SplitSink<WebSocketStream<hyper::upgrade::Upgraded>, Message>>,
+    rx: AsyncRefCell<SplitStream<WebSocketStream<hyper::upgrade::Upgraded>>>,
+  },
+}
+
+pub struct WsStreamResource {
+  pub stream: WebSocketStreamType,
   // 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 WsStreamResource {
+  async fn send(self: &Rc<Self>, message: Message) -> Result<(), AnyError> {
+    match self.stream {
+      WebSocketStreamType::Client { .. } => {
+        let mut tx = RcRef::map(self, |r| match &r.stream {
+          WebSocketStreamType::Client { tx, .. } => tx,
+          WebSocketStreamType::Server { .. } => unreachable!(),
+        }).borrow_mut().await;
+        tx.send(message).await?;
+      },
+      WebSocketStreamType::Server { .. } => {
+        let mut tx = RcRef::map(self, |r| match &r.stream {
+          WebSocketStreamType::Client { .. } => unreachable!(),
+          WebSocketStreamType::Server { tx, .. } => tx,
+        }).borrow_mut().await;
+        tx.send(message).await?;
+      },
+    }
+
+    Ok(())
+  }
+
+  async fn next_message(self: &Rc<Self>, cancel: RcRef<CancelHandle>) -> Result<Option<Result<Message, tokio_tungstenite::tungstenite::Error>>, AnyError> {
+    match &self.stream {
+      WebSocketStreamType::Client { .. } => {
+        let mut rx = RcRef::map(self, |r| match &r.stream {
+          WebSocketStreamType::Client { rx, .. } => rx,
+          WebSocketStreamType::Server { .. } => unreachable!(),
+        }).borrow_mut().await;
+        let message = rx.next().or_cancel(cancel).await?;
+        Ok(message)
+      },
+      WebSocketStreamType::Server { .. } => {
+        let mut rx = RcRef::map(self, |r| match &r.stream {
+          WebSocketStreamType::Client { .. } => unreachable!(),
+          WebSocketStreamType::Server { rx, .. } => rx,
+        }).borrow_mut().await;
+        let message = rx.next().or_cancel(cancel).await?;
+        Ok(message)
+      },
+    }
+  }
 }
 
 impl Resource for WsStreamResource {
@@ -79,8 +133,6 @@ impl Resource for WsStreamResource {
   }
 }
 
-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.
@@ -184,8 +236,10 @@ where
 
   let (ws_tx, ws_rx) = stream.split();
   let resource = WsStreamResource {
-    rx: AsyncRefCell::new(ws_rx),
-    tx: AsyncRefCell::new(ws_tx),
+    stream: WebSocketStreamType::Client {
+      rx: AsyncRefCell::new(ws_rx),
+      tx: AsyncRefCell::new(ws_tx),
+    },
     cancel: Default::default(),
   };
   let mut state = state.borrow_mut();
@@ -227,15 +281,13 @@ pub async fn op_ws_send(
     "pong" => Message::Pong(vec![]),
     _ => unreachable!(),
   };
-  let rid = args.rid;
 
   let resource = state
     .borrow_mut()
     .resource_table
-    .get::<WsStreamResource>(rid)
+    .get::<WsStreamResource>(args.rid)
     .ok_or_else(bad_resource_id)?;
-  let mut tx = RcRef::map(&resource, |r| &r.tx).borrow_mut().await;
-  tx.send(msg).await?;
+  resource.send(msg).await?;
   Ok(())
 }
 
@@ -266,8 +318,7 @@ pub async fn op_ws_close(
     .resource_table
     .get::<WsStreamResource>(rid)
     .ok_or_else(bad_resource_id)?;
-  let mut tx = RcRef::map(&resource, |r| &r.tx).borrow_mut().await;
-  tx.send(msg).await?;
+  resource.send(msg).await?;
   Ok(())
 }
 
@@ -294,9 +345,8 @@ pub async fn op_ws_next_event(
     .get::<WsStreamResource>(rid)
     .ok_or_else(bad_resource_id)?;
 
-  let mut rx = RcRef::map(&resource, |r| &r.rx).borrow_mut().await;
-  let cancel = RcRef::map(resource, |r| &r.cancel);
-  let val = rx.next().or_cancel(cancel).await?;
+  let cancel = RcRef::map(&resource, |r| &r.cancel);
+  let val = resource.next_message(cancel).await?;
   let res = match val {
     Some(Ok(Message::Text(text))) => NextEventResponse::String(text),
     Some(Ok(Message::Binary(data))) => NextEventResponse::Binary(data.into()),

runtime/Cargo.toml

@@ -50,6 +50,7 @@ deno_websocket = { version = "0.15.0", path = "../extensions/websocket" }
 deno_webstorage = { version = "0.5.0", path = "../extensions/webstorage" }
 
 atty = "0.2.14"
+base64 = "0.13.0"
 bytes = "1"
 dlopen = "0.1.8"
 encoding_rs = "0.8.28"

runtime/js/40_http.js

@@ -2,8 +2,11 @@
 "use strict";
 
 ((window) => {
+  const webidl = window.__bootstrap.webidl;
+  const { forgivingBase64Encode } = window.__bootstrap.infra;
   const { InnerBody } = window.__bootstrap.fetchBody;
-  const { Response, fromInnerRequest, toInnerResponse, newInnerRequest } =
+  const { setEventTargetData } = window.__bootstrap.eventTarget;
+  const { Response, fromInnerRequest, toInnerResponse, newInnerRequest, newInnerResponse, fromInnerResponse } =
     window.__bootstrap.fetch;
   const errors = window.__bootstrap.errors.errors;
   const core = window.Deno.core;
@@ -53,7 +56,7 @@
       if (nextRequest === null) return null;
 
       const [
-        requestBodyRid,
+        requestRid,
         responseSenderRid,
         method,
         headersList,
@@ -62,8 +65,8 @@
 
       /** @type {ReadableStream<Uint8Array> | undefined} */
       let body = null;
-      if (typeof requestBodyRid === "number") {
-        body = createRequestBodyStream(requestBodyRid);
+      if (typeof requestRid === "number") {
+        body = createRequestBodyStream(requestRid);
       }
 
       const innerRequest = newInnerRequest(
@@ -74,7 +77,11 @@
       );
       const request = fromInnerRequest(innerRequest, null, "immutable");
 
-      const respondWith = createRespondWith(this, responseSenderRid);
+      const respondWith = createRespondWith(
+        this,
+        responseSenderRid,
+        requestRid,
+      );
 
       return { request, respondWith };
     }
@@ -105,7 +112,7 @@
     );
   }
 
-  function createRespondWith(httpConn, responseSenderRid) {
+  function createRespondWith(httpConn, responseSenderRid, requestRid) {
     return async function respondWith(resp) {
       if (resp instanceof Promise) {
         resp = await resp;
@@ -206,10 +213,50 @@
           } catch { /* pass */ }
         }
       }
+
+      const ws = resp[_ws];
+      if (ws) {
+        const _readyState = Symbol.for("[[readyState]]");
+
+        core.opAsync("op_http_upgrade_websocket", requestRid).then((rid) => {
+          ws[Symbol.for("[[rid]]")] = rid;
+          // TODO: protocols & extensions
+
+          if (ws[_readyState] === WebSocket.CLOSING) {
+            core.opAsync("op_ws_close", {
+              rid,
+            }).then(() => {
+              ws[_readyState] = WebSocket.CLOSED;
+
+              const errEvent = new ErrorEvent("error");
+              errEvent.target = ws;
+              ws.dispatchEvent(errEvent);
+
+              const event = new CloseEvent("close");
+              event.target = ws;
+              ws.dispatchEvent(event);
+
+              try {
+                core.close(rid);
+              } catch (err) {
+                // Ignore error if the socket has already been closed.
+                if (!(err instanceof Deno.errors.BadResource)) throw err;
+              }
+            });
+          } else {
+            ws[_readyState] = WebSocket.OPEN;
+            const event = new Event("open");
+            event.target = ws;
+            ws.dispatchEvent(event);
+
+            ws[Symbol.for("[[eventLoop]]")]();
+          }
+        });
+      }
     };
   }
 
-  function createRequestBodyStream(requestBodyRid) {
+  function createRequestBodyStream(requestRid) {
     return new ReadableStream({
       type: "bytes",
       async pull(controller) {
@@ -218,7 +265,7 @@
           // stream.
           const chunk = new Uint8Array(16 * 1024 + 256);
           const read = await readRequest(
-            requestBodyRid,
+            requestRid,
             chunk,
           );
           if (read > 0) {
@@ -227,23 +274,61 @@
           } else {
             // We have reached the end of the body, so we close the stream.
             controller.close();
-            core.close(requestBodyRid);
+            core.close(requestRid);
           }
         } catch (err) {
           // There was an error while reading a chunk of the body, so we
           // error.
           controller.error(err);
           controller.close();
-          core.close(requestBodyRid);
+          core.close(requestRid);
         }
       },
       cancel() {
-        core.close(requestBodyRid);
+        core.close(requestRid);
       },
     });
   }
 
+  const _ws = Symbol("[[associated_ws]]");
+
+  async function upgradeWebSocket(request) {
+    if (request.headers["Upgrade"] !== "websocket") {
+      // Throw
+    }
+
+    if (request.headers["Connection"] !== "Upgrade") {
+      // Throw
+    }
+
+    if (!request.headers["Sec-WebSocket-Key"]) {
+      // Throw
+    }
+
+    const key = new TextEncoder().encode(
+      request.headers["Sec-WebSocket-Key"] +
+        "258EAFA5-E914-47DA-95CA-C5AB0DC85B11",
+    );
+    const accept = await crypto.subtle.digest("SHA-1", key);
+
+    const r = newInnerResponse(101);
+    r.headerList = [
+      ["Upgrade", "websocket"],
+      ["Connection", "Upgrade"],
+      ["Sec-WebSocket-Accept", forgivingBase64Encode(new Uint8Array(accept))]
+    ];
+
+    const response = fromInnerResponse(r, "immutable");
+
+    const websocket = webidl.createBranded(WebSocket);
+    setEventTargetData(websocket);
+    response[_ws] = websocket;
+
+    return { response, websocket };
+  }
+
   window.__bootstrap.http = {
     serveHttp,
+    upgradeWebSocket,
   };
 })(this);