Repository: denoland/deno
Pull request: fix(ext/node): implement Node worker BroadcastChannel ref/unref

All except one test case in `parallel/test-worker-broadcastchannel-wpt.js` passes. Node.js message dispatch ordering is not same as the Web api as explained in this comment: 
```js
  // Messages are delivered in port creation order.
  // TODO(@jasnell): The ordering here is different than
  // what the browsers would implement due to the different
  // dispatching algorithm under the covers. What's not
  // immediately clear is whether the ordering is spec
  // mandated. In this test, c1 should receive events
  // first, then c2, then c3. In the Node.js dispatching
  // algorithm this means the ordering is:
  //    from c3    (c1 from c3)
  //    done       (c1 from c2)
  //    from c1    (c2 from c1)
  //    from c3    (c2 from c3)
  //    from c1    (c3 from c1)
  //    done       (c3 from c2)
  //
  // Whereas in the browser-ordering (as illustrated in the
  // Web Platform Tests) it would be:
  //    from c1    (c2 from c1)
  //    from c1    (c3 from c1)
  //    from c3    (c1 from c3)
  //    from c3    (c2 from c3)
  //    done       (c1 from c2)
  //    done       (c3 from c2)
```
https://github.com/nodejs/node/blob/f77a709e128334e04278d3b834c1e07bc4686654/test/parallel/test-worker-broadcastchannel-wpt.js#L24-L47

Changed files:
filename: ext/node/polyfills/worker_threads.ts
patch: @@ -26,7 +26,10 @@ import {
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 import { notImplemented } from "ext:deno_node/_utils.ts";
 import { EventEmitter } from "node:events";
-import { BroadcastChannel } from "ext:deno_web/01_broadcast_channel.js";
+import {
+  BroadcastChannel as WebBroadcastChannel,
+  refBroadcastChannel,
+} from "ext:deno_web/01_broadcast_channel.js";
 import { untransferableSymbol } from "ext:deno_node/internal_binding/util.ts";
 import process from "node:process";
 import { createRequire } from "node:module";
@@ -662,6 +665,18 @@ function patchMessagePortIfFound(data: any, seen = new SafeSet<any>()) {
   }
 }
 
+class BroadcastChannel extends WebBroadcastChannel {
+  ref() {
+    this[refBroadcastChannel](true);
+    return this;
+  }
+
+  unref() {
+    this[refBroadcastChannel](false);
+    return this;
+  }
+}
+
 export {
   BroadcastChannel,
   MessagePort,

filename: ext/web/01_broadcast_channel.js
patch: @@ -32,13 +32,22 @@ import { DOMException } from "ext:deno_web/01_dom_exception.js";
 
 const _name = Symbol("[[name]]");
 const _closed = Symbol("[[closed]]");
+const _refed = Symbol("[[refed]]");
+const refBroadcastChannel = Symbol("refBroadcastChannel");
 
 const channels = [];
 let rid = null;
+let recvPromise = null;
+let refedBroadcastChannelsCount = 0;
 
 async function recv() {
   while (channels.length > 0) {
-    const message = await op_broadcast_recv(rid);
+    recvPromise = op_broadcast_recv(rid);
+    if (refedBroadcastChannelsCount === 0) {
+      core.unrefOpPromise(recvPromise);
+    }
+    const message = await recvPromise;
+    recvPromise = null;
 
     if (message === null) {
       break;
@@ -77,6 +86,7 @@ function dispatch(source, name, data) {
 class BroadcastChannel extends EventTarget {
   [_name];
   [_closed] = false;
+  [_refed] = true;
 
   get name() {
     return this[_name];
@@ -93,6 +103,7 @@ class BroadcastChannel extends EventTarget {
     this[webidl.brand] = webidl.brand;
 
     ArrayPrototypePush(channels, this);
+    refedBroadcastChannelsCount++;
 
     if (rid === null) {
       // Create the rid immediately, otherwise there is a time window (and a
@@ -129,15 +140,39 @@ class BroadcastChannel extends EventTarget {
     });
   }
 
+  [refBroadcastChannel](ref) {
+    if (ref && !this[_refed] && !this[_closed]) {
+      refedBroadcastChannelsCount++;
+      if (refedBroadcastChannelsCount === 1 && recvPromise) {
+        core.refOpPromise(recvPromise);
+      }
+      this[_refed] = true;
+    } else if (!ref && this[_refed] && !this[_closed]) {
+      refedBroadcastChannelsCount--;
+      if (refedBroadcastChannelsCount === 0 && recvPromise) {
+        core.unrefOpPromise(recvPromise);
+      }
+      this[_refed] = false;
+    }
+  }
+
   close() {
     webidl.assertBranded(this, BroadcastChannelPrototype);
     this[_closed] = true;
 
     const index = ArrayPrototypeIndexOf(channels, this);
     if (index === -1) return;
 
+    if (this[_refed]) {
+      refedBroadcastChannelsCount--;
+      if (refedBroadcastChannelsCount === 0 && recvPromise) {
+        core.unrefOpPromise(recvPromise);
+      }
+      this[_refed] = false;
+    }
+
     ArrayPrototypeSplice(channels, index, 1);
-    if (channels.length === 0) {
+    if (channels.length === 0 && rid !== null) {
       op_broadcast_unsubscribe(rid);
     }
   }
@@ -162,4 +197,4 @@ defineEventHandler(BroadcastChannel.prototype, "message");
 defineEventHandler(BroadcastChannel.prototype, "messageerror");
 const BroadcastChannelPrototype = BroadcastChannel.prototype;
 
-export { BroadcastChannel };
+export { BroadcastChannel, refBroadcastChannel };

filename: tests/node_compat/config.jsonc
patch: @@ -1536,6 +1536,9 @@
     "parallel/test-whatwg-url-custom-setters.js": {},
     "parallel/test-whatwg-url-custom-tostringtag.js": {},
     "parallel/test-whatwg-url-override-hostname.js": {},
+    // Note: Node.js BroadcastChannel message dispatch ordering differs from the Web spec ordering.
+    // Deno follows the Web spec.
+    // "parallel/test-worker-broadcastchannel-wpt.js": {},
     "parallel/test-worker-cjs-workerdata.js": {},
     "parallel/test-worker-cleanexit-with-js.js": {},
     // TODO(bartlomieju): disabled during work on `node:inspector`, this test didn't actualy run before

filename: tests/unit_node/worker_threads_test.ts
patch: @@ -13,7 +13,11 @@ import { EventEmitter, once } from "node:events";
 import process from "node:process";
 
 Deno.test("[node/worker_threads] BroadcastChannel is exported", () => {
-  assertEquals<unknown>(workerThreads.BroadcastChannel, BroadcastChannel);
+  const bc = new workerThreads.BroadcastChannel("test");
+  assert(bc instanceof BroadcastChannel);
+  assert(typeof bc.ref === "function");
+  assert(typeof bc.unref === "function");
+  bc.close();
 });
 
 Deno.test("[node/worker_threads] MessageChannel are MessagePort are exported", () => {