requestor.hpp

Go to the documentation of this file.

 
#ifndef RAD_MAL_REQUESTOR_HPP
#define RAD_MAL_REQUESTOR_HPP
 
#include <type_traits>
 
#include <rad/logger.hpp>
#include <rad/mal/utils.hpp>
#include <rad/smAdapter.hpp>
 
#include <mal/Cii.hpp>
#include <mal/Mal.hpp>
#include <mal/utility/LoadMal.hpp>
#include <mal/rr/qos/QoS.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/rr/qos/ConnectionTime.hpp>
 
namespace rad {
 
/*
 * Forward declaration to avoid dependency with rad.sm.
 */
class SMAdapter;
 
namespace cii {
 
template <typename INTERFACE_TYPE>
class Requestor {
   public:
    Requestor(const elt::mal::Uri& uri,
              const std::vector<std::shared_ptr<elt::mal::rr::qos::QoS>>& qos,
              const std::optional<elt::mal::Mal::Properties> mal_properties = {})
        : m_client() {
        RAD_TRACE(GetLogger());
 
        m_client = elt::mal::CiiFactory::getInstance().getClient<INTERFACE_TYPE>(
            uri, qos, mal_properties.value_or(elt::mal::Mal::Properties()));
        LOG4CPLUS_DEBUG(GetLogger(), "Created MAL client for <" << uri << ">");
    }
 
    explicit Requestor(const elt::mal::Uri& uri,
              const std::optional<elt::mal::Mal::Properties> mal_properties = {})
        : Requestor(uri, elt::mal::rr::qos::QoS::DEFAULT, mal_properties) {}
 
 
    std::shared_ptr<INTERFACE_TYPE>& GetInterface() {
        RAD_TRACE(GetLogger());
        return m_client;
    }
 
    Requestor(const Requestor&) = delete;
    Requestor& operator=(const Requestor&) = delete;
 
   private:
    std::shared_ptr<INTERFACE_TYPE> m_client;  // Share pointer to MAL Client
};
 
template <class EVENT, class T>
void RoutePartialReply(std::shared_ptr<::elt::mal::rr::Ami<T>> ami, rad::SMAdapter& sm) {
    ::elt::mal::future<T> fut = ami->next();
    fut.then([=, &sm](::elt::mal::future<T> fut) {
        try {
            LOG4CPLUS_DEBUG(GetLogger(),
                            "Received async partial reply, triggering associated event <"
                                << typeid(EVENT).name() << ">");
            sm.PostEvent(std::make_unique<EVENT>(fut.get()));
            LOG4CPLUS_DEBUG(GetLogger(),
                            "Received async partial reply, triggering associated event - done!");
            if (ami->isDone() == false) {
                RoutePartialReply<EVENT, T>(ami, sm);
                LOG4CPLUS_DEBUG(GetLogger(),
                                "Invoked RoutePartialReply for the next partial/final reply");
            }
        } catch (...) {
            LOG4CPLUS_ERROR(GetLogger(), "Unknown exception when receiving partial reply");
        }
    });
    LOG4CPLUS_DEBUG(log4cplus::Logger::getInstance(rad::LOGGER_NAME), "Installed AMI continuation");
}
 
template <class EVENT, class FUT>
CancellationToken RouteReply(FUT&& rep_future, rad::SMAdapter& sm) {
    CancellationToken token;
    rep_future.then([&](FUT res) {
        if (token.IsCancelled()) {
            return;
        }
        try {
            LOG4CPLUS_DEBUG(GetLogger(), "Received async reply, triggering associated event <"
                                             << typeid(EVENT).name() << ">");
            sm.PostEvent(std::make_unique<EVENT>(res));
            LOG4CPLUS_DEBUG(GetLogger(),
                            "Received async reply, triggering associated event - done!");
        } catch (...) {
            LOG4CPLUS_ERROR(GetLogger(), "Unknown exception when receiving the reply");
        }
    });
    LOG4CPLUS_DEBUG(GetLogger(), "Installed future continuation");
    return token;
}
 
template <class EVENT, class EVENT_ERR, class FUT>
CancellationToken RouteReply(FUT&& rep_future, rad::SMAdapter& sm) {
    CancellationToken token;
    rep_future.then([&, token](FUT res) {
        if (token.IsCancelled()) {
            return;
        }
        try {
            if (res.has_value()) {
                // NOLINTNEXTLINE
                if constexpr (std::is_void<typename FUT::value_type>::value) {
                    LOG4CPLUS_DEBUG(GetLogger(),
                                    "Received async VOID reply, triggering associated event <"
                                        << typeid(EVENT).name() << ">");
                    sm.PostEvent(std::make_unique<EVENT>());
                } else {
                    // LOG4CPLUS_DEBUG(GetLogger(), "Received async reply <" << res.get() << ">,
                    // triggering associated event <" << typeid(EVENT).name() << ">");
                    LOG4CPLUS_DEBUG(GetLogger(),
                                    "Received async reply, triggering associated event <"
                                        << typeid(EVENT).name() << ">");
                    sm.PostEvent(std::make_unique<EVENT>(res.get()));
                }
                LOG4CPLUS_DEBUG(GetLogger(),
                                "Received async reply, triggering associated event - done!");
            } else if (res.has_exception()) {
                LOG4CPLUS_DEBUG(GetLogger(), "Exception waiting for reply");
                sm.PostEvent(std::make_unique<EVENT_ERR>(res.get_exception_ptr()));
            } else {
                RAD_ASSERTNEVER();
            }
        } catch (...) {
            LOG4CPLUS_ERROR(GetLogger(), "Unknown exception when receiving the reply");
        }
    });
    LOG4CPLUS_DEBUG(GetLogger(), "Installed future continuation");
    return token;
}
 
template <class EVENT, class EVENT_TIMEOUT, class EVENT_ERR, class FUT>
CancellationToken RouteReplyWithTimeout(FUT&& rep_future, rad::SMAdapter& sm) {
    CancellationToken token;
    rep_future.then([&, token](FUT res) {
        if (token.IsCancelled()) {
            return;
        }
        try {
            if (res.has_value()) {
                // NOLINTNEXTLINE
                if constexpr (std::is_void<typename FUT::value_type>::value) {
                    LOG4CPLUS_DEBUG(GetLogger(),
                                    "Received async VOID reply, triggering associated event <"
                                        << typeid(EVENT).name() << ">");
                    sm.PostEvent(std::make_unique<EVENT>());
                } else {
                    // LOG4CPLUS_DEBUG(GetLogger(), "Received async reply <" << res.get() << ">,
                    // triggering associated event <" << typeid(EVENT).name() << ">");
                    LOG4CPLUS_DEBUG(GetLogger(),
                                    "Received async reply, triggering associated event <"
                                        << typeid(EVENT).name() << ">");
                    sm.PostEvent(std::make_unique<EVENT>(res.get()));
                }
                LOG4CPLUS_DEBUG(GetLogger(),
                                "Received async reply, triggering associated event - done!");
            } else if (res.has_exception()) {
                LOG4CPLUS_DEBUG(GetLogger(), "Received exception as reply, rethrowing it.");
                boost::rethrow_exception(res.get_exception_ptr());
            } else {
                RAD_ASSERTNEVER();
            }
        } catch (const elt::mal::TimeoutException& e) {
            LOG4CPLUS_DEBUG(GetLogger(), "ExceptionTimeout while waiting for reply");
            sm.PostEvent(std::make_unique<EVENT_TIMEOUT>());
        } catch (...) {
            LOG4CPLUS_DEBUG(GetLogger(), "Received exception as reply, post associated event <"
                                             << typeid(EVENT).name() << ">");
            sm.PostEvent(std::make_unique<EVENT_ERR>(res.get_exception_ptr()));
        }
    });
    LOG4CPLUS_DEBUG(GetLogger(), "Installed future continuation");
    return token;
}
 
}  // namespace cii
}  // namespace rad
 
#endif  // RAD_MAL_REQUESTOR_HPP_