From c15cff712102055c0749497c699d9ab923d1e4fa Mon Sep 17 00:00:00 2001 From: Wyatt Johnson Date: Wed, 31 Jan 2018 09:55:54 -0700 Subject: [PATCH] subscription refactor --- graph/connectors.js | 2 - graph/context.js | 4 +- graph/subscriptions/broker.js | 48 +++++++++++++++++++ .../index.js} | 16 +++---- graph/subscriptions/pubsub.js | 29 +++++++++++ graph/{ => subscriptions}/setupFunctions.js | 4 +- middleware/pubsub.js | 11 ----- routes/index.js | 3 +- services/pubsub.js | 24 ---------- 9 files changed, 90 insertions(+), 51 deletions(-) create mode 100644 graph/subscriptions/broker.js rename graph/{subscriptions.js => subscriptions/index.js} (87%) create mode 100644 graph/subscriptions/pubsub.js rename graph/{ => subscriptions}/setupFunctions.js (98%) delete mode 100644 middleware/pubsub.js delete mode 100644 services/pubsub.js diff --git a/graph/connectors.js b/graph/connectors.js index 976f05b71..c0ae9606e 100644 --- a/graph/connectors.js +++ b/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, diff --git a/graph/context.js b/graph/context.js index 35d84dd82..f45e45da1 100644 --- a/graph/context.js +++ b/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; diff --git a/graph/subscriptions/broker.js b/graph/subscriptions/broker.js new file mode 100644 index 000000000..93f57d219 --- /dev/null +++ b/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; diff --git a/graph/subscriptions.js b/graph/subscriptions/index.js similarity index 87% rename from graph/subscriptions.js rename to graph/subscriptions/index.js index 948d10a8f..5ade76feb 100644 --- a/graph/subscriptions.js +++ b/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, diff --git a/graph/subscriptions/pubsub.js b/graph/subscriptions/pubsub.js new file mode 100644 index 000000000..54761bdfb --- /dev/null +++ b/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; diff --git a/graph/setupFunctions.js b/graph/subscriptions/setupFunctions.js similarity index 98% rename from graph/setupFunctions.js rename to graph/subscriptions/setupFunctions.js index 12d9d5685..467c4e98f 100644 --- a/graph/setupFunctions.js +++ b/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) => { diff --git a/middleware/pubsub.js b/middleware/pubsub.js deleted file mode 100644 index 2b7291bf3..000000000 --- a/middleware/pubsub.js +++ /dev/null @@ -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(); -}; diff --git a/routes/index.js b/routes/index.js index 48981faff..902a2138e 100644 --- a/routes/index.js +++ b/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 diff --git a/services/pubsub.js b/services/pubsub.js deleted file mode 100644 index 3bfb9aaa3..000000000 --- a/services/pubsub.js +++ /dev/null @@ -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, -};