391 lines
12 KiB
C++
391 lines
12 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 "bus.h"
|
||
#include "inode.h"
|
||
|
||
#include <mutex>
|
||
#include <queue>
|
||
|
||
namespace sdi_toolBox::desktop::eventBus
|
||
{
|
||
//--------------------------------------------------------------
|
||
/**
|
||
* @class Node
|
||
* @brief Concrete subscriber node in the event bus system.
|
||
*
|
||
* Node is the concrete implementation of @ref INode. It represents a participant
|
||
* in the event bus that can subscribe to specific message types or to broadcast
|
||
* mode, emit and post messages through the bus, and consume received messages
|
||
* from its internal FIFO queue.
|
||
*
|
||
* Each Node holds a reference to the @ref Bus it belongs to. Subscriptions and
|
||
* message transmissions are delegated to the bus. Incoming messages are stored
|
||
* in an internal thread-safe queue and can be retrieved via @ref popMessage().
|
||
*
|
||
* The Node also supports synchronous waiting: a thread can block on
|
||
* @ref syncWaitForMessage() until at least one message is available in the queue.
|
||
*
|
||
* On destruction, the Node automatically unsubscribes from all event types and
|
||
* broadcast mode, preventing dangling pointers in the bus routing table.
|
||
*
|
||
* @note Node is non-copyable and non-movable.
|
||
* @note A @ref Bus reference must be provided at construction time.
|
||
* @note The Node does not take ownership of the @ref Bus.
|
||
*
|
||
* @par Example usage:
|
||
* @code
|
||
* sdi_toolBox::desktop::eventBus::Bus bus;
|
||
* sdi_toolBox::desktop::eventBus::Node node(bus);
|
||
*
|
||
* node.subscribe(MY_EVENT_TYPE);
|
||
* node.emit<MyMessage>(42);
|
||
*
|
||
* node.syncWaitForMessage();
|
||
* auto msg = std::dynamic_pointer_cast<MyMessage>(node.popMessage());
|
||
* @endcode
|
||
*
|
||
* @see Bus
|
||
* @see INode
|
||
* @see Message
|
||
*/
|
||
class Node : public INode
|
||
{
|
||
public:
|
||
///@name Construction & Destruction
|
||
///@{
|
||
|
||
Node() = delete; ///< Default constructor - deleted. A @ref Bus reference must be provided.
|
||
|
||
/**
|
||
* @brief Destructor.
|
||
*
|
||
* Automatically unsubscribes the node from all specific event types and
|
||
* broadcast mode via @ref unsubscribeFromAll(), preventing dangling pointers
|
||
* in the bus routing table. Also notifies any thread blocked in
|
||
* @ref syncWaitForMessage() to unblock it gracefully.
|
||
*/
|
||
virtual ~Node();
|
||
|
||
/**
|
||
* @brief Copy constructor - deleted.
|
||
*
|
||
* Node is non-copyable.
|
||
*/
|
||
Node(const Node &obj) = delete;
|
||
|
||
/**
|
||
* @brief Move constructor - deleted.
|
||
*
|
||
* Node is non-movable.
|
||
*/
|
||
Node(Node &&obj) noexcept = delete;
|
||
|
||
/**
|
||
* @brief Copy assignment operator - deleted.
|
||
*
|
||
* Node is non-copyable.
|
||
*/
|
||
Node &operator=(const Node &obj) = delete;
|
||
|
||
/**
|
||
* @brief Move assignment operator - deleted.
|
||
*
|
||
* Node is non-movable.
|
||
*/
|
||
Node &operator=(Node &&obj) noexcept = delete;
|
||
|
||
/**
|
||
* @brief Construct a Node attached to the given @ref Bus.
|
||
*
|
||
* @param bus Reference to the @ref Bus this node belongs to.
|
||
* The bus must outlive the node.
|
||
*/
|
||
explicit Node(Bus &bus);
|
||
|
||
///@}
|
||
///@name Synchronization
|
||
///@{
|
||
|
||
/**
|
||
* @brief Block the calling thread until a message is received.
|
||
*
|
||
* Suspends the calling thread using an atomic wait until at least one message
|
||
* has been appended to the node's internal queue by the @ref Bus. This method
|
||
* is intended for synchronous event-driven patterns where a thread should idle
|
||
* until work is available.
|
||
*
|
||
* @note Returns immediately if a message is already pending in the queue
|
||
* at the time of the call.
|
||
* @note If the Node is destroyed while a thread is blocked here, the destructor
|
||
* triggers a notification to unblock the waiting thread gracefully.
|
||
* @warning The caller is responsible for checking the queue after this call
|
||
* returns, as the notification may also be triggered by the destructor
|
||
* with an empty queue.
|
||
*/
|
||
void syncWaitForMessage();
|
||
|
||
///@}
|
||
///@name Subscription Management
|
||
///@{
|
||
|
||
/**
|
||
* @brief Subscribe this node to a specific message type.
|
||
*
|
||
* Delegates to @ref Bus::subscribe(). The node will receive all messages
|
||
* of the given type posted to the bus. Duplicate subscriptions are ignored.
|
||
*
|
||
* @param eventType The message type identifier to subscribe to.
|
||
* @see Bus::subscribe()
|
||
*/
|
||
void subscribe(MessageTypeID eventType);
|
||
|
||
/**
|
||
* @brief Unsubscribe this node from a specific message type.
|
||
*
|
||
* Delegates to @ref Bus::unsubscribe(). If the node was not subscribed
|
||
* to the given type, this call has no effect.
|
||
*
|
||
* @param eventType The message type identifier to unsubscribe from.
|
||
* @see Bus::unsubscribe()
|
||
*/
|
||
void unsubscribe(MessageTypeID eventType);
|
||
|
||
/**
|
||
* @brief Unsubscribe this node from all message types and broadcast mode.
|
||
*
|
||
* Delegates to @ref Bus::unsubscribeFromAll(). After this call, the node
|
||
* will no longer receive any messages until it re-subscribes.
|
||
*
|
||
* @see Bus::unsubscribeFromAll()
|
||
*/
|
||
void unsubscribeFromAll();
|
||
|
||
///@}
|
||
///@name Message Transmission
|
||
///@{
|
||
|
||
/**
|
||
* @brief Construct and emit a message of type @p T through the bus.
|
||
*
|
||
* Forwards the call to @ref Bus::emit(). Creates a new message of type @p T
|
||
* using the provided arguments and posts it to the bus.
|
||
*
|
||
* @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.
|
||
*
|
||
* @see Bus::emit()
|
||
*/
|
||
template<class T, class... Args>
|
||
bool emit(Args &&...args);
|
||
|
||
/**
|
||
* @brief Post an already constructed message through the bus.
|
||
*
|
||
* Forwards the call to @ref Bus::post().
|
||
*
|
||
* @param message Shared pointer to the message to post. Must not be @c nullptr.
|
||
* @return @c true if at least one subscriber received the message,
|
||
* @c false otherwise.
|
||
*
|
||
* @see Bus::post()
|
||
*/
|
||
bool post(const std::shared_ptr<Message> &message) const;
|
||
|
||
/**
|
||
* @brief Notify the node that a message has been received.
|
||
*
|
||
* Sets the atomic waiting flag to @c true and triggers a wake-up for any
|
||
* thread blocked in @ref syncWaitForMessage(). Has no effect if the flag
|
||
* is already set.
|
||
*/
|
||
void messageNotify();
|
||
|
||
///@}
|
||
///@name Message queue management
|
||
///@{
|
||
|
||
/**
|
||
* @brief Get the number of messages currently in the node's queue.
|
||
*
|
||
* @return The number of pending messages waiting to be consumed.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
size_t getMessageCount() const;
|
||
|
||
/**
|
||
* @brief Remove and return the front message from the node's queue (FIFO).
|
||
*
|
||
* Retrieves the oldest message in the queue and removes it. If the queue
|
||
* is empty, returns @c nullptr.
|
||
*
|
||
* @return A shared pointer to the front @ref Message, or @c nullptr if the
|
||
* queue is empty.
|
||
* @note This operation is thread-safe.
|
||
*/
|
||
std::shared_ptr<Message> popMessage(); // Pop a message from the node's message queue (remove and return the front message)
|
||
|
||
///@}
|
||
|
||
private:
|
||
/**
|
||
* @brief Append a message to the node's internal queue.
|
||
*
|
||
* Called exclusively by @ref Bus when dispatching a message to this node.
|
||
* Pushes the message onto the queue and notifies any thread waiting in
|
||
* @ref syncWaitForMessage().
|
||
*
|
||
* @param message Shared pointer to the message being delivered.
|
||
*/
|
||
void append(const std::shared_ptr<Message> &message) override;
|
||
|
||
Bus &m_bus; ///< Reference to the event bus
|
||
std::atomic_bool m_nodeWaitingFlag{ false }; ///< Flag to indicate if the node is waiting for a message (used for synchronous waiting)
|
||
|
||
/// @brief Internal message queue, protected by a mutex for thread-safe access.
|
||
struct
|
||
{
|
||
mutable std::mutex mtx; ///< Mutex for thread-safe access to the message queue
|
||
std::queue<std::shared_ptr<Message>> messageQueue; ///< Queue of messages received by the node
|
||
} 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
|