worker: implement `MessagePort` and `MessageChannel`

Implement `MessagePort` and `MessageChannel` along the lines of
the DOM classes of the same names. `MessagePort`s initially
support transferring only `ArrayBuffer`s.

Thanks to Stephen Belanger for reviewing this change in its
original form, to Benjamin Gruenbaum for reviewing the
added tests in their original form, and to Olivia Hugger
for reviewing the documentation in its original form.

Refs: https://github.com/ayojs/ayo/pull/98

PR-URL: https://github.com/nodejs/node/pull/20876
Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Shingo Inoue <leko.noor@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com>
Reviewed-By: John-David Dalton <john.david.dalton@gmail.com>
Reviewed-By: Gus Caplan <me@gus.host>

4 files changed, 104 insertions, 0 deletions

diff --git a/test/parallel/test-message-channel.js b/test/parallel/test-message-channel.js
new file mode 100644
index 0000000000..0facaa1d83
--- /dev/null
+++ b/test/parallel/test-message-channel.js
@@ -0,0 +1,26 @@
+// Flags: --experimental-worker
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { MessageChannel } = require('worker');
+
+{
+  const channel = new MessageChannel();
+
+  channel.port1.on('message', common.mustCall(({ typedArray }) => {
+    assert.deepStrictEqual(typedArray, new Uint8Array([0, 1, 2, 3, 4]));
+  }));
+
+  const typedArray = new Uint8Array([0, 1, 2, 3, 4]);
+  channel.port2.postMessage({ typedArray }, [ typedArray.buffer ]);
+  assert.strictEqual(typedArray.buffer.byteLength, 0);
+  channel.port2.close();
+}
+
+{
+  const channel = new MessageChannel();
+
+  channel.port1.on('close', common.mustCall());
+  channel.port2.on('close', common.mustCall());
+  channel.port2.close();
+}
diff --git a/test/parallel/test-message-port-arraybuffer.js b/test/parallel/test-message-port-arraybuffer.js
new file mode 100644
index 0000000000..4abeb585b4
--- /dev/null
+++ b/test/parallel/test-message-port-arraybuffer.js
@@ -0,0 +1,20 @@
+// Flags: --experimental-worker
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const { MessageChannel } = require('worker');
+
+{
+  const { port1, port2 } = new MessageChannel();
+
+  const arrayBuffer = new ArrayBuffer(40);
+  const typedArray = new Uint32Array(arrayBuffer);
+  typedArray[0] = 0x12345678;
+
+  port1.postMessage(typedArray, [ arrayBuffer ]);
+  port2.on('message', common.mustCall((received) => {
+    assert.strictEqual(received[0], 0x12345678);
+    port2.close(common.mustCall());
+  }));
+}
diff --git a/test/parallel/test-message-port.js b/test/parallel/test-message-port.js
new file mode 100644
index 0000000000..8a7f380520
--- /dev/null
+++ b/test/parallel/test-message-port.js
@@ -0,0 +1,56 @@
+// Flags: --experimental-worker
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const { MessageChannel, MessagePort } = require('worker');
+
+{
+  const { port1, port2 } = new MessageChannel();
+  assert(port1 instanceof MessagePort);
+  assert(port2 instanceof MessagePort);
+
+  const input = { a: 1 };
+  port1.postMessage(input);
+  port2.on('message', common.mustCall((received) => {
+    assert.deepStrictEqual(received, input);
+    port2.close(common.mustCall());
+  }));
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+
+  const input = { a: 1 };
+  port1.postMessage(input);
+  // Check that the message still gets delivered if `port2` has its
+  // `on('message')` handler attached at a later point in time.
+  setImmediate(() => {
+    port2.on('message', common.mustCall((received) => {
+      assert.deepStrictEqual(received, input);
+      port2.close(common.mustCall());
+    }));
+  });
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+
+  const input = { a: 1 };
+
+  const dummy = common.mustNotCall();
+  // Check that the message still gets delivered if `port2` has its
+  // `on('message')` handler attached at a later point in time, even if a
+  // listener was removed previously.
+  port2.addListener('message', dummy);
+  setImmediate(() => {
+    port2.removeListener('message', dummy);
+    port1.postMessage(input);
+    setImmediate(() => {
+      port2.on('message', common.mustCall((received) => {
+        assert.deepStrictEqual(received, input);
+        port2.close(common.mustCall());
+      }));
+    });
+  });
+}
diff --git a/test/sequential/test-async-wrap-getasyncid.js b/test/sequential/test-async-wrap-getasyncid.js
index 971296915c..84a3e3b1f4 100644
--- a/test/sequential/test-async-wrap-getasyncid.js
+++ b/test/sequential/test-async-wrap-getasyncid.js
@@ -35,7 +35,9 @@ common.crashOnUnhandledRejection();
     delete providers.HTTP2STREAM;
     delete providers.HTTP2PING;
     delete providers.HTTP2SETTINGS;
+    // TODO(addaleax): Test for these
     delete providers.STREAMPIPE;
+    delete providers.MESSAGEPORT;
 
     const objKeys = Object.keys(providers);
     if (objKeys.length > 0)