messageBus.hpp

Go to the documentation of this file.

 
#ifndef CCF_MPTK_MESSAGE_BUS_HPP_
#define CCF_MPTK_MESSAGE_BUS_HPP_
 
#include <string>
#include <queue>
#include <iostream>
 
// TODO: Use C++20 std::counting_semaphore, when available.
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
 
#include <core/utils/base/base.hpp>
 
#include <ccf/mptk/message.hpp>
#include <ccf/mptk/response.hpp>
 
 
namespace ccf::mptk {
 
  void SplitDblTime(const double time,
                    int64_t& secs,
                    int64_t& nano_secs);
 
  class MessageBus {
  public:
 
    MessageBus();
 
    ~MessageBus();
 
    void Reset();
 
    MessageBus& RegisterThread(const std::string& thread_id);
 
    bool ThreadRegistered(const std::string& thread_id) const;
 
    void SendMessage(const Message& message);
 
    bool ReceiveMessage(const std::string& receiver_thread_id,
                        const double timeout,
                        bool& timed_out,
                        Message& message);
 
    void SendResponse(Response& response);
 
    bool ReceiveResponse(const std::string& msg_sender_thread_id,
                         const double timeout,
                         bool& timed_out,
                         Response& response);
 
    bool ReceiveResponseByMsgId(const std::string& msg_id,
                                const double timeout,
                                bool& timed_out,
                                Response& response);
 
    std::string ToString() const;
 
  private:
    std::string m_id; // Message Bus ID.
 
    // Message Receiver Thread ID to message objects list.
    std::map<std::string, std::list<Message>> m_message_registry;
    // Message Receiver Registry counting semaphores.
    std::map<std::string, boost::interprocess::interprocess_semaphore*> m_message_registry_sem;
 
    // Response receiver (= sender of msg) Thread ID to response objects list.
    std::map<std::string, std::list<Response>> m_response_registry;
    // Response Receiver Registry counting semaphores.
    std::map<std::string, boost::interprocess::interprocess_semaphore*> m_response_registry_sem;
 
    // Message ID to response objects list.
    std::map<std::string, Response*> m_response_msg_id_registry;
 
    std::vector<std::string> m_threads;
  };
 
}
 
#endif // CCF_MPTK_MESSAGE_BUS_HPP_