refactor: move WebSocket API to an op_crate (#9026)

9 files changed, 38 insertions, 648 deletions

diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml
index 8632019b6..f641dcb42 100644
--- a/runtime/Cargo.toml
+++ b/runtime/Cargo.toml
@@ -22,6 +22,7 @@ deno_core = { path = "../core", version = "0.75.0" }
 deno_crypto = { path = "../op_crates/crypto", version = "0.9.0" }
 deno_fetch = { path = "../op_crates/fetch", version = "0.18.0" }
 deno_web = { path = "../op_crates/web", version = "0.26.0" }
+deno_websocket = { path = "../op_crates/websocket", version = "0.1.0" }
 
 [target.'cfg(windows)'.build-dependencies]
 winres = "0.1.11"
@@ -32,6 +33,7 @@ deno_core = { path = "../core", version = "0.75.0" }
 deno_crypto = { path = "../op_crates/crypto", version = "0.9.0" }
 deno_fetch = { path = "../op_crates/fetch", version = "0.18.0" }
 deno_web = { path = "../op_crates/web", version = "0.26.0" }
+deno_websocket = { path = "../op_crates/websocket", version = "0.1.0" }
 
 atty = "0.2.14"
 dlopen = "0.1.8"
@@ -55,7 +57,6 @@ sys-info = "0.7.0"
 termcolor = "1.1.0"
 tokio = { version = "0.2.22", features = ["full"] }
 tokio-rustls = "0.14.1"
-tokio-tungstenite = "0.11.0"
 uuid = { version = "0.8.1", features = ["v4"] }
 hyper = "0.13.9"
 webpki = "0.21.3"
diff --git a/runtime/build.rs b/runtime/build.rs
index ca4a77c31..a34e9ffa8 100644
--- a/runtime/build.rs
+++ b/runtime/build.rs
@@ -15,6 +15,7 @@ fn create_snapshot(
 ) {
   deno_web::init(&mut js_runtime);
   deno_fetch::init(&mut js_runtime);
+  deno_websocket::init(&mut js_runtime);
   deno_crypto::init(&mut js_runtime);
   // TODO(nayeemrmn): https://github.com/rust-lang/cargo/issues/3946 to get the
   // workspace root.
diff --git a/runtime/inspector.rs b/runtime/inspector.rs
index befaaebee..ae90e0fae 100644
--- a/runtime/inspector.rs
+++ b/runtime/inspector.rs
@@ -23,6 +23,7 @@ use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
 use deno_core::v8;
+use deno_websocket::tokio_tungstenite::tungstenite;
 use std::collections::HashMap;
 use std::ffi::c_void;
 use std::mem::replace;
@@ -40,7 +41,6 @@ use std::sync::Mutex;
 use std::thread;
 use std::{cell::BorrowMutError, convert::Infallible};
 use std::{cell::RefCell, rc::Rc};
-use tokio_tungstenite::tungstenite;
 use uuid::Uuid;
 
 pub struct InspectorServer {
@@ -185,12 +185,13 @@ fn handle_ws_request(
     if resp.is_ok() {
       tokio::task::spawn_local(async move {
         let upgraded = body.on_upgrade().await.unwrap();
-        let websocket = tokio_tungstenite::WebSocketStream::from_raw_socket(
-          upgraded,
-          tungstenite::protocol::Role::Server,
-          None,
-        )
-        .await;
+        let websocket =
+          deno_websocket::tokio_tungstenite::WebSocketStream::from_raw_socket(
+            upgraded,
+            tungstenite::protocol::Role::Server,
+            None,
+          )
+          .await;
         let (proxy, pump) = create_websocket_proxy(websocket);
 
         let _ = new_websocket_tx.unbounded_send(proxy);
@@ -353,7 +354,9 @@ impl WebSocketProxy {
 /// be used to send/receive messages on the websocket, and the second element
 /// is a future that does the forwarding.
 fn create_websocket_proxy(
-  websocket: tokio_tungstenite::WebSocketStream<hyper::upgrade::Upgraded>,
+  websocket: deno_websocket::tokio_tungstenite::WebSocketStream<
+    hyper::upgrade::Upgraded,
+  >,
 ) -> (WebSocketProxy, impl Future<Output = ()> + Send) {
   // The 'outbound' channel carries messages sent to the websocket.
   let (outbound_tx, outbound_rx) = mpsc::unbounded();
diff --git a/runtime/js/27_websocket.js b/runtime/js/27_websocket.js
deleted file mode 100644
index 9f86bdbed..000000000
--- a/runtime/js/27_websocket.js
+++ /dev/null
@@ -1,325 +0,0 @@
-// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
-
-((window) => {
-  const core = window.Deno.core;
-  const { requiredArguments, defineEventHandler } = window.__bootstrap.webUtil;
-  const CONNECTING = 0;
-  const OPEN = 1;
-  const CLOSING = 2;
-  const CLOSED = 3;
-
-  class WebSocket extends EventTarget {
-    #readyState = CONNECTING;
-
-    constructor(url, protocols = []) {
-      super();
-      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.jsonOpSync("op_ws_check_permission", {
-        url: this.#url,
-      });
-
-      if (protocols && typeof protocols === "string") {
-        protocols = [protocols];
-      }
-
-      if (
-        protocols.some((x) => protocols.indexOf(x) !== protocols.lastIndexOf(x))
-      ) {
-        throw new DOMException(
-          "Can't supply multiple times the same protocol.",
-          "SyntaxError",
-        );
-      }
-
-      core.jsonOpAsync("op_ws_create", {
-        url: wsURL.href,
-        protocols: protocols.join(", "),
-      }).then((create) => {
-        if (create.success) {
-          this.#rid = create.rid;
-          this.#extensions = create.extensions;
-          this.#protocol = create.protocol;
-
-          if (this.#readyState === CLOSING) {
-            core.jsonOpAsync("op_ws_close", {
-              rid: this.#rid,
-            }).then(() => {
-              this.#readyState = CLOSED;
-
-              const errEvent = new ErrorEvent("error");
-              errEvent.target = this;
-              this.dispatchEvent(errEvent);
-
-              const event = new CloseEvent("close");
-              event.target = this;
-              this.dispatchEvent(event);
-              core.close(this.#rid);
-            });
-          } else {
-            this.#readyState = OPEN;
-            const event = new Event("open");
-            event.target = this;
-            this.dispatchEvent(event);
-
-            this.#eventLoop();
-          }
-        } else {
-          this.#readyState = CLOSED;
-
-          const errEvent = new ErrorEvent("error");
-          errEvent.target = this;
-          this.dispatchEvent(errEvent);
-
-          const closeEvent = new CloseEvent("close");
-          closeEvent.target = this;
-          this.dispatchEvent(closeEvent);
-        }
-      }).catch((err) => {
-        this.#readyState = CLOSED;
-
-        const errorEv = new ErrorEvent(
-          "error",
-          { error: err, message: err.toString() },
-        );
-        errorEv.target = this;
-        this.dispatchEvent(errorEv);
-
-        const closeEv = new CloseEvent("close");
-        closeEv.target = this;
-        this.dispatchEvent(closeEv);
-      });
-    }
-
-    get CONNECTING() {
-      return CONNECTING;
-    }
-    get OPEN() {
-      return OPEN;
-    }
-    get CLOSING() {
-      return CLOSING;
-    }
-    get CLOSED() {
-      return CLOSED;
-    }
-
-    get readyState() {
-      return this.#readyState;
-    }
-
-    #extensions = "";
-    #protocol = "";
-    #url = "";
-    #rid;
-
-    get extensions() {
-      return this.#extensions;
-    }
-    get protocol() {
-      return this.#protocol;
-    }
-
-    #binaryType = "blob";
-    get binaryType() {
-      return this.#binaryType;
-    }
-    set binaryType(value) {
-      if (value === "blob" || value === "arraybuffer") {
-        this.#binaryType = value;
-      }
-    }
-    #bufferedAmount = 0;
-    get bufferedAmount() {
-      return this.#bufferedAmount;
-    }
-
-    get url() {
-      return this.#url;
-    }
-
-    send(data) {
-      requiredArguments("WebSocket.send", arguments.length, 1);
-
-      if (this.#readyState != OPEN) {
-        throw Error("readyState not OPEN");
-      }
-
-      const sendTypedArray = (ta) => {
-        this.#bufferedAmount += ta.size;
-        core.jsonOpAsync("op_ws_send", {
-          rid: this.#rid,
-          kind: "binary",
-        }, ta).then(() => {
-          this.#bufferedAmount -= ta.size;
-        });
-      };
-
-      if (data instanceof Blob) {
-        data.slice().arrayBuffer().then((ab) =>
-          sendTypedArray(new DataView(ab))
-        );
-      } else if (
-        data instanceof Int8Array || data instanceof Int16Array ||
-        data instanceof Int32Array || data instanceof Uint8Array ||
-        data instanceof Uint16Array || data instanceof Uint32Array ||
-        data instanceof Uint8ClampedArray || data instanceof Float32Array ||
-        data instanceof Float64Array || data instanceof DataView
-      ) {
-        sendTypedArray(data);
-      } else if (data instanceof ArrayBuffer) {
-        sendTypedArray(new DataView(data));
-      } else {
-        const string = String(data);
-        const encoder = new TextEncoder();
-        const d = encoder.encode(string);
-        this.#bufferedAmount += d.size;
-        core.jsonOpAsync("op_ws_send", {
-          rid: this.#rid,
-          kind: "text",
-          text: string,
-        }).then(() => {
-          this.#bufferedAmount -= d.size;
-        });
-      }
-    }
-
-    close(code, reason) {
-      if (code && (code !== 1000 && !(3000 <= 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 (reason && encoder.encode(reason).byteLength > 123) {
-        throw new DOMException(
-          "The close reason may not be longer than 123 bytes.",
-          "SyntaxError",
-        );
-      }
-
-      if (this.#readyState === CONNECTING) {
-        this.#readyState = CLOSING;
-      } else if (this.#readyState === OPEN) {
-        this.#readyState = CLOSING;
-
-        core.jsonOpAsync("op_ws_close", {
-          rid: this.#rid,
-          code,
-          reason,
-        }).then(() => {
-          this.#readyState = CLOSED;
-          const event = new CloseEvent("close", {
-            wasClean: true,
-            code,
-            reason,
-          });
-          event.target = this;
-          this.dispatchEvent(event);
-          core.close(this.#rid);
-        });
-      }
-    }
-
-    async #eventLoop() {
-      if (this.#readyState === OPEN) {
-        const message = await core.jsonOpAsync(
-          "op_ws_next_event",
-          { rid: this.#rid },
-        );
-        if (message.type === "string" || message.type === "binary") {
-          let data;
-
-          if (message.type === "string") {
-            data = message.data;
-          } else {
-            if (this.binaryType === "blob") {
-              data = new Blob([new Uint8Array(message.data)]);
-            } else {
-              data = new Uint8Array(message.data).buffer;
-            }
-          }
-
-          const event = new MessageEvent("message", {
-            data,
-            origin: this.#url,
-          });
-          event.target = this;
-          this.dispatchEvent(event);
-
-          this.#eventLoop();
-        } else if (message.type === "ping") {
-          core.jsonOpAsync("op_ws_send", {
-            rid: this.#rid,
-            kind: "pong",
-          });
-
-          this.#eventLoop();
-        } else if (message.type === "close") {
-          this.#readyState = CLOSED;
-          const event = new CloseEvent("close", {
-            wasClean: true,
-            code: message.code,
-            reason: message.reason,
-          });
-          event.target = this;
-          this.dispatchEvent(event);
-        } else if (message.type === "error") {
-          this.#readyState = CLOSED;
-
-          const errorEv = new ErrorEvent("error");
-          errorEv.target = this;
-          this.dispatchEvent(errorEv);
-
-          this.#readyState = CLOSED;
-          const closeEv = new CloseEvent("close");
-          closeEv.target = this;
-          this.dispatchEvent(closeEv);
-        }
-      }
-    }
-  }
-
-  Object.defineProperties(WebSocket, {
-    CONNECTING: {
-      value: 0,
-    },
-    OPEN: {
-      value: 1,
-    },
-    CLOSING: {
-      value: 2,
-    },
-    CLOSED: {
-      value: 3,
-    },
-  });
-
-  defineEventHandler(WebSocket.prototype, "message");
-  defineEventHandler(WebSocket.prototype, "error");
-  defineEventHandler(WebSocket.prototype, "close");
-  defineEventHandler(WebSocket.prototype, "open");
-  window.__bootstrap.webSocket = {
-    WebSocket,
-  };
-})(this);
diff --git a/runtime/lib.rs b/runtime/lib.rs
index 6745f3ec8..cc6d02681 100644
--- a/runtime/lib.rs
+++ b/runtime/lib.rs
@@ -10,6 +10,7 @@ extern crate log;
 pub use deno_crypto;
 pub use deno_fetch;
 pub use deno_web;
+pub use deno_websocket;
 
 pub mod colors;
 pub mod errors;
diff --git a/runtime/ops/websocket.rs b/runtime/ops/websocket.rs
index b220655ae..3ecdae088 100644
--- a/runtime/ops/websocket.rs
+++ b/runtime/ops/websocket.rs
@@ -1,330 +1,33 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
-
 use crate::permissions::Permissions;
-use deno_core::error::bad_resource_id;
-use deno_core::error::type_error;
-use deno_core::error::AnyError;
-use deno_core::futures::stream::SplitSink;
-use deno_core::futures::stream::SplitStream;
-use deno_core::futures::SinkExt;
-use deno_core::futures::StreamExt;
-use deno_core::serde_json::json;
-use deno_core::serde_json::Value;
-use deno_core::url;
-use deno_core::AsyncRefCell;
-use deno_core::BufVec;
-use deno_core::CancelFuture;
-use deno_core::CancelHandle;
-use deno_core::OpState;
-use deno_core::RcRef;
-use deno_core::Resource;
-use deno_core::{serde_json, ZeroCopyBuf};
-use http::{Method, Request, Uri};
-use serde::Deserialize;
-use std::borrow::Cow;
-use std::cell::RefCell;
-use std::io::BufReader;
-use std::io::Cursor;
-use std::rc::Rc;
-use std::sync::Arc;
-use tokio::net::TcpStream;
-use tokio_rustls::{rustls::ClientConfig, TlsConnector};
-use tokio_tungstenite::stream::Stream as StreamSwitcher;
-use tokio_tungstenite::tungstenite::Error as TungsteniteError;
-use tokio_tungstenite::tungstenite::{
-  handshake::client::Response, protocol::frame::coding::CloseCode,
-  protocol::CloseFrame, Message,
-};
-use tokio_tungstenite::{client_async, WebSocketStream};
-use webpki::DNSNameRef;
-
-#[derive(Clone)]
-struct WsCaData(Vec<u8>);
-#[derive(Clone)]
-struct WsUserAgent(String);
+use deno_websocket::op_ws_check_permission;
+use deno_websocket::op_ws_close;
+use deno_websocket::op_ws_create;
+use deno_websocket::op_ws_next_event;
+use deno_websocket::op_ws_send;
+use deno_websocket::WsCaData;
+use deno_websocket::WsUserAgent;
 
 pub fn init(
   rt: &mut deno_core::JsRuntime,
-  ca_data: Option<Vec<u8>>,
   user_agent: String,
+  ca_data: Option<Vec<u8>>,
 ) {
   {
     let op_state = rt.op_state();
     let mut state = op_state.borrow_mut();
+    state.put::<WsUserAgent>(WsUserAgent(user_agent));
     if let Some(ca_data) = ca_data {
       state.put::<WsCaData>(WsCaData(ca_data));
     }
-    state.put::<WsUserAgent>(WsUserAgent(user_agent));
   }
-  super::reg_json_sync(rt, "op_ws_check_permission", op_ws_check_permission);
-  super::reg_json_async(rt, "op_ws_create", op_ws_create);
+  super::reg_json_sync(
+    rt,
+    "op_ws_check_permission",
+    op_ws_check_permission::<Permissions>,
+  );
+  super::reg_json_async(rt, "op_ws_create", op_ws_create::<Permissions>);
   super::reg_json_async(rt, "op_ws_send", op_ws_send);
   super::reg_json_async(rt, "op_ws_close", op_ws_close);
   super::reg_json_async(rt, "op_ws_next_event", op_ws_next_event);
 }
-
-type MaybeTlsStream =
-  StreamSwitcher<TcpStream, tokio_rustls::client::TlsStream<TcpStream>>;
-
-type WsStream = WebSocketStream<MaybeTlsStream>;
-struct WsStreamResource {
-  tx: AsyncRefCell<SplitSink<WsStream, Message>>,
-  rx: AsyncRefCell<SplitStream<WsStream>>,
-  // 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,
-}
-
-impl Resource for WsStreamResource {
-  fn name(&self) -> Cow<str> {
-    "webSocketStream".into()
-  }
-}
-
-impl WsStreamResource {}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct CheckPermissionArgs {
-  url: String,
-}
-
-// This op is needed because creating a WS instance in JavaScript is a sync
-// operation and should throw error when permissions are not fullfiled,
-// but actual op that connects WS is async.
-pub fn op_ws_check_permission(
-  state: &mut OpState,
-  args: Value,
-  _zero_copy: &mut [ZeroCopyBuf],
-) -> Result<Value, AnyError> {
-  let args: CheckPermissionArgs = serde_json::from_value(args)?;
-
-  state
-    .borrow::<Permissions>()
-    .check_net_url(&url::Url::parse(&args.url)?)?;
-
-  Ok(json!({}))
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct CreateArgs {
-  url: String,
-  protocols: String,
-}
-
-pub async fn op_ws_create(
-  state: Rc<RefCell<OpState>>,
-  args: Value,
-  _bufs: BufVec,
-) -> Result<Value, AnyError> {
-  let args: CreateArgs = serde_json::from_value(args)?;
-
-  {
-    let s = state.borrow();
-    s.borrow::<Permissions>()
-      .check_net_url(&url::Url::parse(&args.url)?)
-      .expect(
-        "Permission check should have been done in op_ws_check_permission",
-      );
-  }
-
-  let ws_ca_data = state.borrow().try_borrow::<WsCaData>().cloned();
-  let user_agent = state.borrow().borrow::<WsUserAgent>().0.clone();
-  let uri: Uri = args.url.parse()?;
-  let mut request = Request::builder().method(Method::GET).uri(&uri);
-
-  request = request.header("User-Agent", user_agent);
-
-  if !args.protocols.is_empty() {
-    request = request.header("Sec-WebSocket-Protocol", args.protocols);
-  }
-
-  let request = request.body(())?;
-  let domain = &uri.host().unwrap().to_string();
-  let port = &uri.port_u16().unwrap_or(match uri.scheme_str() {
-    Some("wss") => 443,
-    Some("ws") => 80,
-    _ => unreachable!(),
-  });
-  let addr = format!("{}:{}", domain, port);
-  let try_socket = TcpStream::connect(addr).await;
-  let tcp_socket = match try_socket.map_err(TungsteniteError::Io) {
-    Ok(socket) => socket,
-    Err(_) => return Ok(json!({"success": false})),
-  };
-
-  let socket: MaybeTlsStream = match uri.scheme_str() {
-    Some("ws") => StreamSwitcher::Plain(tcp_socket),
-    Some("wss") => {
-      let mut config = ClientConfig::new();
-      config
-        .root_store
-        .add_server_trust_anchors(&webpki_roots::TLS_SERVER_ROOTS);
-
-      if let Some(ws_ca_data) = ws_ca_data {
-        let reader = &mut BufReader::new(Cursor::new(ws_ca_data.0));
-        config.root_store.add_pem_file(reader).unwrap();
-      }
-
-      let tls_connector = TlsConnector::from(Arc::new(config));
-      let dnsname =
-        DNSNameRef::try_from_ascii_str(&domain).expect("Invalid DNS lookup");
-      let tls_socket = tls_connector.connect(dnsname, tcp_socket).await?;
-      StreamSwitcher::Tls(tls_socket)
-    }
-    _ => unreachable!(),
-  };
-
-  let (stream, response): (WsStream, Response) =
-    client_async(request, socket).await.map_err(|err| {
-      type_error(format!(
-        "failed to connect to WebSocket: {}",
-        err.to_string()
-      ))
-    })?;
-
-  let (ws_tx, ws_rx) = stream.split();
-  let resource = WsStreamResource {
-    rx: AsyncRefCell::new(ws_rx),
-    tx: AsyncRefCell::new(ws_tx),
-    cancel: Default::default(),
-  };
-  let mut state = state.borrow_mut();
-  let rid = state.resource_table.add(resource);
-
-  let protocol = match response.headers().get("Sec-WebSocket-Protocol") {
-    Some(header) => header.to_str().unwrap(),
-    None => "",
-  };
-  let extensions = response
-    .headers()
-    .get_all("Sec-WebSocket-Extensions")
-    .iter()
-    .map(|header| header.to_str().unwrap())
-    .collect::<String>();
-  Ok(json!({
-    "success": true,
-    "rid": rid,
-    "protocol": protocol,
-    "extensions": extensions
-  }))
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct SendArgs {
-  rid: u32,
-  kind: String,
-  text: Option<String>,
-}
-
-pub async fn op_ws_send(
-  state: Rc<RefCell<OpState>>,
-  args: Value,
-  bufs: BufVec,
-) -> Result<Value, AnyError> {
-  let args: SendArgs = serde_json::from_value(args)?;
-
-  let msg = match args.kind.as_str() {
-    "text" => Message::Text(args.text.unwrap()),
-    "binary" => Message::Binary(bufs[0].to_vec()),
-    "pong" => Message::Pong(vec![]),
-    _ => unreachable!(),
-  };
-  let rid = args.rid;
-
-  let resource = state
-    .borrow_mut()
-    .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?;
-  Ok(json!({}))
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct CloseArgs {
-  rid: u32,
-  code: Option<u16>,
-  reason: Option<String>,
-}
-
-pub async fn op_ws_close(
-  state: Rc<RefCell<OpState>>,
-  args: Value,
-  _bufs: BufVec,
-) -> Result<Value, AnyError> {
-  let args: CloseArgs = serde_json::from_value(args)?;
-  let rid = args.rid;
-  let msg = Message::Close(args.code.map(|c| CloseFrame {
-    code: CloseCode::from(c),
-    reason: match args.reason {
-      Some(reason) => Cow::from(reason),
-      None => Default::default(),
-    },
-  }));
-
-  let resource = state
-    .borrow_mut()
-    .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?;
-  Ok(json!({}))
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct NextEventArgs {
-  rid: u32,
-}
-
-pub async fn op_ws_next_event(
-  state: Rc<RefCell<OpState>>,
-  args: Value,
-  _bufs: BufVec,
-) -> Result<Value, AnyError> {
-  let args: NextEventArgs = serde_json::from_value(args)?;
-
-  let resource = state
-    .borrow_mut()
-    .resource_table
-    .get::<WsStreamResource>(args.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 res = match val {
-    Some(Ok(Message::Text(text))) => json!({
-      "type": "string",
-      "data": text
-    }),
-    Some(Ok(Message::Binary(data))) => {
-      // TODO(ry): don't use json to send binary data.
-      json!({
-        "type": "binary",
-        "data": data
-      })
-    }
-    Some(Ok(Message::Close(Some(frame)))) => json!({
-      "type": "close",
-      "code": u16::from(frame.code),
-      "reason": frame.reason.as_ref()
-    }),
-    Some(Ok(Message::Close(None))) => json!({ "type": "close" }),
-    Some(Ok(Message::Ping(_))) => json!({"type": "ping"}),
-    Some(Ok(Message::Pong(_))) => json!({"type": "pong"}),
-    Some(Err(_)) => json!({"type": "error"}),
-    None => {
-      state.borrow_mut().resource_table.close(args.rid).unwrap();
-      json!({"type": "closed"})
-    }
-  };
-  Ok(res)
-}
diff --git a/runtime/permissions.rs b/runtime/permissions.rs
index 70d243cb0..c50783f9d 100644
--- a/runtime/permissions.rs
+++ b/runtime/permissions.rs
@@ -628,6 +628,12 @@ impl deno_fetch::FetchPermissions for Permissions {
   }
 }
 
+impl deno_websocket::WebSocketPermissions for Permissions {
+  fn check_net_url(&self, url: &url::Url) -> Result<(), AnyError> {
+    Permissions::check_net_url(self, url)
+  }
+}
+
 /// Shows the permission prompt and returns the answer according to the user input.
 /// This loops until the user gives the proper input.
 #[cfg(not(test))]
diff --git a/runtime/web_worker.rs b/runtime/web_worker.rs
index 988845840..313c71177 100644
--- a/runtime/web_worker.rs
+++ b/runtime/web_worker.rs
@@ -237,8 +237,8 @@ impl WebWorker {
       ops::io::init(js_runtime);
       ops::websocket::init(
         js_runtime,
-        options.ca_data.clone(),
         options.user_agent.clone(),
+        options.ca_data.clone(),
       );
 
       if options.use_deno_namespace {
diff --git a/runtime/worker.rs b/runtime/worker.rs
index a05c9f758..9326d632e 100644
--- a/runtime/worker.rs
+++ b/runtime/worker.rs
@@ -143,8 +143,8 @@ impl MainWorker {
       ops::tty::init(js_runtime);
       ops::websocket::init(
         js_runtime,
-        options.ca_data.clone(),
         options.user_agent.clone(),
+        options.ca_data.clone(),
       );
     }
     {