Replace pstreams with Boost.Process

fairmq/CMakeLists.txt

@@ -106,9 +106,10 @@ set(FAIRMQ_HEADER_FILES
     shmem/Region.h
     tools/CppSTL.h
     tools/Network.h
+    tools/Process.h
     tools/Strings.h
-    tools/Version.h
     tools/Unique.h
+    tools/Version.h
     zeromq/FairMQMessageZMQ.h
     zeromq/FairMQPollerZMQ.h
     zeromq/FairMQUnmanagedRegionZMQ.h

fairmq/Tools.h

@@ -12,9 +12,10 @@
 // IWYU pragma: begin_exports
 #include <fairmq/tools/CppSTL.h>
 #include <fairmq/tools/Network.h>
+#include <fairmq/tools/Process.h>
 #include <fairmq/tools/Strings.h>
-#include <fairmq/tools/Version.h>
 #include <fairmq/tools/Unique.h>
+#include <fairmq/tools/Version.h>
 // IWYU pragma: end_exports
 
 #endif // FAIR_MQ_TOOLS_H

fairmq/test/CMakeLists.txt

@@ -7,8 +7,6 @@
 ################################################################################
 
 find_package(GTest REQUIRED)
-set(PSTREAMS_SOURCE_DIR ${CMAKE_SOURCE_DIR}/3rdparty/pstreams)
-find_package(PStreams REQUIRED)
 include(GTestHelper)
 
 #############################
@@ -47,7 +45,7 @@ add_testsuite(FairMQ.Protocols
     protocols/_transfer_timeout.cxx
     protocols/_push_pull_multipart.cxx
 
-    LINKS PStreams FairMQ
+    LINKS FairMQ
     DEPENDS testhelper_runTestDevice
     INCLUDES ${CMAKE_CURRENT_SOURCE_DIR}/protocols
     TIMEOUT 30

fairmq/test/protocols/_poller.cxx

@@ -17,6 +17,7 @@ namespace
 
 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;
 
 auto RunPoller(string transport, int pollType) -> void
 {
@@ -27,7 +28,7 @@ auto RunPoller(string transport, int pollType) -> void
         stringstream cmd;
         cmd << runTestDevice
             << " --id pollout_"<< transport
-            << " --control static --severity DEBUG --color false"
+            << " --control static --color false"
             << " --session " << session << " --mq-config \"" << mqConfig << "\"";
         pollout = execute(cmd.str(), "[POLLOUT]");
     });
@@ -37,14 +38,14 @@ auto RunPoller(string transport, int pollType) -> void
         stringstream cmd;
         cmd << runTestDevice
             << " --id pollin_" << transport
-            << " --control static --severity DEBUG --color false"
+            << " --control static --color false"
             << " --session " << session << " --mq-config \"" << mqConfig << "\" --poll-type " << pollType;
         pollin = execute(cmd.str(), "[POLLIN]");
     });
 
     poll_out_thread.join();
     poll_in_thread.join();
-    cerr << pollout.error_out << pollin.error_out;
+    cerr << pollout.console_out << pollin.console_out;
 
     exit(pollout.exit_code + pollin.exit_code);
 }

fairmq/test/protocols/_pub_sub.cxx

@@ -17,6 +17,7 @@ namespace
 
 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;
 
 auto RunPubSub(string transport) -> void
 {
@@ -25,7 +26,7 @@ auto RunPubSub(string transport) -> void
     auto pub = execute_result{"", 0};
     thread pub_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id pub_" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id pub_" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         pub = execute(cmd.str(), "[PUB]");
     });
@@ -33,7 +34,7 @@ auto RunPubSub(string transport) -> void
     auto sub1 = execute_result{"", 0};
     thread sub1_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id sub_1" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id sub_1" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         sub1 = execute(cmd.str(), "[SUB1]");
     });
@@ -41,7 +42,7 @@ auto RunPubSub(string transport) -> void
     auto sub2 = execute_result{"", 0};
     thread sub2_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id sub_2" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id sub_2" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         sub2 = execute(cmd.str(), "[SUB2]");
     });
@@ -49,7 +50,7 @@ auto RunPubSub(string transport) -> void
     pub_thread.join();
     sub1_thread.join();
     sub2_thread.join();
-    cerr << pub.error_out << sub1.error_out << sub2.error_out;
+    cerr << pub.console_out << sub1.console_out << sub2.console_out << endl;
 
     exit(pub.exit_code + sub1.exit_code + sub2.exit_code);
 }

fairmq/test/protocols/_push_pull.cxx

@@ -17,6 +17,7 @@ namespace
 
 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;
 
 auto RunPushPull(string transport) -> void
 {
@@ -25,7 +26,7 @@ auto RunPushPull(string transport) -> void
     auto push = execute_result{"", 100};
     thread push_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id push_" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id push_" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         push = execute(cmd.str(), "[PUSH]");
     });
@@ -33,14 +34,14 @@ auto RunPushPull(string transport) -> void
     auto pull = execute_result{"", 100};
     thread pull_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id pull_" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id pull_" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         pull = execute(cmd.str(), "[PULL]");
     });
 
     push_thread.join();
     pull_thread.join();
-    cerr << push.error_out << pull.error_out;
+    cerr << push.console_out << pull.console_out;
 
     exit(push.exit_code + pull.exit_code);
 }

fairmq/test/protocols/_req_rep.cxx

@@ -17,6 +17,7 @@ namespace
 
 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;
 
 auto RunReqRep(string transport) -> void
 {
@@ -25,7 +26,7 @@ auto RunReqRep(string transport) -> void
     auto rep = execute_result{ "", 0 };
     thread rep_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id rep_" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id rep_" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         rep = execute(cmd.str(), "[REP]");
     });
@@ -33,7 +34,7 @@ auto RunReqRep(string transport) -> void
     auto req1 = execute_result{ "", 0 };
     thread req1_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id req_1" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id req_1" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         req1 = execute(cmd.str(), "[REQ1]");
     });
@@ -41,7 +42,7 @@ auto RunReqRep(string transport) -> void
     auto req2 = execute_result{ "", 0 };
     thread req2_thread([&]() {
         stringstream cmd;
-        cmd << runTestDevice << " --id req_2" << transport << " --control static --severity DEBUG "
+        cmd << runTestDevice << " --id req_2" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
         req2 = execute(cmd.str(), "[REQ2]");
     });
@@ -49,7 +50,7 @@ auto RunReqRep(string transport) -> void
     rep_thread.join();
     req1_thread.join();
     req2_thread.join();
-    cerr << req1.error_out << req2.error_out << rep.error_out;
+    cerr << req1.console_out << req2.console_out << rep.console_out;
 
     exit(req1.exit_code + req2.exit_code + rep.exit_code);
 }

fairmq/test/protocols/_transfer_timeout.cxx

@@ -16,16 +16,17 @@ namespace
 
 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;
 
 auto RunTransferTimeout(string transport) -> void
 {
     size_t session{fair::mq::tools::UuidHash()};
     stringstream cmd;
-    cmd << runTestDevice << " --id transfer_timeout_" << transport << " --control static --severity DEBUG "
+    cmd << runTestDevice << " --id transfer_timeout_" << transport << " --control static "
         << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
     auto res = execute(cmd.str());
 
-    cerr << res.error_out;
+    cerr << res.console_out;
 
     exit(res.exit_code);
 }

fairmq/test/protocols/runner.cxx.in

@@ -7,10 +7,10 @@
  ********************************************************************************/
 
 #include "runner.h"
+
 #include <gtest/gtest.h>
-#include <sstream>
+
 #include <string>
-#include <pstream.h> // redi::ipstream
 #include <iostream>
 
 namespace fair
@@ -25,43 +25,6 @@ using namespace std;
 string runTestDevice = "@RUN_TEST_DEVICE@";
 string mqConfig = "@MQ_CONFIG@";
 
-auto execute(string cmd, string log_prefix) -> execute_result
-{
-    auto res = execute_result{"", 0};
-    stringstream out;
-
-    // Log cmd
-
-    // print full line thread-safe
-    stringstream printCmd;
-    printCmd << log_prefix << cmd << endl;
-    cout << printCmd.str() << flush;
-
-    out << log_prefix << cmd << endl;
-
-    // Execute command and capture stderr, add log_prefix line by line
-    redi::ipstream in(cmd, redi::pstreams::pstdout);
-    auto line = string{};
-    while (getline(in, line))
-    {
-        // print full line thread-safe
-        stringstream printLine;
-        printLine << log_prefix << line << endl;
-        cout << printLine.str() << flush;
-
-        out << log_prefix << line << endl;
-    }
-    in.close();
-
-    // Capture exit code
-    res.exit_code = in.rdbuf()->status();
-    out << log_prefix << " Exit code: " << res.exit_code << endl;
-    
-    // Return result
-    res.error_out = out.str();
-    return res;
-}
-
 } /* namespace test */
 } /* namespace mq */
 } /* namespace fair */

fairmq/test/protocols/runner.h

@@ -21,24 +21,6 @@ namespace test
 extern std::string runTestDevice; /// Path to test device executable.
 extern std::string mqConfig; /// Path to FairMQ device config file.
 
-/**
- * Result type for execute function. Holds captured stderr output and exit code.
- */
-struct execute_result {
-    std::string error_out;
-    int exit_code;
-};
-
-/**
- * Execute given command in forked process and capture stderr output
- * and exit code.
- *
- * @param[in] cmd Command to execute
- * @param[in] log_prefix How to prefix each captured output line with
- * @return Captured error output and exit code
- */
-auto execute(std::string cmd, std::string log_prefix = "") -> execute_result;
-
 } /* namespace test */
 } /* namespace mq */
 } /* namespace fair */

fairmq/tools/Process.h

@@ -0,0 +1,82 @@
+/********************************************************************************
+ *    Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ *                                                                              *
+ *              This software is distributed under the terms of the             *
+ *              GNU Lesser General Public Licence (LGPL) version 3,             *
+ *                  copied verbatim in the file "LICENSE"                       *
+ ********************************************************************************/
+
+#ifndef FAIR_MQ_TOOLS_PROCESS_H
+#define FAIR_MQ_TOOLS_PROCESS_H
+
+#include <boost/process.hpp>
+
+#include <string>
+
+namespace fair
+{
+namespace mq
+{
+namespace tools
+{
+
+/**
+ * Result type for execute function. Holds captured stdout output and exit code.
+ */
+struct execute_result
+{
+    std::string console_out;
+    int exit_code;
+};
+
+/**
+ * Execute given command in forked process and capture stdout output
+ * and exit code.
+ *
+ * @param[in] cmd Command to execute
+ * @param[in] log_prefix How to prefix each captured output line with
+ * @return Captured stdout output and exit code
+ */
+inline execute_result execute(std::string cmd, std::string prefix = "")
+{
+    execute_result result;
+    std::stringstream out;
+
+    // print full line thread-safe
+    std::stringstream printCmd;
+    printCmd << prefix << cmd << "\n";
+    std::cout << printCmd.str() << std::flush;
+
+    out << prefix << cmd << std::endl;
+
+    // Execute command and capture stdout, add prefix line by line
+    boost::process::ipstream stdout;
+    boost::process::child c(cmd, boost::process::std_out > stdout);
+    std::string line;
+    while (getline(stdout, line))
+    {
+        // print full line thread-safe
+        std::stringstream printLine;
+        printLine << prefix << line << "\n";
+        std::cout << printLine.str() << std::flush;
+
+        out << prefix << line << "\n";
+    }
+
+    c.wait();
+
+    // Capture exit code
+    result.exit_code = c.exit_code();
+    out << prefix << " Exit code: " << result.exit_code << std::endl;
+
+    result.console_out = out.str();
+
+    // Return result
+    return result;
+}
+
+} /* namespace tools */
+} /* namespace mq */
+} /* namespace fair */
+
+#endif /* FAIR_MQ_TOOLS_PROCESS_H */