MessageQueue.h

Go to the documentation of this file.

#ifndef LINTEL_MESSAGEQUEUE_H
#define LINTEL_MESSAGEQUEUE_H
 
#include "lintel/core/models/ProcessName.h"
#include "lintel/features/base/msg/MessageQueueCore.h"
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <string>
 
#include <utility>
 
namespace lintel {
template<typename T> class MessageQueue
{
private:
  // variables
  std::string m_processName;
  std::string m_name;
  static constexpr std::size_t m_msgSize = sizeof(T);
  int32_t m_msgCount;
  MessageQueueCore m_messageQueue;
  std::shared_ptr<ProcessName> m_realProcessName;
  std::atomic<std::uint64_t> m_droppedMessages{ 0 };
 
public:
  static void removeOf(const std::string &messageQueueName) { MessageQueueCore::removeOf(messageQueueName); }
 
  [[nodiscard]] bool isOwner() const { return m_processName == m_realProcessName->getProcessName(); }
 
  [[nodiscard]] const std::string &getProcessName() const { return m_processName; }
 
  [[nodiscard]] const std::string &getName() const { return m_name; }
 
  [[nodiscard]] int32_t getMsgCount() const { return m_msgCount; }
 
  [[nodiscard]] int32_t numberMessagesOf() const { return m_messageQueue.numberMessagesOf(); }
 
  MessageQueue(std::string processName,
    std::string name,
    int32_t msgCount,
    std::shared_ptr<ProcessName> realProcessName)
    : m_processName(std::move(processName)), m_name(std::move(name)), m_msgCount(msgCount),
      m_messageQueue(m_name, msgCount, m_msgSize), m_realProcessName(std::move(realProcessName))
  {}
 
  void sendOf(const T &message) { m_messageQueue.sendOf(&message, m_msgSize); }
 
  bool trySendOf(const T &message) { return m_messageQueue.trySendOf(&message, m_msgSize); }
 
  bool sendOfWithTimeout(const T &message, const std::chrono::milliseconds &timeout)
  {
    const bool success = m_messageQueue.sendOfWithTimeout(&message, m_msgSize, timeout);
    if (!success) { m_droppedMessages.fetch_add(1u, std::memory_order_relaxed); }
    return success;
  }
 
  [[nodiscard]] std::uint64_t droppedMessagesOf() const { return m_droppedMessages.load(std::memory_order_relaxed); }
 
  T receiveOf()
  {
    T temp{};
    m_messageQueue.receiveOf(&temp, m_msgSize);
    return temp;
  }
 
  std::optional<T> tryReceiveOf()
  {
    T temp{};
    if (!m_messageQueue.tryReceiveOf(&temp, m_msgSize)) { return std::nullopt; }
    return std::make_optional(temp);
  }
 
  bool operator<(const MessageQueue &rhs) const
  {
    if (m_processName < rhs.m_processName) return true;
    if (rhs.m_processName < m_processName) return false;
    return m_name < rhs.m_name;
  }
 
  bool operator>(const MessageQueue &rhs) const { return rhs < *this; }
 
  bool operator<=(const MessageQueue &rhs) const { return !(rhs < *this); }
 
  bool operator>=(const MessageQueue &rhs) const { return !(*this < rhs); }
 
  bool operator==(const MessageQueue &rhs) const { return m_processName == rhs.m_processName && m_name == rhs.m_name; }
 
  bool operator!=(const MessageQueue &rhs) const { return !(*this == rhs); }
};
 
}// namespace lintel
#endif// LINTEL_MESSAGEQUEUE_H