feat(extensions/websocket): implement websocketstream

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

@@ -1127,3 +1127,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;
+}

cli/tests/integration/mod.rs

@@ -630,6 +630,27 @@ fn websocket() {
   assert!(status.success());
 }
 
+#[test]
+fn websocketstream() {
+  let _g = util::http_server();
+
+  let script = util::tests_path().join("websocketstream_test.ts");
+  let root_ca = util::tests_path().join("tls/RootCA.pem");
+  let status = util::deno_cmd()
+    .arg("test")
+    .arg("--unstable")
+    .arg("--allow-net")
+    .arg("--cert")
+    .arg(root_ca)
+    .arg(script)
+    .spawn()
+    .unwrap()
+    .wait()
+    .unwrap();
+
+  assert!(status.success());
+}
+
 #[cfg(not(windows))]
 #[test]
 fn set_raw_should_not_panic_on_no_tty() {

cli/tests/websocketstream_test.ts

@@ -0,0 +1,82 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+import {
+  assertEquals,
+  assertThrows,
+  assertThrowsAsync,
+} from "../../test_util/std/testing/asserts.ts";
+
+Deno.test("fragment", () => {
+  assertThrows(() => new WebSocketStream("ws://localhost:4242/#"));
+  assertThrows(() => new WebSocketStream("ws://localhost:4242/#foo"));
+});
+
+Deno.test("duplicate protocols", () => {
+  assertThrows(() =>
+    new WebSocketStream("ws://localhost:4242", {
+      protocols: ["foo", "foo"],
+    })
+  );
+});
+
+Deno.test("connect & close custom valid code", async () => {
+  const ws = new WebSocketStream("ws://localhost:4242");
+  await ws.connection;
+  ws.close({ code: 1000 });
+  await ws.closed;
+});
+
+Deno.test("connect & close custom invalid reason", async () => {
+  const ws = new WebSocketStream("ws://localhost:4242");
+  await ws.connection;
+  assertThrows(() => ws.close({ code: 1000, reason: "".padEnd(124, "o") }));
+  ws.close();
+  await ws.closed;
+});
+
+Deno.test("echo string", async () => {
+  const ws = new WebSocketStream("ws://localhost:4242");
+  const { readable, writable } = await ws.connection;
+  await writable.getWriter().write("foo");
+  const res = await readable.getReader().read();
+  assertEquals(res.value, "foo");
+  ws.close();
+  await ws.closed;
+});
+
+Deno.test("echo string tls", async () => {
+  const ws = new WebSocketStream("wss://localhost:4243");
+  const { readable, writable } = await ws.connection;
+  await writable.getWriter().write("foo");
+  const res = await readable.getReader().read();
+  assertEquals(res.value, "foo");
+  ws.close();
+  await ws.closed;
+});
+
+Deno.test("websocket error", async () => {
+  const ws = new WebSocketStream("wss://localhost:4242");
+  await Promise.all([
+    assertThrowsAsync(
+      () => ws.connection,
+      Deno.errors.UnexpectedEof,
+      "tls handshake eof",
+    ),
+    assertThrowsAsync(
+      () => ws.closed,
+      Deno.errors.UnexpectedEof,
+      "tls handshake eof",
+    ),
+  ]);
+});
+
+Deno.test("echo uint8array", async () => {
+  const ws = new WebSocketStream("ws://localhost:4242");
+  const { readable, writable } = await ws.connection;
+  const uint = new Uint8Array([102, 111, 111]);
+  await writable.getWriter().write(uint);
+  const res = await readable.getReader().read();
+  assertEquals(res.value, uint);
+  ws.close();
+  await ws.closed;
+});

extensions/web/06_streams.js

@@ -4452,6 +4452,8 @@
     isReadableStreamDisturbed,
     errorReadableStream,
     createProxy,
+    writableStreamClose,
+    Deferred,
     // Exposed in global runtime scope
     ByteLengthQueuingStrategy,
     CountQueuingStrategy,

extensions/websocket/01_websocket.js

@@ -233,7 +233,11 @@
 
       this[_url] = wsURL.href;
 
-      core.opSync("op_ws_check_permission", this[_url]);
+      core.opSync(
+        "op_ws_check_permission_and_cancel_handle",
+        this[_url],
+        false,
+      );
 
       if (typeof protocols === "string") {
         protocols = [protocols];