worker: allow synchronously draining message ports

doc/api/worker_threads.md

@@ -49,6 +49,26 @@ The above example spawns a Worker thread for each `parse()` call. In actual
 practice, use a pool of Workers instead for these kinds of tasks. Otherwise, the
 overhead of creating Workers would likely exceed their benefit.
 
+## worker.drainMessagePort(port)
+<!-- YAML
+added: REPLACEME
+-->
+
+* `port` {MessagePort}
+
+Make `port` emit all currently pending messages synchronously.
+
+```js
+const { MessageChannel, drainMessagePort } = require('worker_threads');
+const { port1, port2 } = new MessageChannel();
+
+port1.postMessage({ hello: 'world' });
+port2.once('message', (message) => console.log(message));
+
+drainMessagePort(port2);
+console.log('done emitting messages');  // This is printed *after* the message.
+```
+
 ## worker.isMainThread
 <!-- YAML
 added: v10.5.0

lib/internal/worker.js

@@ -251,6 +251,7 @@ function pipeWithoutWarning(source, dest) {
 }
 
 module.exports = {
+  drainMessagePort,
   ownsProcessState,
   isMainThread,
   threadId,

lib/worker_threads.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  drainMessagePort,
   isMainThread,
   threadId,
   Worker
@@ -13,6 +14,7 @@ const {
 } = require('internal/worker/io');
 
 module.exports = {
+  drainMessagePort,
   isMainThread,
   MessagePort,
   MessageChannel,

test/parallel/test-worker-message-port-drain-manual.js

@@ -0,0 +1,19 @@
+'use strict';
+require('../common');
+const assert = require('assert');
+const { MessageChannel, drainMessagePort } = require('worker_threads');
+
+const { port1, port2 } = new MessageChannel();
+const messages = [];
+port2.on('message', (message) => messages.push(message));
+
+const message = { hello: 'world' };
+
+assert.deepStrictEqual(messages, []);
+port1.postMessage(message);
+port1.postMessage(message);
+assert.deepStrictEqual(messages, []);
+drainMessagePort(port2);
+assert.deepStrictEqual(messages, [ message, message ]);
+
+port1.close();