Repository: taskforcesh/bullmq
Pull request #575 (2021-06-02 00:42:04)

Feat getDependencies with pagination
One job could have many dependencies that are unprocessed, in order to not get all at once, could be nice to have a wait to get those using sscan, this could be helpful to use it in a ui. Also I added getProcessedDependencies to have pagination on those values using hscan, same purpose. Note: this pr is still in progress

manast: Functionality wise this looks good. But I think it would be better to put these two methods inside current ```getDependencies``` in order to keep the API surface more compact.
Maybe you can add more params to getDependencies so that you can control if you want to read only processed, unprocessed and optional pagination?
roggervalf: > Maybe you can add more params to getDependencies so that you can control if you want to read only processed, unprocessed and optional pagination?

Let me think how to address this, also we have a separate method to only return the processed values (getChildrenValues), so this method should also have same feature
roggervalf: > also we have a separate method to only return the processed values (getChildrenValues), so this method should also have same feature

Nevermind, to add this feature using hscan I would need to return the cursor, this should be a breaking change in this method :/
manast: I think it is possible to do it without a breaking change. If no opts are provided return all data as in current implementation, if opts is provided then use the new cursor based code.
roggervalf: > I think it is possible to do it without a breaking change. If no opts are provided return all data as in current implementation, if opts is provided then use the new cursor based code.

Sure no problem, I updated it base on your suggestion
manast: Nice work!

Patch:
diff --git a/src/classes/job.ts b/src/classes/job.ts
index a133782511..0d12572d21 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -54,6 +54,17 @@ export interface MoveToChildrenOpts {
   };
 }
 
+interface DependenciesOpts {
+  processed?: {
+    cursor?: number;
+    count?: number;
+  };
+  unprocessed?: {
+    cursor?: number;
+    count?: number;
+  };
+}
+
 export class Job<T = any, R = any, N extends string = string> {
   /**
    * The progress a job has performed so far.
@@ -547,27 +558,99 @@ export class Job<T = any, R = any, N extends string = string> {
    *
    * @returns dependencies separated by processed and unprocessed.
    */
-  async getDependencies() {
+  async getDependencies(
+    opts: DependenciesOpts = {},
+  ): Promise<{
+    nextProcessedCursor?: number;
+    processed?: Record<string, any>;
+    nextUnprocessedCursor?: number;
+    unprocessed?: 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`));
+
+      const [[err1, processed], [err2, unprocessed]] = (await multi.exec()) as [
+        [null | Error, { [jobKey: string]: string }],
+        [null | Error, string[]],
+      ];
+
+      const transformedProcessed = Object.entries(processed).reduce(
+        (accumulator, [key, value]) => {
+          return { ...accumulator, [key]: JSON.parse(value) };
+        },
+        {},
+      );
+
+      return { processed: transformedProcessed, unprocessed };
+    } else {
+      const defaultOpts = {
+        cursor: 0,
+        count: 20,
+      };
 
-    await multi.hgetall(this.toKey(`${this.id}:processed`));
-    await multi.smembers(this.toKey(`${this.id}:dependencies`));
+      if (opts.processed) {
+        const processedOpts = Object.assign(defaultOpts, opts.processed);
+        await multi.hscan(
+          this.toKey(`${this.id}:processed`),
+          processedOpts.cursor,
+          'COUNT',
+          processedOpts.count,
+        );
+      }
 
-    const [[err1, processed], [err2, unprocessed]] = (await multi.exec()) as [
-      [null | Error, { [jobKey: string]: string }],
-      [null | Error, string[]],
-    ];
+      if (opts.unprocessed) {
+        const unprocessedOpts = Object.assign(defaultOpts, opts.unprocessed);
+        await multi.sscan(
+          this.toKey(`${this.id}:dependencies`),
+          unprocessedOpts.cursor,
+          'COUNT',
+          unprocessedOpts.count,
+        );
+      }
 
-    const transformedProcessed = Object.entries(processed).reduce(
-      (accumulator, [key, value]) => {
-        return { ...accumulator, [key]: JSON.parse(value) };
-      },
-      {},
-    );
+      const [result1, result2] = await multi.exec();
+
+      const [processedCursor, processed = []] = opts.processed
+        ? result1[1]
+        : [];
+      const [unprocessedCursor, unprocessed = []] = opts.unprocessed
+        ? opts.processed
+          ? result2[1]
+          : result1[1]
+        : [];
+
+      const transformedProcessed = processed.reduce(
+        (
+          accumulator: Record<string, any>,
+          currentValue: string,
+          index: number,
+        ) => {
+          if (index % 2) {
+            return {
+              ...accumulator,
+              [processed[index - 1]]: JSON.parse(currentValue),
+            };
+          }
+          return accumulator;
+        },
+        {},
+      );
 
-    return { processed: transformedProcessed, unprocessed };
+      return {
+        ...(processedCursor
+          ? {
+              processed: transformedProcessed,
+              nextProcessedCursor: Number(processedCursor),
+            }
+          : {}),
+        ...(unprocessedCursor
+          ? { unprocessed, nextUnprocessedCursor: Number(unprocessedCursor) }
+          : {}),
+      };
+    }
   }
 
   /**


Test patch:
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
index 59a0dbd7dc..38ce57ffe4 100644
--- a/src/test/test_flow.ts
+++ b/src/test/test_flow.ts
@@ -47,7 +47,10 @@ describe('flows', () => {
     const processingParent = new Promise<void>((resolve, reject) => [
       (parentProcessor = async (job: Job) => {
         try {
-          const { processed } = await job.getDependencies();
+          const { processed, nextProcessedCursor } = await job.getDependencies({
+            processed: {},
+          });
+          expect(nextProcessedCursor).to.be.equal(0);
           expect(Object.keys(processed)).to.have.length(3);
 
           const childrenValues = await job.getChildrenValues();
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index ab8186081a..8466cdebfb 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1886,6 +1886,68 @@ describe('workers', function() {
       await removeAllQueueData(new IORedis(), parentQueueName);
     });
 
+    it('should get paginated unprocessed dependencies keys', async () => {
+      const value = { bar: 'something' };
+      const parentToken = 'parent-token';
+
+      const parentQueueName = `parent-queue-${v4()}`;
+
+      const parentQueue = new Queue(parentQueueName);
+      const parentWorker = new Worker(parentQueueName);
+      const childrenWorker = new Worker(queueName);
+
+      const data = { foo: 'bar' };
+      await Job.create(parentQueue, 'parent', data);
+      const parent = (await parentWorker.getNextJob(parentToken)) as Job;
+      const currentState = await parent.getState();
+
+      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,
+            },
+          },
+        ),
+      );
+      const {
+        nextUnprocessedCursor: nextCursor1,
+        unprocessed: unprocessed1,
+      } = await parent.getDependencies({
+        unprocessed: {
+          cursor: 0,
+          count: 10,
+        },
+      });
+
+      expect(unprocessed1.length).to.be.greaterThanOrEqual(10);
+
+      const {
+        nextUnprocessedCursor: nextCursor2,
+        unprocessed: unprocessed2,
+      } = await parent.getDependencies({
+        unprocessed: {
+          cursor: nextCursor1,
+          count: 10,
+        },
+      });
+
+      expect(unprocessed2.length).to.be.lessThanOrEqual(5);
+      expect(nextCursor2).to.be.equal(0);
+
+      await childrenWorker.close();
+      await parentWorker.close();
+
+      await parentQueue.close();
+      await removeAllQueueData(new IORedis(), parentQueueName);
+    });
+
     it('should allow to fail jobs manually', async () => {
       const worker = new Worker(queueName);
       const token = 'my-token';