repositorySubscriberIf.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_REPOSITORYSUBSCRIBERIF_HPP
#define RTCTK_COMPONENTFRAMEWORK_REPOSITORYSUBSCRIBERIF_HPP
 
#include <rtctk/componentFramework/repositoryIf.hpp>
 
namespace rtctk::componentFramework {
 
class RTCTK_API RepositorySubscriberIf : virtual public RepositoryIf {
public:
    // TODO: flesh out the details of this error type
    // this likely will map to follwoing dds reader callbacks:
    // - on sample_rejected
    // - on_sample_lost
    // - on_request_deadline_missed
    // - on request_incompatible_qos
    class RTCTK_API SubscriptionError : public RtctkException {
    public:
        using RtctkException::RtctkException;
    };
 
    class RTCTK_API SequenceIdMissmatch : public RtctkException {
    public:
        SequenceIdMissmatch(const DataPointPath& path, RtcUInt64 expected, RtcUInt64 actual);
    };
 
    template <typename T>
    using ValueCallbackType = std::function<void(const DataPointPath&, T&, const MetaData&)>;
 
    using NotifyCallbackType = std::function<void(const DataPointPath&, const MetaData&)>;
    using ErrorCallbackType = std::function<void(std::exception_ptr)>;
 
    using SubscriptionId = size_t;
 
    class RTCTK_API Subscription final {
    public:
        using UnsubscribeMethod = std::function<void()>;
 
        explicit Subscription(UnsubscribeMethod);
        ~Subscription();
 
        Subscription(const Subscription&) = delete;
        Subscription& operator=(const Subscription&) = delete;
 
        Subscription(Subscription&&) noexcept;
        Subscription& operator=(Subscription&&) noexcept;
 
        void Unsubscribe();
 
    private:
        UnsubscribeMethod m_unsubscribe;
    };
 
    class RTCTK_API BatchRequest : public RepositoryIf::BatchRequest {
    public:
        SubscriptionId Subscribe(const DataPointPath&, NotifyCallbackType, ErrorCallbackType) const;
 
        SubscriptionId Subscribe(CallbackType, CallbackType, ErrorCallbackType) const;
 
        void Unsubscribe(SubscriptionId) const;
    };
 
    RepositorySubscriberIf() = default;
    virtual ~RepositorySubscriberIf() = default;
 
    template <
        typename T = void,
        typename F,
        std::enable_if_t<
            std::is_convertible_v<F, NotifyCallbackType> or
                (not std::is_same_v<T, void> and
                 std::is_convertible_v<
                     F,
                     ValueCallbackType<std::conditional_t<std::is_same_v<T, void>, std::byte, T>>>),
            bool> = true>
    Subscription Subscribe(const DataPointPath& path,
                           const F& cb_value,
                           const ErrorCallbackType& cb_error) const;
 
    Subscription Subscribe(const CallbackType& cb_created,
                           const CallbackType& cb_deleted,
                           const ErrorCallbackType& cb_error) const;
 
    // The following is part of the legacy API:
 
public:
    class SubscribeRequest {
    public:
        class Parameters {
        public:
            using CallbackType = std::function<void(const DataPointPath&)>;
 
            Parameters() = default;
 
            explicit Parameters(const DataPointPath& path,
                                bool value_requested,
                                void* buffer,
                                const std::type_info& type,
                                const CallbackType& removed_callback,
                                const CallbackType& newvalue_callback)
                : m_path(path)
                , m_value_requested(value_requested)
                , m_buffer(buffer)
                , m_type(&type)
                , m_removed_callback(removed_callback)
                , m_newvalue_callback(newvalue_callback) {
            }
 
            explicit Parameters(const DataPointPath& path,
                                void* buffer,
                                const std::type_info& type,
                                const CallbackType& removed_callback,
                                const CallbackType& newvalue_callback)
                : Parameters(path, true, buffer, type, removed_callback, newvalue_callback) {
            }
 
            inline const DataPointPath& GetPath() const {
                return m_path;
            };
            inline const bool GetValueRequested() const {
                return m_value_requested;
            }
            inline void* GetBuffer() const {
                return m_buffer;
            };
            inline const std::type_info& GetType() const {
                return *m_type;
            };
            inline const CallbackType& GetRemovedCallback() const {
                return m_removed_callback;
            };
            inline const CallbackType& GetNewValueCallback() const {
                return m_newvalue_callback;
            };
 
        private:
            DataPointPath m_path;
 
            bool m_value_requested;
 
            void* m_buffer;
 
            const std::type_info* m_type;
 
            CallbackType m_removed_callback;
            CallbackType m_newvalue_callback;
        };
 
        template <typename F>
        void AddRemovedHandler(const DataPointPath& path, F handler);
 
        template <typename T, typename F>
        void AddNewValueHandler(const DataPointPath& path, T& buffer, F handler);
 
        template <typename F>
        void AddValueChangedHandler(const DataPointPath& path, F handler);
 
        inline const std::vector<Parameters>& GetParams() const {
            return m_params;
        };
 
    private:
        std::vector<Parameters> m_params;
    };
 
    class UnsubscribeRequest {
    public:
        class Parameters {
        public:
            Parameters() = default;
 
            explicit Parameters(const DataPointPath& path,
                                const bool unsubscribe_removed,
                                const bool unsubscribe_newvalue)
                : m_path(path)
                , m_unsubscribe_removed(unsubscribe_removed)
                , m_unsubscribe_newvalue(unsubscribe_newvalue) {
            }
 
            inline const DataPointPath& GetPath() const {
                return m_path;
            };
            inline bool GetUnsubscribeRemoved() const {
                return m_unsubscribe_removed;
            };
            inline bool GetUnsubscribeNewValue() const {
                return m_unsubscribe_newvalue;
            };
 
        private:
            DataPointPath m_path;
 
            bool m_unsubscribe_removed;
 
            bool m_unsubscribe_newvalue;
        };
 
        void AddRemovedHandler(const DataPointPath& path);
 
        void AddNewValueHandler(const DataPointPath& path);
 
        inline const std::vector<Parameters>& GetParams() const {
            return m_params;
        };
 
    private:
        std::vector<Parameters> m_params;
    };
 
    virtual RepositoryIf::Response SendSubscribeRequest(const SubscribeRequest& request) const;
 
    virtual RepositoryIf::Response SendUnsubscribeRequest(const UnsubscribeRequest& request) const;
 
    using NotifyHandler = std::function<void(const DataPointPath& path)>;
 
    template <typename T>
    using ValueHandler = std::function<void(const DataPointPath& path, T& value)>;
 
    template <
        typename T,
        typename F,
        std::enable_if_t<not std::is_convertible_v<T, RepositorySubscriberIf::NotifyCallbackType>,
                         bool> = true>
    void Subscribe(const DataPointPath& path, T& buffer, F handler) const;
 
    template <typename F>
    void Subscribe(const DataPointPath& path, F handler) const {
        SubscribeRequest subscribe;
        if constexpr (std::is_convertible_v<F, NotifyHandler>) {
            subscribe.AddValueChangedHandler(path, handler);
        } else {
            // later we intend to support more callback types
            static_assert(std::is_convertible_v<F, NotifyHandler>, "Feature not supported");
        }
        SendSubscribeRequest(subscribe).Wait();
    }
 
    void Unsubscribe(const DataPointPath& path) const;
 
private:
    // This is a workaround to support Python bindings. We need to associate certain extra objects
    // and buffers under the hood that have to live as long as an instance of this class does.
    friend class PyRepositorySubscriberIf;
    std::any m_extra_objects;
 
    mutable std::mutex m_subscriptions_mutex;
 
    // List of subscriptions made from the legacy API.
    mutable std::map<DataPointPath, std::vector<SubscriptionId>> m_new_value_subscriptions;
    mutable std::map<DataPointPath, std::vector<SubscriptionId>> m_removed_subscriptions;
};
 
}  // namespace rtctk::componentFramework
 
#include <rtctk/componentFramework/repositorySubscriberIf.ipp>
 
#endif  // RTCTK_COMPONENTFRAMEWORK_REPOSITORYSUBSCRIBERIF_HPP