diff --git a/Minecraft-Community-Edition/Core/QEventBus.cpp b/Minecraft-Community-Edition/Core/QEventBus.cpp index 16b1482..eaf1b3f 100644 --- a/Minecraft-Community-Edition/Core/QEventBus.cpp +++ b/Minecraft-Community-Edition/Core/QEventBus.cpp @@ -1 +1,53 @@ #include "QEventBus.hpp" + +namespace mce { + void QEventBus::process() { + eastl::queue> localQueue; + { + std::lock_guard lock(QEventBus::queueMutex); + eastl::swap(localQueue, QEventBus::queue); + } + this; + while (!localQueue.empty()) { + QEventBus::dispatch(*localQueue.front()); + localQueue.pop(); + } + } + + void QEventBus::runAsync() { + QEventBus::bIsRunning = true; + QEventBus::thread = std::thread([this]() { + while (QEventBus::bIsRunning) { + eastl::unique_ptr event; + { + std::unique_lock lock(QEventBus::queueMutex); + QEventBus::conditionalVariable.wait(lock, [&]() { return !QEventBus::queue.empty() || !QEventBus::bIsRunning; }); + if (!QEventBus::bIsRunning) break; + event = std::move(QEventBus::queue.front()); + QEventBus::queue.pop(); + } + QEventBus::dispatch(*event); + } + }); + } + + void QEventBus::stop() { + QEventBus::bIsRunning = false; + QEventBus::conditionalVariable.notify_all(); + if (QEventBus::thread.joinable()) QEventBus::thread.join(); + } + + QEventBus::~QEventBus() { + QEventBus::stop(); + } + + void QEventBus::dispatch(const IEvent& e) { + std::lock_guard lock(QEventBus::mutex); + auto it = QEventBus::handlers.find(typeid(e)); + if (it != QEventBus::handlers.end()) { + for (auto& handler : it->second) { + handler(e); + } + } + } +} diff --git a/Minecraft-Community-Edition/Core/QEventBus.hpp b/Minecraft-Community-Edition/Core/QEventBus.hpp index 8f168b5..dbce7de 100644 --- a/Minecraft-Community-Edition/Core/QEventBus.hpp +++ b/Minecraft-Community-Edition/Core/QEventBus.hpp @@ -1,3 +1,256 @@ #pragma once -class QEventBus {}; +#include +#include +#include +#include +#include +#include +#include +#include "IEvent.hpp" +namespace mce { + + /** + * @brief Token object that controls the lifetime of an event subscription. + * + * The destructor will call the stored unsubscribe callable if it is set, + * causing the associated handler to be removed from the event bus. + * + * This type is move- and copy-assignable depending on the semantics of the + * stored eastl::function. Use subscribeRAII to obtain a token. + */ + struct SubscriptionToken { + /** + * @brief Callable invoked to perform the unsubscribe action. + * + * If empty, destruction of the token has no effect. + */ + eastl::function unsubscribe; + + /** + * @brief Destructor. + * + * Automatically unsubscribes if an unsubscribe callable is present. + */ + ~SubscriptionToken() { if (unsubscribe)unsubscribe(); } + }; + + /** + * @brief Thread-safe event bus. + * + * QEventBus allows subscribing handlers to event types, posting events to a + * queue, and dispatching queued events to all registered handlers. It + * supports both synchronous processing via process() and asynchronous + * processing via runAsync()/stop(). + * + * Handlers are stored per event type (std::type_index of the concrete + * event) and receive events by const-reference to the concrete event type. + * + * @note All public subscribe/post/process operations are designed to be + * thread-safe, but handlers themselves are invoked on the processing + * thread/context and must be safe to execute there. + */ + class QEventBus { + public: + /** + * @brief Alias for an event handler function. + * @tparam EventType Concrete event type handled by the function. + * + * The handler receives a const reference to the event instance. + */ + template + using Handler = eastl::function; + + /** + * @brief Subscribe a handler to EventType and receive a RAII token. + * @tparam EventType The event type to subscribe to. + * @param handler The handler called when an EventType is dispatched. + * @return SubscriptionToken RAII object that will unsubscribe on destruction. + * + * The returned token ensures the handler is removed when the token is + * destroyed. The unsubscribe action captured in the token is safe to + * call from any thread. + */ + template + SubscriptionToken subscribeRAII(Handler handler); + + /** + * @brief Subscribe a handler to EventType without RAII token. + * @tparam EventType The event type to subscribe to. + * @param handler The handler called when an EventType is dispatched. + * + * Use this if you do not need automated unsubscription. To remove the + * handler manually you must extend this class or implement additional + * removal logic. + */ + template + void subscribe(Handler handler); + + /** + * @brief Post an event instance to the bus for later processing. + * @tparam EventType The concrete event type being posted. + * @param event The event instance to enqueue (moved or copied as needed). + * + * The event is stored as a heap-allocated IEvent-derived instance and + * will be dispatched during process() or by the asynchronous thread if + * runAsync() is active. + */ + template + void post(EventType&& event); + + /** + * @brief Process and dispatch all currently queued events. + * + * This method will dequeue all pending events and invoke their + * registered handlers synchronously in the caller's thread. + */ + void process(); + + /** + * @brief Start asynchronous processing of queued events. + * + * This creates a background thread that waits for posted events and + * calls dispatch on them as they arrive. + */ + void runAsync(); + + /** + * @brief Stop asynchronous processing and join the worker thread. + * + * Blocks until the background processing thread has exited. After + * stop() returns, it is safe to destroy the QEventBus or call + * runAsync() again. + */ + void stop(); + + /** + * @brief Destructor. + * + * Ensures asynchronous processing is stopped and resources are cleaned + * up. Any remaining queued events will be destroyed. + */ + ~QEventBus(); + private: + /** + * @brief Dispatch a concrete IEvent to all registered handlers for its type. + * @param e The event to dispatch. + * + * This method performs no locking of the event queue; it only queries + * the handlers map and invokes handlers. Callers are responsible for + * appropriate locking when necessary. + */ + void dispatch(const IEvent& e); + + + struct TypeIndexHash { + size_t operator()(const std::type_index& t) const noexcept { + return t.hash_code(); + } + }; + + struct TypeIndexEqual { + bool operator()(const std::type_index& a, const std::type_index& b) const noexcept { + return a == b; + } + }; + /** + * @brief Map of event type to list of handlers accepting IEvent references. + * + * Key: std::type_index corresponding to the concrete EventType. + * Value: vector of callables accepting const IEvent&; callers cast to the concrete type. + * + * @note Protected by QEventBus::mutex for thread-safe access. + */ + eastl::unordered_map< + std::type_index, + eastl::vector>, + TypeIndexHash, + TypeIndexEqual + > handlers; + /** + * @brief Queue of pending events awaiting processing. + * + * Each entry is a unique_ptr to an IEvent instance holding the concrete event. + * + * @note Protected by QEventBus::queueMutex for thread-safe access. + */ + eastl::queue> queue; + + /** + * @brief Mutex protecting the handlers map and subscription modification. + */ + std::mutex mutex; + + /** + * @brief Mutex protecting the event queue. + */ + std::mutex queueMutex; + + /** + * @brief Condition variable used to wake the background processing thread. + */ + std::condition_variable conditionalVariable; + + /** + * @brief Flag indicating whether the asynchronous processing thread should run. + */ + bool bIsRunning = false; + + /** + * @brief Background thread used for asynchronous processing when runAsync() is called. + */ + std::thread thread; + }; + template + inline SubscriptionToken QEventBus::subscribeRAII(Handler handler) { + std::lock_guard lock(QEventBus::mutex); + + auto& handlers = QEventBus::handlers[typeid(EventType)]; + + handlers.emplace_back([handler](const IEvent& e) { + handler(static_cast(e)); + }); + + auto it = eastl::prev(handlers.end()); + std::type_index typeIdx = std::type_index(typeid(EventType)); + + return SubscriptionToken{ [this, typeIdx, it]() { + std::lock_guard lock(QEventBus::mutex); + auto mapIt = QEventBus::handlers.find(typeIdx); + if (mapIt != QEventBus::handlers.end()) { + auto& vec = mapIt->second; + if (!vec.empty()) { + vec.erase(); + } + + if (vec.empty()) { + QEventBus::handlers.erase(mapIt); + } + } + } + }; + } + + template + inline void QEventBus::subscribe(Handler handler) { + std::lock_guard lock(QEventBus::mutex); + + // Simply store the handler — no token or RAII logic. + auto& handlers = QEventBus::handlers[typeid(EventType)]; + handlers.emplace_back([handler](const IEvent& e) { + handler(static_cast(e)); + }); + } + + template + inline void QEventBus::post(EventType&& event) { + // Create an owned copy / move of the event on the heap + using Decayed = std::decay_t; + auto ptr = eastl::make_unique(eastl::forward(event)); + { + std::lock_guard lock(QEventBus::queueMutex); + QEventBus::queue.push(eastl::move(ptr)); + } + QEventBus::conditionalVariable.notify_one(); + } +}