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 <algorithm>
#include <any>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <typeindex>
#include <unordered_map>
#include <vector>
#include "../interfaces/event_bus_interface.h"
#include "result_types.h"
#include "error_codes.h"
 
namespace kcenon { namespace monitoring {
 
struct event_envelope {
    std::type_index type;
    std::any payload;
    event_priority priority;
    std::chrono::steady_clock::time_point timestamp;
    uint64_t id;
 
    event_envelope(std::type_index t, std::any p, event_priority pr)
        : type(t), payload(std::move(p)), priority(pr),
          timestamp(std::chrono::steady_clock::now()),
          id(generate_id()) {}
 
    bool operator<(const event_envelope& other) const {
        // Higher priority events come first
        if (priority != other.priority) {
            return static_cast<int>(priority) < static_cast<int>(other.priority);
        }
        // For same priority, older events come first
        return timestamp > other.timestamp;
    }
 
private:
    static uint64_t generate_id() {
        static std::atomic<uint64_t> counter{0};
        return counter.fetch_add(1, std::memory_order_relaxed);
    }
};
 
struct event_handler_wrapper {
    std::function<void(const std::any&)> handler;
    event_priority priority;
    uint64_t id;
 
    event_handler_wrapper(std::function<void(const std::any&)> h,
                         event_priority p, uint64_t i)
        : handler(std::move(h)), priority(p), id(i) {}
};
 
struct event_bus_config {
    size_t max_queue_size = 10000;
    size_t worker_thread_count = 2;
    std::chrono::milliseconds processing_interval{10};
    bool auto_start = false;
    bool enable_back_pressure = true;
    size_t back_pressure_threshold = 8000;
};
 
class event_bus : public interface_event_bus {
public:
    using config = event_bus_config;
 
    explicit event_bus(const config& cfg = config())
        : 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();
        }
    }
 
    ~event_bus() {
        stop();
    }
 
    static std::shared_ptr<interface_event_bus> instance() {
        static std::shared_ptr<event_bus> singleton = [] {
            auto bus = std::make_shared<event_bus>();
            (void)bus->start();
            return bus;
        }();
        return singleton;
    }
 
    // Start the event bus
    common::VoidResult start() override {
        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();
    }
 
    // Stop the event bus
    common::VoidResult stop() override {
        {
            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();
    }
 
    // Check if event bus is active
    bool is_active() const override {
        std::lock_guard<std::mutex> lock(bus_mutex_);
        return is_running_;
    }
 
    // Get pending event count
    size_t get_pending_event_count() const override {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        return event_queue_.size();
    }
 
    // Process all pending events synchronously
    common::VoidResult process_pending_events() override {
        process_all_pending();
        return common::ok();
    }
 
    // Unsubscribe from events
    common::VoidResult unsubscribe_event(const subscription_token& token) override {
        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();
    }
 
    // Get statistics
    struct stats {
        uint64_t total_published;
        uint64_t total_processed;
        uint64_t total_dropped;
        size_t current_queue_size;
        size_t subscriber_count;
        bool is_back_pressure_active;
    };
 
    stats get_stats() const {
        // 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
        };
    }
 
protected:
    // Publish event implementation
    common::VoidResult publish_event_impl(std::type_index event_type,
                                  std::any event) override {
        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();
    }
 
    // Subscribe event implementation
    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 {
 
        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);
    }
 
    // Clear subscriptions implementation
    common::VoidResult clear_subscriptions_impl(std::type_index event_type) override {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        handlers_.erase(event_type);
        return common::ok();
    }
 
    // Get subscriber count implementation
    size_t get_subscriber_count_impl(std::type_index event_type) const override {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        auto it = handlers_.find(event_type);
        return it != handlers_.end() ? it->second.size() : 0;
    }
 
private:
    // Process events worker thread
    void process_events_worker() {
        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);
            }
        }
    }
 
    // Process all pending events
    void process_all_pending() {
        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);
        }
    }
 
    // Dispatch event to handlers
    void dispatch_event(const event_envelope& envelope) {
        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
                }
            }
        }
    }
 
    config config_;
    mutable std::mutex bus_mutex_;
    mutable std::mutex queue_mutex_;
    mutable 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_;
};
 
} } // namespace kcenon::monitoring