node.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 "bus.h"
#include "inode.h"
#include <mutex>
#include <queue>
namespace sdi_toolBox::desktop::eventBus
{
//--------------------------------------------------------------
class Node : public INode
{
public:
Node() = delete;
virtual ~Node();
Node(const Node &obj) = delete;
Node(Node &&obj) noexcept = delete;
Node &operator=(const Node &obj) = delete;
Node &operator=(Node &&obj) noexcept = delete;
explicit Node(Bus &bus);
void syncWaitForMessage();
void subscribe(MessageTypeID eventType);
void unsubscribe(MessageTypeID eventType);
void unsubscribeFromAll();
template<class T, class... Args>
bool emit(Args &&...args);
bool post(const std::shared_ptr<Message> &message) const;
size_t getMessageCount() const;
std::shared_ptr<Message> popMessage(); // Pop a message from the node's message queue (remove and return the front message)
private:
void append(const std::shared_ptr<Message> &message) override;
void messageNotify();
Bus &m_bus;
std::atomic_bool m_nodeWaitingFlag{ false };
struct
{
mutable std::mutex mtx;
std::queue<std::shared_ptr<Message>> messageQueue;
} m_busMessages;
};
//--------------------------------------------------------------
/* Constructor */
inline Node::Node(Bus &bus)
: m_bus(bus)
{
// Nothing to do here
}
//--------------------------------------------------------------
/* Default destructor */
inline Node::~Node()
{
// Unsubscribe from all event types when the node is destroyed
unsubscribeFromAll();
// Notify the bus that the node is being destroyed (in case it is waiting for a message)
messageNotify();
}
//--------------------------------------------------------------
/* Synchronously wait for a message to be posted to the bus and received by the node (block the calling thread until a message is received) */
inline void Node::syncWaitForMessage()
{
// Wait until at least one message is received in the node's
// message queue
m_nodeWaitingFlag = false;
m_nodeWaitingFlag.wait(false);
}
//--------------------------------------------------------------
/* Subscribe to receive messages of a specific event type */
inline void Node::subscribe(const MessageTypeID eventType)
{
m_bus.subscribe(this, eventType);
}
//--------------------------------------------------------------
/* Unsubscribe from receiving messages of a specific event type */
inline void Node::unsubscribe(const MessageTypeID eventType)
{
m_bus.unsubscribe(this, eventType);
}
//--------------------------------------------------------------
/* Unsubscribe from receiving messages of all event types */
inline void Node::unsubscribeFromAll()
{
m_bus.unsubscribeFromAll(this);
}
//--------------------------------------------------------------
/* 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 Node::emit(Args &&...args)
{
return m_bus.emit<T>(std::forward<Args>(args)...);
}
//--------------------------------------------------------------
/* Post a message to the bus */
inline bool Node::post(const std::shared_ptr<Message> &message) const
{
return m_bus.post(message);
}
//--------------------------------------------------------------
/* Get the number of messages in the node's message queue */
inline size_t Node::getMessageCount() const
{
std::scoped_lock lock(m_busMessages.mtx);
return m_busMessages.messageQueue.size();
}
//--------------------------------------------------------------
/* Pop a message from the node's message queue (remove and return the front message) */
inline std::shared_ptr<Message> Node::popMessage()
{
std::scoped_lock lock(m_busMessages.mtx);
if (m_busMessages.messageQueue.empty())
return nullptr; // No messages in the queue
auto message = m_busMessages.messageQueue.front(); // Get the front message
m_busMessages.messageQueue.pop(); // Remove the front message from the queue
return message; // Return the popped message
}
//--------------------------------------------------------------
/* Insert a message into the node's message queue (called by the bus when a message is posted to the bus) */
inline void Node::append(const std::shared_ptr<Message> &message)
{
std::scoped_lock lock(m_busMessages.mtx);
m_busMessages.messageQueue.push(message);
// If the node is waiting for a message, notify it that a
// message has been received
messageNotify();
}
//--------------------------------------------------------------
/* Notify the node that a message has been received (used for synchronous waiting) */
inline void Node::messageNotify()
{
if (!m_nodeWaitingFlag.load())
{
m_nodeWaitingFlag = true;
m_nodeWaitingFlag.notify_one();
}
}
//--------------------------------------------------------------
} // namespace sdi_toolBox::desktop::eventBus
bus.h
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::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::unsubscribeFromAll
void unsubscribeFromAll(INode *node)
Unsubscribe a node from all message types and broadcast mode.
Definition bus.h:370
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
sdi_toolBox::desktop::eventBus::Node::operator=
Node & operator=(Node &&obj) noexcept=delete
Move assignment operator - deleted.
sdi_toolBox::desktop::eventBus::Node::Node
Node(Node &&obj) noexcept=delete
Move constructor - deleted.
sdi_toolBox::desktop::eventBus::Node::subscribe
void subscribe(MessageTypeID eventType)
Subscribe this node to a specific message type.
Definition node.h:295
sdi_toolBox::desktop::eventBus::Node::popMessage
std::shared_ptr< Message > popMessage()
Remove and return the front message from the node's queue (FIFO).
Definition node.h:334
sdi_toolBox::desktop::eventBus::Node::~Node
virtual ~Node()
Destructor.
Definition node.h:276
sdi_toolBox::desktop::eventBus::Node::operator=
Node & operator=(const Node &obj)=delete
Copy assignment operator - deleted.
sdi_toolBox::desktop::eventBus::Node::Node
Node(const Node &obj)=delete
Copy constructor - deleted.
sdi_toolBox::desktop::eventBus::Node::messageQueue
std::queue< std::shared_ptr< Message > > messageQueue
Queue of messages received by the node.
Definition node.h:263
sdi_toolBox::desktop::eventBus::Node::syncWaitForMessage
void syncWaitForMessage()
Block the calling thread until a message is received.
Definition node.h:286
sdi_toolBox::desktop::eventBus::Node::post
bool post(const std::shared_ptr< Message > &message) const
Post an already constructed message through the bus.
Definition node.h:320
sdi_toolBox::desktop::eventBus::Node::emit
bool emit(Args &&...args)
Construct and emit a message of type T through the bus.
Definition node.h:314
sdi_toolBox::desktop::eventBus::Node::Node
Node()=delete
Default constructor - deleted. A Bus reference must be provided.
sdi_toolBox::desktop::eventBus::Node::unsubscribe
void unsubscribe(MessageTypeID eventType)
Unsubscribe this node from a specific message type.
Definition node.h:301
sdi_toolBox::desktop::eventBus::Node::getMessageCount
size_t getMessageCount() const
Get the number of messages currently in the node's queue.
Definition node.h:326
sdi_toolBox::desktop::eventBus::Node::mtx
std::mutex mtx
Mutex for thread-safe access to the message queue.
Definition node.h:262
sdi_toolBox::desktop::eventBus::Node::unsubscribeFromAll
void unsubscribeFromAll()
Unsubscribe this node from all message types and broadcast mode.
Definition node.h:307
sdi_toolBox::desktop::eventBus::Node::append
void append(const std::shared_ptr< Message > &message) override
Append a message to the node's internal queue.
Definition node.h:348
inode.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