Process tools: add print helper, support signals

fairmq/tools/Process.cxx

@@ -7,11 +7,14 @@
  ********************************************************************************/
 
 #include <fairmq/tools/Process.h>
+#include <fairmq/tools/Strings.h>
 
 #include <boost/process.hpp>
 #include <boost/asio.hpp>
 #include <boost/date_time/posix_time/posix_time.hpp>
 
+#include <signal.h> // kill, signals
+
 #include <iostream>
 #include <sstream>
 #include <thread>
@@ -22,6 +25,26 @@ namespace bp = boost::process;
 namespace ba = boost::asio;
 namespace bs = boost::system;
 
+class LinePrinter
+{
+  public:
+    LinePrinter(stringstream& out, const string& prefix)
+        : fOut(out)
+        , fPrefix(prefix)
+    {}
+
+    // prints line with prefix on both cout (thread-safe) and output stream
+    void Print(const string& line)
+    {
+        cout << fair::mq::tools::ToString(fPrefix, line, "\n") << flush;
+        fOut << fPrefix << line << endl;
+    }
+
+  private:
+    stringstream& fOut;
+    const string fPrefix;
+};
+
 namespace fair
 {
 namespace mq
@@ -37,16 +60,14 @@ namespace tools
  * @param[in] log_prefix How to prefix each captured output line with
  * @return Captured stdout output and exit code
  */
-execute_result execute(const string& cmd, const string& prefix, const string& input)
+execute_result execute(const string& cmd, const string& prefix, const string& input, int sig)
 {
     execute_result result;
     stringstream out;
 
-    // print full line thread-safe
+    LinePrinter p(out, prefix);
-    stringstream printCmd;
+
-    printCmd << prefix << " " << cmd << "\n";
+    p.Print(cmd);
-    cout << printCmd.str() << flush;
-    out << prefix << cmd << endl;
 
     ba::io_service ios;
 
@@ -61,26 +82,37 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
     bp::async_pipe errorPipe(ios);
 
     const string delimiter = "\n";
-    ba::deadline_timer timer(ios, boost::posix_time::milliseconds(100));
+    ba::deadline_timer inputTimer(ios, boost::posix_time::milliseconds(100));
+    ba::deadline_timer signalTimer(ios, boost::posix_time::milliseconds(100));
 
     // child process
     bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe);
+    int pid = c.id();
+    p.Print(ToString("fair::mq::tools::execute: pid: ", pid));
 
     // handle std_in with a delay
     if (input != "") {
-        timer.async_wait([&](const bs::error_code& ec1) {
+        inputTimer.async_wait([&](const bs::error_code& ec1) {
             if (!ec1) {
                 ba::async_write(inputPipe, inputBuffer, [&](const bs::error_code& ec2, size_t /* n */) {
                     if (!ec2) {
                         // inputPipe.async_close();
                     } else {
-                        cout << prefix << "error in boost::asio::async_write: " << ec2.message() << endl;
+                        p.Print(ToString("error in boost::asio::async_write: ", ec2.message()));
-                        out << prefix << "error in boost::asio::async_write: " << ec2.message() << endl;
                     }
                 });
             } else {
-                cout << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.message() << endl;
+                p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));
-                out << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.message() << endl;
+            }
+        });
+    }
+
+    if (sig != -1) {
+        signalTimer.async_wait([&](const bs::error_code& ec1) {
+            if (!ec1) {
+                kill(pid, sig);
+            } else {
+                p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));
             }
         });
     }
@@ -92,18 +124,14 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
             string line;
             getline(is, line);
 
-            stringstream printLine;
+            p.Print(line);
-            printLine << prefix << line << "\n";
-            cout << printLine.str() << flush;
-            out << prefix << line << endl;
 
             ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
         } else {
             if (ec == ba::error::eof) {
                 // outputPipe.async_close();
             } else {
-                cout << prefix << ec.message() << endl;
+                p.Print(ec.message());
-                out << prefix << ec.message() << endl;
             }
         }
     };
@@ -116,18 +144,14 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
             string line;
             getline(is, line);
 
-            stringstream printLine;
+            p.Print(ToString("error: ", line));
-            printLine << prefix << line << "\n";
-            cerr << printLine.str() << flush;
-            out << prefix << "error: " << line << endl;
 
             ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
         } else {
             if (ec == ba::error::eof) {
                 // errorPipe.async_close();
             } else {
-                cout << prefix << ec.message() << endl;
+                p.Print(ec.message());
-                out << prefix << ec.message() << endl;
             }
         }
     };
@@ -138,10 +162,7 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
 
     result.exit_code = c.exit_code();
 
-    stringstream exitCode;
+    p.Print(ToString("fair::mq::tools::execute: exit code: ", result.exit_code));
-    exitCode << prefix << " fair::mq::tools::execute: exit code: " << result.exit_code << "\n";
-    cout << exitCode.str() << flush;
-    out << prefix << " fair::mq::tools::execute: exit code: " << result.exit_code << endl;
     result.console_out = out.str();
 
     return result;

fairmq/tools/Process.h

@@ -38,7 +38,8 @@ struct execute_result
  */
 execute_result execute(const std::string& cmd,
                        const std::string& prefix = "",
-                       const std::string& input = "");
+                       const std::string& input = "",
+                       int sig = -1);
 
 } /* namespace tools */
 } /* namespace mq */