feat(extensions): Broadcast Channel API

Cargo.toml

@@ -9,6 +9,7 @@ members = [
   "serde_v8",
   "test_plugin",
   "test_util",
+  "extensions/broadcast_channel",
   "extensions/console",
   "extensions/crypto",
   "extensions/fetch",

cli/build.rs

@@ -8,6 +8,7 @@ use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
 use deno_core::JsRuntime;
 use deno_core::RuntimeOptions;
+use deno_runtime::deno_broadcast_channel;
 use deno_runtime::deno_console;
 use deno_runtime::deno_crypto;
 use deno_runtime::deno_fetch;
@@ -74,6 +75,10 @@ fn create_compiler_snapshot(
   op_crate_libs.insert("deno.websocket", deno_websocket::get_declaration());
   op_crate_libs.insert("deno.webstorage", deno_webstorage::get_declaration());
   op_crate_libs.insert("deno.crypto", deno_crypto::get_declaration());
+  op_crate_libs.insert(
+    "deno.broadcast_channel",
+    deno_broadcast_channel::get_declaration(),
+  );
 
   // ensure we invalidate the build properly.
   for (_, path) in op_crate_libs.iter() {
@@ -300,6 +305,10 @@ fn main() {
     "cargo:rustc-env=DENO_CRYPTO_LIB_PATH={}",
     deno_crypto::get_declaration().display()
   );
+  println!(
+    "cargo:rustc-env=DENO_BROADCAST_CHANNEL_LIB_PATH={}",
+    deno_broadcast_channel::get_declaration().display()
+  );
 
   println!("cargo:rustc-env=TARGET={}", env::var("TARGET").unwrap());
   println!("cargo:rustc-env=PROFILE={}", env::var("PROFILE").unwrap());

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

@@ -12,6 +12,7 @@
 /// <reference lib="deno.fetch" />
 /// <reference lib="deno.websocket" />
 /// <reference lib="deno.crypto" />
+/// <reference lib="deno.broadcast_channel" />
 
 declare namespace WebAssembly {
   /**

cli/main.rs

@@ -124,6 +124,7 @@ fn create_web_worker_callback(
       no_color: !colors::use_color(),
       get_error_class_fn: Some(&crate::errors::get_error_class_name),
       blob_url_store: program_state.blob_url_store.clone(),
+      broadcast_channel: program_state.broadcast_channel.clone(),
     };
 
     let mut worker = WebWorker::from_options(
@@ -212,6 +213,7 @@ pub fn create_main_worker(
         .join(checksum::gen(&[loc.to_string().as_bytes()]))
     }),
     blob_url_store: program_state.blob_url_store.clone(),
+    broadcast_channel: program_state.broadcast_channel.clone(),
   };
 
   let mut worker = MainWorker::from_options(main_module, permissions, &options);
@@ -302,7 +304,7 @@ fn print_cache_info(
 
 pub fn get_types(unstable: bool) -> String {
   let mut types = format!(
-    "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
+    "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
     crate::tsc::DENO_NS_LIB,
     crate::tsc::DENO_CONSOLE_LIB,
     crate::tsc::DENO_URL_LIB,
@@ -313,6 +315,7 @@ pub fn get_types(unstable: bool) -> String {
     crate::tsc::DENO_WEBSOCKET_LIB,
     crate::tsc::DENO_WEBSTORAGE_LIB,
     crate::tsc::DENO_CRYPTO_LIB,
+    crate::tsc::DENO_BROADCAST_CHANNEL_LIB,
     crate::tsc::SHARED_GLOBALS_LIB,
     crate::tsc::WINDOW_LIB,
   );

cli/program_state.rs

@@ -15,6 +15,7 @@ use crate::module_graph::TypeLib;
 use crate::source_maps::SourceMapGetter;
 use crate::specifier_handler::FetchHandler;
 use crate::version;
+use deno_runtime::deno_broadcast_channel::InMemoryBroadcastChannel;
 use deno_runtime::deno_file::BlobUrlStore;
 use deno_runtime::inspector::InspectorServer;
 use deno_runtime::permissions::Permissions;
@@ -52,6 +53,7 @@ pub struct ProgramState {
   pub maybe_inspector_server: Option<Arc<InspectorServer>>,
   pub ca_data: Option<Vec<u8>>,
   pub blob_url_store: BlobUrlStore,
+  pub broadcast_channel: InMemoryBroadcastChannel,
 }
 
 impl ProgramState {
@@ -77,6 +79,7 @@ impl ProgramState {
     };
 
     let blob_url_store = BlobUrlStore::default();
+    let broadcast_channel = InMemoryBroadcastChannel::default();
 
     let file_fetcher = FileFetcher::new(
       http_cache,
@@ -143,6 +146,7 @@ impl ProgramState {
       maybe_inspector_server,
       ca_data,
       blob_url_store,
+      broadcast_channel,
     };
     Ok(Arc::new(program_state))
   }

cli/standalone.rs

@@ -15,6 +15,7 @@ use deno_core::v8_set_flags;
 use deno_core::ModuleLoader;
 use deno_core::ModuleSpecifier;
 use deno_core::OpState;
+use deno_runtime::deno_broadcast_channel::InMemoryBroadcastChannel;
 use deno_runtime::deno_file::BlobUrlStore;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::permissions::PermissionsOptions;
@@ -160,6 +161,7 @@ pub async fn run(
   let main_module = resolve_url(SPECIFIER)?;
   let permissions = Permissions::from_options(&metadata.permissions);
   let blob_url_store = BlobUrlStore::default();
+  let broadcast_channel = InMemoryBroadcastChannel::default();
   let module_loader = Rc::new(EmbeddedModuleLoader(source_code));
   let create_web_worker_cb = Arc::new(|_| {
     todo!("Worker are currently not supported in standalone binaries");
@@ -193,6 +195,7 @@ pub async fn run(
     location: metadata.location,
     location_data_dir: None,
     blob_url_store,
+    broadcast_channel,
   };
   let mut worker =
     MainWorker::from_options(main_module.clone(), permissions, &options);

cli/tests/unit/broadcast_channel_test.ts

@@ -0,0 +1,27 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+import { assertEquals } from "../../../test_util/std/testing/asserts.ts";
+import { deferred } from "../../../test_util/std/async/deferred.ts";
+
+Deno.test("broadcastchannel worker", async () => {
+  const intercom = new BroadcastChannel("intercom");
+  let count = 0;
+
+  const url = new URL("../workers/broadcast_channel.ts", import.meta.url);
+  const worker = new Worker(url.href, { type: "module", name: "worker" });
+  worker.onmessage = () => intercom.postMessage(++count);
+
+  const promise = deferred();
+
+  intercom.onmessage = function (e) {
+    assertEquals(count, e.data);
+    if (count < 42) {
+      intercom.postMessage(++count);
+    } else {
+      worker.terminate();
+      intercom.close();
+      promise.resolve();
+    }
+  };
+
+  await promise;
+});

cli/tests/workers/broadcast_channel.ts

@@ -0,0 +1,5 @@
+new BroadcastChannel("intercom").onmessage = function (e) {
+  this.postMessage(e.data);
+};
+
+self.postMessage("go");

cli/tsc.rs

@@ -41,6 +41,8 @@ pub static DENO_WEBSOCKET_LIB: &str =
 pub static DENO_WEBSTORAGE_LIB: &str =
   include_str!(env!("DENO_WEBSTORAGE_LIB_PATH"));
 pub static DENO_CRYPTO_LIB: &str = include_str!(env!("DENO_CRYPTO_LIB_PATH"));
+pub static DENO_BROADCAST_CHANNEL_LIB: &str =
+  include_str!(env!("DENO_BROADCAST_CHANNEL_LIB_PATH"));
 pub static SHARED_GLOBALS_LIB: &str =
   include_str!("dts/lib.deno.shared_globals.d.ts");
 pub static WINDOW_LIB: &str = include_str!("dts/lib.deno.window.d.ts");

extensions/broadcast_channel/01_broadcast_channel.js

@@ -0,0 +1,167 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+"use strict";
+
+((window) => {
+  const core = window.Deno.core;
+  const webidl = window.__bootstrap.webidl;
+  const { setTarget } = window.__bootstrap.event;
+
+  const handlerSymbol = Symbol("eventHandlers");
+  function makeWrappedHandler(handler) {
+    function wrappedHandler(...args) {
+      if (typeof wrappedHandler.handler !== "function") {
+        return;
+      }
+      return wrappedHandler.handler.call(this, ...args);
+    }
+    wrappedHandler.handler = handler;
+    return wrappedHandler;
+  }
+  // TODO(lucacasonato) reuse when we can reuse code between web crates
+  function defineEventHandler(emitter, name) {
+    // HTML specification section 8.1.5.1
+    Object.defineProperty(emitter, `on${name}`, {
+      get() {
+        // TODO(bnoordhuis) The "BroadcastChannel should have an onmessage
+        // event" WPT test expects that .onmessage !== undefined. Returning
+        // null makes it pass but is perhaps not exactly in the spirit.
+        return this[handlerSymbol]?.get(name)?.handler ?? null;
+      },
+      set(value) {
+        if (!this[handlerSymbol]) {
+          this[handlerSymbol] = new Map();
+        }
+        let handlerWrapper = this[handlerSymbol]?.get(name);
+        if (handlerWrapper) {
+          handlerWrapper.handler = value;
+        } else {
+          handlerWrapper = makeWrappedHandler(value);
+          this.addEventListener(name, handlerWrapper);
+        }
+        this[handlerSymbol].set(name, handlerWrapper);
+      },
+      configurable: true,
+      enumerable: true,
+    });
+  }
+
+  const _name = Symbol("[[name]]");
+  const _closed = Symbol("[[closed]]");
+
+  const channels = [];
+  let rid = null;
+
+  async function recv() {
+    while (channels.length > 0) {
+      const message = await core.opAsync("op_broadcast_recv", rid);
+
+      if (message === null) {
+        break;
+      }
+
+      const [name, data] = message;
+      dispatch(null, name, new Uint8Array(data));
+    }
+
+    core.close(rid);
+    rid = null;
+  }
+
+  function dispatch(source, name, data) {
+    for (const channel of channels) {
+      if (channel === source) continue; // Don't self-send.
+      if (channel[_name] !== name) continue;
+      if (channel[_closed]) continue;
+
+      const go = () => {
+        if (channel[_closed]) return;
+        const event = new MessageEvent("message", {
+          data: core.deserialize(data), // TODO(bnoordhuis) Cache immutables.
+          origin: "http://127.0.0.1",
+        });
+        setTarget(event, channel);
+        channel.dispatchEvent(event);
+      };
+
+      defer(go);
+    }
+  }
+
+  // Defer to avoid starving the event loop. Not using queueMicrotask()
+  // for that reason: it lets promises make forward progress but can
+  // still starve other parts of the event loop.
+  function defer(go) {
+    setTimeout(go, 1);
+  }
+
+  class BroadcastChannel extends EventTarget {
+    [_name];
+    [_closed] = false;
+
+    get name() {
+      return this[_name];
+    }
+
+    constructor(name) {
+      super();
+
+      const prefix = "Failed to construct 'broadcastChannel'";
+      webidl.requiredArguments(arguments.length, 1, { prefix });
+
+      this[_name] = webidl.converters["DOMString"](name, {
+        prefix,
+        context: "Argument 1",
+      });
+
+      this[webidl.brand] = webidl.brand;
+
+      channels.push(this);
+
+      if (rid === null) {
+        // Create the rid immediately, otherwise there is a time window (and a
+        // race condition) where messages can get lost, because recv() is async.
+        rid = core.opSync("op_broadcast_subscribe");
+        recv();
+      }
+    }
+
+    postMessage(message) {
+      webidl.assertBranded(this, BroadcastChannel);
+
+      const prefix = "Failed to execute 'postMessage' on 'BroadcastChannel'";
+      webidl.requiredArguments(arguments.length, 1, { prefix });
+
+      if (this[_closed]) {
+        throw new DOMException("Already closed", "InvalidStateError");
+      }
+
+      if (typeof message === "function" || typeof message === "symbol") {
+        throw new DOMException("Uncloneable value", "DataCloneError");
+      }
+
+      const data = core.serialize(message);
+
+      // Send to other listeners in this VM.
+      dispatch(this, this[_name], new Uint8Array(data));
+
+      // Send to listeners in other VMs.
+      defer(() => core.opAsync("op_broadcast_send", [rid, this[_name]], data));
+    }
+
+    close() {
+      webidl.assertBranded(this, BroadcastChannel);
+      this[_closed] = true;
+
+      const index = channels.indexOf(this);
+      if (index === -1) return;
+
+      channels.splice(index, 1);
+      if (channels.length === 0) core.opSync("op_broadcast_unsubscribe", rid);
+    }
+  }
+
+  defineEventHandler(BroadcastChannel.prototype, "message");
+  defineEventHandler(BroadcastChannel.prototype, "messageerror");
+
+  window.__bootstrap.broadcastChannel = { BroadcastChannel };
+})(this);

extensions/broadcast_channel/Cargo.toml

@@ -0,0 +1,20 @@
+# Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+[package]
+name = "deno_broadcast_channel"
+version = "0.1.0"
+edition = "2018"
+description = "Implementation of BroadcastChannel API for Deno"
+authors = ["the Deno authors"]
+license = "MIT"
+readme = "README.md"
+repository = "https://github.com/denoland/deno"
+
+[lib]
+path = "lib.rs"
+
+[dependencies]
+async-trait = "0.1"
+deno_core = { version = "0.88.0", path = "../../core" }
+tokio = { version = "1.4.0", features = ["full"] }
+uuid = { version = "0.8.2", features = ["v4"] }

extensions/broadcast_channel/README.md

@@ -0,0 +1,5 @@
+# deno_broadcast_channel
+
+This crate implements the BroadcastChannel functions of Deno.
+
+Spec: https://html.spec.whatwg.org/multipage/web-messaging.html