feat: improve testability, and test coverage

Author: jimlambie

.gitignore

@@ -1,4 +1,6 @@
 config.*json
 log/*
+*.log
 node_modules/
-coverage/
+coverage/
+scripts/

index.js

@@ -1,2 +1,2 @@
-const QueueHandler = require('./lib/app')
-module.exports = QueueHandler()
+const QueueHandler = require('./lib/queue-handler')
+module.exports = new QueueHandler()

lib/broker.js

@@ -1,37 +1,29 @@
 'use strict'
 
-const conf = require('../config')
+/**
+ * Message Broker
+ */
+const config = require('../config')
 const errs = require('./errors')
 const Rsmq = require('rsmq-worker')
 const Throttle = require('./throttle')
 
 /**
- * Message Broker
- * Polls the queue server for messages, builds requests
+ * @constructor
+ * @classdesc Polls the queue server for messages, builds requests
  * for the app and handles errors
- * @param {function} app - the function to call when a message is received
+ * @param {QueueHandler} queueHandler - the function to call when a message is received
  */
-var Broker = function (app) {
-  // get config options
-  var opts = conf.get('broker')
-
-  // instantiate message queue
-  this.rsmq = new Rsmq(opts.queue)
+var Broker = function (queueHandler) {
+  var opts = config.get('broker')
 
-  // instantiate throttle
-  var throttle = new Throttle(opts.throttle, (start, stop) => {
-    if (start) {
-      this.rsmq.start()
-    }
-
-    if (stop) {
-      this.rsmq.stop()
-    }
-  })
+  this.queueHandler = queueHandler
+  this.rsmq = this.initialiseQueue(opts)
+  this.throttle = this.initialiseThrottle(opts)
 
   // message received
   this.rsmq.on('message', (msg, next) => {
-    throttle.more()
+    this.throttle.more()
     next(false) // next message but don't delete
   })
 
@@ -51,54 +43,67 @@ var Broker = function (app) {
     }
 
     // call the app and delete message on success
-    console.log('Broker calling app')
-    app.handle(null, req, (err) => {
-      console.log('response from app')
-      if (processResponse(req, err)) {
+    this.queueHandler.handle(null, req, (err) => {
+      if (this.processResponse(req, err)) {
         this.rsmq.del(msg.id)
       }
 
-      throttle.less()
+      this.throttle.less()
     })
+  })
 
-    // done if no error and no retries remaining
-    function processResponse (req, err) {
-      var queueError
+  // queue error
+  this.rsmq.on('error', (err, msg) => {
+    this.queueHandler.handle(new errs.BrokerError(err, msg.message))
+  })
 
-      if (err) {
-        queueError = new errs.WorkerError(err, req.message)
-      }
+  // start listening
+  return this.rsmq.start()
+}
 
-      if (err && !req.retries) {
-        queueError = new errs.ExceededError(req.message)
-      }
+/**
+ *  done if no error and no retries remaining
+ */
+Broker.prototype.processResponse = function (req, err) {
+  var queueError
 
-      if (!req.address.length) {
-        queueError = new errs.InvalidError(req.message)
-      }
+  if (err) {
+    queueError = new errs.WorkerError(err, req.message)
+  }
 
-      if (Date.now() / 1000 > req.timeout) {
-        queueError = new errs.TimeoutError(req.message)
-      }
+  if (err && !req.retries) {
+    queueError = new errs.ExceededError(req.message)
+  }
 
-      if (queueError) {
-        app.handle(queueError)
-      }
+  if (!req.address.length) {
+    queueError = new errs.InvalidError(req.message)
+  }
 
-      return !err || !req.retries
-    }
-  })
+  if (Date.now() / 1000 > req.timeout) {
+    queueError = new errs.TimeoutError(req.message)
+  }
 
-  // queue error
-  this.rsmq.on('error', (err, msg) => {
-    console.log(err)
-    app(new errs.BrokerError(err, msg.message))
-  })
+  if (queueError) {
+    this.queueHandler.handle(queueError)
+  }
 
-  // start listening
-  return this.rsmq.start()
+  return !err || !req.retries
 }
 
-module.exports = function (app) {
-  return new Broker(app)
+Broker.prototype.initialiseQueue = function (options) {
+  return new Rsmq(options.queue)
 }
+
+Broker.prototype.initialiseThrottle = function (options) {
+  return new Throttle(options.throttle, (start, stop) => {
+    if (start) {
+      this.rsmq.start()
+    }
+
+    if (stop) {
+      this.rsmq.stop()
+    }
+  })
+}
+
+module.exports = Broker

lib/loader.js

@@ -1,7 +1,10 @@
 'use strict'
 
-const conf = require('../config')
-const logger = require('@dadi/logger')
-logger.init(conf.get('logging'), null, conf.get('env'))
+/**
+ * Logger
+ */
+var config = require('../config')
+var logger = require('@dadi/logger')
+logger.init(config.get('logging'), null, config.get('env'))
 
 module.exports = logger

lib/logger.js

@@ -1,12 +1,15 @@
 'use strict'
 
+/**
+ * QueueHandler
+ */
 const logger = require('./logger')
 const Broker = require('./broker')
 const router = require('./router')()
 
 /**
- * Main App
- * Starts the message broker, delegates messages to
+ * @constructor
+ * @classdesc Starts the message broker, delegates messages to
  * workers via the router, and handles errors
  */
 var QueueHandler = function () {
@@ -22,6 +25,4 @@ QueueHandler.prototype.handle = function (err, req, done) {
   }
 }
 
-module.exports = function () {
-  return new QueueHandler()
-}
+module.exports = QueueHandler

lib/app.js lib/queue-handler.jslib/app.js renamed to lib/queue-handler.js

@@ -1,12 +1,17 @@
 'use strict'
 
-const logger = require('./logger')
+/**
+ * Router
+ */
+var config = require('../config')
+var logger = require('./logger')
+var Workers = require('./workers')
 
 /**
- * Message Router
+ * @constructor
  */
 var Router = function () {
-  this.workers = require('./loader')
+  this.workers = new Workers().load(config.get('workers.path'))
 }
 
 /**

lib/router.js

@@ -1,16 +1,23 @@
-/*
- * Message Throttle -
- * Stops the message broker when it hits max
+/**
+* Throttle
+*/
+
+/**
+ * @constructor
+ * @classdesc Stops the message broker when it hits max
  * Restarts when below max
- * Takes an 'engine' callback with (stop, start) params
+ * @param {number} max -
+ * @param {function} engine - a callback function accepting two Boolean arguments, `start` and `stop`
  */
-
 var Throttle = function (max, engine) {
   this.max = max
   this.engine = engine
   this.val = 0
 }
 
+/**
+ *
+ */
 Throttle.prototype.adjust = function (amt) {
   var adj = this.val + amt
 
@@ -24,10 +31,16 @@ Throttle.prototype.adjust = function (amt) {
   }
 }
 
+/**
+ *
+ */
 Throttle.prototype.less = function () {
   this.adjust(-1)
 }
 
+/**
+ *
+ */
 Throttle.prototype.more = function () {
   this.adjust(1)
 }

lib/throttle.js

@@ -0,0 +1,53 @@
+'use strict'
+
+/**
+ * Workers
+ */
+var fs = require('fs')
+var path = require('path')
+
+/**
+ * @constructor
+ * @classdesc Loads worker functions into an object literal
+ */
+var Workers = function () {
+
+}
+
+/**
+ * @param {string} root - xx
+ * @returns {Array} workers - xx
+ */
+Workers.prototype.load = function (root) {
+  this.workers = {}
+
+  // files must be read before directories so that deeper levels can be
+  // assigned as properties to the initial worker functions
+  fs.readdirSync(root).forEach(fileName => this.loadPath(root, fileName, 'file'))
+  fs.readdirSync(root).forEach(fileName => this.loadPath(root, fileName, 'dir'))
+
+  return this.workers
+}
+
+/**
+ * @param {string} root - xx
+ * @param {string} fileName - xx
+ * @param {string} type - xx
+ */
+Workers.prototype.loadPath = function (root, fileName, type) {
+  var filePath = path.join(root, fileName)
+  var isDir = fs.lstatSync(filePath).isDirectory()
+  var basename = path.basename(filePath, '.js')
+
+  if (type === 'file' && path.extname(filePath) === '.js') {
+    this.workers[basename] = require(path.resolve(filePath))
+  }
+
+  if (type === 'dir' && isDir) { // recurse
+    this.workers[basename] = Object.assign(this.workers[basename] || {}, this.load(filePath))
+  }
+}
+
+module.exports = function () {
+  return new Workers()
+}

lib/workers.js

@@ -4,7 +4,8 @@
   "description": "A lightweight, high-performance task queue built on Node JS and Redis. Includes a broker for queue management and routing, plus a worker framework for processing messages.",
   "main": "index.js",
   "scripts": {
-    "test": "standard 'lib/*.js' && env NODE_ENV=test ./node_modules/.bin/istanbul cover --report cobertura --report text --report html --report lcov ./node_modules/mocha/bin/_mocha",
+    "docs": "jsdoc -c ./docs/conf.json -t ./node_modules/ink-docstrap/template -R README.md -r lib -d docs",
+    "test": "snazzy 'lib/*.js' && env NODE_ENV=test ./node_modules/.bin/istanbul cover --report cobertura --report text --report html --report lcov ./node_modules/mocha/bin/_mocha && npm run docs",
     "posttest": "./scripts/coverage.js",
     "start": "node ./index.js --node-env=development"
   },
@@ -17,12 +18,14 @@
     "babel-register": "^6.18.0",
     "env-test": "^1.0.0",
     "fakeredis": "^1.0.3",
+    "ink-docstrap": "^1.3.0",
     "istanbul": "^1.1.0-alpha.1",
     "istanbul-cobertura-badger": "^1.2.1",
     "mocha": "^3.2.0",
     "redis": "^2.6.4",
     "should": "^11.1.2",
     "sinon": "^1.17.7",
+    "snazzy": "^5.0.0",
     "standard": "^8.6.0",
     "supertest": "^2.0.1"
   },