diff --git a/bin/cli-serve b/bin/cli-serve index eef6526bd..eca23303b 100755 --- a/bin/cli-serve +++ b/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) { diff --git a/graph/index.js b/graph/index.js index 3413f2d78..4e99714d9 100644 --- a/graph/index.js +++ b/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 }; diff --git a/graph/pubsub.js b/graph/pubsub.js new file mode 100644 index 000000000..a5ee95b80 --- /dev/null +++ b/graph/pubsub.js @@ -0,0 +1,5 @@ +const {RedisPubSub} = require('graphql-redis-subscriptions'); + +const {connectionOptions} = require('../services/redis'); + +module.exports = new RedisPubSub(connectionOptions); diff --git a/graph/setupFunctions.js b/graph/setupFunctions.js deleted file mode 100644 index ed4d6ffa9..000000000 --- a/graph/setupFunctions.js +++ /dev/null @@ -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); diff --git a/graph/subscriptions.js b/graph/subscriptions.js new file mode 100644 index 000000000..369b5c058 --- /dev/null +++ b/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 +}; diff --git a/package.json b/package.json index 75e729350..723c6dece 100644 --- a/package.json +++ b/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/* .", diff --git a/services/subscriptions.js b/services/subscriptions.js index eaf2f9b23..8c34507f2 100644 --- a/services/subscriptions.js +++ b/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 };