subscription refactor

graph/connectors.js

@@ -28,7 +28,6 @@ const Moderation = require('../services/moderation');
 const Mongoose = require('../services/mongoose');
 const Passport = require('../services/passport');
 const Plugins = require('../services/plugins');
-const Pubsub = require('../services/pubsub');
 const Redis = require('../services/redis');
 const Regex = require('../services/regex');
 const Scraper = require('../services/scraper');
@@ -67,7 +66,6 @@ const connectors = {
     Mongoose,
     Passport,
     Plugins,
-    Pubsub,
     Redis,
     Regex,
     Scraper,

graph/context.js

@@ -5,7 +5,7 @@ const merge = require('lodash/merge');
 const connectors = require('./connectors');
 
 const plugins = require('../services/plugins');
-const pubsub = require('../services/pubsub');
+const { getBroker } = require('./subscriptions/broker');
 const debug = require('debug')('talk:graph:context');
 
 /**
@@ -68,7 +68,7 @@ class Context {
     this.plugins = decorateContextPlugins(this, contextPlugins);
 
     // Bind the publish/subscribe to the context.
-    this.pubsub = pubsub.getClient();
+    this.pubsub = getBroker();
 
     // Bind the parent context.
     this.parent = parent;

graph/subscriptions/broker.js

@@ -0,0 +1,48 @@
+const { EventEmitter2 } = require('eventemitter2');
+const debug = require('debug')('talk:graph:subscriptions:broker');
+const { getPubsub } = require('./pubsub');
+
+/**
+ * Broker acts as a pubsub client adapter. Any calls to publish will push into
+ * the PubSub client and the local event emitter.
+ */
+class Broker extends EventEmitter2 {
+  constructor(pubsub) {
+    // Create the underlying event emitter.
+    super({
+      wildcard: true, // Allow wildcard listeners.
+      maxListeners: 0, // Disable maximum for listeners.
+    });
+
+    this.pubsub = pubsub;
+  }
+
+  /**
+   * Publishes the event out to the pubsub system and to the broker.
+   *
+   * @param {String} event the name of the event to publish
+   * @param {Any} args the
+   */
+  publish(event, ...args) {
+    debug(`publish:${event}`);
+    this.pubsub.publish(event, ...args);
+    this.emit(event, ...args);
+  }
+}
+
+let client = null;
+const getBroker = () => {
+  if (client !== null) {
+    return client;
+  }
+
+  // Create the new Broker to manage events being published out to
+  // the pubsub so that we may intercept.
+  client = new Broker(getPubsub());
+
+  debug('created');
+
+  return client;
+};
+
+module.exports.getBroker = getBroker;

graph/subscriptions/index.js

@@ -2,18 +2,18 @@ const { SubscriptionManager } = require('graphql-subscriptions');
 const { SubscriptionServer } = require('subscriptions-transport-ws');
 const debug = require('debug')('talk:graph:subscriptions');
 
-const pubsub = require('../services/pubsub');
-const schema = require('./schema');
-const Context = require('./context');
-const plugins = require('../services/plugins');
+const { getPubsub } = require('./pubsub');
+const schema = require('../schema');
+const Context = require('../context');
+const plugins = require('../../services/plugins');
 
-const { deserializeUser } = require('../services/subscriptions');
+const { deserializeUser } = require('../../services/subscriptions');
 const setupFunctions = require('./setupFunctions');
 
 const ms = require('ms');
-const { KEEP_ALIVE } = require('../config');
+const { KEEP_ALIVE } = require('../../config');
 
-const { BASE_PATH } = require('../url');
+const { BASE_PATH } = require('../../url');
 
 // Collect all the plugin hooks that should be executed onConnect and
 // onDisconnect.
@@ -99,7 +99,7 @@ const createSubscriptionManager = server =>
     {
       subscriptionManager: new SubscriptionManager({
         schema,
-        pubsub: pubsub.getClient(),
+        pubsub: getPubsub(),
         setupFunctions,
       }),
       onConnect,

graph/subscriptions/pubsub.js

@@ -0,0 +1,29 @@
+const { RedisPubSub } = require('graphql-redis-subscriptions');
+const { createClient: createRedisClient } = require('../../services/redis');
+const debug = require('debug')('talk:graph:subscriptions:pubsub');
+
+/**
+ * getPubsub returns the pubsub singleton for this instance.
+ */
+let pubsub = null;
+const getPubsub = () => {
+  if (pubsub !== null) {
+    return pubsub;
+  }
+
+  // Create the publisher and subscriber redis clients.
+  const publisher = createRedisClient();
+  const subscriber = createRedisClient();
+
+  // Create the new PubSub client, we only need one per instance of Talk.
+  pubsub = new RedisPubSub({
+    publisher,
+    subscriber,
+  });
+
+  debug('created');
+
+  return pubsub;
+};
+
+module.exports.getPubsub = getPubsub;

graph/subscriptions/setupFunctions.js

@@ -11,11 +11,11 @@ const {
   SUBSCRIBE_ALL_USERNAME_APPROVED,
   SUBSCRIBE_ALL_USERNAME_FLAGGED,
   SUBSCRIBE_ALL_USERNAME_CHANGED,
-} = require('../perms/constants');
+} = require('../../perms/constants');
 
 const merge = require('lodash/merge');
 const debug = require('debug')('talk:graph:setupFunctions');
-const plugins = require('../services/plugins');
+const plugins = require('../../services/plugins');
 
 const setupFunctions = {
   commentAdded: (options, args, comment, context) => {

middleware/pubsub.js

@@ -1,11 +0,0 @@
-const pubsub = require('../services/pubsub');
-
-// To handle dependancy injection safer, we inject the pubsub handle onto the
-// request object.
-module.exports = (req, res, next) => {
-  // Attach the pubsub handle to the requests.
-  req.pubsub = pubsub.getClient();
-
-  // Forward on the request.
-  next();
-};

routes/index.js

@@ -10,7 +10,6 @@ const i18n = require('../middleware/i18n');
 const path = require('path');
 const plugins = require('../services/plugins');
 const staticTemplate = require('../middleware/staticTemplate');
-const pubsub = require('../middleware/pubsub');
 const staticMiddleware = require('express-static-gzip');
 const { DISABLE_STATIC_SERVER } = require('../config');
 const { createGraphOptions } = require('../graph');
@@ -102,7 +101,7 @@ router.use(passport.initialize());
 
 // Attach the authentication middleware, this will be responsible for decoding
 // (if present) the JWT on the request.
-router.use('/api', authentication, pubsub);
+router.use('/api', authentication);
 
 //==============================================================================
 // GraphQL Router

services/pubsub.js

@@ -1,24 +0,0 @@
-const { RedisPubSub } = require('graphql-redis-subscriptions');
-const { createClient } = require('./redis');
-
-/**
- * getClient returns the pubsub singleton for this instance.
- */
-let pubsub = null;
-const getClient = () => {
-  if (pubsub !== null) {
-    return pubsub;
-  }
-
-  // Create the new PubSub client, we only need one per instance of Talk.
-  pubsub = new RedisPubSub({
-    publisher: createClient(),
-    subscriber: createClient(),
-  });
-
-  return pubsub;
-};
-
-module.exports = {
-  getClient,
-};