feat(pg-pubsub): add option to immediately get a result on listen (#812)

Author: ol-teuto

packages/pg-pubsub/__tests__/__snapshots__/subs.test.ts.snap

@@ -1366,7 +1366,13 @@ type Query implements Node {
 The root subscription type: contains realtime events you can subscribe to with the \`subscription\` operation.
 """
 type Subscription {
-  listen(topic: String!): ListenPayload!
+  listen(
+    """
+    If true, this subscription will trigger an event as soon as it initiates.
+    """
+    initialEvent: Boolean! = false
+    topic: String!
+  ): ListenPayload!
 }
 
 """All input for the \`updateBarById\` mutation."""

packages/pg-pubsub/src/PgGenericSubscriptionPlugin.ts

@@ -1,6 +1,6 @@
 import debugFactory from "debug";
 import { Plugin } from "graphile-build";
-import { GraphQLFieldConfig } from "graphql";
+import { GraphQLBoolean, GraphQLFieldConfig } from "graphql";
 import { PubSub } from "graphql-subscriptions";
 
 const debug = debugFactory("pg-pubsub");
@@ -12,6 +12,38 @@ function isPubSub(pubsub: any): pubsub is PubSub {
   return !!pubsub;
 }
 
+const withInitialValue = <T>(
+  initialVal: T,
+  asyncIterator: AsyncIterator<T>
+): AsyncIterable<T> => {
+  return {
+    [Symbol.asyncIterator]: async function* (): AsyncIterator<T> {
+      yield initialVal;
+
+      /* TODO: when we can upgrade to Node 10.3+ we can replace the below with simply:
+      for await (const val of asyncIterator) {
+        yield val;
+      }
+      */
+      try {
+        while (true) {
+          const next = await asyncIterator.next();
+          if (next.done) {
+            return next.value;
+          } else {
+            yield next.value;
+          }
+        }
+      } finally {
+        // Terminate the previous iterator
+        if (typeof asyncIterator.return === "function") {
+          asyncIterator.return();
+        }
+      }
+    },
+  };
+};
+
 const PgGenericSubscriptionPlugin: Plugin = function (
   builder,
   {
@@ -142,6 +174,12 @@ const PgGenericSubscriptionPlugin: Plugin = function (
               topic: {
                 type: new GraphQLNonNull(GraphQLString),
               },
+              initialEvent: {
+                defaultValue: false,
+                type: new GraphQLNonNull(GraphQLBoolean),
+                description:
+                  "If true, this subscription will trigger an event as soon as it initiates.",
+              },
             },
             subscribe: async (_parent, args, _context, _resolveInfo) => {
               const { pgClient } = _context;
@@ -188,7 +226,11 @@ const PgGenericSubscriptionPlugin: Plugin = function (
                   }
                 });
               }
-              return asyncIterator;
+              if (args.initialEvent) {
+                return withInitialValue(null, asyncIterator);
+              } else {
+                return asyncIterator;
+              }
             },
             resolve: async (payload, _args, _context, _resolveInfo) => {
               const result = { ...payload };