ZeroMQServer.hpp

Go to the documentation of this file.

#pragma once
 
#include "zeromq/ZeroMQ.hpp"
#include "zeromq/ZeroMQMonitor.hpp"
#include "zeromq/ZeroMQStats.hpp"
 
#include <functional>
 
using FPTR_MessageCallback = std::function<bool(const std::vector<zmq::message_t> &, std::vector<zmq::message_t> &)>;
 
class ZeroMQServer : private ZeroMQ, private ZeroMQMonitor {
  private:
    // Thread for processing messages
    std::unique_ptr<std::thread> _serverThread;
    // Flag to stop processing messages
    std::atomic_flag _shouldStop{false};
    // Flag to check if the server is running
    std::shared_ptr<std::atomic_flag> _checkFlag;
    // Statistics
    std::unique_ptr<ZeroMQStats> _stats;
    // Called after every message function(std::vector<zmq::message_t>) {}
    FPTR_MessageCallback _m_messageCallback;
 
    void update();
 
    void threadFunc() noexcept;
 
  public:
    ZeroMQServer(const std::string &hostAddr, std::shared_ptr<std::atomic_flag> checkFlag,
                 const std::shared_ptr<prometheus::Registry> &reg = nullptr, const std::string &prependName = "");
 
    ZeroMQServer(const ZeroMQServer & /*unused*/) = delete;
 
    ZeroMQServer(ZeroMQServer && /*unused*/) = delete;
 
    ZeroMQServer &operator=(ZeroMQServer /*unused*/) = delete;
 
    ZeroMQServer &operator=(ZeroMQServer && /*unused*/) = delete;
 
    bool initialise();
 
    void shutdown();
 
    ~ZeroMQServer() override { shutdown(); }
 
    void messageCallback(FPTR_MessageCallback func) { _m_messageCallback = std::move(func); }
 
    [[nodiscard]] FPTR_MessageCallback messageCallback() const { return _m_messageCallback; }
};
 
bool ZeroMQServerMessageCallback(const std::vector<zmq::message_t> &recvMsgs, std::vector<zmq::message_t> &replyMsgs);