主题（Topic）管理类 / Topic management class More...

#include <message.hpp>

Collaboration diagram for LibXR::Topic:

Data Structures
struct	ASyncBlock
	异步订阅块，继承自 SuberBlock Asynchronous subscription block, inheriting from SuberBlock More...

class	ASyncSubscriber
	异步订阅者类，用于订阅异步数据 Asynchronous subscriber class for subscribing to asynchronous data More...

struct	Block
	存储主题运行状态和静态配置。Stores topic runtime state and static configuration. More...

class	Callback

struct	CallbackBlock
	回调订阅块，继承自 SuberBlock Callback subscription block, inheriting from SuberBlock More...

class	Domain
	主题域（Domain）管理器，用于组织多个主题。Domain manager for organizing multiple topics. More...

struct	Message
	带时间戳的类型化消息。Timestamped typed message. More...

struct	MessageView
	带时间戳的类型化消息视图。Timestamped typed message view. More...

struct	QueueBlock
	队列订阅块，继承自 SuberBlock Queue subscription block, inheriting from SuberBlock More...

class	QueuedSubscriber

class	Server
	服务器类，负责解析数据并将其分发到相应的主题 Server class responsible for parsing data and distributing it to corresponding topics More...

struct	SuberBlock
	订阅者信息存储结构。Structure storing subscriber information. More...

struct	SyncBlock
	同步订阅者存储结构。Structure storing synchronous subscriber data. More...

class	SyncSubscriber
	同步订阅者类，允许同步方式接收数据。Synchronous subscriber class allowing data reception in a synchronous manner. More...

Public Types
enum class	SuberType : uint8_t { SYNC , ASYNC , QUEUE , CALLBACK }
	订阅者类型。Subscriber type. More...

enum class	ASyncSubscriberState : uint32_t { IDLE = 0 , WAITING = 1 , DATA_READY = UINT32_MAX }

typedef RBTree< uint32_t >::Node< Block > *	TopicHandle
	主题句柄，指向存储数据的红黑树节点。Handle pointing to a red-black tree node storing data.

Public Member Functions
void	RegisterCallback (Callback &cb)
	注册回调函数 Registers a callback function

	Topic ()
	默认构造函数，创建一个空的 Topic 实例 Default constructor, creates an empty Topic instance

	Topic (const char name, uint32_t max_length, Domain domain=nullptr, bool multi_publisher=false, bool cache=false, bool check_length=false)
	构造函数，使用指定名称、最大长度、域及其他选项初始化主题 Constructor to initialize a topic with the specified name, maximum length, domain, and options

	Topic (TopicHandle topic)
	通过句柄构造主题 Constructs a topic from a topic handle

void	EnableCache ()
	启用主题的缓存功能 Enables caching for the topic

template<typename Data >
void	Publish (Data &data)
	发布数据 Publishes data

template<typename Data >
void	Publish (Data &data, MicrosecondTimestamp timestamp)

void	Publish (void *addr, uint32_t size)
	以原始地址和大小发布数据 Publishes data using raw address and size

void	Publish (void *addr, uint32_t size, MicrosecondTimestamp timestamp)

template<typename Data >
void	PublishFromCallback (Data &data, bool in_isr)
	在回调函数中发布数据 Publishes data in a callback

template<typename Data >
void	PublishFromCallback (Data &data, MicrosecondTimestamp timestamp, bool in_isr)

void	PublishFromCallback (void *addr, uint32_t size, bool in_isr)
	在回调函数中以原始地址和大小发布数据 Publishes data using raw address and size in a callback

void	PublishFromCallback (void *addr, uint32_t size, MicrosecondTimestamp timestamp, bool in_isr)

template<typename Data >
ErrorCode	DumpData (PackedData< Data > &data)
	转储数据到 PackedData Dumps data into PackedData format

template<SizeLimitMode Mode = SizeLimitMode::MORE>
ErrorCode	DumpData (RawData data)
	转储数据到原始缓冲区。Dumps data into a raw buffer.

template<SizeLimitMode Mode = SizeLimitMode::MORE>
ErrorCode	DumpData (RawData data, MicrosecondTimestamp &timestamp)

template<typename Data > requires (!std::same_as<std::remove_cv_t<Data>, RawData>)
ErrorCode	DumpData (Data &data)
	转储数据到普通数据结构 Dumps data into a normal data structure

template<typename Data > requires (!std::same_as<std::remove_cv_t<Data>, RawData>)
ErrorCode	DumpData (Data &data, MicrosecondTimestamp &timestamp)

MicrosecondTimestamp	GetTimestamp () const

	operator TopicHandle ()
	将 Topic 转换为 TopicHandle Converts Topic to TopicHandle

uint32_t	GetKey () const
	获取主题的键值 Gets the key value of the topic

Static Public Member Functions
static void	Lock (TopicHandle topic)
	锁定主题，防止其被多个订阅者同时访问。Lock the topic to prevent it from being accessed by multiple subscribers at the same time.

static void	Unlock (TopicHandle topic)
	解锁主题。Unlock the topic.

static void	LockFromCallback (TopicHandle topic)
	从回调中锁定主题，防止其被多个订阅者同时访问。Lock the topic from a callback to prevent it from being accessed by multiple subscribers at the same time.

static void	UnlockFromCallback (TopicHandle topic)
	从回调中解锁主题。Unlock the topic from a callback.

static MicrosecondTimestamp	NowTimestamp ()
	生成默认发布时刻。Create the default publish timestamp.

template<typename Data >
static Topic	CreateTopic (const char name, Domain domain=nullptr, bool multi_publisher=false, bool cache=false, bool check_length=false)
	创建一个新的主题 Creates a new topic

static TopicHandle	Find (const char name, Domain domain=nullptr)
	在指定域中查找主题 Finds a topic in the specified domain

template<typename Data >
static TopicHandle	FindOrCreate (const char name, Domain domain=nullptr, bool multi_publisher=false, bool cache=false, bool check_length=false)
	在指定域中查找或创建主题 Finds or creates a topic in the specified domain

static void	PackData (uint32_t topic_name_crc32, RawData buffer, MicrosecondTimestamp timestamp, ConstRawData data)
	打包数据

static TopicHandle	WaitTopic (const char name, uint32_t timeout=UINT32_MAX, Domain domain=nullptr)
	等待主题的创建并返回其句柄 Waits for a topic to be created and returns its handle

Private Types
enum class	LockState : uint32_t { UNLOCKED = 0 , LOCKED = 1 , USE_MUTEX = UINT32_MAX }

Private Member Functions
template<SizeLimitMode Mode = SizeLimitMode::MORE>
ErrorCode	DumpPayload (RawData buffer, MicrosecondTimestamp &timestamp)

template<SizeLimitMode Mode = SizeLimitMode::MORE>
ErrorCode	DumpPacket (RawData buffer)

void	PublishRaw (void *addr, uint32_t size, MicrosecondTimestamp timestamp, bool from_callback, bool in_isr)

Static Private Member Functions
static void	EnsureDomainRegistry ()

static Domain *	EnsureDefaultDomain ()

template<typename Data >
static void	CheckSubscriberDataSize (Topic topic)
	校验订阅者数据类型和主题最大长度是否兼容。 Checks whether subscriber data type is compatible with topic max length.

template<typename Data >
static RawData	NewSubscriberBuffer ()
	为异步订阅者分配长期存在的接收缓冲区。 Allocates a long-lived receive buffer for an async subscriber.

static void	CheckPublishSize (TopicHandle topic, uint32_t size)

static RawData	StorePublishedData (TopicHandle topic, void *addr, uint32_t size, MicrosecondTimestamp timestamp)

static void	DispatchSubscriber (SuberBlock &block, MicrosecondTimestamp timestamp, RawData data, bool from_callback, bool in_isr)

static void	DispatchSubscribers (TopicHandle topic, MicrosecondTimestamp timestamp, RawData data, bool from_callback, bool in_isr)

Private Attributes
TopicHandle	block_ = nullptr
	主题句柄，指向当前主题的内存块 Topic handle pointing to the memory block of the current topic

Static Private Attributes
static RBTree< uint32_t > *	domain_ = nullptr
	主题域的红黑树结构，存储不同的主题 Red-Black Tree structure for storing different topics in the domain

static Domain *	def_domain_ = nullptr
	默认的主题域，所有未指定域的主题都会归入此域 Default domain where all topics without a specified domain are assigned

Detailed Description

主题（Topic）管理类 / Topic management class

该类提供了基于发布-订阅模式的主题管理，支持同步、异步、队列和回调订阅者，以及数据的缓存和校验机制。 This class provides topic management based on the publish-subscribe model, supporting synchronous, asynchronous, queue-based, and callback subscribers, as well as data caching and validation mechanisms.

Note: Topic、subscriber 和 server 应在初始化阶段创建，并在应用运行期间长期存在。该消息系统遵循 libxr 的静态生命周期模型：初始化阶段允许分配，发布、回调和解析热路径不应产生动态分配。 Topics, subscribers, and servers are expected to be created during initialization and kept alive for the application lifetime. The message system follows libxr's static lifetime model: allocation is allowed during initialization, while publish, callback, and parser hot paths should not allocate dynamically.

Definition at line 60 of file message.hpp.

Member Typedef Documentation

◆ TopicHandle

RBTree<uint32_t>::Node<Block>* LibXR::Topic::TopicHandle

主题句柄，指向存储数据的红黑树节点。Handle pointing to a red-black tree node storing data.

Definition at line 135 of file message.hpp.

Member Enumeration Documentation

◆ ASyncSubscriberState

enum class LibXR::Topic::ASyncSubscriberState : uint32_t

strong

Definition at line 163 of file subscriber.hpp.

{
  IDLE = 0,
  WAITING = 1,
  DATA_READY = UINT32_MAX
};

◆ LockState

enum class LibXR::Topic::LockState : uint32_t

strongprivate

Definition at line 62 of file message.hpp.

  {
    UNLOCKED = 0,
    LOCKED = 1,
    USE_MUTEX = UINT32_MAX,
  };

◆ SuberType

enum class LibXR::Topic::SuberType : uint8_t

strong

订阅者类型。Subscriber type.

Enumerator
SYNC	同步订阅者。Synchronous subscriber.
ASYNC	异步订阅者。Asynchronous subscriber.
QUEUE	队列订阅者。Queued subscriber.
CALLBACK	回调订阅者。Callback subscriber.

Definition at line 11 of file subscriber.hpp.

{
  SYNC,      
  ASYNC,     
  QUEUE,     
  CALLBACK,  
};

Constructor & Destructor Documentation

◆ Topic() [1/3]

Topic::Topic ( )

默认构造函数，创建一个空的 Topic 实例 Default constructor, creates an empty Topic instance

Definition at line 132 of file topic.cpp.

132{}

◆ Topic() [2/3]

Topic::Topic	(	const char *	name,
		uint32_t	max_length,
		Domain *	domain = nullptr,
		bool	multi_publisher = false,
		bool	cache = false,
		bool	check_length = false )

构造函数，使用指定名称、最大长度、域及其他选项初始化主题 Constructor to initialize a topic with the specified name, maximum length, domain, and options

Parameters

name	主题名称 Topic name
max_length	数据的最大长度 Maximum length of data
domain	主题所属的域（默认为 nullptr）Domain to which the topic belongs (default: nullptr)
multi_publisher	是否允许多个订阅者（默认为 false）Whether to allow multiple subscribers (default: false)
cache	是否启用缓存（默认为 false）Whether to enable caching (default: false)
check_length	是否检查数据长度（默认为 false）Whether to check data length (default: false)

Note: 包含初始化期动态内存分配，主题应长期存在。 Contains initialization-time dynamic allocation; topics are expected to be long-lived.; 未启用缓存时，Topic 只保留最近一次发布数据的地址；调用者必须保证该地址在 DumpData() 前仍然有效。多发布者主题使用 mutex，不支持 PublishFromCallback()。 Without cache, Topic keeps only the last published address; the caller must keep it valid until DumpData(). Multi-publisher topics use a mutex and do not support PublishFromCallback().

Definition at line 134 of file topic.cpp.

{
  ASSERT(name != nullptr);
 
  if (domain == nullptr)
  {
    domain = EnsureDefaultDomain();
  }
 
  auto crc32 = CRC32::Calculate(name, strlen(name));
 
  auto topic = domain->node_->data_.Search<Block>(crc32);
 
  if (topic)
  {
    ASSERT(topic->data_.max_length == max_length);
    ASSERT(topic->data_.check_length == check_length);
 
    if (multi_publisher && !topic->data_.mutex)
    {
      ASSERT(false);
    }
 
    block_ = topic;
  }
  else
  {
    block_ = new RBTree<uint32_t>::Node<Block>;
    block_->data_.max_length = max_length;
    block_->data_.crc32 = crc32;
    block_->data_.data.addr_ = nullptr;
    block_->data_.timestamp = MicrosecondTimestamp();
    block_->data_.cache = false;
    block_->data_.check_length = check_length;
 
    if (multi_publisher)
    {
      block_->data_.mutex = new Mutex();
      block_->data_.busy.store(LockState::USE_MUTEX, std::memory_order_release);
    }
    else
    {
      block_->data_.mutex = nullptr;
      block_->data_.busy.store(LockState::UNLOCKED, std::memory_order_release);
    }
 
    domain->node_->data_.Insert(*block_, crc32);
  }
 
  if (cache && !block_->data_.cache)
  {
    EnableCache();
  }
}

◆ Topic() [3/3]

Topic::Topic ( TopicHandle topic )

通过句柄构造主题 Constructs a topic from a topic handle

Parameters

topic 主题句柄 Topic handle

Definition at line 190 of file topic.cpp.

190: block_(topic) {}

Member Function Documentation

◆ CheckPublishSize()

void Topic::CheckPublishSize	(	TopicHandle	topic,
		uint32_t	size )

staticprivate

Definition at line 15 of file publish.cpp.

{
  if (topic->data_.check_length)
  {
    ASSERT(size == topic->data_.max_length);
  }
  else
  {
    ASSERT(size <= topic->data_.max_length);
  }
}

◆ CheckSubscriberDataSize()

template<typename Data >

static void LibXR::Topic::CheckSubscriberDataSize ( Topic topic )

inlinestaticprivate

校验订阅者数据类型和主题最大长度是否兼容。 Checks whether subscriber data type is compatible with topic max length.

Definition at line 573 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    if (topic.block_->data_.check_length)
    {
      ASSERT(topic.block_->data_.max_length == sizeof(Data));
    }
    else
    {
      ASSERT(topic.block_->data_.max_length <= sizeof(Data));
    }
  }

◆ CreateTopic()

template<typename Data >

static Topic LibXR::Topic::CreateTopic	(	const char *	name,
		Domain *	domain = nullptr,
		bool	multi_publisher = false,
		bool	cache = false,
		bool	check_length = false )

inlinestatic

创建一个新的主题 Creates a new topic

Template Parameters

Data	主题数据类型 Topic data type

Parameters

name	主题名称 Topic name
domain	主题所属的域（默认为 nullptr）Domain to which the topic belongs (default: nullptr)
multi_publisher	是否允许多个订阅者（默认为 false）Whether to allow multiple subscribers (default: false)
cache	是否启用缓存（默认为 false）Whether to enable caching (default: false)
check_length	是否检查数据长度（默认为 false）Whether to check data length (default: false)

Returns: 创建的 Topic 实例 The created Topic instance

Note: 包含初始化期动态内存分配，主题应长期存在。 Contains initialization-time dynamic allocation; topics are expected to be long-lived.

Definition at line 303 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    return Topic(name, sizeof(Data), domain, multi_publisher, cache, check_length);
  }

◆ DispatchSubscriber()

void Topic::DispatchSubscriber	(	SuberBlock &	block,
		MicrosecondTimestamp	timestamp,
		RawData	data,
		bool	from_callback,
		bool	in_isr )

staticprivate

Definition at line 45 of file publish.cpp.

{
  switch (block.type)
  {
    case SuberType::SYNC:
    {
      auto sync = static_cast<SyncBlock*>(&block);
      uint32_t expected = SyncBlock::WAITING;
      auto wake_waiter = sync->wait_state.compare_exchange_strong(
          expected, SyncBlock::WAIT_CLAIMED, std::memory_order_acq_rel,
          std::memory_order_acquire);
 
      if (!wake_waiter)
      {
        break;
      }
 
      LibXR::Memory::FastCopy(sync->buff.addr_, data.addr_, data.size_);
      sync->timestamp = timestamp;
 
      if (from_callback)
      {
        sync->sem.PostFromCallback(in_isr);
      }
      else
      {
        sync->sem.Post();
      }
      break;
    }
    case SuberType::ASYNC:
    {
      auto async = static_cast<ASyncBlock*>(&block);
      if (async->state.load(std::memory_order_acquire) == ASyncSubscriberState::WAITING)
      {
        LibXR::Memory::FastCopy(async->buff.addr_, data.addr_, data.size_);
        async->timestamp = timestamp;
        async->state.store(ASyncSubscriberState::DATA_READY, std::memory_order_release);
      }
      break;
    }
    case SuberType::QUEUE:
    {
      auto queue_block = static_cast<QueueBlock*>(&block);
      queue_block->fun(timestamp, data, *queue_block);
      break;
    }
    case SuberType::CALLBACK:
    {
      auto cb_block = static_cast<CallbackBlock*>(&block);
      cb_block->cb.Run(from_callback && in_isr, timestamp, data);
      break;
    }
  }
}

◆ DispatchSubscribers()

void Topic::DispatchSubscribers	(	TopicHandle	topic,
		MicrosecondTimestamp	timestamp,
		RawData	data,
		bool	from_callback,
		bool	in_isr )

staticprivate

Definition at line 102 of file publish.cpp.

{
  topic->data_.subers.Foreach<SuberBlock>(
      [=](SuberBlock& block)
      {
        DispatchSubscriber(block, timestamp, data, from_callback, in_isr);
        return ErrorCode::OK;
      });
}

◆ DumpData() [1/5]

template<typename Data >
requires (!std::same_as<std::remove_cv_t<Data>, RawData>)

ErrorCode LibXR::Topic::DumpData ( Data & data )

inline

转储数据到普通数据结构 Dumps data into a normal data structure

Note: 允许目标类型只覆盖当前 payload 的前缀，例如把扩展 payload 转储到基础 payload 结构。若目标类型大于当前 payload，则返回 SIZE_ERR。 The destination type may cover only a prefix of the current payload, for example dumping an extended payload into a base payload structure. SIZE_ERR is returned if the destination type is larger than the current payload.

Template Parameters

Data	数据类型 Data type

Parameters

data	存储数据的变量 Variable to store the data

Definition at line 495 of file message.hpp.

  {
    MicrosecondTimestamp timestamp;
    return DumpData(data, timestamp);
  }

◆ DumpData() [2/5]

template<typename Data >
requires (!std::same_as<std::remove_cv_t<Data>, RawData>)

ErrorCode LibXR::Topic::DumpData	(	Data &	data,
		MicrosecondTimestamp &	timestamp )

inline

Definition at line 503 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    return DumpPayload<SizeLimitMode::LESS>(RawData(data), timestamp);
  }

◆ DumpData() [3/5]

template<typename Data >

ErrorCode LibXR::Topic::DumpData ( PackedData< Data > & data )

转储数据到 PackedData Dumps data into PackedData format

Template Parameters

Data	数据类型 Data type

Parameters

data	存储数据的 PackedData 结构 PackedData structure to store data

Definition at line 67 of file packet.hpp.

{
  CheckTopicPayload<Data>();
  if (block_->data_.data.addr_ == nullptr)
  {
    return ErrorCode::EMPTY;
  }
 
  ASSERT(sizeof(Data) == block_->data_.data.size_);
 
  return DumpPacket<SizeLimitMode::NONE>(RawData(data));
}

◆ DumpData() [4/5]

template<SizeLimitMode Mode = SizeLimitMode::MORE>

ErrorCode LibXR::Topic::DumpData ( RawData data )

inline

转储数据到原始缓冲区。Dumps data into a raw buffer.

Definition at line 469 of file message.hpp.

  {
    MicrosecondTimestamp timestamp;
    return DumpPayload<Mode>(data, timestamp);
  }

◆ DumpData() [5/5]

template<SizeLimitMode Mode = SizeLimitMode::MORE>

ErrorCode LibXR::Topic::DumpData	(	RawData	data,
		MicrosecondTimestamp &	timestamp )

inline

Definition at line 476 of file message.hpp.

  {
    return DumpPayload<Mode>(data, timestamp);
  }

◆ DumpPacket()

template<SizeLimitMode Mode = SizeLimitMode::MORE>

ErrorCode LibXR::Topic::DumpPacket ( RawData buffer )

inlineprivate

Definition at line 641 of file message.hpp.

  {
    if (block_->data_.data.addr_ == nullptr)
    {
      return ErrorCode::EMPTY;
    }
 
    Assert::SizeLimitCheck<Mode>(PACK_BASE_SIZE + block_->data_.data.size_, buffer.size_);
    Lock(block_);
    PackData(block_->data_.crc32, buffer, block_->data_.timestamp, block_->data_.data);
    Unlock(block_);
 
    return ErrorCode::OK;
  }

◆ DumpPayload()

template<SizeLimitMode Mode = SizeLimitMode::MORE>

ErrorCode LibXR::Topic::DumpPayload	(	RawData	buffer,
		MicrosecondTimestamp &	timestamp )

inlineprivate

Definition at line 599 of file message.hpp.

  {
    if (block_->data_.data.addr_ == nullptr)
    {
      return ErrorCode::EMPTY;
    }
 
    const size_t payload_size = block_->data_.data.size_;
    size_t copy_size = payload_size;
 
    if constexpr (Mode == SizeLimitMode::EQUAL)
    {
      if (payload_size != buffer.size_)
      {
        return ErrorCode::SIZE_ERR;
      }
    }
    else if constexpr (Mode == SizeLimitMode::LESS)
    {
      if (payload_size < buffer.size_)
      {
        return ErrorCode::SIZE_ERR;
      }
      copy_size = buffer.size_;
    }
    else if constexpr (Mode == SizeLimitMode::MORE)
    {
      if (payload_size > buffer.size_)
      {
        return ErrorCode::SIZE_ERR;
      }
    }
 
    Lock(block_);
    timestamp = block_->data_.timestamp;
    LibXR::Memory::FastCopy(buffer.addr_, block_->data_.data.addr_, copy_size);
    Unlock(block_);
 
    return ErrorCode::OK;
  }

◆ EnableCache()

void Topic::EnableCache ( )

启用主题的缓存功能 Enables caching for the topic

Note: 包含初始化期动态内存分配，缓存启用后应长期存在。 Contains initialization-time dynamic allocation; the cache is expected to remain enabled for the topic lifetime.

Definition at line 210 of file topic.cpp.

{
  Lock(block_);
 
  if (!block_->data_.cache)
  {
    block_->data_.cache = true;
    block_->data_.data.addr_ = new uint8_t[block_->data_.max_length];
  }
 
  Unlock(block_);
}

◆ EnsureDefaultDomain()

Topic::Domain * Topic::EnsureDefaultDomain ( )

staticprivate

Definition at line 20 of file topic.cpp.

{
  if (!def_domain_)
  {
    def_domain_ = new Domain("libxr_def_domain");
  }
 
  return def_domain_;
}

◆ EnsureDomainRegistry()

void Topic::EnsureDomainRegistry ( )

staticprivate

Definition at line 11 of file topic.cpp.

{
  if (!domain_)
  {
    domain_ = new RBTree<uint32_t>([](const uint32_t& a, const uint32_t& b)
                                   { return static_cast<int>(a) - static_cast<int>(b); });
  }
}

◆ Find()

Topic::TopicHandle Topic::Find	(	const char *	name,
		Domain *	domain = nullptr )

static

在指定域中查找主题 Finds a topic in the specified domain

Parameters

name	主题名称 Topic name
domain	主题所属的域（默认为 nullptr）Domain to search in (default: nullptr)

Returns: 主题句柄，如果找到则返回对应的句柄，否则返回 nullptr Topic handle if found, otherwise returns nullptr

Note: 查询不会创建默认域；默认域尚未初始化时返回 nullptr。 Lookup does not create the default domain; it returns nullptr when the default domain has not been initialized.

Definition at line 192 of file topic.cpp.

{
  ASSERT(name != nullptr);
 
  if (domain == nullptr)
  {
    if (def_domain_ == nullptr)
    {
      return nullptr;
    }
    domain = def_domain_;
  }
 
  auto crc32 = CRC32::Calculate(name, strlen(name));
 
  return domain->node_->data_.Search<Block>(crc32);
}

◆ FindOrCreate()

template<typename Data >

static TopicHandle LibXR::Topic::FindOrCreate	(	const char *	name,
		Domain *	domain = nullptr,
		bool	multi_publisher = false,
		bool	cache = false,
		bool	check_length = false )

inlinestatic

在指定域中查找或创建主题 Finds or creates a topic in the specified domain

Template Parameters

Data	数据类型 Data type

Parameters

name	话题名称 Topic name
domain	可选的域指针 Optional domain pointer (default: nullptr)
multi_publisher	可选的多发布标志位 Optional multi-publisher flag
cache	可选的缓存标志位 Optional cache flag (default: false)
check_length	可选的数据长度检查标志位 Optional data length check flag (default: false)

Returns: TopicHandle 主题句柄 Topic handle

Note: 包含初始化期动态内存分配，主题应长期存在。 Contains initialization-time dynamic allocation; topics are expected to be long-lived.

Definition at line 350 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    auto topic = Find(name, domain);
    if (topic == nullptr)
    {
      topic =
          CreateTopic<Data>(name, domain, multi_publisher, cache, check_length).block_;
    }
    return topic;
  }

◆ GetKey()

uint32_t Topic::GetKey ( ) const

获取主题的键值 Gets the key value of the topic

Returns: uint32_t

Definition at line 244 of file topic.cpp.

{
  if (block_)
  {
    return block_->key;
  }
  else
  {
    return 0;
  }
}

◆ GetTimestamp()

MicrosecondTimestamp Topic::GetTimestamp ( ) const

Definition at line 256 of file topic.cpp.

{
  if (block_ == nullptr)
  {
    return MicrosecondTimestamp();
  }
  return block_->data_.timestamp;
}

◆ Lock()

void Topic::Lock ( Topic::TopicHandle topic )

static

锁定主题，防止其被多个订阅者同时访问。Lock the topic to prevent it from being accessed by multiple subscribers at the same time.

Parameters

topic 要锁定的主题。Topic to be locked.

Definition at line 30 of file topic.cpp.

{
  if (topic->data_.mutex)
  {
    topic->data_.mutex->Lock();
  }
  else
  {
    LockState expected = LockState::UNLOCKED;
    if (!topic->data_.busy.compare_exchange_strong(expected, LockState::LOCKED))
    {
      /* Multiple threads are trying to lock the same topic */
      ASSERT(false);
      return;
    }
  }
}

◆ LockFromCallback()

void Topic::LockFromCallback ( Topic::TopicHandle topic )

static

从回调中锁定主题，防止其被多个订阅者同时访问。Lock the topic from a callback to prevent it from being accessed by multiple subscribers at the same time.

Parameters

topic 要锁定的主题。Topic to be locked.

Definition at line 60 of file topic.cpp.

{
  if (topic->data_.mutex)
  {
    ASSERT(false);
  }
  else
  {
    LockState expected = LockState::UNLOCKED;
    if (!topic->data_.busy.compare_exchange_strong(expected, LockState::LOCKED))
    {
      /* Multiple threads are trying to lock the same topic */
      ASSERT(false);
      return;
    }
  }
}

◆ NewSubscriberBuffer()

template<typename Data >

static RawData LibXR::Topic::NewSubscriberBuffer ( )

inlinestaticprivate

为异步订阅者分配长期存在的接收缓冲区。 Allocates a long-lived receive buffer for an async subscriber.

Definition at line 591 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    auto* data = new Data;
    return RawData(*data);
  }

◆ NowTimestamp()

MicrosecondTimestamp Topic::NowTimestamp ( )

static

生成默认发布时刻。Create the default publish timestamp.

Returns: 当前 Timebase 微秒时间戳。Current Timebase microsecond timestamp.

Definition at line 9 of file publish.cpp.

{
  ASSERT(Timebase::timebase != nullptr);
  return Timebase::GetMicroseconds();
}

◆ operator TopicHandle()

LibXR::Topic::operator TopicHandle ( )

inline

将 Topic 转换为 TopicHandle Converts Topic to TopicHandle

Returns: TopicHandle

Definition at line 530 of file message.hpp.

530{ return block_; }

◆ PackData()

void Topic::PackData	(	uint32_t	topic_name_crc32,
		RawData	buffer,
		MicrosecondTimestamp	timestamp,
		ConstRawData	data )

static

打包数据

Parameters

topic_name_crc32	话题名称的 CRC32 校验码
buffer	等待写入的包 Packed data to be written
timestamp	需要打包的消息时间戳 Message timestamp to be packed
data	需要打包的消息数据 Message data to be packed

Definition at line 40 of file packet.cpp.

{
  PackedData<uint8_t>* pack = reinterpret_cast<PackedData<uint8_t>*>(buffer.addr_);
 
  LibXR::Memory::FastCopy(&pack->raw.data_, data.addr_, data.size_);
 
  pack->raw.header_.prefix = PACKET_PREFIX;
  pack->raw.header_.topic_name_crc32 = topic_name_crc32;
  pack->raw.header_.SetDataLen(data.size_);
  pack->raw.header_.SetTimestamp(timestamp);
  pack->raw.header_.pack_header_crc8 =
      CRC8::Calculate(&pack->raw, sizeof(PackedDataHeader) - sizeof(uint8_t));
  uint8_t* crc8_pack = reinterpret_cast<uint8_t*>(
      reinterpret_cast<uint8_t*>(pack) + PACK_BASE_SIZE + data.size_ - sizeof(uint8_t));
  *crc8_pack = CRC8::Calculate(pack, PACK_BASE_SIZE - sizeof(uint8_t) + data.size_);
}

◆ Publish() [1/4]

template<typename Data >

void LibXR::Topic::Publish ( Data & data )

inline

发布数据 Publishes data

Template Parameters

Data	数据类型 Data type

Parameters

data	需要发布的数据 Data to be published

Definition at line 381 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    Publish(static_cast<void*>(&data), static_cast<uint32_t>(sizeof(Data)));
  }

◆ Publish() [2/4]

template<typename Data >

void LibXR::Topic::Publish	(	Data &	data,
		MicrosecondTimestamp	timestamp )

inline

Definition at line 388 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    Publish(static_cast<void*>(&data), static_cast<uint32_t>(sizeof(Data)), timestamp);
  }

◆ Publish() [3/4]

void Topic::Publish	(	void *	addr,
		uint32_t	size )

以原始地址和大小发布数据 Publishes data using raw address and size

Parameters

addr	数据的地址 Address of the data
size	数据大小 Size of the data

Definition at line 139 of file publish.cpp.

139{ Publish(addr, size, NowTimestamp()); }

LibXR::Topic::NowTimestamp

static MicrosecondTimestamp NowTimestamp()

生成默认发布时刻。Create the default publish timestamp.

Definition publish.cpp:9

◆ Publish() [4/4]

void Topic::Publish	(	void *	addr,
		uint32_t	size,
		MicrosecondTimestamp	timestamp )

Definition at line 141 of file publish.cpp.

{
  PublishRaw(addr, size, timestamp, false, false);
}

◆ PublishFromCallback() [1/4]

template<typename Data >

void LibXR::Topic::PublishFromCallback	(	Data &	data,
		bool	in_isr )

inline

在回调函数中发布数据 Publishes data in a callback

Template Parameters

Data	数据类型 Data type

Parameters

data	需要发布的数据 Data to be published
in_isr	是否在中断中发布数据 Whether to publish data in an interrupt

Definition at line 412 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    PublishFromCallback(static_cast<void*>(&data), static_cast<uint32_t>(sizeof(Data)),
                        in_isr);
  }

◆ PublishFromCallback() [2/4]

template<typename Data >

void LibXR::Topic::PublishFromCallback	(	Data &	data,
		MicrosecondTimestamp	timestamp,
		bool	in_isr )

inline

Definition at line 420 of file message.hpp.

  {
    CheckTopicPayload<Data>();
    PublishFromCallback(static_cast<void*>(&data), static_cast<uint32_t>(sizeof(Data)),
                        timestamp, in_isr);
  }

◆ PublishFromCallback() [3/4]

void Topic::PublishFromCallback	(	void *	addr,
		uint32_t	size,
		bool	in_isr )

在回调函数中以原始地址和大小发布数据 Publishes data using raw address and size in a callback

Parameters

addr	数据的地址 Address of the data
size	数据大小 Size of the data
in_isr	是否在中断中发布数据 Whether to publish data in an interrupt

Note: 仅适用于非 mutex 主题。multi_publisher 主题会在回调发布路径触发断言。 Only valid for non-mutex topics. Multi-publisher topics assert on the callback publish path.; 回调发布路径不分配内存，但仍会同步分发到该主题的订阅者。 The callback publish path does not allocate, but still dispatches to the topic subscribers synchronously.

Definition at line 146 of file publish.cpp.

{
  PublishFromCallback(addr, size, NowTimestamp(), in_isr);
}

◆ PublishFromCallback() [4/4]

void Topic::PublishFromCallback	(	void *	addr,
		uint32_t	size,
		MicrosecondTimestamp	timestamp,
		bool	in_isr )

Definition at line 151 of file publish.cpp.

{
  PublishRaw(addr, size, timestamp, true, in_isr);
}

◆ PublishRaw()

void Topic::PublishRaw	(	void *	addr,
		uint32_t	size,
		MicrosecondTimestamp	timestamp,
		bool	from_callback,
		bool	in_isr )

private

Definition at line 113 of file publish.cpp.

{
  if (from_callback)
  {
    LockFromCallback(block_);
  }
  else
  {
    Lock(block_);
  }
 
  CheckPublishSize(block_, size);
  RawData data = StorePublishedData(block_, addr, size, timestamp);
  DispatchSubscribers(block_, timestamp, data, from_callback, in_isr);
 
  if (from_callback)
  {
    UnlockFromCallback(block_);
  }
  else
  {
    Unlock(block_);
  }
}

◆ RegisterCallback()

void LibXR::Topic::RegisterCallback ( Callback & cb )

注册回调函数 Registers a callback function

Parameters

cb	需要注册的回调函数 The callback function to register

Note: 包含初始化期动态内存分配，回调订阅应长期存在。 Contains initialization-time dynamic allocation; callback subscriptions are expected to be long-lived.; 回调在发布锁内运行，不应重入发布同一个主题。 The callback runs while the publish lock is held and should not re-enter publishing on the same topic.

Definition at line 113 of file topic.cpp.

{
  if (cb.PayloadSize() != 0)
  {
    if (block_->data_.check_length)
    {
      ASSERT(block_->data_.max_length == cb.PayloadSize());
    }
    else
    {
      ASSERT(block_->data_.max_length <= cb.PayloadSize());
    }
  }
 
  auto node = new (std::align_val_t(LibXR::CACHE_LINE_SIZE))
      LockFreeList::Node<CallbackBlock>(cb);
  block_->data_.subers.Add(*node);
}

◆ StorePublishedData()

RawData Topic::StorePublishedData	(	TopicHandle	topic,
		void *	addr,
		uint32_t	size,
		MicrosecondTimestamp	timestamp )

staticprivate

Definition at line 27 of file publish.cpp.

{
  if (topic->data_.cache)
  {
    LibXR::Memory::FastCopy(topic->data_.data.addr_, addr, size);
    topic->data_.data.size_ = size;
  }
  else
  {
    topic->data_.data.addr_ = addr;
    topic->data_.data.size_ = size;
  }
 
  topic->data_.timestamp = timestamp;
  return topic->data_.data;
}

◆ Unlock()

void Topic::Unlock ( Topic::TopicHandle topic )

static

解锁主题。Unlock the topic.

Parameters

topic 要解锁的主题。Topic to be unlocked.

Definition at line 48 of file topic.cpp.

{
  if (topic->data_.mutex)
  {
    topic->data_.mutex->Unlock();
  }
  else
  {
    topic->data_.busy.store(LockState::UNLOCKED, std::memory_order_release);
  }
}

◆ UnlockFromCallback()

void Topic::UnlockFromCallback ( Topic::TopicHandle topic )

static

从回调中解锁主题。Unlock the topic from a callback.

Parameters

topic 要解锁的主题。Topic to be unlocked.

Definition at line 78 of file topic.cpp.

{
  if (topic->data_.mutex)
  {
    ASSERT(false);
  }
  else
  {
    topic->data_.busy.store(LockState::UNLOCKED, std::memory_order_release);
  }
}

◆ WaitTopic()

Topic::TopicHandle Topic::WaitTopic	(	const char *	name,
		uint32_t	timeout = UINT32_MAX,
		Domain *	domain = nullptr )

static

等待主题的创建并返回其句柄 Waits for a topic to be created and returns its handle

Parameters

name	主题名称 The name of the topic
timeout	等待的超时时间（默认 UINT32_MAX）Timeout duration to wait (default: UINT32_MAX)
domain	主题所属的域（默认为 nullptr）The domain in which to search for the topic (default: nullptr)

Returns: TopicHandle 如果找到主题，则返回其句柄，否则返回 nullptr TopicHandle if the topic is found, otherwise returns nullptr

Definition at line 223 of file topic.cpp.

{
  const uint32_t start_time = Thread::GetTime();
  TopicHandle topic = nullptr;
  do
  {
    topic = Find(name, domain);
    if (topic == nullptr)
    {
      if (timeout != UINT32_MAX &&
          static_cast<uint32_t>(Thread::GetTime() - start_time) >= timeout)
      {
        return nullptr;
      }
      Thread::Sleep(1);
    }
  } while (topic == nullptr);
 
  return topic;
}

Field Documentation

◆ block_

TopicHandle LibXR::Topic::block_ = nullptr

private

主题句柄，指向当前主题的内存块 Topic handle pointing to the memory block of the current topic

Definition at line 547 of file message.hpp.

◆ def_domain_

Domain* LibXR::Topic::def_domain_ = nullptr

inlinestaticprivate

默认的主题域，所有未指定域的主题都会归入此域 Default domain where all topics without a specified domain are assigned

Definition at line 563 of file message.hpp.

◆ domain_

RBTree<uint32_t>* LibXR::Topic::domain_ = nullptr

inlinestaticprivate

主题域的红黑树结构，存储不同的主题 Red-Black Tree structure for storing different topics in the domain

Note: 初始化期懒创建；调用者应在并发发布前完成 topic/domain 创建。 Lazily created during initialization; callers should finish topic/domain creation before concurrent publishing starts.

Definition at line 557 of file message.hpp.

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

src/middleware/message/message.hpp
src/middleware/message/packet.cpp
src/middleware/message/packet.hpp
src/middleware/message/publish.cpp
src/middleware/message/topic.cpp

Data Structures

Public Types

Public Member Functions

Static Public Member Functions

Private Types

Private Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Typedef Documentation

◆ TopicHandle

Member Enumeration Documentation

◆ ASyncSubscriberState

◆ LockState

◆ SuberType

Constructor & Destructor Documentation

◆ Topic() [1/3]

◆ Topic() [2/3]

◆ Topic() [3/3]

Member Function Documentation

◆ CheckPublishSize()

◆ CheckSubscriberDataSize()

◆ CreateTopic()

◆ DispatchSubscriber()

◆ DispatchSubscribers()

◆ DumpData() [1/5]

◆ DumpData() [2/5]

◆ DumpData() [3/5]

◆ DumpData() [4/5]

◆ DumpData() [5/5]

◆ DumpPacket()

◆ DumpPayload()

◆ EnableCache()

◆ EnsureDefaultDomain()

◆ EnsureDomainRegistry()

◆ Find()

◆ FindOrCreate()

◆ GetKey()

◆ GetTimestamp()

◆ Lock()

◆ LockFromCallback()

◆ NewSubscriberBuffer()

◆ NowTimestamp()

◆ operator TopicHandle()

◆ PackData()

◆ Publish() [1/4]

◆ Publish() [2/4]

◆ Publish() [3/4]

◆ Publish() [4/4]

◆ PublishFromCallback() [1/4]

◆ PublishFromCallback() [2/4]

◆ PublishFromCallback() [3/4]

◆ PublishFromCallback() [4/4]

◆ PublishRaw()

◆ RegisterCallback()

◆ StorePublishedData()

◆ Unlock()

◆ UnlockFromCallback()

◆ WaitTopic()

Field Documentation

◆ block_

◆ def_domain_

◆ domain_