diff --git a/bin/cli.js b/bin/cli.js index f8462890..f9733d35 100755 --- a/bin/cli.js +++ b/bin/cli.js @@ -1,21 +1,21 @@ #!/usr/bin/env node -var fs = require('fs-extra'); -var path = require('path'); -var argv = require('minimist')(process.argv.slice(2)); -var childProcess = require('child_process'); -var inquirer = require('inquirer'); -var prompt = inquirer.createPromptModule(); -var exec = childProcess.exec; -var spawn = childProcess.spawn; -var fork = childProcess.fork; - -var command = argv._[0]; -var commandRawArgs = process.argv.slice(3); -var arg1 = argv._[1]; -var force = argv.force ? true : false; - -var fileExistsSync = function (filePath) { +const fs = require('fs-extra'); +const path = require('path'); +const argv = require('minimist')(process.argv.slice(2)); +const childProcess = require('child_process'); +const inquirer = require('inquirer'); +const prompt = inquirer.createPromptModule(); +const exec = childProcess.exec; +const spawn = childProcess.spawn; +const fork = childProcess.fork; + +let command = argv._[0]; +let commandRawArgs = process.argv.slice(3); +let arg1 = argv._[1]; +let force = argv.force ? true : false; + +let fileExistsSync = function (filePath) { try { fs.accessSync(filePath, fs.constants.F_OK); } catch (err) { @@ -24,8 +24,8 @@ var fileExistsSync = function (filePath) { return true; }; -var parsePackageFile = function (moduleDir) { - var packageFile = moduleDir + '/package.json'; +let parsePackageFile = function (moduleDir) { + let packageFile = moduleDir + '/package.json'; try { if (fileExistsSync(packageFile)) { return JSON.parse(fs.readFileSync(packageFile, {encoding: 'utf8'})); @@ -35,19 +35,19 @@ var parsePackageFile = function (moduleDir) { return {}; }; -var errorMessage = function (message) { +let errorMessage = function (message) { console.log('\x1b[31m[Error]\x1b[0m ' + message); }; -var successMessage = function (message) { +let successMessage = function (message) { console.log('\x1b[32m[Success]\x1b[0m ' + message); }; -var warningMessage = function (message) { +let warningMessage = function (message) { console.log('\x1b[33m[Warning]\x1b[0m ' + message); }; -var showCorrectUsage = function () { +let showCorrectUsage = function () { console.log('Usage: socketcluster [options] [command]\n'); console.log('Options:'); console.log(" -v Get the version of the current SocketCluster installation"); @@ -58,15 +58,15 @@ var showCorrectUsage = function () { console.log(' create Create a new boilerplate app in working directory'); }; -var failedToRemoveDirMessage = function (dirPath) { +let failedToRemoveDirMessage = function (dirPath) { errorMessage('Failed to remove existing directory at ' + dirPath + '. This directory may be used by another program or you may not have the permission to remove it.'); }; -var failedToCreateMessage = function () { +let failedToCreateMessage = function () { errorMessage('Failed to create necessary files. Please check your permissions and try again.'); }; -var promptConfirm = function (message, callback) { +let promptConfirm = function (message, callback) { prompt([ { type: 'confirm', @@ -81,7 +81,7 @@ var promptConfirm = function (message, callback) { }); }; -var copyDirRecursive = function (src, dest) { +let copyDirRecursive = function (src, dest) { try { fs.copySync(src, dest); return true; @@ -91,7 +91,7 @@ var copyDirRecursive = function (src, dest) { return false; }; -var rmdirRecursive = function (dirname) { +let rmdirRecursive = function (dirname) { try { fs.removeSync(dirname); return true; @@ -107,34 +107,34 @@ if (argv.help) { }; if (argv.v) { - var scDir = __dirname + '/../'; - var scPkg = parsePackageFile(scDir); + let scDir = __dirname + '/../'; + let scPkg = parsePackageFile(scDir); console.log('v' + scPkg.version); process.exit(); }; -var wd = process.cwd(); +let wd = process.cwd(); -var sampleDir = __dirname + '/../sample'; -var destDir = path.normalize(wd + '/' + arg1); -var clientFileSourcePath = path.normalize(destDir + '/node_modules/socketcluster-client/socketcluster.js'); -var clientFileDestPath = path.normalize(destDir + '/public/socketcluster.js'); +let sampleDir = __dirname + '/../sample'; +let destDir = path.normalize(wd + '/' + arg1); +let clientFileSourcePath = path.normalize(destDir + '/node_modules/socketcluster-client/socketcluster.js'); +let clientFileDestPath = path.normalize(destDir + '/public/socketcluster.js'); -var createFail = function () { +let createFail = function () { errorMessage("Failed to create SocketCluster sample app."); process.exit(); }; -var createSuccess = function () { +let createSuccess = function () { console.log('Installing app dependencies using npm. This could take a while...'); - var npmCommand = (process.platform === "win32" ? "npm.cmd" : "npm"); - var options = { + let npmCommand = (process.platform === "win32" ? "npm.cmd" : "npm"); + let options = { cwd: destDir, maxBuffer: Infinity }; - var npmProcess = spawn(npmCommand, ['install'], options); + let npmProcess = spawn(npmCommand, ['install'], options); npmProcess.stdout.on('data', function (data) { process.stdout.write(data); @@ -162,11 +162,11 @@ var createSuccess = function () { npmProcess.stdin.end(); }; -var setupMessage = function () { +let setupMessage = function () { console.log('Creating app structure...'); }; -var confirmReplaceSetup = function (confirm) { +let confirmReplaceSetup = function (confirm) { if (confirm) { setupMessage(); if (rmdirRecursive(destDir) && copyDirRecursive(sampleDir, destDir)) { @@ -180,13 +180,13 @@ var confirmReplaceSetup = function (confirm) { } }; -if (command == 'create') { +if (command === 'create') { if (arg1) { if (fileExistsSync(destDir)) { if (force) { confirmReplaceSetup(true); } else { - var message = "There is already a directory at " + destDir + '. Do you want to overwrite it?'; + let message = "There is already a directory at " + destDir + '. Do you want to overwrite it?'; promptConfirm(message, confirmReplaceSetup); } } else { diff --git a/default-workercluster-controller.js b/default-workercluster-controller.js index c69f1340..8b78eb43 100644 --- a/default-workercluster-controller.js +++ b/default-workercluster-controller.js @@ -1,2 +1,2 @@ -var SCWorkerCluster = require('./scworkercluster'); +const SCWorkerCluster = require('./scworkercluster'); new SCWorkerCluster(); diff --git a/fsutil.js b/fsutil.js index da32a689..8f4b6d41 100644 --- a/fsutil.js +++ b/fsutil.js @@ -1,28 +1,28 @@ -var fs = require('fs'); -var scErrors = require('sc-errors'); -var TimeoutError = scErrors.TimeoutError; +const fs = require('fs'); +const scErrors = require('sc-errors'); +const TimeoutError = scErrors.TimeoutError; -var fileExists = function (filePath, callback) { +let fileExists = function (filePath, callback) { fs.access(filePath, fs.constants.F_OK, (err) => { callback(!err); }); }; -var waitForFile = function (filePath, checkInterval, startTime, maxWaitDuration, timeoutErrorMessage) { +let waitForFile = function (filePath, checkInterval, startTime, maxWaitDuration, timeoutErrorMessage) { return new Promise((resolve, reject) => { if (!filePath) { resolve(); return; } - var checkIsReady = () => { - var now = Date.now(); + let checkIsReady = () => { + let now = Date.now(); fileExists(filePath, (exists) => { if (exists) { resolve(); } else { if (now - startTime >= maxWaitDuration) { - var errorMessage; + let errorMessage; if (timeoutErrorMessage != null) { errorMessage = timeoutErrorMessage; @@ -30,7 +30,7 @@ var waitForFile = function (filePath, checkInterval, startTime, maxWaitDuration, errorMessage = `Could not find a file at path ${filePath} ` + `before the timeout was reached`; } - var volumeBootTimeoutError = new TimeoutError(errorMessage); + let volumeBootTimeoutError = new TimeoutError(errorMessage); reject(volumeBootTimeoutError); } else { setTimeout(checkIsReady, checkInterval); @@ -43,6 +43,6 @@ var waitForFile = function (filePath, checkInterval, startTime, maxWaitDuration, }; module.exports = { - fileExists: fileExists, - waitForFile: waitForFile + fileExists, + waitForFile }; diff --git a/index.js b/index.js index 7b82ac17..d543c526 100644 --- a/index.js +++ b/index.js @@ -1,84 +1,81 @@ -var path = require('path'); -var crypto = require('crypto'); -var EventEmitter = require('events').EventEmitter; -var uuid = require('uuid'); -var fork = require('child_process').fork; -var os = require('os'); -var fs = require('fs-extra'); -var uidNumber = require('uid-number'); -var pkg = require('./package.json'); -var argv = require('minimist')(process.argv.slice(2)); -var cluster = require('cluster'); - -var scErrors = require('sc-errors'); -var InvalidOptionsError = scErrors.InvalidOptionsError; -var InvalidActionError = scErrors.InvalidActionError; -var ProcessExitError = scErrors.ProcessExitError; -var UnknownError = scErrors.UnknownError; -var TimeoutError = scErrors.TimeoutError; -var decycle = scErrors.decycle; - -var socketClusterSingleton = null; +const path = require('path'); +const crypto = require('crypto'); +const AsyncStreamEmitter = require('async-stream-emitter'); +const uuid = require('uuid'); +const fork = require('child_process').fork; +const os = require('os'); +const fs = require('fs-extra'); +const uidNumber = require('uid-number'); +const pkg = require('./package.json'); +const argv = require('minimist')(process.argv.slice(2)); +const cluster = require('cluster'); + +const scErrors = require('sc-errors'); +const InvalidOptionsError = scErrors.InvalidOptionsError; +const InvalidActionError = scErrors.InvalidActionError; +const ProcessExitError = scErrors.ProcessExitError; +const UnknownError = scErrors.UnknownError; +const TimeoutError = scErrors.TimeoutError; +const decycle = scErrors.decycle; + +let socketClusterSingleton = null; function SocketCluster(options) { - var self = this; + AsyncStreamEmitter.call(this); + if (socketClusterSingleton) { - var errorMessage = 'The SocketCluster master object is a singleton; ' + + let errorMessage = 'The SocketCluster master object is a singleton; ' + 'it can only be instantiated once per process'; throw new InvalidActionError(errorMessage); } - socketClusterSingleton = self; - - self.EVENT_FAIL = 'fail'; - self.EVENT_WARNING = 'warning'; - self.EVENT_INFO = 'info'; - self.EVENT_READY = 'ready'; - self.EVENT_WORKER_START = 'workerStart'; - self.EVENT_WORKER_EXIT = 'workerExit'; - self.EVENT_BROKER_START = 'brokerStart'; - self.EVENT_BROKER_EXIT = 'brokerExit'; - self.EVENT_WORKER_CLUSTER_START = 'workerClusterStart'; - self.EVENT_WORKER_CLUSTER_READY = 'workerClusterReady'; - self.EVENT_WORKER_CLUSTER_EXIT = 'workerClusterExit'; + socketClusterSingleton = this; - self._pendingResponseHandlers = {}; - self._destroyCallbacks = []; - self.workerClusterMessageBuffer = []; + this._pendingResponseHandlers = {}; + this.workerClusterMessageBuffer = []; - self._shuttingDown = false; + this.isShuttingDown = false; - self._errorAnnotations = { + this._errorAnnotations = { 'EADDRINUSE': 'Failed to bind to a port because it was already used by another process.' }; - self.on('error', function (error) { - self.emitFail(error, { - type: 'Master', - pid: process.pid - }); - }); - // Capture any errors that are thrown during initialization. - new Promise(function () { - self._init(options); - }).catch(function (error) { - self.emit('error', error); - }); + (async () => { + try { + await this._init(options); + } catch (error) { + this.emitError(error, { + type: 'Master', + pid: process.pid + }); + } + })(); } +SocketCluster.prototype = Object.create(AsyncStreamEmitter.prototype); + +SocketCluster.EVENT_ERROR = SocketCluster.prototype.EVENT_ERROR = 'error'; +SocketCluster.EVENT_WARNING = SocketCluster.prototype.EVENT_WARNING = 'warning'; +SocketCluster.EVENT_INFO = SocketCluster.prototype.EVENT_INFO = 'info'; +SocketCluster.EVENT_READY = SocketCluster.prototype.EVENT_READY = 'ready'; +SocketCluster.EVENT_WORKER_START = SocketCluster.prototype.EVENT_WORKER_START = 'workerStart'; +SocketCluster.EVENT_WORKER_EXIT = SocketCluster.prototype.EVENT_WORKER_EXIT = 'workerExit'; +SocketCluster.EVENT_WORKER_MESSAGE = SocketCluster.prototype.EVENT_WORKER_MESSAGE = 'workerMessage'; +SocketCluster.EVENT_WORKER_REQUEST = SocketCluster.prototype.EVENT_WORKER_REQUEST = 'workerRequest'; +SocketCluster.EVENT_BROKER_START = SocketCluster.prototype.EVENT_BROKER_START = 'brokerStart'; +SocketCluster.EVENT_BROKER_EXIT = SocketCluster.prototype.EVENT_BROKER_EXIT = 'brokerExit'; +SocketCluster.EVENT_BROKER_MESSAGE = SocketCluster.prototype.EVENT_BROKER_MESSAGE = 'brokerMessage'; +SocketCluster.EVENT_BROKER_REQUEST = SocketCluster.prototype.EVENT_BROKER_REQUEST = 'brokerRequest'; +SocketCluster.EVENT_WORKER_CLUSTER_START = SocketCluster.prototype.EVENT_WORKER_CLUSTER_START = 'workerClusterStart'; +SocketCluster.EVENT_WORKER_CLUSTER_READY = SocketCluster.prototype.EVENT_WORKER_CLUSTER_READY = 'workerClusterReady'; +SocketCluster.EVENT_WORKER_CLUSTER_EXIT = SocketCluster.prototype.EVENT_WORKER_CLUSTER_EXIT = 'workerClusterExit'; + SocketCluster.create = function (options) { return new SocketCluster(options); }; -SocketCluster.prototype = Object.create(EventEmitter.prototype); - -SocketCluster.prototype._init = function (options) { - var self = this; - - var backslashRegex = /\\/g; - this.appDirPath = path.dirname(require.main.filename).replace(backslashRegex, '/'); - - self.options = { +SocketCluster.prototype._init = async function (options) { + this.options = { port: 8000, workers: null, brokers: null, @@ -94,7 +91,7 @@ SocketCluster.prototype._init = function (options) { authSignAsync: false, authVerifyAsync: true, crashWorkerOnError: true, - rebootWorkerOnCrash: true, + respawnWorkerOnCrash: true, killWorkerMemoryThreshold: null, protocol: 'http', protocolOptions: null, @@ -119,6 +116,9 @@ SocketCluster.prototype._init = function (options) { workerController: null, brokerController: null, brokerConnectRetryErrorThreshold: null, + brokerCommandAckTimeout: null, + brokerPubSubAckTimeout: null, + brokerAutoReconnectOptions: null, workerClusterController: null, rebootOnSignal: true, downgradeToUser: false, @@ -135,24 +135,25 @@ SocketCluster.prototype._init = function (options) { brokerEngine: 'sc-broker-cluster' }; - self._active = false; - self.workerCluster = null; - self.isWorkerClusterReady = false; + this.isActive = false; + this.workerCluster = null; + this.isWorkerClusterReady = false; - self._colorCodes = { + this._colorCodes = { red: 31, green: 32, yellow: 33 }; - Object.assign(self.options, options); + Object.assign(this.options, options); - var maxTimeout = Math.pow(2, 31) - 1; + let maxTimeout = Math.pow(2, 31) - 1; - var verifyDuration = function (propertyName) { - if (self.options[propertyName] > maxTimeout) { - throw new InvalidOptionsError('The ' + propertyName + - ' value provided exceeded the maximum amount allowed'); + let verifyDuration = (propertyName) => { + if (this.options[propertyName] > maxTimeout) { + throw new InvalidOptionsError( + `The ${propertyName} value provided exceeded the maximum amount allowed` + ); } }; @@ -164,45 +165,51 @@ SocketCluster.prototype._init = function (options) { verifyDuration('processTermTimeout'); verifyDuration('forceKillTimeout'); - if (self.options.appName == null) { - self.options.appName = uuid.v4(); + this.appDirPath = path.dirname(require.main.filename).replace(/\\/g, '/'); + + if (this.options.appName == null) { + this.options.appName = uuid.v4(); } - if (self.options.run != null) { - self.run = self.options.run; + if (this.options.run != null) { + this.run = this.options.run; } - if (self.options.workerController == null) { - throw new InvalidOptionsError("Compulsory option 'workerController' was not specified " + - "- It needs to be a path to a JavaScript file which will act as the " + - "boot controller for each worker in the cluster"); + if (this.options.workerController == null) { + throw new InvalidOptionsError( + 'Compulsory option "workerController" was not specified ' + + '- It needs to be a path to a JavaScript file which will act as the ' + + 'boot controller for each worker in the cluster' + ); } - var pathHasher = crypto.createHash('md5'); - pathHasher.update(self.appDirPath, 'utf8'); - var pathHash = pathHasher.digest('hex').substr(0, 10); + let pathHasher = crypto.createHash('md5'); + pathHasher.update(this.appDirPath, 'utf8'); + let pathHash = pathHasher.digest('hex').substr(0, 10); // Trim it because some OSes (e.g. OSX) do not like long path names for domain sockets. - var shortAppName = self.options.appName.substr(0, 13); + let shortAppName = this.options.appName.substr(0, 13); - if (process.platform == 'win32') { - if (self.options.socketRoot) { - self._socketDirPath = self.options.socketRoot + '_'; + if (process.platform === 'win32') { + if (this.options.socketRoot) { + this._socketDirPath = `${this.options.socketRoot}_`; } else { - self._socketDirPath = '\\\\.\\pipe\\socketcluster_' + shortAppName + '_' + pathHash + '_'; + this._socketDirPath = `\\\\.\\pipe\\socketcluster_${shortAppName}_${pathHash}_`; } } else { - var socketDir, socketParentDir; - if (self.options.socketRoot) { - socketDir = self.options.socketRoot.replace(/\/$/, '') + '/'; + let socketDir, socketParentDir; + if (this.options.socketRoot) { + socketDir = this.options.socketRoot.replace(/\/$/, '') + '/'; } else { - socketParentDir = os.tmpdir() + '/socketcluster/'; - socketDir = socketParentDir + shortAppName + '_' + pathHash + '/'; + socketParentDir = path.join(os.tmpdir(), '/socketcluster/'); + socketDir = path.join(socketParentDir, `${shortAppName}_${pathHash}/`); } - if (self._fileExistsSync(socketDir)) { + if (this._fileExistsSync(socketDir)) { try { fs.removeSync(socketDir); } catch (err) { - throw new InvalidActionError('Failed to remove old socket directory ' + socketDir + ' - Try removing it manually'); + throw new InvalidActionError( + `Failed to remove old socket directory ${socketDir} - Try removing it manually` + ); } } fs.mkdirsSync(socketDir); @@ -211,11 +218,11 @@ SocketCluster.prototype._init = function (options) { fs.chmodSync(socketParentDir, '1777'); } catch (err) {} } - self._socketDirPath = socketDir; + this._socketDirPath = socketDir; } - if (self.options.protocolOptions) { - var protoOpts = self.options.protocolOptions; + if (this.options.protocolOptions) { + let protoOpts = this.options.protocolOptions; if (protoOpts.key instanceof Buffer) { protoOpts.key = protoOpts.key.toString(); } @@ -224,7 +231,7 @@ SocketCluster.prototype._init = function (options) { } if (protoOpts.ca) { if (protoOpts.ca instanceof Array) { - protoOpts.ca = protoOpts.ca.map(function (item) { + protoOpts.ca = protoOpts.ca.map((item) => { if (item instanceof Buffer) { return item.toString(); } else { @@ -240,108 +247,118 @@ SocketCluster.prototype._init = function (options) { } if (protoOpts.passphrase == null) { if (protoOpts.key) { - var privKeyEncLine = protoOpts.key.split('\n')[1]; + let privKeyEncLine = protoOpts.key.split('\n')[1]; if (privKeyEncLine && privKeyEncLine.toUpperCase().indexOf('ENCRYPTED') > -1) { - var message = 'The supplied private key is encrypted and cannot be used without a passphrase - ' + + let message = 'The supplied private key is encrypted and cannot be used without a passphrase - ' + 'Please provide a valid passphrase as a property to protocolOptions'; throw new InvalidOptionsError(message); } } else if (protoOpts.pfx) { - var message = 'The supplied pfx certificate cannot be used without a passphrase - ' + + let message = 'The supplied pfx certificate cannot be used without a passphrase - ' + 'Please provide a valid passphrase as a property to protocolOptions'; throw new InvalidOptionsError(message); } else { - var message = 'The supplied protocolOptions were invalid - ' + + let message = 'The supplied protocolOptions were invalid - ' + 'Please provide either a key and cert pair or a pfx certificate'; throw new InvalidOptionsError(message); } } } - if (self.options.authPrivateKey instanceof Buffer) { - self.options.authPrivateKey = self.options.authPrivateKey.toString(); + if (this.options.authPrivateKey instanceof Buffer) { + this.options.authPrivateKey = this.options.authPrivateKey.toString(); } - if (self.options.authPublicKey instanceof Buffer) { - self.options.authPublicKey = self.options.authPublicKey.toString(); + if (this.options.authPublicKey instanceof Buffer) { + this.options.authPublicKey = this.options.authPublicKey.toString(); } - if (!self.options.brokers || self.options.brokers < 1) { - self.options.brokers = 1; + if (!this.options.brokers || this.options.brokers < 1) { + this.options.brokers = 1; } - if (typeof self.options.brokers != 'number') { + if (typeof this.options.brokers !== 'number') { throw new InvalidOptionsError('The brokers option must be a number'); } - if (!self.options.workers || self.options.workers < 1) { - self.options.workers = 1; + if (!this.options.workers || this.options.workers < 1) { + this.options.workers = 1; } - if (typeof self.options.workers != 'number') { + if (typeof this.options.workers !== 'number') { throw new InvalidOptionsError('The workers option must be a number'); } - self._extRegex = /[.][^\/\\]*$/; - self._slashSequenceRegex = /\/+/g; - self._startSlashRegex = /^\//; + this._extRegex = /[.][^\/\\]*$/; + this._slashSequenceRegex = /\/+/g; + this._startSlashRegex = /^\//; - self._dataExpiryAccuracy = 5000; + this._dataExpiryAccuracy = 5000; - self._brokerEngine = require(self.options.brokerEngine); + this._brokerEngine = require(this.options.brokerEngine); - if (self.options.logLevel > 0) { - console.log(' ' + self.colorText('[Busy]', 'yellow') + ' Launching SocketCluster'); + if (this.options.logLevel > 1) { + console.log(' ' + this.colorText('[Busy]', 'yellow') + ' Launching SocketCluster'); } - self._stdinErrorHandler = function (err) { - self.emitWarning(err, { + this._stdinErrorHandler = (err) => { + this.emitWarning(err, { type: 'Master', pid: process.pid }); }; - process.stdin.on('error', self._stdinErrorHandler); + process.stdin.on('error', this._stdinErrorHandler); /* To allow inserting blank lines in console on Windows to aid with debugging. */ - if (process.platform == 'win32') { + if (process.platform === 'win32') { process.stdin.resume(); process.stdin.setEncoding('utf8'); } - if (self.options.secretKey == null) { - self.options.secretKey = crypto.randomBytes(32).toString('hex'); + if (this.options.secretKey == null) { + this.options.secretKey = crypto.randomBytes(32).toString('hex'); } - if (self.options.authKey == null && self.options.authPrivateKey == null && self.options.authPublicKey == null) { - self.options.authKey = crypto.randomBytes(32).toString('hex'); + if (this.options.authKey == null && this.options.authPrivateKey == null && this.options.authPublicKey == null) { + this.options.authKey = crypto.randomBytes(32).toString('hex'); } - if (self.options.instanceId == null) { - self.options.instanceId = uuid.v4(); + if (this.options.instanceId == null) { + this.options.instanceId = uuid.v4(); } - if (self.options.downgradeToUser && process.platform != 'win32') { - if (typeof self.options.downgradeToUser == 'number') { - fs.chownSync(self._socketDirPath, self.options.downgradeToUser, 0); - self._start(); + if (this.options.downgradeToUser && process.platform !== 'win32') { + if (typeof this.options.downgradeToUser === 'number') { + fs.chownSync(this._socketDirPath, this.options.downgradeToUser, 0); + this._start(); } else { - uidNumber(self.options.downgradeToUser, function (err, uid, gid) { - if (self._shuttingDown) { - return; - } - if (err) { - throw new InvalidActionError('Failed to downgrade to user "' + self.options.downgradeToUser + '" - ' + err); - } else { - fs.chownSync(self._socketDirPath, uid, gid); - self._start(); - } - }); + let uidInfo; + try { + uidInfo = await new Promise((resolve, reject) => { + uidNumber(this.options.downgradeToUser, (err, uid, gid) => { + if (err) { + reject(err); + return; + } + resolve({uid, gid}); + }); + }); + } catch (err) { + throw new InvalidActionError( + `Failed to downgrade to user "${this.options.downgradeToUser}" - ${err}` + ); + } + if (this.isShuttingDown) { + return; + } + fs.chownSync(this._socketDirPath, uidInfo.uid, uidInfo.gid); + this._start(); } } else { - self._start(); + this._start(); } }; SocketCluster.prototype._getPaths = function () { - var paths = { + let paths = { appDirPath: this.appDirPath, statusURL: '/~status', appWorkerControllerPath: path.resolve(this.options.workerController) @@ -382,8 +399,8 @@ SocketCluster.prototype._getBrokerSocketName = function (brokerId) { }; SocketCluster.prototype._getBrokerSocketPaths = function () { - var socketPaths = []; - for (var i = 0; i < this.options.brokers; i++) { + let socketPaths = []; + for (let i = 0; i < this.options.brokers; i++) { socketPaths.push(this._socketDirPath + this._getBrokerSocketName(i)); } return socketPaths; @@ -406,28 +423,28 @@ SocketCluster.prototype._logObject = function (obj, objType, time) { if (!obj.origin.type) { obj.origin.type = 'Undefined'; } - var output = obj.stack || obj.message || obj; + let output = obj.stack || obj.message || obj; - var logMessage; + let logMessage; if (obj.origin.pid == null) { - logMessage = 'Origin: ' + this._capitaliseFirstLetter(obj.origin.type) + '\n' + - ' [' + objType + '] ' + output; + logMessage = `Origin: ${this._capitaliseFirstLetter(obj.origin.type)}\n` + + ` [${objType}] ${output}`; } else { - logMessage = 'Origin: ' + this._capitaliseFirstLetter(obj.origin.type) + ' (PID ' + obj.origin.pid + ')\n' + - ' [' + objType + '] ' + output; + logMessage = `Origin: ${this._capitaliseFirstLetter(obj.origin.type)} (PID ${obj.origin.pid})\n` + + ` [${objType}] ${output}`; } this.log(logMessage, time); }; SocketCluster.prototype._convertValueToUnknownError = function (err, origin) { if (!(err instanceof Error)) { - if (err && typeof err == 'object') { + if (err && typeof err === 'object') { if (err.message || err.stack) { err = scErrors.hydrateError(err, true); } else { // If err has neither a stack nor a message property // then the error message will be the JSON stringified object. - var errorMessage; + let errorMessage; try { errorMessage = JSON.stringify(err); } catch (e1) { @@ -439,10 +456,10 @@ SocketCluster.prototype._convertValueToUnknownError = function (err, origin) { } err = new UnknownError(errorMessage); } - } else if (typeof err == 'function') { - var errorMessage = '[function ' + (err.name || 'anonymous') + ']'; + } else if (typeof err === 'function') { + let errorMessage = `[function ${err.name || 'anonymous'}]`; err = new UnknownError(errorMessage); - } else if (typeof err == 'undefined') { + } else if (typeof err === 'undefined') { err = new UnknownError('undefined'); } else if (err === null) { err = new UnknownError('null'); @@ -458,61 +475,73 @@ SocketCluster.prototype._convertValueToUnknownError = function (err, origin) { return err; }; -SocketCluster.prototype.emitFail = function (err, origin) { - err = this._convertValueToUnknownError(err, origin); +SocketCluster.prototype.emitError = function (error, origin) { + error = this._convertValueToUnknownError(error, origin); - var annotation = this._errorAnnotations[err.code]; - if (annotation && err.stack) { - err.stack += '\n ' + this.colorText('!!', 'red') + ' ' + annotation; + let annotation = this._errorAnnotations[error.code]; + if (annotation && error.stack) { + error.stack += `\n ${this.colorText('!!', 'red')} ${annotation}`; } - this.emit(this.EVENT_FAIL, err); + this.emit(this.EVENT_ERROR, {error}); - if (this.options.logLevel > 0 && !this._shuttingDown) { - this._logObject(err, 'Error'); + if (this.options.logLevel > 0 && !this.isShuttingDown) { + this._logObject(error, 'Error'); } }; SocketCluster.prototype.emitWarning = function (warning, origin) { warning = this._convertValueToUnknownError(warning, origin); - this.emit(this.EVENT_WARNING, warning); + this.emit(this.EVENT_WARNING, {warning}); - if (this.options.logLevel > 1 && !this._shuttingDown) { + if (this.options.logLevel > 1 && !this.isShuttingDown) { this._logObject(warning, 'Warning'); } }; +SocketCluster.prototype.emitInfo = function (message, origin) { + let info = { + message, + origin + }; + this.emit(this.EVENT_INFO, info); + + if (this.options.logLevel > 2 && !this.isShuttingDown) { + this._logObject(info, 'Info'); + } +}; + SocketCluster.prototype._workerClusterErrorHandler = function (pid, error) { - this.emitFail(error, { + this.emitError(error, { type: 'WorkerCluster', pid: pid }); }; SocketCluster.prototype._workerErrorHandler = function (workerPid, error) { - this.emitFail(error, { + this.emitError(error, { type: 'Worker', pid: workerPid }); }; SocketCluster.prototype._brokerEngineErrorHandler = function (pid, error) { - this.emitFail(error, { + this.emitError(error, { type: 'BrokerEngine', pid: pid }); }; SocketCluster.prototype._brokerErrorHandler = function (brokerPid, error) { - this.emitFail(error, { + this.emitError(error, { type: 'Broker', pid: brokerPid }); }; SocketCluster.prototype._workerWarningHandler = function (workerPid, warning) { - var origin = { + let origin = { type: 'Worker', pid: workerPid }; @@ -520,29 +549,27 @@ SocketCluster.prototype._workerWarningHandler = function (workerPid, warning) { }; SocketCluster.prototype._workerClusterReadyHandler = function () { - var self = this; - - if (!this._active) { + if (!this.isActive) { if (this.options.rebootOnSignal) { - this._sigusr2SignalHandler = function () { - var warningMessage; - var killOptions = {}; - if (self.options.environment == 'dev') { + this._sigusr2SignalHandler = () => { + let warningMessage; + let killOptions = {}; + if (this.options.environment === 'dev') { warningMessage = 'Master received SIGUSR2 signal - Shutting down all workers immediately'; killOptions.immediate = true; } else { warningMessage = 'Master received SIGUSR2 signal - Shutting down all workers gracefully within processTermTimeout limit'; } - var warning = new ProcessExitError(warningMessage, null); + let warning = new ProcessExitError(warningMessage, null); warning.signal = 'SIGUSR2'; - self.emitWarning(warning, { + this.emitWarning(warning, { type: 'Master', pid: process.pid }); - self.killWorkers(killOptions); - if (self.options.killMasterOnSignal) { + this.killWorkers(killOptions); + if (this.options.killMasterOnSignal) { process.exit(); } }; @@ -550,17 +577,17 @@ SocketCluster.prototype._workerClusterReadyHandler = function () { process.on('SIGUSR2', this._sigusr2SignalHandler); } - this._active = true; + this.isActive = true; this._logDeploymentDetails(); - this.emit(this.EVENT_READY); + this.emit(this.EVENT_READY, {}); this.run(); } this.isWorkerClusterReady = true; this._flushWorkerClusterMessageBuffer(); - var workerClusterInfo = { + let workerClusterInfo = { pid: this.workerCluster.pid, childProcess: this.workerCluster }; @@ -569,9 +596,9 @@ SocketCluster.prototype._workerClusterReadyHandler = function () { SocketCluster.prototype._workerExitHandler = function (workerInfo) { if (this.options.logLevel > 0) { - var message = 'Worker ' + workerInfo.id + ' exited - Exit code: ' + workerInfo.code; + let message = `Worker ${workerInfo.id} exited - Exit code: ${workerInfo.code}`; if (workerInfo.signal) { - message += ', signal: ' + workerInfo.signal; + message += `, signal: ${workerInfo.signal}`; } this.log(message); } @@ -579,8 +606,8 @@ SocketCluster.prototype._workerExitHandler = function (workerInfo) { }; SocketCluster.prototype._workerStartHandler = function (workerInfo, signal) { - if (this._active && this.options.logLevel > 0 && workerInfo.respawn) { - this.log('Worker ' + workerInfo.id + ' was respawned'); + if (this.isActive && this.options.logLevel > 0 && workerInfo.respawn) { + this.log(`Worker ${workerInfo.id} was respawned`); } this.emit(this.EVENT_WORKER_START, workerInfo); }; @@ -589,9 +616,9 @@ SocketCluster.prototype._handleWorkerClusterExit = function (errorCode, signal) this.isWorkerClusterReady = false; this.workerClusterMessageBuffer = []; - var wcPid = this.workerCluster.pid; + let wcPid = this.workerCluster.pid; - var workerClusterInfo = { + let workerClusterInfo = { pid: wcPid, code: errorCode, signal: signal, @@ -600,50 +627,49 @@ SocketCluster.prototype._handleWorkerClusterExit = function (errorCode, signal) this.emit(this.EVENT_WORKER_CLUSTER_EXIT, workerClusterInfo); - var message = 'WorkerCluster exited with code ' + errorCode; + let message = `WorkerCluster exited with code ${errorCode}`; if (signal != null) { - message += ' and signal ' + signal; + message += ` and signal ${signal}`; } - if (errorCode == 0) { - this.log(message); + if (errorCode === 0) { + this.emitInfo(message, { + type: 'WorkerCluster', + pid: wcPid + }); } else { - var error = new ProcessExitError(message, errorCode); + let error = new ProcessExitError(message, errorCode); if (signal != null) { error.signal = signal; } - this.emitFail(error, { + this.emitError(error, { type: 'WorkerCluster', pid: wcPid }); } - if (this._shuttingDown) { - return; + if (!this.isShuttingDown) { + this._launchWorkerCluster(); } - - this._launchWorkerCluster(); }; SocketCluster.prototype._launchWorkerCluster = function () { - var self = this; + let debugPort, inspectPort; - var debugPort, inspectPort; - - var debugRegex = /^--debug(=[0-9]*)?$/; - var debugBrkRegex = /^--debug-brk(=[0-9]*)?$/; - var inspectRegex = /^--inspect(=[0-9]*)?$/; - var inspectBrkRegex = /^--inspect-brk(=[0-9]*)?$/; + let debugRegex = /^--debug(=[0-9]*)?$/; + let debugBrkRegex = /^--debug-brk(=[0-9]*)?$/; + let inspectRegex = /^--inspect(=[0-9]*)?$/; + let inspectBrkRegex = /^--inspect-brk(=[0-9]*)?$/; // Workers should not inherit the master --debug argument // because they have their own --debug-workers option. - var execOptions = { - execArgv: process.execArgv.filter(function (arg) { + let execOptions = { + execArgv: process.execArgv.filter((arg) => { return !debugRegex.test(arg) && !debugBrkRegex.test(arg) && !inspectRegex.test(arg) && !inspectBrkRegex.test(arg); }) }; if (argv['debug-workers']) { - if (argv['debug-workers'] == true) { + if (argv['debug-workers'] === true) { debugPort = this.options.defaultWorkerDebugPort; } else { debugPort = argv['debug-workers']; @@ -652,7 +678,7 @@ SocketCluster.prototype._launchWorkerCluster = function () { } if (argv['inspect-workers']) { - if (argv['inspect-workers'] == true) { + if (argv['inspect-workers'] === true) { inspectPort = this.options.defaultWorkerDebugPort; } else { inspectPort = argv['inspect-workers']; @@ -661,9 +687,9 @@ SocketCluster.prototype._launchWorkerCluster = function () { execOptions.execArgv.push('--inspect=' + inspectPort); } - var paths = this._getPaths(); + let paths = this._getPaths(); - var workerOpts = this._cloneObject(this.options); + let workerOpts = this._cloneObject(this.options); workerOpts.paths = paths; workerOpts.sourcePort = this.options.port; workerOpts.workerCount = this.options.workers; @@ -678,16 +704,16 @@ SocketCluster.prototype._launchWorkerCluster = function () { workerOpts.authSignAsync = this.options.authSignAsync; workerOpts.authVerifyAsync = this.options.authVerifyAsync; - if (typeof workerOpts.schedulingPolicy == 'string') { - if (workerOpts.schedulingPolicy == 'rr') { + if (typeof workerOpts.schedulingPolicy === 'string') { + if (workerOpts.schedulingPolicy === 'rr') { workerOpts.schedulingPolicy = cluster.SCHED_RR; - } else if (workerOpts.schedulingPolicy == 'none') { + } else if (workerOpts.schedulingPolicy === 'none') { workerOpts.schedulingPolicy = cluster.SCHED_NONE; } } execOptions.env = {}; - Object.keys(process.env).forEach(function (key) { + Object.keys(process.env).forEach((key) => { execOptions.env[key] = process.env[key]; }); execOptions.env.workerInitOptions = JSON.stringify(workerOpts); @@ -695,76 +721,87 @@ SocketCluster.prototype._launchWorkerCluster = function () { this.workerCluster = fork(paths.appWorkerClusterControllerPath, process.argv.slice(2), execOptions); this.isWorkerClusterReady = false; - this.workerCluster.on('error', function (err) { - self._workerClusterErrorHandler(self.workerCluster.pid, err); + this.workerCluster.on('error', (err) => { + this._workerClusterErrorHandler(this.workerCluster.pid, err); }); - this.workerCluster.on('message', function workerHandler(m) { - if (m.type == 'error') { - if (m.data.workerPid) { - self._workerErrorHandler(m.data.workerPid, m.data.error); + this.workerCluster.on('message', (message) => { + if (message.type === 'error') { + if (message.data.workerPid) { + this._workerErrorHandler(message.data.workerPid, message.data.error); } else { - self._workerClusterErrorHandler(m.data.pid, m.data.error); + this._workerClusterErrorHandler(message.data.pid, message.data.error); } - } else if (m.type == 'warning') { - var warning = scErrors.hydrateError(m.data.error, true); - self._workerWarningHandler(m.data.workerPid, warning); - } else if (m.type == 'ready') { - self._workerClusterReadyHandler(); - } else if (m.type == 'workerStart') { - self._workerStartHandler(m.data); - } else if (m.type == 'workerExit') { - self._workerExitHandler(m.data); - } else if (m.type == 'workerMessage') { - self.emit('workerMessage', m.workerId, m.data, function (err, data) { - if (m.cid) { - self.respondToWorker(err, data, m.workerId, m.cid); + } else if (message.type === 'warning') { + let warning = scErrors.hydrateError(message.data.error, true); + this._workerWarningHandler(message.data.workerPid, warning); + } else if (message.type === 'ready') { + this._workerClusterReadyHandler(); + } else if (message.type === 'workerStart') { + this._workerStartHandler(message.data); + } else if (message.type === 'workerExit') { + this._workerExitHandler(message.data); + } else if (message.type === 'workerMessage') { + this.emit(this.EVENT_WORKER_MESSAGE, { + workerId: message.workerId, + data: message.data + }); + } else if (message.type === 'workerRequest') { + this.emit(this.EVENT_WORKER_REQUEST, { + workerId: message.workerId, + data: message.data, + end: (data) => { + this.respondToWorker(null, data, message.workerId, message.cid); + }, + error: (err) => { + this.respondToWorker(err, null, message.workerId, message.cid); } }); - } else if (m.type == 'workerResponse' || m.type == 'workerClusterResponse') { - var responseHandler = self._pendingResponseHandlers[m.rid]; + } else if (message.type === 'workerResponse' || message.type === 'workerClusterResponse') { + let responseHandler = this._pendingResponseHandlers[message.rid]; if (responseHandler) { clearTimeout(responseHandler.timeout); - delete self._pendingResponseHandlers[m.rid]; - var properError = scErrors.hydrateError(m.error, true); - responseHandler.callback(properError, m.data, m.workerId); + delete this._pendingResponseHandlers[message.rid]; + let properError = scErrors.hydrateError(message.error, true); + responseHandler.callback(properError, message.data, message.workerId); } } }); - var workerClusterInfo = { + let workerClusterInfo = { pid: this.workerCluster.pid, childProcess: this.workerCluster }; + this.emit(this.EVENT_WORKER_CLUSTER_START, workerClusterInfo); this.workerCluster.on('exit', this._handleWorkerClusterExit.bind(this)); - this.workerCluster.on('disconnect', function () { - self.isWorkerClusterReady = false; - self.workerClusterMessageBuffer = []; + this.workerCluster.on('disconnect', () => { + this.isWorkerClusterReady = false; + this.workerClusterMessageBuffer = []; }); }; SocketCluster.prototype.respondToWorker = function (err, data, workerId, rid) { this.workerCluster.send({ type: 'masterResponse', - workerId: workerId, + workerId, error: scErrors.dehydrateError(err, true), - data: data, - rid: rid + data, + rid }); }; SocketCluster.prototype._logDeploymentDetails = function () { - if (this.options.logLevel > 0) { - console.log(' ' + this.colorText('[Active]', 'green') + ' SocketCluster started'); - console.log(' Version: ' + pkg.version); - console.log(' Environment: ' + this.options.environment); - console.log(' WebSocket engine: ' + this.options.wsEngine); - console.log(' Port: ' + this.options.port); - console.log(' Master PID: ' + process.pid); - console.log(' Worker count: ' + this.options.workers); - console.log(' Broker count: ' + this.options.brokers); + if (this.options.logLevel > 1) { + console.log(` ${this.colorText('[Active]', 'green')} SocketCluster started`); + console.log(` Version: ${pkg.version}`); + console.log(` Environment: ${this.options.environment}`); + console.log(` WebSocket engine: ${this.options.wsEngine}`); + console.log(` Port: ${this.options.port}`); + console.log(` Master PID: ${process.pid}`); + console.log(` Worker count: ${this.options.workers}`); + console.log(` Broker count: ${this.options.brokers}`); console.log(); } }; @@ -773,79 +810,88 @@ SocketCluster.prototype._logDeploymentDetails = function () { SocketCluster.prototype.run = function () {}; SocketCluster.prototype._start = function () { - var self = this; - - var paths = self._getPaths(); + let paths = this._getPaths(); - var brokerEngineServerReady = function () { - self._brokerEngineServer.removeListener('ready', brokerEngineServerReady); - self._launchWorkerCluster(); - }; - - var brokerDebugPort = argv['debug-brokers']; - if (brokerDebugPort == true) { - brokerDebugPort = self.options.defaultBrokerDebugPort; + let brokerDebugPort = argv['debug-brokers']; + if (brokerDebugPort === true) { + brokerDebugPort = this.options.defaultBrokerDebugPort; } - var brokerInspectPort = argv['inspect-brokers']; - if (brokerInspectPort == true) { - brokerInspectPort = self.options.defaultBrokerDebugPort; + let brokerInspectPort = argv['inspect-brokers']; + if (brokerInspectPort === true) { + brokerInspectPort = this.options.defaultBrokerDebugPort; } - self._brokerEngineServer = new self._brokerEngine.Server({ - brokers: self._getBrokerSocketPaths(), + this._brokerEngineServer = new this._brokerEngine.Server({ + brokers: this._getBrokerSocketPaths(), debug: brokerDebugPort, inspect: brokerInspectPort, - instanceId: self.options.instanceId, - secretKey: self.options.secretKey, - expiryAccuracy: self._dataExpiryAccuracy, - downgradeToUser: self.options.downgradeToUser, - processTermTimeout: self.options.processTermTimeout, - forceKillTimeout: self.options.forceKillTimeout, - forceKillSignal: self.options.forceKillSignal, - ipcAckTimeout: self.options.ipcAckTimeout, - brokerOptions: self.options, + instanceId: this.options.instanceId, + secretKey: this.options.secretKey, + expiryAccuracy: this._dataExpiryAccuracy, + downgradeToUser: this.options.downgradeToUser, + processTermTimeout: this.options.processTermTimeout, + forceKillTimeout: this.options.forceKillTimeout, + forceKillSignal: this.options.forceKillSignal, + ipcAckTimeout: this.options.ipcAckTimeout, + brokerOptions: this.options, appBrokerControllerPath: paths.appBrokerControllerPath }); - self._brokerEngineServer.on('error', function (err) { - if (err.brokerPid) { - self._brokerErrorHandler(err.brokerPid, err); - } else { - self._brokerEngineErrorHandler(err.pid, err); + (async () => { + for await (let {error} of this._brokerEngineServer.listener('error')) { + if (error.brokerPid) { + this._brokerErrorHandler(error.brokerPid, error); + } else { + this._brokerEngineErrorHandler(error.pid, error); + } } - }); + })(); - self._brokerEngineServer.on('ready', brokerEngineServerReady); + (async () => { + await this._brokerEngineServer.listener('ready').once(); + this._launchWorkerCluster(); + })(); - self._brokerEngineServer.on('brokerStart', function (brokerInfo) { - self.emit(self.EVENT_BROKER_START, brokerInfo); - }); + (async () => { + for await (let brokerInfo of this._brokerEngineServer.listener('brokerStart')) { + this.emit(this.EVENT_BROKER_START, brokerInfo); + } + })(); - self._brokerEngineServer.on('brokerExit', function (brokerInfo) { - self.emit(self.EVENT_BROKER_EXIT, brokerInfo); - }); + (async () => { + for await (let brokerInfo of this._brokerEngineServer.listener('brokerExit')) { + this.emit(this.EVENT_BROKER_EXIT, brokerInfo); + } + })(); - self._brokerEngineServer.on('brokerMessage', function (brokerId, data, callback) { - self.emit('brokerMessage', brokerId, data, callback); - }); + (async () => { + for await (let event of this._brokerEngineServer.listener('brokerMessage')) { + this.emit(this.EVENT_BROKER_MESSAGE, event); + } + })(); + + (async () => { + for await (let req of this._brokerEngineServer.listener('brokerRequest')) { + this.emit(this.EVENT_BROKER_REQUEST, req); + } + })(); }; SocketCluster.prototype._createIPCResponseHandler = function (callback) { - var self = this; - var cid = uuid.v4(); + let cid = uuid.v4(); - var responseTimeout = setTimeout(function () { - var responseHandler = self._pendingResponseHandlers[cid]; + let responseTimeout = setTimeout(() => { + let responseHandler = this._pendingResponseHandlers[cid]; if (responseHandler) { - delete self._pendingResponseHandlers[cid]; - var timeoutError = new TimeoutError('IPC response timed out'); + delete this._pendingResponseHandlers[cid]; + let timeoutError = new TimeoutError('IPC response timed out'); responseHandler.callback(timeoutError); } }, this.options.ipcAckTimeout); this._pendingResponseHandlers[cid] = { - callback: callback, + callback, timeout: responseTimeout }; @@ -853,36 +899,72 @@ SocketCluster.prototype._createIPCResponseHandler = function (callback) { }; SocketCluster.prototype._flushWorkerClusterMessageBuffer = function () { - var self = this; - - this.workerClusterMessageBuffer.forEach(function (messagePacket) { - self.workerCluster.send(messagePacket); + this.workerClusterMessageBuffer.forEach((messagePacket) => { + this.workerCluster.send(messagePacket); }); this.workerClusterMessageBuffer = []; }; -SocketCluster.prototype.sendToWorker = function (workerId, data, callback) { - var messagePacket = { +SocketCluster.prototype.sendRequestToWorker = function (workerId, data) { + let messagePacket = { + type: 'masterRequest', + workerId: workerId, + data + }; + return new Promise((resolve, reject) => { + messagePacket.cid = this._createIPCResponseHandler((err, data) => { + if (err) { + reject(err); + return; + } + resolve(data); + }); + this.workerClusterMessageBuffer.push(messagePacket); + + if (this.isWorkerClusterReady) { + this._flushWorkerClusterMessageBuffer(); + } + }); +}; + +SocketCluster.prototype.sendMessageToWorker = function (workerId, data) { + let messagePacket = { type: 'masterMessage', workerId: workerId, - data: data + data }; + return new Promise((resolve, reject) => { + messagePacket.cid = this._createIPCResponseHandler((err, data) => { + if (err) { + reject(err); + return; + } + resolve(data); + }); + this.workerClusterMessageBuffer.push(messagePacket); - if (callback) { - messagePacket.cid = this._createIPCResponseHandler(callback); - } - this.workerClusterMessageBuffer.push(messagePacket); + if (this.isWorkerClusterReady) { + this._flushWorkerClusterMessageBuffer(); + } + }); +}; - if (this.isWorkerClusterReady) { - this._flushWorkerClusterMessageBuffer(); +SocketCluster.prototype.sendRequestToBroker = function (brokerId, data) { + if (!this._brokerEngineServer) { + return Promise.reject( + new InvalidActionError('Cannot send a request to a broker until the master instance is ready') + ); } + return this._brokerEngineServer.sendRequestToBroker(brokerId, data); }; -SocketCluster.prototype.sendToBroker = function (brokerId, data, callback) { +SocketCluster.prototype.sendMessageToBroker = function (brokerId, data) { if (!this._brokerEngineServer) { - throw new InvalidActionError('Cannot send a message to a broker until the master instance is ready'); + return Promise.reject( + new InvalidActionError('Cannot send a message to a broker until the master instance is ready') + ); } - this._brokerEngineServer.sendToBroker(brokerId, data, callback); + return this._brokerEngineServer.sendMessageToBroker(brokerId, data); }; // The options object is optional and can have two boolean fields: @@ -916,79 +998,61 @@ SocketCluster.prototype._cloneObject = function (object) { SocketCluster.prototype.colorText = function (message, color) { if (this._colorCodes[color]) { - return '\x1b[0;' + this._colorCodes[color] + 'm' + message + '\x1b[0m'; + return `\x1b[0;${this._colorCodes[color]}m${message}\x1b[0m`; } else if (color) { - return '\x1b[' + color + 'm' + message + '\x1b[0m'; + return `\x1b[${color}m${message}\x1b[0m`; } return message; }; -SocketCluster.prototype.destroy = function (callback) { - var self = this; - - if (callback) { - this._destroyCallbacks.push(callback); - } - - if (this._shuttingDown) { +SocketCluster.prototype.destroy = async function () { + if (this.isShuttingDown) { + await this.listener('destroy').once(); return; } - this._shuttingDown = true; + this.isShuttingDown = true; - Promise.all([ - new Promise(function (resolve, reject) { - if (!self.workerCluster) { - resolve(); - return; - } - self.once(self.EVENT_WORKER_CLUSTER_EXIT, function () { - resolve(); - }); - }), - new Promise(function (resolve, reject) { - if (!self._brokerEngineServer) { - resolve(); - return; - } - var killedBrokerLookup = {}; - var killedBrokerCount = 0; - var handleBrokerExit = function (brokerDetails) { - if (!killedBrokerLookup[brokerDetails.id]) { - killedBrokerLookup[brokerDetails.id] = true; - killedBrokerCount++; + (async () => { + await Promise.all([ + (async () => { + if (this.workerCluster) { + await this.listener(this.EVENT_WORKER_CLUSTER_EXIT).once(); } - if (killedBrokerCount >= self.options.brokers) { - self.removeListener(self.EVENT_BROKER_EXIT, handleBrokerExit); - resolve(); + })(), + (async () => { + if (this._brokerEngineServer) { + let killedBrokerLookup = {}; + let killedBrokerCount = 0; + for await (let event of this.listener(this.EVENT_BROKER_EXIT)) { + if (!killedBrokerLookup[event.brokerId]) { + killedBrokerLookup[event.brokerId] = true; + killedBrokerCount++; + } + if (killedBrokerCount >= this.options.brokers) { + break; + } + } } - }; - self.on(self.EVENT_BROKER_EXIT, handleBrokerExit); - }) - ]) - .then(function () { - return new Promise(function (resolve) { - setTimeout(function () { + })() + ]); + + await new Promise((resolve) => { + setTimeout(() => { resolve(); }, 0); }); - }) - .then(function () { + socketClusterSingleton = null; - self.removeAllListeners(); - if (self._stdinErrorHandler) { - process.stdin.removeListener('error', self._stdinErrorHandler); + this.isShuttingDown = false; + if (this._stdinErrorHandler) { + process.stdin.removeListener('error', this._stdinErrorHandler); } - if (self._sigusr2SignalHandler) { - process.removeListener('SIGUSR2', self._sigusr2SignalHandler); + if (this._sigusr2SignalHandler) { + process.removeListener('SIGUSR2', this._sigusr2SignalHandler); } - self._destroyCallbacks.forEach(function (callback) { - callback(); - }); - self._destroyCallbacks = []; - }) - .catch(function (error) { - self.emit('error', error); - }); + this.emit('destroy', {}); + this.closeAllListeners(); + })(); if (this.workerCluster) { this.killWorkers({killClusterMaster: true}); @@ -996,6 +1060,7 @@ SocketCluster.prototype.destroy = function (callback) { if (this._brokerEngineServer) { this._brokerEngineServer.destroy(); } + await this.listener('destroy').once(); }; module.exports = SocketCluster; diff --git a/package.json b/package.json index 89a39c77..3926827e 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,8 @@ "url": "git://github.com/SocketCluster/socketcluster.git" }, "dependencies": { - "async": "2.3.0", + "async": "2.6.1", + "async-stream-emitter": "^1.1.0", "fs-extra": "6.0.1", "inquirer": "5.2.0", "minimist": "1.2.0", @@ -28,6 +29,9 @@ "bin": { "socketcluster": "bin/cli.js" }, + "scripts": { + "test": "mocha --reporter spec --timeout 3000 --slow 3000" + }, "keywords": [ "websocket", "server", diff --git a/sample/broker.js b/sample/broker.js index 4297c340..ffa2a97b 100644 --- a/sample/broker.js +++ b/sample/broker.js @@ -1,9 +1,9 @@ -var SCBroker = require('socketcluster/scbroker'); -var scClusterBrokerClient = require('scc-broker-client'); +const SCBroker = require('socketcluster/scbroker'); +const scClusterBrokerClient = require('scc-broker-client'); class Broker extends SCBroker { run() { - console.log(' >> Broker PID:', process.pid); + console.log(` >> Broker PID: ${process.pid}`); // This is defined in server.js (taken from environment variable SC_CLUSTER_STATE_SERVER_HOST). // If this property is defined, the broker will try to attach itself to the SC cluster for diff --git a/sample/dockerwait.js b/sample/dockerwait.js index 8e13de3b..3934e2e7 100644 --- a/sample/dockerwait.js +++ b/sample/dockerwait.js @@ -4,16 +4,16 @@ a volume container at runtime and so it is necessary to wait for it before launch. */ -var fsUtil = require('socketcluster/fsutil'); -var waitForFile = fsUtil.waitForFile; +const fsUtil = require('socketcluster/fsutil'); +const waitForFile = fsUtil.waitForFile; -var SOCKETCLUSTER_MASTER_DEFAULT_CONTROLLER = './server.js'; -var masterControllerPath = process.env.SOCKETCLUSTER_MASTER_CONTROLLER || SOCKETCLUSTER_MASTER_DEFAULT_CONTROLLER; -var bootCheckInterval = Number(process.env.SOCKETCLUSTER_BOOT_CHECK_INTERVAL) || 200; -var bootTimeout = Number(process.env.SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT) || 10000; -var bootStartTime = Date.now(); +const SOCKETCLUSTER_MASTER_DEFAULT_CONTROLLER = './server.js'; +const masterControllerPath = process.env.SOCKETCLUSTER_MASTER_CONTROLLER || SOCKETCLUSTER_MASTER_DEFAULT_CONTROLLER; +const bootCheckInterval = Number(process.env.SOCKETCLUSTER_BOOT_CHECK_INTERVAL) || 200; +const bootTimeout = Number(process.env.SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT) || 10000; +const bootStartTime = Date.now(); -var errorMessage = `Failed to locate the master controller file at path ${masterControllerPath} ` + +const errorMessage = `Failed to locate the master controller file at path ${masterControllerPath} ` + `before SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT`; waitForFile(masterControllerPath, bootCheckInterval, bootStartTime, bootTimeout, errorMessage) diff --git a/sample/public/index.html b/sample/public/index.html index 283ec613..7afee79e 100644 --- a/sample/public/index.html +++ b/sample/public/index.html @@ -71,29 +71,39 @@ diff --git a/sample/server.js b/sample/server.js index 194d5f34..3310d196 100644 --- a/sample/server.js +++ b/sample/server.js @@ -7,25 +7,24 @@ each one has a specific meaning within the SC ecosystem. */ -var path = require('path'); -var argv = require('minimist')(process.argv.slice(2)); -var scHotReboot = require('sc-hot-reboot'); +const path = require('path'); +const scHotReboot = require('sc-hot-reboot'); +const fsUtil = require('socketcluster/fsutil'); +const waitForFile = fsUtil.waitForFile; -var fsUtil = require('socketcluster/fsutil'); -var waitForFile = fsUtil.waitForFile; +const SocketCluster = require('socketcluster'); -var SocketCluster = require('socketcluster'); +let argv = require('minimist')(process.argv.slice(2)); -var workerControllerPath = argv.wc || process.env.SOCKETCLUSTER_WORKER_CONTROLLER; -var brokerControllerPath = argv.bc || process.env.SOCKETCLUSTER_BROKER_CONTROLLER; -var workerClusterControllerPath = argv.wcc || process.env.SOCKETCLUSTER_WORKERCLUSTER_CONTROLLER; -var environment = process.env.ENV || 'dev'; +let workerControllerPath = argv.wc || process.env.SOCKETCLUSTER_WORKER_CONTROLLER; +let brokerControllerPath = argv.bc || process.env.SOCKETCLUSTER_BROKER_CONTROLLER; +let workerClusterControllerPath = argv.wcc || process.env.SOCKETCLUSTER_WORKERCLUSTER_CONTROLLER; +let environment = process.env.ENV || 'dev'; -var options = { +let options = { workers: Number(argv.w) || Number(process.env.SOCKETCLUSTER_WORKERS) || 1, brokers: Number(argv.b) || Number(process.env.SOCKETCLUSTER_BROKERS) || 1, port: Number(argv.p) || Number(process.env.SOCKETCLUSTER_PORT) || 8000, - // You can switch to 'sc-uws' for improved performance. wsEngine: process.env.SOCKETCLUSTER_WS_ENGINE || 'ws', appName: argv.n || process.env.SOCKETCLUSTER_APP_NAME || null, workerController: workerControllerPath || path.join(__dirname, 'worker.js'), @@ -42,31 +41,33 @@ var options = { clusterStateServerConnectTimeout: Number(process.env.SCC_STATE_SERVER_CONNECT_TIMEOUT) || null, clusterStateServerAckTimeout: Number(process.env.SCC_STATE_SERVER_ACK_TIMEOUT) || null, clusterStateServerReconnectRandomness: Number(process.env.SCC_STATE_SERVER_RECONNECT_RANDOMNESS) || null, - crashWorkerOnError: argv['auto-reboot'] != false, + crashWorkerOnError: argv['auto-reboot'] !== false, // If using nodemon, set this to true, and make sure that environment is 'dev'. killMasterOnSignal: false, - environment: environment + environment }; -var bootTimeout = Number(process.env.SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT) || 10000; -var SOCKETCLUSTER_OPTIONS; +let bootTimeout = Number(process.env.SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT) || 10000; +let SOCKETCLUSTER_OPTIONS; if (process.env.SOCKETCLUSTER_OPTIONS) { SOCKETCLUSTER_OPTIONS = JSON.parse(process.env.SOCKETCLUSTER_OPTIONS); } -for (var i in SOCKETCLUSTER_OPTIONS) { +for (let i in SOCKETCLUSTER_OPTIONS) { if (SOCKETCLUSTER_OPTIONS.hasOwnProperty(i)) { options[i] = SOCKETCLUSTER_OPTIONS[i]; } } -var start = function () { - var socketCluster = new SocketCluster(options); +let start = function () { + let socketCluster = new SocketCluster(options); - socketCluster.on(socketCluster.EVENT_WORKER_CLUSTER_START, function (workerClusterInfo) { - console.log(' >> WorkerCluster PID:', workerClusterInfo.pid); - }); + (async () => { + for await (let event of socketCluster.listener(socketCluster.EVENT_WORKER_CLUSTER_START)) { + console.log(` >> WorkerCluster PID: ${event.pid}`); + } + })(); if (socketCluster.options.environment === 'dev') { // This will cause SC workers to reboot when code changes anywhere in the app directory. @@ -80,27 +81,30 @@ var start = function () { } }; -var bootCheckInterval = Number(process.env.SOCKETCLUSTER_BOOT_CHECK_INTERVAL) || 200; -var bootStartTime = Date.now(); +let bootCheckInterval = Number(process.env.SOCKETCLUSTER_BOOT_CHECK_INTERVAL) || 200; +let bootStartTime = Date.now(); // Detect when Docker volumes are ready. -var startWhenFileIsReady = (filePath) => { - var errorMessage = `Failed to locate a controller file at path ${filePath} ` + +let startWhenFileIsReady = (filePath) => { + let errorMessage = `Failed to locate a controller file at path ${filePath} ` + `before SOCKETCLUSTER_CONTROLLER_BOOT_TIMEOUT`; return waitForFile(filePath, bootCheckInterval, bootStartTime, bootTimeout, errorMessage); }; -var filesReadyPromises = [ +let filesReadyPromises = [ startWhenFileIsReady(workerControllerPath), startWhenFileIsReady(brokerControllerPath), startWhenFileIsReady(workerClusterControllerPath) ]; -Promise.all(filesReadyPromises) -.then(() => { + +(async () => { + try { + await Promise.all(filesReadyPromises); + } catch (err) { + console.error(err); + process.exit(1); + return; + } start(); -}) -.catch((err) => { - console.error(err.stack); - process.exit(1); -}); +})(); diff --git a/sample/worker.js b/sample/worker.js index 4d277e5d..c1b021a5 100644 --- a/sample/worker.js +++ b/sample/worker.js @@ -1,19 +1,19 @@ -var SCWorker = require('socketcluster/scworker'); -var express = require('express'); -var serveStatic = require('serve-static'); -var path = require('path'); -var morgan = require('morgan'); -var healthChecker = require('sc-framework-health-check'); +const SCWorker = require('socketcluster/scworker'); +const express = require('express'); +const serveStatic = require('serve-static'); +const path = require('path'); +const morgan = require('morgan'); +const healthChecker = require('sc-framework-health-check'); class Worker extends SCWorker { run() { - console.log(' >> Worker PID:', process.pid); - var environment = this.options.environment; + console.log(` >> Worker PID: ${process.pid}`); + let environment = this.options.environment; - var app = express(); + let app = express(); - var httpServer = this.httpServer; - var scServer = this.scServer; + let httpServer = this.httpServer; + let scServer = this.scServer; if (environment === 'dev') { // Log every HTTP request. See https://github.com/expressjs/morgan for other @@ -27,32 +27,37 @@ class Worker extends SCWorker { httpServer.on('request', app); - var count = 0; + let count = 0; /* In here we handle our incoming realtime connections and listen for events. */ - scServer.on('connection', function (socket) { - - // Some sample logic to show how to handle client events, - // replace this with your own logic - - socket.on('sampleClientEvent', function (data) { - count++; - console.log('Handled sampleClientEvent', data); - scServer.exchange.publish('sample', count); - }); - - var interval = setInterval(function () { - socket.emit('random', { - number: Math.floor(Math.random() * 5) - }); - }, 1000); - - socket.on('disconnect', function () { - clearInterval(interval); - }); - }); + (async () => { + for await (let {socket} of scServer.listener('connection')) { + + // Some sample logic to show how to handle client events, + // replace this with your own logic + + (async () => { + for await (let data of socket.receiver('sampleClientEvent')) { + count++; + console.log('Handled sampleClientEvent', data); + scServer.exchange.publish('sample', count); + } + })(); + + let interval = setInterval(() => { + socket.transmit('random', { + number: Math.floor(Math.random() * 5) + }); + }, 1000); + + (async () => { + await socket.listener('disconnect').once(); + clearInterval(interval); + })(); + } + })(); } } diff --git a/scworker.js b/scworker.js index 76568b15..d7529c22 100644 --- a/scworker.js +++ b/scworker.js @@ -1,37 +1,37 @@ -var socketClusterServer = require('socketcluster-server'); -var EventEmitter = require('events').EventEmitter; -var uuid = require('uuid'); -var http = require('http'); -var https = require('https'); -var async = require('async'); -var AuthEngine = require('sc-auth').AuthEngine; - -var scErrors = require('sc-errors'); -var InvalidActionError = scErrors.InvalidActionError; -var ResourceLimitError = scErrors.ResourceLimitError; -var BrokerError = scErrors.BrokerError; -var HTTPServerError = scErrors.HTTPServerError; -var TimeoutError = scErrors.TimeoutError; - -var processTermTimeout = 10000; -var workerInitOptions = JSON.parse(process.env.workerInitOptions); - -var handleError = function (isFatal, err) { - var error = scErrors.dehydrateError(err, true); +const socketClusterServer = require('socketcluster-server'); +const AsyncStreamEmitter = require('async-stream-emitter'); +const uuid = require('uuid'); +const http = require('http'); +const https = require('https'); +const async = require('async'); +const AuthEngine = require('sc-auth').AuthEngine; + +const scErrors = require('sc-errors'); +const InvalidActionError = scErrors.InvalidActionError; +const ResourceLimitError = scErrors.ResourceLimitError; +const BrokerError = scErrors.BrokerError; +const HTTPServerError = scErrors.HTTPServerError; +const TimeoutError = scErrors.TimeoutError; + +const workerInitOptions = JSON.parse(process.env.workerInitOptions); +let processTermTimeout = 10000; + +let handleError = function (isFatal, err) { + let error = scErrors.dehydrateError(err, true); process.send({ type: 'error', data: { error: error, workerPid: process.pid } - }, null, function () { + }, null, () => { if (isFatal) { process.exit(1); } }); }; -var handleWarning = function (warning) { +let handleWarning = function (warning) { warning = scErrors.dehydrateError(warning, true); process.send({ type: 'warning', @@ -42,17 +42,19 @@ var handleWarning = function (warning) { }); }; -var handleReady = function () { +let handleReady = function () { process.send({type: 'ready'}); }; -var handleExit = function () { +let handleExit = function () { process.exit(); }; -var scWorker; +let scWorker; function SCWorker(options) { + AsyncStreamEmitter.call(this); + if (scWorker) { // SCWorker is a singleton; it can only be instantiated once per process. throw new InvalidActionError('Attempted to instantiate a worker which has already been instantiated'); @@ -60,14 +62,6 @@ function SCWorker(options) { options = options || {}; scWorker = this; - this.EVENT_ERROR = 'error'; - this.EVENT_WARNING = 'warning'; - this.EVENT_EXIT = 'exit'; - this.EVENT_READY = 'ready'; - this.EVENT_CONNECTION = 'connection'; - - this.MIDDLEWARE_START = 'start'; - this.type = 'worker'; this.isTerminating = false; this._pendingResponseHandlers = {}; @@ -79,15 +73,22 @@ function SCWorker(options) { this.createHTTPServer = options.createHTTPServer; } - var workerOptions = Object.assign({}, options, workerInitOptions); + let workerOptions = Object.assign({}, options, workerInitOptions); this._init(workerOptions); } +SCWorker.prototype = Object.create(AsyncStreamEmitter.prototype); + SCWorker.create = function (options) { return new SCWorker(options); }; -SCWorker.prototype = Object.create(EventEmitter.prototype); +SCWorker.EVENT_ERROR = SCWorker.prototype.EVENT_ERROR = 'error'; +SCWorker.EVENT_WARNING = SCWorker.prototype.EVENT_WARNING = 'warning'; +SCWorker.EVENT_EXIT = SCWorker.prototype.EVENT_EXIT = 'exit'; +SCWorker.EVENT_READY = SCWorker.prototype.EVENT_READY = 'ready'; +SCWorker.EVENT_CONNECTION = SCWorker.prototype.EVENT_CONNECTION = 'connection'; +SCWorker.MIDDLEWARE_START = SCWorker.prototype.MIDDLEWARE_START = 'start'; SCWorker.prototype.setAuthEngine = function (authEngine) { this.auth = authEngine; @@ -101,9 +102,7 @@ SCWorker.prototype.setCodecEngine = function (codecEngine) { this.scServer.setCodecEngine(this.codec); }; -SCWorker.prototype._init = function (options) { - var self = this; - +SCWorker.prototype._init = async function (options) { this.options = {}; Object.assign(this.options, options); @@ -121,21 +120,38 @@ SCWorker.prototype._init = function (options) { } if (this.options.propagateErrors) { - this.on('error', handleError.bind(null, this.options.crashWorkerOnError)); + (async () => { + for await (let {error} of this.listener('error')) { + handleError(this.options.crashWorkerOnError, error); + } + })(); if (this.options.propagateWarnings) { - this.on('warning', handleWarning); + (async () => { + for await (let {warning} of this.listener('warning')) { + handleWarning(warning); + } + })(); } - this.on('exit', handleExit); + (async () => { + for await (let event of this.listener('exit')) { + handleExit(); + } + })(); } - this.on(this.EVENT_READY, function () { - self.start().then(function () { - handleReady(); - }).catch(this.emitError.bind(this)); - }); + (async () => { + for await (let event of this.listener(this.EVENT_READY)) { + try { + await this.start(); + handleReady(); + } catch (err) { + this.emitError(err); + } + } + })(); this.id = this.options.id; - this.isLeader = this.id == 0; + this.isLeader = this.id === 0; this._middleware = {}; this._middleware[this.MIDDLEWARE_START] = []; @@ -144,9 +160,11 @@ SCWorker.prototype._init = function (options) { try { process.setuid(this.options.downgradeToUser); } catch (err) { - throw new InvalidActionError('Could not downgrade to user "' + this.options.downgradeToUser + - '" - Either this user does not exist or the current process does not have the permission' + - ' to switch to it'); + throw new InvalidActionError( + `Could not downgrade to user "${this.options.downgradeToUser}"` + + ` - Either this user does not exist or the current process does not have the permission` + + ` to switch to it` + ); } } @@ -163,119 +181,134 @@ SCWorker.prototype._init = function (options) { brokers: this.options.brokers, secretKey: this.options.secretKey, pubSubBatchDuration: this.options.pubSubBatchDuration, - connectRetryErrorThreshold: this.options.brokerConnectRetryErrorThreshold + connectRetryErrorThreshold: this.options.brokerConnectRetryErrorThreshold, + commandAckTimeout: this.options.brokerCommandAckTimeout, + pubSubAckTimeout: this.options.brokerPubSubAckTimeout, + autoReconnectOptions: this.options.brokerAutoReconnectOptions }); - this.brokerEngineClient.on('error', function (err) { - var error; - if (typeof err == 'string') { - error = new BrokerError(err); + (async () => { + for await (let {error} of this.brokerEngineClient.listener('error')) { + let err; + if (typeof error === 'string') { + err = new BrokerError(error); + } else { + err = error; + } + this.emitError(err); + } + })(); + + (async () => { + for await (let {warning} of this.brokerEngineClient.listener('warning')) { + this.emitWarning(warning); + } + })(); + + this.exchange = this.brokerEngineClient.exchange(); + + let createHTTPServerResult = this.createHTTPServer(); + try { + this.httpServer = await Promise.resolve(createHTTPServerResult); + } catch (err) { + this.emitError(err); + return; + } + this.httpServer.on('request', this._httpRequestHandler.bind(this)); // TODO 2: support stream emitter + this.httpServer.on('upgrade', this._httpRequestHandler.bind(this)); + + this.httpServer.exchange = this.exchange; + + this.httpServer.on('error', (err) => { + let error; + if (typeof err === 'string') { + error = new HTTPServerError(err); } else { error = err; } - self.emitError(error); - }); - this.brokerEngineClient.on('warning', function (warning) { - self.emitWarning(warning); + this.emitError(error); }); - this.exchange = this.brokerEngineClient.exchange(); - var createHTTPServerResult = this.createHTTPServer(); - Promise.resolve(createHTTPServerResult) - .then(function (httpServer) { - self.httpServer = httpServer; - self.httpServer.on('request', self._httpRequestHandler.bind(self)); - self.httpServer.on('upgrade', self._httpRequestHandler.bind(self)); + this.scServer = socketClusterServer.attach(this.httpServer, { + brokerEngine: this.brokerEngineClient, + wsEngine: this._paths.wsEnginePath, + allowClientPublish: this.options.allowClientPublish, + handshakeTimeout: this.options.handshakeTimeout, + ackTimeout: this.options.ackTimeout, + pingTimeout: this.options.pingTimeout, + pingInterval: this.options.pingInterval, + pingTimeoutDisabled: this.options.pingTimeoutDisabled, + origins: this.options.origins, + appName: this.options.appName, + path: this.options.path, + authKey: this.options.authKey, + authPrivateKey: this.options.authPrivateKey, + authPublicKey: this.options.authPublicKey, + authAlgorithm: this.options.authAlgorithm, + authVerifyAlgorithms: this.options.authVerifyAlgorithms, + authSignAsync: this.options.authSignAsync, + authVerifyAsync: this.options.authVerifyAsync, + authDefaultExpiry: this.options.authDefaultExpiry, + middlewareEmitWarnings: this.options.middlewareEmitWarnings, + socketChannelLimit: this.options.socketChannelLimit, + pubSubBatchDuration: this.options.pubSubBatchDuration, + perMessageDeflate: this.options.perMessageDeflate, + maxPayload: this.options.maxPayload, + wsEngineServerOptions: this.options.wsEngineServerOptions + }); - self.httpServer.exchange = self.exchange; + if (this.brokerEngineClient.setSCServer) { + this.brokerEngineClient.setSCServer(this.scServer); + } - self.httpServer.on('error', function (err) { - var error; - if (typeof err == 'string') { - error = new HTTPServerError(err); - } else { - error = err; - } - self.emitError(error); - }); + if (options.authEngine) { + this.setAuthEngine(options.authEngine); + } else { + // Default authentication engine + this.setAuthEngine(new AuthEngine()); + } + if (options.codecEngine) { + this.setCodecEngine(options.codecEngine); + } else { + this.codec = this.scServer.codec; + } - self.scServer = socketClusterServer.attach(self.httpServer, { - brokerEngine: self.brokerEngineClient, - wsEngine: self._paths.wsEnginePath, - allowClientPublish: self.options.allowClientPublish, - handshakeTimeout: self.options.handshakeTimeout, - ackTimeout: self.options.ackTimeout, - pingTimeout: self.options.pingTimeout, - pingInterval: self.options.pingInterval, - pingTimeoutDisabled: self.options.pingTimeoutDisabled, - origins: self.options.origins, - appName: self.options.appName, - path: self.options.path, - authKey: self.options.authKey, - authPrivateKey: self.options.authPrivateKey, - authPublicKey: self.options.authPublicKey, - authAlgorithm: self.options.authAlgorithm, - authVerifyAlgorithms: self.options.authVerifyAlgorithms, - authSignAsync: self.options.authSignAsync, - authVerifyAsync: self.options.authVerifyAsync, - authDefaultExpiry: self.options.authDefaultExpiry, - middlewareEmitWarnings: self.options.middlewareEmitWarnings, - socketChannelLimit: self.options.socketChannelLimit, - pubSubBatchDuration: self.options.pubSubBatchDuration, - perMessageDeflate: self.options.perMessageDeflate, - maxPayload: self.options.maxPayload, - wsEngineServerOptions: self.options.wsEngineServerOptions - }); + this._socketPath = this.scServer.getPath(); - if (self.brokerEngineClient.setSCServer) { - self.brokerEngineClient.setSCServer(self.scServer); + (async () => { + for await (let {socket} of this.scServer.listener('connection')) { + // The connection event counts as a WS request + this._wsRequestCount++; + (async () => { + for await (let {message} of socket.listener('message')) { + this._wsRequestCount++; + } + })(); + this.emit(this.EVENT_CONNECTION, {socket}); } + })(); - if (options.authEngine) { - self.setAuthEngine(options.authEngine); - } else { - // Default authentication engine - self.setAuthEngine(new AuthEngine()); - } - if (options.codecEngine) { - self.setCodecEngine(options.codecEngine); - } else { - self.codec = self.scServer.codec; + (async () => { + for await (let {error} of this.scServer.listener('error')) { + this.emitError(error); } + })(); - self._socketPath = self.scServer.getPath(); - - self.scServer.on('_connection', function (socket) { - // The connection event counts as a WS request - self._wsRequestCount++; - socket.on('message', function () { - self._wsRequestCount++; - }); - self.emit(self.EVENT_CONNECTION, socket); - }); - - self.scServer.on('warning', function (warning) { - self.emitWarning(warning); - }); - self.scServer.on('error', function (error) { - self.emitError(error); - }); - if (self.scServer.isReady) { - self.emit(self.EVENT_READY); - } else { - self.scServer.once('ready', function () { - self.emit(self.EVENT_READY); - }); + (async () => { + for await (let {warning} of this.scServer.listener('warning')) { + this.emitWarning(warning); } - }) - .catch(function (error) { - self.emitError(error); - }); + })(); + + if (!this.scServer.isReady) { + await this.scServer.listener('ready').once(); + } + this.emit(this.EVENT_READY, {}); }; SCWorker.prototype.createHTTPServer = function () { - var httpServer; - if (this.options.protocol == 'https') { + let httpServer; + if (this.options.protocol === 'https') { httpServer = https.createServer(this.options.protocolOptions); } else { httpServer = http.createServer(); @@ -304,35 +337,35 @@ SCWorker.prototype.addMiddleware = function (type, middleware) { }; SCWorker.prototype.removeMiddleware = function (type, middleware) { - var middlewareFunctions = this._middleware[type]; + let middlewareFunctions = this._middleware[type]; - this._middleware[type] = middlewareFunctions.filter(function (fn) { - return fn != middleware; + this._middleware[type] = middlewareFunctions.filter((fn) => { + return fn !== middleware; }); }; SCWorker.prototype.startHTTPServer = function () { - var self = this; - - var options = this.options; + let options = this.options; - var start = function () { + let start = () => { if (options.tcpSynBacklog != null) { - self.httpServer.listen(options.sourcePort, options.host, options.tcpSynBacklog); + this.httpServer.listen(options.sourcePort, options.host, options.tcpSynBacklog); } else if (options.host != null) { - self.httpServer.listen(options.sourcePort, options.host); + this.httpServer.listen(options.sourcePort, options.host); } else { - self.httpServer.listen(options.sourcePort); + this.httpServer.listen(options.sourcePort); } }; - var startMiddleware = this._middleware[this.MIDDLEWARE_START]; + let startMiddleware = this._middleware[this.MIDDLEWARE_START]; if (startMiddleware.length) { - var callbackInvoked = false; + let callbackInvoked = false; - async.applyEachSeries(startMiddleware, options, function (err) { + async.applyEachSeries(startMiddleware, options, (err) => { if (callbackInvoked) { - self.emit('warning', new InvalidActionError('Callback for ' + self.MIDDLEWARE_START + ' middleware was already invoked')); + this.emitWarning( + new InvalidActionError(`Callback for ${this.MIDDLEWARE_START} middleware was already invoked`) + ); } else { callbackInvoked = true; if (err) { @@ -347,7 +380,7 @@ SCWorker.prototype.startHTTPServer = function () { } }; -SCWorker.prototype.start = function () { +SCWorker.prototype.start = async function () { this._httpRequestCount = 0; this._wsRequestCount = 0; this._httpRPM = 0; @@ -358,10 +391,10 @@ SCWorker.prototype.start = function () { } this._statusInterval = setInterval(this._calculateStatus.bind(this), this.options.workerStatusInterval); - var runResult = this.run(); + let runResult = this.run(); - return Promise.resolve(runResult) - .then(this.startHTTPServer.bind(this)); + await Promise.resolve(runResult); + this.startHTTPServer(); }; SCWorker.prototype._httpRequestHandler = function (req, res) { @@ -369,10 +402,10 @@ SCWorker.prototype._httpRequestHandler = function (req, res) { req.exchange = this.exchange; - var forwardedFor = req.headers['x-forwarded-for']; + let forwardedFor = req.headers['x-forwarded-for']; if (forwardedFor) { - var forwardedClientIP; + let forwardedClientIP; if (forwardedFor.indexOf(',') > -1) { forwardedClientIP = forwardedFor.split(',')[0]; } else { @@ -400,21 +433,21 @@ SCWorker.prototype.getHTTPServer = function () { }; SCWorker.prototype._calculateStatus = function () { - var perMinuteFactor = 60000 / this.options.workerStatusInterval; + let perMinuteFactor = 60000 / this.options.workerStatusInterval; this._httpRPM = this._httpRequestCount * perMinuteFactor; this._wsRPM = this._wsRequestCount * perMinuteFactor; this._httpRequestCount = 0; this._wsRequestCount = 0; - var memThreshold = this.options.killWorkerMemoryThreshold; + let memThreshold = this.options.killWorkerMemoryThreshold; if (memThreshold != null) { - var memoryUsage = process.memoryUsage(); + let memoryUsage = process.memoryUsage(); if (memoryUsage.heapUsed > memThreshold) { - var message = 'Worker killed itself because its memory '; - message += 'usage of ' + memoryUsage.heapUsed + ' exceeded '; - message += 'the killWorkerMemoryThreshold of ' + memThreshold; - var warning = new ResourceLimitError(message); + let message = `Worker killed itself because its memory `; + message += `usage of ${memoryUsage.heapUsed} exceeded `; + message += `the killWorkerMemoryThreshold of ${memThreshold}`; + let warning = new ResourceLimitError(message); this.emitWarning(warning); process.exit(); } @@ -430,13 +463,12 @@ SCWorker.prototype.getStatus = function () { }; SCWorker.prototype._createIPCResponseHandler = function (callback) { - var self = this; - var cid = uuid.v4(); + let cid = uuid.v4(); - var responseTimeout = setTimeout(function () { - var responseHandler = self._pendingResponseHandlers[cid]; - delete self._pendingResponseHandlers[cid]; - var timeoutError = new TimeoutError('IPC response timed out'); + let responseTimeout = setTimeout(() => { + let responseHandler = this._pendingResponseHandlers[cid]; + delete this._pendingResponseHandlers[cid]; + let timeoutError = new TimeoutError('IPC response timed out'); responseHandler.callback(timeoutError); }, this.options.ipcAckTimeout); @@ -449,35 +481,50 @@ SCWorker.prototype._createIPCResponseHandler = function (callback) { }; SCWorker.prototype.handleMasterResponse = function (message) { - var responseHandler = this._pendingResponseHandlers[message.rid]; + let responseHandler = this._pendingResponseHandlers[message.rid]; if (responseHandler) { clearTimeout(responseHandler.timeout); delete this._pendingResponseHandlers[message.rid]; - var properError = scErrors.hydrateError(message.error, true); + let properError = scErrors.hydrateError(message.error, true); responseHandler.callback(properError, message.data); } }; -SCWorker.prototype.sendToMaster = function (data, callback) { - var messagePacket = { - type: 'workerMessage', - data: data, +SCWorker.prototype.sendRequestToMaster = function (data) { + let messagePacket = { + type: 'workerRequest', + data, workerId: this.id }; + return new Promise((resolve, reject) => { + messagePacket.cid = this._createIPCResponseHandler((err, data) => { + if (err) { + reject(err); + return; + } + resolve(data); + }); + process.send(messagePacket); + }); +}; - if (callback) { - messagePacket.cid = this._createIPCResponseHandler(callback); - } +SCWorker.prototype.sendMessageToMaster = function (data) { + let messagePacket = { + type: 'workerMessage', + data, + workerId: this.id + }; process.send(messagePacket); + return Promise.resolve(); }; SCWorker.prototype.respondToMaster = function (err, data, rid) { process.send({ type: 'workerResponse', error: scErrors.dehydrateError(err, true), - data: data, + data, workerId: this.id, - rid: rid + rid }); }; @@ -486,32 +533,38 @@ SCWorker.prototype.handleMasterEvent = function () { }; SCWorker.prototype.handleMasterMessage = function (message) { - var self = this + this.emit('masterMessage', {data: message.data}); +}; - self.emit('masterMessage', message.data, function (err, data) { - if (message.cid) { - self.respondToMaster(err, data, message.cid); +SCWorker.prototype.handleMasterRequest = function (request) { + this.emit('masterRequest', { + data: request.data, + end: (data) => { + this.respondToMaster(null, data, request.cid); + }, + error: (err) => { + this.respondToMaster(err, null, request.cid); } }); }; -SCWorker.prototype.emitError = function (err) { - this.emit(this.EVENT_ERROR, err); +SCWorker.prototype.emitError = function (error) { + this.emit(this.EVENT_ERROR, {error}); }; SCWorker.prototype.emitWarning = function (warning) { - this.emit(this.EVENT_WARNING, warning); + this.emit(this.EVENT_WARNING, {warning}); }; -var handleWorkerClusterMessage = function (wcMessage) { - if (wcMessage.type == 'terminate') { +let handleWorkerClusterMessage = function (wcMessage) { + if (wcMessage.type === 'terminate') { if (scWorker && !wcMessage.data.immediate) { if (!scWorker.isTerminating) { scWorker.isTerminating = true; - scWorker.close(function () { + scWorker.close(() => { process.exit(); }); - setTimeout(function () { + setTimeout(() => { process.exit(); }, processTermTimeout); } @@ -520,17 +573,19 @@ var handleWorkerClusterMessage = function (wcMessage) { } } else { if (!scWorker) { - throw new InvalidActionError(`Attempted to send '${wcMessage.type}' to worker ${workerInitOptions.id} before it was instantiated`); + throw new InvalidActionError(`Attempted to send ${wcMessage.type} to worker ${workerInitOptions.id} before it was instantiated`); } - if (wcMessage.type == 'emit') { + if (wcMessage.type === 'emit') { if (wcMessage.data) { scWorker.handleMasterEvent(wcMessage.event, wcMessage.data); } else { scWorker.handleMasterEvent(wcMessage.event); } - } else if (wcMessage.type == 'masterMessage') { + } else if (wcMessage.type === 'masterMessage') { scWorker.handleMasterMessage(wcMessage); - } else if (wcMessage.type == 'masterResponse') { + } else if (wcMessage.type === 'masterRequest') { + scWorker.handleMasterRequest(wcMessage); + } else if (wcMessage.type === 'masterResponse') { scWorker.handleMasterResponse(wcMessage); } } @@ -538,7 +593,11 @@ var handleWorkerClusterMessage = function (wcMessage) { process.on('message', handleWorkerClusterMessage); -process.on('uncaughtException', function (err) { +process.on('uncaughtException', (err) => { + handleError(workerInitOptions.crashWorkerOnError, err); +}); + +process.on('unhandledRejection', (err) => { handleError(workerInitOptions.crashWorkerOnError, err); }); diff --git a/scworkercluster.js b/scworkercluster.js index 96cc1962..c50dba30 100644 --- a/scworkercluster.js +++ b/scworkercluster.js @@ -1,27 +1,27 @@ -var cluster = require('cluster'); -var scErrors = require('sc-errors'); -var InvalidActionError = scErrors.InvalidActionError; -var ProcessExitError = scErrors.ProcessExitError; +const cluster = require('cluster'); +const scErrors = require('sc-errors'); +const InvalidActionError = scErrors.InvalidActionError; +const ProcessExitError = scErrors.ProcessExitError; -var workerInitOptions = JSON.parse(process.env.workerInitOptions); -var processTermTimeout = 10000; -var forceKillTimeout = 15000; -var forceKillSignal = 'SIGHUP'; +const workerInitOptions = JSON.parse(process.env.workerInitOptions); +let processTermTimeout = 10000; +let forceKillTimeout = 15000; +let forceKillSignal = 'SIGHUP'; -process.on('disconnect', function () { +process.on('disconnect', () => { process.exit(); }); -var scWorkerCluster; -var workers; -var hasExited = false; -var terminatedCount = 0; -var childExitLookup = {}; -var isTerminating = false; -var isForceKillingWorkers = false; +let scWorkerCluster; +let workers; +let hasExited = false; +let terminatedCount = 0; +let childExitLookup = {}; +let isTerminating = false; +let isForceKillingWorkers = false; -var sendErrorToMaster = function (err) { - var error = scErrors.dehydrateError(err, true); +let sendErrorToMaster = function (err) { + let error = scErrors.dehydrateError(err, true); process.send({ type: 'error', data: { @@ -31,14 +31,14 @@ var sendErrorToMaster = function (err) { }); }; -var terminateNow = function () { +let terminateNow = function () { if (!hasExited) { hasExited = true; process.exit(); } }; -var terminate = function (immediate) { +let terminate = function (immediate) { if (immediate) { terminateNow(); return; @@ -47,82 +47,110 @@ var terminate = function (immediate) { return; } isTerminating = true; - setTimeout(function () { + setTimeout(() => { terminateNow(); }, processTermTimeout); }; -var killUnresponsiveWorkersNow = function () { - (workers || []).forEach(function (worker, i) { +let killUnresponsiveWorkersNow = function () { + (workers || []).forEach((worker, i) => { if (!childExitLookup[i]) { process.kill(worker.process.pid, forceKillSignal); - var errorMessage = 'No exit signal was received by worker with id ' + i + - ' (PID: ' + worker.process.pid + ') before forceKillTimeout of ' + forceKillTimeout + - ' ms was reached - As a result, kill signal ' + forceKillSignal + ' was sent to worker'; - var processExitError = new ProcessExitError(errorMessage); + let processExitError = new ProcessExitError( + `No exit signal was received by worker with id ${i} ` + + `(PID: ${worker.process.pid}) before forceKillTimeout of ${forceKillTimeout} ` + + `ms was reached - As a result, kill signal ${forceKillSignal} was sent to worker` + ); sendErrorToMaster(processExitError); } }); isForceKillingWorkers = false; }; -var killUnresponsiveWorkers = function () { +let killUnresponsiveWorkers = function () { childExitLookup = {}; if (isForceKillingWorkers) { return; } isForceKillingWorkers = true; - setTimeout(function () { + setTimeout(() => { killUnresponsiveWorkersNow(); }, forceKillTimeout); }; -process.on('message', function (masterMessage) { - if (masterMessage.type == 'masterMessage' || masterMessage.type == 'masterResponse') { - var targetWorker = workers[masterMessage.workerId]; +process.on('message', (masterPacket) => { + if ( + masterPacket.type === 'masterMessage' || + masterPacket.type === 'masterRequest' || + masterPacket.type === 'masterResponse' + ) { + let targetWorker = workers[masterPacket.workerId]; if (targetWorker) { - targetWorker.send(masterMessage); + targetWorker.send(masterPacket); + if (masterPacket.type === 'masterMessage') { + process.send({ + type: 'workerClusterResponse', + error: null, + workerId: masterPacket.workerId, + rid: masterPacket.cid + }); + } } else { - if (masterMessage.type == 'masterMessage') { - var errorMessage = 'Cannot send message to worker with id ' + masterMessage.workerId + - ' because the worker does not exist'; + if (masterPacket.type === 'masterMessage') { + let notFoundError = new InvalidActionError( + `Cannot send message to worker with id ${masterPacket.workerId} ` + + `because the worker does not exist` + ); + sendErrorToMaster(notFoundError); - var notFoundError = new InvalidActionError(errorMessage); + process.send({ + type: 'workerClusterResponse', + error: scErrors.dehydrateError(notFoundError, true), + workerId: masterPacket.workerId, + rid: masterPacket.cid + }); + } else if (masterPacket.type === 'masterRequest') { + let notFoundError = new InvalidActionError( + `Cannot send request to worker with id ${masterPacket.workerId} ` + + `because the worker does not exist` + ); sendErrorToMaster(notFoundError); - if (masterMessage.cid) { - process.send({ - type: 'workerClusterResponse', - error: scErrors.dehydrateError(notFoundError, true), - data: null, - workerId: masterMessage.workerId, - rid: masterMessage.cid - }); - } + process.send({ + type: 'workerClusterResponse', + error: scErrors.dehydrateError(notFoundError, true), + workerId: masterPacket.workerId, + rid: masterPacket.cid + }); } else { - var errorMessage = 'Cannot send response to worker with id ' + masterMessage.workerId + - ' because the worker does not exist'; - - var notFoundError = new InvalidActionError(errorMessage); + let notFoundError = new InvalidActionError( + `Cannot send response to worker with id ${masterPacket.workerId} ` + + `because the worker does not exist` + ); sendErrorToMaster(notFoundError); } } } else { - if (masterMessage.type == 'terminate') { - if (masterMessage.data.killClusterMaster) { - terminate(masterMessage.data.immediate); + if (masterPacket.type === 'terminate') { + if (masterPacket.data.killClusterMaster) { + terminate(masterPacket.data.immediate); } else { killUnresponsiveWorkers(); } } - (workers || []).forEach(function (worker) { - worker.send(masterMessage); + (workers || []).forEach((worker) => { + worker.send(masterPacket); }); } }); -process.on('uncaughtException', function (err) { +process.on('uncaughtException', (err) => { + sendErrorToMaster(err); + process.exit(1); +}); + +process.on('unhandledRejection', (err) => { sendErrorToMaster(err); process.exit(1); }); @@ -164,25 +192,25 @@ SCWorkerCluster.prototype._init = function (options) { exec: options.paths.appWorkerControllerPath }); - var workerCount = options.workerCount; - var readyCount = 0; - var isReady = false; + let workerCount = options.workerCount; + let readyCount = 0; + let isReady = false; workers = []; this.workers = workers; - var launchWorker = function (i, respawn) { - var workerInitOptions = options; - workerInitOptions.id = i; + let launchWorker = (i, respawn) => { + let initOptions = options; + initOptions.id = i; - var worker = cluster.fork({ - workerInitOptions: JSON.stringify(workerInitOptions) + let worker = cluster.fork({ + workerInitOptions: JSON.stringify(initOptions) }); workers[i] = worker; worker.on('error', sendErrorToMaster); - worker.on('message', function (workerMessage) { - if (workerMessage.type == 'ready') { + worker.on('message', (workerPacket) => { + if (workerPacket.type === 'ready') { process.send({ type: 'workerStart', data: { @@ -199,11 +227,11 @@ SCWorkerCluster.prototype._init = function (options) { }); } } else { - process.send(workerMessage); + process.send(workerPacket); } }); - worker.on('exit', function (code, signal) { + worker.on('exit', (code, signal) => { childExitLookup[i] = true; if (!isTerminating) { process.send({ @@ -216,7 +244,7 @@ SCWorkerCluster.prototype._init = function (options) { } }); - if (options.rebootWorkerOnCrash) { + if (options.respawnWorkerOnCrash) { launchWorker(i, true); } } else if (++terminatedCount >= workers.length) { @@ -228,7 +256,7 @@ SCWorkerCluster.prototype._init = function (options) { }); }; - for (var i = 0; i < workerCount; i++) { + for (let i = 0; i < workerCount; i++) { launchWorker(i); } diff --git a/test/README.md b/test/README.md deleted file mode 100644 index 5da5589b..00000000 --- a/test/README.md +++ /dev/null @@ -1,6 +0,0 @@ -SocketCluster integration tests -=============================== - -The 'external' directory validates the behavioral aspect of SocketCluster (as it would appear to a client/user). -The 'internal' directory verifies that SocketCluster manages its internal resources appropriately. -Launch a test by running the appropriate index.js file using the 'node' command. \ No newline at end of file diff --git a/test/controllers/broker.js b/test/controllers/broker.js new file mode 100644 index 00000000..d20a1c81 --- /dev/null +++ b/test/controllers/broker.js @@ -0,0 +1,44 @@ +const SCBroker = require('socketcluster/scbroker'); + +class Broker extends SCBroker { + run() { + console.log(` >> Broker PID: ${process.pid}`); + + (async () => { + for await (let {data} of this.listener('masterMessage')) { + if (data.sendGoodRequestToMaster) { + let result = await this.sendRequestToMaster({value: 456}); + this.sendMessageToMaster({success: result.value === 45600}); + } else if (data.sendBadRequestToMaster) { + let result; + let error; + try { + result = await this.sendRequestToMaster({fail: true}); + } catch (err) { + error = err; + } + // Send success if we got an error as expected. + this.sendMessageToMaster({ + success: error && error.name === 'MasterFailedToRespondError' + }); + } else { + this.sendMessageToMaster({value: data.value + 1}); + } + } + })(); + + (async () => { + for await (let req of this.listener('masterRequest')) { + if (req.data.fail) { + let error = new Error('Broker could not process request'); + error.name = 'BrokerFailedToRespondError'; + req.error(error); + } else { + req.end({value: req.data.value * 10}); + } + } + })(); + } +} + +new Broker(); diff --git a/test/controllers/worker.js b/test/controllers/worker.js new file mode 100644 index 00000000..b2ffdeca --- /dev/null +++ b/test/controllers/worker.js @@ -0,0 +1,47 @@ +const SCWorker = require('socketcluster/scworker'); + +class Worker extends SCWorker { + run() { + console.log(` >> Worker PID: ${process.pid}`); + + let httpServer = this.httpServer; + let scServer = this.scServer; + + (async () => { + for await (let {data} of this.listener('masterMessage')) { + if (data.sendGoodRequestToMaster) { + let result = await this.sendRequestToMaster({value: 123}); + this.sendMessageToMaster({success: result.value === 12300}); + } else if (data.sendBadRequestToMaster) { + let result; + let error; + try { + result = await this.sendRequestToMaster({fail: true}); + } catch (err) { + error = err; + } + // Send success if we got an error as expected. + this.sendMessageToMaster({ + success: error && error.name === 'MasterFailedToRespondError' + }); + } else { + this.sendMessageToMaster({value: data.value + 1}); + } + } + })(); + + (async () => { + for await (let req of this.listener('masterRequest')) { + if (req.data.fail) { + let error = new Error('Worker could not process request'); + error.name = 'WorkerFailedToRespondError'; + req.error(error); + } else { + req.end({value: req.data.value * 10}); + } + } + })(); + } +} + +new Worker(); diff --git a/test/external/broker.js b/test/external/broker.js deleted file mode 100644 index 75761afa..00000000 --- a/test/external/broker.js +++ /dev/null @@ -1,9 +0,0 @@ -var SCBroker = require('../../scbroker'); - -class Broker extends SCBroker { - run() { - console.log(' >> Broker PID:', process.pid); - } -} - -new Broker(); diff --git a/test/external/index.js b/test/external/index.js deleted file mode 100644 index dd03e277..00000000 --- a/test/external/index.js +++ /dev/null @@ -1,331 +0,0 @@ -var async = require('async'); -var childProcess = require('child_process'); -var scClient = require('socketcluster-client'); -var assert = require('assert'); - -var scServer = childProcess.fork(__dirname + '/server.js'); - -var options = { - hostname: '127.0.0.1', - port: 8000, - autoReconnect: true, - multiplex: false -}; - -scServer.on('message', function (m) { - if (m.event == 'ready') { - var socket = scClient.connect(options); - - var pongChannel; - - var tasks = [ - function (cb) { - socket.on('first', function (data) { - var err; - try { - assert(data == 'This is the first event', 'Received incorrect data from "first" event'); - } catch (e) { - err = e; - } - socket.removeAllListeners('first'); - cb(err); - }); - }, - function (cb) { - pongChannel = socket.subscribe('pong'); - pongChannel.watch(function (data) { - var err; - try { - assert(JSON.stringify(data) == JSON.stringify({message: 'This is pong data'}), - 'Received incorrect data from "pong" event'); - } catch (e) { - err = e; - } - pongChannel.unwatch(); - cb(err); - }); - socket.emit('ping'); - }, - function (cb) { - pongChannel.watch(function (data) { - var err; - try { - assert(JSON.stringify(data) == JSON.stringify({message: 'published pong'}), - 'Received incorrect data from published "pong" event'); - } catch (e) { - err = e; - } - cb(err); - pongChannel.unwatch(); - }); - socket.publish('pong', {message: 'published pong'}); - }, - function (cb) { - socket.unsubscribe('pong'); - var fooChannel = socket.subscribe('foo'); - fooChannel.on('subscribe', function () { - var err; - try { - var subscriptions = socket.subscriptions(); - assert(JSON.stringify(subscriptions) == JSON.stringify(['foo']), - 'Expected subscriptions() array to contain one "foo" channel'); - } catch (e) { - err = e; - } - fooChannel.removeAllListeners('subscribe'); - cb(err); - }); - }, - function (cb) { - socket.subscribe('foo2'); - - setTimeout(function () { - socket.unsubscribe('foo2'); - - setTimeout(function () { - cb(); - }, 1000); - }, 1000); - }, - function (cb) { - socket.emit('killWorker'); - socket.once('error', function (err) { - console.log('Caught:', err); - }); - - var notUnsubscribedTimeout = setTimeout(function () { - cb('Did not unsubscribe from channels on disconnect'); - }, 3000); - - socket.once('unsubscribe', function () { - clearTimeout(notUnsubscribedTimeout); - var err; - - try { - var subscriptions = socket.subscriptions(); - assert(JSON.stringify(subscriptions) == JSON.stringify([]), - 'Did not unsubscribe from channels on disconnect'); - } catch (e) { - err = e; - } - cb(err); - }); - }, - function (cb) { - setTimeout(function () { - socket.emit('new'); - var err; - - setTimeout(function () { - try { - var subscriptions = socket.subscriptions(); - - assert(JSON.stringify(subscriptions) == JSON.stringify(['foo']), - 'Did not automatically resubscribe to the correct channels which were unsubscribed due to disconnection'); - } catch (e) { - err = e; - } - cb(err); - }, 2000); - }, 1000); - }, - function (cb) { - socket.subscribe('test'); - setTimeout(function () { - var unsubscribeEmitted = false; - - socket.on('unsubscribe', function (channel) { - if (channel == 'test') { - unsubscribeEmitted = true; - } - }); - - socket.unsubscribe('test'); - var err; - - setTimeout(function () { - try { - var subscriptions = socket.subscriptions(); - - assert(unsubscribeEmitted, - 'Socket did not emit unsubscribe event after calling socket.unsubscribe(channelName) method'); - } catch (e) { - err = e; - } - cb(err); - }, 2000); - }, 1000); - }, - function (cb) { - var actionSequence = []; - - socket.on('subscribe', function (channel) { - if (channel == 'channel1') { - actionSequence.push('subscribe'); - } - }); - - socket.on('unsubscribe', function (channel) { - if (channel == 'channel1') { - actionSequence.push('unsubscribe'); - } - }); - - socket.subscribe('channel1'); - socket.unsubscribe('channel1'); - socket.subscribe('channel1'); - socket.unsubscribe('channel1'); - - var expectedActionSequence = []; - - var err; - - setTimeout(function () { - socket.off('subscribe'); - socket.off('unsubscribe'); - - try { - assert(JSON.stringify(actionSequence) == JSON.stringify(expectedActionSequence), - 'Subscribing and unsubscribing to channel1 multiple times in a sequence was not handled in an optimal way'); - } catch (e) { - err = e; - } - cb(err); - }, 1000); - }, - function (cb) { - var actionSequence = []; - - socket.subscribe('channel2'); - - setTimeout(function () { - socket.on('subscribe', function (channel) { - if (channel == 'channel2') { - actionSequence.push('subscribe'); - } - }); - - socket.on('unsubscribe', function (channel) { - if (channel == 'channel2') { - actionSequence.push('unsubscribe'); - } - }); - - socket.unsubscribe('channel2'); - socket.subscribe('channel2'); - socket.unsubscribe('channel2'); - socket.subscribe('channel2'); - socket.unsubscribe('channel2'); - socket.subscribe('channel2'); - - var expectedActionSequence = [ - 'unsubscribe', - 'subscribe' - ]; - - var err; - - setTimeout(function () { - socket.off('subscribe'); - socket.off('unsubscribe'); - try { - assert(JSON.stringify(actionSequence) == JSON.stringify(expectedActionSequence), - 'Subscribing and unsubscribing to channel2 multiple times in a sequence was not handled in an optimal way'); - } catch (e) { - err = e; - } - cb(err); - }, 1000); - }, 1000); - }, - function (cb) { - var err; - - socket.once('disconnect', function () { - socket.once('connect', function (status) { - try { - assert(!status.isAuthenticated, - 'Socket should not be authenticated'); - } catch (e) { - err = e; - } - cb(err); - }); - socket.connect(); - }); - socket.disconnect(); - }, - function (cb) { - var err; - - socket.once('connect', function (status) { - - var authTokenIsSet = false; - socket.on('authenticate', function () { - authTokenIsSet = true; - }); - - socket.emit('login', {username: 'john123'}); - - setTimeout(function () { - - socket.once('connect', function (status) { - try { - assert(!!status.isAuthenticated, - 'Socket should be authenticated'); - assert(authTokenIsSet, - 'authenticate event was never emitted'); - } catch (e) { - err = e; - } - cb(err); - }); - - socket.disconnect(); - socket.connect(); - - }, 1000); - }); - - socket.disconnect(); - socket.connect(); - } - ]; - - var timedTasks = []; - var timeoutMs = 20000; - - var timeoutError = function () { - throw new Error('Test timed out'); - }; - - var assertTimeout = null; - var timeoutTask = function (cb) { - clearTimeout(assertTimeout); - assertTimeout = setTimeout(timeoutError, timeoutMs); - cb(); - }; - - for (var i in tasks) { - timedTasks.push(timeoutTask); - timedTasks.push(tasks[i]); - } - timedTasks.push(function (cb) { - clearTimeout(assertTimeout); - socket.disconnect(); - scServer.kill(); - setTimeout(function () { - cb(); - }, 1000); - }); - - async.waterfall(timedTasks, function (err) { - if (err) { - throw err; - } else { - console.log('All tests passed!'); - } - process.exit(); - }); - } -}); diff --git a/test/external/server.js b/test/external/server.js deleted file mode 100644 index 3eebecd2..00000000 --- a/test/external/server.js +++ /dev/null @@ -1,19 +0,0 @@ -var SocketCluster = require('../../index'); - -var socketCluster = new SocketCluster({ - workers: 1, - brokers: 1, - port: 8000, - appName: 'myapp', - workerController: __dirname + '/worker.js', - brokerController: __dirname + '/broker.js', - socketChannelLimit: 100, - crashWorkerOnError: true, - rebootWorkerOnCrash: true, - pingTimeout: 3000, - pingInterval: 1000 -}); - -socketCluster.on('ready', function () { - process.send({event: 'ready'}); -}); diff --git a/test/external/worker.js b/test/external/worker.js deleted file mode 100644 index eb9b8250..00000000 --- a/test/external/worker.js +++ /dev/null @@ -1,41 +0,0 @@ -var SCWorker = require('../../scworker'); -var fs = require('fs'); - -class Worker extends SCWorker { - run() { - console.log(' >> Worker PID:', process.pid); - - // Get a reference to our raw Node HTTP server - var httpServer = this.getHTTPServer(); - // Get a reference to our WebSocket server - var wsServer = this.getSCServer(); - - var pongData = {message: 'This is pong data'}; - - /* - In here we handle our incoming WebSocket connections and listen for events. - From here onwards is just like Socket.io but with some additional features. - */ - wsServer.on('connection', (socket) => { - socket.emit('first', 'This is the first event'); - - socket.on('ping', () => { - wsServer.exchange.publish('pong', pongData); - }); - - socket.on('login', (username) => { - socket.setAuthToken({username: username}); - }); - - socket.on('killWorker', () => { - process.exit(); - }); - - socket.on('new', () => { - console.log('Received new event'); - }); - }); - } -} - -new Worker(); diff --git a/test/integration.js b/test/integration.js new file mode 100644 index 00000000..1af850c1 --- /dev/null +++ b/test/integration.js @@ -0,0 +1,361 @@ +const assert = require('assert'); +const SocketCluster = require('socketcluster'); +const socketClusterClient = require('socketcluster-client'); +const path = require('path'); + +const PORT = 8088; +const WS_ENGINE = 'ws'; +const ENVIRONMENT = 'dev'; +const CLIENT_COUNT = 10; + +function wait(duration) { + return new Promise((resolve) => { + setTimeout(() => { + resolve(); + }, duration); + }); +} + +let socketCluster; +let clients; + +describe('Integration tests', function () { + describe('Single worker and broker', function () { + before(async function () { + let options = { + workers: 1, + brokers: 1, + port: PORT, + wsEngine: WS_ENGINE, + workerController: path.join(__dirname, 'controllers', 'worker.js'), + brokerController: path.join(__dirname, 'controllers', 'broker.js'), + socketChannelLimit: 1000, + crashWorkerOnError: true, + killMasterOnSignal: false, + processTermTimeout: 500, + forceKillTimeout: 1000, + brokerCommandAckTimeout: 1112, + environment: ENVIRONMENT + }; + socketCluster = new SocketCluster(options); + await socketCluster.listener(socketCluster.EVENT_READY).once(); + }); + + after(async function () { + await socketCluster.destroy(); + }); + + beforeEach(async function () { + clients = []; + for (let i = 0; i < CLIENT_COUNT; i++) { + clients.push( + socketClusterClient.create({ + hostname: '127.0.0.1', + port: PORT, + autoReconnect: true, + autoReconnectOptions: { + initialDelay: 50, + randomness: 0, + multiplier: 1, + maxDelay: 50 + } + }) + ); + } + + await Promise.all( + clients.map(async (client) => { + await client.listener('connect').once(); + }) + ); + }); + + afterEach(async function () { + clients.forEach((client) => { + client.disconnect(); + }); + }); + + describe('Pub/sub', function () { + it('Should publish data to and receive data from subscribed channel', async function () { + let channel = clients[0].subscribe('foo'); + await channel.listener('subscribe').once(); + + (async () => { + await wait(10); + clients[1].publish('foo', 'This is a test'); + await wait(10); + clients[2].publish('foo', 'One'); + await wait(20); + clients[5].publish('foo', 'Two'); + })(); + + let data = await channel.once(); + assert.equal(data, 'This is a test'); + data = await channel.once(); + assert.equal(data, 'One'); + data = await channel.once(); + assert.equal(data, 'Two'); + }); + + it('Should handle multiple subscribe and unsubscribe calls in quick succession', async function () { + let channel = clients[0].channel('foo'); + + (async () => { + await wait(10); + channel.subscribe(); + await channel.listener('subscribe').once(); + channel.unsubscribe(); + channel.subscribe(); + channel.unsubscribe(); + channel.subscribe(); + channel.unsubscribe(); + await wait(100); + channel.closeAllListeners(); + })(); + + let stateChanges = []; + for await (let stateChangeData of channel.listener('subscribeStateChange')) { + stateChanges.push(stateChangeData); + } + let expectedStateChanes = [ + {oldChannelState: 'pending', newChannelState: 'subscribed', subscriptionOptions: {}}, + {oldChannelState: 'subscribed', newChannelState: 'unsubscribed'} + ]; + assert.equal(JSON.stringify(stateChanges), JSON.stringify(expectedStateChanes)); + }); + + it.skip('Should clear all subscriptions after all sockets have been disconnected', async function () { + + }); + }); + + describe('During worker restart', function () { + it('Pub/sub should work after worker restart', async function () { + socketCluster.killWorkers(); + await socketCluster.listener(socketCluster.EVENT_WORKER_EXIT).once(); + let channel = clients[0].subscribe('foo'); + await channel.listener('subscribe').once(); + + (async () => { + await wait(10); + clients[1].publish('foo', 'This is a test'); + await wait(10); + clients[2].publish('foo', 'One'); + await wait(20); + clients[5].publish('foo', 'Two'); + await wait(20); + channel.close(); + })(); + + let receivedList = []; + for await (let data of channel) { + receivedList.push(data); + } + + assert.equal(receivedList[0], 'This is a test'); + assert.equal(receivedList[1], 'One'); + assert.equal(receivedList[2], 'Two'); + }); + }); + + describe('During broker restart', function () { + it('Pub/sub should work after broker restart', async function () { + socketCluster.killBrokers(); + await socketCluster.listener(socketCluster.EVENT_BROKER_EXIT).once(); + let channel = clients[7].subscribe('foo'); + await channel.listener('subscribe').once(); + + (async () => { + await wait(10); + clients[1].publish('foo', 'This is a test'); + await wait(10); + clients[2].publish('foo', 'One'); + await wait(20); + clients[5].publish('foo', 'Two'); + await wait(20); + channel.close(); + })(); + + let receivedList = []; + for await (let data of channel) { + receivedList.push(data); + } + + assert.equal(receivedList[0], 'This is a test'); + assert.equal(receivedList[1], 'One'); + assert.equal(receivedList[2], 'Two'); + }); + }); + }); + + describe('Multiple workers and brokers', function () { + before(async function () { + let options = { + workers: 3, + brokers: 3, + port: PORT, + wsEngine: WS_ENGINE, + workerController: path.join(__dirname, 'controllers', 'worker.js'), + brokerController: path.join(__dirname, 'controllers', 'broker.js'), + socketChannelLimit: 1000, + crashWorkerOnError: true, + killMasterOnSignal: false, + processTermTimeout: 500, + forceKillTimeout: 1000, + environment: ENVIRONMENT + }; + socketCluster = new SocketCluster(options); + await socketCluster.listener(socketCluster.EVENT_READY).once(); + }); + + after(async function () { + await socketCluster.destroy(); + }); + + beforeEach(async function () { + clients = []; + for (let i = 0; i < CLIENT_COUNT; i++) { + clients.push( + socketClusterClient.create({ + hostname: '127.0.0.1', + port: PORT + }) + ); + } + + await Promise.all( + clients.map(async (client) => { + await client.listener('connect').once(); + }) + ); + }); + + afterEach(async function () { + clients.forEach((client) => { + client.disconnect(); + }); + }); + + describe('Pub/sub', function () { + it('Should publish data to and receive data from subscribed channel', async function () { + let channel = clients[0].subscribe('foo'); + await channel.listener('subscribe').once(); + + (async () => { + await wait(10); + clients[1].publish('foo', 'This is a test'); + await wait(10); + clients[2].publish('foo', 'One'); + await wait(20); + clients[5].publish('foo', 'Two'); + })(); + + let data = await channel.once(); + assert.equal(data, 'This is a test'); + data = await channel.once(); + assert.equal(data, 'One'); + data = await channel.once(); + assert.equal(data, 'Two'); + }); + }); + + describe('IPC requests and messages', function () { + it('Should support sending a message from the master process to a worker and back', async function () { + socketCluster.sendMessageToWorker(1, {value: 1}); + let {workerId, data} = await socketCluster.listener('workerMessage').once(); + assert.equal(workerId, 1); + assert.notEqual(data, null); + assert.equal(data.value, 2); + }); + + it('Should support sending a message from the master process to a broker and back', async function () { + socketCluster.sendMessageToBroker(0, {value: 10}); + let {brokerId, data} = await socketCluster.listener('brokerMessage').once(); + assert.equal(brokerId, 0); + assert.notEqual(data, null); + assert.equal(data.value, 11); + }); + + it('Should support sending a request from the master process to a worker and back', async function () { + // Send good request. + let result = await socketCluster.sendRequestToWorker(2, {value: 3}); + assert.notEqual(result, null); + assert.equal(result.value, 30); + + // Send bad request. + let error; + result = null; + try { + result = await socketCluster.sendRequestToWorker(1, {fail: true}); + } catch (err) { + error = err; + } + assert.notEqual(error, null); + assert.equal(error.name, 'WorkerFailedToRespondError'); + + // Ask worker to send us a good request. + socketCluster.sendMessageToWorker(0, {sendGoodRequestToMaster: true}); + let req = await socketCluster.listener('workerRequest').once(); + req.end({value: req.data.value * 100}); + + let {workerId, data} = await socketCluster.listener('workerMessage').once(); + assert.equal(workerId, 0); + assert.notEqual(data, null); + assert.equal(data.success, true); + + // Ask worker to send us a bad request. + socketCluster.sendMessageToWorker(2, {sendBadRequestToMaster: true}); + req = await socketCluster.listener('workerRequest').once(); + let failedResponseError = new Error('Master failed to respond'); + failedResponseError.name = 'MasterFailedToRespondError'; + req.error(failedResponseError); + + let brokerResultMessage = await socketCluster.listener('workerMessage').once(); + assert.equal(brokerResultMessage.workerId, 2); + assert.notEqual(brokerResultMessage.data, null); + assert.equal(brokerResultMessage.data.success, true); + }); + + it('Should support sending a request from the master process to a broker and back', async function () { + // Send good request. + let result = await socketCluster.sendRequestToBroker(2, {value: 3}); + assert.notEqual(result, null); + assert.equal(result.value, 30); + + // Send bad request. + let error; + result = null; + try { + result = await socketCluster.sendRequestToBroker(1, {fail: true}); + } catch (err) { + error = err; + } + assert.notEqual(error, null); + assert.equal(error.name, 'BrokerFailedToRespondError'); + + // Ask broker to send us a good request. + socketCluster.sendMessageToBroker(0, {sendGoodRequestToMaster: true}); + let req = await socketCluster.listener('brokerRequest').once(); + req.end({value: req.data.value * 100}); + + let {brokerId, data} = await socketCluster.listener('brokerMessage').once(); + assert.equal(brokerId, 0); + assert.notEqual(data, null); + assert.equal(data.success, true); + + // Ask broker to send us a bad request. + socketCluster.sendMessageToBroker(2, {sendBadRequestToMaster: true}); + req = await socketCluster.listener('brokerRequest').once(); + let failedResponseError = new Error('Master failed to respond'); + failedResponseError.name = 'MasterFailedToRespondError'; + req.error(failedResponseError); + + let brokerResultMessage = await socketCluster.listener('brokerMessage').once(); + assert.equal(brokerResultMessage.brokerId, 2); + assert.notEqual(brokerResultMessage.data, null); + assert.equal(brokerResultMessage.data.success, true); + }); + }); + }); +}); diff --git a/test/internal/broker.js b/test/internal/broker.js deleted file mode 100644 index 0f9385b8..00000000 --- a/test/internal/broker.js +++ /dev/null @@ -1,50 +0,0 @@ -var SCBroker = require('../../scbroker'); -var fs = require('fs'); -var http = require('http'); -var getTestSocketPath = require('./test-socket-path').getTestSocketPath; -var util = require('util'); - -class Broker extends SCBroker { - run() { - console.log(' >> Broker PID:', process.pid); - - var resultSocketPath = getTestSocketPath(); - - // Send test data to index.js every second - var testDataInterval = setInterval(() => { - var socketChannelData = this.subscriptions; - var channels = []; - for (var i in socketChannelData) { - channels = channels.concat(Object.keys(socketChannelData[i])); - } - - var req = http.request({ - socketPath: resultSocketPath, - method: 'POST' - }, (res) => { - res.on('data', (data) => { - if (data.toString() == 'finish') { - clearInterval(testDataInterval); - } - }); - }); - req.write(JSON.stringify([ - { - origin: 'broker', - pid: process.pid, - type: 'channels', - data: channels - }, - { - origin: 'broker', - pid: process.pid, - type: 'all', - data: this.dataMap.getAll() - } - ])); - req.end(); - }, 1000); - } -} - -new Broker(); diff --git a/test/internal/broker2.js b/test/internal/broker2.js deleted file mode 100644 index 57fd6c7e..00000000 --- a/test/internal/broker2.js +++ /dev/null @@ -1,56 +0,0 @@ -var SCBroker = require('../../scbroker'); - -class Broker extends SCBroker { - run() { - console.log(' >> Broker PID:', process.pid); - - this.on('masterMessage', (data, res) => { - console.log(`BROKER ${this.id}::Received data from master:`, data); - if (data.fail) { - var err = new Error('This is an error from broker'); - err.name = 'MyCustomBrokerError'; - res(err); - } else if (!data.doNothing) { - res(null, { - id: 1, - name: 'TestName' - }); - } - }); - - var packet = { - prop: 1234 - }; - - console.log(`BROKER ${this.id}::Sending packet to master`); - this.sendToMaster(packet, (err, data) => { - console.log(`BROKER ${this.id}::Response error from master:`, err); - console.log(`BROKER ${this.id}::Response data packet from master:`, data); - }); - - var timeoutPacket = { - doNothing: true - }; - - console.log(`BROKER ${this.id}::Sending timeout-causing packet to master`); - this.sendToMaster(timeoutPacket, (err, data) => { - console.log(`BROKER ${this.id}::Timeout response error from master:`, err); - console.log(`BROKER ${this.id}::Timeout response data packet from master:`, data); - }); - - var errorPacket = { - fail: true - }; - - console.log(`BROKER ${this.id}::Sending error-causing packet to master`); - this.sendToMaster(errorPacket, (err, data) => { - console.log(`BROKER ${this.id}::Error response error from master:`, err); - console.log(`BROKER ${this.id}::Error response data packet from master:`, data); - }); - - console.log(`BROKER ${this.id}::Sending error-causing packet to master without callback`); - this.sendToMaster(errorPacket); - } -} - -new Broker(); diff --git a/test/internal/index.js b/test/internal/index.js deleted file mode 100644 index 3679b395..00000000 --- a/test/internal/index.js +++ /dev/null @@ -1,238 +0,0 @@ -var async = require('async'); -var childProcess = require('child_process'); -var scClient = require('socketcluster-client'); -var assert = require('assert'); -var http = require('http'); -var util = require('util'); -var fs = require('fs'); -var getTestSocketPath = require('./test-socket-path').getTestSocketPath; - -var scServer = childProcess.fork(__dirname + '/server.js'); -var resultSocketPath = getTestSocketPath(); - -var fileExistsSync = function (filePath) { - try { - fs.accessSync(filePath, fs.constants.F_OK); - } catch (err) { - return false; - } - return true; -}; - -if (process.platform != 'win32') { - if (fileExistsSync(resultSocketPath)) { - fs.unlinkSync(resultSocketPath); - } -} - -var brokerData = {}; -var testIsOver = false; - -var resultsServer = http.createServer(function (req, res) { - var buf = []; - req.on('data', function (data) { - buf.push(data); - }); - req.on('end', function () { - var results = JSON.parse(Buffer.concat(buf).toString()); - var result; - for (var i in results) { - result = results[i]; - if (result.origin == 'broker') { - if (brokerData[result.pid] == null) { - brokerData[result.pid] = {}; - } - brokerData[result.pid][result.type] = result.data; - } - } - res.writeHead(200); - if (testIsOver) { - res.end('finish'); - } else { - res.end(); - } - }); -}).listen(resultSocketPath); - -var endTest = function (callback) { - testIsOver = true; - setTimeout(callback, 4000); -}; - -var options = { - hostname: '127.0.0.1', - port: 8000, - autoReconnect: true, - multiplex: false -}; - -scServer.on('message', function (m) { - var numSockets = 100; - var messageInterval = 2000; - var socketCount = 0; - - if (m.event == 'ready') { - var sockets = []; - var tasks = []; - - for (var i = 0; i < numSockets; i++) { - tasks.push(function (cb) { - var socket = scClient.connect(options); - socket.once('connect', function () { - if (socket.getState() == socket.OPEN) { - // Subscribe to some events - socket.on('event1', function () {}); - socket.on('event2', function () {}); - socket.on('event3', function () {}); - socket.on('event4', function () {}); - - console.log('#' + ++socketCount + ' - Socket ' + socket.id + ' connected'); - - var interval = setInterval(function () { - if (socket.getState() == socket.OPEN) { - socket.emit('test', {id: socket.id}); - } else { - clearInterval(interval); - } - }, messageInterval); - messageInterval += 100; - - cb(); - } else { - cb(new Error('The client socket\'s getState() should return socket.OPEN after it emits a connect event')); - } - }); - sockets.push(socket); - }); - } - - tasks.push(function (cb) { - console.log('Done connecting'); - console.log(); - console.log('Checking that a channel gets cleaned up after all clients unsubscribe from it'); - - var numTest = 5; - var c = 0; - - for (var i in sockets) { - sockets[i].subscribe('foo'); - if (c++ >= numTest) { - break; - } - } - - var checkResults = function () { - var channels = {}; - for (var i in brokerData) { - channels[i] = brokerData[i].channels; - } - - var subscriptionsAsString = util.inspect(channels, {depth: 5}); - console.log('Broker channels after unsubscribe:', subscriptionsAsString); - - var subscriptionsHasFoo = /foo/.test(subscriptionsAsString); - assert(!subscriptionsHasFoo, 'Channel was not cleaned up after all clients unsubscribed from it'); - - console.log('[Success] Broker channel was cleaned up after all clients unsubscribed from it'); - - cb(); - }; - - c = 0; - - setTimeout(function () { - for (var i in sockets) { - sockets[i].unsubscribe('foo'); - if (c++ >= numTest) { - break; - } - } - setTimeout(checkResults, 2000); - }, 3000); - }); - - var assertTimeout = null; - - tasks.push(function (cb) { - console.log(); - clearTimeout(assertTimeout); - assertTimeout = setTimeout(timeoutError, 10000); - - console.log('Checking that broker data is empty after disconnecting all sockets'); - - for (var i in sockets) { - sockets[i].disconnect(); - } - - setTimeout(function () { - var allData = {}; - for (var i in brokerData) { - allData[i] = brokerData[i].all; - } - console.log('Broker data after disconnecting all sockets:', util.inspect(allData, {depth: 5})); - - for (var j in allData) { - var isBrokerDataEmpty = JSON.stringify(allData[j]).length < 70; - assert(isBrokerDataEmpty, 'Broker data was not cleaned up after all sockets were disconnected'); - } - console.log('[Success] Broker data was cleaned up after all sockets were disconnected'); - - cb(); - }, 2000); - }); - - tasks.push(function (cb) { - console.log(); - console.log('Checking that channels get cleaned up after all sockets were disconnected'); - - var channels = {}; - for (var i in brokerData) { - channels[i] = brokerData[i].channels; - } - console.log('Broker channels after disconnecting all sockets:', util.inspect(channels, {depth: 5})); - - for (var j in channels) { - var areSubscriptionsEmpty = JSON.stringify(channels[j]).length < 50; - assert(areSubscriptionsEmpty, 'Channels were not cleaned up after disconnecting all sockets'); - } - console.log('[Success] Broker channels were cleaned up after disconnecting all sockets'); - - cb(); - }); - - var timedTasks = []; - var timeoutMs = 20000; - - var timeoutError = function () { - throw new Error('Test timed out'); - }; - - var timeoutTask = function (cb) { - clearTimeout(assertTimeout); - assertTimeout = setTimeout(timeoutError, timeoutMs); - cb(); - }; - - for (var i in tasks) { - timedTasks.push(timeoutTask); - timedTasks.push(tasks[i]); - } - timedTasks.push(function (cb) { - clearTimeout(assertTimeout); - scServer.kill(); - cb(); - }); - - async.waterfall(timedTasks, function (err) { - if (err) { - throw err; - } else { - console.log(); - console.log('All tests passed!'); - endTest(function () { - process.exit(); - }); - } - }); - } -}); diff --git a/test/internal/master-to-broker.js b/test/internal/master-to-broker.js deleted file mode 100644 index b57cc449..00000000 --- a/test/internal/master-to-broker.js +++ /dev/null @@ -1,76 +0,0 @@ -var SocketCluster = require('../../index'); - -var socketCluster = new SocketCluster({ - workers: 2, - brokers: 2, - port: 8000, - appName: 'myapp', - workerController: __dirname + '/worker2.js', - brokerController: __dirname + '/broker2.js', - socketChannelLimit: 100, - crashWorkerOnError: true, - rebootWorkerOnCrash: true, - ipcAckTimeout: 1000 -}); - -socketCluster.on('brokerMessage', function (brokerId, data, respond) { - console.log(`MASTER::Received data from broker ${brokerId}:`, data); - if (data.fail) { - var err = new Error('Failure from master'); - err.name = 'MasterFailureError'; - respond(err); - } else if (!data.doNothing) { - respond(null, { - qwerty: 123456 - }); - } -}); - -var packet = { - id: 0, - name: 'SocketCluster' -}; - -console.log('MASTER::Sending packet to broker 0'); -socketCluster.sendToBroker(0, packet, function (err, data) { - console.log('MASTER::Response error from broker 0:', err); - console.log('MASTER::Response data from broker 0:', data); -}); - -console.log('MASTER::Sending packet to broker 1'); -socketCluster.sendToBroker(1, packet, function (err, data) { - console.log('MASTER::Response error from broker 1:', err); - console.log('MASTER::Response data from broker 1:', data); -}); - -console.log('MASTER::Sending packet to non-existent broker 2'); -socketCluster.sendToBroker(2, packet, function (err, data) { - console.log('MASTER::Response error from non-existent broker 2:', err); - console.log('MASTER::Response data from non-existent broker 2:', data); -}); - -var errorPacket = { - fail: true -}; - -console.log('MASTER::Sending error-causing packet to broker 0'); -socketCluster.sendToBroker(0, errorPacket, function (err, data) { - console.log('MASTER::Error response error from broker 0:', err); - console.log('MASTER::Error response data from broker 0:', data); -}); - -console.log('MASTER::Sending error-causing packet to broker without callback 0'); -socketCluster.sendToBroker(0, errorPacket); - -var timeoutPacket = { - doNothing: true -}; - -console.log('MASTER::Sending timeout-causing packet to broker 0'); -socketCluster.sendToBroker(0, timeoutPacket, function (err, data) { - console.log('MASTER::Timeout response error from broker 0:', err); - console.log('MASTER::Timeout response data from broker 0:', data); -}); - -console.log('MASTER::Sending timeout-causing packet to broker 0 without callback'); -socketCluster.sendToBroker(0, timeoutPacket); diff --git a/test/internal/master-to-worker.js b/test/internal/master-to-worker.js deleted file mode 100644 index 48c046f4..00000000 --- a/test/internal/master-to-worker.js +++ /dev/null @@ -1,76 +0,0 @@ -var SocketCluster = require('../../index'); - -var socketCluster = new SocketCluster({ - workers: 2, - brokers: 2, - port: 8000, - appName: 'myapp', - workerController: __dirname + '/worker.js', - socketChannelLimit: 100, - crashWorkerOnError: true, - rebootWorkerOnCrash: true, - ipcAckTimeout: 1000 -}); - -socketCluster.on('workerMessage', function (workerId, data, respond) { - console.log(`MASTER::Received data from worker ${workerId}:`, data); - if (data.fail) { - var err = new Error('Failure from master'); - err.name = 'MasterFailureError'; - respond(err); - } else if (!data.doNothing) { - - respond(null, { - qwerty: 123456 - }); - } -}); - -var packet = { - id: 0, - name: 'SocketCluster' -}; - -console.log('MASTER::Sending packet to worker 0'); -socketCluster.sendToWorker(0, packet, function (err, data) { - console.log('MASTER::Response error from worker 0:', err); - console.log('MASTER::Response data from worker 0:', data); -}); - -console.log('MASTER::Sending packet to worker 1'); -socketCluster.sendToWorker(1, packet, function (err, data) { - console.log('MASTER::Response error from worker 1:', err); - console.log('MASTER::Response data from worker 1:', data); -}); - -console.log('MASTER::Sending packet to non-existent worker 2'); -socketCluster.sendToWorker(2, packet, function (err, data) { - console.log('MASTER::Response error from non-existent worker 2:', err); - console.log('MASTER::Response data from non-existent worker 2:', data); -}); - -var errorPacket = { - fail: true -}; - -console.log('MASTER::Sending error-causing packet to worker 0'); -socketCluster.sendToWorker(0, errorPacket, function (err, data) { - console.log('MASTER::Error response error from worker 0:', err); - console.log('MASTER::Error response data from worker 0:', data); -}); - -console.log('MASTER::Sending error-causing packet to worker 0 without callback'); -socketCluster.sendToWorker(0, errorPacket); - -var timeoutPacket = { - doNothing: true -}; - -console.log('MASTER::Sending timeout-causing packet to worker 0'); -socketCluster.sendToWorker(0, timeoutPacket, function (err, data) { - console.log('MASTER::Timeout response error from worker 0:', err); - console.log('MASTER::Timeout response data from worker 0:', data); -}); - -console.log('MASTER::Sending timeout-causing packet to worker 0 without callback'); -socketCluster.sendToWorker(0, timeoutPacket); diff --git a/test/internal/server.js b/test/internal/server.js deleted file mode 100644 index 64554ae3..00000000 --- a/test/internal/server.js +++ /dev/null @@ -1,17 +0,0 @@ -var SocketCluster = require('../../index'); - -var socketCluster = new SocketCluster({ - workers: 3, - brokers: 3, - port: 8000, - appName: 'myapp', - workerController: __dirname + '/worker.js', - brokerController: __dirname + '/broker.js', - socketChannelLimit: 100, - crashWorkerOnError: true, - rebootWorkerOnCrash: true -}); - -socketCluster.on('ready', function () { - process.send({event: 'ready'}); -}); diff --git a/test/internal/test-socket-path.js b/test/internal/test-socket-path.js deleted file mode 100644 index 8cbc4f99..00000000 --- a/test/internal/test-socket-path.js +++ /dev/null @@ -1,8 +0,0 @@ -var os = require('os'); - -module.exports.getTestSocketPath = function () { - if (process.platform == 'win32') { - return '\\\\.\\pipe\\socketclustertest'; - } - return os.tmpdir() + '/socketclustertest'; -}; \ No newline at end of file diff --git a/test/internal/worker.js b/test/internal/worker.js deleted file mode 100644 index 9737255b..00000000 --- a/test/internal/worker.js +++ /dev/null @@ -1,65 +0,0 @@ -var SCWorker = require('../../scworker'); -var fs = require('fs'); - -class Worker extends SCWorker { - run() { - console.log(' >> Worker PID:', process.pid); - - // Get a reference to our raw Node HTTP server - var httpServer = this.getHTTPServer(); - // Get a reference to our WebSocket server - var wsServer = this.getSCServer(); - - this.on('masterMessage', (data, res) => { - console.log(`WORKER ${this.id}::Received data from master:`, data); - if (data.fail) { - var err = new Error('This is an error from worker'); - err.name = 'MyCustomWorkerError'; - res(err); - } else if (!data.doNothing) { - res(null, { - id: 1, - name: 'TestName' - }); - } - }); - - var packet = { - prop: 1234 - }; - - console.log(`WORKER ${this.id}::Sending packet to master`); - this.sendToMaster(packet, (err, data) => { - console.log(`WORKER ${this.id}::Response error from master:`, err); - console.log(`WORKER ${this.id}::Response data packet from master:`, data); - }); - - var timeoutPacket = { - doNothing: true - }; - - console.log(`WORKER ${this.id}::Sending timeout-causing packet to master`); - this.sendToMaster(timeoutPacket, (err, data) => { - console.log(`WORKER ${this.id}::Timeout response error from master:`, err); - console.log(`WORKER ${this.id}::Timeout response data packet from master:`, data); - }); - - console.log(`WORKER ${this.id}::Sending timeout-causing packet to master without callback`); - this.sendToMaster(timeoutPacket); - - var errorPacket = { - fail: true - }; - - console.log(`WORKER ${this.id}::Sending error-causing packet to master`); - this.sendToMaster(errorPacket, (err, data) => { - console.log(`WORKER ${this.id}::Error response error from master:`, err); - console.log(`WORKER ${this.id}::Error response data packet from master:`, data); - }); - - console.log(`WORKER ${this.id}::Sending error-causing packet to master without callback`); - this.sendToMaster(errorPacket); - } -} - -new Worker(); diff --git a/test/internal/worker2.js b/test/internal/worker2.js deleted file mode 100644 index e8185d0f..00000000 --- a/test/internal/worker2.js +++ /dev/null @@ -1,15 +0,0 @@ -var SCWorker = require('../../scworker'); -var fs = require('fs'); - -class Worker extends SCWorker { - run() { - console.log(' >> Worker PID:', process.pid); - - // Get a reference to our raw Node HTTP server - var httpServer = this.getHTTPServer(); - // Get a reference to our WebSocket server - var wsServer = this.getSCServer(); - } -} - -new Worker()