async_event_source.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/concurrent/async_event_source.hpp
/// @brief @copybrief concurrent::AsyncEventSource
 
#include <cstddef>
#include <functional>
#include <string_view>
#include <typeinfo>
#include <utility>
 
USERVER_NAMESPACE_BEGIN
 
namespace concurrent {
 
class FunctionId;
enum class UnsubscribingKind;
 
template <typename... Args>
class AsyncEventSource;
 
namespace impl {
 
class AsyncEventSourceBase {
public:
    virtual void RemoveListener(FunctionId, UnsubscribingKind) noexcept = 0;
 
protected:
    // disallow deletion via pointer to base
    ~AsyncEventSourceBase();
};
 
}  // namespace impl
 
/// Uniquely identifies a subscription (usually the callback method owner) for
/// AsyncEventSource
class FunctionId final {
public:
    constexpr FunctionId() = default;
 
    template <typename Class>
    explicit FunctionId(Class* obj) : FunctionId(obj, typeid(Class)) {}
 
    explicit operator bool() const;
 
    bool operator==(const FunctionId& other) const;
 
    struct Hash final {
        std::size_t operator()(FunctionId id) const noexcept;
    };
 
private:
    FunctionId(void* ptr, const std::type_info& type);
 
    void* ptr_{nullptr};
    const std::type_info* type_{nullptr};
};
 
enum class UnsubscribingKind { kManual, kAutomatic };
 
/// @brief Manages the subscription to events from an AsyncEventSource
///
/// Removes the associated listener automatically on destruction.
///
/// The Scope is usually placed as a member in the subscribing object.
/// `Unsubscribe` should be called manually in the objects destructor, before
/// anything that the callback needs is destroyed.
class [[nodiscard]] AsyncEventSubscriberScope final {
public:
    AsyncEventSubscriberScope() = default;
 
    template <typename... Args>
    AsyncEventSubscriberScope(AsyncEventSource<Args...>& channel, FunctionId id)
        : AsyncEventSubscriberScope(static_cast<impl::AsyncEventSourceBase&>(channel), id) {}
 
    AsyncEventSubscriberScope(AsyncEventSubscriberScope&& scope) noexcept;
 
    AsyncEventSubscriberScope& operator=(AsyncEventSubscriberScope&& other) noexcept;
 
    ~AsyncEventSubscriberScope();
 
    /// Unsubscribes manually. The subscription should be cancelled before
    /// anything that the callback needs is destroyed.
    void Unsubscribe() noexcept;
 
private:
    AsyncEventSubscriberScope(impl::AsyncEventSourceBase& channel, FunctionId id);
 
    impl::AsyncEventSourceBase* channel_{nullptr};
    FunctionId id_;
};
 
/// @ingroup userver_concurrency
///
/// @brief The read-only side of an event channel. Events are delivered to
/// listeners in a strict FIFO order, i.e. only after the event was processed
/// a new event may appear for processing, same listener is never called
/// concurrently.
template <typename... Args>
class AsyncEventSource : public impl::AsyncEventSourceBase {
public:
    using Function = std::function<void(Args... args)>;
 
    virtual ~AsyncEventSource() = default;
 
    /// @brief Subscribes to updates from this event source
    ///
    /// The listener won't be called immediately. To process the current value and
    /// then listen to updates, use `UpdateAndListen` of specific event channels.
    ///
    /// @warning Listeners should not be added or removed while processing the
    /// event inside another listener.
    ///
    /// Example usage:
    /// @snippet concurrent/async_event_channel_test.cpp  AddListener sample
    ///
    /// @param obj the subscriber, which is the owner of the listener method, and
    /// is also used as the unique identifier of the subscription for this
    /// AsyncEventSource
    /// @param name the name of the subscriber, for diagnostic purposes
    /// @param func the listener method, usually called `On<DataName>Update`, e.g.
    /// `OnConfigUpdate` or `OnCacheUpdate`
    /// @returns a AsyncEventSubscriberScope controlling the subscription, which
    /// should be stored as a member in the subscriber; `Unsubscribe` should be
    /// called explicitly
    template <class Class>
    AsyncEventSubscriberScope AddListener(Class* obj, std::string_view name, void (Class::*func)(Args...)) {
        return AddListener(FunctionId(obj), name, [obj, func](Args... args) { (obj->*func)(args...); });
    }
 
    /// @brief A type-erased version of AddListener
    ///
    /// @param id the unique identifier of the subscription
    /// @param name the name of the subscriber, for diagnostic purposes
    /// @param func the callback used for event notifications
    AsyncEventSubscriberScope AddListener(FunctionId id, std::string_view name, Function&& func) {
        return DoAddListener(id, name, std::move(func));
    }
 
    /// Used by AsyncEventSubscriberScope to cancel an event subscription
    void RemoveListener(FunctionId, UnsubscribingKind) noexcept override = 0;
 
protected:
    virtual AsyncEventSubscriberScope DoAddListener(FunctionId id, std::string_view name, Function&& func) = 0;
};
 
}  // namespace concurrent
 
USERVER_NAMESPACE_END