This article presents a robust event dispatching system that supports both synchronous and asynchronous event handling. The implementation extends a synchronous-only design to enable concurrent procesing while maintaining type safety and thread safety.
Core Architecture
The dispatcher uses a template-based approach where each event type gets its own handler. The base class inherits from std::enable_shared_from_this to safely share ownership when spawning worker threads.
template<typename Derived>
class EventDispatcher : public std::enable_shared_from_this<Derived> {
struct HandlerBase {
virtual ~HandlerBase() noexcept = default;
virtual bool empty() const noexcept = 0;
virtual void clear() noexcept = 0;
virtual void join() noexcept = 0;
virtual void shutdown() noexcept = 0;
};
template<typename EventType>
struct EventHandler final : HandlerBase {
using Callback = std::function<void(EventType&, std::shared_ptr<Derived>&)>;
using CallbackEntry = std::pair<bool, Callback>;
using CallbackList = std::list<CallbackEntry>;
using Subscription = typename CallbackList::iterator;
bool empty() const noexcept override {
auto markedForDeletion = [](const auto& entry) { return entry.first; };
return std::all_of(singleShotCallbacks.begin(), singleShotCallbacks.end(), markedForDeletion) &&
std::all_of(persistentCallbacks.begin(), persistentCallbacks.end(), markedForDeletion);
}
void clear() noexcept override {
if (!dispatchMutex.try_lock()) {
auto markDeleted = [](auto& entry) { entry.first = true; };
std::for_each(singleShotCallbacks.begin(), singleShotCallbacks.end(), markDeleted);
std::for_each(persistentCallbacks.begin(), persistentCallbacks.end(), markDeleted);
} else {
singleShotCallbacks.clear();
persistentCallbacks.clear();
}
dispatchMutex.unlock();
}
Subscription registerSingleShot(Callback fn) {
return singleShotCallbacks.emplace(singleShotCallbacks.end(), false, std::move(fn));
}
Subscription registerPersistent(Callback fn) {
return persistentCallbacks.emplace(persistentCallbacks.end(), false, std::move(fn));
}
void unregister(Subscription handle) noexcept {
handle->first = true;
if (dispatchMutex.try_lock()) {
auto markedForDeletion = [](const auto& entry) { return entry.first; };
singleShotCallbacks.remove_if(markedForDeletion);
persistentCallbacks.remove_if(markedForDeletion);
}
dispatchMutex.unlock();
}
void dispatch(EventType event, std::shared_ptr<Derived> instance) {
CallbackList temp;
singleShotCallbacks.swap(temp);
auto invokeCallback = [&event, &instance](auto& entry) {
if (!entry.first) entry.second(event, instance);
};
dispatchMutex.lock();
std::for_each(persistentCallbacks.rbegin(), persistentCallbacks.rend(), invokeCallback);
std::for_each(temp.rbegin(), temp.rend(), invokeCallback);
dispatchMutex.unlock();
persistentCallbacks.remove_if([](const auto& entry) { return entry.first; });
}
void workerLoop(std::shared_ptr<Derived> instance) {
while (true) {
std::unique_lock<std::mutex> lock(queueMutex);
queueCondition.wait_for(lock, std::chrono::seconds(60),
[this]() { return !pendingEvents.empty() || shutdownRequested; });
if (shutdownRequested && pendingEvents.empty()) break;
if (!pendingEvents.empty()) {
EventType event = std::move(pendingEvents.front());
pendingEvents.pop();
lock.unlock();
dispatch(std::move(event), std::move(instance));
}
}
}
void publish(EventType event, std::shared_ptr<Derived> instance, bool async) {
if (async) {
{
std::lock_guard<std::mutex> lock(queueMutex);
pendingEvents.push(std::move(event));
}
queueCondition.notify_all();
if (!workerThread.joinable()) {
workerThread = std::thread(&EventHandler<EventType>::workerLoop, this, std::move(instance));
}
} else {
dispatch(std::move(event), std::move(instance));
}
}
void join() noexcept override {
if (workerThread.joinable()) {
workerThread.join();
}
}
void shutdown() noexcept override {
{
std::lock_guard<std::mutex> lock(queueMutex);
shutdownRequested = true;
}
queueCondition.notify_all();
join();
}
private:
std::mutex dispatchMutex;
CallbackList singleShotCallbacks{};
CallbackList persistentCallbacks{};
std::thread workerThread;
std::queue<EventType> pendingEvents;
std::mutex queueMutex;
std::condition_variable queueCondition;
bool shutdownRequested = false;
};
static std::size_t allocateTypeId() noexcept {
static std::size_t counter = 0;
return counter++;
}
template<typename>
static std::size_t getTypeId() noexcept {
static std::size_t typeId = allocateTypeId();
return typeId;
}
template<typename EventType>
EventHandler<EventType>& getHandler() noexcept {
std::size_t id = getTypeId<EventType>();
if (id >= handlers.size()) {
handlers.resize(id + 1);
}
if (!handlers[id]) {
handlers[id] = std::make_unique<EventHandler<EventType>>();
}
return static_cast<EventHandler<EventType>&>(*handlers[id]);
}
protected:
template<typename EventType>
void publish(EventType event, bool async = false) {
getHandler<EventType>().publish(std::move(event), this->shared_from_this(), async);
}
public:
template<typename EventType>
using Callback = typename EventHandler<EventType>::Callback;
template<typename EventType>
struct Subscription : private EventHandler<EventType>::Subscription {
template<typename> friend class EventDispatcher;
Subscription() = default;
Subscription(const Subscription&) = default;
Subscription(Subscription&&) = default;
Subscription(typename EventHandler<EventType>::Subscription handle)
: EventHandler<EventType>::Subscription{std::move(handle)} {}
Subscription& operator=(const Subscription&) = default;
Subscription& operator=(Subscription&&) = default;
};
virtual ~EventDispatcher() noexcept {
static_assert(std::is_base_of<EventDispatcher<Derived>, Derived>::value, "CRTP requirement violated");
}
template<typename EventType>
Subscription<EventType> on(Callback<EventType> callback) {
return getHandler<EventType>().registerPersistent(std::move(callback));
}
template<typename EventType>
Subscription<EventType> once(Callback<EventType> callback) {
return getHandler<EventType>().registerSingleShot(std::move(callback));
}
template<typename EventType>
void unsubscribe(Subscription<EventType> handle) noexcept {
getHandler<EventType>().unregister(std::move(handle));
}
template<typename EventType>
void clear() noexcept {
getHandler<EventType>().clear();
}
void clearAll() noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto& h) { if (h) h->clear(); });
}
template<typename EventType>
bool hasListeners() const noexcept {
std::size_t id = getTypeId<EventType>();
return id < handlers.size() && handlers[id] && !handlers[id]->empty();
}
bool hasAnyListeners() const noexcept {
return std::any_of(handlers.begin(), handlers.end(),
[](const auto& h) { return h && !h->empty(); });
}
void waitForCompletion() const noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto& h) { if (h) h->join(); });
}
void shutdownAll() const noexcept {
std::for_each(handlers.begin(), handlers.end(),
[](auto& h) { if (h) h->shutdown(); });
}
private:
std::vector<std::unique_ptr<HandlerBase>> handlers{};
};
template<typename EventType>
using EventPtr = std::unique_ptr<EventType>;
template<typename EventType, typename... Args>
EventPtr<EventType> createEvent(Args&&... args) {
return std::make_unique<EventType>(std::forward<Args>(args)...);
}
Event Definitions and Usage
Events are simple structures that carry data. For optimal performance, especially with complex events, use std::unique_ptr to avoid unnecessary copies during dispatch.
struct TextEvent {
explicit TextEvent(std::string msg) : message(std::move(msg)) {
std::cout << "TextEvent constructed\n";
}
void display() const {
std::cout << "Message: " << message << "\n";
}
~TextEvent() {
std::cout << "TextEvent destroyed\n";
}
std::string message;
};
struct NumericEvent {
explicit NumericEvent(int val) : value(val) {
std::cout << "NumericEvent constructed\n";
}
void display() const {
std::cout << "Value: " << value << "\n";
}
~NumericEvent() {
std::cout << "NumericEvent destroyed\n";
}
int value;
};
Practical Example
class EventSource : public EventDispatcher<EventSource> {
public:
EventSource() {
std::cout << "EventSource created\n";
}
void generateEvents() {
publish(TextEvent("Immediate"), false);
publish(createEvent<TextEvent>("Async #1"), true);
publish(std::make_unique<TextEvent>("Async #2"), true);
std::this_thread::sleep_for(std::chrono::seconds(1));
publish(std::make_unique<NumericEvent>(42), true);
publish(std::make_unique<NumericEvent>(100), true);
}
~EventSource() {
std::cout << "EventSource destroyed\n";
}
};
int main() {
auto dispatcher = std::make_shared<EventSource>();
dispatcher->on<TextEvent>([](TextEvent& ev, std::shared_ptr<EventSource>&) {
ev.display();
});
dispatcher->on<EventPtr<TextEvent>>([](EventPtr<TextEvent>& ev, std::shared_ptr<EventSource>&) {
ev->display();
});
dispatcher->on<EventPtr<NumericEvent>>([](EventPtr<NumericEvent>& ev, std::shared_ptr<EventSource>&) {
ev->display();
});
dispatcher->generateEvents();
dispatcher->waitForCompletion();
return 0;
}
Threading and Performance Considerations
The worker thread uses a 60-second timeout to avoid resource waste when idle. If no events arrive within this period, the thread exits. For scenarios with infrequent events, adjust the timeout or implement a persistent thread model.
Critical design decisions:
- Ownership Model: Using
std::shared_ptrensures the dispatcher remains alive while background threads process events - Thread Safety: Separate mutexes protect the callback lists and event queue, preventing deadlocks during nested dispatches
- Event Lifetime: Passing events by value to
publish()then moving them prevents dangling references - Conditional Wait: The condition variable checks both queue state and shutdown flag to ensure clean termination
The async flag is determined at publication time, allowing per-event decisions. For example, log events with large payloads can be processed asynchronously while lightweight metrics are handled synchronously.
When using unique pointers for events, the event is constructed exactly once and moved through the pipeline, eliminating redundant temporary objects.