Cleanup of subscriptions

bin/cli-serve

@@ -8,7 +8,7 @@ const mailer = require('../services/mailer');
 const kue = require('../services/kue');
 const mongoose = require('../services/mongoose');
 const util = require('./util');
-const subscriptions = require('../services/subscriptions');
+const {createSubscriptionManager} = require('../graph/subscriptions');
 
 /**
 * Get port from environment and store in Express.
@@ -77,25 +77,29 @@ function normalizePort(val) {
 function onListening() {
   let addr = server.address();
   let bind = typeof addr === 'string'
-    ? `pipe ${  addr}`
-    : `port ${  addr.port}`;
-  console.log(`Listening on ${  bind}`);
+    ? `pipe ${addr}`
+    : `port ${addr.port}`;
+  console.log(`API Server Listening on ${bind}`);
 }
 
 /**
  * Start the app.
  */
-function startApp() {
+function startApp(program) {
 
   /**
   * Listen on provided port, on all network interfaces.
   */
   server.listen(port, () => {
 
-    // Mount the subscriptions server on the application server.
-    subscriptions.mount(server);
-  });
+    // Mount the websocket server if requested.
+    if (program.websockets) {
+      console.log(`Websocket Server Listening on ${port}`);
 
+      // Mount the subscriptions server on the application server.
+      createSubscriptionManager(server);
+    }
+  });
   server.on('error', onError);
   server.on('listening', onListening);
 }
@@ -106,10 +110,11 @@ function startApp() {
 
 program
   .option('-j, --jobs', 'enable job processing on this thread')
+  .option('-w, --websockets', 'enable the websocket (subscriptions) handler on this thread')
   .parse(process.argv);
 
 // Start the application serving.
-startApp();
+startApp(program);
 
 // Enable job processing on the thread if enabled.
 if (program.jobs) {

graph/index.js

@@ -1,16 +1,9 @@
 const schema = require('./schema');
 const Context = require('./context');
-const setupFunctions = require('./setupFunctions');
-
-const {connectionOptions} = require('../services/redis');
-const {RedisPubSub} = require('graphql-redis-subscriptions');
-const {SubscriptionManager} = require('graphql-subscriptions');
-const {SubscriptionServer} = require('subscriptions-transport-ws');
-
-const pubsub = new RedisPubSub(connectionOptions);
+const pubsub = require('./pubsub');
+const {createSubscriptionManager} = require('./subscriptions');
 
 module.exports = {
-  pubsub,
   createGraphOptions: (req) => ({
 
     // Schema is created already, so just include it.
@@ -20,27 +13,5 @@ module.exports = {
     // the lifespan of this request.
     context: new Context(req, pubsub)
   }),
-  createSubscriptionManager: (server, path, sessionFactory) => new SubscriptionServer({
-    subscriptionManager: new SubscriptionManager({
-      schema,
-      pubsub,
-      setupFunctions,
-    }),
-    onSubscribe: (parsedMessage, baseParams, connection) => {
-
-      // Attach the context per request.
-      baseParams.context = () => sessionFactory(connection.upgradeReq)
-        .then((req) => new Context(req, pubsub))
-        .catch((err) => {
-          console.error(err);
-
-          return new Context({}, pubsub);
-        });
-
-      return baseParams;
-    }
-  }, {
-    server,
-    path
-  })
+  createSubscriptionManager
 };

graph/pubsub.js

@@ -0,0 +1,5 @@
+const {RedisPubSub} = require('graphql-redis-subscriptions');
+
+const {connectionOptions} = require('../services/redis');
+
+module.exports = new RedisPubSub(connectionOptions);

graph/setupFunctions.js

@@ -1,21 +0,0 @@
-const plugins = require('../services/plugins');
-const _ = require('lodash');
-
-// Core setup functions
-let setupFunctions = {
-  commentAdded: (options, args) => ({
-    commentAdded: {
-      filter: (comment) => comment.asset_id === args.asset_id
-    },
-  }),
-};
-
-/**
- * Plugin support requires that we merge in existing setupFunctions with our new
- * plugin based ones. This allows plugins to extend existing setupFunctions as well
- * as provide new ones.
- */
-module.exports = plugins.get('server', 'setupFunctions').reduce((acc, {setupFunctions}) => {
-
-  return _.merge(acc, setupFunctions);
-}, setupFunctions);

graph/subscriptions.js

@@ -0,0 +1,60 @@
+const {SubscriptionManager} = require('graphql-subscriptions');
+const {SubscriptionServer} = require('subscriptions-transport-ws');
+const _ = require('lodash');
+
+const pubsub = require('./pubsub');
+const schema = require('./schema');
+const Context = require('./context');
+const plugins = require('../services/plugins');
+
+const {deserializeUser} = require('../services/subscriptions');
+
+// Core setup functions
+let setupFunctions = {
+  commentAdded: (options, args) => ({
+    commentAdded: {
+      filter: (comment) => comment.asset_id === args.asset_id
+    },
+  }),
+};
+
+/**
+ * Plugin support requires that we merge in existing setupFunctions with our new
+ * plugin based ones. This allows plugins to extend existing setupFunctions as well
+ * as provide new ones.
+ */
+setupFunctions = plugins.get('server', 'setupFunctions').reduce((acc, {setupFunctions}) => {
+
+  return _.merge(acc, setupFunctions);
+}, setupFunctions);
+
+/**
+ * This creates a new subscription manager.
+ */
+const createSubscriptionManager = (server) => new SubscriptionServer({
+  subscriptionManager: new SubscriptionManager({
+    schema,
+    pubsub,
+    setupFunctions,
+  }),
+  onSubscribe: (parsedMessage, baseParams, connection) => {
+
+    // Attach the context per request.
+    baseParams.context = () => deserializeUser(connection.upgradeReq)
+      .then((req) => new Context(req, pubsub))
+      .catch((err) => {
+        console.error(err);
+
+        return new Context({}, pubsub);
+      });
+
+    return baseParams;
+  }
+}, {
+  server,
+  path: '/api/v1/live'
+});
+
+module.exports = {
+  createSubscriptionManager
+};

package.json

@@ -5,8 +5,8 @@
   "main": "app.js",
   "scripts": {
     "postinstall": "./bin/cli plugins reconcile --skip-remote",
-    "start": "./bin/cli serve --jobs",
-    "dev-start": "nodemon --config .nodemon.json --exec \"./bin/cli -c .env serve --jobs\"",
+    "start": "./bin/cli serve -j -w",
+    "dev-start": "nodemon -w . -w bin/cli -w bin/cli-serve --config .nodemon.json --exec \"./bin/cli -c .env serve -j -w\"",
     "build": "NODE_ENV=production webpack --progress -p --config webpack.config.js --bail",
     "build-watch": "NODE_ENV=development webpack --progress --config webpack.config.js --watch",
     "lint": "eslint bin/* .",

services/subscriptions.js

@@ -1,38 +1,36 @@
-const {createSubscriptionManager} = require('../graph');
 const session = require('./session');
 const passport = require('./passport');
 
-module.exports = class Subscriptions {
+// Session data does not automatically attach to websocket req objects.
+// This middleware code looks for a user in the session and, if it exists,
+// attaches it to the graph req.
+const deserializeUser = (req) => {
+  return new Promise((resolve, reject) => {
+    session(req, {}, () => {
 
-  // Session data does not automatically attach to websocket req objects.
-  // This middleware code looks for a user in the session and, if it exists,
-  // attaches it to the graph req.
-  static deserializeUser(req) {
-    return new Promise((resolve, reject) => {
-      session(req, {}, () => {
+      if ('session' in req && 'passport' in req.session && 'user' in req.session.passport) {
+        passport.deserializeUser(req.session.passport.user, (err, user) => {
+          if (err) {
+            return reject(err);
+          }
 
-        if ('session' in req && 'passport' in req.session && 'user' in req.session.passport) {
-          passport.deserializeUser(req.session.passport.user, (err, user) => {
-            if (err) {
-              return reject(err);
-            }
+          req.user = user;
 
-            req.user = user;
+          return resolve(req);
+        });
+      }
 
-            return resolve(req);
-          });
-        } else {
-          resolve(req);
-        }
+        // Remove the user from the request (if there was one)
+      if (req.user) {
+        delete req.user;
+      }
 
-      });
+        // Resolve with the request (user removed possibly).
+      return resolve(req);
     });
-  }
-
-  static mount(server) {
-
-    // Create the SubscriptionManager and mount it on the specified route with
-    // this deserializer.
-    createSubscriptionManager(server, '/api/v1/live', Subscriptions.deserializeUser);
-  }
+  });
+};
+
+module.exports = {
+  deserializeUser
 };