Event Bus for publish-subscribe pattern. More...

#include <event_bus.h>

Collaboration diagram for kcenon::thread::event_bus:

Classes
class	subscription
	Subscription handle for managing subscriptions. More...

Public Types
using	handler_func = std::function<void(const std::any&)>
	Handler function type.

Public Member Functions
	event_bus (std::shared_ptr< thread_pool > pool=nullptr)
	Constructor.

template<typename Event >
void	publish (const Event &event)
	Publish an event asynchronously.

template<typename Event >
void	publish_sync (const Event &event)
	Publish an event synchronously.

template<typename Event >
subscription	subscribe (std::function< void(const Event &)> handler)
	Subscribe to events of a specific type.

template<typename Event >
void	clear_subscriptions ()
	Clear all subscriptions for a specific event type.

void	clear_all_subscriptions ()
	Clear all subscriptions.

template<typename Event >
std::size_t	subscriber_count () const
	Get the number of subscribers for an event type.

Static Public Member Functions
static event_bus &	instance ()
	Get singleton instance.

Private Member Functions
void	unsubscribe (std::type_index type, std::size_t id)
	Unsubscribe a specific handler.

Private Attributes
std::mutex	mutex_

std::unordered_map< std::type_index, std::unordered_map< std::size_t, handler_func > >	handlers_

std::shared_ptr< thread_pool >	thread_pool_

std::size_t	next_handler_id_ {1}

Detailed Description

Event Bus for publish-subscribe pattern.

Provides asynchronous event distribution across system components

Definition at line 55 of file event_bus.h.

Member Typedef Documentation

◆ handler_func

using kcenon::thread::event_bus::handler_func = std::function<void(const std::any&)>

Handler function type.

Definition at line 60 of file event_bus.h.

Constructor & Destructor Documentation

◆ event_bus()

kcenon::thread::event_bus::event_bus ( std::shared_ptr< thread_pool > pool = nullptr )

inlineexplicit

Constructor.

Parameters

thread_pool Optional thread pool for async processing

Definition at line 123 of file event_bus.h.

        : thread_pool_(pool) {
        if (!thread_pool_) {
            // Create default thread pool with 2 threads for event processing
            thread_pool_ = thread_pool_builder("event_bus")
                .with_workers(2)
                .build_and_start();
        }
    }

References kcenon::thread::thread_pool_builder::build_and_start(), thread_pool_, and kcenon::thread::thread_pool_builder::with_workers().

Here is the call graph for this function:

Member Function Documentation

◆ clear_all_subscriptions()

void kcenon::thread::event_bus::clear_all_subscriptions ( )

inline

Clear all subscriptions.

Definition at line 229 of file event_bus.h.

                                   {
        std::lock_guard<std::mutex> lock(mutex_);
        handlers_.clear();
    }

References handlers_, and mutex_.

◆ clear_subscriptions()

template<typename Event >

void kcenon::thread::event_bus::clear_subscriptions ( )

inline

Clear all subscriptions for a specific event type.

Template Parameters

Event Event type

Definition at line 220 of file event_bus.h.

                               {
        auto type = std::type_index(typeid(Event));
        std::lock_guard<std::mutex> lock(mutex_);
        handlers_.erase(type);
    }

References handlers_, and mutex_.

◆ instance()

static event_bus & kcenon::thread::event_bus::instance ( )

inlinestatic

Get singleton instance.

Returns: Event bus instance

Definition at line 254 of file event_bus.h.

                                 {
        static event_bus instance;
        return instance;
    }

References instance().

Referenced by instance().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ publish()

template<typename Event >

void kcenon::thread::event_bus::publish ( const Event & event )

inline

Publish an event asynchronously.

Template Parameters

Event Event type

Parameters

event Event to publish

Definition at line 139 of file event_bus.h.

                                     {
        auto type = std::type_index(typeid(Event));
 
        // Get handlers snapshot to avoid holding lock during callbacks
        std::vector<handler_func> handlers_copy;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            auto it = handlers_.find(type);
            if (it != handlers_.end()) {
                for (const auto& [id, handler] : it->second) {
                    handlers_copy.push_back(handler);
                }
            }
        }
 
        // Process handlers asynchronously
        if (!handlers_copy.empty() && thread_pool_) {
            thread_pool_->submit([handlers_copy, event]() {
                for (const auto& handler : handlers_copy) {
                    try {
                        handler(std::any(event));
                    } catch (...) {
                        // Log error but continue with other handlers
                    }
                }
            });
        }
    }

References handlers_, mutex_, and thread_pool_.

◆ publish_sync()

template<typename Event >

void kcenon::thread::event_bus::publish_sync ( const Event & event )

inline

Publish an event synchronously.

Template Parameters

Event Event type

Parameters

event Event to publish

Definition at line 174 of file event_bus.h.

                                          {
        auto type = std::type_index(typeid(Event));
 
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = handlers_.find(type);
        if (it != handlers_.end()) {
            for (const auto& [id, handler] : it->second) {
                try {
                    handler(std::any(event));
                } catch (...) {
                    // Log error but continue with other handlers
                }
            }
        }
    }

References handlers_, and mutex_.

◆ subscribe()

template<typename Event >

subscription kcenon::thread::event_bus::subscribe ( std::function< void(const Event &)> handler )

inlinenodiscard

Subscribe to events of a specific type.

Template Parameters

Event Event type to subscribe to

Parameters

handler Handler function

Returns: Subscription handle

Definition at line 197 of file event_bus.h.

                                                                                  {
        auto type = std::type_index(typeid(Event));
        auto wrapped_handler = [handler](const std::any& any_event) {
            try {
                const auto& event = std::any_cast<const Event&>(any_event);
                handler(event);
            } catch (const std::bad_any_cast&) {
                // Type mismatch - should not happen
            }
        };
 
        std::lock_guard<std::mutex> lock(mutex_);
        auto id = next_handler_id_++;
        handlers_[type][id] = wrapped_handler;
 
        return subscription(this, type, id);
    }

References handlers_, mutex_, and next_handler_id_.

◆ subscriber_count()

template<typename Event >

std::size_t kcenon::thread::event_bus::subscriber_count ( ) const

inline

Get the number of subscribers for an event type.

Template Parameters

Event Event type

Returns: Number of subscribers

Definition at line 240 of file event_bus.h.

                                       {
        auto type = std::type_index(typeid(Event));
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = handlers_.find(type);
        if (it != handlers_.end()) {
            return it->second.size();
        }
        return 0;
    }

References handlers_, and mutex_.

◆ unsubscribe()

void kcenon::thread::event_bus::unsubscribe	(	std::type_index	type,
		std::size_t	id )

inlineprivate

Unsubscribe a specific handler.

Definition at line 263 of file event_bus.h.

                                                       {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = handlers_.find(type);
        if (it != handlers_.end()) {
            it->second.erase(id);
            if (it->second.empty()) {
                handlers_.erase(it);
            }
        }
    }

References handlers_, and mutex_.

Referenced by kcenon::thread::event_bus::subscription::unsubscribe().

Here is the caller graph for this function:

Member Data Documentation

◆ handlers_

std::unordered_map<std::type_index, std::unordered_map<std::size_t, handler_func> > kcenon::thread::event_bus::handlers_

private

Definition at line 275 of file event_bus.h.

Referenced by clear_all_subscriptions(), clear_subscriptions(), publish(), publish_sync(), subscribe(), subscriber_count(), and unsubscribe().

◆ mutex_

std::mutex kcenon::thread::event_bus::mutex_

mutableprivate

Definition at line 274 of file event_bus.h.

Referenced by clear_all_subscriptions(), clear_subscriptions(), publish(), publish_sync(), subscribe(), subscriber_count(), and unsubscribe().

◆ next_handler_id_

std::size_t kcenon::thread::event_bus::next_handler_id_ {1}

private

Definition at line 277 of file event_bus.h.

277{1};

Referenced by subscribe().

◆ thread_pool_

std::shared_ptr<thread_pool> kcenon::thread::event_bus::thread_pool_

private

Definition at line 276 of file event_bus.h.

Referenced by event_bus(), and publish().

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

include/kcenon/thread/core/event_bus.h

Classes

Public Types

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ handler_func

Constructor & Destructor Documentation

◆ event_bus()

Member Function Documentation

◆ clear_all_subscriptions()

◆ clear_subscriptions()

◆ instance()

◆ publish()

◆ publish_sync()

◆ subscribe()

◆ subscriber_count()

◆ unsubscribe()

Member Data Documentation

◆ handlers_

◆ mutex_

◆ next_handler_id_

◆ thread_pool_