bus.h
Go to the documentation of this file.
/*
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 final
{
friend class Node;
using VectorNode = std::vector<INode *>;
public:
Bus();
~Bus() = default;
Bus(const Bus &obj) = delete;
Bus(Bus &&obj) noexcept = delete;
Bus &operator=(const Bus &obj) = delete;
Bus &operator=(Bus &&obj) noexcept = delete;
void clearAllSubscriptions();
void subscribe(INode *node, MessageTypeID eventType);
void unsubscribe(INode *node, MessageTypeID eventType);
void unsubscribeFromAll(INode *node);
[[nodiscard]] bool isSubscribed(const INode *node, MessageTypeID eventType) const;
void subscribeToBroadcast(INode *node);
void unsubscribeFromBroadcast(INode *node);
[[nodiscard]] bool isSubscribedToBroadcast(INode *node) const; // Check if a node is subscribed to broadcast mode
template<class T, class... Args>
bool emit(Args &&...args);
bool post(const std::shared_ptr<Message> &message) const; // Post a message to the bus
private:
size_t postMessageToSubscribers(MessageTypeID eventType, const std::shared_ptr<Message> &message) const;
size_t postMessageToBroadcastSubscribers(const std::shared_ptr<Message> &message) const;
const VectorNode *getSubscribersForMessageType(MessageTypeID messageType) const; // Get the list of subscribers for a specific message type
TimePoint m_busStartTimestamp;
struct
{
mutable std::mutex mtx;
std::unordered_map<MessageTypeID, VectorNode> nodeList;
std::set<INode *> broadcastNodeList;
} 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
sdi_toolBox::desktop::eventBus::Bus
Central message dispatcher for the event bus system.
Definition bus.h:63
sdi_toolBox::desktop::eventBus::Bus::unsubscribe
void unsubscribe(INode *node, MessageTypeID eventType)
Unsubscribe a node from a specific message type.
Definition bus.h:352
sdi_toolBox::desktop::eventBus::Bus::Bus
Bus()
Default constructor.
Definition bus.h:312
sdi_toolBox::desktop::eventBus::Bus::subscribeToBroadcast
void subscribeToBroadcast(INode *node)
Subscribe a node to broadcast mode.
Definition bus.h:405
sdi_toolBox::desktop::eventBus::Bus::broadcastNodeList
std::set< INode * > broadcastNodeList
Set of nodes subscribed to receive all messages (broadcast mode)
Definition bus.h:305
sdi_toolBox::desktop::eventBus::Bus::Bus
Bus(Bus &&obj) noexcept=delete
Move constructor - deleted.
sdi_toolBox::desktop::eventBus::Bus::Bus
Bus(const Bus &obj)=delete
Copy constructor - deleted.
sdi_toolBox::desktop::eventBus::Bus::nodeList
std::unordered_map< MessageTypeID, VectorNode > nodeList
Map of message type IDs to lists of subscribed nodes.
Definition bus.h:304
sdi_toolBox::desktop::eventBus::Bus::isSubscribedToBroadcast
bool isSubscribedToBroadcast(INode *node) const
Check whether a node is subscribed to broadcast mode.
Definition bus.h:431
sdi_toolBox::desktop::eventBus::Bus::isSubscribed
bool isSubscribed(const INode *node, MessageTypeID eventType) const
Check whether a node is subscribed to a specific message type.
Definition bus.h:387
sdi_toolBox::desktop::eventBus::Bus::post
bool post(const std::shared_ptr< Message > &message) const
Post an already constructed message to the bus.
Definition bus.h:457
sdi_toolBox::desktop::eventBus::Bus::emit
bool emit(Args &&...args)
Construct and emit a message of type T to the bus.
Definition bus.h:445
sdi_toolBox::desktop::eventBus::Bus::subscribe
void subscribe(INode *node, MessageTypeID eventType)
Subscribe a node to a specific message type.
Definition bus.h:328
sdi_toolBox::desktop::eventBus::Bus::~Bus
~Bus()=default
Default destructor.
sdi_toolBox::desktop::eventBus::Bus::clearAllSubscriptions
void clearAllSubscriptions()
Remove all subscriptions from the routing table.
Definition bus.h:319
sdi_toolBox::desktop::eventBus::Bus::operator=
Bus & operator=(Bus &&obj) noexcept=delete
Move assignment operator - deleted.
sdi_toolBox::desktop::eventBus::Bus::mtx
std::mutex mtx
Mutex for thread-safe access to the nodes map.
Definition bus.h:303
sdi_toolBox::desktop::eventBus::Bus::unsubscribeFromBroadcast
void unsubscribeFromBroadcast(INode *node)
Unsubscribe a node from broadcast mode.
Definition bus.h:418
sdi_toolBox::desktop::eventBus::Bus::unsubscribeFromAll
void unsubscribeFromAll(INode *node)
Unsubscribe a node from all message types and broadcast mode.
Definition bus.h:370
sdi_toolBox::desktop::eventBus::Bus::operator=
Bus & operator=(const Bus &obj)=delete
Copy assignment operator - deleted.
sdi_toolBox::desktop::eventBus::INode
Abstract interface representing a subscriber node in the event bus system.
Definition inode.h:42
sdi_toolBox::desktop::eventBus::Node
Concrete subscriber node in the event bus system.
Definition node.h:64
defs.h
inode.h
message.h
sdi_toolBox::desktop::eventBus
Lightweight thread-safe event bus for decoupled message passing.
Definition bus.h:27
sdi_toolBox::desktop::eventBus::MessageTypeID
uint64_t MessageTypeID
Unique identifier for a message type.
Definition defs.h:21
sdi_toolBox::desktop::eventBus::TimePoint
std::chrono::steady_clock::time_point TimePoint
Monotonic timestamp type used throughout the event bus.
Definition defs.h:22