fix(worker): setname on worker blocking connection (#291)

Author: manast

.github/workflows/coverage.yml

@@ -18,7 +18,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [12]
+        node-version: [14]
 
     steps:
       - uses: actions/checkout@v2

src/classes/worker.ts

@@ -85,6 +85,11 @@ export class Worker<T = any> extends QueueBase {
     this.on('error', err => console.error(err));
   }
 
+  async waitUntilReady() {
+    await super.waitUntilReady();
+    return this.blockingConnection.client;
+  }
+
   get repeat() {
     return new Promise<Repeat>(async resolve => {
       if (!this._repeat) {
@@ -99,7 +104,7 @@ export class Worker<T = any> extends QueueBase {
   }
 
   private async run() {
-    const client = await this.client;
+    const client = await this.blockingConnection.client;
 
     if (this.closing) {
       return;

src/test/test_worker.ts

@@ -930,6 +930,57 @@ describe('workers', function() {
     await queueScheduler.close();
   });
 
+  it('continue processing after a worker has stalled', async function() {
+    let first = true;
+    this.timeout(10000);
+
+    const worker = new Worker(
+      queueName,
+      async job => {
+        if (first) {
+          first = false;
+          return delay(2000);
+        }
+      },
+      {
+        lockDuration: 1000,
+        lockRenewTime: 3000, // The lock will not be updated
+      },
+    );
+    await worker.waitUntilReady();
+
+    const queueScheduler = new QueueScheduler(queueName, {
+      stalledInterval: 100,
+    });
+    await queueScheduler.waitUntilReady();
+
+    const job = await queue.add('test', { bar: 'baz' });
+
+    const completed = new Promise(resolve => {
+      worker.on('completed', resolve);
+    });
+
+    await completed;
+
+    await worker.close();
+    await queueScheduler.close();
+  });
+
+  it('stalled interval cannot be zero', function(done) {
+    this.timeout(10000);
+    let queueScheduler;
+
+    try {
+      queueScheduler = new QueueScheduler(queueName, {
+        stalledInterval: 0,
+      });
+      // Fail test if we reach here.
+      done(new Error('Should throw an exception'));
+    } catch (err) {
+      done();
+    }
+  });
+
   describe('Concurrency process', () => {
     it('should run job in sequence if I specify a concurrency of 1', async () => {
       let processing = false;