Repository: taskforcesh/bullmq
Pull request #581 (2021-06-05 02:16:05)

Feat job: adding getDependenciesCount method
Would be nice to have a method to get processed and unprocessed counts without bringing all the dependencies values

ccollie: Hey @roggervalf, while you're at it, could you export `DependenciesOpts` ?

https://github.com/taskforcesh/bullmq/blob/87b360323ed3ab4d43ccd3e278413137479cdf3b/src/classes/job.ts#L57
roggervalf: > Hey @roggervalf, while you're at it, could you export `DependenciesOpts` ?
> 
> https://github.com/taskforcesh/bullmq/blob/87b360323ed3ab4d43ccd3e278413137479cdf3b/src/classes/job.ts#L57

@ccollie, sure no problem
roggervalf: @manast please when you get a change, some :eyes: 
manast: LGTM. Is this ready for merge?
roggervalf: @manast yes it is ready

Patch:
diff --git a/src/classes/flow-producer.ts b/src/classes/flow-producer.ts
index 5a626c3edb..4d42a375b4 100644
--- a/src/classes/flow-producer.ts
+++ b/src/classes/flow-producer.ts
@@ -9,7 +9,7 @@ import { KeysMap, QueueKeys } from './queue-keys';
 import { FlowJob } from '../interfaces/flow-job';
 import { Job } from './job';
 
-interface NodeOpts {
+export interface NodeOpts {
   queueName: string;
   prefix?: string;
   id: string;
@@ -306,7 +306,7 @@ export class FlowProducer extends EventEmitter {
   }
 }
 
-export function getParentKey(opts: { id: string; queue: string }) {
+export function getParentKey(opts: { id: string; queue: string }): string {
   if (opts) {
     return `${opts.queue}:${opts.id}`;
   }
diff --git a/src/classes/job.ts b/src/classes/job.ts
index b2159d4bd1..373c1a00fb 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -54,7 +54,7 @@ export interface MoveToChildrenOpts {
   };
 }
 
-interface DependenciesOpts {
+export interface DependenciesOpts {
   processed?: {
     cursor?: number;
     count?: number;
@@ -525,8 +525,9 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   /**
+   * @method getState
    * Get current state.
-   * @method
+   *
    * @returns {string} Returns one of these values:
    * 'completed', 'failed', 'delayed', 'active', 'waiting', 'waiting-children', 'unknown'.
    */
@@ -554,6 +555,7 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   /**
+   * @method getDependencies
    * Get children job keys if this job is a parent and has children.
    *
    * @returns dependencies separated by processed and unprocessed.
@@ -569,8 +571,8 @@ export class Job<T = any, R = any, N extends string = string> {
     const client = await this.queue.client;
     const multi = client.multi();
     if (!opts.processed && !opts.unprocessed) {
-      await multi.hgetall(this.toKey(`${this.id}:processed`));
-      await multi.smembers(this.toKey(`${this.id}:dependencies`));
+      multi.hgetall(this.toKey(`${this.id}:processed`));
+      multi.smembers(this.toKey(`${this.id}:dependencies`));
 
       const [[err1, processed], [err2, unprocessed]] = (await multi.exec()) as [
         [null | Error, { [jobKey: string]: string }],
@@ -593,7 +595,7 @@ export class Job<T = any, R = any, N extends string = string> {
 
       if (opts.processed) {
         const processedOpts = Object.assign(defaultOpts, opts.processed);
-        await multi.hscan(
+        multi.hscan(
           this.toKey(`${this.id}:processed`),
           processedOpts.cursor,
           'COUNT',
@@ -603,7 +605,7 @@ export class Job<T = any, R = any, N extends string = string> {
 
       if (opts.unprocessed) {
         const unprocessedOpts = Object.assign(defaultOpts, opts.unprocessed);
-        await multi.sscan(
+        multi.sscan(
           this.toKey(`${this.id}:dependencies`),
           unprocessedOpts.cursor,
           'COUNT',
@@ -653,6 +655,62 @@ export class Job<T = any, R = any, N extends string = string> {
     }
   }
 
+  /**
+   * @method getDependenciesCount
+   * Get children job counts if this job is a parent and has children.
+   *
+   * @returns dependencies count separated by processed and unprocessed.
+   */
+  async getDependenciesCount(
+    opts: {
+      processed?: boolean;
+      unprocessed?: boolean;
+    } = {},
+  ): Promise<{
+    processed?: number;
+    unprocessed?: number;
+  }> {
+    const client = await this.queue.client;
+    const multi = client.multi();
+
+    const updatedOpts =
+      !opts.processed && !opts.unprocessed
+        ? { processed: true, unprocessed: true }
+        : opts;
+
+    if (updatedOpts.processed) {
+      multi.hlen(this.toKey(`${this.id}:processed`));
+    }
+
+    if (updatedOpts.unprocessed) {
+      multi.scard(this.toKey(`${this.id}:dependencies`));
+    }
+
+    const [
+      [err1, result1] = [],
+      [err2, result2] = [],
+    ] = (await multi.exec()) as [
+      [null | Error, number],
+      [null | Error, number],
+    ];
+
+    const processed = updatedOpts.processed ? result1 : undefined;
+    const unprocessed = updatedOpts.unprocessed
+      ? updatedOpts.processed
+        ? result2
+        : result1
+      : undefined;
+
+    return {
+      ...(updatedOpts.processed
+        ? {
+            processed,
+          }
+        : {}),
+      ...(updatedOpts.unprocessed ? { unprocessed } : {}),
+    };
+  }
+
   /**
    * Returns a promise the resolves when the job has finished. (completed or failed).
    */
diff --git a/src/classes/queue-getters.ts b/src/classes/queue-getters.ts
index 73494df718..94a51e6744 100644
--- a/src/classes/queue-getters.ts
+++ b/src/classes/queue-getters.ts
@@ -196,8 +196,9 @@ export class QueueGetters extends QueueBase {
   }
 
   /**
+   * @method getWorkers
    * Get worker list related to the queue.
-   * @method
+   *
    * @returns {Object} Returns an array with workers info.
    */
   async getWorkers() {


Test patch:
diff --git a/src/test/test_bulk.ts b/src/test/test_bulk.ts
index 75623915e7..837604f36c 100644
--- a/src/test/test_bulk.ts
+++ b/src/test/test_bulk.ts
@@ -92,9 +92,11 @@ describe('bulk jobs', () => {
     expect(jobs[1].id).to.be.ok;
     expect(jobs[1].data.foo).to.be.eql('baz');
 
-    const { unprocessed } = await parent.getDependencies();
+    const { unprocessed } = await parent.getDependenciesCount({
+      unprocessed: true,
+    });
 
-    expect(unprocessed).to.have.length(2);
+    expect(unprocessed).to.be.equal(2);
 
     await childrenWorker.close();
     await parentWorker.close();
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
index e559dc252b..ff4befd4a5 100644
--- a/src/test/test_flow.ts
+++ b/src/test/test_flow.ts
@@ -404,7 +404,10 @@ describe('flows', () => {
       (resolve, reject) =>
         (parentProcessor = async (job: Job) => {
           try {
-            expect(processedChildren).to.be.equal(3);
+            const { processed, unprocessed } = await job.getDependenciesCount();
+
+            expect(processed).to.be.equal(3);
+            expect(unprocessed).to.be.equal(0);
 
             const childrenValues = await job.getChildrenValues();
 
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 04cf328dff..cbe1552fdd 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1847,12 +1847,18 @@ describe('workers', function() {
         unprocessed: unprocessed3,
       } = await parent.getDependencies();
       const isWaitingChildren1 = await parent.isWaitingChildren();
+      const {
+        processed: processedCount,
+        unprocessed: unprocessedCount,
+      } = await parent.getDependenciesCount();
 
       expect(processed3).to.deep.equal({
         [`bull:${queueName}:${child1.id}`]: 'return value1',
         [`bull:${queueName}:${child2.id}`]: 'return value2',
       });
+      expect(processedCount).to.be.equal(2);
       expect(unprocessed3).to.have.length(1);
+      expect(unprocessedCount).to.be.equal(1);
       expect(isWaitingChildren1).to.be.true;
 
       const isActive3 = await child3.isActive();
@@ -1902,30 +1908,33 @@ describe('workers', function() {
 
       expect(currentState).to.be.equal('active');
 
-      await times(15, async (index: number) =>
-        Job.create(
-          queue,
-          `child${index}`,
-          { idx: index, ...value },
-          {
-            parent: {
-              id: parent.id,
-              queue: 'bull:' + parentQueueName,
+      await Promise.all(
+        Array.from(Array(65).keys()).map((index: number) => {
+          return Job.create(
+            queue,
+            `child${index}`,
+            { idx: index, ...value },
+            {
+              parent: {
+                id: parent.id,
+                queue: 'bull:' + parentQueueName,
+              },
             },
-          },
-        ),
+          );
+        }),
       );
+
       const {
         nextUnprocessedCursor: nextCursor1,
         unprocessed: unprocessed1,
       } = await parent.getDependencies({
         unprocessed: {
           cursor: 0,
-          count: 10,
+          count: 50,
         },
       });
 
-      expect(unprocessed1.length).to.be.greaterThanOrEqual(10);
+      expect(unprocessed1.length).to.be.greaterThanOrEqual(50);
 
       const {
         nextUnprocessedCursor: nextCursor2,
@@ -1933,11 +1942,11 @@ describe('workers', function() {
       } = await parent.getDependencies({
         unprocessed: {
           cursor: nextCursor1,
-          count: 10,
+          count: 50,
         },
       });
 
-      expect(unprocessed2.length).to.be.lessThanOrEqual(5);
+      expect(unprocessed2.length).to.be.lessThanOrEqual(15);
       expect(nextCursor2).to.be.equal(0);
 
       await childrenWorker.close();