subscribe-test: remove dependency on Node's EventEmitter (#2796)

Author: IvanGoncharov

src/subscription/__tests__/eventEmitterAsyncIterator.js

@@ -1,19 +1,16 @@
-import EventEmitter from 'events';
-
 import { expect } from 'chai';
 import { describe, it } from 'mocha';
 
-import eventEmitterAsyncIterator from './eventEmitterAsyncIterator';
+import SimplePubSub from './simplePubSub';
 
-describe('eventEmitterAsyncIterator', () => {
+describe('SimplePubSub', () => {
   it('subscribe async-iterator mock', async () => {
-    // Create an AsyncGenerator from an EventEmitter
-    const emitter = new EventEmitter();
-    const iterator = eventEmitterAsyncIterator(emitter, 'publish');
+    const pubsub = new SimplePubSub();
+    const iterator = pubsub.getSubscriber();
 
     // Queue up publishes
-    expect(emitter.emit('publish', 'Apple')).to.equal(true);
-    expect(emitter.emit('publish', 'Banana')).to.equal(true);
+    expect(pubsub.emit('Apple')).to.equal(true);
+    expect(pubsub.emit('Banana')).to.equal(true);
 
     // Read payloads
     expect(await iterator.next()).to.deep.equal({
@@ -30,8 +27,8 @@ describe('eventEmitterAsyncIterator', () => {
     const i4 = iterator.next().then((x) => x);
 
     // Publish
-    expect(emitter.emit('publish', 'Coconut')).to.equal(true);
-    expect(emitter.emit('publish', 'Durian')).to.equal(true);
+    expect(pubsub.emit('Coconut')).to.equal(true);
+    expect(pubsub.emit('Durian')).to.equal(true);
 
     // Await out of order to get correct results
     expect(await i4).to.deep.equal({ done: false, value: 'Durian' });
@@ -40,11 +37,11 @@ describe('eventEmitterAsyncIterator', () => {
     // Read ahead
     const i5 = iterator.next().then((x) => x);
 
-    // Terminate emitter
+    // Terminate queue
     await iterator.return();
 
     // Publish is not caught after terminate
-    expect(emitter.emit('publish', 'Fig')).to.equal(false);
+    expect(pubsub.emit('Fig')).to.equal(false);
 
     // Find that cancelled read-ahead got a "done" result
     expect(await i5).to.deep.equal({ done: true, value: undefined });

src/subscription/__tests__/eventEmitterAsyncIterator-test.js src/subscription/__tests__/simplePubSub-test.js

@@ -0,0 +1,70 @@
+/**
+ * Create an AsyncIterator from an EventEmitter. Useful for mocking a
+ * PubSub system for tests.
+ */
+export default class SimplePubSub<T> {
+  _subscribers: Set<(T) => void>;
+
+  constructor() {
+    this._subscribers = new Set();
+  }
+
+  emit(event: T): boolean {
+    for (const subscriber of this._subscribers) {
+      subscriber(event);
+    }
+    return this._subscribers.size > 0;
+  }
+
+  getSubscriber<R>(transform?: (T) => R): AsyncGenerator<R, void, void> {
+    const pullQueue = [];
+    const pushQueue = [];
+    let listening = true;
+    this._subscribers.add(pushValue);
+
+    const emptyQueue = () => {
+      listening = false;
+      this._subscribers.delete(pushValue);
+      for (const resolve of pullQueue) {
+        resolve({ value: undefined, done: true });
+      }
+      pullQueue.length = 0;
+      pushQueue.length = 0;
+    };
+
+    /* TODO: Flow doesn't support symbols as keys:
+       https://github.com/facebook/flow/issues/3258 */
+    return ({
+      next() {
+        if (!listening) {
+          return Promise.resolve({ value: undefined, done: true });
+        }
+
+        if (pushQueue.length > 0) {
+          return Promise.resolve({ value: pushQueue.shift(), done: false });
+        }
+        return new Promise((resolve) => pullQueue.push(resolve));
+      },
+      return() {
+        emptyQueue();
+        return Promise.resolve({ value: undefined, done: true });
+      },
+      throw(error: mixed) {
+        emptyQueue();
+        return Promise.reject(error);
+      },
+      [Symbol.asyncIterator]() {
+        return this;
+      },
+    }: any);
+
+    function pushValue(event: T): void {
+      const value = transform != null ? transform(event) : event;
+      if (pullQueue.length > 0) {
+        pullQueue.shift()({ value, done: false });
+      } else {
+        pushQueue.push(value);
+      }
+    }
+  }
+}