PZmqBackend.cpp

Go to the documentation of this file.

/***************************************
        Auteur : Pierre Aubert
        Mail : pierre.aubert@lapp.in2p3.fr
        Licence : CeCILL-C
****************************************/
 
#include "PZmqBackend.h"
 
 
PZmqParam pzmq_createParamClient(const std::string & address, size_t port, zmq::context_t* context, int type, int nbBufferMessage,
                                int bufferSizeByte, size_t threadAffinity, ssize_t dataRate)
{
        PZmqParam param;
        param.address = address;
        param.port = port;
        param.context = context;
        param.type = type;
        param.nbBufferMessage = nbBufferMessage;
        param.bufferSizeByte = bufferSizeByte;
        param.threadAffinity = threadAffinity;
        param.dataRate = dataRate;
        return param;
}
 
 
PZmqParam pzmq_createParamServer(const std::string & address, size_t port, zmq::context_t* context, int type, int nbBufferMessage,
                                int bufferSizeByte, size_t threadAffinity, ssize_t dataRate)
{
        PZmqParam param;
        param.address = address;
        param.port = port;
        param.context = context;
        param.type = type;
        param.nbBufferMessage = nbBufferMessage;
        param.bufferSizeByte = bufferSizeByte;
        param.threadAffinity = threadAffinity;
        param.dataRate = dataRate;
        return param;
}
 
PZmqBackend::PZmqBackend(){
        
}
 
 
PZmqBackend::Param PZmqBackend::client(const std::string & address, size_t port, zmq::context_t* context, int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate){
        return pzmq_createParamClient(address, port, context, type, nbBufferMessage, bufferSizeByte, threadAffinity, dataRate);
}
 
 
PZmqBackend::Param PZmqBackend::server(const std::string & address, size_t port, zmq::context_t* context, int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate){
        return pzmq_createParamServer(address, port, context, type, nbBufferMessage, bufferSizeByte, threadAffinity, dataRate);
}
 
 
bool PZmqBackend::createClientSocket(PZmqBackend::Socket & socket, const std::string & address, size_t port, const PZmqParam & param){
        socket = pzmq_createClientSocket(*param.context, address, port, param.type, param.nbBufferMessage,
                                        param.bufferSizeByte, param.threadAffinity, param.dataRate);
        return socket != NULL;
}
 
 
bool PZmqBackend::createServerSocket(PZmqBackend::Socket & socket, const std::string & address, size_t port, const PZmqParam & param){
        socket = pzmq_createServerSocket(*param.context, port, param.type, param.nbBufferMessage, param.bufferSizeByte,
                                                        param.threadAffinity, param.dataRate);
        return socket != NULL;
}
 
 
bool PZmqBackend::createClientSocket(Socket & socket, const PZmqBackend::Param & param){
        socket = pzmq_createClientSocket(*param.context, param.address, param.port, param.type, param.nbBufferMessage,
                                        param.bufferSizeByte, param.threadAffinity, param.dataRate);
        return socket != NULL;
}
 
 
bool PZmqBackend::createServerSocket(Socket & socket, const PZmqBackend::Param & param){
        socket = pzmq_createServerSocket(*param.context, param.port, param.type, param.nbBufferMessage, param.bufferSizeByte,
                                                        param.threadAffinity, param.dataRate);
        return socket != NULL;
}
 
 
zmq::send_flags convertToSendFlag(PSendFlag::PSendFlag flag){
        if(flag == PSendFlag::NON_BLOCK){return zmq::send_flags::dontwait;}
        else{return zmq::send_flags::none;}
}
 
 
bool PZmqBackend::send(PZmqBackend::Socket & socket, PZmqBackend::Message & msg, PSendFlag::PSendFlag flag){
        if(socket != NULL){
                zmq::send_result_t res = socket->send(msg, convertToSendFlag(flag));
                return res.has_value();         //Seems, if there is no value, there is also no error
        }else{
                return false;
        }
}
 
 
zmq::recv_flags convertToRecvFlag(PRecvFlag::PRecvFlag flag){
        if(flag == PRecvFlag::NON_BLOCK){return zmq::recv_flags::dontwait;}
        else{return zmq::recv_flags::none;}
}
 
 
bool PZmqBackend::recv(PZmqBackend::Socket & socket, PZmqBackend::Message & msg, PRecvFlag::PRecvFlag flag){
        if(socket != NULL){
                zmq::recv_result_t res = socket->recv(msg, convertToRecvFlag(flag));
                return res.has_value();         //Seems, if there is no value, there is also no error
        }else{
                return false;
        }
}
 
 
void PZmqBackend::msgResize(PZmqBackend::Message & msg, size_t sizeMsg){
        msg.rebuild(sizeMsg);
}
 
 
size_t PZmqBackend::msgSize(const PZmqBackend::Message & msg){
        return msg.size();
}
 
 
const DataStreamIter PZmqBackend::msgData(const PZmqBackend::Message & msg){
        return (const DataStreamIter)msg.data();
}
 
 
DataStreamIter PZmqBackend::msgData(PZmqBackend::Message & msg){
        return (DataStreamIter)msg.data();
}
 
 
void PZmqBackend::close(PZmqBackend::Socket & socket){
        pzmq_closeServerSocket(socket);
}
 
 
bool PZmqBackend::isConnected(const PZmqBackend::Socket & socket){
        if(socket != NULL){
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
                return socket->handle() != NULL;
#else
                return socket->connected();
#endif
        }else{
                return false;
        }
}
 
 
void PZmqBackend::msgToMock(DataStreamMsg & mockMsg, const PZmqBackend::Message & msg){
        size_t dataSize(PZmqBackend::msgSize(msg));
        mockMsg.resize(dataSize);
        memcpy(mockMsg.data(), PZmqBackend::msgData(msg), dataSize);
}
 
 
void PZmqBackend::mockToMsg(PZmqBackend::Message & msg, DataStreamMsg & mockMsg){
        size_t dataSize(mockMsg.size());
        PZmqBackend::msgResize(msg, dataSize);
        memcpy(PZmqBackend::msgData(msg), mockMsg.data(), dataSize);
}