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>

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::optional<elt::mal::Mal::Properties> mal_properties = {})
        : m_client() {
        RAD_TRACE(GetLogger());

        m_client = elt::mal::CiiFactory::getInstance().getClient<INTERFACE_TYPE>(
            uri, elt::mal::rr::qos::QoS::DEFAULT,
            mal_properties ? *mal_properties : elt::mal::Mal::Properties());
        LOG4CPLUS_DEBUG(GetLogger(), "Created MAL client for <" << uri << ">");
    }

    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_