event_bus.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <string>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <atomic>
#include <unordered_map>
#include <algorithm>
#include <vector>
#include <typeinfo>
#include <typeindex>
#include <concepts>
 
// Include C++20 concepts for type validation
#include <kcenon/common/concepts/event.h>
 
// Provide a simple standalone implementation
// Note: monitoring_system can extend or wrap this if needed
 
namespace kcenon::common {
namespace detail {
    // Version identifier for ABI compatibility checking
    // This ensures that object files compiled with different macro definitions
    // cannot be linked together, preventing subtle runtime errors.
    constexpr int event_bus_abi_version = 1;  // Standalone implementation
 
    template<typename T>
    struct event_type_id {
        static size_t id() {
            // Use std::type_index hash for deterministic, consistent IDs
            static const size_t type_id = std::hash<std::type_index>{}(std::type_index(typeid(T)));
            return type_id;
        }
    };
} // namespace detail
 
    // Public API - expose detail::event_type_id for backward compatibility
    using detail::event_type_id;
 
    enum class event_priority {
        low = 0,
        normal = 1,
        high = 2
    };
 
    using subscription_id = uint64_t;
 
    struct event {
        std::string type_;
        std::string data_;
 
        event() = default;
        event(const std::string& type, const std::string& data = "")
            : type_(type), data_(data) {}
 
        void set_type(const std::string& type) { type_ = type; }
        void set_data(const std::string& data) { data_ = data; }
        std::string get_type() const { return type_; }
        std::string get_data() const { return data_; }
    };
 
    class simple_event_bus {
    public:
        using error_callback_t = std::function<void(const std::string& error_message,
                                                     size_t event_type_id,
                                                     uint64_t handler_id)>;
 
        template<concepts::EventType EventType>
        void publish(const EventType& evt, event_priority = event_priority::normal) {
            // 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);
                    }
                }
            }
        }
 
        // For generic event (overload for type deduction)
        void publish(event&& evt, event_priority = event_priority::normal) {
            // 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);
                    }
                }
            }
        }
 
        template<concepts::EventType EventType, concepts::EventHandler<EventType> HandlerFunc>
        uint64_t subscribe(HandlerFunc&& func) {
            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;
        }
 
        // Non-template overload for generic event
        uint64_t subscribe(std::function<void(const event&)>&& func) {
            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;
        }
 
        template<concepts::EventType EventType,
                 concepts::EventHandler<EventType> HandlerFunc,
                 concepts::EventFilter<EventType> FilterFunc>
        uint64_t subscribe_filtered(HandlerFunc&& func, FilterFunc&& filter) {
            // 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));
        }
 
        // Non-template overload for generic event filtering
        uint64_t subscribe_filtered(
            std::function<void(const event&)>&& func,
            std::function<bool(const event&)>&& filter) {
            // 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));
        }
 
        void unsubscribe(uint64_t id) {
            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;
                }
            }
        }
 
        void start() { running_ = true; }
        void stop() { running_ = false; }
        bool is_running() const { return running_; }
 
        void set_error_callback(error_callback_t callback) {
            std::lock_guard<std::mutex> lock(mutex_);
            error_callback_ = std::move(callback);
        }
 
        void clear_error_callback() {
            std::lock_guard<std::mutex> lock(mutex_);
            error_callback_ = nullptr;
        }
 
        static simple_event_bus& instance() {
            static simple_event_bus instance;
            return instance;
        }
 
    private:
        struct subscription_info {
            uint64_t id;
            size_t expected_type_id;  // Type safety: store expected type ID
            std::function<void(const void*)> handler;
        };
 
        mutable 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_;  // Optional error callback for exception handling
    };
 
    using event_bus = simple_event_bus;
 
    template<typename T>
    using event_handler = std::function<void(const T&)>;
 
    inline simple_event_bus& get_event_bus() {
        return simple_event_bus::instance();
    }
}
 
// ABI compatibility check functions
namespace kcenon::common {
namespace detail {
    inline constexpr int get_event_bus_abi_version() {
        return event_bus_abi_version;
    }
} // namespace detail
 
    inline bool verify_event_bus_abi(int expected_version) {
        return detail::get_event_bus_abi_version() == expected_version;
    }
} // namespace kcenon::common
 
// Common event types that can be used across modules
namespace kcenon::common {
namespace events {
 
struct module_started_event {
    std::string module_name;
    std::chrono::steady_clock::time_point timestamp;
 
    module_started_event(const std::string& name)
        : module_name(name),
          timestamp(std::chrono::steady_clock::now()) {}
};
 
struct module_stopped_event {
    std::string module_name;
    std::chrono::steady_clock::time_point timestamp;
 
    module_stopped_event(const std::string& name)
        : module_name(name),
          timestamp(std::chrono::steady_clock::now()) {}
};
 
struct error_event {
    std::string module_name;
    std::string error_message;
    int error_code;
    std::chrono::steady_clock::time_point timestamp;
 
    error_event(const std::string& module, const std::string& message, int code)
        : module_name(module), error_message(message), error_code(code),
          timestamp(std::chrono::steady_clock::now()) {}
};
 
struct metric_event {
    std::string name;
    double value;
    std::string unit;
    std::chrono::steady_clock::time_point timestamp;
 
    metric_event(const std::string& n, double v, const std::string& u = "")
        : name(n), value(v), unit(u),
          timestamp(std::chrono::steady_clock::now()) {}
};
 
} // namespace events
} // namespace kcenon::common