Event dispatcher with exception handling and error recovery. More...

#include <safe_event_dispatcher.h>

Collaboration diagram for kcenon::monitoring::safe_event_dispatcher:

Classes
struct	dead_letter_entry

Public Types
using	error_callback = std::function<void(const handler_error_info&)>

Public Member Functions
	safe_event_dispatcher (std::shared_ptr< interface_event_bus > bus=nullptr)

template<typename EventType , typename Handler >
uint64_t	subscribe_safe (Handler &&handler, event_priority priority=event_priority::normal)
	Subscribe to events with automatic exception handling.

template<typename EventType >
bool	publish_safe (const EventType &event, event_priority priority=event_priority::normal)
	Publish event with error recovery.

void	set_error_callback (error_callback callback)
	Set callback for handler errors.

std::vector< handler_error_info >	get_handler_errors () const
	Get recent handler errors.

uint64_t	get_total_exceptions () const
	Get total exception count.

void	clear_errors ()
	Clear error history.

void	set_circuit_breaker_threshold (size_t threshold)
	Set circuit breaker threshold.

size_t	get_dead_letter_queue_size () const
	Get dead letter queue size.

template<typename RecoveryHandler >
void	process_dead_letters (RecoveryHandler &&handler)
	Process dead letter queue with recovery handler.

std::shared_ptr< interface_event_bus >	get_bus () const
	Get underlying event bus.

Private Member Functions
void	handle_exception (uint64_t handler_id, std::type_index event_type, const std::string &error_msg)

void	add_to_dead_letter_queue (std::type_index event_type, const std::string &error_msg)

Private Attributes
std::shared_ptr< interface_event_bus >	bus_

std::atomic< uint64_t >	next_id_ {1}

std::atomic< uint64_t >	total_exceptions_ {0}

std::mutex	error_mutex_

std::vector< handler_error_info >	recent_errors_

std::unordered_map< uint64_t, size_t >	failed_handler_counts_

error_callback	error_callback_

size_t	circuit_breaker_threshold_

std::mutex	dlq_mutex_

std::vector< dead_letter_entry >	dead_letter_queue_

Detailed Description

Event dispatcher with exception handling and error recovery.

Wraps event_bus to provide:

Exception isolation (one handler's failure doesn't affect others)
Error logging and metrics
Circuit breaker for repeatedly failing handlers
Dead letter queue for failed events

Production Safety

auto dispatcher = std::make_shared<safe_event_dispatcher>();
 
// Subscribe with automatic exception handling
dispatcher->subscribe<my_event>([](const my_event& evt) {
    // This might throw, but won't crash the bus
    process_event(evt);
});
 
// Publish events safely
dispatcher->publish(my_event{});
 
// Monitor failed handlers
auto errors = dispatcher->get_handler_errors();
for (const auto& error : errors) {
    log_error("Handler {} failed: {}", error.handler_id, error.error_message);
}

Definition at line 63 of file safe_event_dispatcher.h.

Member Typedef Documentation

◆ error_callback

using kcenon::monitoring::safe_event_dispatcher::error_callback = std::function<void(const handler_error_info&)>

Definition at line 65 of file safe_event_dispatcher.h.

Constructor & Destructor Documentation

◆ safe_event_dispatcher()

kcenon::monitoring::safe_event_dispatcher::safe_event_dispatcher ( std::shared_ptr< interface_event_bus > bus = nullptr )

inlineexplicit

Definition at line 67 of file safe_event_dispatcher.h.

        : bus_(bus ? bus : event_bus::instance())
        , total_exceptions_(0)
        , circuit_breaker_threshold_(10)
    {
    }

Member Function Documentation

◆ add_to_dead_letter_queue()

void kcenon::monitoring::safe_event_dispatcher::add_to_dead_letter_queue	(	std::type_index	event_type,
		const std::string &	error_msg )

inlineprivate

Definition at line 262 of file safe_event_dispatcher.h.

                                                              {
        std::lock_guard<std::mutex> lock(dlq_mutex_);
 
        dead_letter_entry entry{
            event_type,
            error_msg,
            std::chrono::steady_clock::now()
        };
 
        dead_letter_queue_.push_back(entry);
 
        // Limit dead letter queue size
        if (dead_letter_queue_.size() > 1000) {
            dead_letter_queue_.erase(dead_letter_queue_.begin());
        }
    }

References dead_letter_queue_, and dlq_mutex_.

Referenced by publish_safe().

Here is the caller graph for this function:

◆ clear_errors()

void kcenon::monitoring::safe_event_dispatcher::clear_errors ( )

inline

Clear error history.

Definition at line 166 of file safe_event_dispatcher.h.

                        {
        std::lock_guard<std::mutex> lock(error_mutex_);
        recent_errors_.clear();
        failed_handler_counts_.clear();
        total_exceptions_ = 0;
    }

References error_mutex_, failed_handler_counts_, recent_errors_, and total_exceptions_.

◆ get_bus()

std::shared_ptr< interface_event_bus > kcenon::monitoring::safe_event_dispatcher::get_bus ( ) const

inline

Get underlying event bus.

Definition at line 210 of file safe_event_dispatcher.h.

                                                       {
        return bus_;
    }

References bus_.

◆ get_dead_letter_queue_size()

size_t kcenon::monitoring::safe_event_dispatcher::get_dead_letter_queue_size ( ) const

inline

Get dead letter queue size.

Definition at line 184 of file safe_event_dispatcher.h.

                                              {
        std::lock_guard<std::mutex> lock(dlq_mutex_);
        return dead_letter_queue_.size();
    }

References dead_letter_queue_, and dlq_mutex_.

◆ get_handler_errors()

std::vector< handler_error_info > kcenon::monitoring::safe_event_dispatcher::get_handler_errors ( ) const

inline

Get recent handler errors.

Definition at line 151 of file safe_event_dispatcher.h.

                                                             {
        std::lock_guard<std::mutex> lock(error_mutex_);
        return recent_errors_;
    }

References error_mutex_, and recent_errors_.

◆ get_total_exceptions()

uint64_t kcenon::monitoring::safe_event_dispatcher::get_total_exceptions ( ) const

inline

Get total exception count.

Definition at line 159 of file safe_event_dispatcher.h.

                                          {
        return total_exceptions_.load(std::memory_order_acquire);
    }

References total_exceptions_.

◆ handle_exception()

void kcenon::monitoring::safe_event_dispatcher::handle_exception	(	uint64_t	handler_id,
		std::type_index	event_type,
		const std::string &	error_msg )

inlineprivate

Definition at line 221 of file safe_event_dispatcher.h.

                                                     {
        total_exceptions_.fetch_add(1, std::memory_order_relaxed);
 
        std::lock_guard<std::mutex> lock(error_mutex_);
 
        // Record error
        handler_error_info info{
            handler_id,
            error_msg,
            event_type,
            std::chrono::steady_clock::now()
        };
 
        recent_errors_.push_back(info);
 
        // Keep only last 100 errors
        if (recent_errors_.size() > 100) {
            recent_errors_.erase(recent_errors_.begin());
        }
 
        // Track failure count
        auto& count = failed_handler_counts_[handler_id];
        count++;
 
        // Circuit breaker check
        if (count >= circuit_breaker_threshold_) {
            // Handler has failed too many times
            // In production, we would disable this handler
            if (error_callback_) {
                error_callback_(info);
            }
        }
 
        // Notify error callback
        if (error_callback_) {
            error_callback_(info);
        }
    }

References circuit_breaker_threshold_, error_callback_, error_mutex_, failed_handler_counts_, kcenon::monitoring::info, recent_errors_, and total_exceptions_.

Referenced by subscribe_safe().

Here is the caller graph for this function:

◆ process_dead_letters()

template<typename RecoveryHandler >

void kcenon::monitoring::safe_event_dispatcher::process_dead_letters ( RecoveryHandler && handler )

inline

Process dead letter queue with recovery handler.

Definition at line 193 of file safe_event_dispatcher.h.

                                                         {
        std::lock_guard<std::mutex> lock(dlq_mutex_);
 
        for (auto& dlq_entry : dead_letter_queue_) {
            try {
                handler(dlq_entry);
            } catch (...) {
                // Recovery handler failed, skip
            }
        }
 
        dead_letter_queue_.clear();
    }

References dead_letter_queue_, and dlq_mutex_.

◆ publish_safe()

template<typename EventType >

bool kcenon::monitoring::safe_event_dispatcher::publish_safe	(	const EventType &	event,
		event_priority	priority = event_priority::normal )

inline

Publish event with error recovery.

Parameters

event	Event to publish
priority	Event priority

Returns: true if published successfully

Definition at line 104 of file safe_event_dispatcher.h.

                                                                                              {
        try {
            bus_->publish(event, priority);
            return true;
        } catch (const std::exception& e) {
            // Log error
            if (error_callback_) {
                handler_error_info info{
                    0, // No specific handler
                    std::string("Publish failed: ") + e.what(),
                    typeid(EventType),
                    std::chrono::steady_clock::now()
                };
                error_callback_(info);
            }
 
            // Add to dead letter queue
            add_to_dead_letter_queue(typeid(EventType), e.what());
 
            return false;
        } catch (...) {
            if (error_callback_) {
                handler_error_info info{
                    0,
                    "Publish failed: Unknown exception",
                    typeid(EventType),
                    std::chrono::steady_clock::now()
                };
                error_callback_(info);
            }
 
            add_to_dead_letter_queue(typeid(EventType), "Unknown exception");
            return false;
        }
    }

References add_to_dead_letter_queue(), bus_, error_callback_, and kcenon::monitoring::info.

Here is the call graph for this function:

◆ set_circuit_breaker_threshold()

void kcenon::monitoring::safe_event_dispatcher::set_circuit_breaker_threshold ( size_t threshold )

inline

Set circuit breaker threshold.

Parameters

threshold Number of failures before handler is disabled

Definition at line 177 of file safe_event_dispatcher.h.

                                                         {
        circuit_breaker_threshold_ = threshold;
    }

References circuit_breaker_threshold_.

◆ set_error_callback()

void kcenon::monitoring::safe_event_dispatcher::set_error_callback ( error_callback callback )

inline

Set callback for handler errors.

Definition at line 143 of file safe_event_dispatcher.h.

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

References error_callback_, and error_mutex_.

◆ subscribe_safe()

template<typename EventType , typename Handler >

uint64_t kcenon::monitoring::safe_event_dispatcher::subscribe_safe	(	Handler &&	handler,
		event_priority	priority = event_priority::normal )

inline

Subscribe to events with automatic exception handling.

Parameters

handler	Event handler function
priority	Handler priority

Returns: Subscription ID

Definition at line 81 of file safe_event_dispatcher.h.

                                                                                               {
        // Wrap handler with exception handling
        auto safe_handler = [this, h = std::forward<Handler>(handler), id = next_id_++]
                           (const EventType& event) {
            try {
                h(event);
            } catch (const std::exception& e) {
                handle_exception(id, typeid(EventType), e.what());
            } catch (...) {
                handle_exception(id, typeid(EventType), "Unknown exception");
            }
        };
 
        return bus_->subscribe<EventType>(std::move(safe_handler));
    }

References bus_, handle_exception(), and next_id_.

Here is the call graph for this function:

Member Data Documentation

◆ bus_

std::shared_ptr<interface_event_bus> kcenon::monitoring::safe_event_dispatcher::bus_

private

Definition at line 281 of file safe_event_dispatcher.h.

Referenced by get_bus(), publish_safe(), and subscribe_safe().

◆ circuit_breaker_threshold_

size_t kcenon::monitoring::safe_event_dispatcher::circuit_breaker_threshold_

private

Definition at line 289 of file safe_event_dispatcher.h.

Referenced by handle_exception(), and set_circuit_breaker_threshold().

◆ dead_letter_queue_

std::vector<dead_letter_entry> kcenon::monitoring::safe_event_dispatcher::dead_letter_queue_

private

Definition at line 292 of file safe_event_dispatcher.h.

Referenced by add_to_dead_letter_queue(), get_dead_letter_queue_size(), and process_dead_letters().

◆ dlq_mutex_

std::mutex kcenon::monitoring::safe_event_dispatcher::dlq_mutex_

mutableprivate

Definition at line 291 of file safe_event_dispatcher.h.

Referenced by add_to_dead_letter_queue(), get_dead_letter_queue_size(), and process_dead_letters().

◆ error_callback_

error_callback kcenon::monitoring::safe_event_dispatcher::error_callback_

private

Definition at line 288 of file safe_event_dispatcher.h.

Referenced by handle_exception(), publish_safe(), and set_error_callback().

◆ error_mutex_

std::mutex kcenon::monitoring::safe_event_dispatcher::error_mutex_

mutableprivate

Definition at line 285 of file safe_event_dispatcher.h.

Referenced by clear_errors(), get_handler_errors(), handle_exception(), and set_error_callback().

◆ failed_handler_counts_

std::unordered_map<uint64_t, size_t> kcenon::monitoring::safe_event_dispatcher::failed_handler_counts_

private

Definition at line 287 of file safe_event_dispatcher.h.

Referenced by clear_errors(), and handle_exception().

◆ next_id_

std::atomic<uint64_t> kcenon::monitoring::safe_event_dispatcher::next_id_ {1}

private

Definition at line 282 of file safe_event_dispatcher.h.

282{1};

Referenced by subscribe_safe().

◆ recent_errors_

std::vector<handler_error_info> kcenon::monitoring::safe_event_dispatcher::recent_errors_

private

Definition at line 286 of file safe_event_dispatcher.h.

Referenced by clear_errors(), get_handler_errors(), and handle_exception().

◆ total_exceptions_

std::atomic<uint64_t> kcenon::monitoring::safe_event_dispatcher::total_exceptions_ {0}

private

Definition at line 283 of file safe_event_dispatcher.h.

283{0};

Referenced by clear_errors(), get_total_exceptions(), and handle_exception().

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

include/kcenon/monitoring/core/safe_event_dispatcher.h

Classes

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Production Safety

Member Typedef Documentation

◆ error_callback

Constructor & Destructor Documentation

◆ safe_event_dispatcher()

Member Function Documentation

◆ add_to_dead_letter_queue()

◆ clear_errors()

◆ get_bus()

◆ get_dead_letter_queue_size()

◆ get_handler_errors()

◆ get_total_exceptions()

◆ handle_exception()

◆ process_dead_letters()

◆ publish_safe()

◆ set_circuit_breaker_threshold()

◆ set_error_callback()

◆ subscribe_safe()

Member Data Documentation

◆ bus_

◆ circuit_breaker_threshold_

◆ dead_letter_queue_

◆ dlq_mutex_

◆ error_callback_

◆ error_mutex_

◆ failed_handler_counts_

◆ next_id_

◆ recent_errors_

◆ total_exceptions_