generic-library/cppzmq
1
0
Fork 0
mirror of https://github.com/zeromq/cppzmq.git synced 2025-06-27 06:45:40 +02:00
Code Issues Projects Releases Wiki Activity

Merge pull request #405 from gummif/gfa/active-poller-handler

Browse Source 
Problem: Active poller double add mutates handler
This commit is contained in:
Simon Giesecke 2020-09-07 10:01:26 +02:00 committed by GitHub
commit bf4f75b971
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 111 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
tests/active_poller.cpp
View File

@ -14,9 +14,12 @@ TEST_CASE("create destroy", "[active_poller]")
} }
static_assert(!std::is_copy_constructible<zmq::active_poller_t>::value, static_assert(!std::is_copy_constructible<zmq::active_poller_t>::value,
"active_active_poller_t should not be copy-constructible"); "active_poller_t should not be copy-constructible");
static_assert(!std::is_copy_assignable<zmq::active_poller_t>::value, static_assert(!std::is_copy_assignable<zmq::active_poller_t>::value,
"active_active_poller_t should not be copy-assignable"); "active_poller_t should not be copy-assignable");
static const zmq::active_poller_t::handler_type no_op_handler =
[](zmq::event_flags) {};
TEST_CASE("move construct empty", "[active_poller]") TEST_CASE("move construct empty", "[active_poller]")
{ {
@ -47,9 +50,7 @@ TEST_CASE("move construct non empty", "[active_poller]")
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t a; zmq::active_poller_t a;
a.add(socket, zmq::event_flags::pollin, [](zmq::event_flags) a.add(socket, zmq::event_flags::pollin, [](zmq::event_flags) {});
{
});
CHECK_FALSE(a.empty()); CHECK_FALSE(a.empty());
CHECK(1u == a.size()); CHECK(1u == a.size());
zmq::active_poller_t b = std::move(a); zmq::active_poller_t b = std::move(a);
@ -65,9 +66,7 @@ TEST_CASE("move assign non empty", "[active_poller]")
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t a; zmq::active_poller_t a;
a.add(socket, zmq::event_flags::pollin, [](zmq::event_flags) a.add(socket, zmq::event_flags::pollin, no_op_handler);
{
});
CHECK_FALSE(a.empty()); CHECK_FALSE(a.empty());
CHECK(1u == a.size()); CHECK(1u == a.size());
zmq::active_poller_t b; zmq::active_poller_t b;
@ -79,12 +78,22 @@ TEST_CASE("move assign non empty", "[active_poller]")
} }
TEST_CASE("add handler", "[active_poller]") TEST_CASE("add handler", "[active_poller]")
{
zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller;
CHECK_NOTHROW(
active_poller.add(socket, zmq::event_flags::pollin, no_op_handler));
}
TEST_CASE("add null handler fails", "[active_poller]")
{ {
zmq::context_t context; zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::active_poller_t::handler_type handler; zmq::active_poller_t::handler_type handler;
CHECK_NOTHROW(active_poller.add(socket, zmq::event_flags::pollin, handler)); CHECK_THROWS_AS(active_poller.add(socket, zmq::event_flags::pollin, handler),
const std::invalid_argument &);
} }
#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 3, 0) #if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 3, 0)
@ -94,11 +103,11 @@ TEST_CASE("add handler invalid events type", "[active_poller]")
zmq::context_t context; zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::active_poller_t::handler_type handler;
short invalid_events_type = 2 << 10; short invalid_events_type = 2 << 10;
CHECK_THROWS_AS( CHECK_THROWS_AS(
active_poller.add(socket, static_cast<zmq::event_flags>(invalid_events_type), active_poller.add(socket, static_cast<zmq::event_flags>(invalid_events_type),
handler), const zmq::error_t&); no_op_handler),
const zmq::error_t &);
CHECK(active_poller.empty()); CHECK(active_poller.empty());
CHECK(0u == active_poller.size()); CHECK(0u == active_poller.size());
} }
@ -106,22 +115,26 @@ TEST_CASE("add handler invalid events type", "[active_poller]")
TEST_CASE("add handler twice throws", "[active_poller]") TEST_CASE("add handler twice throws", "[active_poller]")
{ {
zmq::context_t context; common_server_client_setup s;
zmq::socket_t socket{context, zmq::socket_type::router};
CHECK(s.client.send(zmq::message_t{}, zmq::send_flags::none));
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::active_poller_t::handler_type handler; bool message_received = false;
active_poller.add(socket, zmq::event_flags::pollin, handler); active_poller.add(
/// \todo the actual error code should be checked s.server, zmq::event_flags::pollin,
CHECK_THROWS_AS(active_poller.add(socket, zmq::event_flags::pollin, handler), [&message_received](zmq::event_flags) { message_received = true; });
const zmq::error_t&); CHECK_THROWS_ZMQ_ERROR(
EINVAL, active_poller.add(s.server, zmq::event_flags::pollin, no_op_handler));
CHECK(1 == active_poller.wait(std::chrono::milliseconds{-1}));
CHECK(message_received); // handler unmodified
} }
TEST_CASE("wait with no handlers throws", "[active_poller]") TEST_CASE("wait with no handlers throws", "[active_poller]")
{ {
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
/// \todo the actual error code should be checked CHECK_THROWS_ZMQ_ERROR(EFAULT,
CHECK_THROWS_AS(active_poller.wait(std::chrono::milliseconds{10}), active_poller.wait(std::chrono::milliseconds{10}));
const zmq::error_t&);
} }
TEST_CASE("remove unregistered throws", "[active_poller]") TEST_CASE("remove unregistered throws", "[active_poller]")
@ -129,8 +142,7 @@ TEST_CASE("remove unregistered throws", "[active_poller]")
zmq::context_t context; zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
/// \todo the actual error code should be checked CHECK_THROWS_ZMQ_ERROR(EINVAL, active_poller.remove(socket));
CHECK_THROWS_AS(active_poller.remove(socket), const zmq::error_t&);
} }
TEST_CASE("remove registered empty", "[active_poller]") TEST_CASE("remove registered empty", "[active_poller]")
@ -138,8 +150,7 @@ TEST_CASE("remove registered empty", "[active_poller]")
zmq::context_t context; zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
active_poller.add(socket, zmq::event_flags::pollin, active_poller.add(socket, zmq::event_flags::pollin, no_op_handler);
zmq::active_poller_t::handler_type{});
CHECK_NOTHROW(active_poller.remove(socket)); CHECK_NOTHROW(active_poller.remove(socket));
} }
@ -148,9 +159,7 @@ TEST_CASE("remove registered non empty", "[active_poller]")
zmq::context_t context; zmq::context_t context;
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
active_poller.add(socket, zmq::event_flags::pollin, [](zmq::event_flags) active_poller.add(socket, zmq::event_flags::pollin, no_op_handler);
{
});
CHECK_NOTHROW(active_poller.remove(socket)); CHECK_NOTHROW(active_poller.remove(socket));
} }
@ -158,8 +167,7 @@ namespace
{ {
struct server_client_setup : common_server_client_setup struct server_client_setup : common_server_client_setup
{ {
zmq::active_poller_t::handler_type handler = [&](zmq::event_flags e) zmq::active_poller_t::handler_type handler = [&](zmq::event_flags e) {
{
events = e; events = e;
}; };
@ -178,9 +186,8 @@ TEST_CASE("poll basic", "[active_poller]")
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
bool message_received = false; bool message_received = false;
zmq::active_poller_t::handler_type handler = [&message_received zmq::active_poller_t::handler_type handler =
](zmq::event_flags events) [&message_received](zmq::event_flags events) {
{
CHECK(zmq::event_flags::none != (events & zmq::event_flags::pollin)); CHECK(zmq::event_flags::none != (events & zmq::event_flags::pollin));
message_received = true; message_received = true;
}; };
@ -200,8 +207,7 @@ TEST_CASE("client server", "[active_poller]")
// Setup active_poller // Setup active_poller
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::event_flags events; zmq::event_flags events;
zmq::active_poller_t::handler_type handler = [&](zmq::event_flags e) zmq::active_poller_t::handler_type handler = [&](zmq::event_flags e) {
{
if (zmq::event_flags::none != (e & zmq::event_flags::pollin)) { if (zmq::event_flags::none != (e & zmq::event_flags::pollin)) {
zmq::message_t zmq_msg; zmq::message_t zmq_msg;
CHECK_NOTHROW(s.server.recv(zmq_msg)); // get message CHECK_NOTHROW(s.server.recv(zmq_msg)); // get message
@ -224,9 +230,8 @@ TEST_CASE("client server", "[active_poller]")
// Re-add server socket with pollout flag // Re-add server socket with pollout flag
CHECK_NOTHROW(active_poller.remove(s.server)); CHECK_NOTHROW(active_poller.remove(s.server));
CHECK_NOTHROW( CHECK_NOTHROW(active_poller.add(
active_poller.add(s.server, zmq::event_flags::pollin | zmq::event_flags:: s.server, zmq::event_flags::pollin | zmq::event_flags::pollout, handler));
pollout, handler));
CHECK(1 == active_poller.wait(std::chrono::milliseconds{-1})); CHECK(1 == active_poller.wait(std::chrono::milliseconds{-1}));
CHECK(events == zmq::event_flags::pollout); CHECK(events == zmq::event_flags::pollout);
} }
@ -237,10 +242,8 @@ TEST_CASE("add invalid socket throws", "[active_poller]")
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::socket_t a{context, zmq::socket_type::router}; zmq::socket_t a{context, zmq::socket_type::router};
zmq::socket_t b{std::move(a)}; zmq::socket_t b{std::move(a)};
CHECK_THROWS_AS( CHECK_THROWS_AS(active_poller.add(a, zmq::event_flags::pollin, no_op_handler),
active_poller.add(a, zmq::event_flags::pollin, zmq::active_poller_t:: const zmq::error_t &);
handler_type{}),
const zmq::error_t&);
} }
TEST_CASE("remove invalid socket throws", "[active_poller]") TEST_CASE("remove invalid socket throws", "[active_poller]")
@ -249,12 +252,11 @@ TEST_CASE("remove invalid socket throws", "[active_poller]")
zmq::socket_t socket{context, zmq::socket_type::router}; zmq::socket_t socket{context, zmq::socket_type::router};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_NOTHROW( CHECK_NOTHROW(
active_poller.add(socket, zmq::event_flags::pollin, zmq::active_poller_t:: active_poller.add(socket, zmq::event_flags::pollin, no_op_handler));
handler_type{}));
CHECK(1u == active_poller.size()); CHECK(1u == active_poller.size());
std::vector<zmq::socket_t> sockets; std::vector<zmq::socket_t> sockets;
sockets.emplace_back(std::move(socket)); sockets.emplace_back(std::move(socket));
CHECK_THROWS_AS(active_poller.remove(socket), const zmq::error_t&); CHECK_THROWS_AS(active_poller.remove(socket), const zmq::error_t &);
CHECK(1u == active_poller.size()); CHECK(1u == active_poller.size());
} }
@ -263,8 +265,8 @@ TEST_CASE("wait on added empty handler", "[active_poller]")
server_client_setup s; server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none)); CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none));
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
zmq::active_poller_t::handler_type handler; CHECK_NOTHROW(
CHECK_NOTHROW(active_poller.add(s.server, zmq::event_flags::pollin, handler)); active_poller.add(s.server, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(active_poller.wait(std::chrono::milliseconds{-1})); CHECK_NOTHROW(active_poller.wait(std::chrono::milliseconds{-1}));
} }
@ -274,7 +276,7 @@ TEST_CASE("modify empty throws", "[active_poller]")
zmq::socket_t socket{context, zmq::socket_type::push}; zmq::socket_t socket{context, zmq::socket_type::push};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_THROWS_AS(active_poller.modify(socket, zmq::event_flags::pollin), CHECK_THROWS_AS(active_poller.modify(socket, zmq::event_flags::pollin),
const zmq::error_t&); const zmq::error_t &);
} }
TEST_CASE("modify invalid socket throws", "[active_poller]") TEST_CASE("modify invalid socket throws", "[active_poller]")
@ -284,7 +286,7 @@ TEST_CASE("modify invalid socket throws", "[active_poller]")
zmq::socket_t b{std::move(a)}; zmq::socket_t b{std::move(a)};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_THROWS_AS(active_poller.modify(a, zmq::event_flags::pollin), CHECK_THROWS_AS(active_poller.modify(a, zmq::event_flags::pollin),
const zmq::error_t&); const zmq::error_t &);
} }
TEST_CASE("modify not added throws", "[active_poller]") TEST_CASE("modify not added throws", "[active_poller]")
@ -293,11 +295,9 @@ TEST_CASE("modify not added throws", "[active_poller]")
zmq::socket_t a{context, zmq::socket_type::push}; zmq::socket_t a{context, zmq::socket_type::push};
zmq::socket_t b{context, zmq::socket_type::push}; zmq::socket_t b{context, zmq::socket_type::push};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_NOTHROW( CHECK_NOTHROW(active_poller.add(a, zmq::event_flags::pollin, no_op_handler));
active_poller.add(a, zmq::event_flags::pollin, zmq::active_poller_t::
handler_type{}));
CHECK_THROWS_AS(active_poller.modify(b, zmq::event_flags::pollin), CHECK_THROWS_AS(active_poller.modify(b, zmq::event_flags::pollin),
const zmq::error_t&); const zmq::error_t &);
} }
TEST_CASE("modify simple", "[active_poller]") TEST_CASE("modify simple", "[active_poller]")
@ -305,12 +305,9 @@ TEST_CASE("modify simple", "[active_poller]")
zmq::context_t context; zmq::context_t context;
zmq::socket_t a{context, zmq::socket_type::push}; zmq::socket_t a{context, zmq::socket_type::push};
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_NOTHROW(active_poller.add(a, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW( CHECK_NOTHROW(
active_poller.add(a, zmq::event_flags::pollin, zmq::active_poller_t:: active_poller.modify(a, zmq::event_flags::pollin | zmq::event_flags::pollout));
handler_type{}));
CHECK_NOTHROW(
active_poller.modify(a, zmq::event_flags::pollin | zmq::event_flags::pollout
));
} }
TEST_CASE("poll client server", "[active_poller]") TEST_CASE("poll client server", "[active_poller]")
@ -330,9 +327,8 @@ TEST_CASE("poll client server", "[active_poller]")
CHECK(s.events == zmq::event_flags::pollin); CHECK(s.events == zmq::event_flags::pollin);
// Modify server socket with pollout flag // Modify server socket with pollout flag
CHECK_NOTHROW( CHECK_NOTHROW(active_poller.modify(s.server, zmq::event_flags::pollin
active_poller.modify(s.server, zmq::event_flags::pollin | zmq::event_flags:: | zmq::event_flags::pollout));
pollout));
CHECK(1 == active_poller.wait(std::chrono::milliseconds{500})); CHECK(1 == active_poller.wait(std::chrono::milliseconds{500}));
CHECK(s.events == (zmq::event_flags::pollin | zmq::event_flags::pollout)); CHECK(s.events == (zmq::event_flags::pollin | zmq::event_flags::pollout));
} }
@ -346,9 +342,8 @@ TEST_CASE("wait one return", "[active_poller]")
// Setup active_poller // Setup active_poller
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
CHECK_NOTHROW( CHECK_NOTHROW(active_poller.add(s.server, zmq::event_flags::pollin,
active_poller.add(s.server, zmq::event_flags::pollin, [&count](zmq:: [&count](zmq::event_flags) { ++count; }));
event_flags) { ++count; }));
// client sends message // client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none)); CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none));
@ -363,12 +358,11 @@ TEST_CASE("wait on move constructed active_poller", "[active_poller]")
server_client_setup s; server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none)); CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none));
zmq::active_poller_t a; zmq::active_poller_t a;
zmq::active_poller_t::handler_type handler; CHECK_NOTHROW(a.add(s.server, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(a.add(s.server, zmq::event_flags::pollin, handler));
zmq::active_poller_t b{std::move(a)}; zmq::active_poller_t b{std::move(a)};
CHECK(1u == b.size()); CHECK(1u == b.size());
/// \todo the actual error code should be checked CHECK(0u == a.size());
CHECK_THROWS_AS(a.wait(std::chrono::milliseconds{10}), const zmq::error_t&); CHECK_THROWS_ZMQ_ERROR(EFAULT, a.wait(std::chrono::milliseconds{10}));
CHECK(b.wait(std::chrono::milliseconds{-1})); CHECK(b.wait(std::chrono::milliseconds{-1}));
} }
@ -377,13 +371,12 @@ TEST_CASE("wait on move assigned active_poller", "[active_poller]")
server_client_setup s; server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none)); CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none));
zmq::active_poller_t a; zmq::active_poller_t a;
zmq::active_poller_t::handler_type handler; CHECK_NOTHROW(a.add(s.server, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(a.add(s.server, zmq::event_flags::pollin, handler));
zmq::active_poller_t b; zmq::active_poller_t b;
b = {std::move(a)}; b = {std::move(a)};
CHECK(1u == b.size()); CHECK(1u == b.size());
/// \todo the actual error code should be checked CHECK(0u == a.size());
CHECK_THROWS_AS(a.wait(std::chrono::milliseconds{10}), const zmq::error_t&); CHECK_THROWS_ZMQ_ERROR(EFAULT, a.wait(std::chrono::milliseconds{10}));
CHECK(b.wait(std::chrono::milliseconds{-1})); CHECK(b.wait(std::chrono::milliseconds{-1}));
} }
@ -394,9 +387,8 @@ TEST_CASE("received on move constructed active_poller", "[active_poller]")
int count = 0; int count = 0;
// Setup active_poller a // Setup active_poller a
zmq::active_poller_t a; zmq::active_poller_t a;
CHECK_NOTHROW( CHECK_NOTHROW(a.add(s.server, zmq::event_flags::pollin,
a.add(s.server, zmq::event_flags::pollin, [&count](zmq::event_flags) { ++ [&count](zmq::event_flags) { ++count; }));
count; }));
// client sends message // client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none)); CHECK_NOTHROW(s.client.send(zmq::message_t{hi_str}, zmq::send_flags::none));
// wait for message and verify it is received // wait for message and verify it is received
@ -425,9 +417,9 @@ TEST_CASE("remove from handler", "[active_poller]")
zmq::active_poller_t active_poller; zmq::active_poller_t active_poller;
int count = 0; int count = 0;
for (size_t i = 0; i < ITER_NO; ++i) { for (size_t i = 0; i < ITER_NO; ++i) {
CHECK_NOTHROW( CHECK_NOTHROW(active_poller.add(
active_poller.add(setup_list[i].server, zmq::event_flags::pollin, [&, i]( setup_list[i].server, zmq::event_flags::pollin,
zmq::event_flags events) { [&, i](zmq::event_flags events) {
CHECK(events == zmq::event_flags::pollin); CHECK(events == zmq::event_flags::pollin);
active_poller.remove(setup_list[ITER_NO - i - 1].server); active_poller.remove(setup_list[ITER_NO - i - 1].server);
CHECK((ITER_NO - i - 1) == active_poller.size()); CHECK((ITER_NO - i - 1) == active_poller.size());

19
tests/testutil.hpp
View File

@ -33,3 +33,22 @@ struct common_server_client_setup
std::string endpoint; std::string endpoint;
}; };
#endif #endif
#define CHECK_THROWS_ZMQ_ERROR(ecode, expr) \
do { \
try { \
expr; \
CHECK(false); \
} \
catch (const zmq::error_t &ze) { \
INFO(std::string("Unexpected error code: ") + ze.what()); \
CHECK(ze.num() == ecode); \
} \
catch (const std::exception &ex) { \
INFO(std::string("Unexpected exception: ") + ex.what()); \
CHECK(false); \
} \
catch (...) { \
CHECK(false); \
} \
} while (false)

5
zmq.hpp
View File

@ -261,12 +261,13 @@ typedef zmq_pollitem_t pollitem_t;
class error_t : public std::exception class error_t : public std::exception
{ {
public: public:
error_t() : errnum(zmq_errno()) {} error_t() ZMQ_NOTHROW : errnum(zmq_errno()) {}
explicit error_t(int err) ZMQ_NOTHROW : errnum(err) {}
virtual const char *what() const ZMQ_NOTHROW ZMQ_OVERRIDE virtual const char *what() const ZMQ_NOTHROW ZMQ_OVERRIDE
{ {
return zmq_strerror(errnum); return zmq_strerror(errnum);
} }
int num() const { return errnum; } int num() const ZMQ_NOTHROW { return errnum; }
private: private:
int errnum; int errnum;

19
zmq_addon.hpp
View File

@ -673,20 +673,19 @@ class active_poller_t
void add(zmq::socket_ref socket, event_flags events, handler_type handler) void add(zmq::socket_ref socket, event_flags events, handler_type handler)
{ {
auto it = decltype(handlers)::iterator{}; if (!handler)
auto inserted = bool{}; throw std::invalid_argument("null handler in active_poller_t::add");
std::tie(it, inserted) = handlers.emplace( auto ret = handlers.emplace(
socket, std::make_shared<handler_type>(std::move(handler))); socket, std::make_shared<handler_type>(std::move(handler)));
if (!ret.second)
throw error_t(EINVAL); // already added
try { try {
base_poller.add(socket, events, base_poller.add(socket, events, ret.first->second.get());
inserted && *(it->second) ? it->second.get() : nullptr); need_rebuild = true;
need_rebuild |= inserted;
} }
catch (const zmq::error_t &) { catch (...) {
// rollback // rollback
if (inserted) {
handlers.erase(socket); handlers.erase(socket);
}
throw; throw;
} }
} }
@ -718,7 +717,7 @@ class active_poller_t
std::for_each(poller_events.begin(), std::for_each(poller_events.begin(),
poller_events.begin() + static_cast<ptrdiff_t>(count), poller_events.begin() + static_cast<ptrdiff_t>(count),
[](decltype(base_poller)::event_type &event) { [](decltype(base_poller)::event_type &event) {
if (event.user_data != nullptr) assert(event.user_data != nullptr);
(*event.user_data)(event.events); (*event.user_data)(event.events);
}); });
return count; return count;