queue.hpp

#pragma once
 
#include "../topic.hpp"
 
namespace LibXR
{
struct Topic::QueueBlock : public Topic::SuberBlock
{
  SPSCQueueBase* queue;  
  void (*fun)(MicrosecondTimestamp, void*,
              QueueBlock&);  
};
 
class Topic::QueuedSubscriber
{
 public:
  template <typename Data>
  QueuedSubscriber(const char* name, SPSCQueue<Data>& queue, Domain* domain = nullptr)
      : QueuedSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)), queue)
  {
  }
 
  template <typename Data>
  QueuedSubscriber(const char* name, SPSCQueue<Message<Data>>& queue,
                   Domain* domain = nullptr)
      : QueuedSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)), queue)
  {
  }
 
  template <typename Data>
  QueuedSubscriber(Topic topic, SPSCQueue<Data>& queue)
  {
    Topic::CheckSubscriberType<Data>(topic);
 
    block_ = new LockFreeList::Node<QueueBlock>;
    block_->data_.type = SuberType::QUEUE;
    block_->data_.queue = &queue;
    block_->data_.fun = [](MicrosecondTimestamp, void* payload_addr, QueueBlock& block)
    { (void)block.queue->PushBytes(payload_addr); };
 
    topic.block_->data_.subers.Add(*block_);
  }
 
  template <typename Data>
  QueuedSubscriber(Topic topic, SPSCQueue<Message<Data>>& queue)
  {
    Topic::CheckSubscriberType<Data>(topic);
 
    block_ = new LockFreeList::Node<QueueBlock>;
    block_->data_.type = SuberType::QUEUE;
    block_->data_.queue = &queue;
    block_->data_.fun =
        [](MicrosecondTimestamp timestamp, void* payload_addr, QueueBlock& block)
    {
      Message<Data> message{timestamp, *reinterpret_cast<Data*>(payload_addr)};
      (void)block.queue->PushBytes(&message);
    };
 
    topic.block_->data_.subers.Add(*block_);
  }
 
  QueuedSubscriber(const QueuedSubscriber& other) = delete;
 
  QueuedSubscriber& operator=(const QueuedSubscriber& other) = delete;
 
  QueuedSubscriber(QueuedSubscriber&& other) noexcept : block_(other.block_)
  {
    other.block_ = nullptr;
  }
 
  QueuedSubscriber& operator=(QueuedSubscriber&& other) noexcept
  {
    if (this != &other)
    {
      block_ = other.block_;
      other.block_ = nullptr;
    }
    return *this;
  }
 
 private:
  LockFreeList::Node<QueueBlock>* block_ =
      nullptr;  
};
}  // namespace LibXR