test: try getChildrenValues in sandbox

src/classes/child-processor.ts

@@ -1,5 +1,5 @@
-import { ParentCommand } from '../enums';
-import { SandboxedJob } from '../interfaces';
+import { ChildCommand, ParentCommand } from '../enums';
+import { SandboxedJob, ParentMessage } from '../interfaces';
 import { JobJsonSandbox } from '../types';
 import { errorToJSON } from '../utils';
 
@@ -160,6 +160,43 @@ export class ChildProcessor {
         });
         wrappedJob.data = 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, reject) => {
+          msgHandler = async (msg: ParentMessage) => {
+            switch (msg.cmd) {
+              case ChildCommand.GetChildrenValues: {
+                resolve(msg.value);
+                break;
+              }
+              case ChildCommand.GetChildrenValuesError: {
+                const err = new Error();
+                Object.assign(err, msg.value);
+                reject(err);
+                break;
+              }
+            }
+          };
+        });
+
+        process.on('message', msgHandler);
+        await send({
+          cmd: ParentCommand.GetChildrenValues,
+        });
+
+        const childrenValues = await done;
+        process.removeListener('message', msgHandler);
+
+        return childrenValues;
+      },
     };
 
     return wrappedJob;

src/classes/sandbox.ts

@@ -1,3 +1,4 @@
+import { errorToJSON } from '../utils';
 import { ChildCommand, ParentCommand } from '../enums';
 import { ChildMessage } from '../interfaces';
 import { ChildPool } from './child-pool';
@@ -43,6 +44,23 @@ const sandbox = <T, R, N extends string>(
           case ParentCommand.Update:
             await job.updateData(msg.value);
             break;
+          case ParentCommand.GetChildrenValues: {
+            try {
+              const value = await job.getChildrenValues();
+
+              await child.send({
+                cmd: ChildCommand.GetChildrenValues,
+                value,
+              });
+              break;
+            } catch (error) {
+              await child.send({
+                cmd: ChildCommand.GetChildrenValuesError,
+                value: errorToJSON(error),
+              });
+            }
+            break;
+          }
         }
       };
 

src/enums/child-command.ts

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

src/enums/parent-command.ts

@@ -8,4 +8,5 @@ export enum ParentCommand {
   MoveToDelayed,
   Progress,
   Update,
+  GetChildrenValues,
 }

src/interfaces/sandboxed-job.ts

@@ -7,6 +7,7 @@ export interface SandboxedJob<T = any, R = any>
   extends Omit<JobJsonSandbox, 'data' | 'opts' | 'returnValue'> {
   data: T;
   opts: JobsOptions;
+  getChildrenValues: <CT = any>() => Promise<{ [jobKey: string]: CT }>;
   moveToDelayed: (timestamp: number, token?: string) => Promise<void>;
   log: (row: any) => void;
   updateData: (data: any) => Promise<void>;

tests/fixtures/fixture_processor_getChildrenValues.js

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

tests/fixtures/fixture_processor_getChildrenValues_child.js

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

tests/test_sandboxed_process.ts

@@ -576,6 +576,59 @@ function sandboxProcessTests(
       await worker.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,
+    prefix,
+    drainDelay: 1,
+    useWorkerThreads,
+  });
+
+  const childWorker = new Worker(queueName, childProcessFile, {
+    connection,
+    prefix,
+    drainDelay: 1,
+    useWorkerThreads,
+  });
+
+  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, prefix });
+  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 move to delayed', async () => {
       const processFile =
         __dirname + '/fixtures/fixture_processor_move_to_delayed.js';