test: expand worker_threads parity coverage

test-parity/node-suite/worker_threads/broadcast-channel/close-and-clone.ts

@@ -0,0 +1,45 @@
+import {
+  BroadcastChannel,
+  MessageChannel,
+  receiveMessageOnPort,
+} from "node:worker_threads";
+
+function outcome(fn: () => void): string {
+  try {
+    fn();
+    return "ok";
+  } catch (error: any) {
+    return `${error?.name}:${error?.code ?? ""}`;
+  }
+}
+
+const sender = new BroadcastChannel("clone");
+const receiver = new BroadcastChannel("clone");
+const view = new Uint8Array([3, 1, 4]);
+sender.postMessage(view);
+view[0] = 9;
+
+const packet = receiveMessageOnPort(receiver);
+const cloned = packet ? packet.message : undefined;
+const clonedValues = typeof cloned?.join === "function"
+  ? cloned.join(",")
+  : "not-typed";
+console.log(
+  "typed clone:",
+  cloned instanceof Uint8Array,
+  clonedValues,
+  view.join(","),
+);
+
+const channel = new MessageChannel();
+console.log(
+  "port rejected:",
+  outcome(() => sender.postMessage({ port: channel.port1 })),
+);
+
+sender.close();
+console.log("post closed:", outcome(() => sender.postMessage("late")));
+
+receiver.close();
+channel.port1.close();
+channel.port2.close();

test-parity/node-suite/worker_threads/broadcast-channel/close-and-ref-idempotence.ts

@@ -0,0 +1,15 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const channel = new BroadcastChannel("idempotent");
+console.log("ref return:", channel.ref() === channel);
+console.log("unref return:", channel.unref() === channel);
+console.log("ref again:", channel.ref() === channel);
+console.log("close return:", channel.close());
+console.log("close twice:", channel.close());
+
+try {
+  channel.postMessage("closed");
+  console.log("post closed: ok");
+} catch (error: any) {
+  console.log("post closed:", error?.name, error?.code ?? "");
+}

test-parity/node-suite/worker_threads/broadcast-channel/close-queued-delivery.ts

@@ -0,0 +1,19 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("close-queued-delivery");
+const closed = new BroadcastChannel("close-queued-delivery");
+const barrier = new BroadcastChannel("close-queued-delivery");
+let closedCalls = 0;
+
+sender.unref();
+closed.onmessage = () => {
+  closedCalls += 1;
+};
+barrier.onmessage = (event) => {
+  console.log("barrier:", event.data, "closed calls:", closedCalls);
+  sender.close();
+  barrier.close();
+};
+
+sender.postMessage("queued");
+closed.close();

test-parity/node-suite/worker_threads/broadcast-channel/constructor-and-name.ts

@@ -0,0 +1,20 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+for (
+  const [label, value] of [
+    ["missing", undefined],
+    ["empty", ""],
+    ["number", 42],
+    ["object", { toString: () => "object-name" }],
+  ] as const
+) {
+  try {
+    const channel = value === undefined
+      ? new (BroadcastChannel as any)()
+      : new BroadcastChannel(value as any);
+    console.log(label, "name", channel.name, typeof channel.name);
+    channel.close();
+  } catch (error: any) {
+    console.log(label, error?.name, error?.code ?? "");
+  }
+}

test-parity/node-suite/worker_threads/broadcast-channel/constructor-coercion-edges.ts

@@ -0,0 +1,19 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+for (
+  const [label, value] of [
+    ["null", null],
+    ["bigint", 1n],
+    ["boolean", false],
+    ["infinity", Number.POSITIVE_INFINITY],
+    ["symbol", Symbol("channel")],
+  ] as const
+) {
+  try {
+    const channel = new BroadcastChannel(value as any);
+    console.log(label, "name", channel.name);
+    channel.close();
+  } catch (error: any) {
+    console.log(label, error?.name, error?.code ?? "");
+  }
+}

test-parity/node-suite/worker_threads/broadcast-channel/create-during-dispatch.ts

@@ -0,0 +1,20 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("create-during-dispatch");
+const first = new BroadcastChannel("create-during-dispatch");
+
+sender.unref();
+first.onmessage = (event) => {
+  console.log("first:", event.data);
+  first.close();
+
+  const created = new BroadcastChannel("create-during-dispatch");
+  created.onmessage = (nextEvent) => {
+    console.log("created:", nextEvent.data);
+    created.close();
+    sender.close();
+  };
+  sender.postMessage("second");
+};
+
+sender.postMessage("first");

test-parity/node-suite/worker_threads/broadcast-channel/custom-inspect.ts

@@ -0,0 +1,17 @@
+import { BroadcastChannel } from "node:worker_threads";
+import { inspect } from "node:util";
+
+function outcome(fn: () => unknown): string {
+  try {
+    return String(fn());
+  } catch (error: any) {
+    return `${error?.name}:${error?.code ?? ""}`;
+  }
+}
+
+const channel = new BroadcastChannel("inspect-channel");
+const custom = (channel as any)[inspect.custom];
+console.log("custom:", typeof custom, outcome(() => custom.call({})));
+console.log("active:", inspect(channel), inspect(channel, { depth: -1 }));
+channel.close();
+console.log("closed:", inspect(channel));

test-parity/node-suite/worker_threads/broadcast-channel/error-precedence.ts

@@ -0,0 +1,15 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+function outcome(channel: BroadcastChannel, value: any): string {
+  try {
+    channel.postMessage(value);
+    return "ok";
+  } catch (error: any) {
+    return `${error?.name}:${error?.code ?? ""}`;
+  }
+}
+
+const channel = new BroadcastChannel("error-precedence");
+console.log("open symbol:", outcome(channel, Symbol("value")));
+channel.close();
+console.log("closed symbol:", outcome(channel, Symbol("value")));

test-parity/node-suite/worker_threads/broadcast-channel/fanout-fifo.ts

@@ -0,0 +1,27 @@
+import {
+  BroadcastChannel,
+  receiveMessageOnPort,
+} from "node:worker_threads";
+
+function receive(channel: BroadcastChannel): string {
+  const packet = receiveMessageOnPort(channel);
+  return packet ? JSON.stringify(packet.message) : "empty";
+}
+
+const sender = new BroadcastChannel("fanout");
+const listenerA = new BroadcastChannel("fanout");
+const listenerB = new BroadcastChannel("fanout");
+const unrelated = new BroadcastChannel("other");
+
+sender.postMessage({ index: 1 });
+sender.postMessage({ index: 2 });
+
+console.log("listener a:", receive(listenerA), receive(listenerA));
+console.log("listener b:", receive(listenerB), receive(listenerB));
+console.log("sender excluded:", receive(sender));
+console.log("name isolated:", receive(unrelated));
+
+sender.close();
+listenerA.close();
+listenerB.close();
+unrelated.close();

test-parity/node-suite/worker_threads/broadcast-channel/listener-management.ts

@@ -0,0 +1,22 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("listener-management");
+const receiver = new BroadcastChannel("listener-management");
+let calls = 0;
+
+function listener(event: MessageEvent) {
+  calls += 1;
+  console.log("listener:", calls, event.data);
+  sender.close();
+  receiver.close();
+}
+
+function removed() {
+  console.log("removed listener fired");
+}
+
+receiver.addEventListener("message", listener);
+receiver.addEventListener("message", listener);
+receiver.addEventListener("message", removed);
+receiver.removeEventListener("message", removed);
+sender.postMessage("once");

test-parity/node-suite/worker_threads/broadcast-channel/listener-once.ts

@@ -0,0 +1,28 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("listener-once");
+const receiver = new BroadcastChannel("listener-once");
+const events: string[] = [];
+
+function removed(event: MessageEvent) {
+  events.push(`removed:${event.data}`);
+}
+
+receiver.addEventListener("message", removed);
+receiver.removeEventListener("message", removed);
+receiver.addEventListener(
+  "message",
+  (event) => events.push(`once:${event.data}`),
+  { once: true },
+);
+receiver.addEventListener("message", (event) => {
+  events.push(`regular:${event.data}`);
+  if (event.data === "second") {
+    console.log("events:", events.join(","));
+    sender.close();
+    receiver.close();
+  }
+});
+
+sender.postMessage("first");
+sender.postMessage("second");

test-parity/node-suite/worker_threads/broadcast-channel/messageevent-metadata.ts

@@ -0,0 +1,28 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("event-metadata");
+const receiver = new BroadcastChannel("event-metadata");
+
+receiver.onmessage = (event: MessageEvent) => {
+  console.log(
+    "event:",
+    event instanceof MessageEvent,
+    event.type,
+    event.data,
+    event.target === receiver,
+    event.currentTarget === receiver,
+  );
+  console.log(
+    "defaults:",
+    JSON.stringify({
+      origin: event.origin ?? "missing",
+      lastEventId: event.lastEventId ?? "missing",
+      source: event.source ?? null,
+      ports: event.ports?.length ?? "missing",
+    }),
+  );
+  sender.close();
+  receiver.close();
+};
+
+sender.postMessage("payload");

test-parity/node-suite/worker_threads/broadcast-channel/method-receivers.ts

@@ -0,0 +1,22 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+function outcome(fn: () => any): string {
+  try {
+    fn();
+    return "ok";
+  } catch (error: any) {
+    return `${error?.name}:${error?.code ?? ""}`;
+  }
+}
+
+const prototype = BroadcastChannel.prototype as any;
+const nameGetter = Object.getOwnPropertyDescriptor(prototype, "name")?.get;
+console.log(
+  "name getter:",
+  typeof nameGetter,
+  outcome(() => (nameGetter as any).call({})),
+);
+
+for (const name of ["postMessage", "close", "ref", "unref"] as const) {
+  console.log(`${name}:`, outcome(() => prototype[name].call({})));
+}

test-parity/node-suite/worker_threads/broadcast-channel/onmessage-replacement.ts

@@ -0,0 +1,23 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const sender = new BroadcastChannel("onmessage-replacement");
+const receiver = new BroadcastChannel("onmessage-replacement");
+const events: string[] = [];
+
+receiver.onmessage = (event) => events.push(`first:${event.data}`);
+receiver.onmessage = (event) => {
+  events.push(`second:${event.data}`);
+  receiver.onmessage = null;
+  sender.postMessage("after-null");
+};
+receiver.addEventListener("message", (event) => {
+  events.push(`observer:${event.data}`);
+  if (event.data === "after-null") {
+    console.log("events:", events.join(","));
+    console.log("handler:", receiver.onmessage);
+    sender.close();
+    receiver.close();
+  }
+});
+
+sender.postMessage("value");

test-parity/node-suite/worker_threads/broadcast-channel/post-message-missing.ts

@@ -0,0 +1,10 @@
+import { BroadcastChannel } from "node:worker_threads";
+
+const channel = new BroadcastChannel("missing-message");
+try {
+  (channel.postMessage as any)();
+  console.log("post: ok");
+} catch (error: any) {
+  console.log("post:", error?.name, error?.code ?? "");
+}
+channel.close();

test-parity/node-suite/worker_threads/broadcast-channel/shared-array-buffer.ts

@@ -0,0 +1,24 @@
+import {
+  BroadcastChannel,
+  receiveMessageOnPort,
+} from "node:worker_threads";
+
+const sender = new BroadcastChannel("shared-buffer");
+const receiver = new BroadcastChannel("shared-buffer");
+const source = new SharedArrayBuffer(4);
+const sourceView = new Int32Array(source);
+Atomics.store(sourceView, 0, 5);
+
+sender.postMessage(source);
+const packet = receiveMessageOnPort(receiver);
+const received = packet ? packet.message : undefined;
+const receivedView = received instanceof SharedArrayBuffer
+  ? new Int32Array(received)
+  : undefined;
+
+console.log("brand/identity:", received instanceof SharedArrayBuffer, received === source);
+Atomics.store(sourceView, 0, 27);
+console.log("shared update:", receivedView ? Atomics.load(receivedView, 0) : "missing");
+
+sender.close();
+receiver.close();

test-parity/node-suite/worker_threads/direct-message/handler-error-worker.cjs

@@ -0,0 +1,7 @@
+const { parentPort } = require("node:worker_threads");
+
+process.on("workerMessage", () => {
+  throw new Error("direct-handler-boom");
+});
+parentPort.once("message", () => {});
+parentPort.postMessage("ready");