From 922f7e9a92300419d78e9d7f6fdbb7a9a439ad7d Mon Sep 17 00:00:00 2001
From: Alexey Rybalchenko <alexryba@gmail.com>
Date: Thu, 14 Mar 2019 14:26:50 +0100
Subject: [PATCH] Use Asio to launch processes in fair::mq::tools::execute

---
 fairmq/tools/Process.cxx | 99 +++++++++++++++++++++++++++++-----------
 1 file changed, 73 insertions(+), 26 deletions(-)

diff --git a/fairmq/tools/Process.cxx b/fairmq/tools/Process.cxx
index da5468e3..1d6062d8 100644
--- a/fairmq/tools/Process.cxx
+++ b/fairmq/tools/Process.cxx
@@ -9,6 +9,8 @@
 #include <fairmq/tools/Process.h>
 
 #include <boost/process.hpp>
+#include <boost/asio.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
 
 #include <iostream>
 #include <sstream>
@@ -17,6 +19,7 @@
 
 using namespace std;
 namespace bp = boost::process;
+namespace ba = boost::asio;
 
 namespace fair
 {
@@ -42,48 +45,92 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
     stringstream printCmd;
     printCmd << prefix << " " << cmd << "\n";
     cout << printCmd.str() << flush;
-
     out << prefix << cmd << endl;
 
-    // Execute command and capture stdout, add prefix line by line
-    bp::ipstream c_stdout;
-    bp::opstream c_stdin;
-    bp::child c(cmd, bp::std_out > c_stdout, bp::std_in < c_stdin);
+    ba::io_service ios;
 
-    while (c.valid() && !c.running()) {
-        ;
-    }
+    // containers for std_in
+    ba::const_buffer inputBuffer(ba::buffer(input));
+    bp::async_pipe inputPipe(ios);
+    // containers for std_out
+    ba::streambuf outputBuffer;
+    bp::async_pipe outputPipe(ios);
+    // containers for std_err
+    ba::streambuf errorBuffer;
+    bp::async_pipe errorPipe(ios);
 
-    if (!c.valid()) {
-        throw runtime_error("Can't execute the given process.");
-    }
+    const string delimiter = "\n";
+    ba::deadline_timer timer(ios, boost::posix_time::milliseconds(100));
 
-    // Optionally, write to stdin of the child
+    // child process
+    bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe, ios);
+
+    // handle std_in with a delay
     if (input != "") {
-        this_thread::sleep_for(chrono::milliseconds(100));
-        c_stdin << input;
-        c_stdin.flush();
+        timer.async_wait([&](const boost::system::error_code& ec1) {
+            if (!ec1) {
+                ba::async_write(inputPipe, inputBuffer, [&](const boost::system::error_code& ec2, size_t /* n */) {
+                    if (!ec2) {
+                        inputPipe.async_close();
+                    } else {
+                        out << prefix << "error in boost::asio::async_write: " << ec2.value() << endl;
+                    }
+                });
+            } else {
+                out << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.value() << endl;
+            }
+        });
     }
 
-    string line;
-    while (c.running() && getline(c_stdout, line)) {
-        // print full line thread-safe
-        stringstream printLine;
-        printLine << prefix << line << "\n";
-        cout << printLine.str() << flush;
+    // handle std_out line by line
+    function<void(const boost::system::error_code&, size_t)> onStdOut = [&](const boost::system::error_code& ec, size_t /* n */) {
+        if (!ec) {
+            istream is(&outputBuffer);
+            string line;
+            getline(is, line);
 
-        out << prefix << line << "\n";
-    }
+            stringstream printLine;
+            printLine << prefix << line << "\n";
+            cout << printLine.str() << flush;
+            out << prefix << line << endl;
 
+            ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
+        } else {
+            outputPipe.async_close();
+        }
+    };
+    ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
+
+    // handle std_err line by line
+    function<void(const boost::system::error_code&, size_t)> onStdErr = [&](const boost::system::error_code& ec, size_t /* n */) {
+        if (!ec) {
+            istream is(&errorBuffer);
+            string line;
+            getline(is, line);
+
+            stringstream printLine;
+            printLine << prefix << line << "\n";
+            cerr << printLine.str() << flush;
+            out << prefix << "error: " << line << endl;
+
+            ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
+        } else {
+            errorPipe.async_close();
+        }
+    };
+    ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
+
+    ios.run();
     c.wait();
 
-    // Capture exit code
     result.exit_code = c.exit_code();
-    out << prefix << " Exit code: " << result.exit_code << endl;
+    stringstream exitCode;
 
+    exitCode << prefix << " Exit code: " << result.exit_code << "\n";
+    cout << exitCode.str() << flush;
+    out << prefix << " Exit code: " << result.exit_code << endl;
     result.console_out = out.str();
 
-    // Return result
     return result;
 }