repositorySubscriberIf.ipp

Go to the documentation of this file.

 
// Note this is a template implementation file and should not be included directly.
// The typical header protection macro is not added to avoid it showing up in Doxygen API
// documentation.
#pragma once
 
#include <exception>
#include <functional>
#include <memory>
#include <type_traits>
 
namespace rtctk::componentFramework {
 
namespace detail {
 
struct ValueSubscribeRequest {
    RepositorySubscriberIf::SubscriptionId id;
    DataPointPath path;
    RepositorySubscriberIf::NotifyCallbackType cb_value;
    RepositorySubscriberIf::ErrorCallbackType cb_error;
};
 
struct CreateDeleteSubscribeRequest {
    RepositorySubscriberIf::SubscriptionId id;
    RepositorySubscriberIf::CallbackType cb_created;
    RepositorySubscriberIf::CallbackType cb_deleted;
    RepositorySubscriberIf::ErrorCallbackType cb_error;
};
 
struct UnsubscribeRequest {
    RepositorySubscriberIf::SubscriptionId id;
};
 
}  // namespace detail
 
// NOTE: The enable_if can be removed once the legacy Subscribe template methods using
// SubscribeRequest have been removed.
template <
    typename T,
    typename F,
    std::enable_if_t<
        std::is_convertible_v<F, RepositorySubscriberIf::NotifyCallbackType> or
            (not std::is_same_v<T, void> and
             std::is_convertible_v<F,
                                   RepositorySubscriberIf::ValueCallbackType<
                                       std::conditional_t<std::is_same_v<T, void>, std::byte, T>>>),
        bool>>
RepositorySubscriberIf::Subscription
RepositorySubscriberIf::Subscribe(const DataPointPath& path,
                                  const F& cb_value,
                                  const ErrorCallbackType& cb_error) const {
    NotifyCallbackType cb_val;
    if constexpr (std::is_convertible_v<F, NotifyCallbackType>) {
        cb_val = cb_value;
    } else {
        static_assert(std::is_convertible_v<F, ValueCallbackType<T>>,
                      "The cb_value callback must have a signature compatible with"
                      " void(const DataPointPath&, T&, const MetaData&).");
        cb_val = [buf = std::make_shared<T>(), cb_value, cb_error, this](const DataPointPath& path2,
                                                                         const MetaData& md) {
            MetaData mdr;
            BatchRequest req;
            req.ReadDataPoint(path2, *buf, mdr);
            SendRequest(req).Wait();
            if (md.Contains("sequence_id")) {
                // If backend supports sequence_id, then check for sequence id missmatch.
                RtcUInt64 sequence_id_expected = md["sequence_id"].Cast<RtcUInt64>();
                RtcUInt64 sequence_id_actual = mdr["sequence_id"].Cast<RtcUInt64>();
                if (sequence_id_expected == sequence_id_actual) {
                    cb_value(path2, *buf, md);
                } else {
                    auto eptr = std::make_exception_ptr(CII_MAKE_EXCEPTION(
                        SequenceIdMissmatch, path2, sequence_id_expected, sequence_id_actual));
                    cb_error(eptr);
                }
            } else {
                // If backend does not support sequence_id, then just invoke cb_value without check.
                cb_value(path2, *buf, md);
            }
        };
    }
    BatchRequest req;
    auto id = req.Subscribe(path, cb_val, cb_error);
    SendRequest(req).Wait();
    Subscription subscription([this, id]() {
        BatchRequest req;
        req.Unsubscribe(id);
        SendRequest(req).Wait();
    });
    return subscription;
}
 
// The following are legacy methods:
 
template <typename F>
void RepositorySubscriberIf::SubscribeRequest::AddRemovedHandler(const DataPointPath& path,
                                                                 F handler) {
    static_assert(
        std::is_convertible_v<F, std::function<void(const DataPointPath&)>>,
        "The handler callback must have a signature compatible with void(const DataPointPath&).");
    m_params.push_back(Parameters(path, nullptr, typeid(void), handler, nullptr));
}
 
template <typename T, typename F>
void RepositorySubscriberIf::SubscribeRequest::AddNewValueHandler(const DataPointPath& path,
                                                                  T& buffer,
                                                                  F handler) {
    static_assert(std::is_convertible_v<F, ValueHandler<T>>,
                  "The handler callback must have a signature compatible with void(const"
                  " DataPointPath&, T&).");
    // We use a lambda here to have type erasure and allow storing callbacks for different types in
    // the m_newvalue_callback.
    auto callback = [handler, &buffer](const DataPointPath& path) -> void {
        handler(path, buffer);
    };
    m_params.push_back(Parameters(path, true, &buffer, typeid(buffer), nullptr, callback));
}
 
template <typename F>
void RepositorySubscriberIf::SubscribeRequest::AddValueChangedHandler(const DataPointPath& path,
                                                                      F handler) {
    static_assert(std::is_convertible_v<F, NotifyHandler>,
                  "The handler callback must have a signature compatible with void(const"
                  " DataPointPath&).");
    // We use a lambda here to have type erasure and allow storing callbacks for different types in
    // the m_newvalue_callback.
    auto callback = [handler](const DataPointPath& path) -> void { handler(path); };
    m_params.push_back(Parameters(path, false, nullptr, typeid(void), nullptr, callback));
}
 
template <typename T,
          typename F,
          std::enable_if_t<not std::is_convertible_v<T, RepositorySubscriberIf::NotifyCallbackType>,
                           bool>>
void RepositorySubscriberIf::Subscribe(const DataPointPath& path, T& buffer, F handler) const {
    SubscribeRequest subscribe;
    subscribe.AddNewValueHandler(path, buffer, handler);
    SendSubscribeRequest(subscribe).Wait();
}
 
}  // namespace rtctk::componentFramework