ZeroMQServer Class Reference

#include <ZeroMQServer.hpp>

Inheritance diagram for ZeroMQServer:

Collaboration diagram for ZeroMQServer:

Public Member Functions
	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 &)=delete
	Copy constructor.
	ZeroMQServer (ZeroMQServer &&)=delete
	Move constructor.
ZeroMQServer &	operator= (ZeroMQServer)=delete
	Copy assignment operator.
ZeroMQServer &	operator= (ZeroMQServer &&)=delete
	Move assignment operator.
bool	initialise ()
void	shutdown ()
	Closes the ZeroMQ Server.
	~ZeroMQServer () override
void	messageCallback (FPTR_MessageCallback func)
FPTR_MessageCallback	messageCallback () const

Private Member Functions
void	update ()
	Processes new messages.
void	threadFunc (const std::stop_token &stopToken) noexcept
	Main thread function.
Private Member Functions inherited from ZeroMQ
	ZeroMQ (const zmq::socket_type &type, const std::string &addr, bool isBind)
	ZeroMQ (const std::shared_ptr< zmq::context_t > &ctx, const zmq::socket_type &type, const std::string &addr, bool isBind)
	ZeroMQ (const ZeroMQ &)=delete
	Copy constructor.
	ZeroMQ (ZeroMQ &&)=delete
	Move constructor.
ZeroMQ &	operator= (ZeroMQ)=delete
	Copy assignment operator.
ZeroMQ &	operator= (ZeroMQ &&)=delete
	Move assignment operator.
bool	start ()
void	stop ()
std::vector< zmq::message_t >	recvMessages ()
size_t	sendMessages (std::vector< zmq::message_t > &msg)
const std::unique_ptr< zmq::socket_t > &	getSocket () const
const std::shared_ptr< zmq::context_t > &	getContext () const
const std::string &	getAddress () const
	~ZeroMQ ()
Private Member Functions inherited from ZeroMQMonitor
	ZeroMQMonitor ()=default
	Constructor.
	ZeroMQMonitor (const ZeroMQMonitor &)=delete
	Copy constructor.
	ZeroMQMonitor (ZeroMQMonitor &&)=delete
	Move constructor.
ZeroMQMonitor &	operator= (ZeroMQMonitor)=delete
	Copy assignment operator.
ZeroMQMonitor &	operator= (ZeroMQMonitor &&)=delete
	Move assignment operator.
void	startMonitoring (zmq::socket_t *socket, const std::string &monitorAddress)
int	getPeerCount () const
void	stopMonitoring ()
void	testInternals ()
	~ZeroMQMonitor () override

Private Attributes
std::shared_ptr< std::atomic_flag >	_checkFlag
std::unique_ptr< ZeroMQStats >	_stats
FPTR_MessageCallback	_m_messageCallback
std::unique_ptr< std::jthread >	_serverThread

Detailed Description

Definition at line 11 of file ZeroMQServer.hpp.

Constructor & Destructor Documentation

ZeroMQServer() [1/3]

ZeroMQServer::ZeroMQServer	(	const std::string &	hostAddr,
		std::shared_ptr< std::atomic_flag >	checkFlag,
		const std::shared_ptr< prometheus::Registry > &	reg = nullptr,
		const std::string &	prependName = "" )

Constructor for server

Parameters

[in]	hostAddr	Host address to connect. Can be anything supported by ZeroMQ reply socket
[in]	checkFlag	Flag to check if the server is running
[in]	reg	Prometheus registry for stats
[in]	prependName	Prefix for Prometheus stats

Definition at line 83 of file ZeroMQServer.cpp.

    : ZeroMQ(zmq::socket_type::rep, hostAddr, true), _checkFlag(std::move(checkFlag))
{
    if (reg)
    {
        _stats = std::make_unique<ZeroMQStats>(reg, prependName);
    }
 
    startMonitoring(getSocket().get(), std::format("{}{}{}", "inproc://", constHasher(hostAddr.c_str()), ".rep"));
}

Here is the call graph for this function:

Here is the caller graph for this function:

ZeroMQServer() [2/3]

ZeroMQServer::ZeroMQServer ( const ZeroMQServer & )

delete

Copy constructor.

Here is the call graph for this function:

ZeroMQServer() [3/3]

ZeroMQServer::ZeroMQServer ( ZeroMQServer && )

delete

Move constructor.

Here is the call graph for this function:

~ZeroMQServer()

ZeroMQServer::~ZeroMQServer ( )

inlineoverride

Deconstructor for server

Definition at line 64 of file ZeroMQServer.hpp.

{ shutdown(); }

Here is the call graph for this function:

Member Function Documentation

initialise()

bool ZeroMQServer::initialise

(

)

Initializes a new ZeroMQ server

Returns

true If initialized

false otherwise

Definition at line 95 of file ZeroMQServer.cpp.

{
    if (start())
    {
        _serverThread = std::make_unique<std::jthread>([this](const std::stop_token &sToken) { threadFunc(sToken); });
        return true;
    }
    return false;
}

Here is the call graph for this function:

messageCallback() [1/2]

FPTR_MessageCallback ZeroMQServer::messageCallback ( ) const

inlinenodiscard

Gets the message callback function

Returns

The message callback function

Definition at line 76 of file ZeroMQServer.hpp.

{ return _m_messageCallback; }

Here is the caller graph for this function:

messageCallback() [2/2]

void ZeroMQServer::messageCallback ( FPTR_MessageCallback func )

inline

Sets the message callback function

Parameters

[in]

func

The message callback function to be set

Definition at line 70 of file ZeroMQServer.hpp.

{ _m_messageCallback = std::move(func); }

operator=() [1/2]

ZeroMQServer & ZeroMQServer::operator= ( ZeroMQServer && )

delete

Move assignment operator.

Here is the call graph for this function:

operator=() [2/2]

ZeroMQServer & ZeroMQServer::operator= ( ZeroMQServer )

delete

Copy assignment operator.

Here is the call graph for this function:

shutdown()

void ZeroMQServer::shutdown

(

)

Closes the ZeroMQ Server.

Definition at line 105 of file ZeroMQServer.cpp.

{
    if (_serverThread)
    {
        _serverThread.reset();
    }
 
    stop();
}

Here is the call graph for this function:

Here is the caller graph for this function:

threadFunc()

void ZeroMQServer::threadFunc ( const std::stop_token & stopToken )

privatenoexcept

Main thread function.

Definition at line 62 of file ZeroMQServer.cpp.

{
    spdlog::info("ZeroMQ server started");
    while (!stopToken.stop_requested())
    {
        try
        {
            update();
            if (_checkFlag)
            {
                _checkFlag->test_and_set();
            }
        }
        catch (const std::exception &e)
        {
            spdlog::error("ZeroMQ server failed: {}", e.what());
        }
    }
    spdlog::info("ZeroMQ server stopped");
}

Here is the call graph for this function:

Here is the caller graph for this function:

update()

void ZeroMQServer::update ( )

private

Processes new messages.

Definition at line 27 of file ZeroMQServer.cpp.

{
    auto recvMsgs = recvMessages();
 
    if (!recvMsgs.empty())
    {
        std::vector<zmq::message_t> replyMsgs;
 
        ZeroMQServerStats serverStats;
        serverStats.processingTimeStart = std::chrono::high_resolution_clock::now();
 
        if (recvMsgs[0].size() != sizeof(uint32_t))
        {
            int errorCode = ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL;
            replyMsgs.emplace_back(&errorCode, sizeof(errorCode));
            replyMsgs.emplace_back("", 0);
        }
        else
        {
            serverStats.isSuccessful = messageCallback() && messageCallback()(recvMsgs, replyMsgs);
        }
 
        if (size_t nSentMsg = sendMessages(replyMsgs); nSentMsg != replyMsgs.size())
        {
            spdlog::warn("Can't send whole reply: Sent messages {} / {}", nSentMsg, replyMsgs.size());
        }
        serverStats.processingTimeEnd = std::chrono::high_resolution_clock::now();
 
        if (_stats)
        {
            _stats->consumeStats(recvMsgs, replyMsgs, serverStats);
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

_checkFlag

std::shared_ptr<std::atomic_flag> ZeroMQServer::_checkFlag

private

Definition at line 14 of file ZeroMQServer.hpp.

_m_messageCallback

FPTR_MessageCallback ZeroMQServer::_m_messageCallback

private

Definition at line 18 of file ZeroMQServer.hpp.

_serverThread

std::unique_ptr<std::jthread> ZeroMQServer::_serverThread

private

Definition at line 20 of file ZeroMQServer.hpp.

_stats

std::unique_ptr<ZeroMQStats> ZeroMQServer::_stats

private

Definition at line 16 of file ZeroMQServer.hpp.

---

The documentation for this class was generated from the following files:

• /__w/Repo-Init/Repo-Init/include/zeromq/ZeroMQServer.hpp
• /__w/Repo-Init/Repo-Init/src/zeromq/ZeroMQServer.cpp