From bf47be0a0c1c71eafd683365c159adaa7eab6b7b Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 09:41:36 +0200 Subject: [PATCH 1/9] Problem: poller_t adds an abstraction layer on zmq_poller_* Solution: extract base_poller_t from poller_t, which provides a direct mapping of zmq_poller_* to C++ only --- tests/poller.cpp | 22 ++++---- zmq.hpp | 133 +++++++++++++++++++++++++++++++---------------- 2 files changed, 99 insertions(+), 56 deletions(-) diff --git a/tests/poller.cpp b/tests/poller.cpp index 18f1346..aa1fc77 100644 --- a/tests/poller.cpp +++ b/tests/poller.cpp @@ -204,7 +204,7 @@ TEST(poller, poll_basic) message_received = true; }; ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler)); - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); + ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); ASSERT_TRUE(message_received); } @@ -237,13 +237,13 @@ TEST(poller, client_server) // client sends message ASSERT_NO_THROW(s.client.send(send_msg)); - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); + ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); ASSERT_EQ(events, ZMQ_POLLIN); // Re-add server socket with pollout flag ASSERT_NO_THROW(poller.remove(s.server)); ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN | ZMQ_POLLOUT, handler)); - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); + ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); ASSERT_EQ(events, ZMQ_POLLOUT); } @@ -335,7 +335,7 @@ TEST(poller, poll_client_server) // Modify server socket with pollout flag ASSERT_NO_THROW(poller.modify(s.server, ZMQ_POLLIN | ZMQ_POLLOUT)); - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{500})); ASSERT_EQ(s.events, ZMQ_POLLIN | ZMQ_POLLOUT); } @@ -356,8 +356,8 @@ TEST(poller, wait_one_return) ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify events - int result = poller.wait(std::chrono::milliseconds{500}); - ASSERT_EQ(count, result); + ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(1u, count); } TEST(poller, wait_on_move_constructed_poller) @@ -401,14 +401,14 @@ TEST(poller, received_on_move_construced_poller) // client sends message ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify it is received - a.wait(std::chrono::milliseconds{500}); + ASSERT_EQ(1, a.wait(std::chrono::milliseconds{500})); ASSERT_EQ(1u, count); // Move construct poller b zmq::poller_t b{std::move(a)}; // client sends message again ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify it is received - b.wait(std::chrono::milliseconds{500}); + ASSERT_EQ(1, b.wait(std::chrono::milliseconds{500})); ASSERT_EQ(2u, count); } @@ -424,12 +424,14 @@ TEST(poller, remove_from_handler) // Setup poller zmq::poller_t poller; + int count = 0; for (auto i = 0; i < ITER_NO; ++i) { ASSERT_NO_THROW(poller.add(setup_list[i].server, ZMQ_POLLIN, [&,i](short events) { ASSERT_EQ(events, ZMQ_POLLIN); poller.remove(setup_list[ITER_NO-i-1].server); ASSERT_EQ(ITER_NO-i-1, poller.size()); })); + ++count; } ASSERT_EQ(ITER_NO, poller.size()); // Clients send messages @@ -444,8 +446,8 @@ TEST(poller, remove_from_handler) } // Fire all handlers in one wait - int count = poller.wait (std::chrono::milliseconds{-1}); - ASSERT_EQ(count, ITER_NO); + ASSERT_EQ(ITER_NO, poller.wait (std::chrono::milliseconds{-1})); + ASSERT_EQ(ITER_NO, count); } #endif diff --git a/zmq.hpp b/zmq.hpp index de9e131..3d2f7bc 100644 --- a/zmq.hpp +++ b/zmq.hpp @@ -577,7 +577,6 @@ namespace zmq class socket_t { friend class monitor_t; - friend class poller_t; public: inline socket_t(context_t& context_, int type_) { @@ -1019,6 +1018,67 @@ namespace zmq }; #if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) + template + class base_poller_t + { + public: + void add (zmq::socket_t &socket, short events, T *user_data) + { + if (0 != zmq_poller_add (poller_ptr.get (), static_cast(socket), user_data, events)) + { + throw error_t (); + } + } + + void remove (zmq::socket_t &socket) + { + if (0 != zmq_poller_remove (poller_ptr.get (), static_cast(socket))) + { + throw error_t (); + } + } + + void modify (zmq::socket_t &socket, short events) + { + if (0 != zmq_poller_modify (poller_ptr.get (), static_cast(socket), events)) + { + throw error_t (); + } + } + + int wait_all (std::vector &poller_events, const std::chrono::microseconds timeout) + { + int rc = zmq_poller_wait_all (poller_ptr.get (), poller_events.data (), + static_cast (poller_events.size ()), + static_cast(timeout.count ())); + if (rc > 0) + return rc; + +#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 2, 3) + if (zmq_errno () == EAGAIN) +#else + if (zmq_errno () == ETIMEDOUT) +#endif + return 0; + + throw error_t (); + } + private: + std::unique_ptr> poller_ptr + { + []() { + auto poller_new = zmq_poller_new (); + if (poller_new) + return poller_new; + throw error_t (); + }(), + [](void *ptr) { + int rc = zmq_poller_destroy (&ptr); + ZMQ_ASSERT (rc == 0); + } + }; + }; + class poller_t { public: @@ -1035,33 +1095,35 @@ namespace zmq void add (zmq::socket_t &socket, short events, handler_t handler) { - auto it = std::end (handlers); - auto inserted = false; - std::tie(it, inserted) = handlers.emplace (socket.ptr, std::make_shared (std::move (handler))); - if (0 == zmq_poller_add (poller_ptr.get (), socket.ptr, inserted && *(it->second) ? it->second.get() : nullptr, events)) { - need_rebuild = true; - return; + auto it = decltype (handlers)::iterator {}; + auto inserted = bool {}; + std::tie(it, inserted) = handlers.emplace (static_cast(socket), std::make_shared (std::move (handler))); + try + { + base_poller.add (socket, events, inserted && *(it->second) ? it->second.get() : nullptr); + need_rebuild |= inserted; + } + catch (const zmq::error_t&) + { + // rollback + if (inserted) + { + handlers.erase (static_cast(socket)); + } + throw; } - // rollback - if (inserted) - handlers.erase (socket.ptr); - throw error_t (); } void remove (zmq::socket_t &socket) { - if (0 == zmq_poller_remove (poller_ptr.get (), socket.ptr)) { - handlers.erase (socket.ptr); - need_rebuild = true; - return; - } - throw error_t (); + base_poller.remove (socket); + handlers.erase (static_cast(socket)); + need_rebuild = true; } void modify (zmq::socket_t &socket, short events) { - if (0 != zmq_poller_modify (poller_ptr.get (), socket.ptr, events)) - throw error_t (); + base_poller.modify (socket, events); } int wait (std::chrono::milliseconds timeout) @@ -1077,25 +1139,15 @@ namespace zmq } need_rebuild = false; } - int rc = zmq_poller_wait_all (poller_ptr.get (), poller_events.data (), - static_cast (poller_events.size ()), - static_cast(timeout.count ())); - if (rc > 0) { - std::for_each (poller_events.begin (), poller_events.begin () + rc, + const int count = base_poller.wait_all (poller_events, timeout); + if (count != 0) { + std::for_each (poller_events.begin (), poller_events.begin () + count, [](zmq_poller_event_t& event) { if (event.user_data != NULL) (*reinterpret_cast (event.user_data)) (event.events); }); - return rc; } -#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 2, 3) - if (zmq_errno () == EAGAIN) -#else - if (zmq_errno () == ETIMEDOUT) -#endif - return 0; - - throw error_t (); + return count; } bool empty () const @@ -1109,20 +1161,9 @@ namespace zmq } private: - std::unique_ptr> poller_ptr - { - []() { - auto poller_new = zmq_poller_new (); - if (poller_new) - return poller_new; - throw error_t (); - }(), - [](void *ptr) { - int rc = zmq_poller_destroy (&ptr); - ZMQ_ASSERT (rc == 0); - } - }; bool need_rebuild {false}; + + base_poller_t base_poller {}; std::unordered_map> handlers {}; std::vector poller_events {}; std::vector> poller_handlers {}; From 89d9db7366982773f8cf3524abafcb6e7673d240 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 10:32:02 +0200 Subject: [PATCH 2/9] Problem: move poller tests are unspecific Solution: remove redundant assertions (tested elsewhere), and add assertion behaviour of calling wait on moved-from poller --- tests/poller.cpp | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/tests/poller.cpp b/tests/poller.cpp index aa1fc77..edd4d7c 100644 --- a/tests/poller.cpp +++ b/tests/poller.cpp @@ -367,28 +367,29 @@ TEST(poller, wait_on_move_constructed_poller) zmq::poller_t a; zmq::poller_t::handler_t handler; ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); - ASSERT_EQ(1u, a.size ()); zmq::poller_t b {std::move (a)}; ASSERT_EQ(1u, b.size ()); - ASSERT_NO_THROW (b.wait (std::chrono::milliseconds {-1})); + /// \todo the actual error code should be checked + ASSERT_THROW(a.wait(std::chrono::milliseconds{10}), zmq::error_t); + ASSERT_TRUE (b.wait (std::chrono::milliseconds {-1})); } -TEST(poller, wait_on_move_assign_poller) +TEST(poller, wait_on_move_assigned_poller) { server_client_setup s; ASSERT_NO_THROW (s.client.send ("Hi")); zmq::poller_t a; zmq::poller_t::handler_t handler; ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); - ASSERT_EQ(1u, a.size ()); zmq::poller_t b; - ASSERT_EQ(0u, b.size ()); b = {std::move (a)}; ASSERT_EQ(1u, b.size ()); - ASSERT_NO_THROW (b.wait (std::chrono::milliseconds {-1})); + /// \todo the actual error code should be checked + ASSERT_THROW(a.wait(std::chrono::milliseconds{10}), zmq::error_t); + ASSERT_TRUE (b.wait (std::chrono::milliseconds {-1})); } -TEST(poller, received_on_move_construced_poller) +TEST(poller, received_on_move_constructed_poller) { // Setup server and client server_client_setup s; From dc996cd2ecf06699b5200d1fea5f064b1a6e93b3 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 11:01:04 +0200 Subject: [PATCH 3/9] Problem: zmq_addon header contains tests (with a syntax error) Solution: convert to googletest based test, and fix syntax error --- tests/CMakeLists.txt | 1 + tests/multipart.cpp | 171 +++++++++++++++++++++++++++++++++++++++++++ zmq_addon.hpp | 166 ----------------------------------------- 3 files changed, 172 insertions(+), 166 deletions(-) create mode 100644 tests/multipart.cpp diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index f664071..51f4197 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -26,6 +26,7 @@ add_executable( context.cpp socket.cpp poller.cpp + multipart.cpp ) target_link_libraries( diff --git a/tests/multipart.cpp b/tests/multipart.cpp new file mode 100644 index 0000000..0c279e9 --- /dev/null +++ b/tests/multipart.cpp @@ -0,0 +1,171 @@ +#include +#include + +#ifdef ZMQ_HAS_RVALUE_REFS +/// \todo split this up into separate test cases +/// +TEST(multipart, legacy_test) +{ + using namespace zmq; + + bool ok = true; (void) ok; + float num = 0; (void) num; + std::string str = ""; + message_t msg; + + // Create two PAIR sockets and connect over inproc + context_t context(1); + socket_t output(context, ZMQ_PAIR); + socket_t input(context, ZMQ_PAIR); + output.bind("inproc://multipart.test"); + input.connect("inproc://multipart.test"); + + // Test send and receive of single-frame message + multipart_t multipart; + assert(multipart.empty()); + + multipart.push(message_t("Hello", 5)); + assert(multipart.size() == 1); + + ok = multipart.send(output); + assert(multipart.empty()); + assert(ok); + + ok = multipart.recv(input); + assert(multipart.size() == 1); + assert(ok); + + msg = multipart.pop(); + assert(multipart.empty()); + assert(std::string(msg.data(), msg.size()) == "Hello"); + + // Test send and receive of multi-frame message + multipart.addstr("A"); + multipart.addstr("BB"); + multipart.addstr("CCC"); + assert(multipart.size() == 3); + + multipart_t copy = multipart.clone(); + assert(copy.size() == 3); + + ok = copy.send(output); + assert(copy.empty()); + assert(ok); + + ok = copy.recv(input); + assert(copy.size() == 3); + assert(ok); + assert(copy.equal(&multipart)); + + multipart.clear(); + assert(multipart.empty()); + + // Test message frame manipulation + multipart.add(message_t("Frame5", 6)); + multipart.addstr("Frame6"); + multipart.addstr("Frame7"); + multipart.addtyp(8.0f); + multipart.addmem("Frame9", 6); + multipart.push(message_t("Frame4", 6)); + multipart.pushstr("Frame3"); + multipart.pushstr("Frame2"); + multipart.pushtyp(1.0f); + multipart.pushmem("Frame0", 6); + assert(multipart.size() == 10); + + msg = multipart.remove(); + assert(multipart.size() == 9); + assert(std::string(msg.data(), msg.size()) == "Frame9"); + + msg = multipart.pop(); + assert(multipart.size() == 8); + assert(std::string(msg.data(), msg.size()) == "Frame0"); + + num = multipart.poptyp(); + assert(multipart.size() == 7); + assert(num == 1.0f); + + str = multipart.popstr(); + assert(multipart.size() == 6); + assert(str == "Frame2"); + + str = multipart.popstr(); + assert(multipart.size() == 5); + assert(str == "Frame3"); + + str = multipart.popstr(); + assert(multipart.size() == 4); + assert(str == "Frame4"); + + str = multipart.popstr(); + assert(multipart.size() == 3); + assert(str == "Frame5"); + + str = multipart.popstr(); + assert(multipart.size() == 2); + assert(str == "Frame6"); + + str = multipart.popstr(); + assert(multipart.size() == 1); + assert(str == "Frame7"); + + num = multipart.poptyp(); + assert(multipart.empty()); + assert(num == 8.0f); + + // Test message constructors and concatenation + multipart_t head("One", 3); + head.addstr("Two"); + assert(head.size() == 2); + + multipart_t tail(std::string("One-hundred")); + tail.pushstr("Ninety-nine"); + assert(tail.size() == 2); + + multipart_t tmp(message_t("Fifty", 5)); + assert(tmp.size() == 1); + + multipart_t mid = multipart_t::create(49.0f); + mid.append(std::move(tmp)); + assert(mid.size() == 2); + assert(tmp.empty()); + + multipart_t merged(std::move(mid)); + merged.prepend(std::move(head)); + merged.append(std::move(tail)); + assert(merged.size() == 6); + assert(head.empty()); + assert(tail.empty()); + + ok = merged.send(output); + assert(merged.empty()); + assert(ok); + + multipart_t received(input); + assert(received.size() == 6); + + str = received.popstr(); + assert(received.size() == 5); + assert(str == "One"); + + str = received.popstr(); + assert(received.size() == 4); + assert(str == "Two"); + + num = received.poptyp(); + assert(received.size() == 3); + assert(num == 49.0f); + + str = received.popstr(); + assert(received.size() == 2); + assert(str == "Fifty"); + + str = received.popstr(); + assert(received.size() == 1); + assert(str == "Ninety-nine"); + + str = received.popstr(); + assert(received.empty()); + assert(str == "One-hundred"); +} +#endif diff --git a/zmq_addon.hpp b/zmq_addon.hpp index 4dd6f05..bc9ae07 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -413,172 +413,6 @@ public: return true; } - // Self test - static int test() - { - bool ok = true; (void) ok; - float num = 0; (void) num; - std::string str = ""; - message_t msg; - - // Create two PAIR sockets and connect over inproc - context_t context(1); - socket_t output(context, ZMQ_PAIR); - socket_t input(context, ZMQ_PAIR); - output.bind("inproc://multipart.test"); - input.connect("inproc://multipart.test"); - - // Test send and receive of single-frame message - multipart_t multipart; - assert(multipart.empty()); - - multipart.push(message_t("Hello", 5)); - assert(multipart.size() == 1); - - ok = multipart.send(output); - assert(multipart.empty()); - assert(ok); - - ok = multipart.recv(input); - assert(multipart.size() == 1); - assert(ok); - - msg = multipart.pop(); - assert(multipart.empty()); - assert(std::string(msg.data(), msg.size()) == "Hello"); - - // Test send and receive of multi-frame message - multipart.addstr("A"); - multipart.addstr("BB"); - multipart.addstr("CCC"); - assert(multipart.size() == 3); - - multipart_t copy = multipart.clone(); - assert(copy.size() == 3); - - ok = copy.send(output); - assert(copy.empty()); - assert(ok); - - ok = copy.recv(input); - assert(copy.size() == 3); - assert(ok); - assert(copy.equal(&multipart)); - - multipart.clear(); - assert(multipart.empty()); - - // Test message frame manipulation - multipart.add(message_t("Frame5", 6)); - multipart.addstr("Frame6"); - multipart.addstr("Frame7"); - multipart.addtyp(8.0f); - multipart.addmem("Frame9", 6); - multipart.push(message_t("Frame4", 6)); - multipart.pushstr("Frame3"); - multipart.pushstr("Frame2"); - multipart.pushtyp(1.0f); - multipart.pushmem("Frame0", 6); - assert(multipart.size() == 10); - - msg = multipart.remove(); - assert(multipart.size() == 9); - assert(std::string(msg.data(), msg.size()) == "Frame9"); - - msg = multipart.pop(); - assert(multipart.size() == 8); - assert(std::string(msg.data(), msg.size()) == "Frame0"); - - num = multipart.poptyp(); - assert(multipart.size() == 7); - assert(num == 1.0f); - - str = multipart.popstr(); - assert(multipart.size() == 6); - assert(str == "Frame2"); - - str = multipart.popstr(); - assert(multipart.size() == 5); - assert(str == "Frame3"); - - str = multipart.popstr(); - assert(multipart.size() == 4); - assert(str == "Frame4"); - - str = multipart.popstr(); - assert(multipart.size() == 3); - assert(str == "Frame5"); - - str = multipart.popstr(); - assert(multipart.size() == 2); - assert(str == "Frame6"); - - str = multipart.popstr(); - assert(multipart.size() == 1); - assert(str == "Frame7"); - - num = multipart.poptyp(); - assert(multipart.empty()); - assert(num == 8.0f); - - // Test message constructors and concatenation - multipart_t head("One", 3); - head.addstr("Two"); - assert(head.size() == 2); - - multipart_t tail("One-hundred"); - tail.pushstr("Ninety-nine"); - assert(tail.size() == 2); - - multipart_t tmp(message_t("Fifty", 5)); - assert(tmp.size() == 1); - - multipart_t mid = multipart_t::create(49.0f); - mid.append(std::move(tmp)); - assert(mid.size() == 2); - assert(tmp.empty()); - - multipart_t merged(std::move(mid)); - merged.prepend(std::move(head)); - merged.append(std::move(tail)); - assert(merged.size() == 6); - assert(head.empty()); - assert(tail.empty()); - - ok = merged.send(output); - assert(merged.empty()); - assert(ok); - - multipart_t received(input); - assert(received.size() == 6); - - str = received.popstr(); - assert(received.size() == 5); - assert(str == "One"); - - str = received.popstr(); - assert(received.size() == 4); - assert(str == "Two"); - - num = received.poptyp(); - assert(received.size() == 3); - assert(num == 49.0f); - - str = received.popstr(); - assert(received.size() == 2); - assert(str == "Fifty"); - - str = received.popstr(); - assert(received.size() == 1); - assert(str == "Ninety-nine"); - - str = received.popstr(); - assert(received.empty()); - assert(str == "One-hundred"); - - return 0; - } - private: // Disable implicit copying (moving is more efficient) multipart_t(const multipart_t& other) ZMQ_DELETED_FUNCTION; From 882f5e844cab058a506ee79f581d39a87096afa1 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 10:56:41 +0200 Subject: [PATCH 4/9] Problem: extra abstraction layer type poller_t is in zmq.hpp Solution: move to zmq_addon.hpp, rename to active_poller_t, and rename base_poller_t to poller_t --- tests/CMakeLists.txt | 2 +- tests/{poller.cpp => active_poller.cpp} | 246 ++++++++++++------------ zmq.hpp | 93 +-------- zmq_addon.hpp | 93 +++++++++ 4 files changed, 218 insertions(+), 216 deletions(-) rename tests/{poller.cpp => active_poller.cpp} (54%) diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 51f4197..30b86b0 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -25,7 +25,7 @@ add_executable( message.cpp context.cpp socket.cpp - poller.cpp + active_poller.cpp multipart.cpp ) diff --git a/tests/poller.cpp b/tests/active_poller.cpp similarity index 54% rename from tests/poller.cpp rename to tests/active_poller.cpp index edd4d7c..87b4d95 100644 --- a/tests/poller.cpp +++ b/tests/active_poller.cpp @@ -1,36 +1,36 @@ #include -#include +#include #if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API) #include #include -TEST(poller, create_destroy) +TEST(active_poller, create_destroy) { - zmq::poller_t poller; - ASSERT_TRUE(poller.empty ()); + zmq::active_poller_t active_poller; + ASSERT_TRUE(active_poller.empty ()); } -static_assert(!std::is_copy_constructible::value, "poller_t should not be copy-constructible"); -static_assert(!std::is_copy_assignable::value, "poller_t should not be copy-assignable"); +static_assert(!std::is_copy_constructible::value, "active_active_poller_t should not be copy-constructible"); +static_assert(!std::is_copy_assignable::value, "active_active_poller_t should not be copy-assignable"); -TEST(poller, move_construct_empty) +TEST(active_poller, move_construct_empty) { - std::unique_ptr a {new zmq::poller_t}; + std::unique_ptr a {new zmq::active_poller_t}; ASSERT_TRUE(a->empty ()); - zmq::poller_t b = std::move (*a); + zmq::active_poller_t b = std::move (*a); ASSERT_TRUE(b.empty ()); ASSERT_EQ(0u, a->size ()); ASSERT_EQ(0u, b.size ()); a.reset (); } -TEST(poller, move_assign_empty) +TEST(active_poller, move_assign_empty) { - std::unique_ptr a{new zmq::poller_t}; + std::unique_ptr a{new zmq::active_poller_t}; ASSERT_TRUE(a->empty()); - zmq::poller_t b; + zmq::active_poller_t b; ASSERT_TRUE(b.empty()); b = std::move(*a); ASSERT_EQ(0u, a->size ()); @@ -40,16 +40,16 @@ TEST(poller, move_assign_empty) a.reset (); } -TEST(poller, move_construct_non_empty) +TEST(active_poller, move_construct_non_empty) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - std::unique_ptr a{new zmq::poller_t}; + std::unique_ptr a{new zmq::active_poller_t}; a->add(socket, ZMQ_POLLIN, [](short) {}); ASSERT_FALSE(a->empty ()); ASSERT_EQ(1u, a->size ()); - zmq::poller_t b = std::move (*a); + zmq::active_poller_t b = std::move (*a); ASSERT_TRUE(a->empty ()); ASSERT_EQ(0u, a->size ()); ASSERT_FALSE(b.empty ()); @@ -57,16 +57,16 @@ TEST(poller, move_construct_non_empty) a.reset (); } -TEST(poller, move_assign_non_empty) +TEST(active_poller, move_assign_non_empty) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - std::unique_ptr a{new zmq::poller_t}; + std::unique_ptr a{new zmq::active_poller_t}; a->add(socket, ZMQ_POLLIN, [](short) {}); ASSERT_FALSE(a->empty()); ASSERT_EQ(1u, a->size ()); - zmq::poller_t b; + zmq::active_poller_t b; b = std::move(*a); ASSERT_TRUE(a->empty ()); ASSERT_EQ(0u, a->size ()); @@ -75,70 +75,70 @@ TEST(poller, move_assign_non_empty) a.reset (); } -TEST(poller, add_handler) +TEST(active_poller, add_handler) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; - zmq::poller_t::handler_t handler; - ASSERT_NO_THROW(poller.add(socket, ZMQ_POLLIN, handler)); + zmq::active_poller_t active_poller; + zmq::active_poller_t::handler_t handler; + ASSERT_NO_THROW(active_poller.add(socket, ZMQ_POLLIN, handler)); } -TEST(poller, add_handler_invalid_events_type) +TEST(active_poller, add_handler_invalid_events_type) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; - zmq::poller_t::handler_t handler; + zmq::active_poller_t active_poller; + zmq::active_poller_t::handler_t handler; short invalid_events_type = 2 << 10; - ASSERT_NO_THROW(poller.add(socket, invalid_events_type, handler)); - ASSERT_FALSE(poller.empty ()); - ASSERT_EQ(1u, poller.size ()); + ASSERT_NO_THROW(active_poller.add(socket, invalid_events_type, handler)); + ASSERT_FALSE(active_poller.empty ()); + ASSERT_EQ(1u, active_poller.size ()); } -TEST(poller, add_handler_twice_throws) +TEST(active_poller, add_handler_twice_throws) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; - zmq::poller_t::handler_t handler; - poller.add(socket, ZMQ_POLLIN, handler); + zmq::active_poller_t active_poller; + zmq::active_poller_t::handler_t handler; + active_poller.add(socket, ZMQ_POLLIN, handler); /// \todo the actual error code should be checked - ASSERT_THROW(poller.add(socket, ZMQ_POLLIN, handler), zmq::error_t); + ASSERT_THROW(active_poller.add(socket, ZMQ_POLLIN, handler), zmq::error_t); } -TEST(poller, wait_with_no_handlers_throws) +TEST(active_poller, wait_with_no_handlers_throws) { - zmq::poller_t poller; + zmq::active_poller_t active_poller; /// \todo the actual error code should be checked - ASSERT_THROW(poller.wait(std::chrono::milliseconds{10}), zmq::error_t); + ASSERT_THROW(active_poller.wait(std::chrono::milliseconds{10}), zmq::error_t); } -TEST(poller, remove_unregistered_throws) +TEST(active_poller, remove_unregistered_throws) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; + zmq::active_poller_t active_poller; /// \todo the actual error code should be checked - ASSERT_THROW(poller.remove(socket), zmq::error_t); + ASSERT_THROW(active_poller.remove(socket), zmq::error_t); } -TEST(poller, remove_registered_empty) +TEST(active_poller, remove_registered_empty) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; - poller.add(socket, ZMQ_POLLIN, zmq::poller_t::handler_t{}); - ASSERT_NO_THROW(poller.remove(socket)); + zmq::active_poller_t active_poller; + active_poller.add(socket, ZMQ_POLLIN, zmq::active_poller_t::handler_t{}); + ASSERT_NO_THROW(active_poller.remove(socket)); } -TEST(poller, remove_registered_non_empty) +TEST(active_poller, remove_registered_non_empty) { zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - zmq::poller_t poller; - poller.add(socket, ZMQ_POLLIN, [](short) {}); - ASSERT_NO_THROW(poller.remove(socket)); + zmq::active_poller_t active_poller; + active_poller.add(socket, ZMQ_POLLIN, [](short) {}); + ASSERT_NO_THROW(active_poller.remove(socket)); } namespace { @@ -178,7 +178,7 @@ struct server_client_setup ASSERT_NO_THROW (client.connect (endpoint)); } - zmq::poller_t::handler_t handler = [&](short e) { + zmq::active_poller_t::handler_t handler = [&](short e) { events = e; }; @@ -191,35 +191,35 @@ struct server_client_setup } //namespace -TEST(poller, poll_basic) +TEST(active_poller, poll_basic) { server_client_setup s; ASSERT_NO_THROW(s.client.send("Hi")); - zmq::poller_t poller; + zmq::active_poller_t active_poller; bool message_received = false; - zmq::poller_t::handler_t handler = [&message_received](short events) { + zmq::active_poller_t::handler_t handler = [&message_received](short events) { ASSERT_TRUE(0 != (events & ZMQ_POLLIN)); message_received = true; }; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler)); - ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN, handler)); + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{-1})); ASSERT_TRUE(message_received); } /// \todo this contains multiple test cases that should be split up -TEST(poller, client_server) +TEST(active_poller, client_server) { const std::string send_msg = "Hi"; // Setup server and client server_client_setup s; - // Setup poller - zmq::poller_t poller; + // Setup active_poller + zmq::active_poller_t active_poller; short events; - zmq::poller_t::handler_t handler = [&](short e) { + zmq::active_poller_t::handler_t handler = [&](short e) { if (0 != (e & ZMQ_POLLIN)) { zmq::message_t zmq_msg; ASSERT_NO_THROW(s.server.recv(&zmq_msg)); // get message @@ -232,123 +232,123 @@ TEST(poller, client_server) events = e; }; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler)); + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN, handler)); // client sends message ASSERT_NO_THROW(s.client.send(send_msg)); - ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{-1})); ASSERT_EQ(events, ZMQ_POLLIN); // Re-add server socket with pollout flag - ASSERT_NO_THROW(poller.remove(s.server)); - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN | ZMQ_POLLOUT, handler)); - ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{-1})); + ASSERT_NO_THROW(active_poller.remove(s.server)); + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN | ZMQ_POLLOUT, handler)); + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{-1})); ASSERT_EQ(events, ZMQ_POLLOUT); } -TEST(poller, add_invalid_socket_throws) +TEST(active_poller, add_invalid_socket_throws) { zmq::context_t context; - zmq::poller_t poller; + zmq::active_poller_t active_poller; zmq::socket_t a {context, zmq::socket_type::router}; zmq::socket_t b {std::move (a)}; - ASSERT_THROW (poller.add (a, ZMQ_POLLIN, zmq::poller_t::handler_t {}), + ASSERT_THROW (active_poller.add (a, ZMQ_POLLIN, zmq::active_poller_t::handler_t {}), zmq::error_t); } -TEST(poller, remove_invalid_socket_throws) +TEST(active_poller, remove_invalid_socket_throws) { zmq::context_t context; zmq::socket_t socket {context, zmq::socket_type::router}; - zmq::poller_t poller; - ASSERT_NO_THROW (poller.add (socket, ZMQ_POLLIN, zmq::poller_t::handler_t {})); - ASSERT_EQ (1u, poller.size ()); + zmq::active_poller_t active_poller; + ASSERT_NO_THROW (active_poller.add (socket, ZMQ_POLLIN, zmq::active_poller_t::handler_t {})); + ASSERT_EQ (1u, active_poller.size ()); std::vector sockets; sockets.emplace_back (std::move (socket)); - ASSERT_THROW (poller.remove (socket), zmq::error_t); - ASSERT_EQ (1u, poller.size ()); + ASSERT_THROW (active_poller.remove (socket), zmq::error_t); + ASSERT_EQ (1u, active_poller.size ()); } -TEST(poller, wait_on_added_empty_handler) +TEST(active_poller, wait_on_added_empty_handler) { server_client_setup s; ASSERT_NO_THROW (s.client.send ("Hi")); - zmq::poller_t poller; - zmq::poller_t::handler_t handler; - ASSERT_NO_THROW (poller.add (s.server, ZMQ_POLLIN, handler)); - ASSERT_NO_THROW (poller.wait (std::chrono::milliseconds {-1})); + zmq::active_poller_t active_poller; + zmq::active_poller_t::handler_t handler; + ASSERT_NO_THROW (active_poller.add (s.server, ZMQ_POLLIN, handler)); + ASSERT_NO_THROW (active_poller.wait (std::chrono::milliseconds {-1})); } -TEST(poller, modify_empty_throws) +TEST(active_poller, modify_empty_throws) { zmq::context_t context; zmq::socket_t socket {context, zmq::socket_type::push}; - zmq::poller_t poller; - ASSERT_THROW (poller.modify (socket, ZMQ_POLLIN), zmq::error_t); + zmq::active_poller_t active_poller; + ASSERT_THROW (active_poller.modify (socket, ZMQ_POLLIN), zmq::error_t); } -TEST(poller, modify_invalid_socket_throws) +TEST(active_poller, modify_invalid_socket_throws) { zmq::context_t context; zmq::socket_t a {context, zmq::socket_type::push}; zmq::socket_t b {std::move (a)}; - zmq::poller_t poller; - ASSERT_THROW (poller.modify (a, ZMQ_POLLIN), zmq::error_t); + zmq::active_poller_t active_poller; + ASSERT_THROW (active_poller.modify (a, ZMQ_POLLIN), zmq::error_t); } -TEST(poller, modify_not_added_throws) +TEST(active_poller, modify_not_added_throws) { zmq::context_t context; zmq::socket_t a {context, zmq::socket_type::push}; zmq::socket_t b {context, zmq::socket_type::push}; - zmq::poller_t poller; - ASSERT_NO_THROW (poller.add (a, ZMQ_POLLIN, zmq::poller_t::handler_t {})); - ASSERT_THROW (poller.modify (b, ZMQ_POLLIN), zmq::error_t); + zmq::active_poller_t active_poller; + ASSERT_NO_THROW (active_poller.add (a, ZMQ_POLLIN, zmq::active_poller_t::handler_t {})); + ASSERT_THROW (active_poller.modify (b, ZMQ_POLLIN), zmq::error_t); } -TEST(poller, modify_simple) +TEST(active_poller, modify_simple) { zmq::context_t context; zmq::socket_t a {context, zmq::socket_type::push}; - zmq::poller_t poller; - ASSERT_NO_THROW (poller.add (a, ZMQ_POLLIN, zmq::poller_t::handler_t {})); - ASSERT_NO_THROW (poller.modify (a, ZMQ_POLLIN|ZMQ_POLLOUT)); + zmq::active_poller_t active_poller; + ASSERT_NO_THROW (active_poller.add (a, ZMQ_POLLIN, zmq::active_poller_t::handler_t {})); + ASSERT_NO_THROW (active_poller.modify (a, ZMQ_POLLIN|ZMQ_POLLOUT)); } -TEST(poller, poll_client_server) +TEST(active_poller, poll_client_server) { // Setup server and client server_client_setup s; - // Setup poller - zmq::poller_t poller; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, s.handler)); + // Setup active_poller + zmq::active_poller_t active_poller; + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN, s.handler)); // client sends message ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify events - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{500})); + ASSERT_NO_THROW(active_poller.wait(std::chrono::milliseconds{500})); ASSERT_TRUE(s.events == ZMQ_POLLIN); ASSERT_EQ(s.events, ZMQ_POLLIN); // Modify server socket with pollout flag - ASSERT_NO_THROW(poller.modify(s.server, ZMQ_POLLIN | ZMQ_POLLOUT)); - ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{500})); + ASSERT_NO_THROW(active_poller.modify(s.server, ZMQ_POLLIN | ZMQ_POLLOUT)); + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{500})); ASSERT_EQ(s.events, ZMQ_POLLIN | ZMQ_POLLOUT); } -TEST(poller, wait_one_return) +TEST(active_poller, wait_one_return) { // Setup server and client server_client_setup s; int count = 0; - // Setup poller - zmq::poller_t poller; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, [&count](short) { + // Setup active_poller + zmq::active_poller_t active_poller; + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN, [&count](short) { ++count; })); @@ -356,32 +356,32 @@ TEST(poller, wait_one_return) ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify events - ASSERT_EQ(1, poller.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{500})); ASSERT_EQ(1u, count); } -TEST(poller, wait_on_move_constructed_poller) +TEST(active_poller, wait_on_move_constructed_active_poller) { server_client_setup s; ASSERT_NO_THROW (s.client.send ("Hi")); - zmq::poller_t a; - zmq::poller_t::handler_t handler; + zmq::active_poller_t a; + zmq::active_poller_t::handler_t handler; ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); - zmq::poller_t b {std::move (a)}; + zmq::active_poller_t b {std::move (a)}; ASSERT_EQ(1u, b.size ()); /// \todo the actual error code should be checked ASSERT_THROW(a.wait(std::chrono::milliseconds{10}), zmq::error_t); ASSERT_TRUE (b.wait (std::chrono::milliseconds {-1})); } -TEST(poller, wait_on_move_assigned_poller) +TEST(active_poller, wait_on_move_assigned_active_poller) { server_client_setup s; ASSERT_NO_THROW (s.client.send ("Hi")); - zmq::poller_t a; - zmq::poller_t::handler_t handler; + zmq::active_poller_t a; + zmq::active_poller_t::handler_t handler; ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); - zmq::poller_t b; + zmq::active_poller_t b; b = {std::move (a)}; ASSERT_EQ(1u, b.size ()); /// \todo the actual error code should be checked @@ -389,13 +389,13 @@ TEST(poller, wait_on_move_assigned_poller) ASSERT_TRUE (b.wait (std::chrono::milliseconds {-1})); } -TEST(poller, received_on_move_constructed_poller) +TEST(active_poller, received_on_move_constructed_active_poller) { // Setup server and client server_client_setup s; int count = 0; - // Setup poller a - zmq::poller_t a; + // Setup active_poller a + zmq::active_poller_t a; ASSERT_NO_THROW(a.add(s.server, ZMQ_POLLIN, [&count](short) { ++count; })); @@ -404,8 +404,8 @@ TEST(poller, received_on_move_constructed_poller) // wait for message and verify it is received ASSERT_EQ(1, a.wait(std::chrono::milliseconds{500})); ASSERT_EQ(1u, count); - // Move construct poller b - zmq::poller_t b{std::move(a)}; + // Move construct active_poller b + zmq::active_poller_t b{std::move(a)}; // client sends message again ASSERT_NO_THROW(s.client.send("Hi")); // wait for message and verify it is received @@ -414,7 +414,7 @@ TEST(poller, received_on_move_constructed_poller) } -TEST(poller, remove_from_handler) +TEST(active_poller, remove_from_handler) { constexpr auto ITER_NO = 10; @@ -423,18 +423,18 @@ TEST(poller, remove_from_handler) for (auto i = 0; i < ITER_NO; ++i) setup_list.emplace_back (server_client_setup{}); - // Setup poller - zmq::poller_t poller; + // Setup active_poller + zmq::active_poller_t active_poller; int count = 0; for (auto i = 0; i < ITER_NO; ++i) { - ASSERT_NO_THROW(poller.add(setup_list[i].server, ZMQ_POLLIN, [&,i](short events) { + ASSERT_NO_THROW(active_poller.add(setup_list[i].server, ZMQ_POLLIN, [&,i](short events) { ASSERT_EQ(events, ZMQ_POLLIN); - poller.remove(setup_list[ITER_NO-i-1].server); - ASSERT_EQ(ITER_NO-i-1, poller.size()); + active_poller.remove(setup_list[ITER_NO-i-1].server); + ASSERT_EQ(ITER_NO-i-1, active_poller.size()); })); ++count; } - ASSERT_EQ(ITER_NO, poller.size()); + ASSERT_EQ(ITER_NO, active_poller.size()); // Clients send messages for (auto & s : setup_list) { ASSERT_NO_THROW(s.client.send("Hi")); @@ -447,7 +447,7 @@ TEST(poller, remove_from_handler) } // Fire all handlers in one wait - ASSERT_EQ(ITER_NO, poller.wait (std::chrono::milliseconds{-1})); + ASSERT_EQ(ITER_NO, active_poller.wait (std::chrono::milliseconds{-1})); ASSERT_EQ(ITER_NO, count); } diff --git a/zmq.hpp b/zmq.hpp index 3d2f7bc..983d0ff 100644 --- a/zmq.hpp +++ b/zmq.hpp @@ -1019,7 +1019,7 @@ namespace zmq #if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) template - class base_poller_t + class poller_t { public: void add (zmq::socket_t &socket, short events, T *user_data) @@ -1078,99 +1078,8 @@ namespace zmq } }; }; - - class poller_t - { - public: - poller_t () = default; - ~poller_t () = default; - - poller_t(const poller_t&) = delete; - poller_t &operator=(const poller_t&) = delete; - - poller_t(poller_t&& src) = default; - poller_t &operator=(poller_t&& src) = default; - - using handler_t = std::function; - - void add (zmq::socket_t &socket, short events, handler_t handler) - { - auto it = decltype (handlers)::iterator {}; - auto inserted = bool {}; - std::tie(it, inserted) = handlers.emplace (static_cast(socket), std::make_shared (std::move (handler))); - try - { - base_poller.add (socket, events, inserted && *(it->second) ? it->second.get() : nullptr); - need_rebuild |= inserted; - } - catch (const zmq::error_t&) - { - // rollback - if (inserted) - { - handlers.erase (static_cast(socket)); - } - throw; - } - } - - void remove (zmq::socket_t &socket) - { - base_poller.remove (socket); - handlers.erase (static_cast(socket)); - need_rebuild = true; - } - - void modify (zmq::socket_t &socket, short events) - { - base_poller.modify (socket, events); - } - - int wait (std::chrono::milliseconds timeout) - { - if (need_rebuild) { - poller_events.clear (); - poller_handlers.clear (); - poller_events.reserve (handlers.size ()); - poller_handlers.reserve (handlers.size ()); - for (const auto &handler : handlers) { - poller_events.emplace_back (zmq_poller_event_t {}); - poller_handlers.push_back (handler.second); - } - need_rebuild = false; - } - const int count = base_poller.wait_all (poller_events, timeout); - if (count != 0) { - std::for_each (poller_events.begin (), poller_events.begin () + count, - [](zmq_poller_event_t& event) { - if (event.user_data != NULL) - (*reinterpret_cast (event.user_data)) (event.events); - }); - } - return count; - } - - bool empty () const - { - return handlers.empty (); - } - - size_t size () const - { - return handlers.size (); - } - - private: - bool need_rebuild {false}; - - base_poller_t base_poller {}; - std::unordered_map> handlers {}; - std::vector poller_events {}; - std::vector> poller_handlers {}; - }; // class poller_t #endif // defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) - inline std::ostream& operator<<(std::ostream& os, const message_t& msg) { return os << msg.str(); diff --git a/zmq_addon.hpp b/zmq_addon.hpp index bc9ae07..9bf8286 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -426,6 +426,99 @@ inline std::ostream& operator<<(std::ostream& os, const multipart_t& msg) return os << msg.str(); } +#if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) + class active_poller_t + { + public: + active_poller_t () = default; + ~active_poller_t () = default; + + active_poller_t(const active_poller_t&) = delete; + active_poller_t &operator=(const active_poller_t&) = delete; + + active_poller_t(active_poller_t&& src) = default; + active_poller_t &operator=(active_poller_t&& src) = default; + + using handler_t = std::function; + + void add (zmq::socket_t &socket, short events, handler_t handler) + { + auto it = decltype (handlers)::iterator {}; + auto inserted = bool {}; + std::tie(it, inserted) = handlers.emplace (static_cast(socket), std::make_shared (std::move (handler))); + try + { + base_poller.add (socket, events, inserted && *(it->second) ? it->second.get() : nullptr); + need_rebuild |= inserted; + } + catch (const zmq::error_t&) + { + // rollback + if (inserted) + { + handlers.erase (static_cast(socket)); + } + throw; + } + } + + void remove (zmq::socket_t &socket) + { + base_poller.remove (socket); + handlers.erase (static_cast(socket)); + need_rebuild = true; + } + + void modify (zmq::socket_t &socket, short events) + { + base_poller.modify (socket, events); + } + + int wait (std::chrono::milliseconds timeout) + { + if (need_rebuild) { + poller_events.clear (); + poller_handlers.clear (); + poller_events.reserve (handlers.size ()); + poller_handlers.reserve (handlers.size ()); + for (const auto &handler : handlers) { + poller_events.emplace_back (zmq_poller_event_t {}); + poller_handlers.push_back (handler.second); + } + need_rebuild = false; + } + const int count = base_poller.wait_all (poller_events, timeout); + if (count != 0) { + std::for_each (poller_events.begin (), poller_events.begin () + count, + [](zmq_poller_event_t& event) { + if (event.user_data != NULL) + (*reinterpret_cast (event.user_data)) (event.events); + }); + } + return count; + } + + bool empty () const + { + return handlers.empty (); + } + + size_t size () const + { + return handlers.size (); + } + + private: + bool need_rebuild {false}; + + poller_t base_poller {}; + std::unordered_map> handlers {}; + std::vector poller_events {}; + std::vector> poller_handlers {}; + }; // class active_poller_t +#endif // defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) + + } // namespace zmq #endif // __ZMQ_ADDON_HPP_INCLUDED__ From ee1cc9a791658c1ccb3b625ddf3a0c8da4108d1f Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 11:09:09 +0200 Subject: [PATCH 5/9] Problem: unnecessary heap allocations for test subject Solution: change to local variables --- tests/active_poller.cpp | 50 +++++++++++++++++++---------------------- 1 file changed, 23 insertions(+), 27 deletions(-) diff --git a/tests/active_poller.cpp b/tests/active_poller.cpp index 87b4d95..65b23cc 100644 --- a/tests/active_poller.cpp +++ b/tests/active_poller.cpp @@ -17,27 +17,25 @@ static_assert(!std::is_copy_assignable::value, "active_act TEST(active_poller, move_construct_empty) { - std::unique_ptr a {new zmq::active_poller_t}; - ASSERT_TRUE(a->empty ()); - zmq::active_poller_t b = std::move (*a); + zmq::active_poller_t a; + ASSERT_TRUE(a.empty ()); + zmq::active_poller_t b = std::move (a); ASSERT_TRUE(b.empty ()); - ASSERT_EQ(0u, a->size ()); + ASSERT_EQ(0u, a.size ()); ASSERT_EQ(0u, b.size ()); - a.reset (); } TEST(active_poller, move_assign_empty) { - std::unique_ptr a{new zmq::active_poller_t}; - ASSERT_TRUE(a->empty()); + zmq::active_poller_t a; + ASSERT_TRUE(a.empty()); zmq::active_poller_t b; ASSERT_TRUE(b.empty()); - b = std::move(*a); - ASSERT_EQ(0u, a->size ()); + b = std::move(a); + ASSERT_EQ(0u, a.size ()); ASSERT_EQ(0u, b.size ()); - ASSERT_TRUE(a->empty()); + ASSERT_TRUE(a.empty()); ASSERT_TRUE(b.empty()); - a.reset (); } TEST(active_poller, move_construct_non_empty) @@ -45,16 +43,15 @@ TEST(active_poller, move_construct_non_empty) zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - std::unique_ptr a{new zmq::active_poller_t}; - a->add(socket, ZMQ_POLLIN, [](short) {}); - ASSERT_FALSE(a->empty ()); - ASSERT_EQ(1u, a->size ()); - zmq::active_poller_t b = std::move (*a); - ASSERT_TRUE(a->empty ()); - ASSERT_EQ(0u, a->size ()); + zmq::active_poller_t a; + a.add(socket, ZMQ_POLLIN, [](short) {}); + ASSERT_FALSE(a.empty ()); + ASSERT_EQ(1u, a.size ()); + zmq::active_poller_t b = std::move (a); + ASSERT_TRUE(a.empty ()); + ASSERT_EQ(0u, a.size ()); ASSERT_FALSE(b.empty ()); ASSERT_EQ(1u, b.size ()); - a.reset (); } TEST(active_poller, move_assign_non_empty) @@ -62,17 +59,16 @@ TEST(active_poller, move_assign_non_empty) zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; - std::unique_ptr a{new zmq::active_poller_t}; - a->add(socket, ZMQ_POLLIN, [](short) {}); - ASSERT_FALSE(a->empty()); - ASSERT_EQ(1u, a->size ()); + zmq::active_poller_t a; + a.add(socket, ZMQ_POLLIN, [](short) {}); + ASSERT_FALSE(a.empty()); + ASSERT_EQ(1u, a.size ()); zmq::active_poller_t b; - b = std::move(*a); - ASSERT_TRUE(a->empty ()); - ASSERT_EQ(0u, a->size ()); + b = std::move(a); + ASSERT_TRUE(a.empty ()); + ASSERT_EQ(0u, a.size ()); ASSERT_FALSE(b.empty ()); ASSERT_EQ(1u, b.size ()); - a.reset (); } TEST(active_poller, add_handler) From f700e5d6b00eb61c110433db4b1ab3379fb4f702 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 11:18:21 +0200 Subject: [PATCH 6/9] Problem: deprecation warning in multipart_t Solution: use non-deprecated operator!= instead --- zmq_addon.hpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/zmq_addon.hpp b/zmq_addon.hpp index 9bf8286..fd2eb8b 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -408,7 +408,7 @@ public: if (size() != other->size()) return false; for (size_t i = 0; i < size(); i++) - if (!peek(i)->equal(other->peek(i))) + if (*peek(i) != *other->peek(i)) return false; return true; } @@ -419,13 +419,13 @@ private: void operator=(const multipart_t& other) ZMQ_DELETED_FUNCTION; }; // class multipart_t -#endif // ZMQ_HAS_RVALUE_REFS - inline std::ostream& operator<<(std::ostream& os, const multipart_t& msg) { return os << msg.str(); } +#endif // ZMQ_HAS_RVALUE_REFS + #if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) class active_poller_t { From 3251d056368375642212dfc4316e198464f1d668 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 11:27:05 +0200 Subject: [PATCH 7/9] Problem: creation of poller_events is unnecessarily complex Solution: simplify code --- zmq_addon.hpp | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/zmq_addon.hpp b/zmq_addon.hpp index fd2eb8b..39d7d68 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -477,12 +477,10 @@ inline std::ostream& operator<<(std::ostream& os, const multipart_t& msg) int wait (std::chrono::milliseconds timeout) { if (need_rebuild) { - poller_events.clear (); + poller_events.resize (handlers.size ()); poller_handlers.clear (); - poller_events.reserve (handlers.size ()); poller_handlers.reserve (handlers.size ()); for (const auto &handler : handlers) { - poller_events.emplace_back (zmq_poller_event_t {}); poller_handlers.push_back (handler.second); } need_rebuild = false; From 762b0131e83447e6f8a633d550784fb8ce12c35a Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 11:43:41 +0200 Subject: [PATCH 8/9] Problem: no tests for poller_t Solution: add tests --- tests/CMakeLists.txt | 1 + tests/active_poller.cpp | 49 +------ tests/poller.cpp | 294 ++++++++++++++++++++++++++++++++++++++++ tests/testutil.hpp | 49 +++++++ 4 files changed, 351 insertions(+), 42 deletions(-) create mode 100644 tests/poller.cpp create mode 100644 tests/testutil.hpp diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 30b86b0..1ed035e 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -25,6 +25,7 @@ add_executable( message.cpp context.cpp socket.cpp + poller.cpp active_poller.cpp multipart.cpp ) diff --git a/tests/active_poller.cpp b/tests/active_poller.cpp index 65b23cc..8cf9de1 100644 --- a/tests/active_poller.cpp +++ b/tests/active_poller.cpp @@ -1,6 +1,7 @@ -#include #include +#include "testutil.hpp" + #if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API) #include @@ -82,6 +83,9 @@ TEST(active_poller, add_handler) TEST(active_poller, add_handler_invalid_events_type) { + /// \todo is it good that this is accepted? should probably already be + /// checked by zmq_poller_add/modify in libzmq: + /// https://github.com/zeromq/libzmq/issues/3088 zmq::context_t context; zmq::socket_t socket{context, zmq::socket_type::router}; zmq::active_poller_t active_poller; @@ -138,54 +142,15 @@ TEST(active_poller, remove_registered_non_empty) } namespace { - -class loopback_ip4_binder +struct server_client_setup : common_server_client_setup { -public: - loopback_ip4_binder(zmq::socket_t &socket) { bind(socket); } - std::string endpoint() { return endpoint_; } -private: - // Helper function used in constructor - // as Gtest allows ASSERT_* only in void returning functions - // and constructor/destructor are not. - void bind(zmq::socket_t &socket) - { - ASSERT_NO_THROW(socket.bind("tcp://127.0.0.1:*")); - std::array endpoint{}; - size_t endpoint_size = endpoint.size(); - ASSERT_NO_THROW(socket.getsockopt(ZMQ_LAST_ENDPOINT, endpoint.data(), - &endpoint_size)); - ASSERT_TRUE(endpoint_size < endpoint.size()); - endpoint_ = std::string{endpoint.data()}; - } - std::string endpoint_; -}; - -struct server_client_setup -{ - server_client_setup () - { - init (); - } - - void init() - { - endpoint = loopback_ip4_binder {server}.endpoint (); - ASSERT_NO_THROW (client.connect (endpoint)); - } - zmq::active_poller_t::handler_t handler = [&](short e) { events = e; }; - zmq::context_t context; - zmq::socket_t server {context, zmq::socket_type::server}; - zmq::socket_t client {context, zmq::socket_type::client}; - std::string endpoint; short events = 0; }; - -} //namespace +} TEST(active_poller, poll_basic) { diff --git a/tests/poller.cpp b/tests/poller.cpp new file mode 100644 index 0000000..a777be3 --- /dev/null +++ b/tests/poller.cpp @@ -0,0 +1,294 @@ +#include "testutil.hpp" + +#if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API) + +#include +#include + +TEST(poller, create_destroy) +{ + zmq::poller_t<> poller; +} + +static_assert(!std::is_copy_constructible>::value, "poller_t should not be copy-constructible"); +static_assert(!std::is_copy_assignable>::value, "poller_t should not be copy-assignable"); + +TEST(poller, move_construct_empty) +{ + zmq::poller_t<> a; + zmq::poller_t<> b = std::move (a); +} + +TEST(poller, move_assign_empty) +{ + zmq::poller_t<> a; + zmq::poller_t<> b; + b = std::move(a); +} + +TEST(poller, move_construct_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + + zmq::poller_t<> a; + a.add(socket, ZMQ_POLLIN, nullptr); + zmq::poller_t<> b = std::move (a); +} + +TEST(poller, move_assign_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + + zmq::poller_t<> a; + a.add(socket, ZMQ_POLLIN, nullptr); + zmq::poller_t<> b; + b = std::move(a); +} + +TEST(poller, add_nullptr) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + ASSERT_NO_THROW(poller.add(socket, ZMQ_POLLIN, nullptr)); +} + +TEST(poller, add_non_nullptr) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + int i; + ASSERT_NO_THROW(poller.add(socket, ZMQ_POLLIN, &i)); +} + +TEST(poller, add_handler_invalid_events_type) +{ + /// \todo is it good that this is accepted? should probably already be + /// checked by zmq_poller_add/modify in libzmq: + /// https://github.com/zeromq/libzmq/issues/3088 + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + short invalid_events_type = 2 << 10; + ASSERT_NO_THROW(poller.add(socket, invalid_events_type, nullptr)); +} + +TEST(poller, add_handler_twice_throws) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + poller.add(socket, ZMQ_POLLIN, nullptr); + /// \todo the actual error code should be checked + ASSERT_THROW(poller.add(socket, ZMQ_POLLIN, nullptr), zmq::error_t); +} + +TEST(poller, wait_with_no_handlers_throws) +{ + zmq::poller_t<> poller; + std::vector events; + /// \todo the actual error code should be checked + ASSERT_THROW(poller.wait_all(events, std::chrono::milliseconds{10}), zmq::error_t); +} + +TEST(poller, remove_unregistered_throws) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + /// \todo the actual error code should be checked + ASSERT_THROW(poller.remove(socket), zmq::error_t); +} + +TEST(poller, remove_registered_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + poller.add(socket, ZMQ_POLLIN, nullptr); + ASSERT_NO_THROW(poller.remove(socket)); +} + +TEST(poller, remove_registered_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::poller_t<> poller; + poller.add(socket, ZMQ_POLLIN, nullptr); + ASSERT_NO_THROW(poller.remove(socket)); +} + +TEST(poller, poll_basic) +{ + common_server_client_setup s; + + ASSERT_NO_THROW(s.client.send("Hi")); + + zmq::poller_t poller; + std::vector events{1}; + int i = 0; + ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, &i)); + ASSERT_EQ(1, poller.wait_all(events, std::chrono::milliseconds{-1})); + ASSERT_EQ(s.server, events[0].socket); + ASSERT_EQ(&i, events[0].user_data); +} + +TEST(poller, add_invalid_socket_throws) +{ + zmq::context_t context; + zmq::poller_t<> poller; + zmq::socket_t a {context, zmq::socket_type::router}; + zmq::socket_t b {std::move (a)}; + ASSERT_THROW (poller.add (a, ZMQ_POLLIN, nullptr), + zmq::error_t); +} + +TEST(poller, remove_invalid_socket_throws) +{ + zmq::context_t context; + zmq::socket_t socket {context, zmq::socket_type::router}; + zmq::poller_t<> poller; + ASSERT_NO_THROW (poller.add (socket, ZMQ_POLLIN, nullptr)); + std::vector sockets; + sockets.emplace_back (std::move (socket)); + ASSERT_THROW (poller.remove (socket), zmq::error_t); + ASSERT_NO_THROW (poller.remove (sockets[0])); +} + +TEST(poller, modify_empty_throws) +{ + zmq::context_t context; + zmq::socket_t socket {context, zmq::socket_type::push}; + zmq::poller_t<> poller; + ASSERT_THROW (poller.modify (socket, ZMQ_POLLIN), zmq::error_t); +} + +TEST(poller, modify_invalid_socket_throws) +{ + zmq::context_t context; + zmq::socket_t a {context, zmq::socket_type::push}; + zmq::socket_t b {std::move (a)}; + zmq::poller_t<> poller; + ASSERT_THROW (poller.modify (a, ZMQ_POLLIN), zmq::error_t); +} + +TEST(poller, modify_not_added_throws) +{ + zmq::context_t context; + zmq::socket_t a {context, zmq::socket_type::push}; + zmq::socket_t b {context, zmq::socket_type::push}; + zmq::poller_t<> poller; + ASSERT_NO_THROW (poller.add (a, ZMQ_POLLIN, nullptr)); + ASSERT_THROW (poller.modify (b, ZMQ_POLLIN), zmq::error_t); +} + +TEST(poller, modify_simple) +{ + zmq::context_t context; + zmq::socket_t a {context, zmq::socket_type::push}; + zmq::poller_t<> poller; + ASSERT_NO_THROW (poller.add (a, ZMQ_POLLIN, nullptr)); + ASSERT_NO_THROW (poller.modify (a, ZMQ_POLLIN|ZMQ_POLLOUT)); +} + +TEST(poller, poll_client_server) +{ + // Setup server and client + common_server_client_setup s; + + // Setup poller + zmq::poller_t<> poller; + ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, s.server)); + + // client sends message + ASSERT_NO_THROW(s.client.send("Hi")); + + // wait for message and verify events + std::vector events(1); + ASSERT_EQ(1, poller.wait_all(events, std::chrono::milliseconds{500})); + ASSERT_EQ(ZMQ_POLLIN, events[0].events); + + // Modify server socket with pollout flag + ASSERT_NO_THROW(poller.modify(s.server, ZMQ_POLLIN | ZMQ_POLLOUT)); + ASSERT_EQ(1, poller.wait_all(events, std::chrono::milliseconds{500})); + ASSERT_EQ(ZMQ_POLLIN | ZMQ_POLLOUT, events[0].events); +} + +TEST(poller, wait_one_return) +{ + // Setup server and client + common_server_client_setup s; + + // Setup poller + zmq::poller_t<> poller; + ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, nullptr)); + + // client sends message + ASSERT_NO_THROW(s.client.send("Hi")); + + // wait for message and verify events + std::vector events(1); + ASSERT_EQ(1, poller.wait_all(events, std::chrono::milliseconds{500})); +} + +TEST(poller, wait_on_move_constructed_poller) +{ + common_server_client_setup s; + ASSERT_NO_THROW (s.client.send ("Hi")); + zmq::poller_t<> a; + ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, nullptr)); + zmq::poller_t<> b {std::move (a)}; + std::vector events(1); + /// \todo the actual error code should be checked + ASSERT_THROW(a.wait_all (events, std::chrono::milliseconds{10}), zmq::error_t); + ASSERT_EQ (1, b.wait_all (events, std::chrono::milliseconds {-1})); +} + +TEST(poller, wait_on_move_assigned_poller) +{ + common_server_client_setup s; + ASSERT_NO_THROW (s.client.send ("Hi")); + zmq::poller_t<> a; + ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, nullptr)); + zmq::poller_t<> b; + b = {std::move (a)}; + /// \todo the actual error code should be checked + std::vector events(1); + ASSERT_THROW(a.wait_all (events, std::chrono::milliseconds{10}), zmq::error_t); + ASSERT_EQ (1, b.wait_all (events, std::chrono::milliseconds {-1})); +} + +TEST(poller, remove_from_handler) +{ + constexpr auto ITER_NO = 10; + + // Setup servers and clients + std::vector setup_list; + for (auto i = 0; i < ITER_NO; ++i) + setup_list.emplace_back (common_server_client_setup{}); + + // Setup poller + zmq::poller_t<> poller; + for (auto i = 0; i < ITER_NO; ++i) { + ASSERT_NO_THROW(poller.add(setup_list[i].server, ZMQ_POLLIN, nullptr)); + } + // Clients send messages + for (auto & s : setup_list) { + ASSERT_NO_THROW(s.client.send("Hi")); + } + + // Wait for all servers to receive a message + for (auto & s : setup_list) { + zmq::pollitem_t items [] = { { s.server, 0, ZMQ_POLLIN, 0 } }; + zmq::poll (&items [0], 1); + } + + // Fire all handlers in one wait + std::vector events(ITER_NO); + ASSERT_EQ(ITER_NO, poller.wait_all (events, std::chrono::milliseconds{-1})); +} + +#endif diff --git a/tests/testutil.hpp b/tests/testutil.hpp new file mode 100644 index 0000000..1ed5432 --- /dev/null +++ b/tests/testutil.hpp @@ -0,0 +1,49 @@ +#pragma once + +#include +#include + +#if defined(ZMQ_CPP11) +#include + +class loopback_ip4_binder +{ +public: + loopback_ip4_binder(zmq::socket_t &socket) { bind(socket); } + std::string endpoint() { return endpoint_; } +private: + // Helper function used in constructor + // as Gtest allows ASSERT_* only in void returning functions + // and constructor/destructor are not. + void bind(zmq::socket_t &socket) + { + ASSERT_NO_THROW(socket.bind("tcp://127.0.0.1:*")); + std::array endpoint{}; + size_t endpoint_size = endpoint.size(); + ASSERT_NO_THROW(socket.getsockopt(ZMQ_LAST_ENDPOINT, endpoint.data(), + &endpoint_size)); + ASSERT_TRUE(endpoint_size < endpoint.size()); + endpoint_ = std::string{endpoint.data()}; + } + std::string endpoint_; +}; + +struct common_server_client_setup +{ + common_server_client_setup () + { + init (); + } + + void init() + { + endpoint = loopback_ip4_binder {server}.endpoint (); + ASSERT_NO_THROW (client.connect (endpoint)); + } + + zmq::context_t context; + zmq::socket_t server {context, zmq::socket_type::server}; + zmq::socket_t client {context, zmq::socket_type::client}; + std::string endpoint; +}; +#endif From 421d66c97e43a438961ad98f2661bec342dc2030 Mon Sep 17 00:00:00 2001 From: Simon Giesecke Date: Fri, 11 May 2018 12:30:34 +0200 Subject: [PATCH 9/9] Problem: redundant assertion Solution: removed redundant assertion --- tests/active_poller.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/active_poller.cpp b/tests/active_poller.cpp index 8cf9de1..1a01ebb 100644 --- a/tests/active_poller.cpp +++ b/tests/active_poller.cpp @@ -291,7 +291,6 @@ TEST(active_poller, poll_client_server) // wait for message and verify events ASSERT_NO_THROW(active_poller.wait(std::chrono::milliseconds{500})); - ASSERT_TRUE(s.events == ZMQ_POLLIN); ASSERT_EQ(s.events, ZMQ_POLLIN); // Modify server socket with pollout flag