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

Closes https://github.com/denoland/deno/issues/23564

2 files changed, 29 insertions, 0 deletions

diff --git a/ext/node/polyfills/worker_threads.ts b/ext/node/polyfills/worker_threads.ts
index 8f531368a..e88e6368b 100644
--- a/ext/node/polyfills/worker_threads.ts
+++ b/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] = () => {
diff --git a/ext/web/13_message_port.js b/ext/web/13_message_port.js
index 4e4184f2a..93145e8f7 100644
--- a/ext/web/13_message_port.js
+++ b/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(