feat: add features to stdin commands

CHANGELOG.md

@@ -4,6 +4,9 @@
 ### Bug Fixes
 
 - Fix `pm2 serve` returning 403 Forbidden on Windows — traversal guard used hardcoded `/` separator #6109
+- Silence cluster worker `error` events to prevent boot crashes
+- Wrap `process.send` before BPM injection to avoid send-on-disconnected during boot
+- BPM IPC transport: log instead of `process.exit(1)` on disconnected send
 
 
 ## 7.0.1

examples/interact-via-stdin/README.md

@@ -11,3 +11,15 @@ Then to attach to it:
 ```
 $ pm2 attach 0
 ```
+
+Or:
+
+```
+$ pm2 logs --attach-input
+```
+
+Then send a message (e.g., *Lorem Ipsum*) to the app:
+
+```
+> 0 Lorem Ipsum
+```

lib/API/Extra.js

@@ -310,74 +310,95 @@ module.exports = function(CLI) {
   }
 
   /**
-   * Description
+   * Send a line to stdin of a process
    * @method sendLineToStdin
    */
-  CLI.prototype.sendLineToStdin = function(pm_id, line, separator, cb) {
+  CLI.prototype.sendLineToStdin = function(id, line, separator, cb) {
     var that = this;
 
     if (!cb && typeof(separator) == 'function') {
       cb = separator;
       separator = null;
     }
 
-    var packet = {
-      pm_id : pm_id,
-      line : line + (separator || '\n')
-    };
+    function handleError(err) {
+      Common.printError(cst.PREFIX_MSG_ERR + err);
+      return cb ? cb(Common.retErr(err)) : that.exitCli(cst.ERROR_EXIT);
+    }
 
-    that.Client.executeRemote('sendLineToStdin', packet, function(err, res) {
-      if (err) {
-        Common.printError(cst.PREFIX_MSG_ERR + err);
-        return cb ? cb(Common.retErr(err)) : that.exitCli(cst.ERROR_EXIT);
-      }
-      return cb ? cb(null, res) : that.speedList();
-    });
+    function sendLine(pm_id) {
+      var packet = {
+        pm_id : pm_id,
+        line : line + (separator || '\n')
+      };
+
+      that.Client.executeRemote('sendLineToStdin', packet, function(err, res) {
+        if (err) return handleError(err);
+        return cb ? cb(null, res) : that.speedList();
+      });
+    }
+
+    if (isNaN(id)) {
+      that.Client.getUniqueProcessIdByName(id, function(err, pm_id) {
+        if (err) return handleError(err);
+        sendLine(pm_id);
+      });
+      return;
+    }
+    sendLine(id);
   };
 
   /**
-   * Description
-   * @method attachToProcess
+   * Attach to a process stdin and stdout
+   * @method attach
    */
-  CLI.prototype.attach = function(pm_id, separator, cb) {
+  CLI.prototype.attach = function(id, separator, cb) {
     var that = this;
     var readline = require('readline');
 
-    if (isNaN(pm_id)) {
-      Common.printError('pm_id must be a process number (not a process name)');
-      return cb ? cb(Common.retErr('pm_id must be number')) : that.exitCli(cst.ERROR_EXIT);
-    }
-
     if (typeof(separator) == 'function') {
       cb = separator;
       separator = null;
     }
 
-    var rl = readline.createInterface({
-      input: process.stdin,
-      output: process.stdout
-    });
+    function handleError(err) {
+      Common.printError(cst.PREFIX_MSG_ERR + err);
+      return cb ? cb(Common.retErr(err)) : that.exitCli(cst.ERROR_EXIT);
+    }
 
-    rl.on('close', function() {
-      return cb ? cb() : that.exitCli(cst.SUCCESS_EXIT);
-    });
+    function attachProcess(pm_id) {
+      var rl = readline.createInterface({
+        input: process.stdin,
+        output: process.stdout
+      });
 
-    that.Client.launchBus(function(err, bus, socket) {
-      if (err) {
-        Common.printError(err);
-        return cb ? cb(Common.retErr(err)) : that.exitCli(cst.ERROR_EXIT);
-      }
+      rl.on('close', function() {
+        return cb ? cb() : that.exitCli(cst.SUCCESS_EXIT);
+      });
+
+      that.Client.launchBus(function(err, bus, socket) {
+        if (err) return handleError(err);
 
-      bus.on('log:*', function(type, packet) {
-        if (packet.process.pm_id !== parseInt(pm_id))
-          return;
-        process.stdout.write(packet.data);
+        bus.on('log:*', function(type, packet) {
+          if (packet.process.pm_id !== parseInt(pm_id))
+            return;
+          process.stdout.write(packet.data);
+        });
       });
-    });
 
-    rl.on('line', function(line) {
-      that.sendLineToStdin(pm_id, line, separator, function() {});
-    });
+      rl.on('line', function(line) {
+        that.sendLineToStdin(pm_id, line, separator, function() {});
+      });
+    }
+
+    if (isNaN(id)) {
+      that.Client.getUniqueProcessIdByName(id, function(err, pm_id) {
+        if (err) return handleError(err);
+        attachProcess(pm_id);
+      });
+      return;
+    }
+    attachProcess(id);
   };
 
   /**
@@ -762,4 +783,33 @@ module.exports = function(CLI) {
       that.exitCli(cst.SUCCESS_EXIT);
     });
   };
+
+  /**
+   * Attach stdin to the client. You could use this to send commands to a process
+   *
+   * Input `<id|name> [line] ...` to send stdin to <id|name>
+   *
+   * Any error will exit the CLI
+   * @method attachInput
+   */
+  CLI.prototype.attachInput = function() {
+    var that = this;
+    var readline = require('readline');
+
+    var rl = readline.createInterface({
+      input: process.stdin,
+      output: process.stdout
+    });
+    rl.on('close', function() {
+      that.exitCli();
+    });
+    rl.on('line', function(input) {
+      var line = input.split(' ');
+      that.sendLineToStdin(line.shift(), line.join(' '), function(err) {
+        if (err) {
+          that.exitCli();
+        }
+      });
+    });
+  };
 };

lib/Client.js

@@ -704,6 +704,19 @@ Client.prototype.getProcessIdByName = function(name, force_all, cb) {
   });
 };
 
+/**
+ * Get process id by name and check if only one process is hitted
+ * @param {string} name Name of the process to search for
+ * @param {(err: string, id: string) => void} cb Callback function to handle the result
+ */
+Client.prototype.getUniqueProcessIdByName = function(name, cb) {
+  this.getProcessIdByName(name, function(err, list) {
+    if (err) return cb(err);
+    if (list.length === 1) return cb(null, list[0]);
+    cb(`Expected exactly one process with name "${name}", but found ${list.length}.`);
+  });
+};
+
 Client.prototype.getProcessIdsByNamespace = function(namespace, force_all, cb) {
   var found_proc   = [];
   var full_details = {};

lib/ProcessContainer.js

@@ -9,6 +9,11 @@
  * - pid
  */
 
+var cluster = require('cluster');
+if (cluster.isWorker) {
+  cluster.worker.on('error', function () {});
+}
+
 var p       = require('path');
 var cst     = require('../constants');
 var Utility = require('./Utility.js');
@@ -36,6 +41,13 @@ delete process.env.pm2_env;
 (function ProcessContainer() {
   var fs          = require('fs');
 
+  var original_send = process.send;
+
+  process.send = function() {
+    if (process.connected)
+      original_send.apply(this, arguments);
+  };
+
   ProcessUtils.injectModules()
 
   var stdFile     = pm2_env.pm_log_path;
@@ -44,13 +56,6 @@ delete process.env.pm2_env;
   var pidFile     = pm2_env.pm_pid_path;
   var script      = pm2_env.pm_exec_path;
 
-  var original_send = process.send;
-
-  process.send = function() {
-    if (process.connected)
-      original_send.apply(this, arguments);
-  };
-
   //send node version
   if (process.versions && process.versions.node) {
     process.send({

lib/ProcessContainerBun.js

@@ -4,6 +4,11 @@
  * can be found in the LICENSE file.
  */
 
+var cluster = require('cluster');
+if (cluster.isWorker) {
+  cluster.worker.on('error', function () {});
+}
+
 var p       = require('path');
 var cst     = require('../constants');
 var Utility = require('./Utility.js');
@@ -31,6 +36,13 @@ delete process.env.pm2_env;
 (function ProcessContainer() {
   var fs          = require('fs');
 
+  var original_send = process.send;
+
+  process.send = function() {
+    if (process.connected)
+      original_send.apply(this, arguments);
+  };
+
   ProcessUtils.injectModules()
 
   var stdFile     = pm2_env.pm_log_path;
@@ -39,13 +51,6 @@ delete process.env.pm2_env;
   var pidFile     = pm2_env.pm_pid_path;
   var script      = pm2_env.pm_exec_path;
 
-  var original_send = process.send;
-
-  process.send = function() {
-    if (process.connected)
-      original_send.apply(this, arguments);
-  };
-
   //send node version
   if (process.versions && process.versions.node) {
     process.send({

lib/binaries/CLI.js

@@ -672,18 +672,18 @@ commander.command('cleardump')
 //
 // Save processes to file
 //
-commander.command('send <pm_id> <line>')
-  .description('send stdin to <pm_id>')
-  .action(function(pm_id, line) {
-    pm2.sendLineToStdin(pm_id, line);
+commander.command('send <id|name> <line>')
+  .description('send stdin to <id|name>')
+  .action(function(id, line) {
+    pm2.sendLineToStdin(id, line);
   });
 
 //
 // Attach to stdin/stdout
 // Not TTY ready
 //
-commander.command('attach <pm_id> [command separator]')
-  .description('attach stdin/stdout to application identified by <pm_id>')
+commander.command('attach <id|name> [command separator]')
+  .description('attach stdin/stdout to <id|name>')
   .action(function(pm_id, separator) {
     pm2.attach(pm_id, separator);
   });
@@ -892,6 +892,7 @@ commander.command('logs [id|name|namespace]')
   .option('--timestamp [format]', 'add timestamps (default format YYYY-MM-DD-HH:mm:ss)')
   .option('--nostream', 'print logs without launching the log stream')
   .option('--highlight [value]', 'highlights the given value')
+  .option('--attach-input', 'input `<id|name> [line] ...` to send stdin to <id|name>')
   .description('stream logs file. Default stream all logs')
   .action(function(id, cmd) {
     var Logs = require('../API/Log.js');
@@ -931,6 +932,10 @@ commander.command('logs [id|name|namespace]')
       Logs.formatStream(pm2.Client, id, false, 'YYYY-MM-DD-HH:mm:ssZZ', exclusive, highlight);
     else
       pm2.streamLogs(id, line, raw, timestamp, exclusive, highlight);
+
+    if (cmd.attachInput) {
+      pm2.attachInput();
+    }
   });
 
 

modules/pm2-io-bpm/transports/IPCTransport.js

@@ -86,18 +86,15 @@ class IPCTransport extends EventEmitter {
 
   send (channel, payload) {
     if (typeof process.send !== 'function') return -1
-    if (process.connected === false) {
-      console.error('Process disconnected from parent! (not connected)')
-      return process.exit(1)
-    }
+    if (process.connected === false) return -1
 
+    this.logger(`Send on channel ${channel}`)
     try {
-      this.logger(`Send on channel ${channel}`)
-      process.send({ type: channel, data: payload })
+      process.send({ type: channel, data: payload }, (err) => {
+        if (err) this.logger('async send failed: %s', err && err.code)
+      })
     } catch (err) {
-      this.logger('Process disconnected from parent !')
-      this.logger(err)
-      return process.exit(1)
+      this.logger('Process disconnected from parent: %s', err && err.message)
     }
   }