Added lock-support for jobs using the redis backend

Author: Jesper Svennevid

lib/backends/redis.js

@@ -16,7 +16,7 @@ var redis_lock = "" +
     "end";
 
 var redis_unlock = "" +
-    "if redis.call('get', KEYS[1]) == ARGV[1] then\n" +
+    "if redis.call('get', KEYS[1]) == ARGV[1]\n" +
     "then\n" +
     " return redis.call('del', KEYS[1])\n" +
     "else\n" +
@@ -103,23 +103,50 @@ _.extend(ProducerBackend.prototype, {
             return;
         }
 
-        var message = this.queue.shift();
-        this.pending[message.payload.index] = message;
-        message.payload.sender = this.producerId;
+        var run = _.bind(function () {
+            this.pending[message.payload.index] = message;
+            message.payload.sender = this.producerId;
+
+            var jobQueue = this.options.prefix + "jobs";
+            this.producerClient.multi().rpush(jobQueue, JSON.stringify(message.payload), _.bind(function (err) {
+                if (err) {
+                    if (message.payload.index) {
+                        delete this.pending[message.payload.index];
+                        this.queue.unshift(message);
+                    }
+                    delete message.payload.sender;
+
+                    process.nextTick(_.bind(function () {
+                        this.processQueue();
+                    }, this));
+                } else {
+                    this.processQueue();
+                }
+            }, this)).expire(jobQueue, 20).exec();
+        }, this);
 
-        var jobQueue = this.options.prefix + "jobs";
-        this.producerClient.multi().rpush(jobQueue, JSON.stringify(message.payload), _.bind(function (err) {
-            if (err) {
-                if (message.payload.index) {
-                    delete this.pending[message.payload.index];
+        var message = this.queue.shift();
+        if ("lock" in message.payload.options) {
+            var lockName = this.options.prefix + "locks." + message.payload.options.lock;
+            var lockData = this.producerId + '.' + message.payload.index;
+            this.producerClient.eval(redis_lock, 1, lockName, lockData, Math.floor(message.lock_timeout * 2 / 1000), _.bind(function (err, response) {
+                if (err) {
                     this.queue.unshift(message);
+                    return;
                 }
-                delete message.payload.sender;
 
-            } else {
-                this.processQueue();
-            }
-        }, this)).expire(jobQueue, 20).exec();
+                if (response > 0) {
+                    run();
+                } else {
+                    this.frontend.onResponse(message, { error: "locked" });
+                    process.nextTick(_.bind(function () {
+                        this.processQueue();
+                    }, this));
+                }
+            }, this));
+        } else {
+            run();
+        }
     },
 
     scheduleResponse: function () {
@@ -133,9 +160,21 @@ _.extend(ProducerBackend.prototype, {
                 var response = JSON.parse(data[1]);
                 var message = this.pending[response.index];
                 if (message) {
-                    delete this.pending[response.index];
-                    this.frontend.onResponse(message, response);
-                    delete message.payload.index;
+                    var complete = _.bind(function () {
+                        delete this.pending[response.index];
+                        this.frontend.onResponse(message, response);
+                        delete message.payload.index;
+                    }, this);
+
+                    if ("lock" in message.payload.options) {
+                        var lockName = this.options.prefix + "locks." + message.payload.options.lock;
+                        var lockData = this.producerId + '.' + message.payload.index;
+                        this.producerClient.eval(redis_unlock, 1, lockName, lockData, function (err) {
+                            complete();
+                        });
+                    } else {
+                        complete();
+                    }
                 }
             }
 

lib/workqueue.js

@@ -70,6 +70,10 @@ _.extend(Producer.prototype, {
             callback: callback
         };
 
+        if ("lock" in message.payload.options) {
+            message.lock_timeout = this.options.timeout;
+        }
+
         message.timeout = setTimeout(_.bind(function () {
             this.backend.onTimeout(message.payload.index);
             delete message.payload.timeout;