diff --git a/Cargo.lock b/Cargo.lock
index c46c6ac9fb..884f890afa 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -595,8 +595,10 @@ dependencies = [
name = "deno_broadcast_channel"
version = "0.1.0"
dependencies = [
+ "async-trait",
"deno_core",
"tokio",
+ "uuid",
]
[[package]]
diff --git a/cli/dts/lib.deno.shared_globals.d.ts b/cli/dts/lib.deno.shared_globals.d.ts
index d3784705eb..e35de2e774 100644
--- a/cli/dts/lib.deno.shared_globals.d.ts
+++ b/cli/dts/lib.deno.shared_globals.d.ts
@@ -12,6 +12,7 @@
///
///
///
+///
declare namespace WebAssembly {
/**
diff --git a/cli/main.rs b/cli/main.rs
index 868805e926..60c202a7ad 100644
--- a/cli/main.rs
+++ b/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);
diff --git a/cli/program_state.rs b/cli/program_state.rs
index 50890b9e48..9f7ddc7494 100644
--- a/cli/program_state.rs
+++ b/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>,
pub ca_data: Option>,
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))
}
diff --git a/cli/standalone.rs b/cli/standalone.rs
index e0b131eb8f..f281c53364 100644
--- a/cli/standalone.rs
+++ b/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);
diff --git a/cli/tests/unit/broadcast_channel_test.ts b/cli/tests/unit/broadcast_channel_test.ts
new file mode 100644
index 0000000000..cfa62c8561
--- /dev/null
+++ b/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;
+});
diff --git a/cli/tests/workers/broadcast_channel.ts b/cli/tests/workers/broadcast_channel.ts
new file mode 100644
index 0000000000..5076e9eb72
--- /dev/null
+++ b/cli/tests/workers/broadcast_channel.ts
@@ -0,0 +1,5 @@
+new BroadcastChannel("intercom").onmessage = function (e) {
+ this.postMessage(e.data);
+};
+
+self.postMessage("go");
diff --git a/extensions/broadcast_channel/01_broadcast_channel.js b/extensions/broadcast_channel/01_broadcast_channel.js
index 34f8b9e196..7670b0cfd0 100644
--- a/extensions/broadcast_channel/01_broadcast_channel.js
+++ b/extensions/broadcast_channel/01_broadcast_channel.js
@@ -4,6 +4,7 @@
((window) => {
const core = window.Deno.core;
const webidl = window.__bootstrap.webidl;
+ const { setTarget } = window.__bootstrap.event;
const handlerSymbol = Symbol("eventHandlers");
function makeWrappedHandler(handler) {
@@ -21,7 +22,10 @@
// HTML specification section 8.1.5.1
Object.defineProperty(emitter, `on${name}`, {
get() {
- return this[handlerSymbol]?.get(name)?.handler;
+ // 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]) {
@@ -43,12 +47,56 @@
const _name = Symbol("[[name]]");
const _closed = Symbol("[[closed]]");
- const _rid = Symbol("[[rid]]");
+
+ 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;
- [_rid];
get name() {
return this[_name];
@@ -57,8 +105,6 @@
constructor(name) {
super();
- window.location;
-
const prefix = "Failed to construct 'broadcastChannel'";
webidl.requiredArguments(arguments.length, 1, { prefix });
@@ -67,46 +113,50 @@
context: "Argument 1",
});
- this[_rid] = core.opSync("op_broadcast_open", this[_name]);
-
this[webidl.brand] = webidl.brand;
- this.#eventLoop();
+ 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");
}
- core.opAsync("op_broadcast_send", this[_rid], core.serialize(message));
+ 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;
- core.close(this[_rid]);
- }
- async #eventLoop() {
- while (!this[_closed]) {
- const message = await core.opAsync(
- "op_broadcast_next_event",
- this[_rid],
- );
+ const index = channels.indexOf(this);
+ if (index === -1) return;
- if (message.length !== 0) {
- const event = new MessageEvent("message", {
- data: core.deserialize(message),
- origin: window.location,
- });
- event.target = this;
- this.dispatchEvent(event);
- }
- }
+ channels.splice(index, 1);
+ if (channels.length === 0) core.opSync("op_broadcast_unsubscribe", rid);
}
}
diff --git a/extensions/broadcast_channel/Cargo.toml b/extensions/broadcast_channel/Cargo.toml
index 72c29f651f..7bc65f3a0c 100644
--- a/extensions/broadcast_channel/Cargo.toml
+++ b/extensions/broadcast_channel/Cargo.toml
@@ -14,5 +14,7 @@ repository = "https://github.com/denoland/deno"
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"] }
diff --git a/extensions/broadcast_channel/in_memory_broadcast_channel.rs b/extensions/broadcast_channel/in_memory_broadcast_channel.rs
new file mode 100644
index 0000000000..34498c8302
--- /dev/null
+++ b/extensions/broadcast_channel/in_memory_broadcast_channel.rs
@@ -0,0 +1,97 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+use crate::BroadcastChannel;
+use async_trait::async_trait;
+use deno_core::error::AnyError;
+use std::sync::Arc;
+use std::sync::Mutex;
+use tokio::sync::broadcast;
+use tokio::sync::mpsc;
+use uuid::Uuid;
+
+#[derive(Clone)]
+pub struct InMemoryBroadcastChannel(Arc>>);
+
+pub struct InMemoryBroadcastChannelResource {
+ rx: tokio::sync::Mutex<(
+ broadcast::Receiver,
+ mpsc::UnboundedReceiver<()>,
+ )>,
+ cancel_tx: mpsc::UnboundedSender<()>,
+ uuid: Uuid,
+}
+
+#[derive(Clone, Debug)]
+struct Message {
+ name: Arc,
+ data: Arc>,
+ uuid: Uuid,
+}
+
+impl Default for InMemoryBroadcastChannel {
+ fn default() -> Self {
+ let (tx, _) = broadcast::channel(256);
+ Self(Arc::new(Mutex::new(tx)))
+ }
+}
+
+#[async_trait]
+impl BroadcastChannel for InMemoryBroadcastChannel {
+ type Resource = InMemoryBroadcastChannelResource;
+
+ fn subscribe(&self) -> Result {
+ let (cancel_tx, cancel_rx) = mpsc::unbounded_channel();
+ let broadcast_rx = self.0.lock().unwrap().subscribe();
+ let rx = tokio::sync::Mutex::new((broadcast_rx, cancel_rx));
+ let uuid = Uuid::new_v4();
+ Ok(Self::Resource {
+ rx,
+ cancel_tx,
+ uuid,
+ })
+ }
+
+ fn unsubscribe(&self, resource: &Self::Resource) -> Result<(), AnyError> {
+ Ok(resource.cancel_tx.send(())?)
+ }
+
+ async fn send(
+ &self,
+ resource: &Self::Resource,
+ name: String,
+ data: Vec,
+ ) -> Result<(), AnyError> {
+ let name = Arc::new(name);
+ let data = Arc::new(data);
+ let uuid = resource.uuid;
+ self.0.lock().unwrap().send(Message { name, data, uuid })?;
+ Ok(())
+ }
+
+ async fn recv(
+ &self,
+ resource: &Self::Resource,
+ ) -> Result