first merge

Author: behrad

lib/queue/job.js

@@ -88,11 +88,12 @@ function get( fn, order, jobType) {
       , jobs    = {};
     if( !pending ) return fn(null, ids);
     ids.forEach(function( id ) {
+      id = redis.client().stripFIFO(id); // turn zid back to regular job id
       exports.get(id, jobType, function( err, job ) {
         if( err ) {
           console.error(err);
         } else {
-          jobs[ job.id ] = job;
+          jobs[ redis.client().createFIFO(job.id) ] = job;
         }
         --pending || fn(null, 'desc' == order
           ? map(jobs, ids).reverse()
@@ -169,6 +170,7 @@ exports.get = function( id, jobType, fn ) {
     , job    = new Job;
 
   job.id = id;
+  job.zid = client.createFIFO(id);
   client.hgetall(client.getKey('job:' + job.id), function( err, hash ) {
     if( err ) return fn(err);
     if( !hash ) {
@@ -224,20 +226,21 @@ exports.get = function( id, jobType, fn ) {
 
 exports.removeBadJob = function( id, jobType) {
   var client = redis.client();
+  var zid = client.createFIFO(id);
   client.multi()
     .del(client.getKey('job:' + id + ':log'))
     .del(client.getKey('job:' + id))
-    .zrem(client.getKey('jobs:inactive'), id)
-    .zrem(client.getKey('jobs:active'), id)
-    .zrem(client.getKey('jobs:complete'), id)
-    .zrem(client.getKey('jobs:failed'), id)
-    .zrem(client.getKey('jobs:delayed'), id)
-    .zrem(client.getKey('jobs'), id)
-    .zrem(client.getKey('jobs:' + jobType + ':inactive'), id)
-    .zrem(client.getKey('jobs:' + jobType+ ':active'), id)
-    .zrem(client.getKey('jobs:' + jobType + ':complete'), id)
-    .zrem(client.getKey('jobs:' + jobType + ':failed'), id)
-    .zrem(client.getKey('jobs:' + jobType + ':delayed'), id)
+    .zrem(client.getKey('jobs:inactive'), zid)
+    .zrem(client.getKey('jobs:active'), zid)
+    .zrem(client.getKey('jobs:complete'), zid)
+    .zrem(client.getKey('jobs:failed'), zid)
+    .zrem(client.getKey('jobs:delayed'), zid)
+    .zrem(client.getKey('jobs'), zid)
+    .zrem(client.getKey('jobs:' + jobType + ':inactive'), zid)
+    .zrem(client.getKey('jobs:' + jobType+ ':active'), zid)
+    .zrem(client.getKey('jobs:' + jobType + ':complete'), zid)
+    .zrem(client.getKey('jobs:' + jobType + ':failed'), zid)
+    .zrem(client.getKey('jobs:' + jobType + ':delayed'), zid)
     .exec();
   if( !exports.disableSearch ) {
     getSearch().remove(id);
@@ -337,7 +340,7 @@ Job.prototype.toJSON = function() {
 
 
 Job.prototype.refreshTtl = function() {
-  ('active' === this.state() && this._ttl > 0) ? this.client.zadd(this.client.getKey('jobs:' + this.state()), Date.now() + parseInt(this._ttl), this.id) : noop();
+  ('active' === this.state() && this._ttl > 0) ? this.client.zadd(this.client.getKey('jobs:' + this.state()), Date.now() + parseInt(this._ttl), this.zid) : noop();
 };
 
 
@@ -625,9 +628,9 @@ Job.prototype.searchKeys = function( keys ) {
 Job.prototype.remove = function( fn ) {
   var client = this.client;
   client.multi()
-    .zrem(client.getKey('jobs:' + this.state()), this.id)
-    .zrem(client.getKey('jobs:' + this.type + ':' + this.state()), this.id)
-    .zrem(client.getKey('jobs'), this.id)
+    .zrem(client.getKey('jobs:' + this.state()), this.zid)
+    .zrem(client.getKey('jobs:' + this.type + ':' + this.state()), this.zid)
+    .zrem(client.getKey('jobs'), this.zid)
     .del(client.getKey('job:' + this.id + ':log'))
     .del(client.getKey('job:' + this.id))
     .exec(function( err ) {
@@ -658,18 +661,18 @@ Job.prototype.state = function( state, fn ) {
   var multi    = client.multi();
   if( oldState && oldState != '' && oldState != state ) {
     multi
-      .zrem(client.getKey('jobs:' + oldState), this.id)
-      .zrem(client.getKey('jobs:' + this.type + ':' + oldState), this.id);
+      .zrem(client.getKey('jobs:' + oldState), this.zid)
+      .zrem(client.getKey('jobs:' + this.type + ':' + oldState), this.zid);
   }
   multi
     .hset(client.getKey('job:' + this.id), 'state', state)
-    .zadd(client.getKey('jobs:' + state), this._priority, this.id)
-    .zadd(client.getKey('jobs:' + this.type + ':' + state), this._priority, this.id);
+    .zadd(client.getKey('jobs:' + state), this._priority, this.zid)
+    .zadd(client.getKey('jobs:' + this.type + ':' + state), this._priority, this.zid);
 
   // use promote_at as score when job moves to delayed
-  ('delayed' === state) ? multi.zadd(client.getKey('jobs:' + state), parseInt(this.promote_at), this.id) : noop();
-  ('active' === state && this._ttl > 0) ? multi.zadd(client.getKey('jobs:' + state), Date.now() + parseInt(this._ttl), this.id) : noop();
-  ('active' === state && !this._ttl) ? multi.zadd(client.getKey('jobs:' + state), this._priority<0?this._priority:-this._priority, this.id) : noop();
+  ('delayed' === state) ? multi.zadd(client.getKey('jobs:' + state), parseInt(this.promote_at), this.zid) : noop();
+  ('active' === state && this._ttl > 0) ? multi.zadd(client.getKey('jobs:' + state), Date.now() + parseInt(this._ttl), this.zid) : noop();
+  ('active' === state && !this._ttl) ? multi.zadd(client.getKey('jobs:' + state), this._priority<0?this._priority:-this._priority, this.zid) : noop();
   ('inactive' === state) ? multi.lpush(client.getKey(this.type + ':jobs'), 1) : noop();
 
   this.set('updated_at', Date.now());
@@ -776,6 +779,7 @@ Job.prototype.save = function( fn ) {
     // add the job for event mapping
     var key = client.getKey('job:' + id);
     self.id = id;
+    self.zid = client.createFIFO(id);
     self.subscribe(function() {
       self._state     = self._state || (this._delay ? 'delayed' : 'inactive');
       if( max ) client.hset(key, 'max_attempts', max);
@@ -836,7 +840,7 @@ Job.prototype.update = function( fn ) {
   // priority
   this.set('priority', this._priority);
 
-  this.client.zadd(this.client.getKey('jobs'), this._priority, this.id);
+  this.client.zadd(this.client.getKey('jobs'), this._priority, this.zid);
 
   // data
   this.set('data', json, function() {

lib/queue/worker.js

@@ -247,8 +247,8 @@ Worker.prototype.zpop = function( key, fn ) {
     .exec(function( err, res ) {
       if( err ) return fn(err);
       var id = res[ 0 ][ 0 ] || res[ 0 ][ 1 ][ 0 ];
-      fn(null, id);
-    });
+      fn(null, this.client.stripFIFO(id));
+    }.bind(this));
 };
 
 /**

lib/redis.js

@@ -60,10 +60,13 @@ exports.configureFactory = function( options, queue ) {
   exports.createClient = function() {
     var clientFactoryMethod = options.redis.createClientFactory || exports.createClientFactory;
     var client              = clientFactoryMethod(options);
+
     client.on('error', function( err ) {
       queue.emit('error', err);
     });
+
     client.prefix           = options.prefix;
+
     // redefine getKey to use the configured prefix
     client.getKey = function( key ) {
       if( client.constructor.name == 'Redis' ) {
@@ -73,6 +76,26 @@ exports.configureFactory = function( options, queue ) {
       }
       return this.prefix + ':' + key;
     };
+
+    client.createFIFO = function( id ) {
+      //Create an id for the zset to preserve FIFO order
+      var idLen = '' + id.toString().length;
+      var len = 2 - idLen.length;
+      while (len--) idLen = '0' + idLen;
+      return idLen + '|' + id;
+    };
+
+    // Parse out original ID from zid
+    client.stripFIFO = function( zid ) {
+      if ( typeof zid === 'string' ) {
+        return zid.substr(zid.indexOf('|')+1);
+      } else {
+        // Sometimes this gets called with an undefined
+        // it seems to be OK to have that not resolve to an id
+        return zid;
+      }
+    };
+
     return client;
   };
 };