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