kcenon::common::simple_event_bus Class Reference

Simple synchronous event bus for testing when monitoring is disabled. More...

#include <event_bus.h>

Collaboration diagram for kcenon::common::simple_event_bus:

Classes
struct	subscription_info

Public Types
using	error_callback_t
	Type for error callback function.

Public Member Functions
template<concepts::EventType EventType>
void	publish (const EventType &evt, event_priority=event_priority::normal)
	Publish an event to all subscribed handlers.
void	publish (event &&evt, event_priority=event_priority::normal)
template<concepts::EventType EventType, concepts::EventHandler< EventType > HandlerFunc>
uint64_t	subscribe (HandlerFunc &&func)
	Subscribe to events of a specific type.
uint64_t	subscribe (std::function< void(const event &)> &&func)
template<concepts::EventType EventType, concepts::EventHandler< EventType > HandlerFunc, concepts::EventFilter< EventType > FilterFunc>
uint64_t	subscribe_filtered (HandlerFunc &&func, FilterFunc &&filter)
	Subscribe to events with a filter function.
uint64_t	subscribe_filtered (std::function< void(const event &)> &&func, std::function< bool(const event &)> &&filter)
void	unsubscribe (uint64_t id)
void	start ()
void	stop ()
bool	is_running () const
void	set_error_callback (error_callback_t callback)
	Set the error callback for handler exceptions and type mismatches.
void	clear_error_callback ()
	Clear the error callback.

Static Public Member Functions
static simple_event_bus &	instance ()
	Get the singleton instance (thread-safe via C++11 magic statics)

Private Attributes
std::mutex	mutex_
std::unordered_multimap< size_t, subscription_info >	handlers_
std::atomic< uint64_t >	next_id_ {1}
std::atomic< bool >	running_ {true}
error_callback_t	error_callback_

Detailed Description

Simple synchronous event bus for testing when monitoring is disabled.

Thread-safety: This class is thread-safe using a mutex to protect subscription management and event dispatch.

Note: This implementation avoids RTTI by using template-based type IDs.

simple_event_bus bus;
 
// Subscribe to an event type
auto id = bus.subscribe<events::module_started_event>(
    [](const events::module_started_event& e) {
        std::cout << "Module started: " << e.module_name << "\n";
    });
 
// Publish an event
bus.publish(events::module_started_event{"network_system"});
 
// Unsubscribe when done
bus.unsubscribe(id);

Definition at line 150 of file event_bus.h.

Member Typedef Documentation

error_callback_t

using kcenon::common::simple_event_bus::error_callback_t

Initial value:

 std::function<void(const std::string& error_message,
                                                     size_t event_type_id,
                                                     uint64_t handler_id)>

Type for error callback function.

This callback is invoked when an exception occurs in an event handler or when a type mismatch is detected. The callback receives:

• error_message: Description of what went wrong
• event_type_id: The type ID of the event being processed
• handler_id: The subscription ID of the failing handler

Definition at line 161 of file event_bus.h.

Member Function Documentation

clear_error_callback()

void kcenon::common::simple_event_bus::clear_error_callback ( )

inline

Clear the error callback.

Definition at line 415 of file event_bus.h.

                                    {
            std::lock_guard<std::mutex> lock(mutex_);
            error_callback_ = nullptr;
        }

References error_callback_, and mutex_.

instance()

static simple_event_bus & kcenon::common::simple_event_bus::instance ( )

inlinestatic

Get the singleton instance (thread-safe via C++11 magic statics)

Returns

Reference to the singleton instance

Definition at line 424 of file event_bus.h.

                                            {
            static simple_event_bus instance;
            return instance;
        }

References instance().

Referenced by kcenon::common::get_event_bus(), and instance().

Here is the call graph for this function:

Here is the caller graph for this function:

is_running()

bool kcenon::common::simple_event_bus::is_running ( ) const

inline

Definition at line 384 of file event_bus.h.

{ return running_; }

References running_.

publish() [1/2]

template<concepts::EventType EventType>

void kcenon::common::simple_event_bus::publish ( const EventType & evt, event_priority = event_priority::normal )

inline

Publish an event to all subscribed handlers.

Template Parameters

EventType

The event type (must satisfy concepts::EventType)

Parameters

evt	The event to publish
priority	Event priority (default: normal)

Note

Uses C++20 concepts for compile-time type validation.

Definition at line 174 of file event_bus.h.

                                                                                  {
            // Snapshot handlers and error callback under the lock, then release
            // before invoking handlers. This prevents reentrancy deadlock when
            // a handler calls subscribe() or unsubscribe() on this bus.
            std::vector<subscription_info> snapshot;
            error_callback_t error_cb;
            {
                std::lock_guard<std::mutex> lock(mutex_);
                auto type_id = event_type_id<EventType>::id();
                auto range = handlers_.equal_range(type_id);
                snapshot.reserve(std::distance(range.first, range.second));
                for (auto it = range.first; it != range.second; ++it) {
                    snapshot.push_back(it->second);
                }
                error_cb = error_callback_;
            }
 
            // Get type identifier without RTTI
            auto type_id = event_type_id<EventType>::id();
 
            // Invoke handlers outside the lock — safe for reentrancy
            for (auto& entry : snapshot) {
                try {
                    // Type safety check: verify expected type matches
                    if (entry.expected_type_id != type_id) {
                        if (error_cb) {
                            error_cb("Type ID mismatch detected in event handler",
                                     type_id, entry.id);
                        }
                        continue;
                    }
 
                    if (entry.handler) {
                        entry.handler(static_cast<const void*>(&evt));
                    }
                } catch (const std::exception& e) {
                    if (error_cb) {
                        std::string error_msg = "Exception in event handler: ";
                        error_msg += e.what();
                        error_cb(error_msg, type_id, entry.id);
                    }
                } catch (...) {
                    if (error_cb) {
                        error_cb("Unknown exception in event handler",
                                 type_id, entry.id);
                    }
                }
            }
        }

References error_callback_, handlers_, kcenon::common::detail::event_type_id< T >::id(), and mutex_.

Here is the call graph for this function:

publish() [2/2]

void kcenon::common::simple_event_bus::publish ( event && evt, event_priority = event_priority::normal )

inline

Definition at line 225 of file event_bus.h.

                                                                         {
            // Snapshot handlers and error callback under the lock
            std::vector<subscription_info> snapshot;
            error_callback_t error_cb;
            {
                std::lock_guard<std::mutex> lock(mutex_);
                auto type_id = event_type_id<event>::id();
                auto range = handlers_.equal_range(type_id);
                snapshot.reserve(std::distance(range.first, range.second));
                for (auto it = range.first; it != range.second; ++it) {
                    snapshot.push_back(it->second);
                }
                error_cb = error_callback_;
            }
 
            auto type_id = event_type_id<event>::id();
 
            // Invoke handlers outside the lock — safe for reentrancy
            for (auto& entry : snapshot) {
                try {
                    if (entry.expected_type_id != type_id) {
                        if (error_cb) {
                            error_cb("Type ID mismatch detected in event handler",
                                     type_id, entry.id);
                        }
                        continue;
                    }
 
                    if (entry.handler) {
                        entry.handler(static_cast<const void*>(&evt));
                    }
                } catch (const std::exception& e) {
                    if (error_cb) {
                        std::string error_msg = "Exception in event handler: ";
                        error_msg += e.what();
                        error_cb(error_msg, type_id, entry.id);
                    }
                } catch (...) {
                    if (error_cb) {
                        error_cb("Unknown exception in event handler",
                                 type_id, entry.id);
                    }
                }
            }
        }

References error_callback_, handlers_, kcenon::common::detail::event_type_id< T >::id(), and mutex_.

Here is the call graph for this function:

set_error_callback()

void kcenon::common::simple_event_bus::set_error_callback ( error_callback_t callback )

inline

Set the error callback for handler exceptions and type mismatches.

This callback will be invoked whenever:

• An exception is thrown by an event handler
• A type ID mismatch is detected (should never happen in correct usage)

The error callback is useful for logging and debugging handler issues without stopping the processing of other handlers.

Parameters

callback

The callback function to invoke on errors

Example usage:

auto& bus = simple_event_bus::instance();
bus.set_error_callback([](const std::string& msg, size_t type_id, uint64_t handler_id) {
    std::cerr << "Event bus error [type=" << type_id
              << ", handler=" << handler_id << "]: " << msg << std::endl;
});

Definition at line 407 of file event_bus.h.

                                                           {
            std::lock_guard<std::mutex> lock(mutex_);
            error_callback_ = std::move(callback);
        }

References error_callback_, and mutex_.

start()

void kcenon::common::simple_event_bus::start ( )

inline

Definition at line 382 of file event_bus.h.

{ running_ = true; }

References running_.

stop()

void kcenon::common::simple_event_bus::stop ( )

inline

Definition at line 383 of file event_bus.h.

{ running_ = false; }

References running_.

subscribe() [1/2]

template<concepts::EventType EventType, concepts::EventHandler< EventType > HandlerFunc>

uint64_t kcenon::common::simple_event_bus::subscribe ( HandlerFunc && func )

inline

Subscribe to events of a specific type.

Template Parameters

EventType	The event type (must satisfy concepts::EventType)
HandlerFunc	The handler function type (must satisfy concepts::EventHandler)

Parameters

func	The handler function to call when the event is published

Returns

Subscription ID for later unsubscription

Note

Uses C++20 concepts for compile-time type validation.

Definition at line 281 of file event_bus.h.

                                               {
            std::lock_guard<std::mutex> lock(mutex_);
 
            auto id = next_id_++;
            auto type_id = event_type_id<EventType>::id();
 
            subscription_info info;
            info.id = id;
            info.expected_type_id = type_id;  // Store expected type ID for validation
            // Wrap the handler in a type-erased function
            info.handler = [f = std::function<void(const EventType&)>(std::forward<HandlerFunc>(func))]
                          (const void* evt_ptr) {
                f(*static_cast<const EventType*>(evt_ptr));
            };
 
            handlers_.emplace(type_id, std::move(info));
 
            return id;
        }

References handlers_, kcenon::common::detail::event_type_id< T >::id(), mutex_, and next_id_.

Referenced by subscribe_filtered(), and subscribe_filtered().

Here is the call graph for this function:

Here is the caller graph for this function:

subscribe() [2/2]

uint64_t kcenon::common::simple_event_bus::subscribe ( std::function< void(const event &)> && func )

inline

Definition at line 302 of file event_bus.h.

                                                                   {
            std::lock_guard<std::mutex> lock(mutex_);
 
            auto id = next_id_++;
            auto type_id = event_type_id<event>::id();
 
            subscription_info info;
            info.id = id;
            info.expected_type_id = type_id;  // Store expected type ID for validation
            info.handler = [f = std::move(func)](const void* evt_ptr) {
                f(*static_cast<const event*>(evt_ptr));
            };
 
            handlers_.emplace(type_id, std::move(info));
 
            return id;
        }

References handlers_, kcenon::common::detail::event_type_id< T >::id(), mutex_, and next_id_.

Here is the call graph for this function:

subscribe_filtered() [1/2]

template<concepts::EventType EventType, concepts::EventHandler< EventType > HandlerFunc, concepts::EventFilter< EventType > FilterFunc>

uint64_t kcenon::common::simple_event_bus::subscribe_filtered ( HandlerFunc && func, FilterFunc && filter )

inline

Subscribe to events with a filter function.

Template Parameters

EventType	The event type (must satisfy concepts::EventType)
HandlerFunc	The handler function type (must satisfy concepts::EventHandler)
FilterFunc	The filter function type (must satisfy concepts::EventFilter)

Parameters

func	The handler function to call when the event passes the filter
filter	The filter function to determine if the handler should be called

Returns

Subscription ID for later unsubscription

Note

The filter function is called before the handler. If it returns false, the handler is not invoked. This allows for efficient event filtering without creating multiple event types.

Uses C++20 concepts for compile-time type validation.

Definition at line 337 of file event_bus.h.

                                                                             {
            // Create a wrapped handler that includes the filtering logic
            auto wrapped_handler = [f = std::forward<HandlerFunc>(func),
                                   flt = std::forward<FilterFunc>(filter)]
                                  (const EventType& evt) {
                // Only call the handler if the filter passes
                if (flt(evt)) {
                    f(evt);
                }
            };
 
            // Subscribe with the wrapped handler
            return subscribe<EventType>(std::move(wrapped_handler));
        }

References subscribe().

Here is the call graph for this function:

subscribe_filtered() [2/2]

uint64_t kcenon::common::simple_event_bus::subscribe_filtered ( std::function< void(const event &)> && func, std::function< bool(const event &)> && filter )

inline

Definition at line 353 of file event_bus.h.

                                                      {
            // Create a wrapped handler that includes the filtering logic
            auto wrapped_handler = [f = std::move(func),
                                   flt = std::move(filter)]
                                  (const event& evt) {
                // Only call the handler if the filter passes
                if (flt(evt)) {
                    f(evt);
                }
            };
 
            return subscribe(std::move(wrapped_handler));
        }

References subscribe().

Here is the call graph for this function:

unsubscribe()

void kcenon::common::simple_event_bus::unsubscribe ( uint64_t id )

inline

Definition at line 369 of file event_bus.h.

                                      {
            std::lock_guard<std::mutex> lock(mutex_);
 
            // Find and remove the handler with this ID
            for (auto it = handlers_.begin(); it != handlers_.end(); ) {
                if (it->second.id == id) {
                    it = handlers_.erase(it);
                } else {
                    ++it;
                }
            }
        }

References handlers_, and mutex_.

Member Data Documentation

error_callback_

error_callback_t kcenon::common::simple_event_bus::error_callback_

private

Definition at line 440 of file event_bus.h.

Referenced by clear_error_callback(), publish(), publish(), and set_error_callback().

handlers_

std::unordered_multimap<size_t, subscription_info> kcenon::common::simple_event_bus::handlers_

private

Definition at line 437 of file event_bus.h.

Referenced by publish(), publish(), subscribe(), subscribe(), and unsubscribe().

mutex_

std::mutex kcenon::common::simple_event_bus::mutex_

mutableprivate

Definition at line 436 of file event_bus.h.

Referenced by clear_error_callback(), publish(), publish(), set_error_callback(), subscribe(), subscribe(), and unsubscribe().

next_id_

std::atomic<uint64_t> kcenon::common::simple_event_bus::next_id_ {1}

private

Definition at line 438 of file event_bus.h.

{1};

Referenced by subscribe(), and subscribe().

running_

std::atomic<bool> kcenon::common::simple_event_bus::running_ {true}

private

Definition at line 439 of file event_bus.h.

{true};

Referenced by is_running(), start(), and stop().

---

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

• include/kcenon/common/patterns/event_bus.h