subscriber.hpp

Go to the documentation of this file.

#ifndef RAD_MAL_SUBSCRIBER_HPP
#define RAD_MAL_SUBSCRIBER_HPP

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

#include <mal/Cii.hpp>
#include <mal/Mal.hpp>
#include <mal/utility/LoadMal.hpp>

#include <functional>

namespace rad {
namespace cii {

template <typename TOPIC_TYPE>
class Subscriber {
   public:
    using TopicHandler_t = std::function<void(elt::mal::ps::Subscriber<TOPIC_TYPE>&,
                                              const elt::mal::ps::DataEvent<TOPIC_TYPE>&)>;

    Subscriber(const elt::mal::Uri& uri, TopicHandler_t handler,
               const std::optional<elt::mal::Mal::Properties> mal_properties = {})
        : m_subscriber(nullptr), m_subscription(nullptr) {
        RAD_TRACE(GetLogger());

        m_subscriber = elt::mal::CiiFactory::getInstance().getSubscriber<TOPIC_TYPE>(
            elt::mal::Uri(uri), elt::mal::ps::qos::QoS::DEFAULT,
            mal_properties ? *mal_properties : elt::mal::Mal::Properties());
        RAD_ASSERTPTR(m_subscriber);

        m_subscription =
            m_subscriber->subscribeAsync(elt::mal::ps::DataEventFilter<TOPIC_TYPE>::all(), handler);
        RAD_ASSERTPTR(m_subscription);

        LOG4CPLUS_DEBUG(GetLogger(), "Subscriber registered async.");
    }

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

    // Enable moving
    Subscriber(Subscriber&& rhs) = default;

   private:
    std::unique_ptr<elt::mal::ps::Subscriber<TOPIC_TYPE>>
        m_subscriber;  // Pointer to MAL Subscriber.
    std::unique_ptr<::elt::mal::ps::Subscription>
        m_subscription;  // Pointer to the topic subscription.
};

template <typename TOPIC_TYPE, typename EVENT_TYPE>
class SMSubscriber : public rad::cii::Subscriber<TOPIC_TYPE> {
   public:
    SMSubscriber(const elt::mal::Uri& uri, SMAdapter& sm,
                 const std::optional<elt::mal::Mal::Properties> mal_properties = {})
        : rad::cii::Subscriber<TOPIC_TYPE>(uri,
                                           std::bind(&SMSubscriber::Callback, this,
                                                     std::placeholders::_1, std::placeholders::_2),
                                           mal_properties),
          m_sm(sm) {
        RAD_TRACE(GetLogger());
    }

    void Callback(elt::mal::ps::Subscriber<TOPIC_TYPE>& subscriber,
                  const elt::mal::ps::DataEvent<TOPIC_TYPE>& event) {
        RAD_TRACE(GetLogger());

        if (event.hasValidData()) {
            LOG4CPLUS_DEBUG(GetLogger(), "Received topic, posting event to SM via ASIO.");
            m_sm.PostEvent(rad::UniqueEvent(new EVENT_TYPE(event.getData())));
        } else {
            LOG4CPLUS_WARN(GetLogger(), "Received invalid data!");
        }
    }

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

    // Enable moving ??
    // SMSubscriber(SMSubscriber&& rhs) = default;

   private:
    rad::SMAdapter& m_sm;
};

#if 0
template<typename TOPIC_TYPE, typename EVENT_TYPE>
class SubscriberSM {
public:
    explicit SubscriberSM(rad::SMAdapter& sm) : m_sm(sm) {
        RAD_TRACE(GetLogger());
    }

    ~SubscriberSM() {
        RAD_TRACE(GetLogger());
    }

    void Start(const std::string& uriStr,
               const std::string& malType) {
        RAD_TRACE(GetLogger());

        elt::mal::CiiFactory& factory = elt::mal::CiiFactory::getInstance();

        elt::mal::Mal::Properties mal_properties;
        auto mal_instance = elt::mal::loadMal(malType, mal_properties);

        auto uri = elt::mal::Uri(uriStr);
        std::string scheme = uri.scheme().to_string();
        factory.registerMal(scheme, mal_instance);

        mSubscriber = factory.getSubscriber<TOPIC_TYPE>(uri, elt::mal::ps::qos::QoS::DEFAULT, mal_properties);
        if (m_subscriber == nullptr) {
            throw std::runtime_error("Cannot create Subscriber");
        }

/*
        auto topicInstance = m_subscriber->createDataEntity();
        if (topicInstance == nullptr) {
            throw std::runtime_error("Subscriber cannot create data entity");
        }

        auto dataEventFilter = elt::mal::ps::DataEventFilter<TOPIC_TYPE>::all();
*/

        m_subscription = m_subscriber->subscribeAsync(elt::mal::ps::DataEventFilter<TOPIC_TYPE>::all(),
                                                      [this](elt::mal::ps::Subscriber<TOPIC_TYPE>& Subscriber,
                                                             const elt::mal::ps::DataEvent<TOPIC_TYPE>& event) {
            this->Callback(Subscriber, event);
        });
        if (m_subscription == nullptr) {
            throw std::runtime_error("Subscriber Async failed");
        }

        LOG4CPLUS_DEBUG(GetLogger(), "Subscriber registered async.");
    }

    void Stop() {
        RAD_TRACE(GetLogger());
        // @todo what happens to the subscription?
        m_subscriber->close();
    }

    void Callback(elt::mal::ps::Subscriber<TOPIC_TYPE>& Subscriber,
            const elt::mal::ps::DataEvent<TOPIC_TYPE>& event) {
        RAD_TRACE(GetLogger());

        if (event.hasValidData()) {
            //auto sample = event.getData();
            //auto altPos = sample->getAlt();
            //auto azPos = sample->getAz();
            //std::cout << "Counter " << ++mNumSamples << " Alt = " << altPos << " Az = " << azPos << std::endl;
            LOG4CPLUS_DEBUG(GetLogger(), "Received topic, posting event to SM ...");
            //rad::UniqueEvent(new MalEvents::DevMeas())
            m_sm.PostEvent(rad::UniqueEvent(new EVENT_TYPE(event.getData())));
        } else {
            LOG4CPLUS_WARN(GetLogger(), "Received invalid data!");
        }

    }

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

    // Enable moving
    //SubscriberSM(SubscriberSM&& rhs) = default;

private:
    rad::SMAdapter&          m_sm;
    UniqueEvent              m_topic_event;
    std::shared_ptr<elt::mal::ps::SubscriberSM<TOPIC_TYPE>> m_subscriber;
    std::unique_ptr<::elt::mal::ps::Subscription> m_subscription;
};
#endif

}  // namespace cii
}  // namespace rad

#endif  // RAD_MAL_SUBSCRIBER_HPP