msgRequestorRaw.hpp

Go to the documentation of this file.

#ifndef RAD_MSG_REQUESTOR_RAW_HPP
#define RAD_MSG_REQUESTOR_RAW_HPP

#include <chrono>

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

#include <boost/asio.hpp>

#include <google/protobuf/message.h>

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

namespace rad {

class MsgRequestorRaw {
   public:
    MsgRequestorRaw(
        const std::string& endpoint, const std::string& identity, boost::asio::io_service& ios,
        std::function<void(const std::error_code& err_code, const std::string& msg_type_id,
                           const void* data, const size_t data_size)>
            reply_handler);
    virtual ~MsgRequestorRaw();

    size_t Send(const std::string& payload_type, const std::string& payload,
                const long timeout = 0);

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

   private:
    void ReplyCallback(const boost::system::error_code& err_code, azmq::message& msg,
                       size_t nbytes);
    void TimeoutCallback(const boost::system::error_code& err_code);
    bool InstallTimer(const long timeout /*msec*/);
    void CancelTimer();

    std::string m_endpoint;
    std::string m_identity;
    boost::asio::io_service& m_ios;
    azmq::dealer_socket m_socket;
    boost::asio::deadline_timer* m_deadline_timer;

    std::function<void(const std::error_code& err_code, std::string const& msg_type_id,
                       const void* data, const size_t data_size)>
        m_reply_handler;
};

struct MsgRequestorRaw2 {
    using handler_t =
        std::function<void(const std::error_code& /*err_code*/, std::string const& /*msg_type_id*/,
                           const void* /*data*/, const size_t /*data_size*/)>;
    MsgRequestorRaw2(const std::string& endpoint, const std::string& identity,
                     boost::asio::io_service& io_ctx);
    ~MsgRequestorRaw2();

    size_t AsyncSendReceive(std::string const& payload_type, std::string const& payload,
                            handler_t handler,
                            std::chrono::milliseconds const timeout = std::chrono::milliseconds(0));
    size_t AsyncSend(std::string const& payload_type, std::string const& payload);
    void AsyncReceive(handler_t handler,
                      std::chrono::milliseconds const timeout = std::chrono::milliseconds(0));
    void Cancel() noexcept;

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

   private:
    std::string m_endpoint;
    std::string m_identity;
    boost::asio::io_service& m_io_ctx;
    azmq::dealer_socket m_socket;
    handler_t m_handler;
};

}  // namespace rad

#endif