async.hpp

#pragma once
 
#include "../topic.hpp"
 
namespace LibXR
{
enum class Topic::ASyncSubscriberState : uint32_t
{
  IDLE = 0,               
  WAITING = 1,            
  DATA_READY = UINT32_MAX 
};
 
struct Topic::ASyncBlock : public Topic::SuberBlock
{
  void* buff_addr;                 
  void (*copy_payload)(void* dst,
                       void* payload_addr);    
  MicrosecondTimestamp timestamp;  
  std::atomic<ASyncSubscriberState> state =
      ASyncSubscriberState::IDLE;  
};
 
template <typename Data>
class Topic::ASyncSubscriber
{
 public:
  ASyncSubscriber(const char* name, Domain* domain = nullptr)
      : ASyncSubscriber(Topic(WaitTopic(name, UINT32_MAX, domain)))
  {
  }
 
  ASyncSubscriber(Topic topic)
  {
    Topic::CheckSubscriberType<Data>(topic);
 
    block_ = new LockFreeList::Node<ASyncBlock>;
    block_->data_.type = SuberType::ASYNC;
    block_->data_.timestamp = MicrosecondTimestamp();
    block_->data_.buff_addr = Topic::AllocateSubscriberBuffer<Data>();
    block_->data_.copy_payload = &Topic::CopyPayload<Data>;
    topic.block_->data_.subers.Add(*block_);
  }
 
  ASyncSubscriber(const ASyncSubscriber& other) = delete;
 
  ASyncSubscriber& operator=(const ASyncSubscriber& other) = delete;
 
  ASyncSubscriber(ASyncSubscriber&& other) noexcept : block_(other.block_)
  {
    other.block_ = nullptr;
  }
 
  ASyncSubscriber& operator=(ASyncSubscriber&& other) noexcept
  {
    if (this != &other)
    {
      block_ = other.block_;
      other.block_ = nullptr;
    }
    return *this;
  }
 
  bool Available()
  {
    return block_->data_.state.load(std::memory_order_acquire) ==
           ASyncSubscriberState::DATA_READY;
  }
 
  Data& GetData()
  {
    if (block_->data_.state.load(std::memory_order_acquire) ==
        ASyncSubscriberState::DATA_READY)
    {
      block_->data_.state.store(ASyncSubscriberState::IDLE, std::memory_order_release);
    }
    return *reinterpret_cast<Data*>(block_->data_.buff_addr);
  }
 
  MicrosecondTimestamp GetTimestamp() const { return block_->data_.timestamp; }
 
  void StartWaiting()
  {
    if (block_->data_.state.load(std::memory_order_acquire) ==
        ASyncSubscriberState::IDLE)
    {
      block_->data_.state.store(ASyncSubscriberState::WAITING,
                                std::memory_order_release);
    }
  }
 
  LockFreeList::Node<ASyncBlock>* block_ = nullptr;  
};
}  // namespace LibXR