Thread-safe event bus implementation. More...

#include <event_bus.h>

Inheritance diagram for kcenon::monitoring::event_bus:

Collaboration diagram for kcenon::monitoring::event_bus:

Classes
struct	stats

Public Types
using	config = event_bus_config

Public Member Functions
	event_bus (const config &cfg=config())

	~event_bus ()

common::VoidResult	start () override
	Start the event bus.

common::VoidResult	stop () override
	Stop the event bus.

bool	is_active () const override
	Check if event bus is active.

size_t	get_pending_event_count () const override
	Get pending event count.

common::VoidResult	process_pending_events () override
	Process all pending events synchronously.

common::VoidResult	unsubscribe_event (const subscription_token &token) override
	Unsubscribe from events using subscription token.

stats	get_stats () const

Public Member Functions inherited from kcenon::monitoring::interface_event_bus
virtual	~interface_event_bus ()=default

template<concepts::EventType E>
common::VoidResult	publish_event (const E &event)
	Publish an event to all subscribers.

template<concepts::EventType E>
common::Result< subscription_token >	subscribe_event (std::function< void(const E &)> handler, event_priority priority=event_priority::normal)
	Subscribe to events of a specific type.

template<concepts::EventType E, concepts::EventHandler< E > H>
common::Result< subscription_token >	subscribe_event (H &&handler, event_priority priority=event_priority::normal)
	Subscribe to events with a callable handler.

template<concepts::EventType E>
common::VoidResult	clear_subscriptions ()
	Clear all subscriptions for a specific event type.

template<concepts::EventType E>
size_t	get_subscriber_count () const
	Get the number of subscribers for an event type.

Static Public Member Functions
static std::shared_ptr< interface_event_bus >	instance ()

Protected Member Functions
common::VoidResult	publish_event_impl (std::type_index event_type, std::any event) override

common::Result< subscription_token >	subscribe_event_impl (std::type_index event_type, std::function< void(const std::any &)> handler, uint64_t handler_id, event_priority priority) override

common::VoidResult	clear_subscriptions_impl (std::type_index event_type) override

size_t	get_subscriber_count_impl (std::type_index event_type) const override

Protected Member Functions inherited from kcenon::monitoring::interface_event_bus

Private Member Functions
void	process_events_worker ()

void	process_all_pending ()

void	dispatch_event (const event_envelope &envelope)

Private Attributes
config	config_

std::mutex	bus_mutex_

std::mutex	queue_mutex_

std::mutex	handlers_mutex_

std::priority_queue< event_envelope >	event_queue_

std::unordered_map< std::type_index, std::vector< event_handler_wrapper > >	handlers_

std::vector< std::thread >	workers_

std::condition_variable	queue_cv_

std::atomic< bool >	is_running_

std::atomic< bool >	stop_requested_

std::atomic< uint64_t >	total_events_published_

std::atomic< uint64_t >	total_events_processed_

std::atomic< uint64_t >	total_events_dropped_

Detailed Description

Thread-safe event bus implementation.

Provides high-performance event distribution with:

Priority-based event processing
Back-pressure management
Async and sync processing modes
Type-safe publish/subscribe

Thread Safety:

All public methods are thread-safe. The event bus uses multiple mutexes for fine-grained locking:

bus_mutex_ for start/stop operations
queue_mutex_ for event queue access
handlers_mutex_ for subscriber management

Definition at line 146 of file event_bus.h.

Member Typedef Documentation

◆ config

using kcenon::monitoring::event_bus::config = event_bus_config

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 148 of file event_bus.h.

Constructor & Destructor Documentation

◆ event_bus()

kcenon::monitoring::event_bus::event_bus ( const config & cfg = config() )

inlineexplicit

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 150 of file event_bus.h.

        : config_(cfg), is_running_(false), stop_requested_(false),
          total_events_published_(0), total_events_processed_(0),
          total_events_dropped_(0) {
 
        if (config_.auto_start) {
            start();
        }
    }

References kcenon::monitoring::event_bus_config::auto_start, and config_.

◆ ~event_bus()

kcenon::monitoring::event_bus::~event_bus ( )

inline

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 160 of file event_bus.h.

                 {
        stop();
    }

References stop().

Here is the call graph for this function:

Member Function Documentation

◆ clear_subscriptions_impl()

common::VoidResult kcenon::monitoring::event_bus::clear_subscriptions_impl ( std::type_index event_type )

inlineoverrideprotectedvirtual

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 347 of file event_bus.h.

                                                                                 {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        handlers_.erase(event_type);
        return common::ok();
    }

References handlers_, and handlers_mutex_.

◆ dispatch_event()

void kcenon::monitoring::event_bus::dispatch_event ( const event_envelope & envelope )

inlineprivate

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 413 of file event_bus.h.

                                                        {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
 
        auto it = handlers_.find(envelope.type);
        if (it != handlers_.end()) {
            for (const auto& wrapper : it->second) {
                try {
                    wrapper.handler(envelope.payload);
                } catch (const std::exception& e) {
                    (void)e;  // Suppress unused variable warning
                    // Log error but continue processing
                    // In production, this would be logged properly
                }
            }
        }
    }

References handlers_, handlers_mutex_, kcenon::monitoring::event_envelope::payload, and kcenon::monitoring::event_envelope::type.

Referenced by process_all_pending(), and process_events_worker().

Here is the caller graph for this function:

◆ get_pending_event_count()

size_t kcenon::monitoring::event_bus::get_pending_event_count ( ) const

inlineoverridevirtual

Get pending event count.

Returns: Number of events waiting to be processed

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 228 of file event_bus.h.

                                                    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        return event_queue_.size();
    }

References event_queue_, and queue_mutex_.

◆ get_stats()

stats kcenon::monitoring::event_bus::get_stats ( ) const

inline

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 272 of file event_bus.h.

                            {
        // Use scoped_lock to acquire both mutexes atomically with consistent ordering
        // This prevents deadlock by acquiring locks in a deterministic order
        // Lock ordering: queue_mutex before handlers_mutex (alphabetical for consistency)
        std::scoped_lock lock(queue_mutex_, handlers_mutex_);
 
        size_t queue_size = event_queue_.size();
        bool back_pressure_active = queue_size >= config_.back_pressure_threshold;
 
        size_t total_subscribers = 0;
        for (const auto& [type, handlers] : handlers_) {
            total_subscribers += handlers.size();
        }
 
        return {
            total_events_published_.load(),
            total_events_processed_.load(),
            total_events_dropped_.load(),
            queue_size,
            total_subscribers,
            back_pressure_active
        };
    }

References kcenon::monitoring::event_bus_config::back_pressure_threshold, config_, event_queue_, handlers_, handlers_mutex_, queue_mutex_, total_events_dropped_, total_events_processed_, and total_events_published_.

◆ get_subscriber_count_impl()

size_t kcenon::monitoring::event_bus::get_subscriber_count_impl ( std::type_index event_type ) const

inlineoverrideprotectedvirtual

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 354 of file event_bus.h.

                                                                              {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        auto it = handlers_.find(event_type);
        return it != handlers_.end() ? it->second.size() : 0;
    }

References handlers_, and handlers_mutex_.

◆ instance()

static std::shared_ptr< interface_event_bus > kcenon::monitoring::event_bus::instance ( )

inlinestatic

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 164 of file event_bus.h.

                                                         {
        static std::shared_ptr<event_bus> singleton = [] {
            auto bus = std::make_shared<event_bus>();
            (void)bus->start();
            return bus;
        }();
        return singleton;
    }

◆ is_active()

bool kcenon::monitoring::event_bus::is_active ( ) const

inlineoverridevirtual

Check if event bus is active.

Returns: True if active, false otherwise

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 222 of file event_bus.h.

                                    {
        std::lock_guard<std::mutex> lock(bus_mutex_);
        return is_running_;
    }

References bus_mutex_, and is_running_.

◆ process_all_pending()

void kcenon::monitoring::event_bus::process_all_pending ( )

inlineprivate

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 395 of file event_bus.h.

                               {
        std::vector<event_envelope> pending;
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            pending.reserve(event_queue_.size());
            while (!event_queue_.empty()) {
                pending.push_back(event_queue_.top());
                event_queue_.pop();
            }
        }
 
        for (auto& envelope : pending) {
            dispatch_event(envelope);
            total_events_processed_.fetch_add(1);
        }
    }

References dispatch_event(), event_queue_, kcenon::monitoring::pending, queue_mutex_, and total_events_processed_.

Referenced by process_pending_events(), and stop().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ process_events_worker()

void kcenon::monitoring::event_bus::process_events_worker ( )

inlineprivate

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 362 of file event_bus.h.

                                 {
        while (!stop_requested_) {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for events or stop signal
            queue_cv_.wait_for(lock, config_.processing_interval,
                              [this] { return !event_queue_.empty() || stop_requested_; });
 
            if (stop_requested_ && event_queue_.empty()) {
                break;
            }
 
            // Process a batch of events
            std::vector<event_envelope> batch;
            size_t batch_size = std::min(size_t(10), event_queue_.size());
 
            for (size_t i = 0; i < batch_size && !event_queue_.empty(); ++i) {
                // Copy element then pop to avoid const_cast UB
                batch.push_back(event_queue_.top());
                event_queue_.pop();
            }
 
            lock.unlock();
 
            // Process the batch
            for (auto& envelope : batch) {
                dispatch_event(envelope);
                total_events_processed_.fetch_add(1);
            }
        }
    }

References config_, dispatch_event(), event_queue_, kcenon::monitoring::event_bus_config::processing_interval, queue_cv_, queue_mutex_, stop_requested_, and total_events_processed_.

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ process_pending_events()

common::VoidResult kcenon::monitoring::event_bus::process_pending_events ( )

inlineoverridevirtual

Process all pending events synchronously.

Returns: Result indicating success or failure

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 234 of file event_bus.h.

                                                       {
        process_all_pending();
        return common::ok();
    }

References process_all_pending().

Here is the call graph for this function:

◆ publish_event_impl()

common::VoidResult kcenon::monitoring::event_bus::publish_event_impl	(	std::type_index	event_type,
		std::any	event )

inlineoverrideprotectedvirtual

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 298 of file event_bus.h.

                                                         {
        bool should_sleep = false;
 
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
 
            if (config_.enable_back_pressure) {
                const auto current_size = event_queue_.size();
                if (current_size >= config_.max_queue_size) {
                    total_events_dropped_.fetch_add(1);
                    return common::VoidResult::err(error_info(monitoring_error_code::resource_exhausted, "Event queue is full").to_common_error());
                }
                should_sleep = current_size >= config_.back_pressure_threshold;
            }
 
            event_queue_.emplace(event_type, std::move(event), event_priority::normal);
            total_events_published_.fetch_add(1);
        }
 
        if (should_sleep) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
 
        queue_cv_.notify_one();
        return common::ok();
    }

References kcenon::monitoring::event_bus_config::back_pressure_threshold, config_, kcenon::monitoring::event_bus_config::enable_back_pressure, event_queue_, kcenon::monitoring::event_bus_config::max_queue_size, kcenon::monitoring::normal, queue_cv_, queue_mutex_, kcenon::monitoring::resource_exhausted, total_events_dropped_, and total_events_published_.

◆ start()

common::VoidResult kcenon::monitoring::event_bus::start ( )

inlineoverridevirtual

Start the event bus.

Returns: Result indicating success or failure

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 174 of file event_bus.h.

                                      {
        std::lock_guard<std::mutex> lock(bus_mutex_);
 
        if (is_running_) {
            return common::VoidResult::err(error_info(monitoring_error_code::already_started, "Event bus is already running").to_common_error());
        }
 
        stop_requested_ = false;
        is_running_ = true;
 
        // Start worker threads
        for (size_t i = 0; i < config_.worker_thread_count; ++i) {
            workers_.emplace_back(&event_bus::process_events_worker, this);
        }
 
        return common::ok();
    }

References kcenon::monitoring::already_started, bus_mutex_, config_, is_running_, process_events_worker(), stop_requested_, kcenon::monitoring::event_bus_config::worker_thread_count, and workers_.

Here is the call graph for this function:

◆ stop()

common::VoidResult kcenon::monitoring::event_bus::stop ( )

inlineoverridevirtual

Stop the event bus.

Returns: Result indicating success or failure

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 193 of file event_bus.h.

                                     {
        {
            std::lock_guard<std::mutex> lock(bus_mutex_);
            if (!is_running_) {
                return common::ok();
            }
 
            stop_requested_ = true;
            is_running_ = false;
        }
 
        // Wake up all worker threads
        queue_cv_.notify_all();
 
        // Wait for all workers to finish
        for (auto& worker : workers_) {
            if (worker.joinable()) {
                worker.join();
            }
        }
        workers_.clear();
 
        // Process any remaining events
        process_all_pending();
 
        return common::ok();
    }

References bus_mutex_, is_running_, process_all_pending(), queue_cv_, stop_requested_, and workers_.

Referenced by ~event_bus().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ subscribe_event_impl()

common::Result< subscription_token > kcenon::monitoring::event_bus::subscribe_event_impl	(	std::type_index	event_type,
		std::function< void(const std::any &)>	handler,
		uint64_t	handler_id,
		event_priority	priority )

inlineoverrideprotectedvirtual

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 327 of file event_bus.h.

                                          {
 
        std::lock_guard<std::mutex> lock(handlers_mutex_);
 
        handlers_[event_type].emplace_back(std::move(handler), priority, handler_id);
 
        // Sort handlers by priority
        std::sort(handlers_[event_type].begin(), handlers_[event_type].end(),
                 [](const event_handler_wrapper& a, const event_handler_wrapper& b) {
                     return static_cast<int>(a.priority) > static_cast<int>(b.priority);
                 });
 
        return subscription_token(event_type, handler_id);
    }

References handlers_, handlers_mutex_, and kcenon::monitoring::event_handler_wrapper::priority.

◆ unsubscribe_event()

common::VoidResult kcenon::monitoring::event_bus::unsubscribe_event ( const subscription_token & token )

inlineoverridevirtual

Unsubscribe from events using subscription token.

Parameters

token The subscription token

Returns: Result indicating success or failure

Implements kcenon::monitoring::interface_event_bus.

Examples: /home/runner/work/monitoring_system/monitoring_system/include/kcenon/monitoring/core/event_bus.h.

Definition at line 240 of file event_bus.h.

                                                                                 {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
 
        auto it = handlers_.find(token.get_event_type());
        if (it != handlers_.end()) {
            auto& handler_list = it->second;
            handler_list.erase(
                std::remove_if(handler_list.begin(), handler_list.end(),
                              [&token](const event_handler_wrapper& wrapper) {
                                  return wrapper.id == token.get_handler_id();
                              }),
                handler_list.end()
            );
 
            if (handler_list.empty()) {
                handlers_.erase(it);
            }
        }
 
        return common::ok();
    }