monitor.cpp

Go to the documentation of this file.

#include "testutil.hpp"
 
#ifdef ZMQ_CPP11
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
 
class mock_monitor_t : public zmq::monitor_t
{
  public:
 
    void on_event_connected(const zmq_event_t &, const char *) ZMQ_OVERRIDE
    {
        ++connected;
        ++total;
    }
 
    int total{0};
    int connected{0};
};
 
#endif
 
TEST_CASE("monitor create destroy", "[monitor]")
{
    zmq::monitor_t monitor;
}
 
#if defined(ZMQ_CPP11)
TEST_CASE("monitor move construct", "[monitor]")
{
    zmq::context_t ctx;
    zmq::socket_t sock(ctx, ZMQ_DEALER);
    SECTION("move ctor empty") {
        zmq::monitor_t monitor1;
        zmq::monitor_t monitor2 = std::move(monitor1);
    }
    SECTION("move ctor init") {
        zmq::monitor_t monitor1;
        monitor1.init(sock, "inproc://monitor-client");
        zmq::monitor_t monitor2 = std::move(monitor1);
    }
}
 
TEST_CASE("monitor move assign", "[monitor]")
{
    zmq::context_t ctx;
    zmq::socket_t sock(ctx, ZMQ_DEALER);
    SECTION("move assign empty") {
        zmq::monitor_t monitor1;
        zmq::monitor_t monitor2;
        monitor1 = std::move(monitor2);
    }
    SECTION("move assign init") {
        zmq::monitor_t monitor1;
        monitor1.init(sock, "inproc://monitor-client");
        zmq::monitor_t monitor2;
        monitor2 = std::move(monitor1);
    }
    SECTION("move assign init both") {
        zmq::monitor_t monitor1;
        monitor1.init(sock, "inproc://monitor-client");
        zmq::monitor_t monitor2;
        zmq::socket_t sock2(ctx, ZMQ_DEALER);
        monitor2.init(sock2, "inproc://monitor-client2");
        monitor2 = std::move(monitor1);
    }
}
 
TEST_CASE("monitor init event count", "[monitor]")
{
    common_server_client_setup s{false};
    mock_monitor_t monitor;
 
    const int expected_event_count = 1;
    monitor.init(s.client, "inproc://foo");
 
    CHECK_FALSE(monitor.check_event(0));
    s.init();
 
    while (monitor.check_event(1000) && monitor.total < expected_event_count) {
    }
    CHECK(monitor.connected == 1);
    CHECK(monitor.total == expected_event_count);
}
 
TEST_CASE("monitor init abort", "[monitor]")
{
    class mock_monitor : public mock_monitor_t
    {
      public:
        mock_monitor(std::function<void(void)> handle_connected) :
            handle_connected{std::move(handle_connected)}
        {
        }
 
        void on_event_connected(const zmq_event_t &e, const char *m) ZMQ_OVERRIDE
        {
            mock_monitor_t::on_event_connected(e, m);
            handle_connected();
        }
 
        std::function<void(void)> handle_connected;
    };
 
    common_server_client_setup s(false);
 
    std::mutex mutex;
    std::condition_variable cond_var;
    bool done{false};
 
    mock_monitor monitor([&]()
    {
        std::lock_guard<std::mutex> lock(mutex);
        done = true;
        cond_var.notify_one();
    });
    monitor.init(s.client, "inproc://foo");
 
    auto thread = std::thread([&monitor]
    {
        while (monitor.check_event(-1)) {
        }
    });
 
    s.init();
    {
        std::unique_lock<std::mutex> lock(mutex);
        CHECK(cond_var.wait_for(lock, std::chrono::seconds(1),
                                [&done] { return done; }));
    }
    CHECK(monitor.connected == 1);
    monitor.abort();
    thread.join();
}
 
 
TEST_CASE("monitor from move assigned socket", "[monitor]")
{
    zmq::context_t ctx;
    zmq::socket_t sock;
    sock = std::move([&ctx] {
        zmq::socket_t sock(ctx, ZMQ_DEALER);
        return sock;
    }());
    zmq::monitor_t monitor1;
    monitor1.init(sock, "inproc://monitor-client");
    // On failure, this test might hang indefinitely instead of immediately
    // failing
}
#endif
 
#if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11)                              \
  && !defined(ZMQ_CPP11_PARTIAL) && defined(ZMQ_HAVE_POLLER)
#include "zmq_addon.hpp"
 
TEST_CASE("poll monitor events using active poller", "[monitor]")
{
    // define necessary class for test
    class test_monitor : public zmq::monitor_t
    {
      public:
        void init(zmq::socket_t &socket,
                  const char *const addr_,
                  int events = ZMQ_EVENT_ALL)
        {
            zmq::monitor_t::init(socket, addr_, events);
        }
 
        void addToPoller(zmq::active_poller_t &inActivePoller)
        {
            inActivePoller.add(
              monitor_socket(), zmq::event_flags::pollin,
              [&](zmq::event_flags ef) { process_event(static_cast<short>(ef)); });
        }
 
        void on_event_accepted(const zmq_event_t &event_, const char *addr_) override
        {
            clientAccepted++;
        }
        void on_event_disconnected(const zmq_event_t &event,
                                   const char *const addr) override
        {
            clientDisconnected++;
        }
 
        int clientAccepted = 0;
        int clientDisconnected = 0;
    };
 
    //Arrange
    int messageCounter = 0;
    const char monitorAddress[] = "inproc://monitor-server";
 
    auto addToPoller = [&](zmq::socket_t &socket, zmq::active_poller_t &poller) {
        poller.add(socket, zmq::event_flags::pollin, [&](zmq::event_flags ef) {
            zmq::message_t msg;
            auto result = socket.recv(msg, zmq::recv_flags::dontwait);
            messageCounter++;
        });
    };
 
    common_server_client_setup sockets(false);
 
    test_monitor monitor;
    monitor.init(sockets.server, monitorAddress,
                 ZMQ_EVENT_ACCEPTED | ZMQ_EVENT_DISCONNECTED);
 
    zmq::active_poller_t poller;
    monitor.addToPoller(poller);
    addToPoller(sockets.server, poller);
 
    sockets.init();
    sockets.client.send(zmq::message_t(0), zmq::send_flags::dontwait);
    CHECK(monitor.clientAccepted == 0);
    CHECK(monitor.clientDisconnected == 0);
 
    //Act
    for (int i = 0; i < 100; i++) {
        poller.wait(std::chrono::milliseconds(50));
        if (monitor.clientAccepted > 0) {
            break;
        }
    }
    CHECK(monitor.clientAccepted == 1);
    CHECK(monitor.clientDisconnected == 0);
 
    sockets.client.close();
 
    for (int i = 0; i < 100; i++) {
        poller.wait(std::chrono::milliseconds(50));
        if (monitor.clientDisconnected > 0) {
            break;
        }
    }
    sockets.server.close();
 
    // Assert
    CHECK(messageCounter == 1);
    CHECK(monitor.clientAccepted == 1);
    CHECK(monitor.clientDisconnected == 1);
}
#endif