node-resque - npm Package Compare versions

Comparing version 0.11.5 to 0.11.6

lib/multiWorker.js

 var os = require('os');
-var util = require("util");
+var util = require('util');
 var EventEmitter = require('events').EventEmitter;
-var Worker = require(__dirname + "/worker.js").worker;
+var Worker = require(__dirname + '/worker.js').worker;
+var eventLoopDelay = require(__dirname + '/eventLoopDelay');
 
@@ -19,6 +20,17 @@ var multiWorker = function(options, jobs, callback){
   self.working = false;
+  self.eventLoopBlocked = true;
+  self.eventLoopCheckCounter = 0;
   process.nextTick(function(){ 
+    eventLoopDelay(
+      self.options.maxEventLoopDelay, 
+      self.options.checkTimeout, 
+    function(blocked, ms){
+      self.eventLoopBlocked = blocked;
+      self.eventLoopDelay = ms;
+      self.eventLoopCheckCounter++;
+    });
+
     callback(); 
   });
-}
+};
 
@@ -37,12 +49,5 @@ util.inherits(multiWorker, EventEmitter);
     toDisconnectProcessors: true,
-  }
-}
+  };
+};
 
-multiWorker.prototype.eventLoopDelay = function(callback){
-  var start = Date.now();
-  setImmediate(function(){
-    callback(null, (Date.now() - start));
-  });
-}
-
 multiWorker.prototype.startWorker = function(callback){
@@ -64,15 +69,15 @@ var self = this;
 
-  worker.on('start',           function(){                    self.emit('start', worker.id);                         })
-  worker.on('end',             function(){                    self.emit('end', worker.id);                           })
-  worker.on('cleaning_worker', function(worker, pid){         self.emit('cleaning_worker', worker.id, worker, pid);  })
-  worker.on('poll',            function(queue){               self.emit('poll', worker.id, queue);                   })
-  worker.on('job',             function(queue, job){          self.emit('job', worker.id, queue, job);               })
-  worker.on('reEnqueue',       function(queue, job, plugin){  self.emit('reEnqueue', worker.id, queue, job, plugin); })
-  worker.on('success',         function(queue, job, result){  self.emit('success', worker.id, queue, job, result);   })
-  worker.on('failure',         function(queue, job, failure){ self.emit('failure', worker.id, queue, job, failure);  })
-  worker.on('error',           function(queue, job, error){   self.emit('error', worker.id, queue, job, error);      })
-  worker.on('pause',           function(){                    self.emit('pause', worker.id);                         })
+  worker.on('start',           function(){                    self.emit('start', worker.id);                         });
+  worker.on('end',             function(){                    self.emit('end', worker.id);                           });
+  worker.on('cleaning_worker', function(worker, pid){         self.emit('cleaning_worker', worker.id, worker, pid);  });
+  worker.on('poll',            function(queue){               self.emit('poll', worker.id, queue);                   });
+  worker.on('job',             function(queue, job){          self.emit('job', worker.id, queue, job);               });
+  worker.on('reEnqueue',       function(queue, job, plugin){  self.emit('reEnqueue', worker.id, queue, job, plugin); });
+  worker.on('success',         function(queue, job, result){  self.emit('success', worker.id, queue, job, result);   });
+  worker.on('failure',         function(queue, job, failure){ self.emit('failure', worker.id, queue, job, failure);  });
+  worker.on('error',           function(queue, job, error){   self.emit('error', worker.id, queue, job, error);      });
+  worker.on('pause',           function(){                    self.emit('pause', worker.id);                         });
 
   self.workers.push(worker);
-}
+};
 
@@ -83,4 +88,5 @@ multiWorker.prototype.checkWorkers = function(callback){
   var workingCount = 0;
-  self.eventLoopDelay(function(err, delay){
 
+  setImmediate(function(){
+
     self.workers.forEach(function(worker){
@@ -98,8 +104,8 @@ if(worker.working === true){ workingCount++; }
     else if(self.running === false && self.workers.length === 0){                                             verb = 'x'; }
-    else if(delay > self.options.maxEventLoopDelay && self.workers.length > self.options.minTaskProcessors){  verb = '-'; }
-    else if(delay > self.options.maxEventLoopDelay && self.workers.length == self.options.minTaskProcessors){ verb = 'x'; }
-    else if(delay < self.options.maxEventLoopDelay && self.workers.length == self.options.maxTaskProcessors){ verb = 'x'; }
-    else if(delay < self.options.maxEventLoopDelay && self.workers.length < self.options.minTaskProcessors){  verb = '+'; }
+    else if(self.eventLoopBlocked  && self.workers.length > self.options.minTaskProcessors){  verb = '-'; }
+    else if(self.eventLoopBlocked  && self.workers.length == self.options.minTaskProcessors){ verb = 'x'; }
+    else if(!self.eventLoopBlocked && self.workers.length == self.options.maxTaskProcessors){ verb = 'x'; }
+    else if(!self.eventLoopBlocked && self.workers.length < self.options.minTaskProcessors){  verb = '+'; }
     else if(
-      delay < self.options.maxEventLoopDelay && 
+      !self.eventLoopBlocked && 
       self.workers.length < self.options.maxTaskProcessors && 
@@ -112,3 +118,3 @@ (
     else if(
-      delay < self.options.maxEventLoopDelay && 
+      !self.eventLoopBlocked && 
       self.workers.length > self.options.minTaskProcessors && 
@@ -121,3 +127,3 @@ workingCount / self.workers.length < 0.5
 
-    if(verb === 'x'){ callback(null, verb, delay); }
+    if(verb === 'x'){ callback(null, verb, self.eventLoopDelay); }
     if(verb === '-'){
@@ -127,3 +133,3 @@ var worker = self.workers.pop();
         self.cleanupWorker(worker);
-        callback(err, verb, delay);
+        callback(err, verb, self.eventLoopDelay);
       });
@@ -133,5 +139,5 @@ }
       var touched = 0;
-      if(self.workers.length === 0){ callback(err, verb, delay); }
+      if(self.workers.length === 0){ callback(err, verb, self.eventLoopDelay); }
       self.workers.forEach(function(worker){
-        touched++
+        touched++;
         worker.end(function(err){
@@ -142,3 +148,3 @@ touched--;
             self.workers = [];
-            callback(err, verb, delay); 
+            callback(err, verb, self.eventLoopDelay); 
           }
@@ -150,7 +156,7 @@ });
       self.startWorker(function(err){
-        callback(err, verb, delay);
+        callback(err, verb, self.eventLoopDelay);
       });
     }
   });
-}
+};
 
@@ -180,3 +186,3 @@ multiWorker.prototype.cleanupWorker = function(worker){
   }
-}
+};
 
@@ -193,3 +199,3 @@ multiWorker.prototype.checkWraper = function(){
   });
-}
+};
 
@@ -203,3 +209,3 @@ multiWorker.prototype.start = function(callback){
   });
-}
+};
 
@@ -210,3 +216,3 @@ multiWorker.prototype.stop = function(callback){
   self.stopWait(callback);
-}
+};
 
@@ -216,3 +222,3 @@ multiWorker.prototype.end = function(callback){
   self.stop(callback);
-}
+};
 
@@ -225,10 +231,10 @@ multiWorker.prototype.stopWait = function(callback){
       if(typeof callback === 'function'){ callback(); }
-    }, self.options.checkTimeout * 2)
+    }, self.options.checkTimeout * 2);
   }else{
     setTimeout(function(){
       self.stopWait(callback);
-    }, self.options.checkTimeout)
+    }, self.options.checkTimeout);
   }
-}
+};
 
 exports.multiWorker = multiWorker;

package.json

@@ -5,3 +5,3 @@ {
   "description": "an opinionated implementation of resque in node",
-  "version": "0.11.5",
+  "version": "0.11.6",
   "homepage": "http://github.com/taskrabbit/node-resque",
@@ -8,0 +8,0 @@ "repository": {

test/core/multiWorker.js

@@ -5,3 +5,3 @@ var specHelper = require(__dirname + "/../_specHelper.js").specHelper;
 if(specHelper.package === 'fakeredis'){
-  console.log("multiWorker does not work with fakeredis for now...")
+  console.log("multiWorker does not work with fakeredis for now...");
 }else{
@@ -13,4 +13,4 @@
     var checkTimeout = specHelper.timeout / 10;
-    var minTaskProcessors = 1
-    var maxTaskProcessors = 5
+    var minTaskProcessors = 1;
+    var maxTaskProcessors = 5;
 
@@ -27,5 +27,5 @@ var toDisconnectProcessors = true;
         var alarmMSeconds = alarm.getTime();
-        if(alarmMSeconds - startingMSeconds > naptime){ sleeping = false }
+        if(alarmMSeconds - startingMSeconds > naptime){ sleeping = false; }
       }
-    }
+    };
 
@@ -82,3 +82,3 @@ var jobs = {
         should.not.exist(err);
-        done()
+        done();
       });
@@ -127,3 +127,3 @@ });
           multiWorker.end(done);
-        }, checkTimeout * 20);
+        }, checkTimeout * 30);
       });
@@ -130,0 +130,0 @@ });

New alerts

Dynamic require

Supply chain risk

Dynamic require can indicate the package is performing dangerous or unsafe dynamic code execution.

Found 1 instance in 1 package

Fixed alerts

Dynamic require

Supply chain risk

Dynamic require can indicate the package is performing dangerous or unsafe dynamic code execution.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

130852

increased by0.49%

Number of package files

39

increased by2.63%

Lines of code

2993

increased by0.67%

Worsened metrics

Number of low supply chain risk alerts

37

increased by2.78%

No dependency changes