fix(ext/node): add support for MessagePort.removeListener/off (#23598)

Closes #23564

ext/node/polyfills/worker_threads.ts

@@ -473,6 +473,12 @@ class NodeMessageChannel {
   }
 }
 
+const listeners = new SafeWeakMap<
+  // deno-lint-ignore no-explicit-any
+  (...args: any[]) => void,
+  // deno-lint-ignore no-explicit-any
+  (ev: any) => any
+>();
 function webMessagePortToNodeMessagePort(port: MessagePort) {
   port.on = port.addListener = function (this: MessagePort, name, listener) {
     // deno-lint-ignore no-explicit-any
@@ -486,6 +492,24 @@ function webMessagePortToNodeMessagePort(port: MessagePort) {
     } else {
       throw new Error(`Unknown event: "${name}"`);
     }
+    listeners.set(listener, _listener);
+    return this;
+  };
+  port.off = port.removeListener = function (
+    this: MessagePort,
+    name,
+    listener,
+  ) {
+    if (name == "message") {
+      port.removeEventListener("message", listeners.get(listener)!);
+    } else if (name == "messageerror") {
+      port.removeEventListener("messageerror", listeners.get(listener)!);
+    } else if (name == "close") {
+      port.removeEventListener("close", listeners.get(listener)!);
+    } else {
+      throw new Error(`Unknown event: "${name}"`);
+    }
+    listeners.delete(listener);
     return this;
   };
   port[nodeWorkerThreadCloseCb] = () => {

ext/web/13_message_port.js

@@ -190,6 +190,11 @@ class MessagePort extends EventTarget {
       this[_enabled] = true;
       while (true) {
         if (this[_id] === null) break;
+        // Exit if no message event listeners are present in Node compat mode.
+        if (
+          typeof this[nodeWorkerThreadCloseCb] == "function" &&
+          messageEventListenerCount === 0
+        ) break;
         let data;
         try {
           data = await op_message_port_recv_message(

tests/specs/node/worker_threads/__test__.jsonc

@@ -0,0 +1,8 @@
+{
+  "steps": [{
+    "args": "run message_port_removelistener.mjs",
+    "output": "message_port_removelistener.mjs.out",
+    // Note: successful exit asserts that the test passed
+    "exitCode": 0
+  }]
+}

tests/specs/node/worker_threads/message_port_removelistener.mjs

@@ -0,0 +1,9 @@
+import { MessageChannel } from "node:worker_threads";
+
+const { port1, port2 } = new MessageChannel();
+const listener = (message) => {
+  console.log(message);
+  port1.off("message", listener);
+};
+port1.on("message", listener);
+port2.postMessage("Hello World!");

tests/specs/node/worker_threads/message_port_removelistener.mjs.out

@@ -0,0 +1 @@
+Hello World!