lib: make diagnostics_channel async iterable

Author: Qard

lib/diagnostics_channel.js

@@ -3,10 +3,14 @@
 const {
   ArrayPrototypeIndexOf,
   ArrayPrototypePush,
+  ArrayPrototypeShift,
   ArrayPrototypeSplice,
   ObjectCreate,
   ObjectGetPrototypeOf,
   ObjectSetPrototypeOf,
+  Promise,
+  PromiseResolve,
+  SymbolAsyncIterator,
   SymbolHasInstance,
   WeakRefPrototypeGet
 } = primordials;
@@ -21,8 +25,72 @@ const { triggerUncaughtException } = internalBinding('errors');
 
 const { WeakReference } = internalBinding('util');
 
+class AsyncIterableChannel {
+  constructor(channel) {
+    this.channel = channel;
+    this.events = [];
+    this.waiting = [];
+
+    this.subscriber = (message) => {
+      const resolve = ArrayPrototypeShift(this.waiting);
+      if (resolve) {
+        return resolve({
+          value: message,
+          done: false
+        });
+      }
+
+      ArrayPrototypePush(this.events, message);
+    };
+
+    channel.subscribe(this.subscriber);
+  }
+
+  [SymbolAsyncIterator]() {
+    return this;
+  }
+
+  return() {
+    const data = { done: true };
+    this.done = true;
+
+    this.channel.unsubscribe(this.subscriber);
+
+    for (let i = 0; i < this.waiting.length; i++) {
+      const resolve = this.waiting[i];
+      resolve(data);
+    }
+
+    return PromiseResolve(data);
+  }
+
+  next() {
+    const event = ArrayPrototypeShift(this.events);
+    if (event) {
+      return PromiseResolve({
+        value: event,
+        done: false
+      });
+    }
+
+    if (this.done) {
+      return PromiseResolve({
+        done: true
+      });
+    }
+
+    return new Promise((resolve) => {
+      ArrayPrototypePush(this.waiting, resolve);
+    });
+  }
+}
+
 // TODO(qard): should there be a C++ channel interface?
 class ActiveChannel {
+  [SymbolAsyncIterator]() {
+    return new AsyncIterableChannel(this);
+  }
+
   subscribe(subscription) {
     if (typeof subscription !== 'function') {
       throw new ERR_INVALID_ARG_TYPE('subscription', ['function'],
@@ -71,7 +139,11 @@ class Channel {
   static [SymbolHasInstance](instance) {
     const prototype = ObjectGetPrototypeOf(instance);
     return prototype === Channel.prototype ||
-           prototype === ActiveChannel.prototype;
+      prototype === ActiveChannel.prototype;
+  }
+
+  [SymbolAsyncIterator]() {
+    return new AsyncIterableChannel(this);
   }
 
   subscribe(subscription) {

test/parallel/test-diagnostics-channel-async-iterable.js

@@ -0,0 +1,31 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const input = {
+  foo: 'bar'
+};
+
+const channel = dc.channel('test');
+
+const done = common.mustCall();
+
+async function main() {
+  for await (const message of channel) {
+    assert.strictEqual(channel.hasSubscribers, true);
+    assert.strictEqual(message, input);
+    break;
+  }
+
+  // Make sure the subscription is cleaned up when breaking the loop!
+  assert.strictEqual(channel.hasSubscribers, false);
+  done();
+}
+
+main();
+
+setTimeout(common.mustCall(() => {
+  channel.publish(input);
+}), 1);