Repository: taskforcesh/bullmq
Pull request #517 (2021-04-30 19:40:07)

feat: create option for non-blocking getNextJob
Closes https://github.com/taskforcesh/bullmq/issues/516



Patch:
diff --git a/docs/gitbook/CHANGELOG.md b/docs/gitbook/CHANGELOG.md
index 391b304b62..cb3f66235d 100644
--- a/docs/gitbook/CHANGELOG.md
+++ b/docs/gitbook/CHANGELOG.md
@@ -1,3 +1,10 @@
+## [1.23.1](https://github.com/taskforcesh/bullmq/compare/v1.23.0...v1.23.1) (2021-05-03)
+
+
+### Bug Fixes
+
+* add return type for job.waitUntilFinished() ([59ede97](https://github.com/taskforcesh/bullmq/commit/59ede976061a738503f70d9eb0c92a4b1d6ae4a3))
+
 # [1.23.0](https://github.com/taskforcesh/bullmq/compare/v1.22.2...v1.23.0) (2021-04-30)
 
 
diff --git a/docs/gitbook/api/bullmq.job.waituntilfinished.md b/docs/gitbook/api/bullmq.job.waituntilfinished.md
index ab5183f2a1..e962c5d10b 100644
--- a/docs/gitbook/api/bullmq.job.waituntilfinished.md
+++ b/docs/gitbook/api/bullmq.job.waituntilfinished.md
@@ -9,7 +9,7 @@ Returns a promise the resolves when the job has finished. (completed or failed).
 <b>Signature:</b>
 
 ```typescript
-waitUntilFinished(queueEvents: QueueEvents, ttl?: number): Promise<any>;
+waitUntilFinished(queueEvents: QueueEvents, ttl?: number): Promise<R>;
 ```
 
 ## Parameters
@@ -21,5 +21,5 @@ waitUntilFinished(queueEvents: QueueEvents, ttl?: number): Promise<any>;
 
 <b>Returns:</b>
 
-Promise&lt;any&gt;
+Promise&lt;R&gt;
 
diff --git a/package.json b/package.json
index 0f10a1d0c1..3529767255 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "bullmq",
-  "version": "1.23.0",
+  "version": "1.23.1",
   "description": "Queue for messages and jobs based on Redis",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
diff --git a/src/classes/job.ts b/src/classes/job.ts
index 06cb790cbd..047f983a1d 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -445,7 +445,7 @@ export class Job<T = any, R = any, N extends string = string> {
   /**
    * Returns a promise the resolves when the job has finished. (completed or failed).
    */
-  async waitUntilFinished(queueEvents: QueueEvents, ttl?: number) {
+  async waitUntilFinished(queueEvents: QueueEvents, ttl?: number): Promise<R> {
     await this.queue.waitUntilReady();
 
     const jobId = this.id;
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
index c5879de58d..65382b09fb 100644
--- a/src/classes/worker.ts
+++ b/src/classes/worker.ts
@@ -1,7 +1,7 @@
 import * as fs from 'fs';
 import { Redis } from 'ioredis';
 import * as path from 'path';
-import { Processor, WorkerOptions } from '../interfaces';
+import { Processor, WorkerOptions, GetNextJobOptions } from '../interfaces';
 import { QueueBase, Repeat } from './';
 import { ChildPool } from './child-pool';
 import { Job, JobJsonRaw } from './job';
@@ -173,16 +173,20 @@ export class Worker<
    * @param token worker token to be assigned to retrieved job
    * @returns a Job or undefined if no job was available in the queue.
    */
-  async getNextJob(token: string) {
+  async getNextJob(token: string, { block = true }: GetNextJobOptions = {}) {
     if (this.paused) {
-      await this.paused;
+      if (block) {
+        await this.paused;
+      } else {
+        return;
+      }
     }
 
     if (this.closing) {
       return;
     }
 
-    if (this.drained) {
+    if (this.drained && block) {
       try {
         const jobId = await this.waitForJob();
 
diff --git a/src/interfaces/worker-options.ts b/src/interfaces/worker-options.ts
index 6797c02d52..85a240fb3b 100644
--- a/src/interfaces/worker-options.ts
+++ b/src/interfaces/worker-options.ts
@@ -27,3 +27,7 @@ export interface WorkerOptions extends QueueBaseOptions {
   lockRenewTime?: number;
   settings?: AdvancedOptions; // FIXME only backoffStrategies is used
 }
+
+export interface GetNextJobOptions {
+  block?: boolean;
+}


Test patch:
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 81e0e4e3c4..687f6493b6 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1707,4 +1707,64 @@ describe('workers', function() {
       await worker.close();
     });
   });
+
+  describe('non-blocking', async () => {
+    it('should block by default', async () => {
+      const worker = new Worker(queueName);
+      const token = 'my-token';
+
+      // make sure worker is in drained state
+      await worker.getNextJob(token);
+
+      const [
+        job
+      ] = await Promise.all([
+        worker.getNextJob(token) as Promise<Job>,
+        delay(100).then(() => queue.add('test', { foo: 'bar' }))
+      ]);
+      expect(job).not.to.be.equal(undefined);
+      const isActive = await job.isActive();
+      expect(isActive).to.be.equal(true);
+    });
+
+    it('shouldn\'t block when disabled', async () => {
+      const worker = new Worker(queueName);
+      const token = 'my-token';
+
+      // make sure worker is in drained state
+      await worker.getNextJob(token, { block: false });
+
+      const [
+        job1
+      ] = await Promise.all([
+        worker.getNextJob(token, { block: false }) as Promise<Job>,
+        delay(100).then(() => queue.add('test', { foo: 'bar' }))
+      ]);
+      expect(job1).to.be.equal(undefined);
+
+      const job2 = (await worker.getNextJob(token, { block: false })) as Job;
+      const isActive = await job2.isActive();
+      expect(isActive).to.be.equal(true);
+    });
+
+    it('shouldn\'t block when disabled and paused', async () => {
+      const worker = new Worker(queueName);
+      const token = 'my-token';
+
+      // make sure worker is in drained state
+      await worker.getNextJob(token, { block: false });
+      await queue.add('test', { foo: 'bar' })
+
+      worker.pause();
+
+      const job1 = (await worker.getNextJob(token, { block: false })) as Job;
+      expect(job1).to.be.equal(undefined);
+
+      worker.resume();
+
+      const job2 = (await worker.getNextJob(token, { block: false })) as Job;
+      const isActive = await job2.isActive();
+      expect(isActive).to.be.equal(true);
+    });
+  });
 });