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 <core/utils/time/time.hpp>
 
#include <ccf/mptk/message.hpp>
#include <ccf/mptk/response.hpp>
 
 
// TODO: Dummy implementation. Should use C++20 counting semaphore.
class TmpCountingSem {
  public:
  
    TmpCountingSem() {
      m_count = 0;
    };
 
    ~TmpCountingSem() {
      // Release all blocked on the semaphore?
    };
 
    void release() {
      m_count_mutex.lock(); {
          m_count++;
      } m_count_mutex.unlock();
    };
 
    void acquire() {
      m_acquire_mutex.lock(); {
 
        while (m_count == 0) {
          core::utils::time::Sleep(0.0001);
        }
 
        m_count_mutex.lock(); {
          m_count--;
        } m_count_mutex.unlock();
 
      } m_acquire_mutex.unlock();
    };
 
    bool try_acquire_for(const double time) {
      m_acquire_mutex.lock(); {
 
        double start_time = core::utils::time::Time();
        while ((m_count == 0) && ((core::utils::time::Time() - start_time) < time)) {
          core::utils::time::Sleep(0.0001);
        }
        if ((core::utils::time::Time() - start_time) >= time) {
          m_acquire_mutex.unlock();
          return false;
        }
 
        m_count_mutex.lock(); {
          m_count--;
        } m_count_mutex.unlock();
 
      } m_acquire_mutex.unlock();
 
      return true;
    };
 
  private:
    std::mutex m_acquire_mutex;
    std::mutex m_count_mutex;
    int32_t m_count;
};
 
 
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_name);
 
    bool ThreadRegistered(const std::string& thread_name) const;
 
    void SendMessage(const Message& message);
 
    bool ReceiveMessage(const std::string& receiver_thread_name,
                        const double timeout,
                        bool& timed_out,
                        Message& message);
 
    void SendResponse(Response& response);
 
    bool ReceiveResponse(const std::string& msg_sender_thread_name,
                         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.
    // boost: std::map<std::string, boost::interprocess::interprocess_semaphore*> m_message_registry_sem;
    std::map<std::string, TmpCountingSem*> 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.
    // boost: std::map<std::string, boost::interprocess::interprocess_semaphore*> m_response_registry_sem;
    std::map<std::string, TmpCountingSem*> 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_