apollo::cyber::Reader< MessageT > 模板类 参考

Reader subscribes a channel, it has two main functions: 更多...

#include <reader.h>

类 apollo::cyber::Reader< MessageT > 继承关系图:

apollo::cyber::Reader< MessageT > 的协作图:

Public 类型
using	BlockerPtr = std::unique_ptr< blocker::Blocker< MessageT > >
using	ReceiverPtr = std::shared_ptr< transport::Receiver< MessageT > >
using	ChangeConnection = typename service_discovery::Manager::ChangeConnection
using	Iterator = typename std::list< std::shared_ptr< MessageT > >::const_iterator

Public 成员函数
	Reader (const proto::RoleAttributes &role_attr, const CallbackFunc< MessageT > &reader_func=nullptr, uint32_t pending_queue_size=DEFAULT_PENDING_QUEUE_SIZE)
	Constructor a Reader object.
virtual	~Reader ()
bool	Init () override
	Init Reader
void	Shutdown () override
	Shutdown Reader
void	Observe () override
	Get All data that Blocker stores
void	ClearData () override
	Clear Blocker's data
bool	HasReceived () const override
	Query whether we have received data since last clear
bool	Empty () const override
	Query whether the Reader has data to be handled
double	GetDelaySec () const override
	Get time interval of since last receive message
uint32_t	PendingQueueSize () const override
	Get pending_queue_size configuration
virtual void	Enqueue (const std::shared_ptr< MessageT > &msg)
	Push msg to Blocker's PublishQueue
virtual void	SetHistoryDepth (const uint32_t &depth)
	Set Blocker's PublishQueue's capacity to depth
virtual uint32_t	GetHistoryDepth () const
	Get Blocker's PublishQueue's capacity
virtual std::shared_ptr< MessageT >	GetLatestObserved () const
	Get the latest message we Observe
virtual std::shared_ptr< MessageT >	GetOldestObserved () const
	Get the oldest message we Observe
virtual Iterator	Begin () const
	Get the begin iterator of ObserveQueue, used to traverse
virtual Iterator	End () const
	Get the end iterator of ObserveQueue, used to traverse
bool	HasWriter () override
	Is there is at least one writer publish the channel that we subscribes?
void	GetWriters (std::vector< proto::RoleAttributes > *writers) override
	Get all writers pushlish the channel we subscribes
Public 成员函数 继承自 apollo::cyber::ReaderBase
	ReaderBase (const proto::RoleAttributes &role_attr)
virtual	~ReaderBase ()
const std::string &	GetChannelName () const
	Get Reader's Channel name
uint64_t	ChannelId () const
	Get Reader's Channel id
const proto::QosProfile &	QosProfile () const
	Get qos profile.
bool	IsInit () const
	Query whether the Reader is initialized

Protected 属性
double	latest_recv_time_sec_ = -1.0
double	second_to_lastest_recv_time_sec_ = -1.0
uint32_t	pending_queue_size_
Protected 属性 继承自 apollo::cyber::ReaderBase
proto::RoleAttributes	role_attr_
std::atomic< bool >	init_

详细描述

template<typename MessageT>
class apollo::cyber::Reader< MessageT >

Reader subscribes a channel, it has two main functions:

1. You can pass a CallbackFunc to handle the message then it arrived
2. You can Observe messages that Blocker cached. Reader automatically push the message to Blocker's PublishQueue, and we can use Observe to fetch messages from PublishQueue to ObserveQueue. But, if you have set CallbackFunc, you can ignore this. One Reader uses one ChannelBuffer, the message we are handling is stored in ChannelBuffer Reader will Join the topology when init and Leave the topology when shutdown

   警告

   To save resource, ChannelBuffer has limited length, it's passed through the pending_queue_size param. pending_queue_size is default set to 1, So, If you handle slower than writer sending, older messages that are not handled will be lost. You can increase pending_queue_size to resolve this problem.

在文件 reader.h 第 69 行定义.

成员类型定义说明

BlockerPtr

template<typename MessageT >

using apollo::cyber::Reader< MessageT >::BlockerPtr = std::unique_ptr<blocker::Blocker<MessageT> >

在文件 reader.h 第 71 行定义.

ChangeConnection

template<typename MessageT >

using apollo::cyber::Reader< MessageT >::ChangeConnection = typename service_discovery::Manager::ChangeConnection

在文件 reader.h 第 73 行定义.

Iterator

template<typename MessageT >

using apollo::cyber::Reader< MessageT >::Iterator = typename std::list<std::shared_ptr<MessageT> >::const_iterator

在文件 reader.h 第 75 行定义.

ReceiverPtr

template<typename MessageT >

using apollo::cyber::Reader< MessageT >::ReceiverPtr = std::shared_ptr<transport::Receiver<MessageT> >

在文件 reader.h 第 72 行定义.

构造及析构函数说明

Reader()

template<typename MessageT >

apollo::cyber::Reader< MessageT >::Reader ( const proto::RoleAttributes &  role_attr, const CallbackFunc< MessageT > &  reader_func = nullptr, uint32_t  pending_queue_size = DEFAULT_PENDING_QUEUE_SIZE  )

explicit

Constructor a Reader object.

参数

role_attr	is a protobuf message RoleAttributes, which includes the channel name and other info.
reader_func	is the callback function, when the message is received.
pending_queue_size	is the max depth of message cache queue.

警告

the received messages is enqueue a queue,the queue's depth is pending_queue_size

在文件 reader.h 第 231 行定义.

    : ReaderBase(role_attr),
      pending_queue_size_(pending_queue_size),
      reader_func_(reader_func) {
  blocker_.reset(new blocker::Blocker<MessageT>(blocker::BlockerAttr(
      role_attr.qos_profile().depth(), role_attr.channel_name())));
}

~Reader()

template<typename MessageT >

apollo::cyber::Reader< MessageT >::~Reader ( )

virtual

在文件 reader.h 第 242 行定义.

                          {
  Shutdown();
}

成员函数说明

Begin()

template<typename MessageT >

virtual Iterator apollo::cyber::Reader< MessageT >::Begin ( ) const

inlinevirtual

Get the begin iterator of ObserveQueue, used to traverse

返回

Iterator begin iterator

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 185 行定义.

{ return blocker_->ObservedBegin(); }

ClearData()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::ClearData ( )

overridevirtual

Clear Blocker's data

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 411 行定义.

                                 {
  blocker_->ClearPublished();
  blocker_->ClearObserved();
}

Empty()

template<typename MessageT >

bool apollo::cyber::Reader< MessageT >::Empty ( ) const

overridevirtual

Query whether the Reader has data to be handled

返回

true if blocker is empty

false if blocker has data

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 379 行定义.

                                   {
  return blocker_->IsObservedEmpty();
}

End()

template<typename MessageT >

virtual Iterator apollo::cyber::Reader< MessageT >::End ( ) const

inlinevirtual

Get the end iterator of ObserveQueue, used to traverse

返回

Iterator begin iterator

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 192 行定义.

{ return blocker_->ObservedEnd(); }

Enqueue()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::Enqueue ( const std::shared_ptr< MessageT > &  msg )

virtual

Push msg to Blocker's PublishQueue

参数

msg	message ptr to be pushed

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 247 行定义.

                                                                 {
  second_to_lastest_recv_time_sec_ = latest_recv_time_sec_;
  latest_recv_time_sec_ = Time::Now().ToSecond();
  blocker_->Publish(msg);
}

GetDelaySec()

template<typename MessageT >

double apollo::cyber::Reader< MessageT >::GetDelaySec ( ) const

overridevirtual

Get time interval of since last receive message

返回

double seconds delay

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 384 行定义.

                                           {
  if (latest_recv_time_sec_ < 0) {
    return -1.0;
  }
  if (second_to_lastest_recv_time_sec_ < 0) {
    return Time::Now().ToSecond() - latest_recv_time_sec_;
  }
  return std::max((Time::Now().ToSecond() - latest_recv_time_sec_),
                  (latest_recv_time_sec_ - second_to_lastest_recv_time_sec_));
}

GetHistoryDepth()

template<typename MessageT >

uint32_t apollo::cyber::Reader< MessageT >::GetHistoryDepth ( ) const

virtual

Get Blocker's PublishQueue's capacity

返回

uint32_t depth of the history

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 422 行定义.

                                                 {
  return static_cast<uint32_t>(blocker_->capacity());
}

GetLatestObserved()

template<typename MessageT >

std::shared_ptr< MessageT > apollo::cyber::Reader< MessageT >::GetLatestObserved ( ) const

virtual

Get the latest message we Observe

返回

std::shared_ptr<MessageT> the latest message

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 401 行定义.

                                                                  {
  return blocker_->GetLatestObservedPtr();
}

GetOldestObserved()

template<typename MessageT >

std::shared_ptr< MessageT > apollo::cyber::Reader< MessageT >::GetOldestObserved ( ) const

virtual

Get the oldest message we Observe

返回

std::shared_ptr<MessageT> the oldest message

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 406 行定义.

                                                                  {
  return blocker_->GetOldestObservedPtr();
}

GetWriters()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::GetWriters ( std::vector< proto::RoleAttributes > *  writers )

overridevirtual

Get all writers pushlish the channel we subscribes

参数

writers

result vector of RoleAttributes

重载 apollo::cyber::ReaderBase .

在文件 reader.h 第 436 行定义.

                                                                         {
  if (writers == nullptr) {
    return;
  }
 
  if (!init_.load()) {
    return;
  }
 
  channel_manager_->GetWritersOfChannel(role_attr_.channel_name(), writers);
}

HasReceived()

template<typename MessageT >

bool apollo::cyber::Reader< MessageT >::HasReceived ( ) const

overridevirtual

Query whether we have received data since last clear

返回

true if the reader has received data

false if the reader has not received data

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 374 行定义.

                                         {
  return !blocker_->IsPublishedEmpty();
}

HasWriter()

template<typename MessageT >

bool apollo::cyber::Reader< MessageT >::HasWriter ( )

overridevirtual

Is there is at least one writer publish the channel that we subscribes?

返回

true if the channel has writer

false if the channel has no writer

重载 apollo::cyber::ReaderBase .

在文件 reader.h 第 427 行定义.

                                 {
  if (!init_.load()) {
    return false;
  }
 
  return channel_manager_->HasWriter(role_attr_.channel_name());
}

Init()

template<typename MessageT >

bool apollo::cyber::Reader< MessageT >::Init ( )

overridevirtual

Init Reader

返回

true if init successfully

false if init failed

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 259 行定义.

                            {
  if (init_.exchange(true)) {
    return true;
  }
  auto statistics_center = statistics::Statistics::Instance();
  if (!statistics_center->RegisterChanVar(role_attr_)) {
    AWARN << "Failed to register reader var!";
  }
  std::function<void(const std::shared_ptr<MessageT>&)> func;
  if (reader_func_ != nullptr) {
    func = [this](const std::shared_ptr<MessageT>& msg) {
      uint64_t process_start_time;
      uint64_t proc_done_time;
      uint64_t proc_start_time;
 
      this->Enqueue(msg);
      this->reader_func_(msg);
      // sampling proc latency in microsecond
      proc_done_time = Time::Now().ToMicrosecond();
      proc_start_time =
          static_cast<uint64_t>(latest_recv_time_sec_ * 1000000UL);
 
      statistics::Statistics::Instance()->SamplingProcLatency<uint64_t>(
          this->role_attr_, (proc_done_time - proc_start_time));
      if (statistics::Statistics::Instance()->GetProcStatus(
              this->role_attr_, &process_start_time)) {
        auto cyber_latency = proc_start_time - process_start_time;
        if (process_start_time > 0 && cyber_latency > 0) {
          statistics::Statistics::Instance()->SamplingCyberLatency(
              this->role_attr_, cyber_latency);
        }
      }
    };
  } else {
    func = [this](const std::shared_ptr<MessageT>& msg) { this->Enqueue(msg); };
  }
  auto sched = scheduler::Instance();
  croutine_name_ = role_attr_.node_name() + "_" + role_attr_.channel_name();
  auto dv = std::make_shared<data::DataVisitor<MessageT>>(
      role_attr_.channel_id(), pending_queue_size_);
  // Using factory to wrap templates.
  croutine::RoutineFactory factory =
      croutine::CreateRoutineFactory<MessageT>(std::move(func), dv);
  if (!sched->CreateTask(factory, croutine_name_)) {
    AERROR << "Create Task Failed!";
    init_.store(false);
    return false;
  }
 
  receiver_ = ReceiverManager<MessageT>::Instance()->GetReceiver(role_attr_);
  this->role_attr_.set_id(receiver_->id().HashValue());
  channel_manager_ =
      service_discovery::TopologyManager::Instance()->channel_manager();
  JoinTheTopology();
 
  return true;
}

Observe()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::Observe ( )

overridevirtual

Get All data that Blocker stores

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 254 行定义.

                               {
  blocker_->Observe();
}

PendingQueueSize()

template<typename MessageT >

uint32_t apollo::cyber::Reader< MessageT >::PendingQueueSize ( ) const

overridevirtual

Get pending_queue_size configuration

返回

uint32_t the value of pending queue size

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 396 行定义.

                                                  {
  return pending_queue_size_;
}

SetHistoryDepth()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::SetHistoryDepth ( const uint32_t &  depth )

virtual

Set Blocker's PublishQueue's capacity to depth

参数

depth

the value you want to set

被 apollo::cyber::blocker::IntraReader< MessageT > 重载.

在文件 reader.h 第 417 行定义.

                                                            {
  blocker_->set_capacity(depth);
}

Shutdown()

template<typename MessageT >

void apollo::cyber::Reader< MessageT >::Shutdown ( )

overridevirtual

Shutdown Reader

实现了 apollo::cyber::ReaderBase.

在文件 reader.h 第 318 行定义.

                                {
  if (!init_.exchange(false)) {
    return;
  }
  LeaveTheTopology();
  receiver_ = nullptr;
  channel_manager_ = nullptr;
 
  if (!croutine_name_.empty()) {
    scheduler::Instance()->RemoveTask(croutine_name_);
  }
}

类成员变量说明

latest_recv_time_sec_

template<typename MessageT >

double apollo::cyber::Reader< MessageT >::latest_recv_time_sec_ = -1.0

protected

在文件 reader.h 第 211 行定义.

pending_queue_size_

template<typename MessageT >

uint32_t apollo::cyber::Reader< MessageT >::pending_queue_size_

protected

在文件 reader.h 第 213 行定义.

second_to_lastest_recv_time_sec_

template<typename MessageT >

double apollo::cyber::Reader< MessageT >::second_to_lastest_recv_time_sec_ = -1.0

protected

在文件 reader.h 第 212 行定义.

---

该类的文档由以下文件生成:

• cyber/node/reader.h