phoenix_zmq.cpp File Reference

#include <sstream>
#include "phoenix_zmq.h"

Include dependency graph for phoenix_zmq.cpp:

Go to the source code of this file.

Functions
void	pzmq_closeServerSocket (zmq::socket_t *&socket)
	Close the given server socket.
zmq::socket_t *	pzmq_createClientSocket (zmq::context_t &context, const std::string &address, size_t port, int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int immediate)
	Add a client socket to the manager.
zmq::socket_t *	pzmq_createClientSocket (zmq::context_t &context, int type, const std::string &address, size_t port, int immediate)
	Create a client socket to be used by the SocketManagerZMQ.
zmq::socket_t *	pzmq_createServerSocket (zmq::context_t &context, int type, size_t port)
	Create a server socket to be used by the SocketManagerZMQ.
zmq::socket_t *	pzmq_createServerSocket (zmq::context_t &context, size_t port, int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate)
	Add a server socket to the manager.
void	pzmq_setBufferSize (zmq::socket_t *socket, int type, int nbBufferMessage, int dataRate, size_t bufferSizeByte)
	Set the size of the buffer to send messages.
void	pzmq_setDataRate (zmq::socket_t *socket, int type, int dataRate)
	Set the data rate of the socket.
void	pzmq_setNbMessageBuffer (zmq::socket_t *socket, int nbBufferMessage)
	Set the number of messages in the messages buffer.
void	pzmq_setRecvBufferSize (zmq::socket_t *socket, int bufferSizeByte)
	Set the size of the buffer to received messages.
void	pzmq_setSendBufferSize (zmq::socket_t *socket, int bufferSizeByte)
	Set the size of the buffer to send messages.
void	pzmq_setThreadAffinity (zmq::socket_t *socket, size_t threadAffinity)
	Set the thread affinity of zmq.

Function Documentation

pzmq_closeServerSocket()

void pzmq_closeServerSocket

(

zmq::socket_t *&

socket

)

Close the given server socket.

Parameters

[out]

socket

: pointer to the server socket to be closed (will be set to NULL at then end of the function)

Definition at line 117 of file phoenix_zmq.cpp.

                                                  {
        if(socket != NULL){
                socket->close();
                delete socket;
                socket = NULL;
        }
}

pzmq_createClientSocket() [1/2]

zmq::socket_t * pzmq_createClientSocket	(	zmq::context_t &	context,
		const std::string &	address,
		size_t	port,
		int	type,
		int	nbBufferMessage,
		int	bufferSizeByte,
		size_t	threadAffinity,
		ssize_t	dataRate,
		int	immediate )

Add a client socket to the manager.

Parameters

context	: zmq context where to create socket
address	: address of the server to be connected to
port	: port to be used
type	: type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
nbBufferMessage	: number of messages to be buffered
bufferSizeByte	: size of the zmq buffer in bytes
threadAffinity	: bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
dataRate	: expected data rate (in kilobytes per second)

Returns

zmq socket

Definition at line 101 of file phoenix_zmq.cpp.

{
        zmq::socket_t* socket = pzmq_createClientSocket(context, type, address, port, immediate);
        bool b(socket != NULL);
        if(b){
                pzmq_setBufferSize(socket, type, nbBufferMessage, dataRate, bufferSizeByte);
                pzmq_setThreadAffinity(socket, threadAffinity);
        }
        return socket;
}

References pzmq_createClientSocket(), pzmq_setBufferSize(), and pzmq_setThreadAffinity().

Here is the call graph for this function:

pzmq_createClientSocket() [2/2]

zmq::socket_t * pzmq_createClientSocket	(	zmq::context_t &	context,
		int	type,
		const std::string &	address,
		size_t	port,
		int	immediate )

Create a client socket to be used by the SocketManagerZMQ.

Parameters

context	: zeromq context which defines the number of thread to be used in the data transfert
type	: type of the socket (ZMQ_PULL, ZMQ_PUSH, etc)
address	: address of the socket (example localhost or 127.0.0.1)
port	: port to be used

Returns

create socket

Definition at line 18 of file phoenix_zmq.cpp.

                                                                                                                             {
        std::stringstream socketAddressData;
        socketAddressData << "tcp://" << address <<":" << port;
        zmq::socket_t  *socket = new zmq::socket_t(context, type);
        if(socket == NULL){return NULL;}
 
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
                socket->set(zmq::sockopt::immediate, immediate);
                socket->set(zmq::sockopt::linger, -1);  //1 ms to stop
#else
                int immediateVal(immediate);
                socket->setsockopt(ZMQ_IMMEDIATE, &immediateVal, sizeof(int));
                socket->setsockopt(ZMQ_LINGER, -1);     //1 ms to stop
#endif
        socket->connect(socketAddressData.str());   // ← connect APRÈS les options
 
        if(type == ZMQ_SUB){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::subscribe, "");
                socket->set(zmq::sockopt::conflate, 1);
#else
                socket->setsockopt(ZMQ_SUBSCRIBE, "", 0);
                int conflate(1);
                socket->setsockopt(ZMQ_CONFLATE, &conflate, sizeof(int));
#endif
        }
        return socket;
}

Referenced by PZmqSocket::createClientSocket(), and pzmq_createClientSocket().

Here is the caller graph for this function:

pzmq_createServerSocket() [1/2]

zmq::socket_t * pzmq_createServerSocket	(	zmq::context_t &	context,
		int	type,
		size_t	port )

Create a server socket to be used by the SocketManagerZMQ.

Parameters

context	: zeromq context which defines the number of thread to be used in the data transfert
type	: type of the socket (ZMQ_PULL, ZMQ_PUSH, etc)
port	: port to be used

Returns

create socket

Definition at line 52 of file phoenix_zmq.cpp.

                                                                                   {
        std::stringstream socketAddressData;
        socketAddressData << "tcp://127.0.0.1:" << port;
        zmq::socket_t  *socket = new zmq::socket_t(context, type);
        if(socket != NULL){
                socket->bind(socketAddressData.str());
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::linger, -1);  //1 ms to stop
#else
                socket->setsockopt(ZMQ_LINGER, -1);     //1 ms to stop
#endif
        }
        return socket;
}

Referenced by PZmqSocket::createServerSocket(), and pzmq_createServerSocket().

Here is the caller graph for this function:

pzmq_createServerSocket() [2/2]

zmq::socket_t * pzmq_createServerSocket	(	zmq::context_t &	context,
		size_t	port,
		int	type,
		int	nbBufferMessage,
		int	bufferSizeByte,
		size_t	threadAffinity,
		ssize_t	dataRate )

Add a server socket to the manager.

Parameters

context	: zmq context where to create socket
port	: port to be used
type	: type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
nbBufferMessage	: number of messages to be buffered
bufferSizeByte	: size of the zmq buffer in bytes
threadAffinity	: bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
dataRate	: expected data rate (in kilobytes per second)

Returns

zmq socket

Definition at line 77 of file phoenix_zmq.cpp.

{
        zmq::socket_t* socket = pzmq_createServerSocket(context, type, port);
        bool b(socket != NULL);
        if(b){
                pzmq_setBufferSize(socket, type, nbBufferMessage, dataRate, bufferSizeByte);
                pzmq_setThreadAffinity(socket, threadAffinity);
        }
        return socket;
}

References pzmq_createServerSocket(), pzmq_setBufferSize(), and pzmq_setThreadAffinity().

Here is the call graph for this function:

pzmq_setBufferSize()

void pzmq_setBufferSize	(	zmq::socket_t *	socket,
		int	type,
		int	nbBufferMessage,
		int	dataRate,
		size_t	bufferSizeByte )

Set the size of the buffer to send messages.

Parameters

[out]	socket	: socket to be modified
	type	: type of the socket to be used
	nbBufferMessage	: number of messages to be buffered
	dataRate	: expected data rate (in kilobytes per second)
	bufferSizeByte	: size of the zmq buffer in bytes

Definition at line 210 of file phoenix_zmq.cpp.

                                                                                                                {
        pzmq_setNbMessageBuffer(socket, nbBufferMessage);
        pzmq_setDataRate(socket, type, dataRate);
        if(type == ZMQ_PULL || type == ZMQ_SUB){
                pzmq_setRecvBufferSize(socket, bufferSizeByte);
        }else if(type == ZMQ_PUSH || type == ZMQ_PUB){
                pzmq_setSendBufferSize(socket, bufferSizeByte);
        }
}

References pzmq_setDataRate(), pzmq_setNbMessageBuffer(), pzmq_setRecvBufferSize(), and pzmq_setSendBufferSize().

Referenced by pzmq_createClientSocket(), and pzmq_createServerSocket().

Here is the call graph for this function:

Here is the caller graph for this function:

pzmq_setDataRate()

void pzmq_setDataRate	(	zmq::socket_t *	socket,
		int	type,
		int	dataRate )

Set the data rate of the socket.

Parameters

[out]	socket	: socket to be modified
	type	: type of the socket to be used
	dataRate	: expected data rate (in kilobytes per second)

Definition at line 146 of file phoenix_zmq.cpp.

                                                                  {
        if(socket != NULL && (type == ZMQ_PUB || type == ZMQ_SUB)){
                int dataRateKbit(dataRate*8l);
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::rate, dataRateKbit);
#else
                //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
                socket->setsockopt(ZMQ_RATE, dataRateKbit);
#endif
        }
}

Referenced by pzmq_setBufferSize().

Here is the caller graph for this function:

pzmq_setNbMessageBuffer()

void pzmq_setNbMessageBuffer	(	zmq::socket_t *	socket,
		int	nbBufferMessage )

Set the number of messages in the messages buffer.

Parameters

[out]	socket	: socket to be modified
	nbBufferMessage	: number of messages to be buffered

Definition at line 129 of file phoenix_zmq.cpp.

                                                                      {
        if(socket != NULL){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::rcvhwm, nbBufferMessage);
#else
                //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
                socket->setsockopt(ZMQ_RCVHWM, &nbBufferMessage, sizeof(int));
                socket->setsockopt(ZMQ_SNDHWM, &nbBufferMessage, sizeof(int));
#endif
        }
}

Referenced by pzmq_setBufferSize().

Here is the caller graph for this function:

pzmq_setRecvBufferSize()

void pzmq_setRecvBufferSize	(	zmq::socket_t *	socket,
		int	bufferSizeByte )

Set the size of the buffer to received messages.

Parameters

[out]	socket	: socket to be modified
	bufferSizeByte	: size of the zmq buffer in bytes

Definition at line 162 of file phoenix_zmq.cpp.

                                                                    {
        if(socket != NULL){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::rcvbuf, bufferSizeByte);
#else
                //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
                socket->setsockopt(ZMQ_RCVBUF, bufferSizeByte);
#endif
        }
}

Referenced by pzmq_setBufferSize().

Here is the caller graph for this function:

pzmq_setSendBufferSize()

void pzmq_setSendBufferSize	(	zmq::socket_t *	socket,
		int	bufferSizeByte )

Set the size of the buffer to send messages.

Parameters

[out]	socket	: socket to be modified
	bufferSizeByte	: size of the zmq buffer in bytes

Definition at line 177 of file phoenix_zmq.cpp.

                                                                    {
        if(socket != NULL){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::sndbuf, bufferSizeByte);
#else
                //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
                socket->setsockopt(ZMQ_SNDBUF, bufferSizeByte);
#endif
        }
}

Referenced by pzmq_setBufferSize().

Here is the caller graph for this function:

pzmq_setThreadAffinity()

void pzmq_setThreadAffinity	(	zmq::socket_t *	socket,
		size_t	threadAffinity )

Set the thread affinity of zmq.

Parameters

socket	: socket to be modified
threadAffinity	: bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)

Definition at line 192 of file phoenix_zmq.cpp.

                                                                       {
        if(socket != NULL){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                socket->set(zmq::sockopt::affinity, threadAffinity);
#else
                //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
                socket->setsockopt(ZMQ_AFFINITY, threadAffinity);
#endif
        }
}

Referenced by pzmq_createClientSocket(), and pzmq_createServerSocket().

Here is the caller graph for this function: