ZeroMQMonitor.hpp

Go to the documentation of this file.

#pragma once
 
#include <zmq.hpp>
 
#include <atomic>
#include <string>
#include <thread>

class ZeroMQMonitor : private zmq::monitor_t {
  private:
    std::atomic_int _peerCount{0};                  
    std::unique_ptr<std::jthread> _monitorThread; 
 
    void threadFunc(const std::stop_token &stopToken);
 
    static void on_event(const std::string &messageStr, int level, const char *addr = nullptr);
 
    void on_monitor_started() override;
 
    void on_event_connected(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_connect_delayed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_connect_retried(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_listening(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_bind_failed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_accepted(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_accept_failed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_closed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_close_failed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_disconnected(const zmq_event_t & /*unused*/, const char *addr_) override;
 
#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 3, 0) ||                                                                        \
    (defined(ZMQ_BUILD_DRAFT_API) && ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 2, 3))
    void on_event_handshake_failed_no_detail(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_handshake_failed_protocol(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_handshake_failed_auth(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_handshake_succeeded(const zmq_event_t & /*unused*/, const char *addr_) override;
 
#elif defined(ZMQ_BUILD_DRAFT_API) && ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 2, 1)
    void on_event_handshake_failed(const zmq_event_t & /*unused*/, const char *addr_) override;
 
    void on_event_handshake_succeed(const zmq_event_t & /*unused*/, const char *addr_) override;
#endif
 
    void on_event_unknown(const zmq_event_t & /*unused*/, const char *addr_) override;
 
  public:
    ZeroMQMonitor() = default;

    ZeroMQMonitor(const ZeroMQMonitor & /*unused*/) = delete;

    ZeroMQMonitor(ZeroMQMonitor && /*unused*/) = delete;

    ZeroMQMonitor &operator=(ZeroMQMonitor /*unused*/) = delete;

    ZeroMQMonitor &operator=(ZeroMQMonitor && /*unused*/) = delete;

    void startMonitoring(zmq::socket_t *socket, const std::string &monitorAddress);

    [[nodiscard]] int getPeerCount() const { return _peerCount; }

    void stopMonitoring();

    void testInternals();

    ~ZeroMQMonitor() override { stopMonitoring(); }
};