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::unique_ptr<std::thread> _monitorThread; 
    std::atomic_flag _shouldStop{false};         
    std::atomic_int _peerCount{0};               
    void threadFunc();
 
    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();
 
    ~ZeroMQMonitor() override { stopMonitoring(); }
};