Initial commit: migration from private repo

Author: rcrichton

.gitignore

@@ -0,0 +1,5 @@
+*.log
+node_modules/
+queue/
+working/
+error/

.jshintrc

@@ -0,0 +1,15 @@
+{
+  "node": true,
+
+  "camelcase": true,
+  "curly": true,
+  "eqeqeq": true,
+  "expr": true,
+  "indent": 2,
+  "latedef": true,
+  "newcap": false,
+  "quotmark": "single",
+  "strict": true,
+  "undef": true,
+  "unused": true
+}

.tern-project

@@ -0,0 +1,11 @@
+{
+  "libs": [
+    "ecma5"
+  ],
+  "loadEagerly": [
+    "**.js"
+  ],
+  "plugins": {
+    "node": {}
+  }
+}

LICENSE.txt

@@ -0,0 +1,25 @@
+# File queue
+
+To install, execute `npm install -g openhim-mediator-file-queue`
+
+Once installed you will have to manually edit the config files in the location where global npm modules are installed.
+
+* With NVM this is usually: `~/.nvm/versions/node/v4.1.0/lib/node_modules/openhim-mediator-file-queue/config`
+* With a standard node install this is usually: `/usr/local/lib/node_modules/openhim-mediator-file-queue/config`
+
+You may add a `production.json` file to the config folder to override or add any config values.
+
+## How it works
+
+The file queue simply handles incoming requests, writing the files to a directory on the filesystem, and then processes the queue, sending the files to a configured endpoint. If the file is successfully sent then the queued file is deleted from the filesystem, otherwise it is moved to an error directory.
+
+Multiple "endpoints" can be configured. Each endpoint handles incoming requests for a specific URL, queues them, and then sends them to another configured URL. An endpoint has a "worker" which is responsible for reading the files from the queue and processing them. Workers can process multiple files in parallel as configured (by default 2 at a time). Workers can be paused/unpaused or repopulated via a RESTlike endpoint. Pausing a worker will stop it from processing files from the queue, but the endpoint will continue accepting requests and writing the files to the queue. Repopulating a worker will cause it to refresh its queue from the files on the filesystem. This is useful when manually adding files to or removing files from the queue.
+
+## Endpoint config
+
+An array of endpoints should be configured in the config file. An endpoint can have the following properties:
+* `name` (required) - The name of the endpoint which is used for setting up the RESTlike routes for the worker.
+* `path` (required) - The path to use for handling incoming requests.
+* `url` (required) - The URL to send the files to when processing them from the queue.
+* `paused` - Whether or not the endpoint's worker should be paused by default. This must be a boolean value.
+* `parallel` - The number of files that the worker should process in parallel. Defaults to 2.

README.md

@@ -0,0 +1,32 @@
+{
+  "port": 4003,
+  "endpoints": [],
+  "log_level": "info",
+  "statsd": {
+    "host": "localhost",
+    "port": 8125
+  },
+  "api": {
+    "apiURL": "https://localhost:8080",
+    "username": "[email protected]",
+    "password": "password"
+  },
+  "mediatorConf": {
+    "urn": "urn:uuid:a15c3d48-0686-4c9b-b375-f68d2f244a33",
+    "version": "1.1.1",
+    "name": "file-queue",
+    "description": "Async file queue mediator for the OpenHIM",
+    "defaultChannelConfig": [],
+    "endpoints": [
+      {
+        "name": "File queue",
+        "host": "localhost",
+        "path": "/test",
+        "port": "4002",
+        "primary": true,
+        "type": "http"
+      }
+    ],
+    "configDefs": []
+  }
+}

config/config.json

@@ -0,0 +1,12 @@
+{
+  "port": 4002,
+  "log_level": "debug",
+  "endpoints": [
+    {
+      "name": "test",
+      "path": "/test",
+      "url": "http://localhost:9999/test",
+      "updateTx": true
+    }
+  ]
+}

config/development.json

@@ -0,0 +1,164 @@
+#!/usr/bin/env node
+'use strict';
+
+var BodyParser = require('body-parser');
+var Confit = require('confit');
+var Crypto = require('crypto');
+var Express = require('express');
+var FS = require('graceful-fs');
+var MUtils = require('openhim-mediator-utils');
+var OnFinished = require('on-finished');
+var Path = require('path');
+var Stats = require('./stats');
+var Type = require('type-is');
+var Utils = require('./utils');
+var Worker = require('./worker');
+var Winston = require('winston');
+
+var app = Express();
+
+// Adds an extension to the filename based on the content-type request header
+function addExt(filename, req) {
+  switch (Type(req, ['json', 'xml'])) {
+    case 'json':
+      return filename + '.json';
+    case 'xml':
+      return filename + '.xml';
+    default:
+      return filename + '.txt';
+  }
+}
+
+// Set up an endpoint based on the config
+function setUpEndpoint(endpoint, apiOpts) {
+  var updateTx;
+  if (endpoint.updateTx && endpoint.updateTx === true) {
+    updateTx = true;
+  } else {
+    updateTx = false;
+  }
+  var worker = new Worker({
+    name: endpoint.name,
+    url: endpoint.url,
+    paused: endpoint.paused,
+    parallel: endpoint.parallel,
+    updateTx: updateTx,
+    apiOpts: apiOpts
+  });
+
+  // Clear the worker's queue and repopulate it
+  app.post('/workers/' + worker.name + '/repopulate', function(req, res) {
+    worker.repopulate();
+    res.status(200).send('Worker repopulated');
+  });
+
+  // Register an endpoint for pausing/resuming the worker
+  app.put('/workers/' + worker.name, BodyParser.json(), function(req, res) {
+    if (typeof req.body.paused !== 'boolean') {
+      return res.status(400).send('Missing or invalid property: paused');
+    }
+    if (req.body.paused) {
+      worker.pause();
+      res.status(200).send('Worker paused');
+    } else {
+      worker.resume();
+      res.status(200).send('Worker resumed');
+    }
+  });
+  Winston.info('Worker for endpoint %s available at /workers/%s', endpoint.path, worker.name);
+
+  // Register an endpoint for handling requests
+  app.post(endpoint.path, function(req, res, next) {
+    Winston.info('Handling request for %s', endpoint.path);
+
+    var filename;
+    if (req.headers['x-openhim-transactionid']) {
+      // set file name to transaction ID
+      filename = req.headers['x-openhim-transactionid'];
+    } else {
+      // Generate an invalid transaction ID
+      filename = Crypto.randomBytes(12).toString('hex').replace(/./, 'x');
+    }
+    filename = addExt(filename, req);
+
+    var filePath = Path.join(worker.queuePath, filename);
+    var stream = req.pipe(FS.createWriteStream(filePath));
+    stream.on('error', function(err) {
+      Stats.increment('errors');
+      Winston.error('Handling request for %s failed', endpoint.path);
+      return next(err);
+    });
+    stream.on('finish', function() {
+      Winston.info('File saved to ./%s', Path.relative(process.cwd(), filePath));
+      worker.addToQueue(filename, function(err) {
+        if (err) {
+          Stats.increment('errors');
+          Winston.error(err, {path: filename});
+        }
+      });
+
+      var mediatorResponse = {
+        'x-mediator-urn': apiOpts.urn,
+        status: 'Processing',
+        response: {
+          status: 202,
+          body: 'Request added to queue',
+          timestamp: new Date().toString()
+        }
+      };
+      res.status(202).type('application/json+openhim').send(mediatorResponse);
+    });
+  });
+}
+
+app.get('/heartbeat', function(req, res) {
+  res.send({
+    uptime: process.uptime()
+  });
+});
+
+app.use(function(req, res, next) {
+  var start = process.hrtime();
+  Stats.increment('requests');
+  OnFinished(res, function(err, res) {
+    Stats.increment('response_codes.' + res.statusCode);
+    Stats.timing('response_time', Utils.millisecondsSince(start));
+  });
+  next();
+});
+
+// Create the config and start up the app
+Confit(Path.join(__dirname, '..', 'config')).create(function(err, config) {
+  if (err) {
+    throw err;
+  }
+
+  Object.defineProperty(app.locals, 'config', {value: config});
+
+  Winston.clear();
+  Winston.add(Winston.transports.Console, {timestamp: true, level: config.get('log_level')});
+
+  Stats.init(config.get('statsd'));
+
+  var apiOpts = config.get('api');
+  var mediatorConf = config.get('mediatorConf');
+  apiOpts.urn = mediatorConf.urn;
+
+  MUtils.registerMediator(apiOpts, mediatorConf, function(err) {
+    if (err) {
+      Winston.error('Could not register mediator');
+      process.exit(1);
+    }
+
+    MUtils.activateHeartbeat(apiOpts);
+
+    config.get('endpoints').forEach(function(endpoint) {
+      setUpEndpoint(endpoint, apiOpts);
+    });
+
+    var port = process.env.PORT || config.get('port');
+    app.listen(port, function() {
+      Winston.info('App started on port %s', port);
+    });
+  });
+});

lib/index.js

@@ -0,0 +1,52 @@
+'use strict';
+
+var OS = require('os');
+var StatsD = require('node-statsd');
+
+var stats = null;
+
+module.exports = {
+  init: function(options) {
+    stats = new StatsD({
+      host: options.host,
+      port: options.port,
+      prefix: OS.hostname() + '.momconnect_queue.'
+    });
+  },
+
+  /**
+   * Gauges a stat by a specified amount
+   * @param stat {String|Array} The stat(s) to send
+   * @param value The value to send
+   * @param sampleRate {Number=} The Number of times to sample (0 to 1). Optional.
+   * @param tags {Array=} The Array of tags to add to metrics. Optional.
+   * @param callback {Function=} Callback when message is done being delivered. Optional.
+   */
+  gauge: function() {
+    stats && stats.gauge.apply(stats, arguments);
+  },
+
+  /**
+   * Increments a stat by a specified amount
+   * @param stat {String|Array} The stat(s) to send
+   * @param value The value to send
+   * @param sampleRate {Number=} The Number of times to sample (0 to 1). Optional.
+   * @param tags {Array=} The Array of tags to add to metrics. Optional.
+   * @param callback {Function=} Callback when message is done being delivered. Optional.
+   */
+  increment: function() {
+    stats && stats.increment.apply(stats, arguments);
+  },
+
+  /**
+   * Represents the timing stat
+   * @param stat {String|Array} The stat(s) to send
+   * @param time {Number} The time in milliseconds to send
+   * @param sampleRate {Number=} The Number of times to sample (0 to 1). Optional.
+   * @param tags {Array=} The Array of tags to add to metrics. Optional.
+   * @param callback {Function=} Callback when message is done being delivered. Optional.
+   */
+  timing: function() {
+    stats && stats.timing.apply(stats, arguments);
+  }
+};

lib/stats.js

@@ -0,0 +1,7 @@
+'use strict';
+
+// Return the number of milliseconds since a start time
+exports.millisecondsSince = function millisecondsSince(start) {
+  var diff = process.hrtime(start);
+  return diff[0] * 1e3 + diff[1] / 1e6;
+};