Problem: poller_t does not support modify

Solution: Added `modify` method based on `zmq_poller_modify` and test cases
covering it.
Reduced code duplication in existing test cases by introducing
`client_server_setup` helper struct.
This commit is contained in:
Pawel Kurdybacha 2018-04-24 21:16:01 +01:00
parent dcdf828230
commit 85e05b0c88
2 changed files with 104 additions and 43 deletions

View File

@ -113,7 +113,6 @@ TEST(poller, remove_unregistered_throws)
ASSERT_THROW(poller.remove(socket), zmq::error_t); ASSERT_THROW(poller.remove(socket), zmq::error_t);
} }
/// \todo this should lead to an exception instead
TEST(poller, remove_registered_empty) TEST(poller, remove_registered_empty)
{ {
zmq::context_t context; zmq::context_t context;
@ -156,19 +155,37 @@ private:
std::string endpoint_; 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::router};
zmq::socket_t client {context, zmq::socket_type::dealer};
std::string endpoint;
short events = 0;
};
} //namespace } //namespace
TEST(poller, poll_basic) TEST(poller, poll_basic)
{ {
zmq::context_t context; server_client_setup s;
zmq::socket_t vent{context, zmq::socket_type::push}; ASSERT_NO_THROW(s.client.send("Hi"));
auto endpoint = loopback_ip4_binder(vent).endpoint();
zmq::socket_t sink{context, zmq::socket_type::pull};
ASSERT_NO_THROW(sink.connect(endpoint));
ASSERT_NO_THROW(vent.send("Hi"));
zmq::poller_t poller; zmq::poller_t poller;
bool message_received = false; bool message_received = false;
@ -176,7 +193,7 @@ TEST(poller, poll_basic)
ASSERT_TRUE(0 != (events & ZMQ_POLLIN)); ASSERT_TRUE(0 != (events & ZMQ_POLLIN));
message_received = true; message_received = true;
}; };
ASSERT_NO_THROW(poller.add(sink, ZMQ_POLLIN, handler)); ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler));
ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1}));
ASSERT_TRUE(message_received); ASSERT_TRUE(message_received);
} }
@ -184,48 +201,41 @@ TEST(poller, poll_basic)
/// \todo this contains multiple test cases that should be split up /// \todo this contains multiple test cases that should be split up
TEST(poller, client_server) TEST(poller, client_server)
{ {
zmq::context_t context;
const std::string send_msg = "Hi"; const std::string send_msg = "Hi";
// Setup server // Setup server and client
zmq::socket_t server{context, zmq::socket_type::router}; server_client_setup s;
auto endpoint = loopback_ip4_binder(server).endpoint();
// Setup poller // Setup poller
zmq::poller_t poller; zmq::poller_t poller;
bool got_pollin = false; short events;
bool got_pollout = false; zmq::poller_t::handler_t handler = [&](short e) {
zmq::poller_t::handler_t handler = [&](short events) { if (0 != (e & ZMQ_POLLIN)) {
if (0 != (events & ZMQ_POLLIN)) {
zmq::message_t zmq_msg; zmq::message_t zmq_msg;
ASSERT_NO_THROW(server.recv(&zmq_msg)); // skip msg id ASSERT_NO_THROW(s.server.recv(&zmq_msg)); // skip msg id
ASSERT_NO_THROW(server.recv(&zmq_msg)); // get message ASSERT_NO_THROW(s.server.recv(&zmq_msg)); // get message
std::string recv_msg(zmq_msg.data<char>(), std::string recv_msg(zmq_msg.data<char>(),
zmq_msg.size()); zmq_msg.size());
ASSERT_EQ(send_msg, recv_msg); ASSERT_EQ(send_msg, recv_msg);
got_pollin = true; } else if (0 != (e & ~ZMQ_POLLOUT)) {
} else if (0 != (events & ZMQ_POLLOUT)) {
got_pollout = true;
} else {
ASSERT_TRUE(false) << "Unexpected event type " << events; ASSERT_TRUE(false) << "Unexpected event type " << events;
} }
events = e;
}; };
ASSERT_NO_THROW(poller.add(server, ZMQ_POLLIN, handler));
// Setup client and send message ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler));
zmq::socket_t client{context, zmq::socket_type::dealer};
ASSERT_NO_THROW(client.connect(endpoint)); // client sends message
ASSERT_NO_THROW(client.send(send_msg)); ASSERT_NO_THROW(s.client.send(send_msg));
ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1}));
ASSERT_TRUE(got_pollin); ASSERT_EQ(events, ZMQ_POLLIN);
ASSERT_FALSE(got_pollout);
// Re-add server socket with pollout flag // Re-add server socket with pollout flag
ASSERT_NO_THROW(poller.remove(server)); ASSERT_NO_THROW(poller.remove(s.server));
ASSERT_NO_THROW(poller.add(server, ZMQ_POLLIN | ZMQ_POLLOUT, handler)); ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN | ZMQ_POLLOUT, handler));
ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1}));
ASSERT_TRUE(got_pollout); ASSERT_EQ(events, ZMQ_POLLOUT);
} }
TEST(poller, poller_add_invalid_socket_throws) TEST(poller, poller_add_invalid_socket_throws)
@ -254,17 +264,62 @@ TEST(poller, poller_remove_invalid_socket_throws)
TEST(poller, wait_on_added_empty_handler) TEST(poller, wait_on_added_empty_handler)
{ {
zmq::context_t context; server_client_setup s;
zmq::socket_t vent{context, zmq::socket_type::push}; ASSERT_NO_THROW(s.client.send("Hi"));
auto endpoint = loopback_ip4_binder(vent).endpoint();
zmq::socket_t sink{context, zmq::socket_type::pull};
ASSERT_NO_THROW(sink.connect(endpoint));
ASSERT_NO_THROW(vent.send("Hi"));
zmq::poller_t poller; zmq::poller_t poller;
std::function<void(void)> handler; std::function<void(void)> handler;
ASSERT_NO_THROW(poller.add(sink, ZMQ_POLLIN, handler)); ASSERT_NO_THROW(poller.add(s.server, ZMQ_POLLIN, handler));
ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1})); ASSERT_NO_THROW(poller.wait(std::chrono::milliseconds{-1}));
} }
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, modified_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);
}
TEST(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));
// 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);
// 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);
}
#endif #endif

View File

@ -1083,6 +1083,12 @@ namespace zmq
throw error_t (); throw error_t ();
} }
void modify (zmq::socket_t &socket, short events)
{
if (0 != zmq_poller_modify (poller_ptr, socket.ptr, events))
throw error_t ();
}
bool wait (std::chrono::milliseconds timeout) bool wait (std::chrono::milliseconds timeout)
{ {
int rc = zmq_poller_wait_all (poller_ptr, poller_events.data (), static_cast<int> (poller_events.size ()), static_cast<long>(timeout.count ())); int rc = zmq_poller_wait_all (poller_ptr, poller_events.data (), static_cast<int> (poller_events.size ()), static_cast<long>(timeout.count ()));