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