concurrent::AsyncEventChannel< Args > Class Template Reference

#include <userver/concurrent/async_event_channel.hpp>

Detailed Description

template<typename... Args>
class concurrent::AsyncEventChannel< Args >

AsyncEventChannel is an in-process pub-sub with strict FIFO serialization, i.e. only after the event was processed a new event may appear for processing, same listener is never called concurrently.

Example usage:

enum class WeatherKind { kSunny, kRainy };
 
class WeatherStorage final {
 public:
  explicit WeatherStorage(WeatherKind value)
      : value_(value), channel_("weather") {}
 
  WeatherKind Get() const { return value_.load(); }
 
  concurrent::AsyncEventSource<WeatherKind>& GetSource() { return channel_; }
 
  template <typename Class>
  concurrent::AsyncEventSubscriberScope UpdateAndListen(
      Class* obj, std::string_view name, void (Class::*func)(WeatherKind)) {
    return channel_.DoUpdateAndListen(obj, name, func,
                                      [&] { (obj->*func)(Get()); });
  }
 
  void Set(WeatherKind value) {
    value_.store(value);
    channel_.SendEvent(value);
  }
 
 private:
  std::atomic<WeatherKind> value_;
  concurrent::AsyncEventChannel<WeatherKind> channel_;
};
 
enum class CoatKind { kJacket, kRaincoat };
 
class CoatStorage final {
 public:
  explicit CoatStorage(WeatherStorage& weather_storage) {
    weather_subscriber_ = weather_storage.UpdateAndListen(
        this, "coats", &CoatStorage::OnWeatherUpdate);
  }
 
  ~CoatStorage() { weather_subscriber_.Unsubscribe(); }
 
  CoatKind Get() const { return value_.load(); }
 
 private:
  void OnWeatherUpdate(WeatherKind weather) {
    value_.store(ComputeCoat(weather));
  }
 
  static CoatKind ComputeCoat(WeatherKind weather);
 
  std::atomic<CoatKind> value_{};
  concurrent::AsyncEventSubscriberScope weather_subscriber_;
};
 
UTEST(AsyncEventChannel, UpdateAndListenSample) {
  WeatherStorage weather_storage(WeatherKind::kSunny);
  CoatStorage coat_storage(weather_storage);
  EXPECT_EQ(coat_storage.Get(), CoatKind::kJacket);
  weather_storage.Set(WeatherKind::kRainy);
  EXPECT_EQ(coat_storage.Get(), CoatKind::kRaincoat);
}

Definition at line 74 of file async_event_channel.hpp.

Inheritance diagram for concurrent::AsyncEventChannel< Args >:

Collaboration diagram for concurrent::AsyncEventChannel< Args >:

Public Types
using	Function = typename AsyncEventSource<Args...>::Function
using	OnRemoveCallback = std::function<void(Function&)>
Public Types inherited from concurrent::AsyncEventSource< Args... >
using	Function

Public Member Functions
	AsyncEventChannel (std::string name)
	The primary constructor.
	AsyncEventChannel (std::string name, OnRemoveCallback on_listener_removal)
	The constructor with AsyncEventSubscriberScope usage checking.
template<typename UpdaterFunc >
AsyncEventSubscriberScope	DoUpdateAndListen (FunctionId id, std::string_view name, Function &&func, UpdaterFunc &&updater)
	For use in UpdateAndListen of specific event channels.
template<typename Class , typename UpdaterFunc >
AsyncEventSubscriberScope	DoUpdateAndListen (Class *obj, std::string_view name, void(Class::*func)(Args...), UpdaterFunc &&updater)
	This is an overloaded member function, provided for convenience. It differs from the above function only in what argument(s) it accepts.
void	SendEvent (Args... args) const
const std::string &	Name () const noexcept
Public Member Functions inherited from concurrent::AsyncEventSource< Args... >
AsyncEventSubscriberScope	AddListener (Class *obj, std::string_view name, void(Class::*func)(Args...))
	Subscribes to updates from this event source.
AsyncEventSubscriberScope	AddListener (FunctionId id, std::string_view name, Function &&func)
	A type-erased version of AddListener.

Additional Inherited Members
Protected Member Functions inherited from concurrent::AsyncEventSource< Args... >
virtual AsyncEventSubscriberScope	DoAddListener (FunctionId id, std::string_view name, Function &&func)=0

Member Typedef Documentation

Function

template<typename... Args>

using concurrent::AsyncEventChannel< Args >::Function = typename AsyncEventSource<Args...>::Function

Definition at line 76 of file async_event_channel.hpp.

OnRemoveCallback

template<typename... Args>

using concurrent::AsyncEventChannel< Args >::OnRemoveCallback = std::function<void(Function&)>

Definition at line 77 of file async_event_channel.hpp.

Constructor & Destructor Documentation

AsyncEventChannel() [1/2]

template<typename... Args>

concurrent::AsyncEventChannel< Args >::AsyncEventChannel ( std::string name )

inlineexplicit

The primary constructor.

Parameters

name	used for diagnostic purposes and is also accessible with Name

Definition at line 81 of file async_event_channel.hpp.

AsyncEventChannel() [2/2]

template<typename... Args>

concurrent::AsyncEventChannel< Args >::AsyncEventChannel ( std::string name, OnRemoveCallback on_listener_removal )

inline

The constructor with AsyncEventSubscriberScope usage checking.

The constructor with a callback that is called on listener removal. The callback takes a reference to ‘Function’ as input. This is useful for checking the lifetime of data captured by the listener update function.

Note

Works only in debug mode.

Warning

Data captured by on_listener_removal function must be valid until the AsyncEventChannel object is completely destroyed.

Example usage:

  auto on_remove = [](std::function<void(int)> func) { func(1); };
 
  concurrent::AsyncEventChannel<int> channel("channel", on_remove);
 
  int value = 0;
  {
    concurrent::AsyncEventSubscriberScope sub =
        channel.AddListener(concurrent::FunctionId(&sub), "sub",
                            [&value](int new_value) { value = new_value; });
  }
 
  if constexpr (concurrent::impl::kCheckSubscriptionUB) {
    EXPECT_EQ(value, 1);
  } else {
    EXPECT_EQ(value, 0);
  }

Parameters

name	used for diagnostic purposes and is also accessible with Name
on_listener_removal	the callback used for check

See also

impl::CheckDataUsedByCallbackHasNotBeenDestroyedBeforeUnsubscribing

Definition at line 102 of file async_event_channel.hpp.

Member Function Documentation

DoUpdateAndListen() [1/2]

template<typename... Args>

template<typename Class , typename UpdaterFunc >

AsyncEventSubscriberScope concurrent::AsyncEventChannel< Args >::DoUpdateAndListen ( Class * obj, std::string_view name, void(Class::*)(Args...) func, UpdaterFunc && updater )

inline

This is an overloaded member function, provided for convenience. It differs from the above function only in what argument(s) it accepts.

Definition at line 124 of file async_event_channel.hpp.

DoUpdateAndListen() [2/2]

template<typename... Args>

template<typename UpdaterFunc >

AsyncEventSubscriberScope concurrent::AsyncEventChannel< Args >::DoUpdateAndListen ( FunctionId id, std::string_view name, Function && func, UpdaterFunc && updater )

inline

For use in UpdateAndListen of specific event channels.

Atomically calls updater, which should invoke func with the previously sent event, and subscribes to new events as if using AddListener.

See also

AsyncEventSource::AddListener

Definition at line 113 of file async_event_channel.hpp.

Name()

template<typename... Args>

const std::string & concurrent::AsyncEventChannel< Args >::Name ( ) const

inlinenoexcept

Returns

the name of this event channel

Definition at line 159 of file async_event_channel.hpp.

SendEvent()

template<typename... Args>

void concurrent::AsyncEventChannel< Args >::SendEvent ( Args... args ) const

inline

Send the next event and wait until all the listeners process it.

Strict FIFO serialization is guaranteed, i.e. only after this event is processed a new event may be delivered for the subscribers, same listener/subscriber is never called concurrently.

Definition at line 138 of file async_event_channel.hpp.

---

The documentation for this class was generated from the following file:

• userver/concurrent/async_event_channel.hpp