#include #include #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) { 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 { 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::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) { 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_TRUE(s.events == ZMQ_POLLIN); 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