event_bus.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <any>
#include <functional>
#include <memory>
#include <mutex>
#include <typeindex>
#include <unordered_map>
#include <vector>
#include <kcenon/thread/core/thread_pool.h>
#include <kcenon/thread/core/thread_pool_builder.h>
 
namespace kcenon::thread {
 
class event_base {
public:
    virtual ~event_base() = default;
 
    virtual std::string type_name() const = 0;
 
    std::chrono::steady_clock::time_point timestamp() const {
        return timestamp_;
    }
 
protected:
    std::chrono::steady_clock::time_point timestamp_{std::chrono::steady_clock::now()};
};
 
class event_bus {
public:
    using handler_func = std::function<void(const std::any&)>;
 
    class subscription {
    public:
        subscription() = default;
        subscription(event_bus* bus, std::type_index type, std::size_t id)
            : bus_(bus), type_(type), id_(id) {}
 
        void unsubscribe() {
            if (bus_) {
                bus_->unsubscribe(type_, id_);
                bus_ = nullptr;
            }
        }
 
        bool is_active() const {
            return bus_ != nullptr;
        }
 
        // Move operations
        subscription(subscription&& other) noexcept
            : bus_(other.bus_), type_(other.type_), id_(other.id_) {
            other.bus_ = nullptr;
        }
 
        subscription& operator=(subscription&& other) noexcept {
            if (this != &other) {
                unsubscribe();
                bus_ = other.bus_;
                type_ = other.type_;
                id_ = other.id_;
                other.bus_ = nullptr;
            }
            return *this;
        }
 
        // Disable copy
        subscription(const subscription&) = delete;
        subscription& operator=(const subscription&) = delete;
 
        ~subscription() {
            unsubscribe();
        }
 
    private:
        event_bus* bus_{nullptr};
        std::type_index type_{typeid(void)};
        std::size_t id_{0};
    };
 
    explicit event_bus(std::shared_ptr<thread_pool> pool = nullptr)
        : 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();
        }
    }
 
    template<typename Event>
    void publish(const Event& event) {
        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
                    }
                }
            });
        }
    }
 
    template<typename Event>
    void publish_sync(const Event& event) {
        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
                }
            }
        }
    }
 
    template<typename Event>
    [[nodiscard]] subscription subscribe(std::function<void(const Event&)> handler) {
        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);
    }
 
    template<typename Event>
    void clear_subscriptions() {
        auto type = std::type_index(typeid(Event));
        std::lock_guard<std::mutex> lock(mutex_);
        handlers_.erase(type);
    }
 
    void clear_all_subscriptions() {
        std::lock_guard<std::mutex> lock(mutex_);
        handlers_.clear();
    }
 
    template<typename Event>
    std::size_t subscriber_count() const {
        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;
    }
 
    static event_bus& instance() {
        static event_bus instance;
        return instance;
    }
 
private:
    void unsubscribe(std::type_index type, std::size_t id) {
        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);
            }
        }
    }
 
    mutable 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};
};
 
// Common event types
 
struct system_startup_event : event_base {
    std::string system_name;
 
    explicit system_startup_event(std::string name)
        : system_name(std::move(name)) {}
 
    std::string type_name() const override {
        return "SystemStartupEvent";
    }
};
 
struct system_shutdown_event : event_base {
    std::string system_name;
    std::string reason;
 
    system_shutdown_event(std::string name, std::string reason_msg = "")
        : system_name(std::move(name)), reason(std::move(reason_msg)) {}
 
    std::string type_name() const override {
        return "SystemShutdownEvent";
    }
};
 
struct config_changed_event : event_base {
    std::string config_path;
    std::any old_value;
    std::any new_value;
 
    config_changed_event(std::string path, std::any old_val, std::any new_val)
        : config_path(std::move(path))
        , old_value(std::move(old_val))
        , new_value(std::move(new_val)) {}
 
    std::string type_name() const override {
        return "ConfigChangedEvent";
    }
};
 
struct performance_alert_event : event_base {
    enum class severity { info, warning, critical };
 
    severity level;
    std::string message;
    double metric_value;
 
    performance_alert_event(severity lvl, std::string msg, double value)
        : level(lvl), message(std::move(msg)), metric_value(value) {}
 
    std::string type_name() const override {
        return "PerformanceAlertEvent";
    }
};
 
} // namespace kcenon::thread