conflated_event_channel.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/concurrent/conflated_event_channel.hpp
/// @brief @copybrief concurrent::ConflatedEventChannel
 
#include <atomic>
#include <vector>
 
#include <userver/concurrent/async_event_channel.hpp>
#include <userver/engine/single_consumer_event.hpp>
#include <userver/engine/task/task_with_result.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace concurrent {
 
/// @ingroup userver_concurrency
///
/// @brief A non-blocking version of 'AsyncEventChannel'
///
/// The difference is that 'SendEvent' returns immediately, without waiting for
/// subscribers to finish. If 'SendEvent' is called multiple times while
/// subscribers are handling the previous event, new events will be conflated
/// (all events except for the last one will be ignored). This class can be used
/// instead of 'AsyncEventChannel' when we've got a "heavy" subscriber and we
/// don't want to slow down the pipeline.
class ConflatedEventChannel : private AsyncEventChannel<> {
public:
    explicit ConflatedEventChannel(std::string name, OnRemoveCallback on_listener_removal = {});
    ~ConflatedEventChannel() override;
 
    /// For convenient forwarding of events from other channels
    template <typename... Args>
    void AddChannel(concurrent::AsyncEventSource<Args...>& channel);
 
    using AsyncEventChannel<>::AddListener;
 
    /// Subscribes to updates using a member function. Also immediately invokes
    /// the function with the current config snapshot.
    ///
    /// Further updates are delivered after @ref utils::ResourceScopeStorage::AfterConstruction, including those updates
    /// that arrived before it. Unsubscribe runs in @ref utils::ResourceScopeStorage::BeforeDestruction.
    ///
    /// @param scopes storage that owns the subscription lifetime. In a component constructor pass `context.Scopes()`
    /// or @ref components::GetResourceScopes.
    template <typename Class>
    void UpdateAndListen(utils::ResourceScopeStorage& scopes, Class* obj, std::string_view name, void (Class::*func)());
 
    /// @overload
    /// @deprecated Use the overload that takes @ref utils::ResourceScopeStorage.
    ///
    /// Store the returned scope as a member and call `Unsubscribe` explicitly.
    template <typename Class>
    concurrent::AsyncEventSubscriberScope UpdateAndListen(Class* obj, std::string_view name, void (Class::*func)());
 
    void SendEvent();
 
private:
    template <typename... Args>
    void OnChannelEvent(Args...);
 
    std::atomic<bool> stop_flag_;
    engine::TaskWithResult<void> task_;
    std::vector<concurrent::AsyncEventSubscriberScope> subscriptions_;
    engine::SingleConsumerEvent event_;
};
 
template <typename... Args>
void ConflatedEventChannel::AddChannel(concurrent::AsyncEventSource<Args...>& channel) {
    subscriptions_.push_back(channel.AddListener(this, Name(), &ConflatedEventChannel::OnChannelEvent<Args...>));
}
 
template <typename Class>
void ConflatedEventChannel::UpdateAndListen(
    utils::ResourceScopeStorage& scopes,
    Class* obj,
    std::string_view name,
    void (Class::*func)()
) {
    DoUpdateAndListenScoped(scopes, obj, name, func, [obj, func] { (obj->*func)(); });
}
 
template <typename Class>
concurrent::AsyncEventSubscriberScope ConflatedEventChannel::UpdateAndListen(
    Class* obj,
    std::string_view name,
    void (Class::*func)()
) {
    return DoUpdateAndListen(obj, name, func, [&] { (obj->*func)(); });
}
 
template <typename... Args>
void ConflatedEventChannel::OnChannelEvent(Args...) {
    SendEvent();
}
 
}  // namespace concurrent
 
USERVER_NAMESPACE_END