530 lines
17 KiB
C++
530 lines
17 KiB
C++
/*
|
||
Copyright (c) 2026 - SD-Innovation S.A.S. - FRANCE
|
||
*/
|
||
|
||
/*
|
||
ver: 2.x.x - build: 2026-04-28
|
||
*/
|
||
|
||
/*
|
||
The zlib License
|
||
|
||
Copyright (c) 2026 SD-Innovation S.A.S.
|
||
|
||
This software is provided ‘as-is’, without any express or implied
|
||
warranty. In no event will the authors be held liable for any damages
|
||
arising from the use of this software.
|
||
|
||
Permission is granted to anyone to use this software for any purpose,
|
||
including commercial applications, and to alter it and redistribute it
|
||
freely, subject to the following restrictions:
|
||
|
||
1. The origin of this software must not be misrepresented; you must not
|
||
claim that you wrote the original software. If you use this software
|
||
in a product, an acknowledgment in the product documentation would be
|
||
appreciated but is not required.
|
||
|
||
2. Altered source versions must be plainly marked as such, and must not be
|
||
misrepresented as being the original software.
|
||
|
||
3. This notice may not be removed or altered from any source
|
||
distribution.
|
||
*/
|
||
|
||
#pragma once
|
||
|
||
#include "defs.h"
|
||
#include "inode.h"
|
||
#include "message.h"
|
||
|
||
#include <chrono>
|
||
#include <memory>
|
||
#include <mutex>
|
||
#include <ranges>
|
||
#include <set>
|
||
#include <unordered_map>
|
||
|
||
namespace sdi_toolBox::desktop::eventBus
|
||
{
|
||
//--------------------------------------------------------------
|
||
/**
|
||
* @class Bus
|
||
* @brief Central message dispatcher for the event bus system.
|
||
*
|
||
* The Bus class is the core component of the event bus architecture.
|
||
* It manages a routing table that maps message type identifiers to lists
|
||
* of subscribed nodes, and dispatches messages to the appropriate nodes
|
||
* when they are emitted or posted.
|
||
*
|
||
* Nodes can subscribe to specific message types or to broadcast mode,
|
||
* in which case they receive all messages regardless of their type.
|
||
*
|
||
* The Bus is thread-safe: all operations on the routing table are
|
||
* protected by an internal mutex.
|
||
*
|
||
* @note The Bus is non-copyable and non-movable.
|
||
* @note The Bus does not take ownership of the nodes it manages.
|
||
*
|
||
* @par Example usage:
|
||
* @code
|
||
* sdi_toolBox::desktop::eventBus::Bus bus;
|
||
* sdi_toolBox::desktop::eventBus::Node node(bus);
|
||
*
|
||
* node.subscribe(MY_EVENT_TYPE);
|
||
* bus.emit<MyMessage>(arg1, arg2);
|
||
*
|
||
* auto msg = std::dynamic_pointer_cast<MyMessage>(node.popMessage());
|
||
* @endcode
|
||
*
|
||
* @see Node
|
||
* @see INode
|
||
* @see Message
|
||
*/
|
||
class Bus final
|
||
{
|
||
friend class Node; ///< Allow the Node class to access private members
|
||
|
||
using VectorNode = std::vector<INode *>;
|
||
|
||
public:
|
||
///@name Construction & Destruction
|
||
///@{
|
||
|
||
/**
|
||
* @brief Default constructor.
|
||
*
|
||
* Initializes the Bus and records the construction timestamp
|
||
* used as the bus start reference time.
|
||
*/
|
||
Bus();
|
||
|
||
/**
|
||
* @brief Default destructor.
|
||
*/
|
||
~Bus() = default;
|
||
|
||
/**
|
||
* @brief Copy constructor - deleted.
|
||
*
|
||
* The Bus is non-copyable.
|
||
*/
|
||
Bus(const Bus &obj) = delete;
|
||
|
||
/**
|
||
* @brief Move constructor - deleted.
|
||
*
|
||
* The Bus is non-movable.
|
||
*/
|
||
Bus(Bus &&obj) noexcept = delete;
|
||
|
||
/**
|
||
* @brief Copy assignment operator - deleted.
|
||
*
|
||
* The Bus is non-copyable.
|
||
*/
|
||
Bus &operator=(const Bus &obj) = delete;
|
||
|
||
/**
|
||
* @brief Move assignment operator - deleted.
|
||
*
|
||
* The Bus is non-movable.
|
||
*/
|
||
Bus &operator=(Bus &&obj) noexcept = delete;
|
||
|
||
///@}
|
||
///@name Subscription Management
|
||
///@{
|
||
|
||
/**
|
||
* @brief Remove all subscriptions from the routing table.
|
||
*
|
||
* Clears both the specific event subscriptions and the broadcast
|
||
* subscription list. After this call, no node will receive any message
|
||
* until it re-subscribes.
|
||
*
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void clearAllSubscriptions();
|
||
|
||
/**
|
||
* @brief Subscribe a node to a specific message type.
|
||
*
|
||
* Registers the given node to receive messages of the specified type.
|
||
* If the node is already subscribed to this type, this call has no effect
|
||
* (no duplicates are created).
|
||
*
|
||
* @param node Pointer to the node to subscribe. Must not be @c nullptr.
|
||
* @param eventType The message type identifier to subscribe to.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void subscribe(INode *node, MessageTypeID eventType);
|
||
|
||
/**
|
||
* @brief Unsubscribe a node from a specific message type.
|
||
*
|
||
* Removes the given node from the list of subscribers for the specified
|
||
* message type. If the node was not subscribed to this type, this call
|
||
* has no effect.
|
||
*
|
||
* @param node Pointer to the node to unsubscribe. Must not be @c nullptr.
|
||
* @param eventType The message type identifier to unsubscribe from.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void unsubscribe(INode *node, MessageTypeID eventType);
|
||
|
||
/**
|
||
* @brief Unsubscribe a node from all message types and broadcast mode.
|
||
*
|
||
* Removes the given node from all specific event subscription lists
|
||
* and from the broadcast subscription list. This is automatically called
|
||
* by the Node destructor to ensure no dangling pointers remain in the
|
||
* routing table.
|
||
*
|
||
* @param node Pointer to the node to unsubscribe. Must not be @c nullptr.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void unsubscribeFromAll(INode *node);
|
||
|
||
/**
|
||
* @brief Check whether a node is subscribed to a specific message type.
|
||
*
|
||
* @param node Pointer to the node to check. Must not be @c nullptr.
|
||
* @param eventType The message type identifier to check.
|
||
* @return @c true if the node is subscribed to the given message type,
|
||
* @c false otherwise.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
[[nodiscard]] bool isSubscribed(const INode *node, MessageTypeID eventType) const;
|
||
|
||
///@}
|
||
///@name Broadcast management
|
||
///@{
|
||
|
||
/**
|
||
* @brief Subscribe a node to broadcast mode.
|
||
*
|
||
* A node in broadcast mode receives all messages posted to the bus,
|
||
* regardless of their type. A node can be subscribed to both broadcast
|
||
* mode and specific message types simultaneously, in which case it will
|
||
* receive the message twice for matching types.
|
||
*
|
||
* @param node Pointer to the node to subscribe. Must not be @c nullptr.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void subscribeToBroadcast(INode *node);
|
||
|
||
/**
|
||
* @brief Unsubscribe a node from broadcast mode.
|
||
*
|
||
* Removes the given node from the broadcast subscription list.
|
||
* If the node was not subscribed to broadcast mode, this call has no effect.
|
||
*
|
||
* @param node Pointer to the node to unsubscribe. Must not be @c nullptr.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
void unsubscribeFromBroadcast(INode *node);
|
||
|
||
/**
|
||
* @brief Check whether a node is subscribed to broadcast mode.
|
||
*
|
||
* @param node Pointer to the node to check. Must not be @c nullptr.
|
||
* @return @c true if the node is subscribed to broadcast mode,
|
||
* @c false otherwise.
|
||
*
|
||
* @throws std::runtime_error if @p node is @c nullptr.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
[[nodiscard]] bool isSubscribedToBroadcast(INode *node) const; // Check if a node is subscribed to broadcast mode
|
||
|
||
///@}
|
||
///@name Message transmission
|
||
///@{
|
||
|
||
/**
|
||
* @brief Construct and emit a message of type @p T to the bus.
|
||
*
|
||
* Creates a new message of type @p T by forwarding the provided arguments
|
||
* to its constructor, then posts it to the bus via @ref post().
|
||
*
|
||
* @tparam T The message type to emit. Must be derived from @ref Message.
|
||
* @tparam Args Constructor argument types for @p T.
|
||
* @param args Arguments forwarded to the constructor of @p T.
|
||
* @return @c true if at least one subscriber received the message,
|
||
* @c false otherwise.
|
||
*
|
||
* @note Enforced at compile time: @p T must derive from @ref Message.
|
||
* @note This operation is thread-safe.
|
||
*
|
||
* @par Example:
|
||
* @code
|
||
* bus.emit<MyMessage>(arg1, arg2);
|
||
* @endcode
|
||
*/
|
||
template<class T, class... Args>
|
||
bool emit(Args &&...args);
|
||
|
||
/**
|
||
* @brief Post an already constructed message to the bus.
|
||
*
|
||
* Updates the message timestamp and dispatches it to all nodes subscribed
|
||
* to the message type, as well as all nodes in broadcast mode.
|
||
*
|
||
* @param message Shared pointer to the message to post. Must not be @c nullptr.
|
||
* @return @c true if at least one specific subscriber received the message,
|
||
* @c false otherwise.
|
||
*
|
||
* @note This operation is thread-safe.
|
||
* @see emit()
|
||
*/
|
||
bool post(const std::shared_ptr<Message> &message) const; // Post a message to the bus
|
||
|
||
///@}
|
||
|
||
private:
|
||
/**
|
||
* @brief Dispatch a message to all nodes subscribed to its type.
|
||
* @param eventType The message type identifier.
|
||
* @param message The message to dispatch.
|
||
* @return The number of nodes that received the message.
|
||
*/
|
||
size_t postMessageToSubscribers(MessageTypeID eventType, const std::shared_ptr<Message> &message) const;
|
||
|
||
/**
|
||
* @brief Dispatch a message to all nodes in broadcast mode.
|
||
* @param message The message to dispatch.
|
||
* @return The number of broadcast nodes that received the message.
|
||
*/
|
||
size_t postMessageToBroadcastSubscribers(const std::shared_ptr<Message> &message) const;
|
||
|
||
/**
|
||
* @brief Retrieve the subscriber list for a given message type.
|
||
* @param messageType The message type identifier.
|
||
* @return Pointer to the vector of subscribed nodes, or @c nullptr if none.
|
||
*/
|
||
const VectorNode *getSubscribersForMessageType(MessageTypeID messageType) const; // Get the list of subscribers for a specific message type
|
||
|
||
/// @brief Timestamp recorded when the Bus was constructed.
|
||
TimePoint m_busStartTimestamp;
|
||
|
||
/// @brief Internal routing table, protected by a mutex for thread-safe access.
|
||
struct
|
||
{
|
||
mutable std::mutex mtx; ///< Mutex for thread-safe access to the nodes map
|
||
std::unordered_map<MessageTypeID, VectorNode> nodeList; ///< Map of message type IDs to lists of subscribed nodes
|
||
std::set<INode *> broadcastNodeList; ///< Set of nodes subscribed to receive all messages (broadcast mode)
|
||
} m_routingTable;
|
||
};
|
||
//--------------------------------------------------------------
|
||
|
||
//--------------------------------------------------------------
|
||
/* Default constructor */
|
||
inline Bus::Bus()
|
||
{
|
||
// Initialization
|
||
m_busStartTimestamp = std::chrono::steady_clock::now();
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Clear all subscriptions from the bus (remove all nodes from the routing table) */
|
||
inline void Bus::clearAllSubscriptions()
|
||
{
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
m_routingTable.nodeList.clear(); // Clear all event subscriptions
|
||
m_routingTable.broadcastNodeList.clear(); // Clear all broadcast subscriptions
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Subscribe a listener to a specific event type */
|
||
inline void Bus::subscribe(INode *node, MessageTypeID eventType)
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
if (!m_routingTable.nodeList.contains(eventType)) // Event type not registered yet, create a new entry with the node
|
||
{
|
||
m_routingTable.nodeList[eventType] = { node };
|
||
}
|
||
else // Event type already registered, add the node if not already subscribed
|
||
{
|
||
auto &nodes = m_routingTable.nodeList.at(eventType);
|
||
if (std::ranges::find(nodes, node) == nodes.end())
|
||
{
|
||
// Node not already subscribed, add it to the list
|
||
nodes.push_back(node);
|
||
}
|
||
}
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Unsubscribe a listener from a specific event type */
|
||
inline void Bus::unsubscribe(INode *node, MessageTypeID eventType)
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Event type not recorded, nothing to do
|
||
if (!m_routingTable.nodeList.contains(eventType))
|
||
return;
|
||
|
||
// Event type registered, remove the listener if subscribed
|
||
auto &nodes = m_routingTable.nodeList.at(eventType);
|
||
std::erase(nodes, node);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Unsubscribe a listener from a specific event type */
|
||
inline void Bus::unsubscribeFromAll(INode *node)
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Iterate through all event types and remove the node from each list
|
||
for (auto &nodeList : m_routingTable.nodeList | std::views::values)
|
||
std::erase(nodeList, node);
|
||
|
||
// Remove from broadcast subscriptions as well
|
||
m_routingTable.broadcastNodeList.erase(node);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Check if a listener is subscribed to a specific event type */
|
||
inline bool Bus::isSubscribed(const INode *node, MessageTypeID eventType) const
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Event type not recorded, node not subscribed
|
||
if (!m_routingTable.nodeList.contains(eventType))
|
||
return false;
|
||
|
||
// Event type recorded, check if the node is in the list
|
||
const auto &nodes = m_routingTable.nodeList.at(eventType);
|
||
return std::ranges::find(nodes, node) != nodes.end();
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Subscribe a node to receive all messages (broadcast mode) */
|
||
inline void Bus::subscribeToBroadcast(INode *node)
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Add the node to the broadcast nodes set
|
||
m_routingTable.broadcastNodeList.insert(node);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Unsubscribe a node from broadcast mode */
|
||
inline void Bus::unsubscribeFromBroadcast(INode *node)
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Remove the node from the broadcast nodes set
|
||
m_routingTable.broadcastNodeList.erase(node);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Check if a node is subscribed to broadcast mode */
|
||
inline bool Bus::isSubscribedToBroadcast(INode *node) const
|
||
{
|
||
// Sanity check
|
||
if (!node)
|
||
throw std::runtime_error("invalid node handle");
|
||
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Check if the node is in the broadcast nodes set
|
||
return m_routingTable.broadcastNodeList.contains(node);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Emit a message of type T with the given arguments(create a message and post it to the bus) */
|
||
template<class T, class... Args>
|
||
bool Bus::emit(Args &&...args)
|
||
{
|
||
static_assert(std::derived_from<T, Message>, "T must be derived from IMessage");
|
||
|
||
// Create a message of type T with the given arguments
|
||
auto message = std::make_shared<T>(std::forward<Args>(args)...);
|
||
|
||
// Post the message to the bus
|
||
return post(message);
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Post a message to the bus */
|
||
inline bool Bus::post(const std::shared_ptr<Message> &message) const
|
||
{
|
||
std::scoped_lock lock(m_routingTable.mtx);
|
||
|
||
// Update message state
|
||
message->updateTimestamp(); // Update the timestamp of when the message was posted to the bus
|
||
|
||
// Post the message to all subscribers of the specific
|
||
// message type and get the number of subscribers that
|
||
// received the message
|
||
const auto subscriberCount = postMessageToSubscribers(message->getMessageTypeID(), message);
|
||
|
||
// Post the message to all broadcast subscribers and
|
||
// get the number of subscribers that received the
|
||
// message
|
||
(void)postMessageToBroadcastSubscribers(message);
|
||
|
||
return subscriberCount > 0;
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Post a message to all subscribers of a specific event type and return the number of subscribers that received the message */
|
||
inline size_t Bus::postMessageToSubscribers(const MessageTypeID eventType, const std::shared_ptr<Message> &message) const
|
||
{
|
||
const auto subscriberList = getSubscribersForMessageType(eventType);
|
||
if (!subscriberList)
|
||
return 0;
|
||
|
||
for (const auto &node : *subscriberList)
|
||
node->append(message);
|
||
|
||
return subscriberList->size();
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Post a message to all broadcast subscribers and return the number of subscribers that received the message */
|
||
inline size_t Bus::postMessageToBroadcastSubscribers(const std::shared_ptr<Message> &message) const
|
||
{
|
||
for (const auto &node : m_routingTable.broadcastNodeList)
|
||
node->append(message);
|
||
|
||
return m_routingTable.broadcastNodeList.size();
|
||
}
|
||
//--------------------------------------------------------------
|
||
/* Get the list of subscribers for a specific message type */
|
||
inline const Bus::VectorNode *Bus::getSubscribersForMessageType(const MessageTypeID messageType) const
|
||
{
|
||
if (!m_routingTable.nodeList.contains(messageType))
|
||
return nullptr; // No subscribers for this message type
|
||
|
||
return &m_routingTable.nodeList.at(messageType);
|
||
}
|
||
//--------------------------------------------------------------
|
||
} // namespace sdi_toolBox::desktop::eventBus
|