bull - npm Package Compare versions

Comparing version 3.3.7 to 3.3.8

CHANGELOG.md

@@ -0,1 +1,8 @@
+v.3.3.8
+=======
+
+- Fixed #812. External process doesn't terminate on `queue.close()`.
+- Fixed #830. Named Process Sent to Wrong Processor.
+- Fixed #572. Do not close external connections.
+
 v.3.3.7
@@ -2,0 +9,0 @@ =======

lib/job.js

@@ -161,3 +161,3 @@ /*eslint-env node */
     if(unlocked != 1){
-      throw Error('Could not release lock for job ' + _this.id);
+      throw new Error('Could not release lock for job ' + _this.id);
     }
@@ -337,3 +337,3 @@ });
     }else{
-      throw Error('Could not remove job ' + job.id);
+      throw new Error('Could not remove job ' + job.id);
     }
@@ -340,0 +340,0 @@ });

lib/process/child-pool.js

@@ -14,6 +14,6 @@ 'use strict';
   this.retained = {};
-  this.free = [];
+  this.free = {};
 
   this.retain = Promise.method(function(processFile){
-    var child = this.free.pop();
+    var child = this.getFree(processFile).pop();
 
@@ -25,3 +25,4 @@ if (child) {
     child = fork(path.join(__dirname, './master.js'));
-
+    child.processFile = processFile;
+  
     this.retained[child.pid] = child;
@@ -42,3 +43,3 @@
     delete this.retained[child.pid];
-    this.free.push(child);
+    this.getFree(child.processFile).push(child);
   };
@@ -48,5 +49,8 @@
     delete this.retained[child.pid];
-    var childIndex = this.free.indexOf(child);
+
+    var free = this.getFree(child.processFile);
+    
+    var childIndex = free.indexOf(child);
     if (childIndex > -1) {
-      this.free.splice(childIndex, 1);
+      free.splice(childIndex, 1);
     }
@@ -61,11 +65,20 @@ };
   this.clean = function(){
-    var children = _.values(this.retained).concat(this.free);
+    var children = _.values(this.retained).concat(this.getAllFree());
     var _this = this;
     children.forEach(function(child){
-      _this.kill(child, 0);
+      // TODO: We may want to use SIGKILL if the process does not die after some time.
+      _this.kill(child, 'SIGTERM');
     });
 
     this.retained = {};
-    this.free = [];
+    this.free = {};
   };
+
+  this.getFree = function(id) {
+    return this.free[id] = this.free[id] || [];
+  };
+
+  this.getAllFree = function(){
+    return _.flatten(_.values(this.free));
+  };
 };

lib/queue.js

@@ -237,3 +237,5 @@ /*eslint-env node */
       var client = connections[type] = createClient(type, options.redis);
-      queue.clients.push(client);
+      if(!options.createClient){
+        queue.clients.push(client);
+      }
       return initCallback(type, client), client;
@@ -268,2 +270,3 @@ };
         }
+        return null;
       }).catch(function(err){
@@ -319,2 +322,3 @@ queue.emit('error', err);
         }
+        return null;
       }).return(null);
@@ -321,0 +325,0 @@ }).then(function(){

package.json

 {
   "name": "bull",
-  "version": "3.3.7",
+  "version": "3.3.8",
   "description": "Job manager",
@@ -5,0 +5,0 @@ "main": "./index.js",

PATTERNS.md

@@ -70,5 +70,8 @@
 ```js
-var client = new redis();
-var subscriber = new redis();
+var {REDIS_URL} = process.env
 
+var Redis = require('ioredis')
+var client = new Redis(REDIS_URL);
+var subscriber = new Redis(REDIS_URL);
+
 var opts = {
@@ -82,3 +85,3 @@ createClient: function(type){
         default:
-          return new redis();
+          return new Redis();
       }
@@ -85,0 +88,0 @@ }

README.md

@@ -63,2 +63,5 @@
   </a>
+  <a href="http://taskforce.sh"  style="margin-left: 50px;">
+    <img src="http://taskforce.sh/assets/logo_square.png" width="160" alt="Taskforce.sh, Inc" />
+  </a>
 </div>
@@ -99,2 +102,3 @@
 
+- [NEW (Preview) Taskforce](https://taskforce.sh)
 - [Arena](https://github.com/mixmaxhq/arena)
@@ -101,0 +105,0 @@

REFERENCE.md

@@ -58,3 +58,3 @@
 ```typescript
-interface QueueOpts{
+interface QueueOptions {
   limiter?: RateLimiter;
@@ -121,3 +121,3 @@ redis?: RedisOpts;
  * comply with one of the below defined patterns.
- * 
+ *
  * Note: Concurrency defaults to 1 if not specified.
@@ -124,0 +124,0 @@ */

test/test_connection.js

@@ -74,2 +74,34 @@ /*eslint-env node */
 
+  it('should not close external connections', function () {
+
+    var client = new redis();
+    var subscriber = new redis();
+    
+    var opts = {
+      createClient: function(type){
+        switch(type){
+          case 'client':
+            return client;
+          case 'subscriber':
+            return subscriber;
+          default:
+            return new redis();
+        }
+      }
+    };
+
+    var testQueue = utils.buildQueue('external connections', opts);
+
+    return testQueue.isReady().then(function(){
+      return testQueue.add({'foo': 'bar'});
+    }).then(function(){
+      expect(testQueue.client).to.be.eql(client);
+      expect(testQueue.eclient).to.be.eql(subscriber);
+
+      return testQueue.close();
+    }).then(function(){
+      expect(client.status).to.be.eql('ready');
+      expect(subscriber.status).to.be.eql('ready');
+    });
+  });
 });

test/test_repeat.js

@@ -15,2 +15,3 @@ /*eslint-env node */
 var ONE_DAY = 24 * ONE_HOUR;
+var MAX_INT = 2147483647;
 
@@ -25,4 +26,4 @@ describe('repeat', function () {
       queue = utils.buildQueue('repeat', {settings: {
-        guardInterval: Number.MAX_VALUE,
-        stalledInterval: Number.MAX_VALUE,
+        guardInterval: MAX_INT,
+        stalledInterval: MAX_INT,
         drainDelay: 1 // Small delay so that .close is faster.
@@ -29,0 +30,0 @@ }});

test/test_sandboxed_process.js

@@ -17,4 +17,4 @@ /*eslint-env node */
       queue = utils.buildQueue('test process', {settings: {
-        guardInterval: Number.MAX_VALUE,
-        stalledInterval: Number.MAX_VALUE
+        guardInterval: 300000,
+        stalledInterval: 300000,
       }});
@@ -33,3 +33,4 @@ return queue;
   it('should process and complete', function (done) {
-    queue.process(__dirname + '/fixtures/fixture_processor.js');
+    var processFile = __dirname + '/fixtures/fixture_processor.js';
+    queue.process(processFile);
 
@@ -41,3 +42,3 @@ queue.on('completed', function(job, value){
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.free[processFile]).to.have.lengthOf(1);
         done();
@@ -53,3 +54,4 @@ } catch (err) {
   it('should process with named processor', function (done) {
-    queue.process('foobar', __dirname + '/fixtures/fixture_processor.js');
+    var processFile = __dirname + '/fixtures/fixture_processor.js';
+    queue.process('foobar', processFile);
 
@@ -61,3 +63,3 @@ queue.on('completed', function(job, value){
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.free[processFile]).to.have.lengthOf(1);
         done();
@@ -72,5 +74,53 @@ } catch (err) {
 
+  it('should process with several named processors', function (done) {
+    var processFileFoo =  __dirname + '/fixtures/fixture_processor_foo.js';
+    var processFileBar =  __dirname + '/fixtures/fixture_processor_bar.js';
+    
+    queue.process('foo', processFileFoo);
+    queue.process('bar', processFileBar);
+    
+    var count = 0;
+    queue.on('completed', function(job, value){
+      var data, result, processFile, retainedLength;
+      count++;
+      if(count == 1){
+        data = {foo: 'bar'};
+        result = 'foo';
+        processFile = processFileFoo;
+        retainedLength = 1;
+      }else {
+        data = {bar: 'qux'};
+        result = 'bar';
+        processFile = processFileBar;
+        retainedLength = 0;
+      }
+
+      try {
+        expect(job.data).to.be.eql(data);
+        expect(value).to.be.eql(result);
+        expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(retainedLength);
+        expect(queue.childPool.free[processFile]).to.have.lengthOf(1);
+        if(count === 2){
+          done();
+        }
+      } catch (err) {
+        console.error(err);
+        done(err);
+      }
+    });
+
+    queue.add('foo', {foo: 'bar'}).then(function(){
+      Promise.delay(500).then(function(){
+        queue.add('bar', {bar: 'qux'});
+      });
+    });
+
+    queue.on('error', function(err){
+      console.error(err);
+    });
+  });
+
   it('should process with concurrent processors', function (done) {
     var after = _.after(4, function(){
-      expect(queue.childPool.free.length).to.eql(4);
+      expect(queue.childPool.getAllFree().length).to.eql(4);
       done();
@@ -81,3 +131,3 @@ });
         expect(value).to.be.eql(42);
-        expect(Object.keys(queue.childPool.retained).length + queue.childPool.free.length).to.eql(4);
+        expect(Object.keys(queue.childPool.retained).length + queue.childPool.getAllFree().length).to.eql(4);
         after();
@@ -107,3 +157,3 @@ } catch (err) {
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
         done();
@@ -128,3 +178,3 @@ } catch (err) {
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
         done();
@@ -153,3 +203,3 @@ } catch (err) {
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
         done();
@@ -173,3 +223,3 @@ } catch (err) {
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
         done();
@@ -190,8 +240,7 @@ } catch (err) {
         expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-        expect(queue.childPool.free).to.have.lengthOf(1);
+        expect(queue.childPool.getAllFree()).to.have.lengthOf(1);
         Promise.delay(500).then(function(){
           expect(Object.keys(queue.childPool.retained)).to.have.lengthOf(0);
-          expect(queue.childPool.free).to.have.lengthOf(0);
-        })
-          .asCallback(done);
+          expect(queue.childPool.getAllFree()).to.have.lengthOf(0);
+        }).asCallback(done);
       } catch (err) {
@@ -198,0 +247,0 @@ done(err);

test/test_worker.js

@@ -15,4 +15,4 @@ /*eslint-env node */
       queue = utils.buildQueue('test workers', {settings: {
-        guardInterval: Number.MAX_VALUE,
-        stalledInterval: Number.MAX_VALUE
+        guardInterval: 300000,
+        stalledInterval: 300000
       }});
@@ -19,0 +19,0 @@ return queue;

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

596037

increased by0.68%

Number of package files

83

increased by3.75%

Lines of code

6452

increased by1.64%

Number of lines in readme file

419

increased by0.96%

No dependency changes