phoenix_zmq.cpp

Go to the documentation of this file.

/***************************************
        Auteur : Pierre Aubert
        Mail : pierre.aubert@lapp.in2p3.fr
        Licence : CeCILL-C
****************************************/
 
#include <sstream>
#include "phoenix_zmq.h"
 
 
 
zmq::socket_t* pzmq_createClientSocket(zmq::context_t & context, int type, const std::string & address, size_t port){
        std::stringstream socketAddressData;
        socketAddressData << "tcp://" << address <<":" << port;
        zmq::socket_t  *socket = new zmq::socket_t(context, type);
        socket->connect(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
        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;
}
 
 
zmq::socket_t* pzmq_createServerSocket(zmq::context_t & context, int type, size_t port){
        std::stringstream socketAddressData;
        socketAddressData << "tcp://*:" << port;
        zmq::socket_t  *socket = new zmq::socket_t(context, type);
        socket->bind(socketAddressData.str());
        return socket;
}
 
 
zmq::socket_t* pzmq_createServerSocket(zmq::context_t & context, size_t port, int type, int nbBufferMessage, int bufferSizeByte,
                                        size_t threadAffinity, ssize_t dataRate)
{
        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;
}
 
 
 
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)
{
        zmq::socket_t* socket = pzmq_createClientSocket(context, type, address, port);
        bool b(socket != NULL);
        if(b){
                pzmq_setBufferSize(socket, type, nbBufferMessage, dataRate, bufferSizeByte);
                pzmq_setThreadAffinity(socket, threadAffinity);
        }
        return socket;
}
 
 
void pzmq_closeServerSocket(zmq::socket_t *& socket){
        if(socket != NULL){
                delete socket;
                socket = NULL;
        }
}
 
 
void pzmq_setNbMessageBuffer(zmq::socket_t* socket, int nbBufferMessage){
        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
        }
}
 
 
void pzmq_setDataRate(zmq::socket_t* socket, int type, int dataRate){
        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
        }
}
 
 
void pzmq_setRecvBufferSize(zmq::socket_t* socket, int bufferSizeByte){
        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
        }
}
 
 
void pzmq_setSendBufferSize(zmq::socket_t* socket, int bufferSizeByte){
        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
        }
}
 
 
void pzmq_setThreadAffinity(zmq::socket_t* socket, size_t threadAffinity){
        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
        }
}
 
 
void pzmq_setBufferSize(zmq::socket_t* socket, int type, int nbBufferMessage, int dataRate, size_t bufferSizeByte){
        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);
        }
}