feat(js): improve @nx/js:node executor to be more resilient to many f…

…ile change events

packages/js/executors.json

@@ -12,7 +12,7 @@
       "description": "Build a project using SWC."
     },
     "node": {
-      "implementation": "./src/executors/node/node.impl",
+      "implementation": "./src/executors/node/node-v2.impl",
       "schema": "./src/executors/node/schema.json",
       "description": "Execute a Node application."
     }

packages/js/src/executors/node/node-v2.impl.ts

@@ -0,0 +1,213 @@
+import * as chalk from 'chalk';
+import { ChildProcess, exec, spawn } from 'child_process';
+import { ExecutorContext, logger, parseTargetString } from '@nx/devkit';
+import { daemonClient } from 'nx/src/daemon/client/client';
+import { randomUUID } from 'crypto';
+import { join } from 'path';
+
+import { InspectType, NodeExecutorOptions } from './schema';
+import { createAsyncIterable } from '@nx/devkit/src/utils/async-iterable';
+
+interface ActiveTask {
+  id: string;
+  killed: boolean;
+  promise: Promise<void>;
+  childProcess: null | ChildProcess;
+  start: () => Promise<void>;
+  stop: () => Promise<void>;
+}
+
+function debounce(fn: () => void, wait: number) {
+  let timeoutId: NodeJS.Timeout;
+  return () => {
+    clearTimeout(timeoutId);
+    timeoutId = setTimeout(fn, wait);
+  };
+}
+
+export async function* nodeExecutor(
+  options: NodeExecutorOptions,
+  context: ExecutorContext
+) {
+  const project = context.projectGraph.nodes[context.projectName];
+  const buildTarget = parseTargetString(
+    options.buildTarget,
+    context.projectGraph
+  );
+
+  const buildOptions = project.data.targets[buildTarget.target]?.options;
+  if (!buildOptions) {
+    throw new Error(
+      `Cannot find build target ${chalk.bold(
+        options.buildTarget
+      )} for project ${chalk.bold(context.projectName)}`
+    );
+  }
+
+  const outputPath = join(
+    context.root,
+    buildOptions.outputPath,
+    buildOptions.outputFileName ?? 'main.js'
+  );
+
+  const tasks: ActiveTask[] = [];
+  let currentTask: ActiveTask = null;
+
+  yield* createAsyncIterable<{ success: boolean }>(
+    async ({ done, next, error }) => {
+      const processQueue = async () => {
+        if (tasks.length === 0) return;
+
+        const previousTask = currentTask;
+        const task = tasks.shift();
+        currentTask = task;
+        await previousTask?.stop();
+        await task.start();
+      };
+
+      const debouncedProcessQueue = debounce(processQueue, 1_000);
+
+      const addToQueue = async () => {
+        const task: ActiveTask = {
+          id: randomUUID(),
+          killed: false,
+          childProcess: null,
+          promise: null,
+          start: async () => {
+            // Run the build
+            task.promise = new Promise<void>(async (resolve, reject) => {
+              task.childProcess = exec(
+                `npx nx run ${context.projectName}:${buildTarget.target}${
+                  buildTarget.configuration
+                    ? `:${buildTarget.configuration}`
+                    : ''
+                }`,
+                {
+                  cwd: context.root,
+                },
+                (error, stdout, stderr) => {
+                  if (error) {
+                    // If process receive termination signal, it means another task has started.
+                    if (error.signal === 'SIGTERM') {
+                      resolve();
+                    } else {
+                      if (options.watch) {
+                        logger.error(
+                          `Build failed, waiting for changes to restart...`
+                        );
+                      } else {
+                        logger.error(`Build failed. See above for errors.`);
+                      }
+                      logger.error(error);
+                      reject();
+                    }
+                  } else {
+                    resolve();
+                  }
+                }
+              );
+            });
+
+            // Wait for build to finish
+            await task.promise;
+
+            // Task may have been stopped due to another running task
+            if (task.killed) return;
+
+            // Run the program
+            task.promise = new Promise<void>((resolve, reject) => {
+              task.childProcess = spawn(
+                'node',
+                [...getExecArgv(options), outputPath],
+                {
+                  stdio: 'inherit',
+                }
+              );
+
+              task.childProcess.once('exit', (code) => {
+                if (options.watch && !task.killed) {
+                  logger.info(
+                    `process exited with code ${code}, waiting for changes to restart...`
+                  );
+                }
+                if (!options.watch) done();
+                resolve();
+              });
+
+              next({ success: true });
+            });
+          },
+          stop: async () => {
+            task.killed = true;
+            // Request termination and wait for process to finish gracefully.
+            // NOTE: `childProcess` may not have been set yet if the task did not have a chance to start.
+            // e.g. multiple file change events in a short time (like git checkout).
+            task.childProcess?.kill('SIGTERM');
+            await task.promise;
+          },
+        };
+
+        tasks.push(task);
+      };
+
+      const stopWatch = await daemonClient.registerFileWatcher(
+        {
+          watchProjects: [context.projectName],
+          includeDependentProjects: true,
+        },
+        async (err, data) => {
+          if (err === 'closed') {
+            logger.error(`Watch error: Daemon closed the connection`);
+            process.exit(1);
+          } else if (err) {
+            logger.error(`Watch error: ${err?.message ?? 'Unknown'}`);
+          } else {
+            await addToQueue();
+            await debouncedProcessQueue();
+          }
+        }
+      );
+
+      const stopAllTasks = () => {
+        for (const task of tasks) {
+          task.stop();
+        }
+      };
+
+      process.on('SIGTERM', async () => {
+        stopWatch();
+        stopAllTasks();
+        process.exit(128 + 15);
+      });
+      process.on('SIGINT', async () => {
+        stopWatch();
+        stopAllTasks();
+        process.exit(128 + 2);
+      });
+      process.on('SIGHUP', async () => {
+        stopWatch();
+        stopAllTasks();
+        process.exit(128 + 1);
+      });
+
+      await addToQueue();
+      await processQueue();
+    }
+  );
+}
+
+function getExecArgv(options: NodeExecutorOptions) {
+  const args = [...options.runtimeArgs];
+
+  if (options.inspect === true) {
+    options.inspect = InspectType.Inspect;
+  }
+
+  if (options.inspect) {
+    args.push(`--${options.inspect}=${options.host}:${options.port}`);
+  }
+
+  return args;
+}
+
+export default nodeExecutor;

packages/js/src/executors/node/schema.json

@@ -67,6 +67,11 @@
       "type": "boolean",
       "description": "Enable re-building when files change.",
       "default": true
+    },
+    "debounce": {
+      "type": "number",
+      "description": "Delay in milliseconds to wait before restarting. Useful to batch multiple file changes events together. Set to zero (0) to disable.",
+      "default": 1000
     }
   },
   "additionalProperties": false,