FairMQ: Add unit tests for PAIR channel type

fairmq/test/CMakeLists.txt

@@ -16,6 +16,8 @@ include(GTestHelper)
 add_testhelper(runTestDevice
     SOURCES
     helper/runTestDevice.cxx
+    helper/devices/TestPairLeft.cxx
+    helper/devices/TestPairRight.cxx
     helper/devices/TestPollIn.cxx
     helper/devices/TestPollOut.cxx
     helper/devices/TestPub.cxx
@@ -38,6 +40,7 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/protocols/runner.cxx.in ${CMAKE_CURRE
 add_testsuite(FairMQ.Protocols
     SOURCES
     ${CMAKE_CURRENT_BINARY_DIR}/protocols/runner.cxx
+    protocols/_pair.cxx
     protocols/_poller.cxx
     protocols/_pub_sub.cxx
     protocols/_push_pull.cxx

fairmq/test/helper/devices/TestPairLeft.cxx

@@ -0,0 +1,40 @@
+/********************************************************************************
+ * Copyright (C) 2015-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"                       *
+ ********************************************************************************/
+
+#include <FairMQDevice.h>
+
+namespace fair
+{
+namespace mq
+{
+namespace test
+{
+
+class PairLeft : public FairMQDevice
+{
+  protected:
+    auto Init() -> void override
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+
+    auto Reset() -> void override
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+
+    auto Run() -> void
+    {
+        auto msg = FairMQMessagePtr{NewMessage()};
+        Send(msg, "data");
+    };
+};
+
+} // namespace test
+} // namespace mq
+} // namespace fair

fairmq/test/helper/devices/TestPairRight.cxx

@@ -0,0 +1,44 @@
+/********************************************************************************
+ * Copyright (C) 2015-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"                       *
+ ********************************************************************************/
+
+#include <FairMQDevice.h>
+
+namespace fair
+{
+namespace mq
+{
+namespace test
+{
+
+class PairRight : public FairMQDevice
+{
+  protected:
+    auto Init() -> void override
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+
+    auto Reset() -> void override
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+
+    auto Run() -> void
+    {
+        auto msg = FairMQMessagePtr{NewMessage()};
+
+        if (Receive(msg, "data") >= 0)
+        {
+            LOG(info) << "PAIR test successfull";
+        }
+    };
+};
+
+} // namespace test
+} // namespace mq
+} // namespace fair

fairmq/test/helper/runTestDevice.cxx

@@ -6,6 +6,8 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
+#include "devices/TestPairLeft.cxx"
+#include "devices/TestPairRight.cxx"
 #include "devices/TestPollIn.cxx"
 #include "devices/TestPollOut.cxx"
 #include "devices/TestPub.cxx"
@@ -70,6 +72,14 @@ auto getDevice(const FairMQProgOptions& config) -> FairMQDevicePtr
     {
         return new PollIn;
     }
+    else if (0 == id.find("pairleft_"))
+    {
+        return new PairLeft;
+    }
+    else if (0 == id.find("pairright_"))
+    {
+        return new PairRight;
+    }
     else
     {
         cerr << "Don't know id '" << id << "'" << endl;

fairmq/test/protocols/_pair.cxx

@@ -0,0 +1,71 @@
+/********************************************************************************
+ *    Copyright (C) 2018 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"                       *
+ ********************************************************************************/
+
+#include "runner.h"
+#include <fairmq/Tools.h>
+#include <gtest/gtest.h>
+#include <sstream> // std::stringstream
+#include <thread>
+
+namespace
+{
+
+using namespace std;
+using namespace fair::mq::test;
+using namespace fair::mq::tools;
+
+auto RunPair(string transport) -> void
+{
+    size_t session{fair::mq::tools::UuidHash()};
+
+    auto pairleft = execute_result{"", 100};
+    thread pairleft_thread([&]() {
+        stringstream cmd;
+        cmd << runTestDevice << " --id pairleft_" << transport << " --control static "
+        << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
+        pairleft = execute(cmd.str(), "[PAIR L]");
+    });
+
+    auto pairright = execute_result{"", 100};
+    thread pairright_thread([&]() {
+        stringstream cmd;
+        cmd << runTestDevice << " --id pairright_" << transport << " --control static "
+        << "--session " << session << " --color false --mq-config \"" << mqConfig << "\"";
+        pairright = execute(cmd.str(), "[PAIR R]");
+    });
+
+    pairleft_thread.join();
+    pairright_thread.join();
+    cerr << pairleft.console_out << pairright.console_out;
+
+    exit(pairleft.exit_code + pairright.exit_code);
+}
+
+TEST(Pair, MP_ZeroMQ__tcp____SingleMsg)
+{
+    EXPECT_EXIT(RunPair("zeromq"), ::testing::ExitedWithCode(0), "PAIR test successfull");
+}
+
+TEST(Pair, MP_ShMem___tcp____SingleMsg)
+{
+    EXPECT_EXIT(RunPair("shmem"), ::testing::ExitedWithCode(0), "PAIR test successfull");
+}
+
+#ifdef NANOMSG_FOUND
+TEST(Pair, MP_Nanomsg_tcp____SingleMsg)
+{
+    EXPECT_EXIT(RunPair("nanomsg"), ::testing::ExitedWithCode(0), "PAIR test successfull");
+}
+#endif /* NANOMSG_FOUND */
+
+TEST(Pair, MP_Ofi_____tcp____SingleMsg)
+{
+    EXPECT_EXIT(RunPair("ofi"), ::testing::ExitedWithCode(0), "PAIR test successfull");
+}
+
+} // namespace

fairmq/test/protocols/config.json.in

@@ -1,6 +1,110 @@
 {
     "fairMQOptions": {
         "devices": [
+            {
+                "id": "pairleft_zeromq",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5557",
+                        "method": "bind",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "zeromq",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairright_zeromq",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5557",
+                        "method": "connect",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "zeromq",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairleft_nanomsg",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5757",
+                        "method": "bind",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "nanomsg",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairright_nanomsg",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5757",
+                        "method": "connect",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "nanomsg",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairleft_shmem",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5857",
+                        "method": "bind",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "shmem",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairright_shmem",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5857",
+                        "method": "connect",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "shmem",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairleft_ofi",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5857",
+                        "method": "bind",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "ofi",
+                        "type": "pair"
+                    }
+                ]
+            },
+            {
+                "id": "pairright_ofi",
+                "channels": [
+                    {
+                        "address": "tcp://127.0.0.1:5857",
+                        "method": "connect",
+                        "name": "data",
+                        "rateLogging": 0,
+                        "transport": "ofi",
+                        "type": "pair"
+                    }
+                ]
+            },
             {
                 "id": "push_zeromq",
                 "channels": [