msgRequestor.hpp

Go to the documentation of this file.

#ifndef RAD_MSG_REQUESTOR_HPP
#define RAD_MSG_REQUESTOR_HPP

#include <rad/assert.hpp>
#include <rad/errors.hpp>
#include <rad/exceptions.hpp>
#include <rad/logger.hpp>
#include <rad/msgRequestorRaw.hpp>

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

#include <boost/asio.hpp>

#include <google/protobuf/message.h>

#include <chrono>
#include <memory>

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)> reply_handler);
    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& err_code, const std::string& msg_type_id, const void* data,
                  const size_t data_size);

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

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)> reply_handler)
    : m_msg_requestor_raw(
          endpoint, identity, ios,
          std::bind(&MsgRequestor::Callback, this, std::placeholders::_1, std::placeholders::_2,
                    std::placeholders::_3, std::placeholders::_4)),
      m_reply_handler(reply_handler) {
    RAD_TRACE(GetLogger());
}

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

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

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

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

template <typename TYPEREQ, typename TYPEREP>
void MsgRequestor<TYPEREQ, TYPEREP>::Callback(const std::error_code& err_code,
                                              const std::string& msg_type_id, const void* data,
                                              const size_t data_size) {
    RAD_TRACE(GetLogger());

    TYPEREP reply;

    if (err_code) {
        m_reply_handler(err_code, reply);
        return;
    }

    if (reply.ParseFromArray(data, data_size)) {
        m_reply_handler({}, reply);
    } else {
        LOG4CPLUS_ERROR(GetLogger(), "Failed to parse reply type <" << msg_type_id << ">");
        m_reply_handler(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_TRACE(GetLogger());
        RAD_ASSERTPTR(payload.GetDescriptor());
        std::string payload_type = payload.GetDescriptor()->full_name();

        std::string str;
        if (payload.SerializeToString(&str) == false) {
            LOG4CPLUS_ERROR(GetLogger(),
                            "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_TRACE(GetLogger());
                reply_t reply;
                if (ec) {
                    handler(ec, reply);
                    return;
                }

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

   private:
    MsgRequestorRaw2 m_raw_requestor;
};

}  // namespace rad

#endif  // RAD_MSG_REQUESTOR_HPP