Repository: taskforcesh/bullmq
Pull request #153 (2020-03-03 21:02:50)

fix(worker): return this.closing when calling close


Patch:
diff --git a/src/classes/queue-base.ts b/src/classes/queue-base.ts
index f65ab60bff..a8696e953c 100644
--- a/src/classes/queue-base.ts
+++ b/src/classes/queue-base.ts
@@ -72,7 +72,10 @@ export class QueueBase extends EventEmitter {
   }
 
   close() {
-    return (this.closing = this.connection.close());
+    if (!this.closing) {
+      this.closing = this.connection.close();
+    }
+    return this.closing;
   }
 
   disconnect() {
diff --git a/src/classes/queue-events.ts b/src/classes/queue-events.ts
index b87774c48e..f64743e5bb 100644
--- a/src/classes/queue-events.ts
+++ b/src/classes/queue-events.ts
@@ -68,6 +68,9 @@ export class QueueEvents extends QueueBase {
   }
 
   async close() {
-    return (this.closing = this.disconnect());
+    if (!this.closing) {
+      this.closing = this.disconnect();
+    }
+    return this.closing;
   }
 }
diff --git a/src/classes/queue-scheduler.ts b/src/classes/queue-scheduler.ts
index 7b745a2157..b94441f33f 100644
--- a/src/classes/queue-scheduler.ts
+++ b/src/classes/queue-scheduler.ts
@@ -170,10 +170,13 @@ export class QueueScheduler extends QueueBase {
   }
 
   async close() {
+    if (this.closing) {
+      return this.closing;
+    }
     if (this.isBlocked) {
       this.closing = this.disconnect();
     } else {
-      super.close();
+      this.closing = super.close();
     }
     return this.closing;
   }
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
index bc00a05c2d..3c855269d1 100644
--- a/src/classes/worker.ts
+++ b/src/classes/worker.ts
@@ -372,13 +372,12 @@ export class Worker<T = any> extends QueueBase {
           const client = await this.blockingConnection.client;
 
           await this.resume();
+
           if (!force) {
             await this.whenCurrentJobsFinished(false);
           } else {
             await client.disconnect();
           }
-          // await this.disconnect();
-          await super.close();
         } catch (err) {
           reject(err);
         } finally {


Test patch:
diff --git a/src/test/test_stalled_jobs.ts b/src/test/test_stalled_jobs.ts
index 6f16d60e12..1ac1fbe876 100644
--- a/src/test/test_stalled_jobs.ts
+++ b/src/test/test_stalled_jobs.ts
@@ -135,17 +135,17 @@ describe('stalled jobs', function() {
   });
 
   it('jobs not stalled while lock is extended', async function() {
-    this.timeout(6000);
+    this.timeout(5000);
 
     const concurrency = 4;
 
     const worker = new Worker(
       queueName,
       async job => {
-        return delay(10000);
+        return delay(4000);
       },
       {
-        lockDuration: 1000, // lockRenewTime would be half of it i.e. 500
+        lockDuration: 100, // lockRenewTime would be half of it i.e. 500
         concurrency,
       },
     );
@@ -154,8 +154,6 @@ describe('stalled jobs', function() {
       worker.on('active', after(concurrency, resolve));
     });
 
-    await worker.waitUntilReady();
-
     await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
@@ -166,15 +164,14 @@ describe('stalled jobs', function() {
     await allActive;
 
     const queueScheduler = new QueueScheduler(queueName, {
-      stalledInterval: 100,
+      stalledInterval: 50,
     });
-    await queueScheduler.waitUntilReady();
 
-    const allStalled = new Promise(resolve => {
-      queueScheduler.on('stalled', after(concurrency, resolve));
-    });
+    const allStalled = new Promise(resolve =>
+      queueScheduler.on('stalled', after(concurrency, resolve)),
+    );
 
-    await delay(2000); // let worker to extend lock for several times
+    await delay(500); // Wait for jobs to become active
 
     const active = await queue.getActiveCount();
     expect(active).to.be.equal(4);