watch: enable passthrough ipc in watch mode

znewsham · targos · commit e895f7cf3295 · 2024-05-11T18:55:49.000+02:00
PR-URL: #50890 Reviewed-By: Moshe Atlow <moshe@atlow.co.il> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
diff --git a/lib/internal/main/watch_mode.js b/lib/internal/main/watch_mode.js
@@ -81,6 +81,7 @@ function start() {
       process.stdout.write(`${red}Failed running ${kCommandStr}${white}\n`);
     }
   });
+  return child;
 }
 
 async function killAndWait(signal = kKillSignal, force = false) {
@@ -113,34 +114,43 @@ function reportGracefulTermination() {
   };
 }
 
-async function stop() {
+async function stop(child) {
+  // Without this line, the child process is still able to receive IPC, but is unable to send additional messages
+  watcher.destroyIPC(child);
   watcher.clearFileFilters();
   const clearGraceReport = reportGracefulTermination();
   await killAndWait();
   clearGraceReport();
 }
 
 let restarting = false;
-async function restart() {
+async function restart(child) {
   if (restarting) return;
   restarting = true;
   try {
     if (!kPreserveOutput) process.stdout.write(clear);
     process.stdout.write(`${green}Restarting ${kCommandStr}${white}\n`);
-    await stop();
-    start();
+    await stop(child);
+    return start();
   } finally {
     restarting = false;
   }
 }
 
-start();
-watcher
-  .on('changed', restart)
-  .on('error', (error) => {
-    watcher.off('changed', restart);
-    triggerUncaughtException(error, true /* fromPromise */);
-  });
+async function init() {
+  let child = start();
+  const restartChild = async () => {
+    child = await restart(child);
+  };
+  watcher
+    .on('changed', restartChild)
+    .on('error', (error) => {
+      watcher.off('changed', restartChild);
+      triggerUncaughtException(error, true /* fromPromise */);
+    });
+}
+
+init();
 
 // Exiting gracefully to avoid stdout/stderr getting written after
 // parent process is killed.
diff --git a/lib/internal/watch_mode/files_watcher.js b/lib/internal/watch_mode/files_watcher.js
@@ -3,8 +3,10 @@
 const {
   ArrayIsArray,
   ArrayPrototypeForEach,
+  Boolean,
   SafeMap,
   SafeSet,
+  SafeWeakMap,
   StringPrototypeStartsWith,
 } = primordials;
 
@@ -31,6 +33,8 @@ class FilesWatcher extends EventEmitter {
   #debounce;
   #mode;
   #signal;
+  #passthroughIPC = false;
+  #ipcHandlers = new SafeWeakMap();
 
   constructor({ debounce = 200, mode = 'filter', signal } = kEmptyObject) {
     super({ __proto__: null, captureRejections: true });
@@ -40,6 +44,7 @@ class FilesWatcher extends EventEmitter {
     this.#debounce = debounce;
     this.#mode = mode;
     this.#signal = signal;
+    this.#passthroughIPC = Boolean(process.send);
 
     if (signal) {
       addAbortListener(signal, () => this.clear());
@@ -128,7 +133,31 @@ class FilesWatcher extends EventEmitter {
       this.#ownerDependencies.set(owner, dependencies);
     }
   }
+
+
+  #setupIPC(child) {
+    const handlers = {
+      __proto__: null,
+      parentToChild: (message) => child.send(message),
+      childToParent: (message) => process.send(message),
+    };
+    this.#ipcHandlers.set(child, handlers);
+    process.on('message', handlers.parentToChild);
+    child.on('message', handlers.childToParent);
+  }
+
+  destroyIPC(child) {
+    const handlers = this.#ipcHandlers.get(child);
+    if (this.#passthroughIPC && handlers !== undefined) {
+      process.off('message', handlers.parentToChild);
+      child.off('message', handlers.childToParent);
+    }
+  }
+
   watchChildProcessModules(child, key = null) {
+    if (this.#passthroughIPC) {
+      this.#setupIPC(child);
+    }
     if (this.#mode !== 'filter') {
       return;
     }
diff --git a/test/sequential/test-watch-mode.mjs b/test/sequential/test-watch-mode.mjs
@@ -8,6 +8,7 @@ import { spawn } from 'node:child_process';
 import { writeFileSync, readFileSync, mkdirSync } from 'node:fs';
 import { inspect } from 'node:util';
 import { pathToFileURL } from 'node:url';
+import { once } from 'node:events';
 import { createInterface } from 'node:readline';
 
 if (common.isIBMi)
@@ -574,4 +575,84 @@ console.log(values.random);
       `Completed running ${inspect(file)}`,
     ]);
   });
+
+  it('should pass IPC messages from a spawning parent to the child and back', async () => {
+    const file = createTmpFile(`console.log('running');
+process.on('message', (message) => {
+  if (message === 'exit') {
+    process.exit(0);
+  } else {
+    console.log('Received:', message);
+    process.send(message);
+  }
+})`);
+
+    const child = spawn(
+      execPath,
+      [
+        '--watch',
+        '--no-warnings',
+        file,
+      ],
+      {
+        encoding: 'utf8',
+        stdio: ['pipe', 'pipe', 'pipe', 'ipc'],
+      },
+    );
+
+    let stderr = '';
+    let stdout = '';
+
+    child.stdout.on('data', (data) => stdout += data);
+    child.stderr.on('data', (data) => stderr += data);
+    async function waitForEcho(msg) {
+      const receivedPromise = new Promise((resolve) => {
+        const fn = (message) => {
+          if (message === msg) {
+            child.off('message', fn);
+            resolve();
+          }
+        };
+        child.on('message', fn);
+      });
+      child.send(msg);
+      await receivedPromise;
+    }
+
+    async function waitForText(text) {
+      const seenPromise = new Promise((resolve) => {
+        const fn = (data) => {
+          if (data.toString().includes(text)) {
+            resolve();
+            child.stdout.off('data', fn);
+          }
+        };
+        child.stdout.on('data', fn);
+      });
+      await seenPromise;
+    }
+
+    await waitForText('running');
+    await waitForEcho('first message');
+    const stopRestarts = restart(file);
+    await waitForText('running');
+    stopRestarts();
+    await waitForEcho('second message');
+    const exitedPromise = once(child, 'exit');
+    child.send('exit');
+    await waitForText('Completed');
+    child.disconnect();
+    child.kill();
+    await exitedPromise;
+    assert.strictEqual(stderr, '');
+    const lines = stdout.split(/\r?\n/).filter(Boolean);
+    assert.deepStrictEqual(lines, [
+      'running',
+      'Received: first message',
+      `Restarting ${inspect(file)}`,
+      'running',
+      'Received: second message',
+      `Completed running ${inspect(file)}`,
+    ]);
+  });
 });
