typedEventService.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_TYPEDEVENTSERVICE_HPP
#define RTCTK_COMPONENTFRAMEWORK_TYPEDEVENTSERVICE_HPP
 
#include <rtctk/componentFramework/eventDefinitions.hpp>
#include <rtctk/componentFramework/eventServiceIf.hpp>
#include <rtctk/componentFramework/jsonPayload.hpp>
 
#include <map>
#include <mutex>
#include <set>
#include <string>
 
namespace rtctk::componentFramework {
 
namespace detail {
 
// internal helper class to keep track of publishers
class TypedEventPublisherManager {
public:
    explicit TypedEventPublisherManager(EventServiceIf& es) : m_es(es) {
    }
 
    template <typename EventType>
    void RegisterPublisher() {
        std::lock_guard guard(m_mutex);
 
        const std::string tpc_name = EventType::TOPIC_NAME;
 
        if (m_hl_publishers.count(tpc_name) == 0) {
            m_hl_publishers[tpc_name] = 1;
            m_ll_publishers[tpc_name] = m_es.MakePublisher(tpc_name);
        } else {
            m_hl_publishers[tpc_name] += 1;
        }
    }
 
    template <typename EventType>
    void UnregisterPublisher() {
        std::lock_guard guard(m_mutex);
 
        const std::string tpc_name = EventType::TOPIC_NAME;
 
        if (m_hl_publishers.count(tpc_name)) {
            if (m_hl_publishers.at(tpc_name) >= 1) {
                m_hl_publishers.at(tpc_name) -= 1;
            }
 
            if (m_hl_publishers.at(tpc_name) == 0) {
                m_hl_publishers.erase(tpc_name);
                m_ll_publishers.erase(tpc_name);
            }
        }
    }
 
    template <typename EventType>
    void Publish(const EventType& sample) {
        std::lock_guard guard(m_mutex);
 
// we dont need any deprecation warnings in already deprecated code
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
        const std::string tpc_name = EventType::TOPIC_NAME;
        m_ll_publishers.at(tpc_name)->Publish(sample.ToJson());
#pragma GCC diagnostic pop
    }
 
private:
    // handle to low-level event service
    EventServiceIf& m_es;
 
    // to synchronise access on the data structure
    std::mutex m_mutex;
 
    // to keep track of publishers
    std::map<std::string, unsigned> m_hl_publishers;
    std::map<std::string, std::unique_ptr<EventPublisherIf>> m_ll_publishers;
};
 
// internal helper class to keep track of subscribers
class TypedEventSubscriberManager {
public:
    explicit TypedEventSubscriberManager(EventServiceIf& es) : m_es(es) {
    }
 
    template <typename EventType>
    void RegisterSubscriber(uintptr_t id) {
        std::lock_guard guard(m_mutex);
 
        const std::string tpc_name = EventType::TOPIC_NAME;
 
        if (m_hl_subscribers.count(tpc_name) == 0) {
            m_hl_subscribers[tpc_name].insert(id);
            m_ll_subscribers[tpc_name] = m_es.MakeSubscriber(tpc_name);
        } else {
            m_hl_subscribers[tpc_name].insert(id);
        }
    }
 
    template <typename EventType>
    void UnregisterSubscriber(uintptr_t id) {
        std::lock_guard guard(m_mutex);
 
        UnsubscribePrivate<EventType>(id);
 
        // delete subscriber
        const std::string tpc_name = EventType::TOPIC_NAME;
 
        if (m_hl_subscribers.count(tpc_name)) {
            m_hl_subscribers.at(tpc_name).erase(id);
 
            if (m_hl_subscribers.at(tpc_name).empty()) {
                m_hl_subscribers.erase(tpc_name);
                m_ll_subscribers.erase(tpc_name);
            }
        }
    }
 
    template <typename EventType>
    void Subscribe(uintptr_t id, std::function<void(const EventType&)> cb) {
        std::lock_guard guard(m_mutex);
 
        const std::string tpc_name = EventType::TOPIC_NAME;
        const std::string evt_type = PrettyTypeName<EventType>();
 
        if (m_callbacks.count(tpc_name) == 0) {
            m_ll_subscribers.at(tpc_name)->Subscribe([this, tpc_name](const JsonPayload& pl) {
                std::lock_guard guard(m_mutex);
 
                if (m_callbacks.count(tpc_name) != 0) {
                    for (const auto& [key, val] : m_callbacks.at(tpc_name)) {
                        for (const auto& [k, obj] : val) {
                            (*obj)(pl);  // invoke the callback
                        }
                    }
                }
            });
        }
 
        m_callbacks[tpc_name][evt_type][id] = std::make_shared<FuncObj<EventType>>(cb);
    }
 
    template <typename EventType>
    void Unsubscribe(uintptr_t id) {
        std::lock_guard guard(m_mutex);
 
        UnsubscribePrivate<EventType>(id);
    }
 
private:
    template <typename EventType>
    void UnsubscribePrivate(uintptr_t id) {
        const std::string tpc_name = EventType::TOPIC_NAME;
        const std::string evt_type = PrettyTypeName<EventType>();
 
        if (m_callbacks.count(tpc_name) and m_callbacks.at(tpc_name).count(evt_type) and
            m_callbacks.at(tpc_name).at(evt_type).count(id)) {
            m_callbacks.at(tpc_name).at(evt_type).erase(id);
            if (m_callbacks.at(tpc_name).at(evt_type).empty()) {
                m_callbacks.at(tpc_name).erase(evt_type);
                if (m_callbacks.at(tpc_name).empty()) {
                    m_callbacks.erase(tpc_name);
                    m_ll_subscribers.at(tpc_name)->Unsubscribe();
                }
            }
        }
    }
 
    // Basic functor interface for book-keeping
    class FuncObjBase {
    public:
        virtual ~FuncObjBase() = default;
        virtual void operator()(const JsonPayload& sample) = 0;
    };
 
    // Specific functor classes that host event callbacks
    template <typename EventType>
    class FuncObj : public FuncObjBase {
    public:
        static_assert(std::is_base_of_v<AbstractEvent, EventType>,
                      "EventType must extend AbstractEvent");
 
        explicit FuncObj(std::function<void(const EventType&)> cb) : m_cb(cb) {
        }
 
        void operator()(const JsonPayload& sample) override {
            if (std::count(
                    sample["type"].begin(), sample["type"].end(), PrettyTypeName<EventType>())) {
// we dont need any deprecation warnings in already deprecated code
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
                m_cb(EventType(sample));
#pragma GCC diagnostic pop
            }
        }
 
    private:
        std::function<void(const EventType&)> m_cb;
    };
 
    // handle to low-level event service
    EventServiceIf& m_es;
 
    // to keep track of subscribers
    std::map<std::string, std::set<uintptr_t>> m_hl_subscribers;
    std::map<std::string, std::unique_ptr<EventSubscriberIf>> m_ll_subscribers;
 
    // We keep track of our event type subscriptions in a map of maps of maps of function objects:
    // - the first level map keys are the currently subscribed 'topic names'
    // - the second level map keys are the currently subscribed 'event types' of a 'topic name'
    //   note that multiple 'event types' can share the same 'topic name'
    // - the third level map keys are the ids of the subscriber instance
    //   note that multiple subscribers can subscribe to same 'topic name' and 'event type'
    // - the final values are the function objects used to invoke the right callback
    std::map<std::string, std::map<std::string, std::map<uintptr_t, std::shared_ptr<FuncObjBase>>>>
        m_callbacks;
 
    // used to synchronise access on the above data structure
    std::mutex m_mutex;
};
 
}  // namespace detail
 
// forward declarations
 
template <typename EventType>
class TypedEventPublisher;
 
template <typename EventType>
class TypedEventSubscriber;
 
class [[deprecated("Use EventServiceIf instead.")]] TypedEventService {
public:
    explicit TypedEventService(EventServiceIf& es) : m_pub_man(es), m_sub_man(es) {
    }
 
    template <typename EventType>
    std::unique_ptr<TypedEventPublisher<EventType>> MakePublisher() {
        return std::unique_ptr<TypedEventPublisher<EventType>>(
            new TypedEventPublisher<EventType>(m_pub_man));
    }
 
    template <typename EventType>
    std::unique_ptr<TypedEventSubscriber<EventType>> MakeSubscriber() {
        return std::unique_ptr<TypedEventSubscriber<EventType>>(
            new TypedEventSubscriber<EventType>(m_sub_man));
    }
 
private:
    detail::TypedEventPublisherManager m_pub_man;
    detail::TypedEventSubscriberManager m_sub_man;
};
 
template <typename EventType>
class [[deprecated("Use EventPublisherIf instead.")]] TypedEventPublisher {
public:
    // Use the factory method from the TypedEventService to construct publisher objects
    friend TypedEventService;
 
    ~TypedEventPublisher() {
        m_man.UnregisterPublisher<EventType>();
    }
 
    void Publish(const EventType& sample) {
        m_man.Publish(sample);
    }
 
private:
    explicit TypedEventPublisher(detail::TypedEventPublisherManager& man) : m_man(man) {
        man.RegisterPublisher<EventType>();
    }
 
    detail::TypedEventPublisherManager& m_man;
};
 
template <typename EventType>
class [[deprecated("Use EventSubscriberIf instead.")]] TypedEventSubscriber {
public:
    // Use the factory method from the TypedEventService to construct subscriber objects
    friend TypedEventService;
 
    ~TypedEventSubscriber() {
        m_man.UnregisterSubscriber<EventType>(reinterpret_cast<uintptr_t>(this));
    }
 
    void Subscribe(std::function<void(const EventType&)> cb) {
        m_man.Subscribe<EventType>(reinterpret_cast<uintptr_t>(this), cb);
    }
 
    void Unsubscribe() {
        m_man.Unsubscribe<EventType>(reinterpret_cast<uintptr_t>(this));
    }
 
private:
    explicit TypedEventSubscriber(detail::TypedEventSubscriberManager& man) : m_man(man) {
        m_man.RegisterSubscriber<EventType>(reinterpret_cast<uintptr_t>(this));
    }
 
    detail::TypedEventSubscriberManager& m_man;
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_TYPEDEVENTSERVICE_HPP