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)
{
        PZmqParam param;
        param.type = type;
        param.nbBufferMessage = nbBufferMessage;
        param.bufferSizeByte = bufferSizeByte;
        param.threadAffinity = threadAffinity;
        param.dataRate = dataRate;
        return param;
}


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

PZmqSocket::PZmqSocket()
:p_socket(nullptr)
{
 
}

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);
#else
                p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
                p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
#endif
    return p_socket != nullptr;
}


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);
#else
                p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
                p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
#endif
    return p_socket != nullptr;
}


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){
                        return PRecvStatus::BROKEN_SOCKET;
                }
                else if(e.num() == ENOTSUP || e.num() == EFSM || e.num() == ETERM || e.num() == EINVAL){
                        return PRecvStatus::BROKEN_BACKEND;
                }
                else if(e.num() == EINTR){
                        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 != nullptr){
        return p_socket->handle() != nullptr;
    }
    return false;
}

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

PZmqBackend::PZmqBackend()
:p_context(1)
{
 
}


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


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


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


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


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


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