Repository: taskforcesh/bullmq
Pull request #574 (2021-06-01 02:00:13)

Fix QueueScheduler: adding initial event typing
For now I added only 2 events, at least those that I could check from tests

manast: these are nice little improvements, thanks!.
manast: I am going to merge is as a "feat" though, since it could actually break a CI if they have using a wrong signature.


Patch:
diff --git a/src/classes/queue-scheduler.ts b/src/classes/queue-scheduler.ts
index 1eadebb0f3..e2c9dbbf2d 100644
--- a/src/classes/queue-scheduler.ts
+++ b/src/classes/queue-scheduler.ts
@@ -5,7 +5,14 @@ import { Scripts } from './scripts';
 import { StreamReadRaw } from '../interfaces/redis-streams';
 import { RedisClient } from './redis-connection';
 
-const MAX_TIMEOUT_MS = Math.pow(2, 31) - 1; // 32 bit signed
+export declare interface QueueScheduler {
+  on(event: 'stalled', listener: (jobId: string, prev: string) => void): this;
+  on(
+    event: 'failed',
+    listener: (jobId: string, failedReason: Error, prev: string) => void,
+  ): this;
+  on(event: string, listener: Function): this;
+}
 
 /**
  * This class is just used for some automatic bookkeeping of the queue,
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
index 841bf9e46f..6aa95ec9a8 100644
--- a/src/classes/worker.ts
+++ b/src/classes/worker.ts
@@ -17,6 +17,12 @@ import { isRedisInstance, delay } from '../utils';
 
 export const clientCommandMessageReg = /ERR unknown command ['`]\s*client\s*['`]/;
 
+export declare interface Worker {
+  on(event: 'completed', listener: (job: Job) => void): this;
+  on(event: 'error', listener: (failedReason: Error) => void): this;
+  on(event: string, listener: Function): this;
+}
+
 export class Worker<
   T = any,
   R = any,


Test patch:
diff --git a/src/test/test_stalled_jobs.ts b/src/test/test_stalled_jobs.ts
index 8ac0cce6a5..055f915afc 100644
--- a/src/test/test_stalled_jobs.ts
+++ b/src/test/test_stalled_jobs.ts
@@ -61,7 +61,13 @@ describe('stalled jobs', function() {
     await worker.close(true);
 
     const allStalled = new Promise(resolve => {
-      queueScheduler.on('stalled', after(concurrency, resolve));
+      queueScheduler.on(
+        'stalled',
+        after(concurrency, (jobId, prev) => {
+          expect(prev).to.be.equal('active');
+          resolve();
+        }),
+      );
     });
 
     const allStalledGlobalEvent = new Promise(resolve => {
@@ -132,7 +138,8 @@ describe('stalled jobs', function() {
     const allFailed = new Promise(resolve => {
       queueScheduler.on(
         'failed',
-        after(concurrency, (jobId, failedReason) => {
+        after(concurrency, (jobId, failedReason, prev) => {
+          expect(prev).to.be.equal('active');
           expect(failedReason.message).to.be.equal(errorMessage);
           resolve();
         }),
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 7b5ec787c6..5426ed0f76 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -942,16 +942,15 @@ describe('workers', function() {
 
     const job = await queue.add('test', { bar: 'baz' });
 
+    const errorMessage = `Missing lock for job ${job.id} failed`;
     const workerError = new Promise(resolve => {
-      worker.on('error', resolve);
+      worker.on('error', error => {
+        expect(error.message).to.be.equal(errorMessage);
+        resolve();
+      });
     });
 
-    const error = await workerError;
-
-    expect(error).to.be.instanceOf(Error);
-    expect((error as Error).message).to.be.eql(
-      `Missing lock for job ${job.id} failed`,
-    );
+    await workerError;
 
     await worker.close();
     await queueScheduler.close();