LibXR::Topic Class Reference

主题（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
	存储主题（Topic）数据的结构体。Structure storing topic data. More...
struct	CallbackBlock
	回调订阅块，继承自 SuberBlock Callback subscription block, inheriting from SuberBlock More...
class	Domain
	主题域（Domain）管理器，用于组织多个主题。Domain manager for organizing multiple topics. More...
struct	QueueBlock
	队列订阅块，继承自 SuberBlock Queue subscription block, inheriting from SuberBlock More...
class	QueuedSubscriber
	构造函数，使用名称和无锁队列进行初始化 Constructor using a name and a lock-free queue More...
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...
typedef RBTree< uint32_t >::Node< Block > *	TopicHandle
	主题句柄，指向存储数据的红黑树节点。Handle pointing to a red-black tree node storing data.
typedef LibXR::Topic::ASyncBlock	ASyncBlock
typedef LibXR::Topic::QueueBlock	QueueBlock
typedef LibXR::Topic::CallbackBlock	CallbackBlock

Public Member Functions
void	RegisterCallback (Callback< RawData & > &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 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	PublishFromCallback (Data &data, bool in_isr)
	从回调函数发布数据 Publishes data from a callback function
void	Publish (void *addr, uint32_t size)
	以原始地址和大小发布数据 Publishes data using raw address and size
void	PublishFromCallback (void *addr, uint32_t size, bool in_isr)
	从回调函数发布数据 Publishes data from a callback function
template<SizeLimitMode Mode = SizeLimitMode::MORE>
ErrorCode	DumpData (RawData data, bool pack=false)
	转储数据 Dump data
template<typename Data >
ErrorCode	DumpData (PackedData< Data > &data)
	转储数据到 PackedData Dumps data into PackedData format
template<typename Data >
ErrorCode	DumpData (Data &data)
	转储数据到普通数据结构 Dumps data into a normal data structure
	operator TopicHandle ()
	将 Topic 转换为 TopicHandle Converts Topic to TopicHandle

Static Public Member Functions
template<typename Data >
static Topic	CreateTopic (const char *name, Domain *domain=nullptr, bool cache=false, bool check_length=true)
	创建一个新的主题 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 cache=false, bool check_length=true)
	在指定域中查找或创建主题 Finds or creates a topic in the specified domain
static void	PackData (uint32_t topic_name_crc32, RawData buffer, RawData source)
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 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 SpinLock	domain_lock_
	主题域访问的自旋锁，确保多线程安全 SpinLock for domain access to ensure thread safety
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.

Definition at line 28 of file message.hpp.

Member Typedef Documentation

TopicHandle

LibXR::Topic::TopicHandle

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

Definition at line 125 of file message.hpp.

Member Enumeration Documentation

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 181 of file message.hpp.

  {
    SYNC,      
    ASYNC,     
    QUEUE,     
    CALLBACK,  
  };

Constructor & Destructor Documentation

Topic() [1/3]

LibXR::Topic::Topic ( )

inline

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

Definition at line 489 of file message.hpp.

{}

Topic() [2/3]

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

inline

构造函数，使用指定名称、最大长度、域及其他选项初始化主题 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)
cache	是否启用缓存（默认为 false）Whether to enable caching (default: false)
check_length	是否检查数据长度（默认为 false）Whether to check data length (default: false)

Definition at line 503 of file message.hpp.

  {
    if (!def_domain_)
    {
      domain_lock_.Lock();
      if (!domain_)
      {
        domain_ =
            new RBTree<uint32_t>([](const uint32_t &a, const uint32_t &b)
                                 { return static_cast<int>(a) - static_cast<int>(b); });
      }
      if (!def_domain_)
      {
        def_domain_ = new Domain("libxr_def_domain");
      }
      domain_lock_.Unlock();
    }
 
    if (domain == nullptr)
    {
      domain = def_domain_;
    }
 
    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);
 
      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_.cache = false;
      block_->data_.check_length = check_length;
 
      domain->node_->data_.Insert(*block_, crc32);
    }
 
    if (cache && !block_->data_.cache)
    {
      EnableCache();
    }
  }

Topic() [3/3]

LibXR::Topic::Topic ( TopicHandle  topic )

inline

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

Parameters

topic

主题句柄 Topic handle

Definition at line 580 of file message.hpp.

: block_(topic) {}

Member Function Documentation

CreateTopic()

template<typename Data >

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

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)
cache	是否启用缓存（默认为 false）Whether to enable caching (default: false)
check_length	是否检查数据长度（默认为 true）Whether to check data length (default: true)

Returns

创建的 Topic 实例 The created Topic instance

Definition at line 569 of file message.hpp.

  {
    return Topic(name, sizeof(Data), domain, cache, check_length);
  }

DumpData() [1/3]

template<typename Data >

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

inline

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

Template Parameters

Data	数据类型 Data type

Parameters

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

Definition at line 893 of file message.hpp.

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

DumpData() [2/3]

template<typename Data >

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

inline

转储数据到 PackedData Dumps data into PackedData format

Template Parameters

Data	数据类型 Data type

Parameters

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

Definition at line 874 of file message.hpp.

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

DumpData() [3/3]

template<SizeLimitMode Mode = SizeLimitMode::MORE>

ErrorCode LibXR::Topic::DumpData ( RawData  data, bool  pack = false  )

inline

转储数据 Dump data

Template Parameters

Mode	数据大小检查模式 Size limit check mode

Parameters

data	需要转储的数据 Data to be dumped
pack	是否打包数据 Pack data

Returns

ErrorCode

Definition at line 824 of file message.hpp.

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

EnableCache()

void LibXR::Topic::EnableCache ( )

inline

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

Definition at line 630 of file message.hpp.

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

Find()

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

inlinestatic

在指定域中查找主题 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

Definition at line 590 of file message.hpp.

  {
    if (domain == 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  cache = false, bool  check_length = true  )

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)
cache	可选的缓存标志位 Optional cache flag (default: false)
check_length	可选的数据长度检查标志位 Optional data length check flag (default: true)

Returns

TopicHandle 主题句柄 Topic handle

Definition at line 615 of file message.hpp.

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

operator TopicHandle()

LibXR::Topic::operator TopicHandle ( )

inline

将 Topic 转换为 TopicHandle Converts Topic to TopicHandle

Returns

TopicHandle

Definition at line 941 of file message.hpp.

{ return block_; }

PackData()

static void LibXR::Topic::PackData ( uint32_t  topic_name_crc32, RawData  buffer, RawData  source  )

inlinestatic

Definition at line 850 of file message.hpp.

  {
    PackedData<uint8_t> *pack = reinterpret_cast<PackedData<uint8_t> *>(buffer.addr_);
 
    memcpy(&pack->raw.data_, source.addr_, source.size_);
 
    pack->raw.header.prefix = 0xa5;
    pack->raw.header.topic_name_crc32 = topic_name_crc32;
    pack->raw.header.data_len = source.size_;
    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 +
                                    source.size_ - sizeof(uint8_t));
    *crc8_pack = CRC8::Calculate(pack, PACK_BASE_SIZE - sizeof(uint8_t) + source.size_);
  }

Publish() [1/2]

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 648 of file message.hpp.

  {
    Publish(&data, sizeof(Data));
  }

Publish() [2/2]

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

inline

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

Parameters

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

Definition at line 672 of file message.hpp.

  {
    block_->data_.mutex.Lock();
    if (block_->data_.check_length)
    {
      ASSERT(size == block_->data_.max_length);
    }
    else
    {
      ASSERT(size <= block_->data_.max_length);
    }
 
    if (block_->data_.cache)
    {
      memcpy(block_->data_.data.addr_, addr, size);
      block_->data_.data.size_ = size;
    }
    else
    {
      block_->data_.data.addr_ = addr;
      block_->data_.data.size_ = size;
    }
 
    RawData data = block_->data_.data;
 
    auto foreach_fun = [&](SuberBlock &block)
    {
      switch (block.type)
      {
        case SuberType::SYNC:
        {
          auto sync = reinterpret_cast<SyncBlock *>(&block);
          memcpy(sync->buff.addr_, data.addr_, data.size_);
          sync->sem.Post();
          break;
        }
        case SuberType::ASYNC:
        {
          auto async = reinterpret_cast<ASyncBlock *>(&block);
          if (async->waiting)
          {
            memcpy(async->buff.addr_, data.addr_, data.size_);
            async->data_ready = true;
          }
          break;
        }
        case SuberType::QUEUE:
        {
          auto queue_block = reinterpret_cast<QueueBlock *>(&block);
          queue_block->fun(data, queue_block->queue, false);
          break;
        }
        case SuberType::CALLBACK:
        {
          auto cb_block = reinterpret_cast<CallbackBlock *>(&block);
          cb_block->cb.Run(false, data);
          break;
        }
      }
      return ErrorCode::OK;
    };
 
    block_->data_.subers.Foreach<SuberBlock>(foreach_fun);
 
    block_->data_.mutex.Unlock();
  }

PublishFromCallback() [1/2]

template<typename Data >

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

inline

从回调函数发布数据 Publishes data from a callback function

Template Parameters

Data	数据类型 Data type

Parameters

data	需要发布的数据 Data to be published
in_isr	是否在中断服务程序（ISR）中发布 Whether publishing inside an ISR

Definition at line 661 of file message.hpp.

  {
    PublishFromCallback(&data, sizeof(Data), in_isr);
  }

PublishFromCallback() [2/2]

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

inline

从回调函数发布数据 Publishes data from a callback function

Parameters

addr	数据的地址 Address of the data
size	数据大小 Size of the data
in_isr	是否在中断服务程序（ISR）中发布 Whether publishing inside an ISR

Definition at line 746 of file message.hpp.

  {
    if (block_->data_.mutex.TryLockInCallback(in_isr) != ErrorCode::OK)
    {
      return;
    }
 
    if (block_->data_.check_length)
    {
      ASSERT(size == block_->data_.max_length);
    }
    else
    {
      ASSERT(size <= block_->data_.max_length);
    }
 
    if (block_->data_.cache)
    {
      memcpy(block_->data_.data.addr_, addr, size);
      block_->data_.data.size_ = size;
    }
    else
    {
      block_->data_.data.addr_ = addr;
      block_->data_.data.size_ = size;
    }
 
    RawData data = block_->data_.data;
 
    auto foreach_fun = [&](SuberBlock &block)
    {
      switch (block.type)
      {
        case SuberType::SYNC:
        {
          auto sync = reinterpret_cast<SyncBlock *>(&block);
          memcpy(sync->buff.addr_, data.addr_, data.size_);
          sync->sem.PostFromCallback(in_isr);
          break;
        }
        case SuberType::ASYNC:
        {
          auto async = reinterpret_cast<ASyncBlock *>(&block);
          memcpy(async->buff.addr_, data.addr_, data.size_);
          async->data_ready = true;
          break;
        }
        case SuberType::QUEUE:
        {
          auto queue_block = reinterpret_cast<QueueBlock *>(&block);
          queue_block->fun(data, queue_block->queue, in_isr);
          break;
        }
        case SuberType::CALLBACK:
        {
          auto cb_block = reinterpret_cast<CallbackBlock *>(&block);
          cb_block->cb.Run(in_isr, data);
          break;
        }
      }
      return ErrorCode::OK;
    };
 
    block_->data_.subers.ForeachFromCallback<SuberBlock>(foreach_fun, in_isr);
 
    block_->data_.mutex.UnlockFromCallback(in_isr);
  }

RegisterCallback()

void LibXR::Topic::RegisterCallback ( Callback< RawData & > &  cb )

inline

注册回调函数 Registers a callback function

Parameters

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

Definition at line 476 of file message.hpp.

  {
    CallbackBlock block;
    block.cb = cb;
    block.type = SuberType::CALLBACK;
    auto node = new List::Node<CallbackBlock>(block);
    block_->data_.subers.Add(*node);
  }

WaitTopic()

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

inlinestatic

等待主题的创建并返回其句柄 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 916 of file message.hpp.

  {
    TopicHandle topic = nullptr;
    do
    {
      topic = Find(name, domain);
      if (topic == nullptr)
      {
        if (timeout <= Thread::GetTime())
        {
          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 1113 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 1131 of file message.hpp.

domain_

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

inlinestaticprivate

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

Definition at line 1119 of file message.hpp.

domain_lock_

SpinLock LibXR::Topic::domain_lock_

inlinestaticprivate

主题域访问的自旋锁，确保多线程安全 SpinLock for domain access to ensure thread safety

Definition at line 1125 of file message.hpp.

---

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

• src/middleware/message.hpp