event_bus_interface.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 <atomic>
#include <chrono>
#include <concepts>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <type_traits>
#include <typeindex>
#include <unordered_map>
#include <vector>
#include "../core/result_types.h"
#include "../config/feature_flags.h"
 
#if KCENON_HAS_COMMON_SYSTEM
#include <kcenon/common/concepts/event.h>
#endif
 
namespace kcenon { namespace monitoring {
 
namespace concepts {
 
template <typename T>
concept EventType = std::is_class_v<T> && std::is_copy_constructible_v<T>;
 
template <typename H, typename E>
concept EventHandler =
    std::invocable<H, const E&> && std::is_void_v<std::invoke_result_t<H, const E&>>;
 
template <typename F, typename E>
concept EventFilter =
    std::invocable<F, const E&> && std::convertible_to<std::invoke_result_t<F, const E&>, bool>;
 
} // namespace concepts
 
class event_base {
public:
    virtual ~event_base() = default;
 
    virtual std::string get_type_name() const = 0;
 
    std::chrono::steady_clock::time_point get_timestamp() const {
        return timestamp_;
    }
 
    uint64_t get_id() const {
        return id_;
    }
 
protected:
    event_base()
        : timestamp_(std::chrono::steady_clock::now()),
          id_(generate_id()) {}
 
private:
    static uint64_t generate_id() {
        static std::atomic<uint64_t> counter{0};
        return counter.fetch_add(1, std::memory_order_relaxed);
    }
 
    std::chrono::steady_clock::time_point timestamp_;
    uint64_t id_;
};
 
enum class event_priority {
    low,
    normal,
    high,
    critical
};
 
template<typename EventType>
class event_handler {
public:
    using handler_func = std::function<void(const EventType&)>;
 
    event_handler(handler_func handler, event_priority priority = event_priority::normal)
        : handler_(handler), priority_(priority), id_(generate_id()) {}
 
    void operator()(const EventType& event) const {
        if (handler_) {
            handler_(event);
        }
    }
 
    event_priority get_priority() const { return priority_; }
    uint64_t get_id() const { return id_; }
 
private:
    static uint64_t generate_id() {
        static std::atomic<uint64_t> counter{0};
        return counter.fetch_add(1, std::memory_order_relaxed);
    }
 
    handler_func handler_;
    event_priority priority_;
    uint64_t id_;
};
 
class subscription_token {
public:
    subscription_token(std::type_index event_type, uint64_t handler_id)
        : event_type_(event_type), handler_id_(handler_id) {}
 
    std::type_index get_event_type() const { return event_type_; }
    uint64_t get_handler_id() const { return handler_id_; }
 
private:
    std::type_index event_type_;
    uint64_t handler_id_;
};
 
class interface_event_bus {
public:
    virtual ~interface_event_bus() = default;
 
    template <concepts::EventType E>
    common::VoidResult publish_event(const E& event) {
        return publish_event_impl(std::type_index(typeid(E)), std::make_any<E>(event));
    }
 
    template <concepts::EventType E>
    common::Result<subscription_token> subscribe_event(std::function<void(const E&)> handler,
                                               event_priority priority = event_priority::normal) {
        auto wrapped_handler = event_handler<E>(handler, priority);
        return subscribe_event_impl(
            std::type_index(typeid(E)),
            [wrapped_handler](const std::any& event) { wrapped_handler(std::any_cast<E>(event)); },
            wrapped_handler.get_id(), priority);
    }
 
    template <concepts::EventType E, concepts::EventHandler<E> H>
    common::Result<subscription_token> subscribe_event(H&& handler,
                                               event_priority priority = event_priority::normal) {
        return subscribe_event(std::function<void(const E&)>(std::forward<H>(handler)), priority);
    }
 
    virtual common::VoidResult unsubscribe_event(const subscription_token& token) = 0;
 
    template <concepts::EventType E>
    common::VoidResult clear_subscriptions() {
        return clear_subscriptions_impl(std::type_index(typeid(E)));
    }
 
    template <concepts::EventType E>
    size_t get_subscriber_count() const {
        return get_subscriber_count_impl(std::type_index(typeid(E)));
    }
 
    virtual bool is_active() const = 0;
 
    virtual common::VoidResult start() = 0;
 
    virtual common::VoidResult stop() = 0;
 
    virtual size_t get_pending_event_count() const = 0;
 
    virtual common::VoidResult process_pending_events() = 0;
 
protected:
    // Implementation methods to be overridden by concrete classes
    virtual common::VoidResult publish_event_impl(
        std::type_index event_type,
        std::any event) = 0;
 
    virtual 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) = 0;
 
    virtual common::VoidResult clear_subscriptions_impl(
        std::type_index event_type) = 0;
 
    virtual size_t get_subscriber_count_impl(
        std::type_index event_type) const = 0;
};
 
class interface_event_publisher {
public:
    virtual ~interface_event_publisher() = default;
 
    virtual common::VoidResult set_event_bus(std::shared_ptr<interface_event_bus> bus) = 0;
 
    virtual std::shared_ptr<interface_event_bus> get_event_bus() const = 0;
};
 
class interface_event_subscriber {
public:
    virtual ~interface_event_subscriber() = default;
 
    virtual common::VoidResult subscribe_to_events(std::shared_ptr<interface_event_bus> bus) = 0;
 
    virtual common::VoidResult unsubscribe_from_events() = 0;
 
    virtual std::vector<subscription_token> get_subscriptions() const = 0;
};
 
} } // namespace kcenon::monitoring