Implemented Generic AsyncIterator

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+### 1.1.0
+
+- Replaced `eventEmitterAsynIterator` with default generic `AsyncIterator` named `PubSubAsyncIterator`. `extends PubSubEngine` automatically uses generic implementation. No breaking changes for those who continue to use `implements PubSubEngine`. See PR [#78](https://github.com/apollographql/graphql-subscriptions/pull/78).
+
 ### 1.0.0
 
 - BREAKING CHANGE: Changed return type of `publish`.  <br/>

README.md

@@ -225,7 +225,7 @@ For more information about `AsyncIterator`:
 
 ### PubSub Implementations
 
-It can be easily replaced with some other implementations of [PubSubEngine interface](https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub.ts#L21-L25). There are a couple of them out there:
+It can be easily replaced with some other implementations of [PubSubEngine abstract class](https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub-engine.ts). Here are a few of them:
 - Use Redis with https://github.com/davidyaha/graphql-redis-subscriptions
 - Use Google PubSub with https://github.com/axelspringer/graphql-google-pubsub
 - Use MQTT enabled broker with https://github.com/davidyaha/graphql-mqtt-subscriptions
@@ -236,7 +236,7 @@ It can be easily replaced with some other implementations of [PubSubEngine inter
 - Use multiple backends with https://github.com/jcoreio/graphql-multiplex-subscriptions
 - [Add your implementation...](https://github.com/apollographql/graphql-subscriptions/pull/new/master)
 
-You can also implement a `PubSub` of your own, by using the exported interface `PubSubEngine` from this package.
+You can also implement a `PubSub` of your own, by using the exported abstract class `PubSubEngine` from this package. By using `extends PubSubEngine` you use the default `asyncIterator` method implementation; by using `implements PubSubEngine` you must implement your own `AsyncIterator`.
 
 #### SubscriptionManager **@deprecated**
 

package.json

@@ -31,7 +31,7 @@
     "@types/graphql": "^14.0.0",
     "@types/mocha": "^2.2.39",
     "@types/node": "^8.0.28",
-    "@types/sinon": "^5.0.1",
+    "@types/sinon": "5.0.2",
     "@types/sinon-chai": "^3.2.0",
     "chai": "^4.1.2",
     "chai-as-promised": "^7.1.1",

src/pubsub-async-iterator.ts

@@ -0,0 +1,121 @@
+import { $$asyncIterator } from 'iterall';
+import {PubSubEngine} from './pubsub-engine';
+
+/**
+ * A class for digesting PubSubEngine events via the new AsyncIterator interface.
+ * This implementation is a generic version of the AsyncIterator, so any PubSubEngine may
+ * be used.
+ * @class
+ *
+ * @constructor
+ *
+ * @property pullQueue @type {Function[]}
+ * A queue of resolve functions waiting for an incoming event which has not yet arrived.
+ * This queue expands as next() calls are made without PubSubEngine events occurring in-between.
+ *
+ * @property pushQueue @type {T[]}
+ * A queue of PubSubEngine events waiting for next() calls to be made, which returns the queued events
+ * for handling. This queue expands as PubSubEngine events arrive without next() calls occurring in-between.
+ *
+ * @property eventsArray @type {string[]}
+ * An array of PubSubEngine event names that this PubSubAsyncIterator should watch.
+ *
+ * @property allSubscribed @type {Promise<number[]>}
+ * undefined until next() called for the first time, afterwards is a promise of an array of all
+ * subscription ids, where each subscription id identified a subscription on the PubSubEngine.
+ * The undefined initialization ensures that subscriptions are not made to the PubSubEngine
+ * before next() has ever been called.
+ *
+ * @property running @type {boolean}
+ * Whether or not the PubSubAsynIterator is in running mode (responding to incoming PubSubEngine events and next() calls).
+ * running begins as true and turns to false once the return method is called.
+ *
+ * @property pubsub @type {PubSubEngine}
+ * The PubSubEngine whose events will be observed.
+ */
+export class PubSubAsyncIterator<T> implements AsyncIterator<T> {
+
+  private pullQueue: ((value?: any) => Promise<IteratorResult<T>>)[];
+  private pushQueue: T[];
+  private eventsArray: string[];
+  private allSubscribed: Promise<number[]>;
+  private running: boolean;
+  private pubsub: PubSubEngine;
+
+  constructor(pubsub: PubSubEngine, eventNames: string | string[]) {
+    this.pubsub = pubsub;
+    this.pullQueue = [];
+    this.pushQueue = [];
+    this.running = true;
+    this.allSubscribed = null;
+    this.eventsArray = typeof eventNames === 'string' ? [eventNames] : eventNames;
+  }
+
+  public async next(): Promise<IteratorResult<T>> {
+    if (!this.allSubscribed) await (this.allSubscribed = this.subscribeAll());
+    return this.pullValue()
+  }
+
+  public async return(): Promise<IteratorResult<T>> {
+    await this.emptyQueue();
+    return { value: undefined, done: true };
+  }
+
+  public async throw(error) {
+    await this.emptyQueue();
+    return Promise.reject(error);
+  }
+
+  public [$$asyncIterator]() {
+    return this;
+  }
+
+  private async pushValue(event: T) {
+    await this.allSubscribed;
+    if (this.pullQueue.length !== 0) {
+      this.pullQueue.shift()(this.running
+        ? { value: event, done: false }
+        : { value: undefined, done: true }
+      );
+    } else {
+      this.pushQueue.push(event);
+    }
+  }
+
+  private pullValue(): Promise<IteratorResult<T>> {
+    return new Promise((resolve => {
+      if (this.pushQueue.length !== 0) {
+        resolve(this.running
+          ? { value: this.pushQueue.shift(), done: false }
+          : { value: undefined, done: true }
+        );
+      } else {
+        this.pullQueue.push(resolve);
+      }
+    }).bind(this));
+  }
+
+  private async emptyQueue() {
+    if (this.running) {
+      this.running = false;
+      this.pullQueue.forEach(resolve => resolve({ value: undefined, done: true }));
+      this.pullQueue.length = 0;
+      this.pushQueue.length = 0;
+      const subscriptionIds = await this.allSubscribed;
+      if (subscriptionIds) this.unsubscribeAll(subscriptionIds);
+    }
+  }
+
+  private subscribeAll() {
+    return Promise.all(this.eventsArray.map(
+      eventName => this.pubsub.subscribe(eventName, this.pushValue.bind(this), {}),
+    ));
+  }
+
+  private unsubscribeAll(subscriptionIds: number[]) {
+    for (const subscriptionId of subscriptionIds) {
+      this.pubsub.unsubscribe(subscriptionId);
+    }
+  }
+
+}

src/pubsub-engine.ts

@@ -1,6 +1,10 @@
-export interface PubSubEngine {
-  publish(triggerName: string, payload: any): Promise<void>;
-  subscribe(triggerName: string, onMessage: Function, options: Object): Promise<number>;
-  unsubscribe(subId: number);
-  asyncIterator<T>(triggers: string | string[]): AsyncIterator<T>;
+import {PubSubAsyncIterator} from './pubsub-async-iterator';
+
+export abstract class PubSubEngine {
+  public abstract publish(triggerName: string, payload: any): Promise<void>;
+  public abstract subscribe(triggerName: string, onMessage: Function, options: Object): Promise<number>;
+  public abstract unsubscribe(subId: number);
+  public asyncIterator<T>(triggers: string | string[]): AsyncIterator<T> {
+    return new PubSubAsyncIterator<T>(this, triggers);
+  }
 }

src/pubsub.ts

@@ -1,17 +1,17 @@
 import { EventEmitter } from 'events';
 import { PubSubEngine } from './pubsub-engine';
-import { eventEmitterAsyncIterator } from './event-emitter-to-async-iterator';
 
 export interface PubSubOptions {
   eventEmitter?: EventEmitter;
 }
 
-export class PubSub implements PubSubEngine {
+export class PubSub extends PubSubEngine {
   protected ee: EventEmitter;
   private subscriptions: { [key: string]: [string, (...args: any[]) => void] };
   private subIdCounter: number;
 
   constructor(options: PubSubOptions = {}) {
+    super();
     this.ee = options.eventEmitter || new EventEmitter();
     this.subscriptions = {};
     this.subIdCounter = 0;
@@ -35,8 +35,4 @@ export class PubSub implements PubSubEngine {
     delete this.subscriptions[subId];
     this.ee.removeListener(triggerName, onMessage);
   }
-
-  public asyncIterator<T>(triggers: string | string[]): AsyncIterator<T> {
-    return eventEmitterAsyncIterator<T>(this.ee, triggers);
-  }
 }

src/test/tests.ts

@@ -88,19 +88,21 @@ describe('AsyncIterator', () => {
     const iterator = ps.asyncIterator(eventName);
 
     iterator.next().then(result => {
-      expect(result).to.not.be.undefined;
-      expect(result.value).to.not.be.undefined;
-      expect(result.done).to.be.false;
-    });
+      expect(result).to.deep.equal({
+        value: undefined,
+        done: true,
+      });
+    }).catch(done);
 
     ps.publish(eventName, { test: true });
 
     iterator.next().then(result => {
-      expect(result).to.not.be.undefined;
-      expect(result.value).to.be.undefined;
-      expect(result.done).to.be.true;
+      expect(result).to.deep.equal({
+        value: undefined,
+        done: true,
+      });
       done();
-    });
+    }).catch(done);
 
     iterator.return();
 

src/validation.ts

@@ -0,0 +1,43 @@
+import {
+  ValidationContext,
+  SelectionNode,
+  GraphQLError,
+} from 'graphql';
+
+// XXX I don't know how else to do this. Can't seem to import from GraphQL.
+const FIELD = 'Field';
+
+/**
+ * @deprecated
+ */
+export function tooManySubscriptionFieldsError(subscriptionName: string): string {
+  return `Subscription "${subscriptionName}" must have only one field.`;
+}
+
+// XXX we temporarily use this validation rule to make our life a bit easier.
+
+/**
+ * @deprecated
+ */
+export function subscriptionHasSingleRootField(context: ValidationContext): any {
+  const schema = context.getSchema();
+  schema.getSubscriptionType();
+  return {
+    OperationDefinition(node) {
+      const operationName = node.name ? node.name.value : '';
+      let numFields = 0;
+      node.selectionSet.selections.forEach( (selection: SelectionNode) => {
+        if (selection.kind === FIELD) {
+          numFields++;
+        } else {
+          // why the heck use a fragment on the Subscription type? Just ... don't
+          context.reportError(new GraphQLError('Apollo subscriptions do not support fragments on the root field', [node]));
+        }
+      });
+      if (numFields > 1) {
+        context.reportError(new GraphQLError(tooManySubscriptionFieldsError(operationName), [node]));
+      }
+      return false;
+    },
+  };
+}