fix(core): pass-through ipc messages to and from task processes (#11744)

FrozenPandaz · web-flow · commit 8b02670a3962 · 2022-08-26T17:49:34.000-04:00
diff --git a/packages/nx/src/tasks-runner/forked-process-task-runner.ts b/packages/nx/src/tasks-runner/forked-process-task-runner.ts
@@ -1,6 +1,6 @@
 import { readFileSync, writeFileSync } from 'fs';
 import * as dotenv from 'dotenv';
-import { ChildProcess, fork } from 'child_process';
+import { ChildProcess, fork, Serializable } from 'child_process';
 import { workspaceRoot } from '../utils/workspace-root';
 import { DefaultTasksRunnerOptions } from './default-tasks-runner';
 import { output } from '../utils/output';
@@ -30,7 +30,7 @@ export class ForkedProcessTaskRunner {
   private processes = new Set<ChildProcess>();
 
   constructor(private readonly options: DefaultTasksRunnerOptions) {
-    this.setupOnProcessExitListener();
+    this.setupProcessEventListeners();
   }
 
   // TODO: vsavkin delegate terminal output printing
@@ -79,6 +79,14 @@ export class ForkedProcessTaskRunner {
           switch (message.type) {
             case BatchMessageType.Complete: {
               res(message.results);
+              break;
+            }
+            case BatchMessageType.Tasks: {
+              break;
+            }
+            default: {
+              // Re-emit any non-batch messages from the task process
+              process.send(message);
             }
           }
         });
@@ -126,6 +134,12 @@ export class ForkedProcessTaskRunner {
           ),
         });
         this.processes.add(p);
+
+        // Re-emit any messages from the task process
+        p.on('message', (message) => {
+          process.send(message);
+        });
+
         let out = [];
         let outWithErr = [];
         p.stdout.on('data', (chunk) => {
@@ -199,6 +213,12 @@ export class ForkedProcessTaskRunner {
           ),
         });
         this.processes.add(p);
+
+        // Re-emit any messages from the task process
+        p.on('message', (message) => {
+          process.send(message);
+        });
+
         p.on('exit', (code, signal) => {
           if (code === null) code = this.signalToCode(signal);
           // we didn't print any output as we were running the command
@@ -369,7 +389,15 @@ export class ForkedProcessTaskRunner {
     return 128;
   }
 
-  private setupOnProcessExitListener() {
+  private setupProcessEventListeners() {
+    // When the nx process gets a message, it will be sent into the task's process
+    process.on('message', (message: Serializable) => {
+      this.processes.forEach((p) => {
+        p.send(message);
+      });
+    });
+
+    // Terminate any task processes on exit
     process.on('SIGINT', () => {
       this.processes.forEach((p) => {
         p.kill('SIGTERM');
