feat(extensions/websocket): implement websocketstream

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

@@ -1265,3 +1265,28 @@ declare interface WorkerOptions {
     };
   };
 }
+
+declare interface WebSocketStreamOptions {
+  protocols?: string[];
+  signal?: AbortSignal;
+}
+
+declare interface WebSocketConnection {
+  readable: ReadableStream<string | Uint8Array>;
+  writable: WritableStream<string | Uint8Array>;
+  extensions: string;
+  protocol: string;
+}
+
+declare interface WebSocketCloseInfo {
+  code?: number;
+  reason?: string;
+}
+
+declare class WebSocketStream {
+  constructor(url: string, options?: WebSocketStreamOptions);
+  url: string;
+  connection: Promise<WebSocketConnection>;
+  closed: Promise<WebSocketCloseInfo>;
+  close(closeInfo: WebSocketCloseInfo): void;
+}

op_crates/websocket/02_websocketstream.js

@@ -0,0 +1,247 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+"use strict";
+
+((window) => {
+  const core = window.Deno.core;
+
+  /** @template T */
+  class Deferred {
+    /** @type {Promise<T>} */
+    #promise;
+    /** @type {(reject?: any) => void} */
+    #reject;
+    /** @type {(value: T | PromiseLike<T>) => void} */
+    #resolve;
+    /** @type {"pending" | "fulfilled"} */
+    #state = "pending";
+
+    constructor() {
+      this.#promise = new Promise((resolve, reject) => {
+        this.#resolve = resolve;
+        this.#reject = reject;
+      });
+    }
+
+    /** @returns {Promise<T>} */
+    get promise() {
+      return this.#promise;
+    }
+
+    /** @returns {"pending" | "fulfilled"} */
+    get state() {
+      return this.#state;
+    }
+
+    /** @param {any=} reason */
+    reject(reason) {
+      // already settled promises are a no-op
+      if (this.#state !== "pending") {
+        return;
+      }
+      this.#state = "fulfilled";
+      this.#reject(reason);
+    }
+
+    /** @param {T | PromiseLike<T>} value */
+    resolve(value) {
+      // already settled promises are a no-op
+      if (this.#state !== "pending") {
+        return;
+      }
+      this.#state = "fulfilled";
+      this.#resolve(value);
+    }
+  }
+
+  /**
+   * Tries to close the resource (and ignores BadResource errors).
+   * @param {number} rid
+   */
+  function tryClose(rid) {
+    try {
+      core.close(rid);
+    } catch (err) {
+      // Ignore error if the socket has already been closed.
+      if (!(err instanceof Deno.errors.BadResource)) throw err;
+    }
+  }
+
+  class WebSocketStream {
+    #rid;
+
+    #url;
+    get url() {
+      return this.#url;
+    }
+
+    constructor(url, options) {
+      requiredArguments("WebSocket", arguments.length, 1);
+
+      const wsURL = new URL(url);
+
+      if (wsURL.protocol !== "ws:" && wsURL.protocol !== "wss:") {
+        throw new DOMException(
+          "Only ws & wss schemes are allowed in a WebSocket URL.",
+          "SyntaxError",
+        );
+      }
+
+      if (wsURL.hash !== "" || wsURL.href.endsWith("#")) {
+        throw new DOMException(
+          "Fragments are not allowed in a WebSocket URL.",
+          "SyntaxError",
+        );
+      }
+
+      this.#url = wsURL.href;
+
+      core.opSync("op_ws_check_permission", this.#url);
+
+      if (
+        options?.protocols?.some((x) =>
+          options.protocols.indexOf(x) !== options.protocols.lastIndexOf(x)
+        )
+      ) {
+        throw new DOMException(
+          "Can't supply multiple times the same protocol.",
+          "SyntaxError",
+        );
+      }
+
+      core.opAsync("op_ws_create", {
+        url: wsURL.href,
+        protocols: options?.protocols?.join(", "),
+      }).then((create) => {
+        if (create.success) {
+          options.abort.addEventListener("abort", () => this.close());
+
+          const readable = new ReadableStream({
+            pull: async (controller) => {
+              const { kind, value } = await core.opAsync(
+                "op_ws_next_event",
+                this.#rid,
+              );
+
+              switch (kind) {
+                case "string": {
+                  controller.enqueue(value);
+                  break;
+                }
+                case "binary": {
+                  controller.enqueue(new Uint8Array(value));
+                  break;
+                }
+                case "ping": {
+                  core.opAsync("op_ws_send", {
+                    rid: this.#rid,
+                    kind: "pong",
+                  });
+                  break;
+                }
+                case "close": {
+                  this.#closed.resolve(value);
+                  tryClose(this.#rid);
+                  break;
+                }
+                case "error": {
+                  this.#closed.reject();
+                  controller.error(value);
+                  tryClose(this.#rid);
+                  break;
+                }
+              }
+            },
+            cancel: (reason) => {
+              this.close(reason);
+            },
+          });
+          const writable = new WritableStream({
+            write: async (chunk) => {
+              if (typeof chunk === "string") {
+                await core.opAsync("op_ws_send", {
+                  rid: this.#rid,
+                  kind: "text",
+                  text: chunk,
+                });
+              } else if (chunk instanceof Uint8Array) {
+                await core.opAsync("op_ws_send", {
+                  rid: this.#rid,
+                  kind: "binary",
+                }, chunk);
+              }
+            },
+            cancel: (reason) => {
+              this.close(reason);
+            },
+            abort: (reason) => {
+              this.close(reason);
+            },
+          });
+
+          this.#connection.resolve({
+            readable,
+            writable,
+            extensions: create.extensions ?? "",
+            protocol: create.protocol ?? "",
+          });
+        } else {
+          this.#connection.reject();
+          this.#closed.reject();
+        }
+      });
+    }
+
+    #connection = new Deferred();
+    get connection() {
+      return this.#connection.promise;
+    }
+
+    #closed = new Deferred();
+    get closed() {
+      return this.#closed.promise;
+    }
+
+    close(closeInfo) {
+      if (
+        closeInfo?.code &&
+        !(closeInfo.code === 1000 ||
+          (3000 <= closeInfo.code && closeInfo.code < 5000))
+      ) {
+        throw new DOMException(
+          "The close code must be either 1000 or in the range of 3000 to 4999.",
+          "NotSupportedError",
+        );
+      }
+
+      const encoder = new TextEncoder();
+      if (
+        closeInfo?.reason && encoder.encode(closeInfo.reason).byteLength > 123
+      ) {
+        throw new DOMException(
+          "The close reason may not be longer than 123 bytes.",
+          "SyntaxError",
+        );
+      }
+
+      let code = closeInfo?.code;
+      if (closeInfo?.reason && code === undefined) {
+        code = 1000;
+      }
+
+      if (this.#closed.state === "pending") {
+        core.opAsync("op_ws_close", {
+          rid: this.#rid,
+          code,
+          reason: closeInfo?.reason,
+        }).then(() => {
+          this.#closed.resolve({
+            code: closeInfo?.code ?? 1005,
+            reason: closeInfo?.reason,
+          });
+        });
+      }
+    }
+  }
+
+  window.__bootstrap.webSocket.WebSocketStream = WebSocketStream;
+})(this);

op_crates/websocket/lib.rs

@@ -289,7 +289,7 @@ pub enum NextEventResponse {
   Close { code: u16, reason: String },
   Ping,
   Pong,
-  Error,
+  Error(String),
   Closed,
 }
 
@@ -323,7 +323,7 @@ pub async fn op_ws_next_event(
     },
     Some(Ok(Message::Ping(_))) => NextEventResponse::Ping,
     Some(Ok(Message::Pong(_))) => NextEventResponse::Pong,
-    Some(Err(_)) => NextEventResponse::Error,
+    Some(Err(e)) => NextEventResponse::Error(e.to_string()),
     None => {
       state.borrow_mut().resource_table.close(rid).unwrap();
       NextEventResponse::Closed
@@ -334,12 +334,19 @@ pub async fn op_ws_next_event(
 
 /// Load and execute the javascript code.
 pub fn init(isolate: &mut JsRuntime) {
-  isolate
-    .execute(
+  let files = vec![
+    (
       "deno:op_crates/websocket/01_websocket.js",
       include_str!("01_websocket.js"),
-    )
-    .unwrap();
+    ),
+    (
+      "deno:op_crates/websocket/02_websocketstream.js",
+      include_str!("02_websocketstream.js"),
+    ),
+  ];
+  for (url, source_code) in files {
+    isolate.execute(url, source_code).unwrap();
+  }
 }
 
 pub fn get_declaration() -> PathBuf {

runtime/js/99_main.js

@@ -277,6 +277,7 @@ delete Object.prototype.__proto__;
     URL: util.nonEnumerable(url.URL),
     URLSearchParams: util.nonEnumerable(url.URLSearchParams),
     WebSocket: util.nonEnumerable(webSocket.WebSocket),
+    WebSocketStream: util.nonEnumerable(webSocket.WebSocketStream),
     Worker: util.nonEnumerable(worker.Worker),
     WritableStream: util.nonEnumerable(streams.WritableStream),
     WritableStreamDefaultWriter: util.nonEnumerable(