Merge pull request #5818

f91a06c Dropping cppzmq dependency; adding some zmq utils (vtnerd)

2 files changed, 132 insertions, 1 deletions

diff --git a/tests/unit_tests/CMakeLists.txt b/tests/unit_tests/CMakeLists.txt
index a5d179040..cac1fa943 100644
--- a/tests/unit_tests/CMakeLists.txt
+++ b/tests/unit_tests/CMakeLists.txt
@@ -117,7 +117,8 @@ target_link_libraries(unit_tests
     ${Boost_THREAD_LIBRARY}
     ${GTEST_LIBRARIES}
     ${CMAKE_THREAD_LIBS_INIT}
-    ${EXTRA_LIBRARIES})
+    ${EXTRA_LIBRARIES}
+    ${ZMQ_LIB})
 set_property(TARGET unit_tests
   PROPERTY
     FOLDER "tests")
diff --git a/tests/unit_tests/net.cpp b/tests/unit_tests/net.cpp
index 7e6ba4f03..253280d4d 100644
--- a/tests/unit_tests/net.cpp
+++ b/tests/unit_tests/net.cpp
@@ -40,6 +40,7 @@
 #include <boost/range/adaptor/sliced.hpp>
 #include <boost/range/combine.hpp>
 #include <boost/system/error_code.hpp>
+#include <boost/thread/scoped_thread.hpp>
 #include <boost/thread/thread.hpp>
 #include <boost/uuid/nil_generator.hpp>
 #include <boost/uuid/random_generator.hpp>
@@ -59,6 +60,7 @@
 #include "net/socks_connect.h"
 #include "net/parse.h"
 #include "net/tor_address.h"
+#include "net/zmq.h"
 #include "p2p/net_peerlist_boost_serialization.h"
 #include "serialization/keyvalue_serialization.h"
 #include "storages/portable_storage.h"
@@ -1259,3 +1261,131 @@ TEST(dandelionpp_map, dropped_all_connections)
             EXPECT_EQ(3u, entry.second);
     }
 }
+
+TEST(zmq, error_codes)
+{
+    EXPECT_EQ(
+        std::addressof(net::zmq::error_category()),
+        std::addressof(net::zmq::make_error_code(0).category())
+    );
+    EXPECT_EQ(
+        std::make_error_condition(std::errc::not_a_socket),
+        net::zmq::make_error_code(ENOTSOCK)
+    );
+
+    EXPECT_TRUE(
+        []() -> expect<void>
+        {
+            MONERO_ZMQ_CHECK(zmq_msg_send(nullptr, nullptr, 0));
+            return success();
+        }().matches(std::errc::not_a_socket)
+    );
+
+    bool thrown = false;
+    try
+    {
+        MONERO_ZMQ_THROW("stuff");
+    }
+    catch (const std::system_error& e)
+    {
+        thrown = true;
+        EXPECT_EQ(std::make_error_condition(std::errc::not_a_socket), e.code());
+    }
+    EXPECT_TRUE(thrown);
+}
+
+TEST(zmq, read_write)
+{
+    net::zmq::context context{zmq_init(1)};
+    ASSERT_NE(nullptr, context);
+
+    net::zmq::socket send_socket{zmq_socket(context.get(), ZMQ_REQ)};
+    net::zmq::socket recv_socket{zmq_socket(context.get(), ZMQ_REP)};
+    ASSERT_NE(nullptr, send_socket);
+    ASSERT_NE(nullptr, recv_socket);
+
+    ASSERT_EQ(0u, zmq_bind(recv_socket.get(), "inproc://testing"));
+    ASSERT_EQ(0u, zmq_connect(send_socket.get(), "inproc://testing"));
+
+    std::string message;
+    message.resize(1024);
+    crypto::rand(message.size(), reinterpret_cast<std::uint8_t*>(std::addressof(message[0])));
+
+    ASSERT_TRUE(bool(net::zmq::send(epee::strspan<std::uint8_t>(message), send_socket.get())));
+
+    const expect<std::string> received = net::zmq::receive(recv_socket.get());
+    ASSERT_TRUE(bool(received));
+    EXPECT_EQ(message, *received);
+}
+
+TEST(zmq, read_write_multipart)
+{
+    net::zmq::context context{zmq_init(1)};
+    ASSERT_NE(nullptr, context);
+
+    net::zmq::socket send_socket{zmq_socket(context.get(), ZMQ_REQ)};
+    net::zmq::socket recv_socket{zmq_socket(context.get(), ZMQ_REP)};
+    ASSERT_NE(nullptr, send_socket);
+    ASSERT_NE(nullptr, recv_socket);
+
+    ASSERT_EQ(0u, zmq_bind(recv_socket.get(), "inproc://testing"));
+    ASSERT_EQ(0u, zmq_connect(send_socket.get(), "inproc://testing"));
+
+    std::string message;
+    message.resize(999);
+    crypto::rand(message.size(), reinterpret_cast<std::uint8_t*>(std::addressof(message[0])));
+
+    for (unsigned i = 0; i < 3; ++i)
+    {
+        const expect<std::string> received = net::zmq::receive(recv_socket.get(), ZMQ_DONTWAIT);
+        ASSERT_FALSE(bool(received));
+        EXPECT_EQ(net::zmq::make_error_code(EAGAIN), received.error());
+
+        const epee::span<const std::uint8_t> bytes{
+            reinterpret_cast<const std::uint8_t*>(std::addressof(message[0])) + (i * 333), 333
+        };
+        ASSERT_TRUE(bool(net::zmq::send(bytes, send_socket.get(), (i == 2 ? 0 : ZMQ_SNDMORE))));
+    }
+
+    const expect<std::string> received = net::zmq::receive(recv_socket.get(), ZMQ_DONTWAIT);
+    ASSERT_TRUE(bool(received));
+    EXPECT_EQ(message, *received);
+}
+
+TEST(zmq, read_write_termination)
+{
+    net::zmq::context context{zmq_init(1)};
+    ASSERT_NE(nullptr, context);
+
+    // must be declared before sockets and after context
+    boost::scoped_thread<> thread{};
+
+    net::zmq::socket send_socket{zmq_socket(context.get(), ZMQ_REQ)};
+    net::zmq::socket recv_socket{zmq_socket(context.get(), ZMQ_REP)};
+    ASSERT_NE(nullptr, send_socket);
+    ASSERT_NE(nullptr, recv_socket);
+
+    ASSERT_EQ(0u, zmq_bind(recv_socket.get(), "inproc://testing"));
+    ASSERT_EQ(0u, zmq_connect(send_socket.get(), "inproc://testing"));
+
+    std::string message;
+    message.resize(1024);
+    crypto::rand(message.size(), reinterpret_cast<std::uint8_t*>(std::addressof(message[0])));
+
+    ASSERT_TRUE(bool(net::zmq::send(epee::strspan<std::uint8_t>(message), send_socket.get(), ZMQ_SNDMORE)));
+
+    expect<std::string> received = net::zmq::receive(recv_socket.get(), ZMQ_DONTWAIT);
+    ASSERT_FALSE(bool(received));
+    EXPECT_EQ(net::zmq::make_error_code(EAGAIN), received.error());
+
+    thread = boost::scoped_thread<>{
+        boost::thread{
+            [&context] () { context.reset(); }
+        }
+    };
+
+    received = net::zmq::receive(recv_socket.get());
+    ASSERT_FALSE(bool(received));
+    EXPECT_EQ(net::zmq::make_error_code(ETERM), received.error());
+}
+