Merge pull request #60 from Santinell/features

new features: addQueue and universal worker

lib/connection.js

@@ -12,18 +12,23 @@ function Connection(uri, options) {
 Connection.prototype.worker = function (queues, options) {
     var self = this;
 
-    if (!Array.isArray(queues)) {
-      queues = [queues];
-    }
-
-    var queues = queues.map(function (queue) {
-        if (typeof queue === 'string') {
-            queue = self.queue(queue);
+    if (queues === "*") {
+        var opts = {universal: true, collection: options.collection || 'jobs' };
+        options.universal = true;
+        queues = [new Queue('*', opts)];
+    } else {
+        if (!Array.isArray(queues)) {
+            queues = [queues];
         }
 
-        return queue;
-    });
+        var queues = queues.map(function (queue) {
+            if (typeof queue === 'string') {
+                queue = self.queue(queue);
+            }
 
+            return queue;
+        });
+    }
     return new Worker(queues, options);
 };
 

lib/queue.js

@@ -12,6 +12,7 @@ function Queue(connection, name, options) {
 
     options || (options = {});
     options.collection || (options.collection = 'jobs');
+    options.universal || (options.universal = false);
 
     this.connection = connection;
     this.name = name || 'default';
@@ -35,7 +36,12 @@ Queue.prototype.get = function (id, callback) {
         id = new mongoskin.helper.toObjectID(id);
     }
 
-    this.collection.findOne({ _id: id, queue: this.name }, function (err, data) {
+    var query = { _id: id };
+    if (!this.options.universal) {
+        query.queue = this.name;
+    }
+
+    this.collection.findOne(query, function (err, data) {
         if (err) return callback(err);
 
         var job = new Job(self.collection, data);
@@ -72,14 +78,17 @@ Queue.prototype.dequeue = function (options, callback) {
 
     var query = {
         status: Job.QUEUED,
-        queue: this.name,
         delay: { $lte: new Date() }
     };
 
+    if (!this.options.universal) {
+        query.queue = this.name;
+    }
+
     if (options.minPriority !== undefined) {
         query.priority = { $gte: options.minPriority };
     }
-    
+
     if (options.callbacks !== undefined) {
         var callback_names = Object.keys(options.callbacks);
         query.name = { $in: callback_names };

lib/worker.js

@@ -6,14 +6,14 @@ module.exports = Worker;
 
 function Worker(queues, options) {
     options || (options = {});
-    options.queues || (options.queues = []);
 
     this.empty = 0;
-    this.queues = queues;
+    this.queues = queues || [];
     this.interval = options.interval || 5000;
 
     this.callbacks = options.callbacks || {};
     this.strategies = options.strategies || {};
+    this.universal = options.universal || false;
 
     // Default retry strategies
     this.strategies.linear || (this.strategies.linear = linear);
@@ -22,9 +22,6 @@ function Worker(queues, options) {
     // This worker will only process jobs of this priority or higher
     this.minPriority = options.minPriority;
 
-    if (this.queues.length === 0) {
-        throw new Error('Worker must have at least one queue.');
-    }
 }
 
 util.inherits(Worker, events.EventEmitter);
@@ -42,6 +39,9 @@ Worker.prototype.strategies = function (strategies) {
 };
 
 Worker.prototype.start = function () {
+    if (this.queues.length === 0) {
+        return setTimeout(this.start.bind(this), this.interval);
+    }
     this.working = true;
     this.poll();
 };
@@ -65,6 +65,11 @@ Worker.prototype.stop = function (callback) {
     this.once('stopped', done);
 };
 
+Worker.prototype.addQueue = function (queue) {
+  if (!this.universal)
+    this.queues.push(queue);
+};
+
 Worker.prototype.poll = function () {
     if (!this.working) {
         return this.emit('stopped');

package.json

@@ -17,8 +17,8 @@
     "mongoskin": "^2.0.0"
   },
   "devDependencies": {
-    "async": "0.2.10",
-    "mocha": "1.18.2",
-    "sinon": "1.9.0"
+    "async": "^1.5.0",
+    "mocha": "^2.3.4",
+    "sinon": "^1.17.2"
   }
 }