MsgRequestor.hpp

Go to the documentation of this file.

#ifndef RAD_MSG_REQUESTOR_HPP
#define RAD_MSG_REQUESTOR_HPP

#include <chrono>
#include <memory>

#include <rad/Logger.hpp>
#include <rad/Assert.hpp>
#include <rad/Exceptions.hpp>
#include <rad/Errors.hpp>
#include <rad/MsgRequestorRaw.hpp>

#include <azmq/socket.hpp>
#include <azmq/message.hpp>

#include <boost/asio.hpp>

#include <google/protobuf/message.h>

namespace rad {

template<typename TYPEREQ, typename TYPEREP>
class MsgRequestor {

public:
    MsgRequestor(const std::string& endpoint,
                 const std::string& identity,
                 boost::asio::io_service& ios,
                 std::function<void(const std::error_code&, TYPEREP)> replyHandler);
    virtual ~MsgRequestor();

    size_t  Send(const TYPEREQ& payload, const long timeout = 0);

    MsgRequestor(const MsgRequestor&) = delete;
    MsgRequestor& operator= (const MsgRequestor&) = delete;

private:
    void Callback(const std::error_code& errCode, const std::string& msgTypeId, const void* pData, const size_t dataSize);

    MsgRequestorRaw              mMsgRequestorRaw;
    std::function<void(const std::error_code& , TYPEREP)> mReplyHandler;
};

template<typename TYPEREQ, typename TYPEREP>
MsgRequestor<TYPEREQ, TYPEREP>::MsgRequestor(const std::string& endpoint,
                                             const std::string& identity,
                                             boost::asio::io_service& ios,
                                             std::function<void(const std::error_code& , TYPEREP)> replyHandler)
: mMsgRequestorRaw(endpoint, identity, ios, std::bind(&MsgRequestor::Callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)),
  mReplyHandler(replyHandler)
{
    RAD_LOG_TRACE();
}

template<typename TYPEREQ, typename TYPEREP>
MsgRequestor<TYPEREQ, TYPEREP>::~MsgRequestor()
{
    RAD_LOG_TRACE();
}

template<typename TYPEREQ, typename TYPEREP>
size_t MsgRequestor<TYPEREQ, TYPEREP>::Send(const TYPEREQ& payload, const long timeout)
{
    RAD_LOG_TRACE();

    /*
     * Important: never call google::protobuf::ShutdownProtobufLibrary()
     * before accessing the message descriptor!
     */
    RAD_ASSERTPTR(payload.GetDescriptor());
    std::string payloadType = payload.GetDescriptor()->full_name();

    std::string str;
    if (payload.SerializeToString(&str) == false) {
        RAD_LOG_ERROR() << "Failed serializing to string payload type <" << payloadType << ">";
        return 0;
    }
    return mMsgRequestorRaw.Send(payloadType, str, timeout);
}

template<typename TYPEREQ, typename TYPEREP>
void MsgRequestor<TYPEREQ, TYPEREP>::Callback(const std::error_code& errCode, const std::string& msgTypeId, const void* pData, const size_t dataSize)
{
    RAD_LOG_TRACE();

    TYPEREP reply;

    if (errCode) {
        mReplyHandler(errCode, reply);
    }

    if (reply.ParseFromArray(pData, dataSize)) {
        mReplyHandler({}, reply);
    } else {
        RAD_LOG_ERROR() << "Failed to parse reply type <" << msgTypeId << ">";
        mReplyHandler(rad::ErrorCodes::DESERIALIZATION_ERR, reply);
        // @todo throw exception?
    }
}


template<typename TREQ, typename TREP>
struct MsgRequestor2 {
     using request_t = TREQ;
     using reply_t = TREP;
     using handler_t = std::function<void(const std::error_code&, reply_t)>;

     MsgRequestor2(const std::string& endpoint,
                   const std::string& identity,
                   boost::asio::io_service& ios)
         : m_raw_requestor(endpoint, identity, ios) {
     }
     virtual ~MsgRequestor2(){};

     MsgRequestor2(const MsgRequestor2&) = delete;
     MsgRequestor2& operator=(const MsgRequestor2&) = delete;
     MsgRequestor2(MsgRequestor2&&) = default;
     MsgRequestor2& operator=(MsgRequestor2&&) = default;

     size_t
     AsyncSendReceive(const request_t& payload,
                      handler_t handler,
                      std::chrono::milliseconds const timeout = std::chrono::milliseconds(0)) {
         RAD_LOG_TRACE();
         RAD_ASSERTPTR(payload.GetDescriptor());
         std::string payload_type = payload.GetDescriptor()->full_name();

         std::string str;
         if (payload.SerializeToString(&str) == false) {
             RAD_LOG_ERROR() << "Failed serializing to string payload type <" << payload_type
                             << ">";
             return 0;
         }
         return m_raw_requestor.AsyncSendReceive(
             payload_type, str,
             [handler](std::error_code const& ec, std::string const& msg_type_id,
                       const void* p_data, const size_t size) {
                 RAD_LOG_TRACE();
                 reply_t reply;
                 if (ec) {
                     handler(ec, reply);
                     return;
                 }


                 // Assert p_data != nullptr
                 if (reply.ParseFromArray(p_data, size)) {
                     handler({}, reply);
                 } else {
                     RAD_LOG_ERROR() << "Failed to parse reply type <" << msg_type_id << ">";
                     handler(rad::ErrorCodes::DESERIALIZATION_ERR, reply);
                 }
             },
             timeout);
     }

   private:
     MsgRequestorRaw2 m_raw_requestor;
 };

} // namespace rad

#endif