发布订阅主题 / Publish-subscribe topic More...

#include <topic.hpp>

Collaboration diagram for LibXR::Topic:

Data Structures
struct	ASyncBlock
	异步订阅者自己挂的数据块 / Data block owned by one asynchronous subscriber More...

class	ASyncSubscriber
	先 `StartWaiting()`，再自己来取数据的订阅者 / Subscriber that first calls `StartWaiting()` and later pulls the data itself More...

struct	Block
	topic 运行时状态块 / Runtime state block of one topic More...

class	Callback
	每次发布时直接执行函数的订阅句柄 / Subscription handle that runs a function on each publish More...

struct	CallbackBlock
	挂在 topic 订阅链表里的回调记录 / Callback record stored in the topic subscriber list More...

class	Domain
	topic 所属的命名域 / Naming domain that groups topics More...

struct	Message
	带时间戳和 payload 副本的消息对象 / Message object carrying a timestamp and a payload copy More...

struct	MessageView
	带时间戳和 payload 指针的只读消息视图 / Read-only message view carrying a timestamp and a payload pointer More...

struct	QueueBlock
	队列订阅者自己挂的数据块 / Data block owned by one queued subscriber More...

class	QueuedSubscriber
	每次发布都往队列里塞一份数据的订阅者 / Subscriber that pushes one entry into a queue on each publish More...

struct	RawMessageView
	带时间戳和只读原始 payload 视图的消息 / Message carrying a timestamp and a read-only raw payload view More...

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

struct	SuberBlock
	所有订阅块共用的公共头 / Common header shared by all subscriber blocks More...

struct	SyncBlock
	同步订阅者自己挂的数据块 / Data block owned by one synchronous subscriber More...

class	SyncSubscriber
	调用 `Wait()` 收消息的订阅者 / Subscriber that receives messages by calling `Wait()` More...

Public Types
enum class	ASyncSubscriberState : uint32_t { IDLE = 0 , WAITING = 1 , DATA_READY = UINT32_MAX }
	异步订阅者本地缓冲区的状态 / State of the async subscriber's local buffer More...

enum class	SuberType : uint8_t { SYNC , ASYNC , QUEUE , CALLBACK }
	topic 支持的订阅者种类 / Subscriber kinds supported by a topic More...

typedef RBTree< uint32_t >::Node< Block > *	TopicHandle
	指向一个 topic 运行时状态块的句柄 / Handle pointing to one topic runtime state block

Public Member Functions
void	RegisterCallback (Callback &cb)
	注册一个回调订阅者 / Register one callback subscriber

	Topic ()
	构造一个空 topic 视图 / Construct one empty topic view

	Topic (const char name, TypeID::ID payload_type_id, size_t payload_size, size_t payload_alignment, Domain domain=nullptr, bool multi_publisher=false)
	用显式类型契约构造或查找一个 topic / Construct or look up one topic with an explicit runtime type contract

	Topic (TopicHandle topic)
	从已有句柄构造一个 topic 视图 / Construct one topic view from an existing handle

size_t	PayloadSize () const
	获取该 topic 固定 payload 字节数 / Get the fixed payload size of this topic

size_t	PayloadAlignment () const
	获取该 topic payload 的对齐要求 / Get the payload alignment requirement of this topic

template<typename Data >
void	Publish (Data &data)
	在普通上下文里发布一条消息，并自动取当前时间戳 / Publish one message in normal context and stamp it with the current time

template<typename Data >
void	Publish (Data &data, MicrosecondTimestamp timestamp)
	在普通上下文里按指定时间戳发布一条消息 / Publish one message in normal context with an explicit timestamp

template<typename Data >
void	PublishFromCallback (Data &data, bool in_isr)
	在回调或 ISR 路径里发布一条消息，并自动取当前时间戳 / Publish one message from callback or ISR context and stamp it with the current time

template<typename Data >
void	PublishFromCallback (Data &data, MicrosecondTimestamp timestamp, bool in_isr)
	在回调或 ISR 路径里按指定时间戳发布一条消息 / Publish one message from callback or ISR context with an explicit timestamp

void	PublishBytesFromServer (void *payload_addr, size_t payload_size, MicrosecondTimestamp timestamp)
	供 packet/server 路径按字节发布一条消息 / Publish one message from the packet/server path using bytes already arranged as the exact payload object

void	PublishBytesFromServerCallback (void *payload_addr, size_t payload_size, MicrosecondTimestamp timestamp, bool in_isr)
	供回调/ISR 上下文里的 packet/server 路径按字节发布一条消息 / Publish one packet/server message from callback/ISR context using bytes already arranged as the exact payload object

template<typename Data >
ErrorCode	PackData (const Data &data, PackedData< Data > &packet)
	将一个精确类型消息打包成 packet / Pack one exact-typed message into one packet using the topic's runtime contract and current timestamp

template<typename Data >
ErrorCode	PackData (const Data &data, PackedData< Data > &packet, MicrosecondTimestamp timestamp)
	将一个精确类型消息按指定时间戳打包成 packet / Pack one exact-typed message into one packet with the given timestamp

ErrorCode	PackRaw (ConstRawData data, RawData packet)
	将一段 raw payload 按当前 topic 元数据打包成 packet / Pack a raw payload byte view into one packet using the current topic metadata and current timestamp

ErrorCode	PackRaw (ConstRawData data, RawData packet, MicrosecondTimestamp timestamp)
	将一段 raw payload 按当前 topic 元数据和指定时间戳打包成 packet / Pack a raw payload byte view into one packet using the current topic metadata and timestamp

	operator TopicHandle ()
	把 topic 视图转换为底层句柄 / Convert this topic view to the underlying handle

uint32_t	GetKey () const
	读取 topic 键值 / Read the key value of this topic

Static Public Member Functions
static void	Lock (TopicHandle topic)
	在普通上下文里锁住一个 topic 发布路径 / Lock one topic publish path in normal context

static void	Unlock (TopicHandle topic)
	在普通上下文里释放一个 topic 发布路径 / Unlock one topic publish path in normal context

static void	LockFromCallback (TopicHandle topic)
	在回调或 ISR 路径里锁住一个 topic 发布路径 / Lock one topic publish path from callback or ISR context

static void	UnlockFromCallback (TopicHandle topic)
	在回调或 ISR 路径里释放一个 topic 发布路径 / Unlock one topic publish path from callback or ISR context

static MicrosecondTimestamp	NowTimestamp ()
	读取当前时间戳 / Read the current timestamp

template<typename Data >
static Topic	CreateTopic (const char name, Domain domain=nullptr, bool multi_publisher=false)
	用精确类型创建或查找一个 topic / Create or look up one topic using one exact payload type

static TopicHandle	Find (const char name, Domain domain=nullptr)
	按名称查找一个已存在 topic / Find one existing topic by name

template<typename Data >
static TopicHandle	FindOrCreate (const char name, Domain domain=nullptr, bool multi_publisher=false)
	按精确类型查找或创建一个 topic / Find or create one topic with an exact payload type

static TopicHandle	WaitTopic (const char name, uint32_t timeout=UINT32_MAX, Domain domain=nullptr)
	等待指定名称的 topic 出现 / Wait until a topic with the given name exists

Private Types
enum class	LockState : uint32_t { UNLOCKED = 0 , LOCKED = 1 , USE_MUTEX = UINT32_MAX }
	topic 发布路径的内部锁状态 / Internal lock state of the topic publish path More...

Private Member Functions
template<typename Data >
void	PublishTyped (Data &data, MicrosecondTimestamp timestamp, bool from_callback, bool in_isr)
	强类型发布入口的共享实现 / Shared implementation of typed publish entry points

void	PublishServerBytes (void *payload_addr, size_t payload_size, MicrosecondTimestamp timestamp, bool from_callback, bool in_isr)
	`PublishBytesFromServer()` 的共享实现 / Shared implementation behind `PublishBytesFromServer()`

Static Private Member Functions
static void	EnsureDomainRegistry ()
	确保全局域注册表已创建 / Ensure the global domain registry exists

static Domain *	EnsureDefaultDomain ()
	确保默认域已创建 / Ensure the default domain exists

static void	CheckServerPublishContract (TopicHandle topic, void *payload_addr, size_t payload_size)
	校验 server 侧字节发布前提 / Check the preconditions of one server-side byte publish

template<typename Data >
static void	CheckSubscriberType (Topic topic)
	断言订阅者看到的精确 payload 类型与 topic 契约一致 / Assert that the exact payload type seen by a subscriber matches the topic contract

template<typename Data >
static void *	AllocateSubscriberBuffer ()
	为订阅者分配一个长期存在的本地接收对象 / Allocate one long-lived local receive object for a subscriber

template<typename Data >
static void	CopyPayload (void dst, void payload_addr)
	按精确类型把一份 payload 拷到订阅者缓冲区 / Copy one payload into a subscriber buffer using the exact type

static void	CheckPublishContract (TopicHandle topic, TypeID::ID payload_type_id, size_t payload_size, size_t payload_alignment)
	校验一次强类型发布的运行时契约 / Check the runtime contract of one typed publish

static void	PackBytes (uint32_t topic_name_crc32, RawData buffer, MicrosecondTimestamp timestamp, ConstRawData data)
	将一段 payload 字节和 topic 元数据拼成 packet / Pack one payload byte range together with topic metadata into one packet

static void	DispatchSubscriber (SuberBlock &block, MicrosecondTimestamp timestamp, void *payload_addr, bool from_callback, bool in_isr)
	将一条消息分发给一个订阅块 / Dispatch one message to one subscriber block

static void	DispatchSubscribers (TopicHandle topic, MicrosecondTimestamp timestamp, void *payload_addr, bool from_callback, bool in_isr)
	将一条消息分发给一个 topic 上的全部订阅者 / Dispatch one message to all subscribers attached to one topic

Private Attributes
TopicHandle	block_ = nullptr

Static Private Attributes
static RBTree< uint32_t > *	domain_
	全局 topic 域注册表。Global registry of topic domains.

static Domain *	def_domain_ = nullptr
	缺省 topic 域。Default topic domain.

Detailed Description

发布订阅主题 / Publish-subscribe topic

一个 Topic 表示一种精确类型的消息通道：它在每次发布时同步通知挂在上面的同步、异步、队列或回调订阅者，但自身不缓存最近一次消息。 One Topic represents one exact-typed message channel: it synchronously notifies the attached synchronous, asynchronous, queued, or callback subscribers on each publish, but does not cache the latest message itself.

Definition at line 55 of file topic.hpp.

Member Typedef Documentation

◆ TopicHandle

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

指向一个 topic 运行时状态块的句柄 / Handle pointing to one topic runtime state block

Definition at line 125 of file topic.hpp.

Member Enumeration Documentation

◆ ASyncSubscriberState

enum class LibXR::Topic::ASyncSubscriberState : uint32_t

strong

异步订阅者本地缓冲区的状态 / State of the async subscriber's local buffer

异步订阅者本地缓冲区状态 / State of one asynchronous subscriber local buffer

Enumerator
IDLE	当前没有等待，也没有待消费的新数据。No wait is pending and no unread data is buffered.
WAITING	等待下一次发布填充本地缓冲区。Waiting for the next publish to fill the local buffer.
DATA_READY	本地缓冲区已有一份待消费的新数据。One unread fresh sample is buffered locally.

Definition at line 11 of file async.hpp.

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

◆ LockState

enum class LibXR::Topic::LockState : uint32_t

strongprivate

topic 发布路径的内部锁状态 / Internal lock state of the topic publish path

Enumerator
UNLOCKED	当前未持有发布锁。The publish path is currently unlocked.
LOCKED	当前通过原子快路径持有发布权。The publish path is locked through the atomic fast path.
USE_MUTEX	当前主题改走互斥量串行化。This topic currently serializes publishers through a mutex.

Definition at line 61 of file topic.hpp.

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

◆ SuberType

enum class LibXR::Topic::SuberType : uint8_t

strong

topic 支持的订阅者种类 / Subscriber kinds supported by a topic

Enumerator
SYNC	同步等待型订阅者。Synchronous wait-based subscriber.
ASYNC	异步本地缓冲型订阅者。Asynchronous local-buffer subscriber.
QUEUE	队列转发型订阅者。Queue-forwarding subscriber.
CALLBACK	回调执行型订阅者。Callback-executing subscriber.

Definition at line 224 of file topic.hpp.

  {
    SYNC,      
    ASYNC,     
    QUEUE,     
    CALLBACK,  
  };

Constructor & Destructor Documentation

◆ Topic() [1/3]

Topic::Topic ( )

构造一个空 topic 视图 / Construct one empty topic view

Definition at line 114 of file topic.cpp.

114{}

◆ Topic() [2/3]

Topic::Topic	(	const char *	name,
		TypeID::ID	payload_type_id,
		size_t	payload_size,
		size_t	payload_alignment,
		Domain *	domain = nullptr,
		bool	multi_publisher = false )

用显式类型契约构造或查找一个 topic / Construct or look up one topic with an explicit runtime type contract

Parameters

name	topic 名称 / Topic name
payload_type_id	payload 类型标识 / Payload type identifier
payload_size	该 topic 固定 payload 字节数 / Fixed payload size of this topic
payload_alignment	该 topic payload 所需对齐，必须是非零 2 次幂 / Required payload alignment of this topic; must be a non-zero power of two
domain	可选主题域 / Optional topic domain
multi_publisher	是否允许多发布者串行化 / Whether to allow serialized multi-publisher use

Definition at line 116 of file topic.cpp.

{
  ASSERT(name != nullptr);
  ASSERT(payload_type_id != nullptr);
  ASSERT(payload_size != 0);
  ASSERT(payload_alignment != 0);
  ASSERT((payload_alignment & (payload_alignment - 1)) == 0);
 
  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_.payload_type_id == payload_type_id);
    ASSERT(topic->data_.payload_size == payload_size);
    ASSERT(topic->data_.payload_alignment == payload_alignment);
 
    if (multi_publisher && !topic->data_.mutex)
    {
      ASSERT(false);
    }
 
    block_ = topic;
  }
  else
  {
    block_ = new RBTree<uint32_t>::Node<Block>;
    block_->data_.payload_type_id = payload_type_id;
    block_->data_.payload_size = payload_size;
    block_->data_.payload_alignment = payload_alignment;
    block_->data_.crc32 = crc32;
 
    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);
  }
}

◆ Topic() [3/3]

Topic::Topic ( TopicHandle topic )

从已有句柄构造一个 topic 视图 / Construct one topic view from an existing handle

Parameters

topic 既有 topic 句柄 / Existing topic handle

Definition at line 170 of file topic.cpp.

170: block_(topic) {}

Member Function Documentation

◆ AllocateSubscriberBuffer()

template<typename Data >

static void * LibXR::Topic::AllocateSubscriberBuffer ( )

inlinestaticprivate

为订阅者分配一个长期存在的本地接收对象 / Allocate one long-lived local receive object for a subscriber

Template Parameters

Data	对象类型 / Object type

Returns: 新分配对象的地址 / Address of the newly allocated object

Definition at line 669 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
    return new Data;
  }

◆ CheckPublishContract()

void Topic::CheckPublishContract	(	TopicHandle	topic,
		TypeID::ID	payload_type_id,
		size_t	payload_size,
		size_t	payload_alignment )

staticprivate

校验一次强类型发布的运行时契约 / Check the runtime contract of one typed publish

Parameters

topic	目标 topic 句柄 / Target topic handle
payload_type_id	本次发布的精确 payload 类型标识 / Exact payload type ID of this publish
payload_size	本次发布的 payload 固定字节数 / Fixed payload size of this publish
payload_alignment	本次发布的 payload 对齐要求 / Payload alignment requirement of this publish

Definition at line 82 of file publish.cpp.

{
  ASSERT(topic != nullptr);
  ASSERT(payload_type_id != nullptr);
  ASSERT(topic->data_.payload_type_id == payload_type_id);
  ASSERT(topic->data_.payload_size == payload_size);
  ASSERT(topic->data_.payload_alignment == payload_alignment);
}

◆ CheckServerPublishContract()

static void LibXR::Topic::CheckServerPublishContract	(	TopicHandle	topic,
		void *	payload_addr,
		size_t	payload_size )

inlinestaticprivate

校验 server 侧字节发布前提 / Check the preconditions of one server-side byte publish

Parameters

topic	目标 topic / Target topic
payload_addr	已满足对齐的 payload 地址 / Already-aligned payload address
payload_size	payload 字节数 / Payload size in bytes

Definition at line 635 of file topic.hpp.

  {
    ASSERT(topic != nullptr);
    ASSERT(payload_addr != nullptr);
    ASSERT(payload_size == topic->data_.payload_size);
    ASSERT(topic->data_.payload_alignment != 0);
    ASSERT(reinterpret_cast<uintptr_t>(payload_addr) % topic->data_.payload_alignment ==
           0);
  }

◆ CheckSubscriberType()

template<typename Data >

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

inlinestaticprivate

断言订阅者看到的精确 payload 类型与 topic 契约一致 / Assert that the exact payload type seen by a subscriber matches the topic contract

Template Parameters

Data	订阅类型 / Subscriber payload type

Parameters

topic 目标 topic 视图 / Target topic view

Definition at line 653 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
    ASSERT(topic.block_ != nullptr);
    ASSERT(topic.block_->data_.payload_type_id == TypeID::GetID<Data>());
    ASSERT(topic.block_->data_.payload_size == sizeof(Data));
    ASSERT(topic.block_->data_.payload_alignment == alignof(Data));
  }

◆ CopyPayload()

template<typename Data >

static void LibXR::Topic::CopyPayload	(	void *	dst,
		void *	payload_addr )

inlinestaticprivate

按精确类型把一份 payload 拷到订阅者缓冲区 / Copy one payload into a subscriber buffer using the exact type

Template Parameters

Data	payload 类型 / Payload type

Parameters

dst	订阅者缓冲区地址 / Subscriber buffer address
payload_addr	本次发布 payload 地址 / Address of the payload object of the current publish

Definition at line 684 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
    ASSERT(dst != nullptr);
    ASSERT(payload_addr != nullptr);
    *reinterpret_cast<Data*>(dst) = *reinterpret_cast<Data*>(payload_addr);
  }

◆ CreateTopic()

template<typename Data >

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

inlinestatic

用精确类型创建或查找一个 topic / Create or look up one topic using one exact payload type

Template Parameters

Data	payload 类型 / Payload type

Parameters

name	topic 名称 / Topic name
domain	可选主题域 / Optional topic domain
multi_publisher	是否允许多发布者串行化 / Whether to allow serialized multi-publisher use

Returns: 对应的 topic 视图 / Topic view of the requested topic

Definition at line 357 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
    return Topic(name, TypeID::GetID<Data>(), sizeof(Data), alignof(Data), domain,
                 multi_publisher);
  }

◆ DispatchSubscriber()

void Topic::DispatchSubscriber	(	SuberBlock &	block,
		MicrosecondTimestamp	timestamp,
		void *	payload_addr,
		bool	from_callback,
		bool	in_isr )

staticprivate

将一条消息分发给一个订阅块 / Dispatch one message to one subscriber block

Parameters

block	目标订阅块 / Target subscriber block
timestamp	消息时间戳 / Message timestamp
payload_addr	本次发布 payload 地址 / Address of the payload object of the current publish
from_callback	是否来自回调路径 / Whether this publish comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 12 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;
      }
 
      sync->copy_payload(sync->buff_addr, payload_addr);
      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)
      {
        async->copy_payload(async->buff_addr, payload_addr);
        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, payload_addr, *queue_block);
      break;
    }
    case SuberType::CALLBACK:
    {
      auto cb_block = static_cast<CallbackBlock*>(&block);
      cb_block->Run(from_callback && in_isr, timestamp, payload_addr);
      break;
    }
  }
}

◆ DispatchSubscribers()

void Topic::DispatchSubscribers	(	TopicHandle	topic,
		MicrosecondTimestamp	timestamp,
		void *	payload_addr,
		bool	from_callback,
		bool	in_isr )

staticprivate

将一条消息分发给一个 topic 上的全部订阅者 / Dispatch one message to all subscribers attached to one topic

Parameters

topic	目标 topic 句柄 / Target topic handle
timestamp	消息时间戳 / Message timestamp
payload_addr	本次发布 payload 地址 / Address of the payload object of the current publish
from_callback	是否来自回调路径 / Whether this publish comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 69 of file publish.cpp.

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

◆ EnsureDefaultDomain()

Topic::Domain * Topic::EnsureDefaultDomain ( )

staticprivate

确保默认域已创建 / Ensure the default domain exists

Returns: 默认域指针 / Default domain pointer

Definition at line 22 of file topic.cpp.

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

◆ EnsureDomainRegistry()

void Topic::EnsureDomainRegistry ( )

staticprivate

确保全局域注册表已创建 / Ensure the global domain registry exists

Definition at line 13 of file topic.cpp.

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

◆ Find()

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

static

按名称查找一个已存在 topic / Find one existing topic by name

Parameters

name	topic 名称 / Topic name
domain	可选主题域 / Optional topic domain

Returns: 找到则返回句柄，否则返回空 / Returns the handle when found, otherwise null

Definition at line 172 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 )

inlinestatic

按精确类型查找或创建一个 topic / Find or create one topic with an exact payload type

Template Parameters

Data	payload 类型 / Payload type

Parameters

name	topic 名称 / Topic name
domain	可选主题域 / Optional topic domain
multi_publisher	是否允许多发布者串行化 / Whether to allow serialized multi-publisher use

Returns: topic 句柄 / Topic handle

Definition at line 391 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
    auto topic = Find(name, domain);
    if (topic != nullptr)
    {
      CheckSubscriberType<Data>(Topic(topic));
      if (multi_publisher && !topic->data_.mutex)
      {
        ASSERT(false);
      }
    }
    else
    {
      topic = CreateTopic<Data>(name, domain, multi_publisher).block_;
    }
    return topic;
  }

◆ GetKey()

uint32_t Topic::GetKey ( ) const

读取 topic 键值 / Read the key value of this topic

Returns: topic 键值 / Topic key value

Definition at line 211 of file topic.cpp.

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

◆ Lock()

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

static

在普通上下文里锁住一个 topic 发布路径 / Lock one topic publish path in normal context

Parameters

topic 目标 topic 句柄 / Target topic handle

Definition at line 32 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))
    {
      // 非 mutex topic 禁止并发发布
      ASSERT(false);
      return;
    }
  }
}

◆ LockFromCallback()

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

static

在回调或 ISR 路径里锁住一个 topic 发布路径 / Lock one topic publish path from callback or ISR context

Parameters

topic 目标 topic 句柄 / Target topic handle

Definition at line 62 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))
    {
      // 回调发布路径要求外围先串行化
      ASSERT(false);
      return;
    }
  }
}

◆ NowTimestamp()

MicrosecondTimestamp Topic::NowTimestamp ( )

static

读取当前时间戳 / Read the current timestamp

Returns: 当前时间戳 / Current timestamp

Definition at line 80 of file publish.cpp.

80{ return Timebase::GetMicroseconds(); }

LibXR::Timebase::GetMicroseconds

static MicrosecondTimestamp GetMicroseconds()

获取当前时间的微秒级时间戳。 Gets the current timestamp in microseconds.

Definition ch32_timebase.cpp:12

◆ operator TopicHandle()

LibXR::Topic::operator TopicHandle ( )

inline

把 topic 视图转换为底层句柄 / Convert this topic view to the underlying handle

Returns: topic 句柄 / Topic handle

Definition at line 601 of file topic.hpp.

601{ return block_; }

◆ PackBytes()

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

staticprivate

将一段 payload 字节和 topic 元数据拼成 packet / Pack one payload byte range together with topic metadata into one packet

Parameters

topic_name_crc32	目标 topic 的 CRC32 键 / CRC32 key of the target topic
buffer	输出原始缓冲区 / Output raw buffer
timestamp	消息时间戳 / Message timestamp
data	待打包的 payload 字节 / Payload bytes to pack

Definition at line 42 of file packet.cpp.

{
  ASSERT(buffer.addr_ != nullptr);
  ASSERT(buffer.size_ >= PACK_BASE_SIZE + data.size_);
 
  auto* 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_.version = PACKET_VERSION;
  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_);
}

◆ PackData() [1/2]

template<typename Data >

ErrorCode LibXR::Topic::PackData	(	const Data &	data,
		PackedData< Data > &	packet )

inline

将一个精确类型消息打包成 packet / Pack one exact-typed message into one packet using the topic's runtime contract and current timestamp

Template Parameters

Data	负载类型 / Payload type

Parameters

data	待打包 payload / Payload to pack
packet	输出 packet 对象 / Output packed message object

Returns: 操作结果错误码 / Error code

Definition at line 525 of file topic.hpp.

  {
    return PackData(data, packet, NowTimestamp());
  }

◆ PackData() [2/2]

template<typename Data >

ErrorCode LibXR::Topic::PackData	(	const Data &	data,
		PackedData< Data > &	packet,
		MicrosecondTimestamp	timestamp )

将一个精确类型消息按指定时间戳打包成 packet / Pack one exact-typed message into one packet with the given timestamp

按当前 topic 的名字和类型契约把一条消息打包成 packet / Pack one message into one packet using the current topic name and type contract

Template Parameters

Data	负载类型 / Payload type

Parameters

data	待打包 payload / Payload to pack
packet	输出 packet 对象 / Output packed message object
timestamp	指定时间戳 / Explicit timestamp

Returns: 操作结果错误码 / Error code

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待打包 payload / Payload to pack
packet	输出 packet / Output packed message
timestamp	要写入包头的时间戳 / Timestamp to store into the packet header

Returns: 操作结果错误码 / Error code

Definition at line 112 of file packet.hpp.

{
  CheckTopicPayload<Data>();
  ASSERT(block_ != nullptr);
  ASSERT(block_->data_.payload_type_id == TypeID::GetID<Data>());
  ASSERT(block_->data_.payload_size == sizeof(Data));
 
  PackBytes(block_->data_.crc32, RawData(packet), timestamp, ConstRawData(data));
  return ErrorCode::OK;
}

◆ PackRaw() [1/2]

ErrorCode LibXR::Topic::PackRaw	(	ConstRawData	data,
		RawData	packet )

inline

将一段 raw payload 按当前 topic 元数据打包成 packet / Pack a raw payload byte view into one packet using the current topic metadata and current timestamp

Parameters

data	待打包 payload 字节 / Payload bytes to pack
packet	输出 packet 缓冲区 / Output packet buffer

Returns: 操作结果错误码 / Error code

Definition at line 551 of file topic.hpp.

  {
    return PackRaw(data, packet, NowTimestamp());
  }

◆ PackRaw() [2/2]

ErrorCode LibXR::Topic::PackRaw	(	ConstRawData	data,
		RawData	packet,
		MicrosecondTimestamp	timestamp )

inline

将一段 raw payload 按当前 topic 元数据和指定时间戳打包成 packet / Pack a raw payload byte view into one packet using the current topic metadata and timestamp

Parameters

data	待打包 payload 字节 / Payload bytes to pack
packet	输出 packet 缓冲区 / Output packet buffer
timestamp	指定时间戳 / Explicit timestamp

Returns: 操作结果错误码 / Error code

Definition at line 565 of file topic.hpp.

  {
    if (block_ == nullptr || data.addr_ == nullptr || packet.addr_ == nullptr)
    {
      return ErrorCode::PTR_NULL;
    }
 
    if (data.size_ != block_->data_.payload_size)
    {
      return ErrorCode::SIZE_ERR;
    }
 
    if (packet.size_ < PACK_BASE_SIZE + data.size_)
    {
      return ErrorCode::NO_BUFF;
    }
 
    PackBytes(block_->data_.crc32, packet, timestamp, data);
    return ErrorCode::OK;
  }

◆ PayloadAlignment()

size_t LibXR::Topic::PayloadAlignment ( ) const

inlinenodiscard

获取该 topic payload 的对齐要求 / Get the payload alignment requirement of this topic

Returns: payload 对齐要求 / Payload alignment requirement

Definition at line 426 of file topic.hpp.

  {
    ASSERT(block_ != nullptr);
    return block_->data_.payload_alignment;
  }

◆ PayloadSize()

size_t LibXR::Topic::PayloadSize ( ) const

inlinenodiscard

获取该 topic 固定 payload 字节数 / Get the fixed payload size of this topic

Returns: payload 字节数 / Payload size in bytes

Definition at line 415 of file topic.hpp.

  {
    ASSERT(block_ != nullptr);
    return block_->data_.payload_size;
  }

◆ Publish() [1/2]

template<typename Data >

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

inline

在普通上下文里发布一条消息，并自动取当前时间戳 / Publish one message in normal context and stamp it with the current time

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待发布的 payload 对象 / Payload object to publish

Definition at line 439 of file topic.hpp.

  {
    PublishTyped(data, NowTimestamp(), false, false);
  }

◆ Publish() [2/2]

template<typename Data >

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

inline

在普通上下文里按指定时间戳发布一条消息 / Publish one message in normal context with an explicit timestamp

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待发布的 payload 对象 / Payload object to publish
timestamp	消息时间戳 / Message timestamp

Definition at line 452 of file topic.hpp.

  {
    PublishTyped(data, timestamp, false, false);
  }

◆ PublishBytesFromServer()

void LibXR::Topic::PublishBytesFromServer	(	void *	payload_addr,
		size_t	payload_size,
		MicrosecondTimestamp	timestamp )

inline

供 packet/server 路径按字节发布一条消息 / Publish one message from the packet/server path using bytes already arranged as the exact payload object

Parameters

payload_addr	已满足对齐的 payload 起始地址 / Start address of the already-aligned payload object
payload_size	payload 字节数，必须等于 topic 固定大小 / Payload size, must equal the fixed topic size
timestamp	这条消息的时间戳 / Timestamp of this message

Definition at line 493 of file topic.hpp.

  {
    PublishServerBytes(payload_addr, payload_size, timestamp, false, false);
  }

◆ PublishBytesFromServerCallback()

void LibXR::Topic::PublishBytesFromServerCallback	(	void *	payload_addr,
		size_t	payload_size,
		MicrosecondTimestamp	timestamp,
		bool	in_isr )

inline

供回调/ISR 上下文里的 packet/server 路径按字节发布一条消息 / Publish one packet/server message from callback/ISR context using bytes already arranged as the exact payload object

Parameters

payload_addr	已满足对齐的 payload 起始地址 / Start address of the already-aligned payload object
payload_size	payload 字节数，必须等于 topic 固定大小 / Payload size, must equal the fixed topic size
timestamp	这条消息的时间戳 / Timestamp of this message
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 510 of file topic.hpp.

  {
    PublishServerBytes(payload_addr, payload_size, timestamp, true, in_isr);
  }

◆ PublishFromCallback() [1/2]

template<typename Data >

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

inline

在回调或 ISR 路径里发布一条消息，并自动取当前时间戳 / Publish one message from callback or ISR context and stamp it with the current time

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待发布的 payload 对象 / Payload object to publish
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 465 of file topic.hpp.

  {
    PublishTyped(data, NowTimestamp(), true, in_isr);
  }

◆ PublishFromCallback() [2/2]

template<typename Data >

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

inline

在回调或 ISR 路径里按指定时间戳发布一条消息 / Publish one message from callback or ISR context with an explicit timestamp

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待发布的 payload 对象 / Payload object to publish
timestamp	消息时间戳 / Message timestamp
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 479 of file topic.hpp.

  {
    PublishTyped(data, timestamp, true, in_isr);
  }

◆ PublishServerBytes()

void LibXR::Topic::PublishServerBytes	(	void *	payload_addr,
		size_t	payload_size,
		MicrosecondTimestamp	timestamp,
		bool	from_callback,
		bool	in_isr )

inlineprivate

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

Parameters

payload_addr	已满足对齐的 payload 地址 / Already-aligned payload address
payload_size	payload 字节数 / Payload size in bytes
timestamp	消息时间戳 / Message timestamp
from_callback	是否来自回调发布路径 / Whether this publish comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 792 of file topic.hpp.

  {
    CheckServerPublishContract(block_, payload_addr, payload_size);
 
    if (from_callback)
    {
      LockFromCallback(block_);
    }
    else
    {
      Lock(block_);
    }
 
    DispatchSubscribers(block_, timestamp, payload_addr, from_callback, in_isr);
 
    if (from_callback)
    {
      UnlockFromCallback(block_);
    }
    else
    {
      Unlock(block_);
    }
  }

◆ PublishTyped()

template<typename Data >

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

inlineprivate

强类型发布入口的共享实现 / Shared implementation of typed publish entry points

Template Parameters

Data	payload 类型 / Payload type

Parameters

data	待发布 payload / Payload to publish
timestamp	消息时间戳 / Message timestamp
from_callback	是否来自回调路径 / Whether this publish comes from callback path
in_isr	当前是否位于 ISR / Whether the current path is in ISR context

Definition at line 703 of file topic.hpp.

  {
    CheckTopicPayload<Data>();
 
    if (from_callback)
    {
      LockFromCallback(block_);
    }
    else
    {
      Lock(block_);
    }
 
    CheckPublishContract(block_, TypeID::GetID<Data>(), sizeof(Data), alignof(Data));
    DispatchSubscribers(block_, timestamp, &data, from_callback, in_isr);
 
    if (from_callback)
    {
      UnlockFromCallback(block_);
    }
    else
    {
      Unlock(block_);
    }
  }

◆ RegisterCallback()

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

inline

注册一个回调订阅者 / Register one callback subscriber

注册回调函数 / Register a callback function

Parameters

cb	要注册的回调句柄 / Callback handle to register
cb	需要注册的回调函数 / 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 606 of file callback.hpp.

{
  if (!cb.IsRawPayloadView())
  {
    ASSERT(block_->data_.payload_type_id == cb.PayloadTypeID());
  }
 
  auto node = new (std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT))
      LockFreeList::Node<CallbackBlock>(cb, block_->data_.payload_size);
  block_->data_.subers.Add(*node);
}

◆ Unlock()

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

static

在普通上下文里释放一个 topic 发布路径 / Unlock one topic publish path in normal context

Parameters

topic 目标 topic 句柄 / Target topic handle

Definition at line 50 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

在回调或 ISR 路径里释放一个 topic 发布路径 / Unlock one topic publish path from callback or ISR context

Parameters

topic 目标 topic 句柄 / Target topic handle

Definition at line 80 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

等待指定名称的 topic 出现 / Wait until a topic with the given name exists

Parameters

name	topic 名称 / Topic name
timeout	等待超时，默认永久等待 / Timeout in ticks, default wait forever
domain	可选主题域 / Optional topic domain

Returns: 找到则返回句柄，否则返回空 / Returns the handle when found, otherwise null

Definition at line 190 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 视图绑定的状态块。Runtime state block bound to the current topic view.

Definition at line 610 of file topic.hpp.

◆ def_domain_

Domain* LibXR::Topic::def_domain_ = nullptr

inlinestaticprivate

缺省 topic 域。Default topic domain.

Definition at line 615 of file topic.hpp.

◆ domain_

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

inlinestaticprivate

Initial value:

=

nullptr

全局 topic 域注册表。Global registry of topic domains.

Definition at line 613 of file topic.hpp.

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

src/middleware/message/topic.hpp
src/middleware/message/packet/packet.cpp
src/middleware/message/packet/packet.hpp
src/middleware/message/publish.cpp
src/middleware/message/subscriber/callback.hpp
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

◆ AllocateSubscriberBuffer()

◆ CheckPublishContract()

◆ CheckServerPublishContract()

◆ CheckSubscriberType()

◆ CopyPayload()

◆ CreateTopic()

◆ DispatchSubscriber()

◆ DispatchSubscribers()

◆ EnsureDefaultDomain()

◆ EnsureDomainRegistry()

◆ Find()

◆ FindOrCreate()

◆ GetKey()

◆ Lock()

◆ LockFromCallback()

◆ NowTimestamp()

◆ operator TopicHandle()

◆ PackBytes()

◆ PackData() [1/2]

◆ PackData() [2/2]

◆ PackRaw() [1/2]

◆ PackRaw() [2/2]

◆ PayloadAlignment()

◆ PayloadSize()

◆ Publish() [1/2]

◆ Publish() [2/2]

◆ PublishBytesFromServer()

◆ PublishBytesFromServerCallback()

◆ PublishFromCallback() [1/2]

◆ PublishFromCallback() [2/2]

◆ PublishServerBytes()

◆ PublishTyped()

◆ RegisterCallback()

◆ Unlock()

◆ UnlockFromCallback()

◆ WaitTopic()

Field Documentation

◆ block_

◆ def_domain_

◆ domain_