CyberRt源码剖析--04基于FastRTPS的通信实现

4.1 FastRTPS通信流程

在基于FastRTPS实现跨进程通信时，发送端和接收端需遵循特定流程完成配置，以实现数据通信。其具体步骤如下所述，在发送端，首先要创建RtpsParticipant。接着，创建RtpsWriter的配置信息实例并完成填充，随后创建RtpsWriter和RtpsWriter History，最后将RtpsWriter进行注册。准备工作完成后，就可以进行数据装载与发送。在接收端，同样需要先创建RtpsParticipant，然后创建RtpsReader的配置信息实例并填充相关信息，创建RtpsReader并为其设置回调函数，再创建RtpsReader History，完成这些步骤后，就能在接收到数据时执行回调操作。当发送端和接收端都按照上述流程配置完成，双方即可开始通信。

4.2 基于FastRTPS的发送端的实现

在本通信架构中，RtpsTransmitter负责实现基于FastRTPS的发送端功能，具体设计思路如下：在FastRTPS通信模型中，一个RtpsParticipant可创建多个RtpsWriter。因此无需为每个RtpsTransmitter单独创建RtpsParticipant，同一进程内的所有RtpsTransmitter可共享由Transport单例类创建的全局RtpsParticipant。在具体实现时，本文基于RtpsParticipant的向上继续抽象了一层Participant。

RtpsTransmitter重载的Enable函数按照创建RtpsWriter的流程执行初始化操作，完成后生成RtpsWriter及WriterHistory实例。Enable操作完成后，RtpsTransmitter即可通过Transmit函数发布数据。Transmit函数发送的数据包含两部分：Message（用户定义且支持序列化的数据结构）和MessageInfo（附加消息）。这两部分数据最终被写入WriterHistory。在FastRTPS中，WriterHistory缓存的数据结构由附加消息（WriteParams）和缓存消息（CacheChange_t）组成，其中MessageInfo被转换为WriteParams，Message被转换为CacheChange_t。随后，这两部分数据通过RtpsWriter发送至网络。

4.3 基于FastRTPS的接收端的实现

基于FastRTPS的接收端设计时引入了RtpsReader的抽象层Reader，其内部包含RtpsReader、ReaderHistory和ReaListener三个组件。ReaListener是RtpsReader的回调处理类：当RtpsReader从ReaderHistory读取数据后，ReaListener的onNewCacheChangeAdded函数被自动触发。该函数调用ReaListener内部维护的回调函数（callback）完成处理。通过Reader抽象层，统一了基于FastRTPS接收端的组件设计。

4.4 基于FastRTPS的数据订阅端的实现

数据订阅端由RtpsReceiver和RtpsDispatcher共同构成，二者需与Reader建立关联，具体实现如下：RtpsReceiver的回调函数首先在OnNewMessage函数中进行初步封装，随后通过Enable函数完成注册。注册过程调用RtpsDispatcher的AddListener函数实现，该函数以当前Receiver的channel_id为索引，将RtpsReceiver的OnNewMessage函数注册至前文所述的二级回调索引表。注册完成后，调用AddReader函数创建新的Reader实例，并在创建时将RtpsDispatcher的OnMessage函数指定为Reader的回调函数。由此，当channel上的Reader接收到数据时，onNewCacheChangeAdded函数被触发，继而调用RtpsDispatcher的OnMessage函数。OnMessage函数根据channel_id，在二级回调索引表中依次执行订阅该channel的所有Receiver的OnNewMessage函数，从而实现数据的分发与回调执行。