Merge pull request #225 from sigiesec/split-poller-layer

Split poller_t into two layers

tests/CMakeLists.txt

@@ -26,6 +26,8 @@ add_executable(
     context.cpp
     socket.cpp
     poller.cpp
+    active_poller.cpp
+    multipart.cpp
     )
 
 target_link_libraries(

tests/active_poller.cpp

@@ -0,0 +1,414 @@
+#include <zmq_addon.hpp>
+
+#include "testutil.hpp"
+
+#if defined(ZMQ_CPP11) && defined(ZMQ_BUILD_DRAFT_API)
+
+#include <array>
+#include <memory>
+
+TEST(active_poller, create_destroy)
+{
+    zmq::active_poller_t active_poller;
+    ASSERT_TRUE(active_poller.empty ());
+}
+
+static_assert(!std::is_copy_constructible<zmq::active_poller_t>::value, "active_active_poller_t should not be copy-constructible");
+static_assert(!std::is_copy_assignable<zmq::active_poller_t>::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<char>(),
+                                 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<zmq::socket_t> 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<server_client_setup> 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

tests/multipart.cpp

@@ -0,0 +1,171 @@
+#include <gtest/gtest.h>
+#include <zmq_addon.hpp>
+
+#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<char>(), 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<char>(), msg.size()) == "Frame9");
+
+    msg = multipart.pop();
+    assert(multipart.size() == 8);
+    assert(std::string(msg.data<char>(), msg.size()) == "Frame0");
+
+    num = multipart.poptyp<float>();
+    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<float>();
+    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<float>();
+    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

tests/poller.cpp

@@ -1,5 +1,4 @@
-#include <gtest/gtest.h>
-#include <zmq.hpp>
+#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<zmq::poller_t>::value, "poller_t should not be copy-constructible");
-static_assert(!std::is_copy_assignable<zmq::poller_t>::value, "poller_t should not be copy-assignable");
+static_assert(!std::is_copy_constructible<zmq::poller_t<>>::value, "poller_t should not be copy-constructible");
+static_assert(!std::is_copy_assignable<zmq::poller_t<>>::value, "poller_t should not be copy-assignable");
 
 TEST(poller, move_construct_empty)
 {
-    std::unique_ptr<zmq::poller_t> 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<zmq::poller_t> 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<zmq::poller_t> 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<zmq::poller_t> 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<zmq_poller_event_t> 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<char, 100> 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<char>(),
-                                 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<int> poller;
+    std::vector<zmq_poller_event_t> 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<zmq::socket_t> 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<zmq_poller_event_t> 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<zmq_poller_event_t> 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<zmq_poller_event_t> 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<zmq_poller_event_t> 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<server_client_setup> setup_list;
+    std::vector<common_server_client_setup> 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<zmq_poller_event_t> events(ITER_NO);
+    ASSERT_EQ(ITER_NO, poller.wait_all (events, std::chrono::milliseconds{-1}));
 }
 
 #endif

tests/testutil.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <gtest/gtest.h>
+#include <zmq.hpp>
+
+#if defined(ZMQ_CPP11)
+#include <array>
+
+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<char, 100> 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

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 <typename T = void>
     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(short)>;
-
-        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<handler_t> (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<void*>(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<void*>(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<void*>(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<zmq_poller_event_t> &poller_events, const std::chrono::microseconds timeout)
+        {
             int rc = zmq_poller_wait_all (poller_ptr.get (), poller_events.data (),
                                           static_cast<int> (poller_events.size ()),
                                           static_cast<long>(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<handler_t*> (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<void, std::function<void(void*)>> poller_ptr
         {
@@ -1122,14 +1077,9 @@ namespace zmq
                 ZMQ_ASSERT (rc == 0);
             }
         };
-        bool need_rebuild {false};
-        std::unordered_map<void*, std::shared_ptr<handler_t>> handlers {};
-        std::vector<zmq_poller_event_t> poller_events {};
-        std::vector<std::shared_ptr<handler_t>> 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();

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<char>(), 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<char>(), msg.size()) == "Frame9");
-
-        msg = multipart.pop();
-        assert(multipart.size() == 8);
-        assert(std::string(msg.data<char>(), msg.size()) == "Frame0");
-
-        num = multipart.poptyp<float>();
-        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<float>();
-        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<float>();
-        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(short)>;
+
+        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<void*>(socket), std::make_shared<handler_t> (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<void*>(socket));
+                }
+                throw;
+            }
+        }
+
+        void remove (zmq::socket_t &socket)
+        {
+            base_poller.remove (socket);
+            handlers.erase (static_cast<void*>(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<handler_t*> (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<handler_t> base_poller {};
+        std::unordered_map<void*, std::shared_ptr<handler_t>> handlers {};
+        std::vector<zmq_poller_event_t> poller_events {};
+        std::vector<std::shared_ptr<handler_t>> 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__