conflated_event_channel.hpp

#pragma once
 
#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 {
 
class ConflatedEventChannel : private AsyncEventChannel<> {
 public:
  explicit ConflatedEventChannel(std::string name);
  ~ConflatedEventChannel() override;
 
  template <typename... Args>
  void AddChannel(concurrent::AsyncEventSource<Args...>& channel);
 
  using AsyncEventChannel<>::AddListener;
 
  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>
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