fix(ext/node): support multiple message listeners on MessagePort

ext/node/polyfills/worker_threads.ts

@@ -484,9 +484,17 @@ function webMessagePortToNodeMessagePort(port: MessagePort) {
     // deno-lint-ignore no-explicit-any
     const _listener = (ev: any) => listener(ev.data);
     if (name == "message") {
-      port.onmessage = _listener;
+      if (port.onmessage === null) {
+        port.onmessage = _listener;
+      } else {
+        port.addEventListener("message", _listener);
+      }
     } else if (name == "messageerror") {
-      port.onmessageerror = _listener;
+      if (port.onmessageerror === null) {
+        port.onmessageerror = _listener;
+      } else {
+        port.addEventListener("messageerror", _listener);
+      }
     } else if (name == "close") {
       port.addEventListener("close", _listener);
     } else {

tests/unit_node/worker_threads_test.ts

@@ -515,3 +515,24 @@ Deno.test({
     await worker.terminate();
   },
 });
+
+Deno.test({
+  name:
+    "[node/worker_threads] MessagePort.on all message listeners are invoked",
+  async fn() {
+    const output: string[] = [];
+    const deferred = Promise.withResolvers<void>();
+    const { port1, port2 } = new workerThreads.MessageChannel();
+    port1.on("message", (msg) => output.push(msg));
+    port1.on("message", (msg) => output.push(msg + 2));
+    port1.on("message", (msg) => {
+      output.push(msg + 3);
+      deferred.resolve();
+    });
+    port2.postMessage("hi!");
+    await deferred.promise;
+    assertEquals(output, ["hi!", "hi!2", "hi!3"]);
+    port2.close();
+    port1.close();
+  },
+});