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, 117 insertions, 1 deletions

diff --git a/lib/internal/bootstrap/loaders.js b/lib/internal/bootstrap/loaders.js
index ff809a9129..417e8594e1 100644
--- a/lib/internal/bootstrap/loaders.js
+++ b/lib/internal/bootstrap/loaders.js
@@ -194,7 +194,8 @@
     };
 
     NativeModule.isInternal = function(id) {
-      return id.startsWith('internal/');
+      return id.startsWith('internal/') ||
+          (id === 'worker' && !process.binding('config').experimentalWorker);
     };
   }
 
diff --git a/lib/internal/modules/cjs/helpers.js b/lib/internal/modules/cjs/helpers.js
index 60346c5841..55eaed7d37 100644
--- a/lib/internal/modules/cjs/helpers.js
+++ b/lib/internal/modules/cjs/helpers.js
@@ -105,6 +105,11 @@ const builtinLibs = [
   'v8', 'vm', 'zlib'
 ];
 
+if (process.binding('config').experimentalWorker) {
+  builtinLibs.push('worker');
+  builtinLibs.sort();
+}
+
 if (typeof process.binding('inspector').open === 'function') {
   builtinLibs.push('inspector');
   builtinLibs.sort();
diff --git a/lib/internal/worker.js b/lib/internal/worker.js
new file mode 100644
index 0000000000..73f7525aa7
--- /dev/null
+++ b/lib/internal/worker.js
@@ -0,0 +1,105 @@
+'use strict';
+
+const EventEmitter = require('events');
+const util = require('util');
+
+const { internalBinding } = require('internal/bootstrap/loaders');
+const { MessagePort, MessageChannel } = internalBinding('messaging');
+const { handle_onclose } = internalBinding('symbols');
+
+util.inherits(MessagePort, EventEmitter);
+
+const kOnMessageListener = Symbol('kOnMessageListener');
+
+const debug = util.debuglog('worker');
+
+// A MessagePort consists of a handle (that wraps around an
+// uv_async_t) which can receive information from other threads and emits
+// .onmessage events, and a function used for sending data to a MessagePort
+// in some other thread.
+MessagePort.prototype[kOnMessageListener] = function onmessage(payload) {
+  debug('received message', payload);
+  // Emit the deserialized object to userland.
+  this.emit('message', payload);
+};
+
+// This is for compatibility with the Web's MessagePort API. It makes sense to
+// provide it as an `EventEmitter` in Node.js, but if somebody overrides
+// `onmessage`, we'll switch over to the Web API model.
+Object.defineProperty(MessagePort.prototype, 'onmessage', {
+  enumerable: true,
+  configurable: true,
+  get() {
+    return this[kOnMessageListener];
+  },
+  set(value) {
+    this[kOnMessageListener] = value;
+    if (typeof value === 'function') {
+      this.ref();
+      this.start();
+    } else {
+      this.unref();
+      this.stop();
+    }
+  }
+});
+
+// This is called from inside the `MessagePort` constructor.
+function oninit() {
+  setupPortReferencing(this, this, 'message');
+}
+
+Object.defineProperty(MessagePort.prototype, 'oninit', {
+  enumerable: true,
+  writable: false,
+  value: oninit
+});
+
+// This is called after the underlying `uv_async_t` has been closed.
+function onclose() {
+  if (typeof this.onclose === 'function') {
+    // Not part of the Web standard yet, but there aren't many reasonable
+    // alternatives in a non-EventEmitter usage setting.
+    // Refs: https://github.com/whatwg/html/issues/1766
+    this.onclose();
+  }
+  this.emit('close');
+}
+
+Object.defineProperty(MessagePort.prototype, handle_onclose, {
+  enumerable: false,
+  writable: false,
+  value: onclose
+});
+
+const originalClose = MessagePort.prototype.close;
+MessagePort.prototype.close = function(cb) {
+  if (typeof cb === 'function')
+    this.once('close', cb);
+  originalClose.call(this);
+};
+
+function setupPortReferencing(port, eventEmitter, eventName) {
+  // Keep track of whether there are any workerMessage listeners:
+  // If there are some, ref() the channel so it keeps the event loop alive.
+  // If there are none or all are removed, unref() the channel so the worker
+  // can shutdown gracefully.
+  port.unref();
+  eventEmitter.on('newListener', (name) => {
+    if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {
+      port.ref();
+      port.start();
+    }
+  });
+  eventEmitter.on('removeListener', (name) => {
+    if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {
+      port.stop();
+      port.unref();
+    }
+  });
+}
+
+module.exports = {
+  MessagePort,
+  MessageChannel
+};
diff --git a/lib/worker.js b/lib/worker.js
new file mode 100644
index 0000000000..d67fb4efe4
--- /dev/null
+++ b/lib/worker.js
@@ -0,0 +1,5 @@
+'use strict';
+
+const { MessagePort, MessageChannel } = require('internal/worker');
+
+module.exports = { MessagePort, MessageChannel };