fix(node): Cooperative Scheduling between startBuild and subProcess generators

docs/angular/api-node/executors/execute.md

@@ -18,6 +18,14 @@ Type: `array`
 
 Extra args when starting the app
 
+### emitSubprocessEvents
+
+Default: `false`
+
+Type: `boolean`
+
+Emit events from subprocess
+
 ### host
 
 Default: `localhost`

docs/node/api-node/executors/execute.md

@@ -18,6 +18,14 @@ Type: `array`
 
 Extra args when starting the app
 
+### emitSubprocessEvents
+
+Default: `false`
+
+Type: `boolean`
+
+Emit events from subprocess
+
 ### host
 
 Default: `localhost`

docs/react/api-node/executors/execute.md

@@ -18,6 +18,14 @@ Type: `array`
 
 Extra args when starting the app
 
+### emitSubprocessEvents
+
+Default: `false`
+
+Type: `boolean`
+
+Emit events from subprocess
+
 ### host
 
 Default: `localhost`

packages/node/src/executors/execute/execute.impl.spec.ts

@@ -19,6 +19,7 @@ import {
 describe('NodeExecuteBuilder', () => {
   let testOptions: NodeExecuteBuilderOptions;
   let context: ExecutorContext;
+  let mockSubProcess: { on: jest.Mock; exitCode: number };
 
   beforeEach(async () => {
     buildOptions = {};
@@ -34,13 +35,16 @@ describe('NodeExecuteBuilder', () => {
     );
 
     fork.mockImplementation(() => {
-      return {
-        on: (eventName, cb) => {
+      mockSubProcess = {
+        on: jest.fn().mockImplementation((eventName, cb) => {
           if (eventName === 'exit') {
-            cb();
+            mockSubProcess.exitCode = 0;
+            cb(mockSubProcess.exitCode);
           }
-        },
+        }),
+        exitCode: null,
       };
+      return mockSubProcess;
     });
 
     treeKill.mockImplementation((pid, signal, callback) => {
@@ -74,6 +78,7 @@ describe('NodeExecuteBuilder', () => {
       waitUntilTargets: [],
       host: 'localhost',
       watch: true,
+      emitSubprocessEvents: false,
     };
   });
 
@@ -83,6 +88,7 @@ describe('NodeExecuteBuilder', () => {
 
   it('should build the application and start the built file', async () => {
     for await (const event of executeExecutor(testOptions, context)) {
+      if (!('success' in event)) throw new Error('Got Unexpected event');
       expect(event.success).toEqual(true);
     }
     expect(require('@nrwl/devkit').runExecutor).toHaveBeenCalledWith(
@@ -105,7 +111,6 @@ describe('NodeExecuteBuilder', () => {
     expect(treeKill).toHaveBeenCalledTimes(0);
     expect(fork).toHaveBeenCalledTimes(1);
   });
-
   describe('--inspect', () => {
     describe('inspect', () => {
       it('should inspect the process', async () => {
@@ -215,6 +220,10 @@ describe('NodeExecuteBuilder', () => {
   });
 
   it('should log errors from killing the process', async () => {
+    (devkit.runExecutor as any).mockImplementation(function* () {
+      yield { success: true, outfile: 'outfile.js' };
+      yield { success: true, outfile: 'outfile.js' };
+    });
     treeKill.mockImplementation((pid, signal, callback) => {
       callback(new Error('Error Message'));
     });
@@ -227,6 +236,10 @@ describe('NodeExecuteBuilder', () => {
   });
 
   it('should log errors from killing the process on windows', async () => {
+    (devkit.runExecutor as any).mockImplementation(function* () {
+      yield { success: true, outfile: 'outfile.js' };
+      yield { success: true, outfile: 'outfile.js' };
+    });
     treeKill.mockImplementation((pid, signal, callback) => {
       callback([new Error('error'), '', 'Error Message']);
     });
@@ -329,4 +342,235 @@ describe('NodeExecuteBuilder', () => {
       }
     });
   });
+
+  describe('--watch', () => {
+    let mockSubProcessExit;
+    let mockSubProcessMessage;
+    let mockSubProcessError;
+
+    beforeEach(() => {
+      fork.mockImplementation(() => {
+        mockSubProcess = {
+          on: jest.fn().mockImplementation((eventName, cb) => {
+            if (eventName === 'exit') {
+              mockSubProcessExit = (exitCode: number) => {
+                cb(exitCode);
+                mockSubProcess.exitCode = exitCode;
+              };
+            }
+
+            if (eventName === 'error') {
+              mockSubProcessError = cb;
+            }
+
+            if (eventName === 'message') {
+              mockSubProcessMessage = cb;
+            }
+          }),
+          exitCode: null,
+        };
+
+        return mockSubProcess;
+      });
+    });
+
+    describe('false', () => {
+      it('Should not complete until the child process has exited', async () => {
+        const firstEvent = Symbol('first');
+        const secondEvent = Symbol('second');
+        const events = [];
+        const executorIterator = executeExecutor(
+          {
+            ...testOptions,
+            watch: false,
+          },
+          context
+        );
+
+        expect(await executorIterator.next()).toEqual({
+          done: false,
+          value: {
+            outfile: 'outfile.js',
+            success: true,
+          },
+        });
+
+        const result = await Promise.race([
+          executorIterator.next(),
+          new Promise(() => {
+            mockSubProcessExit(1);
+            events.push(firstEvent);
+          }),
+        ]);
+
+        events.push(secondEvent);
+
+        expect(result).toEqual({
+          value: undefined,
+          done: true,
+        });
+        expect(events).toEqual([firstEvent, secondEvent]);
+      });
+
+      it('When sub process emits, it should propagate when configured', async () => {
+        const expectedExitCode = 42;
+        const executorIterator = executeExecutor(
+          {
+            ...testOptions,
+            watch: false,
+            emitSubprocessEvents: true,
+          },
+          context
+        );
+
+        expect(await executorIterator.next()).toEqual({
+          done: false,
+          value: {
+            outfile: 'outfile.js',
+            success: true,
+          },
+        });
+
+        mockSubProcessExit(expectedExitCode);
+
+        expect(await executorIterator.next()).toEqual({
+          done: false,
+          value: {
+            exitCode: expectedExitCode,
+          },
+        });
+      });
+
+      it('Should propagate sub process events in expected order', async () => {
+        const expectedSubProcessEvents = [
+          {
+            message: 'hello',
+          },
+          {
+            message: 'world',
+          },
+          {
+            error: new Error('error'),
+          },
+          {
+            exitCode: 1,
+          },
+        ];
+        const expectedTotalEvents = 5;
+
+        let subProcessEvents = [];
+        let totalEvents = 0;
+
+        const doSubProcessWork = () => {
+          mockSubProcessMessage('hello');
+          mockSubProcessMessage('world');
+          mockSubProcessError(new Error('error'));
+          mockSubProcessExit(1);
+        };
+
+        for await (const event of executeExecutor(
+          {
+            ...testOptions,
+            watch: false,
+            emitSubprocessEvents: true,
+          },
+          context
+        )) {
+          if (totalEvents === 0) {
+            expect(event).toEqual({
+              outfile: 'outfile.js',
+              success: true,
+            });
+            doSubProcessWork();
+          } else {
+            subProcessEvents.push(event);
+          }
+
+          totalEvents += 1;
+        }
+
+        expect(subProcessEvents).toEqual(expectedSubProcessEvents);
+        expect(totalEvents).toEqual(expectedTotalEvents);
+      });
+    });
+
+    describe('true', () => {
+      beforeEach(() => {
+        (devkit.runExecutor as any).mockImplementation(async function* () {
+          yield { success: true, outfile: 'outfile.js' };
+          await new Promise((resolve) => setTimeout(resolve, 1));
+          yield { success: true, outfile: 'outfile.js' };
+        });
+      });
+
+      it('Should not block build events with process events', async () => {
+        const executorIterator = executeExecutor(
+          {
+            ...testOptions,
+            watch: true,
+          },
+          context
+        );
+
+        expect(await executorIterator.next()).toEqual({
+          done: false,
+          value: {
+            outfile: 'outfile.js',
+            success: true,
+          },
+        });
+
+        // sub process has not exited yet, but we'll receive a new build event
+
+        expect(await executorIterator.next()).toEqual({
+          done: false,
+          value: {
+            outfile: 'outfile.js',
+            success: true,
+          },
+        });
+
+        // there's no more build events, but process still needs to exit
+
+        const result = await Promise.race([
+          executorIterator.next(),
+          new Promise(() => {
+            // this promise intentionally doesn't resolve
+            mockSubProcessExit();
+          }),
+        ]);
+
+        expect(result).toEqual({
+          value: undefined,
+          done: true,
+        });
+      });
+
+      it('Yield build and process events cooperatively', async () => {
+        const expectedEvents = [
+          { outfile: 'outfile.js', success: true },
+          { exitCode: 1 },
+          { outfile: 'outfile.js', success: true },
+          { exitCode: 1 },
+        ];
+        const events = [];
+        for await (const event of executeExecutor(
+          {
+            ...testOptions,
+            watch: true,
+            emitSubprocessEvents: true,
+          },
+          context
+        )) {
+          if ('outfile' in event) {
+            mockSubProcessExit(1);
+          }
+
+          events.push(event);
+        }
+
+        expect(events).toEqual(expectedEvents);
+      });
+    });
+  });
 });