diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index f664071..1ed035e 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -26,6 +26,8 @@ add_executable( context.cpp socket.cpp poller.cpp + active_poller.cpp + multipart.cpp ) target_link_libraries( diff --git a/tests/active_poller.cpp b/tests/active_poller.cpp new file mode 100644 index 0000000..1a01ebb --- /dev/null +++ b/tests/active_poller.cpp @@ -0,0 +1,414 @@ +#include + +#include "testutil.hpp" + +#if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API) + +#include +#include + +TEST(active_poller, create_destroy) +{ + zmq::active_poller_t active_poller; + ASSERT_TRUE(active_poller.empty ()); +} + +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(active_poller, move_construct_empty) +{ + 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, b.size ()); +} + +TEST(active_poller, move_assign_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 ()); + ASSERT_EQ(0u, b.size ()); + ASSERT_TRUE(a.empty()); + ASSERT_TRUE(b.empty()); +} + +TEST(active_poller, move_construct_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + + 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 ()); +} + +TEST(active_poller, move_assign_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + + 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 ()); + ASSERT_FALSE(b.empty ()); + ASSERT_EQ(1u, b.size ()); +} + +TEST(active_poller, add_handler) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::active_poller_t active_poller; + zmq::active_poller_t::handler_t handler; + ASSERT_NO_THROW(active_poller.add(socket, ZMQ_POLLIN, 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; + zmq::active_poller_t::handler_t handler; + short invalid_events_type = 2 << 10; + ASSERT_NO_THROW(active_poller.add(socket, invalid_events_type, handler)); + ASSERT_FALSE(active_poller.empty ()); + ASSERT_EQ(1u, active_poller.size ()); +} + +TEST(active_poller, add_handler_twice_throws) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + 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(active_poller.add(socket, ZMQ_POLLIN, handler), zmq::error_t); +} + +TEST(active_poller, wait_with_no_handlers_throws) +{ + zmq::active_poller_t active_poller; + /// \todo the actual error code should be checked + ASSERT_THROW(active_poller.wait(std::chrono::milliseconds{10}), zmq::error_t); +} + +TEST(active_poller, remove_unregistered_throws) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::active_poller_t active_poller; + /// \todo the actual error code should be checked + ASSERT_THROW(active_poller.remove(socket), zmq::error_t); +} + +TEST(active_poller, remove_registered_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + 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(active_poller, remove_registered_non_empty) +{ + zmq::context_t context; + zmq::socket_t socket{context, zmq::socket_type::router}; + zmq::active_poller_t active_poller; + active_poller.add(socket, ZMQ_POLLIN, [](short) {}); + ASSERT_NO_THROW(active_poller.remove(socket)); +} + +namespace { +struct server_client_setup : common_server_client_setup +{ + zmq::active_poller_t::handler_t handler = [&](short e) { + events = e; + }; + + short events = 0; +}; +} + +TEST(active_poller, poll_basic) +{ + server_client_setup s; + + ASSERT_NO_THROW(s.client.send("Hi")); + + zmq::active_poller_t active_poller; + bool message_received = false; + zmq::active_poller_t::handler_t handler = [&message_received](short events) { + ASSERT_TRUE(0 != (events & ZMQ_POLLIN)); + message_received = true; + }; + 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(active_poller, client_server) +{ + const std::string send_msg = "Hi"; + + // Setup server and client + server_client_setup s; + + // Setup active_poller + zmq::active_poller_t active_poller; + short events; + 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 + std::string recv_msg(zmq_msg.data(), + zmq_msg.size()); + ASSERT_EQ(send_msg, recv_msg); + } else if (0 != (e & ~ZMQ_POLLOUT)) { + ASSERT_TRUE(false) << "Unexpected event type " << events; + } + events = e; + }; + + 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, active_poller.wait(std::chrono::milliseconds{-1})); + ASSERT_EQ(events, ZMQ_POLLIN); + + // Re-add server socket with pollout flag + 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(active_poller, add_invalid_socket_throws) +{ + zmq::context_t context; + zmq::active_poller_t active_poller; + zmq::socket_t a {context, zmq::socket_type::router}; + zmq::socket_t b {std::move (a)}; + ASSERT_THROW (active_poller.add (a, ZMQ_POLLIN, zmq::active_poller_t::handler_t {}), + zmq::error_t); +} + +TEST(active_poller, remove_invalid_socket_throws) +{ + zmq::context_t context; + zmq::socket_t socket {context, zmq::socket_type::router}; + 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 (active_poller.remove (socket), zmq::error_t); + ASSERT_EQ (1u, active_poller.size ()); +} + +TEST(active_poller, wait_on_added_empty_handler) +{ + server_client_setup s; + ASSERT_NO_THROW (s.client.send ("Hi")); + 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(active_poller, modify_empty_throws) +{ + zmq::context_t context; + zmq::socket_t socket {context, zmq::socket_type::push}; + zmq::active_poller_t active_poller; + ASSERT_THROW (active_poller.modify (socket, ZMQ_POLLIN), zmq::error_t); +} + +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::active_poller_t active_poller; + ASSERT_THROW (active_poller.modify (a, ZMQ_POLLIN), zmq::error_t); +} + +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::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(active_poller, modify_simple) +{ + zmq::context_t context; + zmq::socket_t a {context, zmq::socket_type::push}; + 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(active_poller, poll_client_server) +{ + // Setup server and client + server_client_setup s; + + // 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(active_poller.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(s.events, ZMQ_POLLIN); + + // Modify server socket with pollout flag + 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(active_poller, wait_one_return) +{ + // Setup server and client + server_client_setup s; + + int count = 0; + + // Setup active_poller + zmq::active_poller_t active_poller; + ASSERT_NO_THROW(active_poller.add(s.server, ZMQ_POLLIN, [&count](short) { + ++count; + })); + + // client sends message + ASSERT_NO_THROW(s.client.send("Hi")); + + // wait for message and verify events + ASSERT_EQ(1, active_poller.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(1u, count); +} + +TEST(active_poller, wait_on_move_constructed_active_poller) +{ + server_client_setup s; + ASSERT_NO_THROW (s.client.send ("Hi")); + zmq::active_poller_t a; + zmq::active_poller_t::handler_t handler; + ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); + 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(active_poller, wait_on_move_assigned_active_poller) +{ + server_client_setup s; + ASSERT_NO_THROW (s.client.send ("Hi")); + zmq::active_poller_t a; + zmq::active_poller_t::handler_t handler; + ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, handler)); + zmq::active_poller_t b; + 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(active_poller, received_on_move_constructed_active_poller) +{ + // Setup server and client + server_client_setup s; + int count = 0; + // Setup active_poller a + zmq::active_poller_t a; + ASSERT_NO_THROW(a.add(s.server, ZMQ_POLLIN, [&count](short) { + ++count; + })); + // client sends message + ASSERT_NO_THROW(s.client.send("Hi")); + // wait for message and verify it is received + ASSERT_EQ(1, a.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(1u, count); + // 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 + ASSERT_EQ(1, b.wait(std::chrono::milliseconds{500})); + ASSERT_EQ(2u, count); +} + + +TEST(active_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 (server_client_setup{}); + + // Setup active_poller + zmq::active_poller_t active_poller; + int count = 0; + for (auto i = 0; i < ITER_NO; ++i) { + ASSERT_NO_THROW(active_poller.add(setup_list[i].server, ZMQ_POLLIN, [&,i](short events) { + ASSERT_EQ(events, ZMQ_POLLIN); + active_poller.remove(setup_list[ITER_NO-i-1].server); + ASSERT_EQ(ITER_NO-i-1, active_poller.size()); + })); + ++count; + } + ASSERT_EQ(ITER_NO, active_poller.size()); + // 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 + ASSERT_EQ(ITER_NO, active_poller.wait (std::chrono::milliseconds{-1})); + ASSERT_EQ(ITER_NO, count); +} + +#endif 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/tests/poller.cpp b/tests/poller.cpp index 18f1346..a777be3 100644 --- a/tests/poller.cpp +++ b/tests/poller.cpp @@ -1,5 +1,4 @@ -#include -#include +#include "testutil.hpp" #if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API) @@ -8,36 +7,23 @@ TEST(poller, create_destroy) { - zmq::poller_t poller; - ASSERT_TRUE(poller.empty ()); + 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"); +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) { - std::unique_ptr a {new zmq::poller_t}; - ASSERT_TRUE(a->empty ()); - zmq::poller_t b = std::move (*a); - ASSERT_TRUE(b.empty ()); - ASSERT_EQ(0u, a->size ()); - ASSERT_EQ(0u, b.size ()); - a.reset (); + zmq::poller_t<> a; + zmq::poller_t<> b = std::move (a); } TEST(poller, move_assign_empty) { - std::unique_ptr a{new zmq::poller_t}; - ASSERT_TRUE(a->empty()); - zmq::poller_t b; - ASSERT_TRUE(b.empty()); - b = std::move(*a); - ASSERT_EQ(0u, a->size ()); - ASSERT_EQ(0u, b.size ()); - ASSERT_TRUE(a->empty()); - ASSERT_TRUE(b.empty()); - a.reset (); + zmq::poller_t<> a; + zmq::poller_t<> b; + b = std::move(a); } TEST(poller, move_construct_non_empty) @@ -45,16 +31,9 @@ TEST(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}; - a->add(socket, ZMQ_POLLIN, [](short) {}); - ASSERT_FALSE(a->empty ()); - ASSERT_EQ(1u, a->size ()); - zmq::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 (); + zmq::poller_t<> a; + a.add(socket, ZMQ_POLLIN, nullptr); + zmq::poller_t<> b = std::move (a); } TEST(poller, move_assign_non_empty) @@ -62,63 +41,64 @@ TEST(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}; - a->add(socket, ZMQ_POLLIN, [](short) {}); - ASSERT_FALSE(a->empty()); - ASSERT_EQ(1u, a->size ()); - zmq::poller_t b; - b = std::move(*a); - ASSERT_TRUE(a->empty ()); - ASSERT_EQ(0u, a->size ()); - ASSERT_FALSE(b.empty ()); - ASSERT_EQ(1u, b.size ()); - a.reset (); + zmq::poller_t<> a; + a.add(socket, ZMQ_POLLIN, nullptr); + zmq::poller_t<> b; + b = std::move(a); } -TEST(poller, add_handler) +TEST(poller, add_nullptr) { 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::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; - zmq::poller_t::handler_t handler; + zmq::poller_t<> poller; 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(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; - zmq::poller_t::handler_t handler; - poller.add(socket, ZMQ_POLLIN, handler); + 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, handler), zmq::error_t); + ASSERT_THROW(poller.add(socket, ZMQ_POLLIN, nullptr), zmq::error_t); } TEST(poller, wait_with_no_handlers_throws) { - zmq::poller_t poller; + zmq::poller_t<> poller; + std::vector events; /// \todo the actual error code should be checked - ASSERT_THROW(poller.wait(std::chrono::milliseconds{10}), zmq::error_t); + 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; + zmq::poller_t<> poller; /// \todo the actual error code should be checked ASSERT_THROW(poller.remove(socket), zmq::error_t); } @@ -127,8 +107,8 @@ 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, zmq::poller_t::handler_t{}); + zmq::poller_t<> poller; + poller.add(socket, ZMQ_POLLIN, nullptr); ASSERT_NO_THROW(poller.remove(socket)); } @@ -136,124 +116,33 @@ 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, [](short) {}); + zmq::poller_t<> poller; + poller.add(socket, ZMQ_POLLIN, nullptr); ASSERT_NO_THROW(poller.remove(socket)); } -namespace { - -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 server_client_setup -{ - server_client_setup () - { - init (); - } - - void init() - { - endpoint = loopback_ip4_binder {server}.endpoint (); - ASSERT_NO_THROW (client.connect (endpoint)); - } - - zmq::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(poller, poll_basic) { - server_client_setup s; + common_server_client_setup s; ASSERT_NO_THROW(s.client.send("Hi")); - zmq::poller_t poller; - bool message_received = false; - zmq::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_NO_THROW(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) -{ - const std::string send_msg = "Hi"; - - // Setup server and client - server_client_setup s; - - // Setup poller - zmq::poller_t poller; - short events; - zmq::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 - std::string recv_msg(zmq_msg.data(), - zmq_msg.size()); - ASSERT_EQ(send_msg, recv_msg); - } else if (0 != (e & ~ZMQ_POLLOUT)) { - ASSERT_TRUE(false) << "Unexpected event type " << events; - } - events = e; - }; - - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler)); - - // client sends message - ASSERT_NO_THROW(s.client.send(send_msg)); - - ASSERT_NO_THROW(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(events, ZMQ_POLLOUT); + 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::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, zmq::poller_t::handler_t {}), + ASSERT_THROW (poller.add (a, ZMQ_POLLIN, nullptr), zmq::error_t); } @@ -261,30 +150,19 @@ 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, zmq::poller_t::handler_t {})); - ASSERT_EQ (1u, poller.size ()); + 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_EQ (1u, poller.size ()); -} - -TEST(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})); + 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; + zmq::poller_t<> poller; ASSERT_THROW (poller.modify (socket, ZMQ_POLLIN), zmq::error_t); } @@ -293,7 +171,7 @@ 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; + zmq::poller_t<> poller; ASSERT_THROW (poller.modify (a, ZMQ_POLLIN), zmq::error_t); } @@ -302,8 +180,8 @@ 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, zmq::poller_t::handler_t {})); + zmq::poller_t<> poller; + ASSERT_NO_THROW (poller.add (a, ZMQ_POLLIN, nullptr)); ASSERT_THROW (poller.modify (b, ZMQ_POLLIN), zmq::error_t); } @@ -311,127 +189,92 @@ 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, zmq::poller_t::handler_t {})); + 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 - server_client_setup s; + common_server_client_setup s; // Setup poller - zmq::poller_t poller; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, s.handler)); + 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 - ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{500})); - ASSERT_TRUE(s.events == ZMQ_POLLIN); - ASSERT_EQ(s.events, ZMQ_POLLIN); + 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_NO_THROW(poller.wait(std::chrono::milliseconds{500})); - ASSERT_EQ(s.events, 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 - server_client_setup s; - - int count = 0; + common_server_client_setup s; // Setup poller - zmq::poller_t poller; - ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, [&count](short) { - ++count; - })); + 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 - int result = poller.wait(std::chrono::milliseconds{500}); - ASSERT_EQ(count, result); + std::vector events(1); + ASSERT_EQ(1, poller.wait_all(events, std::chrono::milliseconds{500})); } TEST(poller, wait_on_move_constructed_poller) { - server_client_setup s; + common_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 {std::move (a)}; - ASSERT_EQ(1u, b.size ()); - ASSERT_NO_THROW (b.wait (std::chrono::milliseconds {-1})); + 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_assign_poller) +TEST(poller, wait_on_move_assigned_poller) { - server_client_setup s; + common_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 ()); + zmq::poller_t<> a; + ASSERT_NO_THROW (a.add (s.server, ZMQ_POLLIN, nullptr)); + zmq::poller_t<> b; 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 + 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, received_on_move_construced_poller) -{ - // Setup server and client - server_client_setup s; - int count = 0; - // Setup poller a - zmq::poller_t a; - ASSERT_NO_THROW(a.add(s.server, ZMQ_POLLIN, [&count](short) { - ++count; - })); - // 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(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(2u, count); -} - - TEST(poller, remove_from_handler) { constexpr auto ITER_NO = 10; // Setup servers and clients - std::vector setup_list; + std::vector setup_list; for (auto i = 0; i < ITER_NO; ++i) - setup_list.emplace_back (server_client_setup{}); + setup_list.emplace_back (common_server_client_setup{}); // Setup poller - zmq::poller_t poller; + zmq::poller_t<> poller; 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()); - })); + ASSERT_NO_THROW(poller.add(setup_list[i].server, ZMQ_POLLIN, nullptr)); } - ASSERT_EQ(ITER_NO, poller.size()); // Clients send messages for (auto & s : setup_list) { ASSERT_NO_THROW(s.client.send("Hi")); @@ -444,8 +287,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); + 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 diff --git a/zmq.hpp b/zmq.hpp index de9e131..983d0ff 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,75 +1018,42 @@ namespace zmq }; #if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER) + template 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) + void add (zmq::socket_t &socket, short events, T *user_data) { - 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; + if (0 != zmq_poller_add (poller_ptr.get (), static_cast(socket), user_data, events)) + { + throw error_t (); } - // 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; + if (0 != zmq_poller_remove (poller_ptr.get (), static_cast(socket))) + { + throw error_t (); } - throw error_t (); } void modify (zmq::socket_t &socket, short events) { - if (0 != zmq_poller_modify (poller_ptr.get (), socket.ptr, events)) + if (0 != zmq_poller_modify (poller_ptr.get (), static_cast(socket), events)) + { throw error_t (); - } - - 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; } + } + + 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) { - std::for_each (poller_events.begin (), poller_events.begin () + rc, - [](zmq_poller_event_t& event) { - if (event.user_data != NULL) - (*reinterpret_cast (event.user_data)) (event.events); - }); + if (rc > 0) return rc; - } + #if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 2, 3) if (zmq_errno () == EAGAIN) #else @@ -1097,17 +1063,6 @@ namespace zmq throw error_t (); } - - bool empty () const - { - return handlers.empty (); - } - - size_t size () const - { - return handlers.size (); - } - private: std::unique_ptr> poller_ptr { @@ -1122,14 +1077,9 @@ namespace zmq ZMQ_ASSERT (rc == 0); } }; - bool need_rebuild {false}; - 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 4dd6f05..39d7d68 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -408,190 +408,115 @@ 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; } - // 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; 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 + { + 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.resize (handlers.size ()); + poller_handlers.clear (); + poller_handlers.reserve (handlers.size ()); + for (const auto &handler : handlers) { + 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__