Merge pull request #361 from mixmaxhq/add-removeOnComplete

Add new 'removeOnComplete' option to atomically remove a job when it …

README.md

@@ -433,6 +433,8 @@ __Arguments__
                           If you use this option, it is up to you to ensure the
                           jobId is unique. If you attempt to add a job with an id that
                           already exists, it will not be added.
+    removeOnComplete {Boolean} A boolean which, if true, removes the job when it successfully
+                               completes. Default behavior is to keep the job in the completed queue.
   }
   returns {Promise} A promise that resolves when the job has been succesfully
     added to the queue (or rejects if some error occured). On success, the promise

lib/job.js

@@ -162,7 +162,7 @@ Job.prototype.delayIfNeeded = function(){
 
 Job.prototype.moveToCompleted = function(returnValue, token){
   this.returnvalue = returnValue || 0;
-  return scripts.moveToCompleted(this, token || 0);
+  return scripts.moveToCompleted(this, token || 0, this.opts.removeOnComplete);
 };
 
 Job.prototype.moveToFailed = function(err){

lib/scripts.js

@@ -134,7 +134,7 @@ var scripts = {
 
     return execScript.apply(scripts, args);
   },
-  moveToCompleted: function(job, token){
+  moveToCompleted: function(job, token, removeOnComplete){
     var keys = _.map([
       'active',
       'completed',
@@ -143,20 +143,23 @@ var scripts = {
         return job.queue.toKey(name);
       }
     );
-
     keys.push(job.lockKey());
 
-    //
-    // INVESTIGATE: Should'nt we check if we have the lock before trying to move the
-    // job?
-    //
+    var deleteJob = 'redis.call("DEL", KEYS[3])';
+    var moveJob = [
+      'redis.call("SADD", KEYS[2], ARGV[1])',
+      'redis.call("HSET", KEYS[3], "returnvalue", ARGV[2])',
+    ].join('\n');
+
     var script = [
-      'if redis.call("EXISTS", KEYS[3]) == 1 then',
-      ' redis.call("SADD", KEYS[2], ARGV[1])',
-      ' redis.call("HSET", KEYS[3], "returnvalue", ARGV[2])',
-      ' redis.call("LREM", KEYS[1], 0, ARGV[1])',
-      ' if redis.call("get", KEYS[4]) == ARGV[3] then', // Remove the lock
-      '  return redis.call("del", KEYS[4])',
+      'if redis.call("EXISTS", KEYS[3]) == 1 then', // Make sure job exists
+      ' if redis.call("GET", KEYS[4]) == ARGV[3] then', // Makes sure we own the lock
+      '  redis.call("LREM", KEYS[1], -1, ARGV[1])',
+      removeOnComplete ? deleteJob : moveJob,
+      '  redis.call("DEL", KEYS[4])',
+      '  return 0',
+      ' else',
+      '  return -1',
       ' end',
       'else',
       ' return -1',
@@ -165,7 +168,7 @@ var scripts = {
 
     var args = [
       job.queue.client,
-      'moveToCompletedSet',
+      'moveToCompletedSet' + (removeOnComplete ? 'RemoveOnComplete' : ''),
       script,
       keys.length,
       keys[0],