开始
这一节定义了DDS和RTPS的概念,也提供了一个逐步讲解的教程,这个教程中讲解了如何开发一个简单的FastDDS发布订阅应用程序。
1.1 什么是DDS
DDS是一个以数据为中心的通信一些,主要用在分布式软件的通信领域。它定义了应用程序的通信API和通信语义,这使得数据提供者和数据消费者之间通信成为可能。
因为它是一个DCPS模型,DDS的实现中定义了3个关键的应用实体:发布实体、订阅实体、配置实体(它定义了消息类型、创建带有QoS的发布者和订阅者,用来确保发布者和订阅者之间正确的数据传输)。
DDS使用QoS开定义DDS实体之间的行为特征,QoS是由单个的QoS策略组成的(QoSPolicy类的子类的对象),这些都在Policy一节进行了详细定义。
1.1.1 DCPS的概念模型
在dcps的模型中,为了开发一个通信系统,必须定义4个基本的元素。
- Publisher. 它是负责创建和配置DataWriter的实体。DataWriter是负责实际发布消息的实体。每个DataWriter都被指定了一个Topic,而消息就是通过这个Topic进行发布的。从Publisher这一节会详细讲解。
- Subscriber. 它是负责接收自己订阅的topic中的消息的实体。它服务于一个或者多个DataReader对象,DataReader对象是负责为应用程序读取数据的。在Subscriber这一节会详细解释。
- Topic. 它将发布和订阅连接起来,在一个domain中它是唯一的。通过TopicDescription,可以让发布和订阅的数据统一。在topic一节详细讲解。
- Domain. 这是把所有的publisher和subscriber连接起来的概念,属于一个或者多个应用程序,这些应用程序在多个topic下交换数据。这些处在一个domain中的应用程序被称为DomainParticipant.Domain是被domainid唯一标识的。DomainParticipant通过定义domainId来指定dds域。两个不同的domainparticipant即时在同一个网络中,也是不知道彼此的。因此,多个通信通道都可以被创建。domainparicipant是其他dcps实体的容器,是publisher、subscriber和topc等实体的创建者,而且在domain中提供管理服务。在domain一节进行详细了解。
这些元素在下图中详细展示:
1.2 什么是RTPS
RTPS协议开发出来是为了支撑DDS应用程序,它是一个发布订阅通信中间件,构建在尽力交付的运输层UDP/IP之上。而且,FastDDS也支持TCP和共享内存(SHM Shared Memory)运输层。
RTPS既可以支持单播通信,也可以支持多播通信。
在RTPS的上层,可以看到有Domain,这个domain是从dds继承而来的,它定义了一个隔离的通信平面。同一时刻多个domain可以独立存在。一个domain中可以包含任意数量的RTPSParticipant,RTPSParticipant是能够发送和接收数据的元素。为此,RTPSParticipant使用它们的Endpoint:
- RTPSWriter:发送数据的Endpoint。
- RTPSReader:接收数据的Endpoint。
一个RTPSParticipant可以包含任意多个writer和reader endpoint。
通信围绕topic进行,topic中定义了数据。topc是不属于任何一个具体的participant的。参与者通过RTPSWriter对主题下发布的数据进行更改,并通过RTPSReader接收与其订阅的主题相关的数据。通信单元称为Change,它表示在Topic下写入的数据的更新。RTPSReader/RTPSWriter在其历史记录History中注册这些更改,历史记录是一种数据结构,用作最近更改的缓存。
在eProsima Fast DDS的默认配置中,当您通过RTPSWriter端点发布更改时,会在幕后执行以下步骤:
- 1.Change将添加到RTPSWriter的历史缓存中
- 2.RTPSWriter将更改发送给它知道的任何RTPSReader。
- 3.接收到数据后,RTPSReader将使用新的Change更新其历史缓存。
但是,Fast DDS支持多种配置,允许您更改RTPSWriter/RTPSReader的行为。修改RTPS实体的默认配置意味着RTPSWriter和RTPSReader之间的数据交换流发生了变化。此外,通过选择服务质量(QoS)策略,可以以多种方式影响这些历史缓存的管理方式,但通信循环保持不变。您可以继续阅读RTPS层一节,以了解有关快速DDS中RTPS协议实现的更多信息。
1.3 编写一个简单的C++发布者和订阅者应用程序
本节详细介绍了如何使用C++API一步一步地创建一个简单的FastDDS应用程序,其中包含发布者和订阅者。也可以使用eProsima Fast DDS Gen工具自行生成与本节中实现的示例类似的示例。在构建发布/订阅应用程序中解释了这种额外的方法。
1.3.1 背景
DDS是一种以数据为中心的通信中间件,它实现了DCPS模型。该模型基于发布者的开发,这是一个数据生成元素;以及订户、数据消费元件。这些实体通过主题进行通信,主题是绑定两个DDS实体的元素。发布者在主题下生成信息,订阅者订阅该主题以接收信息。
1.3.2 前提条件
首先,您需要按照安装手册中概述的步骤安装eProsima Fast DDS及其所有依赖项。您还需要完成《安装手册》中概述的安装eProsima Fast DDS Gen工具的步骤。此外,本教程中提供的所有命令都是针对Linux环境的概述。
1.3.3 创建应用程序工作空间
在项目结束时,应用程序工作区将具有以下结构。文件build/DDSHelloWorldPublisher和build/DDSelloWorldSubscriber分别是发布服务器应用程序和订阅服务器应用程序。
.
└── workspace_DDSHelloWorld
├── build
│ ├── CMakeCache.txt
│ ├── CMakeFiles
│ ├── cmake_install.cmake
│ ├── DDSHelloWorldPublisher
│ ├── DDSHelloWorldSubscriber
│ └── Makefile
├── CMakeLists.txt
└── src
├── HelloWorld.cxx
├── HelloWorld.h
├── HelloWorld.idl
├── HelloWorldPublisher.cpp
├── HelloWorldPubSubTypes.cxx
├── HelloWorldPubSubTypes.h
└── HelloWorldSubscriber.cpp
让我们首先创建目录树:
mkdir workspace_DDSHelloWorld && cd workspace_DDSHelloWorld
mkdir src build
1.3.4 导入链接库及其依赖项
DDS应用程序需要Fast DDS和Fast CDR库。根据所遵循的安装过程,在DDS应用程序使用这些库的过程将略有不同。
1.3.4.1 从二进制文件安装和手动安装
如果我们从二进制文件或手动安装完成了安装,那么这些库已经可以从工作区访问。在Linux上,可以分别在用于Fast DDS和Fast CDR的/usr/include/fastrtps/和/usr/include/Fast CDR/目录中找到头文件。两者的编译库都可以在/usr/lib/目录中找到。
1.3.4.2 Colcon安装
从Colcon安装中,有几种方法可以导入库。如果库需要仅用于当前会话,请运行以下命令。
source <path/to/Fast-DDS/workspace>/install/setup.bash
通过将Fast DDS安装目录添加到当前用户运行以下命令的shell配置文件中的$PATH变量,可以从任何会话访问它们。
echo 'source <path/to/Fast-DDS/workspace>/install/setup.bash' >> ~/.bashrc
这将在该用户每次登录后设置环境。
1.3.5 配置CMake工程
我们将使用CMake工具来管理项目的构建。使用首选的文本编辑器,创建一个名为CMakeLists.txt的新文件,然后复制并粘贴以下代码段。将此文件保存在工作区的根目录中。如果您遵循了这些步骤,那么应该是workspace_DDSHelloWorld。
cmake_minimum_required(VERSION 3.12.4)
if(NOT CMAKE_VERSION VERSION_LESS 3.0)
cmake_policy(SET CMP0048 NEW)
endif()
project(DDSHelloWorld)
# Find requirements
if(NOT fastcdr_FOUND)
find_package(fastcdr REQUIRED)
endif()
if(NOT fastrtps_FOUND)
find_package(fastrtps REQUIRED)
endif()
# Set C++11
include(CheckCXXCompilerFlag)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_COMPILER_IS_CLANG OR
CMAKE_CXX_COMPILER_ID MATCHES "Clang")
check_cxx_compiler_flag(-std=c++11 SUPPORTS_CXX11)
if(SUPPORTS_CXX11)
add_compile_options(-std=c++11)
else()
message(FATAL_ERROR "Compiler doesn't support C++11")
endif()
endif()
在每一节中,我们将完善此文件以包括特定生成的文件。
1.3.6. 构建topc数据类型
eProsima Fast DDS Gen是一个Java应用程序,它使用接口描述语言(IDL)文件中定义的数据类型生成源代码。此应用程序可以执行两种不同的操作:
- 1 为自定义topic生成C++定义。
- 2 生成使用topic数据的函数示例。
在本教程中将遵循前者。要查看后者的应用示例,可以查看另一个示例。有关详细信息,请参见简介。对于这个项目,我们将使用Fast DDS Gen应用程序来定义发布者发送和订阅者接收的消息的数据类型。
在工作区目录中,执行以下命令:
cd src && touch HelloWorld.idl
这将在src目录中创建HelloWorld.idl文件。在文本编辑器中打开文件,复制并粘贴以下代码段。
struct HelloWorld
{
unsigned long index;
string message;
};
通过这样做,我们定义了HelloWorld数据类型,它有两个元素:uint32_t类型的索引和std::string类型的消息。剩下的就是生成在C++11中实现这种数据类型的源代码。为此,从src目录运行以下命令。
<path/to/Fast DDS-Gen>/scripts/fastddsgen HelloWorld.idl
这必须已生成以下文件:
-
HelloWorld.cxx: HelloWorld 类型定义.
-
HelloWorld.h: HelloWorld.cxx的头文件.
-
HelloWorldPubSubTypes.cxx: HelloWorld类型的序列化和反序列化代码.
-
HelloWorldPubSubTypes.h: HelloWorldPubSubTypes.cxx的头文件.
1.3.6.1 CMakeLists.txt
在前面创建的CMakeList.txt文件末尾包含以下代码段。这包括我们刚刚创建的文件。
message(STATUS "Configuring HelloWorld publisher/subscriber example...")
file(GLOB DDS_HELLOWORLD_SOURCES_CXX "src/*.cxx")
1.3.7 编写Fast DDS发布者
从工作区的src目录中,运行以下命令下载HelloWorldPublisher.cpp文件。
wget -O HelloWorldPublisher.cpp https://raw.githubusercontent.com/eProsima/Fast-RTPS-docs/master/code/Examples/C++/DDSHelloWorld/src/HelloWorldPublisher.cpp
这是发布者应用程序的C++源代码。它将在主题HelloWorldTopic下发送10个消息。
// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/**
* @file HelloWorldPublisher.cpp
*
*/
#include "HelloWorldPubSubTypes.h"
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
using namespace eprosima::fastdds::dds;
class HelloWorldPublisher
{
private:
HelloWorld hello_;
DomainParticipant* participant_;
Publisher* publisher_;
Topic* topic_;
DataWriter* writer_;
TypeSupport type_;
class PubListener : public DataWriterListener
{
public:
PubListener()
: matched_(0)
{
}
~PubListener() override
{
}
void on_publication_matched(
DataWriter*,
const PublicationMatchedStatus& info) override
{
if (info.current_count_change == 1)
{
matched_ = info.total_count;
std::cout << "Publisher matched." << std::endl;
}
else if (info.current_count_change == -1)
{
matched_ = info.total_count;
std::cout << "Publisher unmatched." << std::endl;
}
else
{
std::cout << info.current_count_change
<< " is not a valid value for PublicationMatchedStatus current count change." << std::endl;
}
}
std::atomic_int matched_;
} listener_;
public:
HelloWorldPublisher()
: participant_(nullptr)
, publisher_(nullptr)
, topic_(nullptr)
, writer_(nullptr)
, type_(new HelloWorldPubSubType())
{
}
virtual ~HelloWorldPublisher()
{
if (writer_ != nullptr)
{
publisher_->delete_datawriter(writer_);
}
if (publisher_ != nullptr)
{
participant_->delete_publisher(publisher_);
}
if (topic_ != nullptr)
{
participant_->delete_topic(topic_);
}
DomainParticipantFactory::get_instance()->delete_participant(participant_);
}
//!Initialize the publisher
bool init()
{
hello_.index(0);
hello_.message("HelloWorld");
DomainParticipantQos participantQos;
participantQos.name("Participant_publisher");
participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
if (participant_ == nullptr)
{
return false;
}
// Register the Type
type_.register_type(participant_);
// Create the publications Topic
topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
if (topic_ == nullptr)
{
return false;
}
// Create the Publisher
publisher_ = participant_->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr);
if (publisher_ == nullptr)
{
return false;
}
// Create the DataWriter
writer_ = publisher_->create_datawriter(topic_, DATAWRITER_QOS_DEFAULT, &listener_);
if (writer_ == nullptr)
{
return false;
}
return true;
}
//!Send a publication
bool publish()
{
if (listener_.matched_ > 0)
{
hello_.index(hello_.index() + 1);
writer_->write(&hello_);
return true;
}
return false;
}
//!Run the Publisher
void run(
uint32_t samples)
{
uint32_t samples_sent = 0;
while (samples_sent < samples)
{
if (publish())
{
samples_sent++;
std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
<< " SENT" << std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds( 0));
}
}
};
int main(
int argc,
char** argv)
{
std::cout << "Starting publisher." << std::endl;
int samples = 10;
HelloWorldPublisher* mypub = new HelloWorldPublisher();
if(mypub->init())
{
mypub->run(static_cast<uint32_t>(samples));
}
delete mypub;
return 0;
}
1.3.7.1 检查呆马
在文件的开头,我们有一个Doxygen风格的注释块,带有@file字段,告诉文件的名称。
/**
* @file HelloWorldPublisher.cpp
*
*/
下面是C++头文件的内容。第一个文件包括HelloWorldPubSubTypes.h文件,其中包含我们在上一节中定义的数据类型的序列化和反序列化函数。
#include "HelloWorldPubSubTypes.h"
下一块包含允许使用Fast DDS API的C++头文件。
-
DomainParticipantFactory. 允许创建和销毁DomainParticipant对象。
-
DomainParticipant. 充当所有其他实体对象的容器,以及发布服务器、订阅服务器和主题对象的工厂。
-
TypeSupport. 为参与者提供序列化、反序列化和获取特定数据类型的密钥的函数。
-
Publisher. 这是负责创建DataWriter的对象。
-
DataWriter. 允许应用程序设置要在给定主题下发布的数据的值。
-
DataWriterListener. 允许重新定义DataWriterListener的函数。
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
接下来,我们定义包含eProsima Fast DDS类和函数的命名空间,这些类和函数将在我们的应用程序中使用。
using namespace eprosima::fastdds::dds;
下一行创建实现发布者的HelloWorldPublisher类。
class HelloWorldPublisher
继续看类的私有数据成员,hello_数据成员被定义为HelloWorld类的一个对象,该对象定义了我们使用IDL文件创建的数据类型。接下来,定义与participant、publisher、topic、DataWriter和data type相对应的私有数据成员。TypeSupport类的type_对象是将用于在DomainParticipant中注册主题数据类型的对象。
private:
HelloWorld hello_;
DomainParticipant* participant_;
Publisher* publisher_;
Topic* topic_;
DataWriter* writer_;
TypeSupport type_;
然后,通过从DataWriterListener类继承来定义PubListener类。此类重写默认的DataWriter侦听器回调,该回调允许在发生事件时执行例程。重写的回调on_publication_matched()允许在检测到新的DataReader时定义一系列操作。info.current_count_change()检测与DataWriter匹配的DataReader的这些更改。这是MatchedStatus结构中的一个成员,允许跟踪订阅状态的更改。最后,类的listener_对象被定义为PubListener的实例。
class PubListener : public DataWriterListener
{
public:
PubListener()
: matched_(0)
{
}
~PubListener() override
{
}
void on_publication_matched(
DataWriter*,
const PublicationMatchedStatus& info) override
{
if (info.current_count_change == 1)
{
matched_ = info.total_count;
std::cout << "Publisher matched." << std::endl;
}
else if (info.current_count_change == -1)
{
matched_ = info.total_count;
std::cout << "Publisher unmatched." << std::endl;
}
else
{
std::cout << info.current_count_change
<< " is not a valid value for PublicationMatchedStatus current count change." << std::endl;
}
}
std::atomic_int matched_;
} listener_;
HelloWorldPublisher类的公共构造函数和析构函数定义如下。构造函数将类的私有数据成员初始化为nullptr,TypeSupport对象除外,该对象被初始化为HelloWorldPubSubType类的实例。类析构函数删除这些数据成员,从而清理系统内存。
HelloWorldPublisher()
: participant_(nullptr)
, publisher_(nullptr)
, topic_(nullptr)
, writer_(nullptr)
, type_(new HelloWorldPubSubType())
{
}
virtual ~HelloWorldPublisher()
{
if (writer_ != nullptr)
{
publisher_->delete_datawriter(writer_);
}
if (publisher_ != nullptr)
{
participant_->delete_publisher(publisher_);
}
if (topic_ != nullptr)
{
participant_->delete_topic(topic_);
}
DomainParticipantFactory::get_instance()->delete_participant(participant_);
}
继续HelloWorldPublisher类的公共成员函数,下一段代码定义了公共发布者的初始化成员函数。此函数执行几个操作:
- 初始化HelloWorld类型hello_ 结构成员的内容。
- 通过DomainParticipant的QoS为参与者分配名称。
- 使用DomainParticipantFactory创建参与者。
- 注册IDL中定义的数据类型。
- 为发布者创建topic.
- 创建publisher。
- 使用先前创建的侦听器创建DataWriter。
如您所见,除了参与者的名称之外,所有实体的QoS配置都是默认配置(participant_QoS_default、PUBLISHER_QoS_DEF、TOPIC_QoS_AULT、DATAWRITER_QoS_DEFULT)。可以在DDS标准中检查每个DDS实体的QoS默认值。
//!Initialize the publisher
bool init()
{
hello_.index(0);
hello_.message("HelloWorld");
DomainParticipantQos participantQos;
participantQos.name("Participant_publisher");
participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
if (participant_ == nullptr)
{
return false;
}
// Register the Type
type_.register_type(participant_);
// Create the publications Topic
topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
if (topic_ == nullptr)
{
return false;
}
// Create the Publisher
publisher_ = participant_->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr);
if (publisher_ == nullptr)
{
return false;
}
// Create the DataWriter
writer_ = publisher_->create_datawriter(topic_, DATAWRITER_QOS_DEFAULT, &listener_);
if (writer_ == nullptr)
{
return false;
}
return true;
}
为了发布,实现了公共成员函数publish()。在DataWriter的侦听器回调中,数据成员matched_将更新, 该回调表示DataWriter已与侦听发布主题的DataReader匹配。 它包含发现的DataReader的数量。因此,当发现第一个DataReader时,应用程序开始发布。这只是DataWriter对象对更改的写入。
//!Send a publication
bool publish()
{
if (listener_.matched_ > 0)
{
hello_.index(hello_.index() + 1);
writer_->write(&hello_);
return true;
}
return false;
}
公共运行函数执行发布给定次数的操作,在发布之间等待1秒.
//!Run the Publisher
void run(
uint32_t samples)
{
uint32_t samples_sent = 0;
while (samples_sent < samples)
{
if (publish())
{
samples_sent++;
std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
<< " SENT" << std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
}
最后,HelloWorldPublisher被初始化并在main中运行。
int main(int argc, char** argv)
{
std::cout << "Starting publisher." << std::endl;
int samples = 10;
HelloWorldPublisher* mypub = new HelloWorldPublisher();
if(mypub->init())
{
mypub->run(static_cast<uint32_t>(samples));
}
delete mypub;
return 0;
}
1.3.7.2 CMakeLists.txt
在前面创建的CMakeList.txt文件的末尾包含以下代码段。这将添加构建可执行文件所需的所有源文件,并将可执行文件和库链接在一起。
add_executable(DDSHelloWorldPublisher src/HelloWorldPublisher.cpp ${DDS_HELLOWORLD_SOURCES_CXX})
target_link_libraries(DDSHelloWorldPublisher fastrtps fastcdr)
此时,项目已准备好构建、编译和运行发布者应用程序。在工作区的构建目录中,运行以下命令。
cmake ..
cmake --build .
./DDSHelloWorldPublisher
1.3.8 编写FastDDS订阅者
在工作区的src目录中,执行以下命令以下载HelloWorldSubscriber.cpp文件。
wget -O HelloWorldSubscriber.cpp https://raw.githubusercontent.com/eProsima/Fast-RTPS-docs/master/code/Examples/C++/DDSHelloWorld/src/HelloWorldSubscriber.cpp
这是订阅者应用程序的C++源代码。应用程序运行订阅者,直到收到主题HelloWorldTopic下的10个样本。此时,用户停止。
// Copyright 16 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/**
* @file HelloWorldSubscriber.cpp
*
*/
#include "HelloWorldPubSubTypes.h"
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/DataReaderListener.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
using namespace eprosima::fastdds::dds;
class HelloWorldSubscriber
{
private:
DomainParticipant* participant_;
Subscriber* subscriber_;
DataReader* reader_;
Topic* topic_;
TypeSupport type_;
class SubListener : public DataReaderListener
{
public:
SubListener()
: samples_(0)
{
}
~SubListener() override
{
}
void on_subscription_matched(
DataReader*,
const SubscriptionMatchedStatus& info) override
{
if (info.current_count_change == 1)
{
std::cout << "Subscriber matched." << std::endl;
}
else if (info.current_count_change == -1)
{
std::cout << "Subscriber unmatched." << std::endl;
}
else
{
std::cout << info.current_count_change
<< " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
}
}
void on_data_available(
DataReader* reader) override
{
SampleInfo info;
if (reader->take_next_sample(&hello_, &info) == ReturnCode_t::RETCODE_OK)
{
if (info.valid_data)
{
samples_++;
std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
<< " RECEIVED." << std::endl;
}
}
}
HelloWorld hello_;
std::atomic_int samples_;
} listener_;
public:
HelloWorldSubscriber()
: participant_(nullptr)
, subscriber_(nullptr)
, topic_(nullptr)
, reader_(nullptr)
, type_(new HelloWorldPubSubType())
{
}
virtual ~HelloWorldSubscriber()
{
if (reader_ != nullptr)
{
subscriber_->delete_datareader(reader_);
}
if (topic_ != nullptr)
{
participant_->delete_topic(topic_);
}
if (subscriber_ != nullptr)
{
participant_->delete_subscriber(subscriber_);
}
DomainParticipantFactory::get_instance()->delete_participant(participant_);
}
//!Initialize the subscriber
bool init()
{
DomainParticipantQos participantQos;
participantQos.name("Participant_subscriber");
participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
if (participant_ == nullptr)
{
return false;
}
// Register the Type
type_.register_type(participant_);
// Create the subscriptions Topic
topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
if (topic_ == nullptr)
{
return false;
}
// Create the Subscriber
subscriber_ = participant_->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr);
if (subscriber_ == nullptr)
{
return false;
}
// Create the DataReader
reader_ = subscriber_->create_datareader(topic_, DATAREADER_QOS_DEFAULT, &listener_);
if (reader_ == nullptr)
{
return false;
}
return true;
}
//!Run the Subscriber
void run(
uint32_t samples)
{
while(listener_.samples_ < samples)
{
std::this_thread::sleep_for(std::chrono::milliseconds( ));
}
}
};
int main(
int argc,
char** argv)
{
std::cout << "Starting subscriber." << std::endl;
int samples = ;
HelloWorldSubscriber* mysub = new HelloWorldSubscriber();
if(mysub->init())
{
mysub->run(static_cast<uint32_t>(samples));
}
delete mysub;
return 0;
}
1.3.8.1 检查代码
由于发布者和订阅者应用程序的源代码基本相同,本文档将重点介绍它们之间的主要区别,省略已经解释过的部分代码。遵循与发布者说明中相同的结构,第一步是包含C++头文件。在这些文件中,包含发布者类的文件由订阅者类替换,数据写入器类由数据读取器类替换。
-
Subscriber. 它是负责创建和配置DataReader的对象。
-
DataReader. 它是负责实际接收数据的对象。它在应用程序中注册主题(TopicDescription),该主题标识要读取的数据并访问订户接收的数据。
-
DataReaderListener. 这是分配给数据读取器的侦听器。
-
DataReaderQoS. 定义DataReader的QoS的结构。
-
SampleInfo. “读取”或“获取”每个样本所附带的信息DataReader的QoS。
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
下一行定义实现订阅者的HelloWorldSubscriber类。
class HelloWorldSubscriber
从类的私有数据成员开始,值得一提的是数据读取器侦听器的实现。类的私有数据成员将是参与者、订阅者、主题、数据读取器和数据类型。与数据编写器一样,侦听器实现了在事件发生时要执行的回调。SubListener的第一个重写回调是on_subscription_matched(),它是DataWriter的on_spublication_matched)回调的模拟。
void on_subscription_matched(
DataReader*,
const SubscriptionMatchedStatus& info) override
{
if (info.current_count_change == 1)
{
std::cout << "Subscriber matched." << std::endl;
}
else if (info.current_count_change == -1)
{
std::cout << "Subscriber unmatched." << std::endl;
}
else
{
std::cout << info.current_count_change
<< " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
}
}
第二个重写回调是on_data_available()。在这种情况下,获取并处理数据读取器可以访问的下一个接收样本以显示其内容。这里定义了SampleInfo类的对象,该对象确定是否已经读取或获取了样本。每次读取样本时,接收的样本计数器都会增加。
void on_data_available(
DataReader* reader) override
{
SampleInfo info;
if (reader->take_next_sample(&hello_, &info) == ReturnCode_t::RETCODE_OK)
{
if (info.valid_data)
{
samples_++;
std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
<< " RECEIVED." << std::endl;
}
}
}
类的公共构造函数和析构函数定义如下。
HelloWorldSubscriber()
: participant_(nullptr)
, subscriber_(nullptr)
, topic_(nullptr)
, reader_(nullptr)
, type_(new HelloWorldPubSubType())
{
}
virtual ~HelloWorldSubscriber()
{
if (reader_ != nullptr)
{
subscriber_->delete_datareader(reader_);
}
if (topic_ != nullptr)
{
participant_->delete_topic(topic_);
}
if (subscriber_ != nullptr)
{
participant_->delete_subscriber(subscriber_);
}
DomainParticipantFactory::get_instance()->delete_participant(participant_);
}
接下来是订阅者初始化公共成员函数。这与为HelloWorldPublisher定义的初始化公共成员函数相同。除参与者名称外,所有实体的QoS配置均为默认QoS(participant_QoS_default、SUBSCRIBER_QoS_DEAULT、TOPIC_QoS_DEF、DATAREADER_QoS-default)。可以在DDS标准中检查每个DDS实体的QoS默认值。
//!Initialize the subscriber
bool init()
{
DomainParticipantQos participantQos;
participantQos.name("Participant_subscriber");
participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
if (participant_ == nullptr)
{
return false;
}
// Register the Type
type_.register_type(participant_);
// Create the subscriptions Topic
topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
if (topic_ == nullptr)
{
return false;
}
// Create the Subscriber
subscriber_ = participant_->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr);
if (subscriber_ == nullptr)
{
return false;
}
// Create the DataReader
reader_ = subscriber_->create_datareader(topic_, DATAREADER_QOS_DEFAULT, &listener_);
if (reader_ == nullptr)
{
return false;
}
return true;
}
公共成员函数run()确保订阅者在收到所有样本之前一直运行。此成员函数实现了用户的主动等待,并具有100ms的睡眠间隔以减轻CPU负担。
//!Run the Subscriber
void run(
uint32_t samples)
{
while(listener_.samples_ < samples)
{
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
最后,实现订阅者的参与者被初始化并在main中运行。
int main(
int argc,
char** argv)
{
std::cout << "Starting subscriber." << std::endl;
int samples = 10;
HelloWorldSubscriber* mysub = new HelloWorldSubscriber();
if(mysub->init())
{
mysub->run(static_cast<uint32_t>(samples));
}
delete mysub;
return 0;
}
1.3.8.2 CMakeLists.txt
在前面创建的CMakeList.txt文件的末尾包含以下代码段。这将添加构建可执行文件所需的所有源文件,并将可执行文件和库链接在一起。
add_executable(DDSHelloWorldSubscriber src/HelloWorldSubscriber.cpp ${DDS_HELLOWORLD_SOURCES_CXX})
target_link_libraries(DDSHelloWorldSubscriber fastrtps fastcdr)
此时,项目已准备好构建、编译和运行订户应用程序。在工作区的构建目录中,运行以下命令。
cmake ..
cmake --build .
./DDSHelloWorldSubscriber
1.3.9 将所有内容放在一起
最后,从构建目录中,从两个终端运行发布者和订阅者应用程序。
./DDSHelloWorldPublisher
./DDSHelloWorldSubscriber
1.3.10 总结
在本教程中,您构建了一个发布者和订阅者DDS应用程序。您还学习了如何为源代码编译构建CMake文件,以及如何在项目中包含和使用Fast DDS和Fast CDR库。
1.3.11 下一步
在eProsima Fast DDS Github存储库中,您可以找到更多复杂的示例,这些示例可以为多种用例和场景实现DDS通信。你可以在这里找到它们。