Added lock-support for jobs in the local backend (only run a single instance of a job at the same time)

Reordered timeout setup to allow for proper cleanup when a job cannot run due to a lock
Added missing parameter to acquireMaster on local backend

Author: Jesper Svennevid

lib/backends/local.js

@@ -6,6 +6,7 @@ function ProducerBackend(frontend) {
     this.pending = {};
     this.active = {};
     this.queue = [];
+    this.locks = {};
     this.isMaster = false;
 }
 
@@ -15,6 +16,9 @@ _.extend(ProducerBackend.prototype, {
             var message = this.active[worker.id];
             delete this.active[worker.id];
             if (message) {
+                if ("lock" in message.payload.options) {
+                    delete this.locks[message.payload.options.lock];
+                }
                 delete this.pending[message.payload.index];
                 this.frontend.onResponse(message, response);
             }
@@ -36,6 +40,12 @@ _.extend(ProducerBackend.prototype, {
 
         var message = this.queue[0];
 
+        if (("lock" in message.payload.options) && this.locks[message.payload.options.lock]) {
+            this.queue.shift();
+            this.frontend.onResponse(message, { error: "locked" });
+            return;
+        }
+
         for (var id in cluster.workers) {
             if (!this.active[id]) {
                 cluster.workers[id].send(message.payload);
@@ -52,16 +62,24 @@ _.extend(ProducerBackend.prototype, {
 
         this.queue.shift();
         this.pending[message.payload.index] = message;
+        if ("lock" in message.payload.options) {
+            this.locks[message.payload.options.lock] = message;
+        }
     },
 
     onTimeout: function (id) {
         var pending = this.pending[id];
         if (pending) {
             delete this.pending[id];
+
             if (pending.worker) {
                 delete this.active[pending.worker];
             }
 
+            if ("lock" in pending.payload.options) {
+                delete this.locks[pending.payload.options.lock];
+            }
+
             var worker = cluster.workers[pending.worker];
             if (worker) {
                 worker.kill();
@@ -74,7 +92,7 @@ _.extend(ProducerBackend.prototype, {
         this.processQueue();
     },
 
-    acquireMaster: function (callback) {
+    acquireMaster: function (timeout, callback) {
         this.isMaster = true;
         callback(true);
     },

lib/workqueue.js

@@ -64,23 +64,22 @@ _.extend(Producer.prototype, {
         var message = {
             payload: {
                 type: type,
-                options: options,
+                options: _.clone(options),
                 index: this.index++
             },
             callback: callback
         };
 
-        this.backend.queueMessage(message);
-
         message.timeout = setTimeout(_.bind(function () {
             this.backend.onTimeout(message.payload.index);
             delete message.payload.timeout;
             this.onResponse(message, {
-                error: "job timeout",
+                error: "timeout",
                 index: message.payload.index
             });
         }, this), this.options.timeout);
 
+        this.backend.queueMessage(message);
         return message.payload.index;
     },