feat(sandboxed-job): add support for getChildrenValues

src/classes/child-processor.ts

@@ -1,5 +1,11 @@
 import { promisify } from 'util';
-import { JobJson, ParentCommand, SandboxedJob } from '../interfaces';
+import {
+  ChildCommand,
+  JobJson,
+  ParentCommand,
+  ParentMessage,
+  SandboxedJob,
+} from '../interfaces';
 import { childSend } from '../utils';
 
 enum ChildStatus {
@@ -187,5 +193,41 @@ function wrapJob(job: JobJson): SandboxedJob {
         value: data,
       });
     },
+    /*
+     * Emulate the real job `getChildrenValues` function.
+     */
+    getChildrenValues: async <CT = any>(): Promise<{
+      [jobKey: string]: CT;
+    }> => {
+      let msgHandler: any;
+
+      const done = new Promise<{
+        [jobKey: string]: CT;
+      }>(resolve => {
+        msgHandler = async (msg: ParentMessage) => {
+          switch (msg.cmd) {
+            case ChildCommand.GetChildrenValues: {
+              resolve(msg.value);
+              break;
+            }
+          }
+        };
+      });
+
+      process.on('message', msgHandler);
+
+      try {
+        childSend(process, {
+          cmd: ParentCommand.GetChildrenValues,
+        });
+
+        const childrenValues = await done;
+        process.removeListener('message', msgHandler);
+
+        return childrenValues;
+      } catch (err) {
+        // TODO: how to handle this error? Is this try/catch needed at all?
+      }
+    },
   };
 }

src/classes/sandbox.ts

@@ -38,6 +38,16 @@ const sandbox = <T, R, N extends string>(
             break;
           case ParentCommand.Update:
             await job.update(msg.value);
+            break;
+          case ParentCommand.GetChildrenValues:
+            {
+              const value = await job.getChildrenValues();
+              await parentSend(child, {
+                cmd: ChildCommand.GetChildrenValues,
+                value,
+              });
+            }
+
             break;
         }
       };

src/interfaces/child-command.ts

@@ -2,4 +2,5 @@ export enum ChildCommand {
   Init,
   Start,
   Stop,
+  GetChildrenValues,
 }

src/interfaces/parent-command.ts

@@ -7,4 +7,5 @@ export enum ParentCommand {
   Log,
   Progress,
   Update,
+  GetChildrenValues,
 }

src/interfaces/sandboxed-job.ts

@@ -14,5 +14,6 @@ export interface SandboxedJob<T = any, R = any>
   updateProgress: (value: object | number) => Promise<void>;
   log: (row: any) => void;
   update: (data: any) => Promise<void>;
+  getChildrenValues: <CT = any>() => Promise<{ [jobKey: string]: CT }>;
   returnValue: R;
 }

tests/fixtures/fixture_processor_getChildrenValues.js

@@ -0,0 +1,11 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+const delay = require('./delay');
+
+module.exports = async function (job) {
+  return job.getChildrenValues();
+};

tests/fixtures/fixture_processor_getChildrenValues_child.js

@@ -0,0 +1,11 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+const delay = require('./delay');
+
+module.exports = function (job) {
+  return { childResult: 'bar' };
+};

tests/test_sandboxed_process.ts

@@ -513,6 +513,53 @@ describe('sandboxed process', () => {
     await flow.close();
   });
 
+  it('can get children values by calling getChildrenValues', async () => {
+    const childJobId = 'child-job-id';
+    const childProcessFile =
+      __dirname + '/fixtures/fixture_processor_getChildrenValues_child.js';
+    const parentProcessFile =
+      __dirname + '/fixtures/fixture_processor_getChildrenValues.js';
+    const parentQueueName = `parent-queue-${v4()}`;
+
+    const parentWorker = new Worker(parentQueueName, parentProcessFile, {
+      connection,
+      drainDelay: 1,
+    });
+
+    const childWorker = new Worker(queueName, childProcessFile, {
+      connection,
+      drainDelay: 1,
+    });
+
+    const parentCompleting = new Promise<void>((resolve, reject) => {
+      parentWorker.on('completed', async (job: Job, value: any) => {
+        try {
+          expect(value).to.be.eql({
+            [`bull:${queueName}:${childJobId}`]: { childResult: 'bar' },
+          });
+          await parentWorker.close();
+          resolve();
+        } catch (err) {
+          await parentWorker.close();
+          reject(err);
+        }
+      });
+    });
+
+    const flow = new FlowProducer({ connection });
+    await flow.add({
+      name: 'parent-job',
+      queueName: parentQueueName,
+      opts: { jobId: 'job-id' },
+      children: [{ name: 'child-job', queueName, opts: { jobId: childJobId } }],
+    });
+
+    await parentCompleting;
+    await parentWorker.close();
+    await childWorker.close();
+    await flow.close();
+  });
+
   it('should process and fail', async () => {
     const processFile = __dirname + '/fixtures/fixture_processor_fail.js';