PZmqBackend.cpp

Go to the documentation of this file.

/***************************************
        Auteur : Pierre Aubert
        Mail : pierre.aubert@lapp.in2p3.fr
        Licence : CeCILL-C
****************************************/
 
#include "PZmqBackend.h"


PSendStatus::PSendStatus checkSendStatus(zmq::send_result_t res){
        if(res.has_value()){
                return PSendStatus::OK;
        }else{
                int err = zmq_errno();
                if(err == EAGAIN){
                        return PSendStatus::SOCKET_NOT_AVAILABLE;
                }else{
                        std::cerr << "Unknown ZMQ error in send: '" <<  err << "'"<< std::endl;
                        return PSendStatus::BROKEN_BACKEND;
                }
        }
}


PRecvStatus::PRecvStatus checkRecvStatus(zmq::recv_result_t res){
        if(res.has_value()){
                return PRecvStatus::OK;
        }else{
                int err = zmq_errno();
                if(err == EAGAIN){
                        return PRecvStatus::NO_MESSAGE_RECEIVED;
                }else{
                        return PRecvStatus::BROKEN_SOCKET;
                }
        }
}


zmq::send_flags convertToSendFlag(PSendFlag::PSendFlag flag){
        if(flag == PSendFlag::NON_BLOCK){return zmq::send_flags::dontwait;}
        else{return zmq::send_flags::none;}
}


zmq::recv_flags convertToRecvFlag(PRecvFlag::PRecvFlag flag){
        if(flag == PRecvFlag::NON_BLOCK){return zmq::recv_flags::dontwait;}
        else{return zmq::recv_flags::none;}
}


PZmqParam pzmq_createParamClient(int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int linger){
        PZmqParam param;
        param.type = type;
        param.nbBufferMessage = nbBufferMessage;
        param.bufferSizeByte = bufferSizeByte;
        param.threadAffinity = threadAffinity;
        param.dataRate = dataRate;
        param.linger = linger;
        return param;
}


PZmqParam pzmq_createParamServer(int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int linger){
        PZmqParam param;
        param.type = type;
        param.nbBufferMessage = nbBufferMessage;
        param.bufferSizeByte = bufferSizeByte;
        param.threadAffinity = threadAffinity;
        param.dataRate = dataRate;
        param.linger = linger;
        return param;
}

PZmqSocket::PZmqSocket()
        :p_socket(NULL)
{
 
}

PZmqSocket::~PZmqSocket(){
        close();
}


bool PZmqSocket::createClientSocket(zmq::context_t & context, const PSocketParam & socketParam, const Param & extraParam){
        p_socket = pzmq_createClientSocket(context, socketParam.hostname, socketParam.port, extraParam.type, extraParam.nbBufferMessage,
                extraParam.bufferSizeByte, extraParam.threadAffinity, extraParam.dataRate);
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
        p_socket->set(zmq::sockopt::rcvtimeo, socketParam.recvTimeOut);
        p_socket->set(zmq::sockopt::sndtimeo, socketParam.sendTimeOut);
        p_socket->set(zmq::sockopt::linger, extraParam.linger);   //number of ms to stop
#else
        p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
        p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
        p_socket->setsockopt(ZMQ_LINGER, extraParam.linger);              //number of ms to stop
#endif
        return p_socket != NULL;
}


bool PZmqSocket::createServerSocket(zmq::context_t & context, const PSocketParam & socketParam, const Param & extraParam){
        p_socket = pzmq_createServerSocket(context, socketParam.port, extraParam.type, extraParam.nbBufferMessage, extraParam.bufferSizeByte,
                extraParam.threadAffinity, extraParam.dataRate);
#if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471 
        p_socket->set(zmq::sockopt::rcvtimeo, socketParam.recvTimeOut);
        p_socket->set(zmq::sockopt::sndtimeo, socketParam.sendTimeOut);
        p_socket->set(zmq::sockopt::linger, extraParam.linger);   //number of ms to stop
#else
        p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
        p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
        p_socket->setsockopt(ZMQ_LINGER, extraParam.linger);              //number of ms to stop
#endif
        return p_socket != NULL;
}


PSendStatus::PSendStatus PZmqSocket::sendMsg(Message & msg, PSendFlag::PSendFlag flag){
        try {
                zmq::send_result_t result = p_socket->send(msg, convertToSendFlag(flag));
                return checkSendStatus(result);
        }catch(const zmq::error_t& e) {
                if(e.num() == ENOTSOCK){
                        return PSendStatus::BROKEN_SOCKET;
                }else if(e.num() == ENOTSUP || e.num() == EFSM || e.num() == ETERM || e.num() == EINVAL){
                        return PSendStatus::BROKEN_BACKEND;
                }else if(e.num() == EINTR){
                        return PSendStatus::SIGNAL_INTERRUPTION;
                }else if(e.num() == EHOSTUNREACH){
                        return PSendStatus::NO_ROUTE_TO_RECEIVER;
                }else{
                        std::cerr << "Unknown ZMQ error in send: " << e.what() << std::endl;
                        return PSendStatus::BROKEN_BACKEND;
                }
        }
}


PRecvStatus::PRecvStatus PZmqSocket::recvMsg(Message & msg, PRecvFlag::PRecvFlag flag){
        try {
                zmq::recv_result_t result = p_socket->recv(msg, convertToRecvFlag(flag));
                return checkRecvStatus(result);
        }catch(const zmq::error_t& e){
                if(e.num() == ENOTSOCK){
                        std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
                        return PRecvStatus::BROKEN_SOCKET;
                }
                else if(e.num() == ENOTSUP || e.num() == EFSM || e.num() == ETERM || e.num() == EINVAL){
                        std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
                        return PRecvStatus::BROKEN_BACKEND;
                }
                else if(e.num() == EINTR){
                        std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
                        return PRecvStatus::SIGNAL_INTERRUPTION;
                }
                else{
                        std::cerr << "Unknown ZMQ error in recv: " << e.what() << std::endl;
                        return PRecvStatus::BROKEN_BACKEND;
                }
        }
}


bool PZmqSocket::isConnected() const{
        if(p_socket != NULL){
                return p_socket->handle() != NULL;
        }
        return false;
}

void PZmqSocket::close(){
        if(p_socket != NULL){
                p_socket->close();
                delete p_socket;
                p_socket = NULL;
        }
}

PZmqSocketGenerator::PZmqSocketGenerator()
:p_context(1)
{
 
}


PZmqSocketGenerator::Param PZmqSocketGenerator::client(){
        return pzmq_createParamClient(ZMQ_PULL);
}


PZmqSocketGenerator::Param PZmqSocketGenerator::server(){
        return pzmq_createParamServer(ZMQ_PUSH);
}


bool PZmqSocketGenerator::createClientSocket(PZmqSocketGenerator::Socket & socket, const PSocketParam & socketParam, const PZmqParam & param){
    return socket.createClientSocket(p_context, socketParam, param);
}


bool PZmqSocketGenerator::createServerSocket(PZmqSocketGenerator::Socket & socket, const PSocketParam & socketParam, const PZmqParam & param){
        return socket.createServerSocket(p_context, socketParam, param);
}


void PZmqSocketGenerator::msgToMock(DataStreamMsg & mockMsg, const PZmqSocketGenerator::Message & msg){
        size_t dataSize(msg.size());
        mockMsg.resize(dataSize);
        memcpy(mockMsg.data(), (const void*)msg.data(), dataSize);
}


void PZmqSocketGenerator::mockToMsg(PZmqSocketGenerator::Message & msg, DataStreamMsg & mockMsg){
        size_t dataSize(mockMsg.size());
        msg.rebuild(dataSize);
        memcpy((void*)msg.data(), mockMsg.data(), dataSize);
}