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
 
template <typename T, typename F>
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;
}
 
}  // namespace rtctk::componentFramework