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:
    // 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;
    // Thread for processing messages
    std::unique_ptr<std::jthread> _serverThread;

    void update();

    void threadFunc(const std::stop_token &stopToken) 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);