Merge pull request #40 from AnyFetch/tasks-on-redis

Neamar · Neamar · commit 1fe185c63d84 · 2014-11-26T09:53:37.000+01:00
Tasks on redis
diff --git a/.jshintrc b/.jshintrc
@@ -66,6 +66,7 @@
     "describe",
     "it",
     "after",
+    "afterEach",
     "before",
     "beforeEach"
   ],
diff --git a/.travis.yml b/.travis.yml
@@ -1,6 +1,8 @@
 language: node_js
 node_js:
   - '0.10'
+services:
+  - redis-server
 before_script:
   - npm install -g istanbul
   - npm install coveralls
diff --git a/lib/handlers/hydrater.js b/lib/handlers/hydrater.js
@@ -7,7 +7,6 @@
  */
 
 var restify = require('restify');
-
 /**
  * This handler receives a document to hydrate on a POST request and processes it.
  *
@@ -16,28 +15,36 @@ var restify = require('restify');
  * @param {Object} server Current server. See implementation in index.js
  * @param {Function} next Callback to call once res has been populated.
  */
-module.exports = function(req, res, server, logger, next) {
-  if(!req.params.callback && !req.params.long_poll) {
+module.exports = function(req, res, server, logger, errLogger, next) {
+  if(!req.params.callback) {
     return next(new restify.MissingParameterError('No specified callback'));
   }
 
+
   res.send(202);
   next();
 
   // Prepare the new task
+
   var task = {};
   task.file_path = (req.params.file_path) ? req.params.file_path : null;
   task.callback = req.params.callback;
 
-
   task.document = req.params.document || {};
   task.document.metadata = task.document ? task.document.metadata || {} : {};
   task.document.data = task.document ? task.document.data || {} : {};
 
   task.priority = (req.params.priority) ? parseInt(req.params.priority) : 0;
 
   // Push it to the queue
-  server.queue.push(task, -task.priority);
+  var job = server.queue.createJob(task, {priority: task.priority});
+  job.save(function(err) {
+    if(err) {
+      errLogger("Error while queuing: " + ((task.file_path) ? task.file_path : task.document.id) + "\nError :" + err.toString());
+    }
+    else {
+      logger("Queuing: " + ((task.file_path) ? task.file_path : task.document.id));
+    }
+  });
 
-  logger("Queuing: " + ((task.file_path) ? task.file_path : task.document.id));
 };
diff --git a/lib/handlers/status.js b/lib/handlers/status.js
@@ -13,11 +13,14 @@
  * @param {Function} next Callback to call once res has been populated.
  */
 module.exports = function(req, res, server, logger, next) {
-  var ret = {
-    'status': 'ok',
-    'queued_items': server.queue.length()
-  };
-
-  res.send(ret);
-  next();
+  server.queue.stats(function(err, stats) {
+    if(err) {
+      return next(err);
+    }
+    else {
+      stats.status = 'ok';
+      res.send(stats);
+      next();
+    }
+  });
 };
diff --git a/lib/helpers/Childs.js b/lib/helpers/Childs.js
@@ -55,6 +55,12 @@ var Childs = function(concurrency, tasksPerProcess) {
       return child;
     }
   };
+
+  this.stopAllChilds = function()  {
+    this.childs.forEach(function(child) {
+      child.process.kill('SIGTERM');
+    });
+  };
 };
 
 
diff --git a/lib/helpers/child-process.js b/lib/helpers/child-process.js
@@ -1,6 +1,6 @@
 "use strict";
 /*
- * A new process is spawned with this file for each task.
+ * A new process is spawned with this file for some tasks.
  * Inter process comunication is used to send and receive message from the parent process regarding the task.
  * This let us finely control the behavior of the hydration function.
  * If it happens to get stuck in a subprocess or in an HTTP blackhole,
diff --git a/lib/helpers/hydrater.js b/lib/helpers/hydrater.js
@@ -14,22 +14,16 @@ var rarity = require('rarity');
 var util = require('util');
 
 var lib = require('../index.js');
-var Childs = require('./Childs');
 
 var HydrationError = lib.HydrationError;
 
 
 
-module.exports = function(hydraterFunction, concurrency, logger, errLogger) {
+module.exports = function(hydraterFunction, childs, logger, errLogger) {
   if(!errLogger) {
     errLogger = logger;
   }
 
-  var tasksPerProcess = process.env.TASKS_PER_PROCESS || 100;
-
-  var childs = new Childs(concurrency, tasksPerProcess);
-
-
   /**
    * Handle a hydration task:
    * - Download the file
@@ -40,7 +34,8 @@ module.exports = function(hydraterFunction, concurrency, logger, errLogger) {
    * @param {Object} task Task object, keys must be `file_path` (file URL) and `callback` (URL)
    * @param {Function} done(err)
    */
-  return function(task, done) {
+  return function(job, done) {
+    var task = job.data;
     logger("Starting task: " + ((task.file_path) ? task.file_path : task.document.id));
 
     async.waterfall([
@@ -222,7 +217,6 @@ module.exports = function(hydraterFunction, concurrency, logger, errLogger) {
         if(internalErr) {
           errLogger("INTERNAL ERR", internalErr);
         }
-
         done(err || internalErr, changes);
       });
     });
diff --git a/lib/index.js b/lib/index.js
@@ -2,7 +2,8 @@
 
 var util = require('util');
 var restify = require('restify');
-var async = require('async');
+var yaqs = require('yaqs');
+var Childs = require('./helpers/Childs');
 
 /**
  * Create a new hydration server.
@@ -23,24 +24,49 @@ module.exports.createServer = function(config) {
   config.errLogger = config.errLogger || console.error;
 
   var concurrency = config.concurrency || 1;
+  var tasksPerProcess = config.tasksPerProcess || 100;
 
   // Load configuration and initialize server
   var hydraterEndpoint = require('./handlers/hydrater.js');
   var statusEndpoint = require('./handlers/status.js');
-  var hydraterHelper = require('./helpers/hydrater.js')(config.hydrater_function, concurrency, config.logger, config.errLogger);
-  var server = restify.createServer();
 
+  var childs = new Childs(concurrency, tasksPerProcess);
+
+  var hydraterHelper = require('./helpers/hydrater.js')(config.hydrater_function, childs, config.logger, config.errLogger);
+
+  var server = restify.createServer();
   // Middleware Goes Here
   server.use(restify.acceptParser(server.acceptable));
   server.use(restify.queryParser());
   server.use(restify.bodyParser());
   server.use(require('./middlewares/logger.js'));
 
-  server.queue = async.priorityQueue(hydraterHelper, concurrency);
+  server.yaqsClient = yaqs({
+    prefix: config.hydraterUrl,
+    redis: config.redisUrl,
+    defaultConcurrency: concurrency
+  });
+
+  server.queue = server.yaqsClient.createQueue('hydration', {});
+  server.queue.setWorker(hydraterHelper).start();
+
+
+  function sigtermYaqs() {
+    server.yaqsClient.stopAllQueues(function(err) {
+      if(err) {
+        config.errLogger(err);
+      }
+      childs.stopAllChilds();
+      config.logger('YAQS has stopped.');
+      process.exit(0);
+    });
+  }
+
+  process.once('SIGTERM', sigtermYaqs);
 
   // Load routes
   server.post('/hydrate', function(req, res, next) {
-    hydraterEndpoint(req, res, server, config.logger, next);
+    hydraterEndpoint(req, res, server, config.logger, config.errLogger, next);
   });
   server.get('/status', function(req, res, next) {
     statusEndpoint(req, res, server, config.logger, next);
diff --git a/package.json b/package.json
@@ -19,9 +19,11 @@
   "dependencies": {
     "async": "^0.9.0",
     "rarity": "^2.1.1",
+    "redis": "^0.12.1",
     "restify": "^2.8.3",
     "restify-logger": "^2.0.1",
-    "supertest": "^0.14.0"
+    "supertest": "^0.14.0",
+    "yaqs": "^1.0.3"
   },
   "devDependencies": {
     "should": "^4.1.0",
diff --git a/test/cleaning.js b/test/cleaning.js
@@ -4,6 +4,8 @@ var path = require("path");
 var async = require("async");
 var rarity = require("rarity");
 var shellExec = require('child_process').exec;
+
+var Childs = require('../lib/helpers/Childs');
 var createFakeApi = require('./helpers/fake-api.js');
 
 var concurrencies = [1, 2];
@@ -14,17 +16,24 @@ concurrencies.forEach(function(concurrency) {
   tasksPerProcess.forEach(function(_tasksPerProcess) {
     describe('Hydration should be cleaned every time with concurrency = ' + concurrency + ' & tasksPerProcess = ' + _tasksPerProcess , function() {
       var fakeApi = createFakeApi();
+      var childs;
 
       fakeApi.patch('/result', function(req, res, next) {
         res.send(204);
         next();
       });
+
       before(function() {
         fakeApi.listen(4243);
       });
 
-      after(function() {
-        fakeApi.close();
+      afterEach(function(done) {
+        childs.stopAllChilds();
+        done();
+      });
+
+      after(function(done) {
+        fakeApi.close(done);
       });
 
       it('on normal workflow', function(done) {
@@ -35,11 +44,12 @@ concurrencies.forEach(function(concurrency) {
           concurrency: concurrency,
           logger: function() {},
         };
+        childs = new Childs(config.concurrency, _tasksPerProcess);
 
-        process.env.TASKS_PER_PROCESS = _tasksPerProcess;
-        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, config.concurrency, config.logger);
+        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, childs, config.logger);
 
-        var task = {
+        var task = {};
+        task.data = {
           file_path: "http://127.0.0.1:4243/afile",
           callback: "http://127.0.0.1:4243/result",
           document: {
@@ -82,11 +92,12 @@ concurrencies.forEach(function(concurrency) {
           concurrency: concurrency,
           logger: function() {},
         };
-        process.env.TASKS_PER_PROCESS = _tasksPerProcess;
+        var childs = new Childs(config.concurrency, _tasksPerProcess);
 
-        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, config.concurrency, config.logger);
+        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, childs, config.logger);
 
-        var task = {
+        var task = {};
+        task.data = {
           file_path: "http://127.0.0.1:4243/afile",
           callback: "http://127.0.0.1:4243/result",
           document: {
@@ -129,11 +140,12 @@ concurrencies.forEach(function(concurrency) {
           concurrency: concurrency,
           logger: function() {},
         };
-        process.env.TASKS_PER_PROCESS = _tasksPerProcess;
+        var childs = new Childs(config.concurrency, _tasksPerProcess);
 
-        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, config.concurrency, config.logger);
+        var hydrate = require('../lib/helpers/hydrater.js')(config.hydrater_function, childs, config.logger);
 
-        var task = {
+        var task = {};
+        task.data = {
           file_path: "http://127.0.0.1:4243/afile",
           callback: "http://127.0.0.1:4243/result",
           document: {
diff --git a/test/errors.js b/test/errors.js
@@ -5,16 +5,24 @@ var request = require('supertest');
 var anyfetchFileHydrater = require('../lib/');
 
 describe('Errors', function() {
-  var config = {
-    hydrater_function: __dirname + '/hydraters/buggy-hydrater.js',
-    concurrency: 1,
-    logger: function() {// Will be pinged with error. We don't care.
-    },
-    errLogger: function() {// Will be pinged with error. We don't care.
-    }
-  };
-  var hydrationServer = anyfetchFileHydrater.createServer(config);
+  var hydrationServer;
 
+  before(function() {
+    var config = {
+      hydraterUrl: "test-hydrater",
+      hydrater_function: __dirname + '/hydraters/buggy-hydrater.js',
+      concurrency: 1,
+      logger: function() {}, // Will be pinged with error. We don't care.
+      errLogger: function() {} // Will be pinged with error. We don't care.
+
+    };
+
+    hydrationServer = anyfetchFileHydrater.createServer(config);
+  });
+
+  after(function(done) {
+    hydrationServer.queue.remove(done);
+  });
 
   describe('in lib', function() {
     var fakeApi = require('./helpers/fake-api.js')();
@@ -42,16 +50,14 @@ describe('Errors', function() {
             }
           }
         })
-        .expect(204)
+        .expect(202)
         .end(function() {});
 
-      hydrationServer.queue.drain = function(err) {
-        hydrationServer.queue.drain = null;
-        done(err);
-      };
+      hydrationServer.queue.once('empty', function() {
+        done();
+      });
     });
 
-
     it('should be handled gracefully if file does not exists', function(done) {
       this.timeout(10000);
 
@@ -87,12 +93,21 @@ describe('Errors', function() {
   });
 
   describe('in hydrators', function() {
-    var config = {
-      hydrater_function: __dirname + '/hydraters/errored-hydrater.js',
-      logger: function() {// Will be pinged with error. We don't care.
-      }
-    };
-    var hydrationErrorServer = anyfetchFileHydrater.createServer(config);
+
+    var hydrationErrorServer;
+
+    before(function() {
+      var config = {
+        hydrater_function: __dirname + '/hydraters/errored-hydrater.js',
+        logger: function() {// Will be pinged with error. We don't care.
+        }
+      };
+      hydrationErrorServer = anyfetchFileHydrater.createServer(config);
+    });
+
+    after(function(done) {
+      hydrationErrorServer.queue.remove(done);
+    });
 
     it('should be handled gracefully while hydrating', function(done) {
       var fakeApi = require('./helpers/fake-api.js')();
diff --git a/test/helpers/hydrater.js b/test/helpers/hydrater.js
diff --git a/test/hook.js b/test/hook.js
diff --git a/test/index.js b/test/index.js
diff --git a/test/setup.js b/test/setup.js
