fix(ext/node): MessagePort works (#22999)

Closes https://github.com/denoland/deno/issues/22951
Closes https://github.com/denoland/deno/issues/23001

Co-authored-by: Divy Srivastava <dj.srivastava23@gmail.com>

8 files changed, 95 insertions, 1 deletions

diff --git a/tests/integration/worker_tests.rs b/tests/integration/worker_tests.rs
index dd0c2d409..abd07da16 100644
--- a/tests/integration/worker_tests.rs
+++ b/tests/integration/worker_tests.rs
@@ -1,6 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
 use test_util::itest;
+use test_util::itest_flaky;
 
 itest!(worker_error {
   args: "run -A workers/worker_error.ts",
@@ -125,3 +126,15 @@ itest!(node_worker_auto_exits {
   output: "workers/node_worker_auto_exits.mjs.out",
   exit_code: 0,
 });
+
+itest_flaky!(node_worker_message_port {
+  args: "run --quiet --allow-read workers/node_worker_message_port.mjs",
+  output: "workers/node_worker_message_port.mjs.out",
+  exit_code: 0,
+});
+
+itest!(node_worker_transfer_port {
+  args: "run --quiet --allow-read workers/node_worker_transfer_port.mjs",
+  output: "workers/node_worker_transfer_port.mjs.out",
+  exit_code: 0,
+});
diff --git a/tests/testdata/workers/node_worker_message_port.mjs b/tests/testdata/workers/node_worker_message_port.mjs
new file mode 100644
index 000000000..71640fb40
--- /dev/null
+++ b/tests/testdata/workers/node_worker_message_port.mjs
@@ -0,0 +1,41 @@
+import workerThreads from "node:worker_threads";
+
+const { port1: mainPort, port2: workerPort } = new workerThreads
+  .MessageChannel();
+
+// Note: not using Promise.withResolver() because it's not available in Node.js
+const deferred = createDeferred();
+
+const worker = new workerThreads.Worker(
+  import.meta.resolve("./node_worker_message_port_1.cjs"),
+  {
+    workerData: { workerPort },
+    transferList: [workerPort],
+  },
+);
+
+worker.on("message", (data) => {
+  console.log("worker:", data);
+  mainPort.on("message", (msg) => {
+    console.log("mainPort:", msg);
+    deferred.resolve();
+  });
+  mainPort.on("close", (_msg) => {
+    console.log("mainPort closed");
+  });
+});
+
+worker.postMessage("Hello from parent");
+await deferred.promise;
+await worker.terminate();
+mainPort.close();
+
+function createDeferred() {
+  let resolveCallback;
+  let rejectCallback;
+  const promise = new Promise((resolve, reject) => {
+    resolveCallback = resolve;
+    rejectCallback = reject;
+  });
+  return { promise, resolve: resolveCallback, reject: rejectCallback };
+}
diff --git a/tests/testdata/workers/node_worker_message_port.mjs.out b/tests/testdata/workers/node_worker_message_port.mjs.out
new file mode 100644
index 000000000..5317b65a0
--- /dev/null
+++ b/tests/testdata/workers/node_worker_message_port.mjs.out
@@ -0,0 +1,4 @@
+worker port closed
+worker: Hello from worker on parentPort!
+mainPort: Hello from worker on workerPort!
+mainPort closed
diff --git a/tests/testdata/workers/node_worker_message_port_1.cjs b/tests/testdata/workers/node_worker_message_port_1.cjs
new file mode 100644
index 000000000..01739c51e
--- /dev/null
+++ b/tests/testdata/workers/node_worker_message_port_1.cjs
@@ -0,0 +1,9 @@
+const { parentPort, workerData } = require("worker_threads");
+
+parentPort.on("message", (msg) => {
+  const workerPort = workerData.workerPort;
+  parentPort.postMessage("Hello from worker on parentPort!");
+  workerPort.postMessage("Hello from worker on workerPort!");
+  workerPort.on("close", () => console.log("worker port closed"));
+  workerPort.close();
+});
diff --git a/tests/testdata/workers/node_worker_transfer_port.mjs b/tests/testdata/workers/node_worker_transfer_port.mjs
new file mode 100644
index 000000000..1b17ed1ab
--- /dev/null
+++ b/tests/testdata/workers/node_worker_transfer_port.mjs
@@ -0,0 +1,14 @@
+import { MessageChannel, Worker } from "node:worker_threads";
+
+const { port1, port2 } = new MessageChannel();
+const worker = new Worker(
+  import.meta.resolve("./node_worker_transfer_port_1.mjs"),
+);
+// Send the port directly after the worker is created
+worker.postMessage(port2, [port2]);
+// Send a message to the worker using the transferred port
+port1.postMessage("Hello from main thread!");
+worker.on("message", (message) => {
+  console.log("Received message from worker:", message);
+  worker.terminate();
+});
diff --git a/tests/testdata/workers/node_worker_transfer_port.mjs.out b/tests/testdata/workers/node_worker_transfer_port.mjs.out
new file mode 100644
index 000000000..8e8f11940
--- /dev/null
+++ b/tests/testdata/workers/node_worker_transfer_port.mjs.out
@@ -0,0 +1,3 @@
+Worker thread started!
+Received message from main thread: Hello from main thread!
+Received message from worker: Reply from worker
diff --git a/tests/testdata/workers/node_worker_transfer_port_1.mjs b/tests/testdata/workers/node_worker_transfer_port_1.mjs
new file mode 100644
index 000000000..4d0a38bd5
--- /dev/null
+++ b/tests/testdata/workers/node_worker_transfer_port_1.mjs
@@ -0,0 +1,10 @@
+import { parentPort } from "node:worker_threads";
+
+parentPort.on("message", (message) => {
+  const transferredPort = message;
+  transferredPort.on("message", (message) => {
+    console.log("Received message from main thread:", message);
+    parentPort.postMessage("Reply from worker");
+  });
+  console.log("Worker thread started!");
+});
diff --git a/tests/unit_node/worker_threads_test.ts b/tests/unit_node/worker_threads_test.ts
index 52c9cfffd..2351e1052 100644
--- a/tests/unit_node/worker_threads_test.ts
+++ b/tests/unit_node/worker_threads_test.ts
@@ -16,7 +16,7 @@ Deno.test("[node/worker_threads] BroadcastChannel is exported", () => {
 });
 
 Deno.test("[node/worker_threads] MessageChannel are MessagePort are exported", () => {
-  assertEquals<unknown>(workerThreads.MessageChannel, MessageChannel);
+  assert(workerThreads.MessageChannel);
   assertEquals<unknown>(workerThreads.MessagePort, MessagePort);
 });