fix(queue): enabled queues to share childPool instance

REFERENCE.md

@@ -107,6 +107,7 @@ interface AdvancedSettings {
   retryProcessDelay: number = 5000; // delay before processing next job in case of internal error.
   backoffStrategies: {}; // A set of custom backoff strategies keyed by name.
   drainDelay: number = 5; // A timeout for when the queue is in drained state (empty waiting for jobs).
+  isSharedChildPool: boolean = false; // enables multiple queues on the same instance of child pool to share the same instance.
 }
 ```
 
@@ -592,8 +593,8 @@ for the job when it was added.
 removeRepeatableByKey(key: string): Promise<void>
 ```
 
-Removes a given Repeatable Job configuration by its key so that no more repeatable jobs will be processed for this 
-particular configuration. 
+Removes a given Repeatable Job configuration by its key so that no more repeatable jobs will be processed for this
+particular configuration.
 
 There are currently two ways to get the "key" of a repeatable job.
 
@@ -691,7 +692,7 @@ Returns a promise that will return the waiting job counts for the given queue.
 
 ### Queue#getPausedCount
 
-*DEPRECATED* Since only the queue can be paused, getWaitingCount gives the same 
+*DEPRECATED* Since only the queue can be paused, getWaitingCount gives the same
 result.
 
 ```ts
@@ -970,7 +971,7 @@ A queue emits also some useful events:
 })
 
 .on('lock-extension-failed', function (job, err) {
-  // A job failed to extend lock. This will be useful to debug redis 
+  // A job failed to extend lock. This will be useful to debug redis
   // connection issues and jobs getting restarted because workers
   // are not able to extend locks.
 });

lib/process/child-pool.js

@@ -121,4 +121,15 @@ async function initChild(child, processFile) {
   );
   await onComplete;
 }
-module.exports = ChildPool;
+function ChildPoolSingleton(isSharedChildPool = false) {
+  if (
+    isSharedChildPool === false ||
+    (!(this instanceof ChildPool) && ChildPoolSingleton.instance === undefined)
+  ) {
+    ChildPoolSingleton.instance = new ChildPool();
+  }
+
+  return ChildPoolSingleton.instance;
+}
+
+module.exports = ChildPoolSingleton;

lib/queue.js

@@ -71,6 +71,7 @@ const MINIMUM_REDIS_VERSION = '2.8.18';
       guardInterval?: number = 5000,
       retryProcessDelay?: number = 5000,
       drainDelay?: number = 5
+      isSharedChildPool?: boolean = false
     }
   }
 
@@ -215,7 +216,8 @@ const Queue = function Queue(name, url, opts) {
     guardInterval: 5000,
     retryProcessDelay: 5000,
     drainDelay: 5,
-    backoffStrategies: {}
+    backoffStrategies: {},
+    isSharedChildPool: false
   });
 
   this.settings.lockRenewTime =
@@ -662,8 +664,9 @@ Queue.prototype.setHandler = function(name, handler) {
     if (!fs.existsSync(processorFile)) {
       throw new Error('File ' + processorFile + ' does not exist');
     }
-
-    this.childPool = this.childPool || require('./process/child-pool')();
+    const isSharedChildPool = this.settings.isSharedChildPool;
+    this.childPool =
+      this.childPool || require('./process/child-pool')(isSharedChildPool);
 
     const sandbox = require('./process/sandbox');
     this.handlers[name] = sandbox(handler, this.childPool).bind(this);

lib/scripts.js

@@ -326,17 +326,13 @@ const scripts = {
     return allRemoved;
   },
 
-  extendLock(
-    queue,
-    jobId,
-    duration,
-  ) {
+  extendLock(queue, jobId, duration) {
     return queue.client.extendLock([
       queue.toKey(jobId) + ':lock',
       queue.keys.stalled,
       queue.token,
       duration,
-      jobId,
+      jobId
     ]);
   },
 

test/test_child-pool.js

@@ -141,4 +141,12 @@ describe('Child pool', () => {
         expect(children).to.include(child);
       });
   });
+
+  it('should returned a shared child pool is isSharedChildPool is true', () => {
+    expect(childPool(true)).to.be.equal(new childPool(true));
+  });
+
+  it('should return a different childPool if isSharedChildPool is false', () => {
+    expect(childPool()).to.not.be.equal(childPool());
+  });
 });

test/test_queue.js

@@ -405,7 +405,7 @@ describe('Queue', () => {
     });
   });
 
-  describe(' a worker', () => {
+  describe('a worker', () => {
     let queue;
 
     beforeEach(() => {
@@ -2121,6 +2121,17 @@ describe('Queue', () => {
       queue.on('failed', cb);
       queue.on('error', done);
     });
+
+    it('should share child pool across all different queues created', async () => {
+      const [queueA, queueB] = await Promise.all([
+        utils.newQueue('queueA', { settings: { isSharedChildPool: true } }),
+        utils.newQueue('queueB', { settings: { isSharedChildPool: true } })
+      ]);
+
+      await Promise.all([queueA.add(), queueB.add()]);
+
+      expect(queueA.childPool).to.be.eql(queueB.childPool);
+    });
   });
 
   describe('Retries and backoffs', () => {
@@ -2837,11 +2848,14 @@ describe('Queue', () => {
         });
     });
 
-    it('should clean the number of jobs requested even if first jobs timestamp doesn\'t match', async () => {
+    it("should clean the number of jobs requested even if first jobs timestamp doesn't match", async () => {
       // This job shouldn't get deleted due to the 5000 grace
       await queue.add({ some: 'data' });
       // This job should get cleaned since 10000 > 5000 grace
-      const jobToClean = await queue.add({ some: 'data' }, { timestamp: Date.now() - 10000 });
+      const jobToClean = await queue.add(
+        { some: 'data' },
+        { timestamp: Date.now() - 10000 }
+      );
       // This job shouldn't get deleted due to the 5000 grace
       await queue.add({ some: 'data' });
 
@@ -2853,7 +2867,7 @@ describe('Queue', () => {
       expect(len).to.be.eql(2);
     });
 
-    it('shouldn\'t clean anything if all jobs are in grace period', async () => {
+    it("shouldn't clean anything if all jobs are in grace period", async () => {
       await queue.add({ some: 'data' });
       await queue.add({ some: 'data' });