queue.hpp

#pragma once
 
#include "../topic.hpp"
 
namespace LibXR
{
struct Topic::QueueBlock : public Topic::SuberBlock
{
  void* queue;  
  void (*fun)(MicrosecondTimestamp, void*,
              QueueBlock&);  
};
 
class Topic::QueuedSubscriber
{
 public:
  template <typename Data>
  QueuedSubscriber(const char* name, LockFreeQueue<Data>& queue, Domain* domain = nullptr)
      : QueuedSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)), queue)
  {
  }
 
  template <typename Data>
  QueuedSubscriber(const char* name, LockFreeQueue<Message<Data>>& queue,
                   Domain* domain = nullptr)
      : QueuedSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)), queue)
  {
  }
 
  template <typename Data>
  QueuedSubscriber(Topic topic, LockFreeQueue<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)
    {
      LockFreeQueue<Data>* queue = reinterpret_cast<LockFreeQueue<Data>*>(block.queue);
      (void)queue->Push(*reinterpret_cast<Data*>(payload_addr));
    };
 
    topic.block_->data_.subers.Add(*block_);
  }
 
  template <typename Data>
  QueuedSubscriber(Topic topic, LockFreeQueue<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)
    {
      LockFreeQueue<Message<Data>>* queue =
          reinterpret_cast<LockFreeQueue<Message<Data>>*>(block.queue);
      (void)queue->Push(
          Message<Data>{timestamp, *reinterpret_cast<Data*>(payload_addr)});
    };
 
    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