GraphQL Subscriptions

src/Controllers/ParseGraphQLController.js

@@ -24,7 +24,7 @@ class ParseGraphQLController {
         `ParseGraphQLController requires a "databaseController" to be instantiated.`
       );
     this.cacheController = params.cacheController;
-    this.isMounted = !!params.mountGraphQL;
+    this.isMounted = !!params.mountGraphQL || !!params.mountSubscriptions;
     this.configCacheKey = GraphQLConfigKey;
   }
 
@@ -145,7 +145,14 @@ class ParseGraphQLController {
     if (!isValidSimpleObject(classConfig)) {
       return 'it must be a valid object';
     } else {
-      const { className, type = null, query = null, mutation = null, ...invalidKeys } = classConfig;
+      const {
+        className,
+        type = null,
+        query = null,
+        mutation = null,
+        subscription = null,
+        ...invalidKeys
+      } = classConfig;
       if (Object.keys(invalidKeys).length) {
         return `"invalidKeys" [${Object.keys(invalidKeys)}] should not be present`;
       }
@@ -287,6 +294,20 @@ class ParseGraphQLController {
           return `"mutation" must be a valid object`;
         }
       }
+      if (subscription !== null) {
+        if (isValidSimpleObject(subscription)) {
+          const { enabled = null, alias = null, ...invalidKeys } = query;
+          if (Object.keys(invalidKeys).length) {
+            return `"subscription" contains invalid keys, [${Object.keys(invalidKeys)}]`;
+          } else if (enabled !== null && typeof enabled !== 'boolean') {
+            return `"subscription.enabled" must be a boolean`;
+          } else if (alias !== null && typeof alias !== 'string') {
+            return `"subscription.alias" must be a string`;
+          }
+        } else {
+          return `"subscription" must be a valid object`;
+        }
+      }
     }
   }
 }
@@ -355,6 +376,11 @@ export interface ParseGraphQLClassConfig {
     updateAlias: ?String,
     destroyAlias: ?String,
   };
+  /* The `subscription` object contains options for which class subscriptions are generated */
+  subscription: ?{
+    enabled: ?boolean,
+    alias: ?String,
+  };
 }
 
 export default ParseGraphQLController;

src/Controllers/index.js

@@ -127,6 +127,7 @@ export function getParseGraphQLController(
 ): ParseGraphQLController {
   return new ParseGraphQLController({
     mountGraphQL: options.mountGraphQL,
+    mountSubscriptions: options.mountSubscriptions,
     ...controllerDeps,
   });
 }

src/GraphQL/ParseGraphQLSchema.js

@@ -7,6 +7,7 @@ import * as defaultGraphQLTypes from './loaders/defaultGraphQLTypes';
 import * as parseClassTypes from './loaders/parseClassTypes';
 import * as parseClassQueries from './loaders/parseClassQueries';
 import * as parseClassMutations from './loaders/parseClassMutations';
+import * as parseClassSubscriptions from './loaders/parseClassSubscriptions';
 import * as defaultGraphQLQueries from './loaders/defaultGraphQLQueries';
 import * as defaultGraphQLMutations from './loaders/defaultGraphQLMutations';
 import ParseGraphQLController, { ParseGraphQLConfig } from '../Controllers/ParseGraphQLController';
@@ -58,6 +59,7 @@ const RESERVED_GRAPHQL_MUTATION_NAMES = [
   'updateClass',
   'deleteClass',
 ];
+const RESERVED_GRAPHQL_SUBSCRIPTION_NAMES = [];
 
 class ParseGraphQLSchema {
   databaseController: DatabaseController;
@@ -66,6 +68,7 @@ class ParseGraphQLSchema {
   log: any;
   appId: string;
   graphQLCustomTypeDefs: ?(string | GraphQLSchema | DocumentNode | GraphQLNamedType[]);
+  liveQueryClassNames: any;
 
   constructor(
     params: {
@@ -74,6 +77,7 @@ class ParseGraphQLSchema {
       log: any,
       appId: string,
       graphQLCustomTypeDefs: ?(string | GraphQLSchema | DocumentNode | GraphQLNamedType[]),
+      liveQueryClassNames: any,
     } = {}
   ) {
     this.parseGraphQLController =
@@ -85,6 +89,7 @@ class ParseGraphQLSchema {
     this.log = params.log || requiredParameter('You must provide a log instance!');
     this.graphQLCustomTypeDefs = params.graphQLCustomTypeDefs;
     this.appId = params.appId || requiredParameter('You must provide the appId!');
+    this.liveQueryClassNames = params.liveQueryClassNames;
   }
 
   async load() {
@@ -132,6 +137,9 @@ class ParseGraphQLSchema {
         parseClassTypes.load(this, parseClass, parseClassConfig);
         parseClassQueries.load(this, parseClass, parseClassConfig);
         parseClassMutations.load(this, parseClass, parseClassConfig);
+        if (this.liveQueryClassNames && this.liveQueryClassNames.includes(parseClass.className)) {
+          parseClassSubscriptions.load(this, parseClass, parseClassConfig);
+        }
       }
     );
 
@@ -342,6 +350,22 @@ class ParseGraphQLSchema {
     return field;
   }
 
+  addGraphQLSubscription(fieldName, field, throwError = false, ignoreReserved = false) {
+    if (
+      (!ignoreReserved && RESERVED_GRAPHQL_SUBSCRIPTION_NAMES.includes(fieldName)) ||
+      this.graphQLSubscriptions[fieldName]
+    ) {
+      const message = `Subscription ${fieldName} could not be added to the auto schema because it collided with an existing field.`;
+      if (throwError) {
+        throw new Error(message);
+      }
+      this.log.warn(message);
+      return undefined;
+    }
+    this.graphQLSubscriptions[fieldName] = field;
+    return field;
+  }
+
   handleError(error) {
     if (error instanceof Parse.Error) {
       this.log.error('Parse error: ', error);

src/GraphQL/ParseGraphQLServer.js

@@ -8,8 +8,10 @@ import { SubscriptionServer } from 'subscriptions-transport-ws';
 import { handleParseErrors, handleParseHeaders } from '../middlewares';
 import requiredParameter from '../requiredParameter';
 import defaultLogger from '../logger';
+import { ParseLiveQueryServer } from '../LiveQuery/ParseLiveQueryServer';
 import { ParseGraphQLSchema } from './ParseGraphQLSchema';
 import ParseGraphQLController, { ParseGraphQLConfig } from '../Controllers/ParseGraphQLController';
+import { WSSAdapter } from '../Adapters/WebSocketServer/WSSAdapter';
 
 class ParseGraphQLServer {
   parseGraphQLController: ParseGraphQLController;
@@ -29,6 +31,8 @@ class ParseGraphQLServer {
       log: this.log,
       graphQLCustomTypeDefs: this.config.graphQLCustomTypeDefs,
       appId: this.parseServer.config.appId,
+      liveQueryClassNames:
+        this.parseServer.config.liveQuery && this.parseServer.config.liveQuery.classNames,
     });
   }
 
@@ -114,12 +118,132 @@ class ParseGraphQLServer {
   }
 
   createSubscriptions(server) {
+    const wssAdapter = new WSSAdapter();
+
+    new ParseLiveQueryServer(
+      undefined,
+      {
+        ...this.parseServer.config.liveQueryServerOptions,
+        wssAdapter,
+      },
+      this.parseServer.config
+    );
+
     SubscriptionServer.create(
       {
         execute,
         subscribe,
-        onOperation: async (_message, params, webSocket) =>
-          Object.assign({}, params, await this._getGraphQLOptions(webSocket.upgradeReq)),
+        onConnect: async connectionParams => {
+          const keyPairs = {
+            applicationId: connectionParams['X-Parse-Application-Id'],
+            sessionToken: connectionParams['X-Parse-Session-Token'],
+            masterKey: connectionParams['X-Parse-Master-Key'],
+            installationId: connectionParams['X-Parse-Installation-Id'],
+            clientKey: connectionParams['X-Parse-Client-Key'],
+            javascriptKey: connectionParams['X-Parse-Javascript-Key'],
+            windowsKey: connectionParams['X-Parse-Windows-Key'],
+            restAPIKey: connectionParams['X-Parse-REST-API-Key'],
+          };
+
+          const listeners = [];
+
+          let connectResolve, connectReject;
+          let connectIsResolved = false;
+          const connectPromise = new Promise((resolve, reject) => {
+            connectResolve = resolve;
+            connectReject = reject;
+          });
+
+          const liveQuery = {
+            OPEN: 'OPEN',
+            readyState: 'OPEN',
+            on: () => {},
+            ping: () => {},
+            onmessage: () => {},
+            onclose: () => {},
+            send: message => {
+              message = JSON.parse(message);
+              if (message.op === 'connected') {
+                connectResolve();
+                connectIsResolved = true;
+                return;
+              } else if (message.op === 'error' && !connectIsResolved) {
+                connectReject({
+                  code: message.code,
+                  message: message.error,
+                });
+                return;
+              }
+              const requestId = message && message.requestId;
+              if (
+                requestId &&
+                typeof requestId === 'number' &&
+                requestId > 0 &&
+                requestId <= listeners.length
+              ) {
+                const listener = listeners[requestId - 1];
+                if (listener) {
+                  listener(message);
+                }
+              }
+            },
+            subscribe: async (query, sessionToken, listener) => {
+              await connectPromise;
+              listeners.push(listener);
+              liveQuery.onmessage(
+                JSON.stringify({
+                  op: 'subscribe',
+                  requestId: listeners.length,
+                  query,
+                  sessionToken,
+                })
+              );
+            },
+            unsubscribe: async listener => {
+              await connectPromise;
+              const index = listeners.indexOf(listener);
+              if (index > 0) {
+                liveQuery.onmessage(
+                  JSON.stringify({
+                    op: 'unsubscribe',
+                    requestId: index + 1,
+                  })
+                );
+                listeners[index] = null;
+              }
+            },
+          };
+
+          wssAdapter.onConnection(liveQuery);
+
+          liveQuery.onmessage(
+            JSON.stringify({
+              op: 'connect',
+              ...keyPairs,
+            })
+          );
+
+          await connectPromise;
+
+          return { liveQuery, keyPairs };
+        },
+        onDisconnect: (_webSocket, context) => {
+          const { liveQuery } = context;
+
+          if (liveQuery) {
+            liveQuery.onclose();
+          }
+        },
+        onOperation: async (_message, params) => {
+          return {
+            ...params,
+            schema: await this.parseGraphQLSchema.load(),
+            formatError: error => {
+              // Allow to console.log here to debug
+              return error;
+            },
+          };
+        },
       },
       {
         server,

src/GraphQL/loaders/defaultGraphQLTypes.js

@@ -1222,6 +1222,29 @@ const loadArrayResult = (parseGraphQLSchema, parseClasses) => {
   parseGraphQLSchema.graphQLTypes.push(ARRAY_RESULT);
 };
 
+const EVENT_KIND = new GraphQLEnumType({
+  name: 'EventKind',
+  description: 'The EventKind enum type is used in subscriptions to identify listened events.',
+  values: {
+    create: { value: 'create' },
+    enter: { value: 'enter' },
+    update: { value: 'update' },
+    leave: { value: 'leave' },
+    delete: { value: 'delete' },
+    all: { value: 'all' },
+  },
+});
+
+const EVENT_KIND_ATT = {
+  description: 'The event kind that was fired.',
+  type: new GraphQLNonNull(EVENT_KIND),
+};
+
+const EVENT_KINDS_ATT = {
+  description: 'The event kinds to be listened in the subscription.',
+  type: new GraphQLNonNull(new GraphQLList(new GraphQLNonNull(EVENT_KIND))),
+};
+
 const load = parseGraphQLSchema => {
   parseGraphQLSchema.addGraphQLType(GraphQLUpload, true);
   parseGraphQLSchema.addGraphQLType(ANY, true);
@@ -1266,6 +1289,7 @@ const load = parseGraphQLSchema => {
   parseGraphQLSchema.addGraphQLType(PUBLIC_ACL, true);
   parseGraphQLSchema.addGraphQLType(SUBQUERY_INPUT, true);
   parseGraphQLSchema.addGraphQLType(SELECT_INPUT, true);
+  parseGraphQLSchema.addGraphQLType(EVENT_KIND, true);
 };
 
 export {
@@ -1358,6 +1382,9 @@ export {
   USER_ACL,
   ROLE_ACL,
   PUBLIC_ACL,
+  EVENT_KIND,
+  EVENT_KIND_ATT,
+  EVENT_KINDS_ATT,
   load,
   loadArrayResult,
 };