1.3. Writing a simple C++ publisher and subscriber application¶
This section details how to create a simple Fast DDS application with a publisher and a subscriber using C++ API step by step. It is also possible to self-generate a similar example to the one implemented in this section by using the eProsima Fast DDS-Gen tool. This additional approach is explained in Building a publish/subscribe application.
1.3.1. Background¶
DDS is a data-centric communications middleware that implements the DCPS model. This model is based on the development of a publisher, a data generating element; and a subscriber, a data consuming element. These entities communicate by means of the topic, an element that binds both DDS entities. Publishers generate information under a topic and subscribers subscribe to this same topic to receive information.
1.3.2. Prerequisites¶
First of all, you need to follow the steps outlined in the Installation Manual for the installation of eProsima Fast DDS and all its dependencies. You also need to have completed the steps outlined in the Installation Manual for the installation of the eProsima Fast DDS-Gen tool. Moreover, all the commands provided in this tutorial are outlined for a Linux environment.
1.3.3. Create the application workspace¶
The application workspace will have the following structure at the end of the project.
Files build/DDSHelloWorldPublisher
and build/DDSHelloWorldSubscriber
are the Publisher application and
Subscriber application respectively.
.
└── 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
Let’s create the directory tree first.
mkdir workspace_DDSHelloWorld && cd workspace_DDSHelloWorld
mkdir src build
1.3.4. Import linked libraries and its dependencies¶
The DDS application requires the Fast DDS and Fast CDR libraries. Depending on the installation procedure followed the process of making these libraries available for our DDS application will be slightly different.
1.3.4.1. Installation from binaries and manual installation¶
If we have followed the installation from binaries or the manual installation, these libraries are already accessible from the workspace. On Linux, the header files can be found in directories /usr/include/fastrtps/ and /usr/include/fastcdr/ for Fast DDS and Fast CDR respectively. The compiled libraries of both can be found in the directory /usr/lib/.
1.3.4.2. Colcon installation¶
From a Colcon installation there are several ways to import the libraries. If the libraries need to be available just for the current session, run the following command.
source <path/to/Fast-DDS/workspace>/install/setup.bash
They can be made accessible from any session by adding the Fast DDS installation directory to your $PATH
variable in the shell configuration files for the current user running the following command.
echo 'source <path/to/Fast-DDS/workspace>/install/setup.bash' >> ~/.bashrc
This will set up the environment after each of this user’s logins.
1.3.5. Configure the CMake project¶
We will use the CMake tool to manage the building of the project. With your preferred text editor, create a new file called CMakeLists.txt and copy and paste the following code snippet. Save this file in the root directory of your workspace. If you have followed these steps, it should be 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()
In each section we will complete this file to include the specific generated files.
1.3.6. Build the topic data type¶
eProsima Fast DDS-Gen is a Java application that generates source code using the data types defined in an Interface Description Language (IDL) file. This application can do two different things:
Generate C++ definitions for your custom topic.
Generate a functional example that uses your topic data.
It will be the former that will be followed in this tutorial. To see an example of application of the latter you can check this other example. See Introduction for further details. For this project, we will use the Fast DDS-Gen application to define the data type of the messages that will be sent by the publishers and received by the subscribers.
In the workspace directory, execute the following commands:
cd src && touch HelloWorld.idl
This creates the HelloWorld.idl file in the src directory. Open the file in a text editor and copy and paste the following snippet of code.
struct HelloWorld
{
unsigned long index;
string message;
};
By doing this we have defined the HelloWorld
data type, which has two elements: an index of type uint32_t
and a message of type std::string
.
All that remains is to generate the source code that implements this data type in C++11.
To do this, run the following command from the src
directory.
<path/to/Fast DDS-Gen>/scripts/fastddsgen HelloWorld.idl
This must have generated the following files:
HelloWorld.cxx: HelloWorld type definition.
HelloWorld.h: Header file for HelloWorld.cxx.
HelloWorldPubSubTypes.cxx: Serialization and Deserialization code for the HelloWorld type.
HelloWorldPubSubTypes.h: Header file for HelloWorldPubSubTypes.cxx.
1.3.6.1. CMakeLists.txt¶
Include the following code snippet at the end of the CMakeList.txt file you created earlier. This includes the files we have just created.
message(STATUS "Configuring HelloWorld publisher/subscriber example...")
file(GLOB DDS_HELLOWORLD_SOURCES_CXX "src/*.cxx")
1.3.7. Write the Fast DDS publisher¶
From the src directory in the workspace, run the following command to download the HelloWorldPublisher.cpp file.
wget -O HelloWorldPublisher.cpp \
https://raw.githubusercontent.com/eProsima/Fast-RTPS-docs/master/code/Examples/C++/DDSHelloWorld/src/HelloWorldPublisher.cpp
This is the C++ source code for the publisher application. It is going to send 10 publications under the topic HelloWorldTopic.
1// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15/**
16 * @file HelloWorldPublisher.cpp
17 *
18 */
19
20#include "HelloWorldPubSubTypes.h"
21
22#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
23#include <fastdds/dds/domain/DomainParticipant.hpp>
24#include <fastdds/dds/topic/TypeSupport.hpp>
25#include <fastdds/dds/publisher/Publisher.hpp>
26#include <fastdds/dds/publisher/DataWriter.hpp>
27#include <fastdds/dds/publisher/DataWriterListener.hpp>
28
29using namespace eprosima::fastdds::dds;
30
31class HelloWorldPublisher
32{
33private:
34
35 HelloWorld hello_;
36
37 DomainParticipant* participant_;
38
39 Publisher* publisher_;
40
41 Topic* topic_;
42
43 DataWriter* writer_;
44
45 TypeSupport type_;
46
47 class PubListener : public DataWriterListener
48 {
49 public:
50
51 PubListener()
52 : matched_(0)
53 {
54 }
55
56 ~PubListener() override
57 {
58 }
59
60 void on_publication_matched(
61 DataWriter*,
62 const PublicationMatchedStatus& info) override
63 {
64 if (info.current_count_change == 1)
65 {
66 matched_ = info.total_count;
67 std::cout << "Publisher matched." << std::endl;
68 }
69 else if (info.current_count_change == -1)
70 {
71 matched_ = info.total_count;
72 std::cout << "Publisher unmatched." << std::endl;
73 }
74 else
75 {
76 std::cout << info.current_count_change
77 << " is not a valid value for PublicationMatchedStatus current count change." << std::endl;
78 }
79 }
80
81 std::atomic_int matched_;
82
83 } listener_;
84
85public:
86
87 HelloWorldPublisher()
88 : participant_(nullptr)
89 , publisher_(nullptr)
90 , topic_(nullptr)
91 , writer_(nullptr)
92 , type_(new HelloWorldPubSubType())
93 {
94 }
95
96 virtual ~HelloWorldPublisher()
97 {
98 if (writer_ != nullptr)
99 {
100 publisher_->delete_datawriter(writer_);
101 }
102 if (publisher_ != nullptr)
103 {
104 participant_->delete_publisher(publisher_);
105 }
106 if (topic_ != nullptr)
107 {
108 participant_->delete_topic(topic_);
109 }
110 DomainParticipantFactory::get_instance()->delete_participant(participant_);
111 }
112
113 //!Initialize the publisher
114 bool init()
115 {
116 hello_.index(0);
117 hello_.message("HelloWorld");
118
119 DomainParticipantQos participantQos;
120 participantQos.name("Participant_publisher");
121 participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
122
123 if (participant_ == nullptr)
124 {
125 return false;
126 }
127
128 // Register the Type
129 type_.register_type(participant_);
130
131 // Create the publications Topic
132 topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
133
134 if (topic_ == nullptr)
135 {
136 return false;
137 }
138
139 // Create the Publisher
140 publisher_ = participant_->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr);
141
142 if (publisher_ == nullptr)
143 {
144 return false;
145 }
146
147 // Create the DataWriter
148 writer_ = publisher_->create_datawriter(topic_, DATAWRITER_QOS_DEFAULT, &listener_);
149
150 if (writer_ == nullptr)
151 {
152 return false;
153 }
154 return true;
155 }
156
157 //!Send a publication
158 bool publish()
159 {
160 if (listener_.matched_ > 0)
161 {
162 hello_.index(hello_.index() + 1);
163 writer_->write(&hello_);
164 return true;
165 }
166 return false;
167 }
168
169 //!Run the Publisher
170 void run(
171 uint32_t samples)
172 {
173 uint32_t samples_sent = 0;
174 while (samples_sent < samples)
175 {
176 if (publish())
177 {
178 samples_sent++;
179 std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
180 << " SENT" << std::endl;
181 }
182 std::this_thread::sleep_for(std::chrono::milliseconds(1000));
183 }
184 }
185};
186
187int main(
188 int argc,
189 char** argv)
190{
191 std::cout << "Starting publisher." << std::endl;
192 int samples = 10;
193
194 HelloWorldPublisher* mypub = new HelloWorldPublisher();
195 if(mypub->init())
196 {
197 mypub->run(static_cast<uint32_t>(samples));
198 }
199
200 delete mypub;
201 return 0;
202}
1.3.7.1. Examining the code¶
At the beginning of the file we have a Doxygen style comment block with the @file
field that tells us the name of
the file.
/**
* @file HelloWorldPublisher.cpp
*
*/
Below are the includes of the C++ headers. The first one includes the HelloWorldPubSubTypes.h file with the serialization and deserialization functions of the data type that we have defined in the previous section.
#include "HelloWorldPubSubTypes.h"
The next block includes the C++ header files that allow the use of the Fast DDS API.
DomainParticipantFactory
. Allows for the creation and destruction of DomainParticipant objects.DomainParticipant
. Acts as a container for all other Entity objects and as a factory for the Publisher, Subscriber, and Topic objects.TypeSupport
. Provides the participant with the functions to serialize, deserialize and get the key of a specific data type.Publisher
. It is the object responsible for the creation of DataWriters.DataWriter
. Allows the application to set the value of the data to be published under a given Topic.DataWriterListener
. Allows the redefinition of the functions of the 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>
Next, we define the namespace that contains the eProsima Fast DDS classes and functions that we are going to use in our application.
using namespace eprosima::fastdds::dds;
The next line creates the HelloWorldPublisher
class that implements a publisher.
class HelloWorldPublisher
Continuing with the private data members of the class, the hello_
data member is defined as an object of the
HelloWorld
class that defines the data type
we created with the IDL file.
Next, the private data members corresponding to the participant, publisher, topic, DataWriter and data type are
defined.
The type_
object of the TypeSupport
class is the object that will be used to register the topic data type
in the DomainParticipant.
private:
HelloWorld hello_;
DomainParticipant* participant_;
Publisher* publisher_;
Topic* topic_;
DataWriter* writer_;
TypeSupport type_;
Then, the PubListener
class is defined by inheriting from the DataWriterListener
class.
This class overrides the default DataWriter listener callbacks, which allows the execution of routines in case of an
event.
The overridden callback on_publication_matched()
allows the definition of a series of actions when a new DataReader
is detected listening to the topic under which the DataWriter is publishing.
The info.current_count_change()
detects these changes of DataReaders that are matched to the
DataWriter.
This is a member in the MatchedStatus
structure that allows tracking changes in the status of subscriptions.
Finally, the listener_
object of the class is defined as an instance of 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_;
The public constructor and destructor of the HelloWorldPublisher
class are defined below.
The constructor initializes the private data members of the class to nullptr
, with the exception of the TypeSupport
object, that is initialized as an instance of the HelloWorldPubSubType
class.
The class destructor removes these data members and thus cleans the system memory.
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_);
}
Continuing with the public member functions of the HelloWorldPublisher
class, the next snippet of code defines
the public publisher’s initialization member function.
This function performs several actions:
Initializes the content of the HelloWorld type
hello_
structure members.Assigns a name to the participant through the QoS of the DomainParticipant.
Uses the
DomainParticipantFactory
to create the participant.Registers the data type defined in the IDL.
Creates the topic for the publications.
Creates the publisher.
Creates the DataWriter with the listener previously created.
As you can see, the QoS configuration for all entities, except for the participant’s name, is the default configuration
(PARTICIPANT_QOS_DEFAULT
, PUBLISHER_QOS_DEFAULT
, TOPIC_QOS_DEFAULT
, DATAWRITER_QOS_DEFAULT
).
The default value of the QoS of each DDS Entity can be checked in the
DDS standard.
//!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;
}
To make the publication, the public member function publish()
is implemented.
In the DataWriter’s listener callback which states that the DataWriter has matched with a DataReader
that listens to the publication topic, the data member matched_
is updated. It contains the number of DataReaders
discovered.
Therefore, when the first DataReader has been discovered, the application starts to publish.
This is simply the writing of a change by the DataWriter object.
//!Send a publication
bool publish()
{
if (listener_.matched_ > 0)
{
hello_.index(hello_.index() + 1);
writer_->write(&hello_);
return true;
}
return false;
}
The public run function executes the action of publishing a given number of times, waiting for 1 second between publications.
//!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));
}
}
Finally, the HelloWorldPublisher is initialized and run in 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¶
Include at the end of the CMakeList.txt file you created earlier the following code snippet. This adds all the source files needed to build the executable, and links the executable and the library together.
add_executable(DDSHelloWorldPublisher src/HelloWorldPublisher.cpp ${DDS_HELLOWORLD_SOURCES_CXX})
target_link_libraries(DDSHelloWorldPublisher fastrtps fastcdr)
At this point the project is ready for building, compiling and running the publisher application. From the build directory in the workspace, run the following commands.
cmake ..
cmake --build .
./DDSHelloWorldPublisher
1.3.8. Write the Fast DDS subscriber¶
From the src directory in the workspace, execute the following command to download the HelloWorldSubscriber.cpp file.
wget -O HelloWorldSubscriber.cpp \
https://raw.githubusercontent.com/eProsima/Fast-RTPS-docs/master/code/Examples/C++/DDSHelloWorld/src/HelloWorldSubscriber.cpp
This is the C++ source code for the subscriber application. The application runs a subscriber until it receives 10 samples under the topic HelloWorldTopic. At this point the subscriber stops.
1// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15/**
16 * @file HelloWorldSubscriber.cpp
17 *
18 */
19
20#include "HelloWorldPubSubTypes.h"
21
22#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
23#include <fastdds/dds/domain/DomainParticipant.hpp>
24#include <fastdds/dds/topic/TypeSupport.hpp>
25#include <fastdds/dds/subscriber/Subscriber.hpp>
26#include <fastdds/dds/subscriber/DataReader.hpp>
27#include <fastdds/dds/subscriber/DataReaderListener.hpp>
28#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
29#include <fastdds/dds/subscriber/SampleInfo.hpp>
30
31using namespace eprosima::fastdds::dds;
32
33class HelloWorldSubscriber
34{
35private:
36
37 DomainParticipant* participant_;
38
39 Subscriber* subscriber_;
40
41 DataReader* reader_;
42
43 Topic* topic_;
44
45 TypeSupport type_;
46
47 class SubListener : public DataReaderListener
48 {
49 public:
50
51 SubListener()
52 : samples_(0)
53 {
54 }
55
56 ~SubListener() override
57 {
58 }
59
60 void on_subscription_matched(
61 DataReader*,
62 const SubscriptionMatchedStatus& info) override
63 {
64 if (info.current_count_change == 1)
65 {
66 std::cout << "Subscriber matched." << std::endl;
67 }
68 else if (info.current_count_change == -1)
69 {
70 std::cout << "Subscriber unmatched." << std::endl;
71 }
72 else
73 {
74 std::cout << info.current_count_change
75 << " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
76 }
77 }
78
79 void on_data_available(
80 DataReader* reader) override
81 {
82 SampleInfo info;
83 if (reader->take_next_sample(&hello_, &info) == ReturnCode_t::RETCODE_OK)
84 {
85 if (info.valid_data)
86 {
87 samples_++;
88 std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
89 << " RECEIVED." << std::endl;
90 }
91 }
92 }
93
94 HelloWorld hello_;
95
96 std::atomic_int samples_;
97
98 } listener_;
99
100public:
101
102 HelloWorldSubscriber()
103 : participant_(nullptr)
104 , subscriber_(nullptr)
105 , topic_(nullptr)
106 , reader_(nullptr)
107 , type_(new HelloWorldPubSubType())
108 {
109 }
110
111 virtual ~HelloWorldSubscriber()
112 {
113 if (reader_ != nullptr)
114 {
115 subscriber_->delete_datareader(reader_);
116 }
117 if (topic_ != nullptr)
118 {
119 participant_->delete_topic(topic_);
120 }
121 if (subscriber_ != nullptr)
122 {
123 participant_->delete_subscriber(subscriber_);
124 }
125 DomainParticipantFactory::get_instance()->delete_participant(participant_);
126 }
127
128 //!Initialize the subscriber
129 bool init()
130 {
131 DomainParticipantQos participantQos;
132 participantQos.name("Participant_subscriber");
133 participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
134
135 if (participant_ == nullptr)
136 {
137 return false;
138 }
139
140 // Register the Type
141 type_.register_type(participant_);
142
143 // Create the subscriptions Topic
144 topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
145
146 if (topic_ == nullptr)
147 {
148 return false;
149 }
150
151 // Create the Subscriber
152 subscriber_ = participant_->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr);
153
154 if (subscriber_ == nullptr)
155 {
156 return false;
157 }
158
159 // Create the DataReader
160 reader_ = subscriber_->create_datareader(topic_, DATAREADER_QOS_DEFAULT, &listener_);
161
162 if (reader_ == nullptr)
163 {
164 return false;
165 }
166
167 return true;
168 }
169
170 //!Run the Subscriber
171 void run(
172 uint32_t samples)
173 {
174 while(listener_.samples_ < samples)
175 {
176 std::this_thread::sleep_for(std::chrono::milliseconds(100));
177 }
178 }
179};
180
181int main(
182 int argc,
183 char** argv)
184{
185 std::cout << "Starting subscriber." << std::endl;
186 int samples = 10;
187
188 HelloWorldSubscriber* mysub = new HelloWorldSubscriber();
189 if(mysub->init())
190 {
191 mysub->run(static_cast<uint32_t>(samples));
192 }
193
194 delete mysub;
195 return 0;
196}
1.3.8.1. Examining the code¶
Since the source code of both the publisher and subscriber applications is mostly identical, this document will focus on the main differences between them, omitting the parts of the code that have already been explained.
Following the same structure as in the publisher explanation, the first step is the includes of the C++ header files. In these, the files that include the publisher class are replaced by the subscriber class and the data writer class by the data reader class.
Subscriber
. It is the object responsible for the creation and configuration of DataReaders.DataReader
. It is the object responsible for the actual reception of the data. It registers in the application the topic (TopicDescription) that identifies the data to be read and accesses the data received by the subscriber.DataReaderListener
. This is the listener assigned to the data reader.DataReaderQoS
. Structure that defines the QoS of the DataReader.SampleInfo
. It is the information that accompanies each sample that is ‘read’ or ‘taken.’
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
The next line defines the HelloWorldSubscriber
class that implements a subscriber.
class HelloWorldSubscriber
Starting with the private data members of the class, it is worth mentioning the implementation of the data reader
listener.
The private data members of the class will be the participant, the subscriber, the topic, the data reader, and the
data type.
As it was the case with the data writer, the listener implements the callbacks to be executed in case an event
occurs.
The first overridden callback of the SubListener is the on_subscription_matched()
, which is the
analog of the on_publication_matched()
callback of the DataWriter.
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;
}
}
The second overridden callback is on_data_available()
.
In this, the next received sample that the data reader can access is taken and processed to display its content.
It is here that the object of the SampleInfo
class is defined, which determines whether a sample has already
been read or taken.
Each time a sample is read, the counter of samples received is increased.
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;
}
}
}
The public constructor and destructor of the class is defined below.
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_);
}
Next comes the subscriber initialization public member function.
This is the same as the initialization public member function defined for the HelloWorldPublisher
.
The QoS configuration for all entities, except for the participant’s name, is the default QoS
(PARTICIPANT_QOS_DEFAULT
, SUBSCRIBER_QOS_DEFAULT
, TOPIC_QOS_DEFAULT
, DATAREADER_QOS_DEFAULT
).
The default value of the QoS of each DDS Entity can be checked in the
DDS standard.
//!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;
}
The public member function run()
ensures that the subscriber runs until all the samples have been received.
This member function implements an active wait of the subscriber, with a 100ms sleep interval to ease the CPU.
//!Run the Subscriber
void run(
uint32_t samples)
{
while(listener_.samples_ < samples)
{
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
Finally, the participant that implements a subscriber is initialized and run in 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¶
Include at the end of the CMakeList.txt file you created earlier the following code snippet. This adds all the source files needed to build the executable, and links the executable and the library together.
add_executable(DDSHelloWorldSubscriber src/HelloWorldSubscriber.cpp ${DDS_HELLOWORLD_SOURCES_CXX})
target_link_libraries(DDSHelloWorldSubscriber fastrtps fastcdr)
At this point the project is ready for building, compiling and running the subscriber application. From the build directory in the workspace, run the following commands.
cmake ..
cmake --build .
./DDSHelloWorldSubscriber
1.3.9. Putting all together¶
Finally, from the build directory, run the publisher and subscriber applications from two terminals.
./DDSHelloWorldPublisher
./DDSHelloWorldSubscriber
1.3.10. Summary¶
In this tutorial you have built a publisher and a subscriber DDS application. You have also learned how to build the CMake file for source code compilation, and how to include and use the Fast DDS and Fast CDR libraries in your project.