将字节流解析成 packet 并发布到已注册 topic 的状态机 / State machine that parses byte streams into packets and publishes them into registered topics More...

#include <server.hpp>

Collaboration diagram for LibXR::Topic::Server:

[legend]

Public Types
enum class	Status : uint8_t { WAIT_START , WAIT_TOPIC , WAIT_DATA_CRC }
	parser 当前所在阶段 / Current stage of the parser More...

Public Member Functions
	Server (size_t buffer_length)
	构造 parser 并分配内部暂存队列 / Construct the parser and allocate its internal staging queue

void	Register (TopicHandle topic)
	注册一个可接收 packet 的 topic / Register one topic that may receive parsed packets

size_t	ParseData (ConstRawData data)
	在普通上下文里喂入一批新字节 / Feed one new byte batch in normal context

size_t	ParseDataFromCallback (ConstRawData data, bool in_isr)
	在回调/ISR 路径里喂入一批新字节 / Feed one new byte batch in callback/ISR path

Private Types
enum class	ParseResult : uint8_t { NEED_MORE , DROPPED , DELIVERED }
	一次 payload 阶段处理结果 / Result of one payload-stage handling step More...

Private Member Functions
size_t	ParseDataRaw (ConstRawData data, bool from_callback, bool in_isr)
	`ParseData()` 的共享实现 / Shared implementation behind `ParseData()`

bool	SyncToPacketStart ()
	把输入流同步到下一条 packet 起点 / Synchronize the input stream to the next packet start

bool	ReadHeader ()
	在已对齐前缀后继续读取并校验完整头部 / Read and validate the full header after the prefix is aligned

ParseResult	ReadPayload (bool from_callback, bool in_isr)
	读取当前包的 payload 和尾 CRC，并在成功时发布 / Read the payload and trailing CRC of the current packet and publish it on success

void	ResetParser ()
	清空当前包的解析上下文并回到找起点状态 / Clear the current packet parsing context and return to the start-search state

Private Attributes
Status	status_ = Status::WAIT_START
	当前 parser 阶段。Current parser stage.

uint32_t	data_len_
	当前包头声明的 payload 长度。Payload length declared by the current header.

RBTree< uint32_t >	topic_map_

QueueBase	queue_
	输入字节 FIFO。Input byte FIFO.

RawData	parse_buff_

TopicHandle	current_topic_
	当前包命中的目标 topic。Target topic matched by the current packet.

MicrosecondTimestamp	current_timestamp_

Detailed Description

将字节流解析成 packet 并发布到已注册 topic 的状态机 / State machine that parses byte streams into packets and publishes them into registered topics

把字节流解析成 packet 并投递到 topic 的 parser / Parser that turns byte streams into packets and delivers them into topics

Note: 当前 Server 自己不维护 raw topic/cache 语义；它只依赖注册 topic 的： payload_size、payload_alignment 和名字 CRC 键。 The current Server does not restore old raw topic/cache semantics by itself; it relies only on each registered topic's payload_size, payload_alignment, and name CRC key.; 当前兼容规则：收到的 packet payload 短于 topic 固定大小时，仅保证前缀部分有效，后半段保持未定义；长于 topic 固定大小时，仅保留前缀部分，其余字节直接截断。 Current compatibility rule: when an incoming packet payload is shorter than the topic's fixed size, only the leading prefix is guaranteed valid and the remaining tail stays unspecified; when it is longer, only the prefix matching the topic size is kept and the rest is truncated.

Definition at line 26 of file server.hpp.

Member Enumeration Documentation

◆ ParseResult

enum class LibXR::Topic::Server::ParseResult : uint8_t

strongprivate

一次 payload 阶段处理结果 / Result of one payload-stage handling step

Enumerator
NEED_MORE	当前包还没收全。Current packet is still incomplete.
DROPPED	当前包被丢弃。Current packet is dropped.
DELIVERED	当前包已发布。Current packet is delivered.

Definition at line 87 of file server.hpp.

  {
    NEED_MORE,  
    DROPPED,    
    DELIVERED   
  };

◆ Status

enum class LibXR::Topic::Server::Status : uint8_t

strong

parser 当前所在阶段 / Current stage of the parser

Enumerator
WAIT_START	正在找下一包前缀。Searching for the next packet prefix.
WAIT_TOPIC	已读到前缀，正在等完整头。Prefix received; waiting for the full header.
WAIT_DATA_CRC	头已接受，正在等 payload 和尾 CRC。Header accepted; waiting for payload plus trailing CRC.

Definition at line 33 of file server.hpp.

  {
    WAIT_START,     
    WAIT_TOPIC,     
    WAIT_DATA_CRC,  
  };

Constructor & Destructor Documentation

◆ Server()

Topic::Server::Server ( size_t buffer_length )

构造 parser 并分配内部暂存队列 / Construct the parser and allocate its internal staging queue

Parameters

buffer_length 内部字节队列和暂存缓冲区大小 / Size of the internal byte queue and staging buffer

Definition at line 12 of file server.cpp.

    : topic_map_([](const uint32_t& a, const uint32_t& b) { return (a > b) - (a < b); }),
      queue_(1, buffer_length)
{
  ASSERT(buffer_length > PACK_BASE_SIZE);
  parse_buff_.size_ = buffer_length;
  parse_buff_.addr_ =
      new (std::align_val_t(LibXR::CACHE_LINE_SIZE)) uint8_t[parse_buff_.size_];
}

Member Function Documentation

◆ ParseData()

size_t Topic::Server::ParseData ( ConstRawData data )

在普通上下文里喂入一批新字节 / Feed one new byte batch in normal context

Parameters

data	新收到的原始字节 / Newly received raw bytes

Returns: 成功解析并发布的包数量 / Number of packets parsed and published

Definition at line 35 of file server.cpp.

{
  return ParseDataRaw(data, false, false);
}

◆ ParseDataFromCallback()

size_t Topic::Server::ParseDataFromCallback	(	ConstRawData	data,
		bool	in_isr )

在回调/ISR 路径里喂入一批新字节 / Feed one new byte batch in callback/ISR path

Parameters

data	新收到的原始字节 / Newly received raw bytes
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Returns: 成功解析并发布的包数量 / Number of packets parsed and published

Definition at line 40 of file server.cpp.

{
  return ParseDataRaw(data, true, in_isr);
}

◆ ParseDataRaw()

size_t Topic::Server::ParseDataRaw	(	ConstRawData	data,
		bool	from_callback,
		bool	in_isr )

private

ParseData*() 的共享实现 / Shared implementation behind ParseData*()

Parameters

data	新收到的原始字节 / Newly received raw bytes
from_callback	是否来自回调路径 / Whether the current parse comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Returns: 成功解析并发布的包数量 / Number of packets parsed and published

Definition at line 45 of file server.cpp.

{
  size_t count = 0;
 
  (void)queue_.PushBatchBytes(data.addr_, data.size_);
 
  while (true)
  {
    if (status_ == Status::WAIT_START && !SyncToPacketStart())
    {
      return count;
    }
 
    if (status_ == Status::WAIT_TOPIC && !ReadHeader())
    {
      return count;
    }
 
    if (status_ == Status::WAIT_DATA_CRC)
    {
      switch (ReadPayload(from_callback, in_isr))
      {
        case ParseResult::NEED_MORE:
          return count;
        case ParseResult::DROPPED:
          continue;
        case ParseResult::DELIVERED:
          count++;
          continue;
      }
    }
  }
}

◆ ReadHeader()

bool Topic::Server::ReadHeader ( )

private

在已对齐前缀后继续读取并校验完整头部 / Read and validate the full header after the prefix is aligned

Returns: 头部有效返回 true，否则返回 false / Returns true when the header is valid, otherwise false

Definition at line 97 of file server.cpp.

{
  if (queue_.Size() < sizeof(PackedDataHeader))
  {
    return false;
  }
 
  queue_.PopBatchBytes(parse_buff_.addr_, sizeof(PackedDataHeader));
  if (!CRC8::Verify(parse_buff_.addr_, sizeof(PackedDataHeader)))
  {
    ResetParser();
    return true;
  }
 
  auto* header = reinterpret_cast<PackedDataHeader*>(parse_buff_.addr_);
  if (header->version != PACKET_VERSION)
  {
    ResetParser();
    return true;
  }
 
  auto* node = topic_map_.Search<TopicHandle>(header->topic_name_crc32);
  if (node == nullptr)
  {
    ResetParser();
    return true;
  }
 
  data_len_ = header->GetDataLen();
  current_timestamp_ = header->GetTimestamp();
  current_topic_ = *node;
  const auto target_size = current_topic_->data_.payload_size;
 
  if (target_size + PACK_BASE_SIZE > parse_buff_.size_)
  {
    ResetParser();
    return true;
  }
 
  if (data_len_ + PACK_BASE_SIZE > queue_.length_)
  {
    ResetParser();
    return true;
  }
 
  status_ = Status::WAIT_DATA_CRC;
  return true;
}

◆ ReadPayload()

Topic::Server::ParseResult Topic::Server::ReadPayload	(	bool	from_callback,
		bool	in_isr )

private

读取当前包的 payload 和尾 CRC，并在成功时发布 / Read the payload and trailing CRC of the current packet and publish it on success

Parameters

from_callback	是否来自回调路径 / Whether the current parse comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Returns: 当前 payload 阶段的处理结果 / Result of the current payload stage

Definition at line 146 of file server.cpp.

{
  if (queue_.Size() < data_len_ + sizeof(uint8_t))
  {
    return ParseResult::NEED_MORE;
  }
 
  auto* payload_addr =
      reinterpret_cast<uint8_t*>(parse_buff_.addr_) + sizeof(PackedDataHeader);
  queue_.PopBatchBytes(payload_addr, data_len_ + sizeof(uint8_t));
 
  if (!CRC8::Verify(parse_buff_.addr_,
                    data_len_ + sizeof(PackedDataHeader) + sizeof(uint8_t)))
  {
    ResetParser();
    return ParseResult::DROPPED;
  }
 
  const auto target_size = current_topic_->data_.payload_size;
  void* publish_addr = payload_addr;
  if (reinterpret_cast<uintptr_t>(payload_addr) %
          current_topic_->data_.payload_alignment !=
      0)
  {
    publish_addr = parse_buff_.addr_;
    if (data_len_ >= target_size)
    {
      LibXR::Memory::FastMove(publish_addr, payload_addr, target_size);
    }
    else
    {
      LibXR::Memory::FastMove(publish_addr, payload_addr, data_len_);
    }
  }
 
  auto topic = Topic(current_topic_);
  if (from_callback)
  {
    topic.PublishBytesFromServerCallback(publish_addr, target_size, current_timestamp_,
                                         in_isr);
  }
  else
  {
    topic.PublishBytesFromServer(publish_addr, target_size, current_timestamp_);
  }
 
  ResetParser();
  return ParseResult::DELIVERED;
}

◆ Register()

void Topic::Server::Register ( TopicHandle topic )

注册一个可接收 packet 的 topic / Register one topic that may receive parsed packets

Parameters

topic 目标 topic 句柄 / Target topic handle

Note: 注册时会断言该 topic 的 payload_size + PACK_BASE_SIZE 能放进本 server 的暂存缓冲区。 Registration asserts that the topic's payload_size + PACK_BASE_SIZE fits in this server's staging buffer.; server 的内部暂存缓冲区在构造时固定按 CACHE_LINE_SIZE 对齐分配；注册时还会断言该 topic 的 payload_alignment <= CACHE_LINE_SIZE。 The server's internal staging buffer is allocated once at CACHE_LINE_SIZE alignment during construction; registration also asserts that the topic's payload_alignment <= CACHE_LINE_SIZE.

Definition at line 22 of file server.cpp.

{
  ASSERT(topic != nullptr);
  ASSERT(topic->data_.payload_size != 0);
  ASSERT(topic->data_.payload_alignment != 0);
  ASSERT(topic->data_.payload_alignment <= LibXR::CACHE_LINE_SIZE);
 
  ASSERT(topic->data_.payload_size + PACK_BASE_SIZE <= parse_buff_.size_);
 
  auto* node = new RBTree<uint32_t>::Node<TopicHandle>(topic);
  topic_map_.Insert(*node, topic->key);
}

◆ ResetParser()

void Topic::Server::ResetParser ( )

private

清空当前包的解析上下文并回到找起点状态 / Clear the current packet parsing context and return to the start-search state

Definition at line 196 of file server.cpp.

{
  status_ = Status::WAIT_START;
  data_len_ = 0;
  current_topic_ = nullptr;
  current_timestamp_ = MicrosecondTimestamp();
}

◆ SyncToPacketStart()

bool Topic::Server::SyncToPacketStart ( )

private

把输入流同步到下一条 packet 起点 / Synchronize the input stream to the next packet start

Returns: 若已找到起始字节则返回 true，否则返回 false / Returns true when a packet start byte is found, otherwise false

Definition at line 79 of file server.cpp.

{
  auto queue_size = queue_.Size();
  for (uint32_t i = 0; i < queue_size; i++)
  {
    uint8_t prefix = 0;
    queue_.PeekBytes(&prefix);
    if (prefix == PACKET_PREFIX)
    {
      status_ = Status::WAIT_TOPIC;
      return true;
    }
    queue_.PopBytes();
  }
 
  return false;
}

Field Documentation

◆ current_timestamp_

MicrosecondTimestamp LibXR::Topic::Server::current_timestamp_

private

当前包头里的时间戳。Timestamp carried by the current packet header.

Definition at line 146 of file server.hpp.

◆ current_topic_

TopicHandle LibXR::Topic::Server::current_topic_

private

Initial value:

=

nullptr

当前包命中的目标 topic。Target topic matched by the current packet.

Definition at line 144 of file server.hpp.

◆ data_len_

uint32_t LibXR::Topic::Server::data_len_

private

Initial value:

=

0

当前包头声明的 payload 长度。Payload length declared by the current header.

Definition at line 137 of file server.hpp.

◆ parse_buff_

RawData LibXR::Topic::Server::parse_buff_

private

当前包头和 payload 的暂存缓冲区。Staging buffer holding the current header and payload.

Definition at line 142 of file server.hpp.

◆ queue_

QueueBase LibXR::Topic::Server::queue_

private

输入字节 FIFO。Input byte FIFO.

Definition at line 141 of file server.hpp.

◆ status_

Status LibXR::Topic::Server::status_ = Status::WAIT_START

private

当前 parser 阶段。Current parser stage.

Definition at line 136 of file server.hpp.

◆ topic_map_

RBTree<uint32_t> LibXR::Topic::Server::topic_map_

private

从 topic 名称 CRC32 到 topic 句柄的映射。Map from topic-name CRC32 to topic handle.

Definition at line 139 of file server.hpp.

The documentation for this class was generated from the following files:

src/middleware/message/server/server.hpp
src/middleware/message/server/server.cpp

Public Types

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Enumeration Documentation

◆ ParseResult

◆ Status

Constructor & Destructor Documentation

◆ Server()

Member Function Documentation

◆ ParseData()

◆ ParseDataFromCallback()

◆ ParseDataRaw()

◆ ReadHeader()

◆ ReadPayload()

◆ Register()

◆ ResetParser()

◆ SyncToPacketStart()

Field Documentation

◆ current_timestamp_

◆ current_topic_

◆ data_len_

◆ parse_buff_

◆ queue_

◆ status_

◆ topic_map_