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
        ├── HelloWorldCdrAux.hpp
        ├── HelloWorldCdrAux.ipp
        ├── 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.20)

project(DDSHelloWorld)

# Find requirements
if(NOT fastcdr_FOUND)
    find_package(fastcdr 2 REQUIRED)
endif()

if(NOT fastrtps_FOUND)
    find_package(fastrtps 2.12 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()

message(STATUS "Configuring HelloWorld publisher/subscriber example...")
file(GLOB DDS_HELLOWORLD_SOURCES_CXX "src/*.cxx")

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:

  1. Generate C++ definitions for your custom topic.

  2. 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: Interface used by Fast DDS to support HelloWorld type.

  • HelloWorldPubSubTypes.h: Header file for HelloWorldPubSubTypes.cxx.

  • HelloWorldCdrAux.ipp: Serialization and Deserialization code for the HelloWorld type.

  • HelloWorldCdrAux.hpp: Header file for HelloWorldCdrAux.ipp.

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.

target_link_libraries(DDSHelloWorldPublisher fastrtps fastcdr)

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 <chrono>
 23#include <thread>
 24
 25#include <fastdds/dds/domain/DomainParticipant.hpp>
 26#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
 27#include <fastdds/dds/publisher/DataWriter.hpp>
 28#include <fastdds/dds/publisher/DataWriterListener.hpp>
 29#include <fastdds/dds/publisher/Publisher.hpp>
 30#include <fastdds/dds/topic/TypeSupport.hpp>
 31
 32using namespace eprosima::fastdds::dds;
 33
 34class HelloWorldPublisher
 35{
 36private:
 37
 38    HelloWorld hello_;
 39
 40    DomainParticipant* participant_;
 41
 42    Publisher* publisher_;
 43
 44    Topic* topic_;
 45
 46    DataWriter* writer_;
 47
 48    TypeSupport type_;
 49
 50    class PubListener : public DataWriterListener
 51    {
 52    public:
 53
 54        PubListener()
 55            : matched_(0)
 56        {
 57        }
 58
 59        ~PubListener() override
 60        {
 61        }
 62
 63        void on_publication_matched(
 64                DataWriter*,
 65                const PublicationMatchedStatus& info) override
 66        {
 67            if (info.current_count_change == 1)
 68            {
 69                matched_ = info.total_count;
 70                std::cout << "Publisher matched." << std::endl;
 71            }
 72            else if (info.current_count_change == -1)
 73            {
 74                matched_ = info.total_count;
 75                std::cout << "Publisher unmatched." << std::endl;
 76            }
 77            else
 78            {
 79                std::cout << info.current_count_change
 80                        << " is not a valid value for PublicationMatchedStatus current count change." << std::endl;
 81            }
 82        }
 83
 84        std::atomic_int matched_;
 85
 86    } listener_;
 87
 88public:
 89
 90    HelloWorldPublisher()
 91        : participant_(nullptr)
 92        , publisher_(nullptr)
 93        , topic_(nullptr)
 94        , writer_(nullptr)
 95        , type_(new HelloWorldPubSubType())
 96    {
 97    }
 98
 99    virtual ~HelloWorldPublisher()
100    {
101        if (writer_ != nullptr)
102        {
103            publisher_->delete_datawriter(writer_);
104        }
105        if (publisher_ != nullptr)
106        {
107            participant_->delete_publisher(publisher_);
108        }
109        if (topic_ != nullptr)
110        {
111            participant_->delete_topic(topic_);
112        }
113        DomainParticipantFactory::get_instance()->delete_participant(participant_);
114    }
115
116    //!Initialize the publisher
117    bool init()
118    {
119        hello_.index(0);
120        hello_.message("HelloWorld");
121
122        DomainParticipantQos participantQos;
123        participantQos.name("Participant_publisher");
124        participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
125
126        if (participant_ == nullptr)
127        {
128            return false;
129        }
130
131        // Register the Type
132        type_.register_type(participant_);
133
134        // Create the publications Topic
135        topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
136
137        if (topic_ == nullptr)
138        {
139            return false;
140        }
141
142        // Create the Publisher
143        publisher_ = participant_->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr);
144
145        if (publisher_ == nullptr)
146        {
147            return false;
148        }
149
150        // Create the DataWriter
151        writer_ = publisher_->create_datawriter(topic_, DATAWRITER_QOS_DEFAULT, &listener_);
152
153        if (writer_ == nullptr)
154        {
155            return false;
156        }
157        return true;
158    }
159
160    //!Send a publication
161    bool publish()
162    {
163        if (listener_.matched_ > 0)
164        {
165            hello_.index(hello_.index() + 1);
166            writer_->write(&hello_);
167            return true;
168        }
169        return false;
170    }
171
172    //!Run the Publisher
173    void run(
174            uint32_t samples)
175    {
176        uint32_t samples_sent = 0;
177        while (samples_sent < samples)
178        {
179            if (publish())
180            {
181                samples_sent++;
182                std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
183                            << " SENT" << std::endl;
184            }
185            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
186        }
187    }
188};
189
190int main(
191        int argc,
192        char** argv)
193{
194    std::cout << "Starting publisher." << std::endl;
195    uint32_t samples = 10;
196
197    HelloWorldPublisher* mypub = new HelloWorldPublisher();
198    if(mypub->init())
199    {
200        mypub->run(samples);
201    }
202
203    delete mypub;
204    return 0;
205}

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 <chrono>
#include <thread>

#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/topic/TypeSupport.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:

  1. Initializes the content of the HelloWorld type hello_ structure members.

  2. Assigns a name to the participant through the QoS of the DomainParticipant.

  3. Uses the DomainParticipantFactory to create the participant.

  4. Registers the data type defined in the IDL.

  5. Creates the topic for the publications.

  6. Creates the publisher.

  7. 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;
    uint32_t samples = 10;

    HelloWorldPublisher* mypub = new HelloWorldPublisher();
    if(mypub->init())
    {
        mypub->run(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 <chrono>
 23#include <thread>
 24
 25#include <fastdds/dds/domain/DomainParticipant.hpp>
 26#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
 27#include <fastdds/dds/subscriber/DataReader.hpp>
 28#include <fastdds/dds/subscriber/DataReaderListener.hpp>
 29#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
 30#include <fastdds/dds/subscriber/SampleInfo.hpp>
 31#include <fastdds/dds/subscriber/Subscriber.hpp>
 32#include <fastdds/dds/topic/TypeSupport.hpp>
 33
 34using namespace eprosima::fastdds::dds;
 35
 36class HelloWorldSubscriber
 37{
 38private:
 39
 40    DomainParticipant* participant_;
 41
 42    Subscriber* subscriber_;
 43
 44    DataReader* reader_;
 45
 46    Topic* topic_;
 47
 48    TypeSupport type_;
 49
 50    class SubListener : public DataReaderListener
 51    {
 52    public:
 53
 54        SubListener()
 55            : samples_(0)
 56        {
 57        }
 58
 59        ~SubListener() override
 60        {
 61        }
 62
 63        void on_subscription_matched(
 64                DataReader*,
 65                const SubscriptionMatchedStatus& info) override
 66        {
 67            if (info.current_count_change == 1)
 68            {
 69                std::cout << "Subscriber matched." << std::endl;
 70            }
 71            else if (info.current_count_change == -1)
 72            {
 73                std::cout << "Subscriber unmatched." << std::endl;
 74            }
 75            else
 76            {
 77                std::cout << info.current_count_change
 78                        << " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
 79            }
 80        }
 81
 82        void on_data_available(
 83                DataReader* reader) override
 84        {
 85            SampleInfo info;
 86            if (reader->take_next_sample(&hello_, &info) == ReturnCode_t::RETCODE_OK)
 87            {
 88                if (info.valid_data)
 89                {
 90                    samples_++;
 91                    std::cout << "Message: " << hello_.message() << " with index: " << hello_.index()
 92                                << " RECEIVED." << std::endl;
 93                }
 94            }
 95        }
 96
 97        HelloWorld hello_;
 98
 99        std::atomic_int samples_;
100
101    } listener_;
102
103public:
104
105    HelloWorldSubscriber()
106        : participant_(nullptr)
107        , subscriber_(nullptr)
108        , topic_(nullptr)
109        , reader_(nullptr)
110        , type_(new HelloWorldPubSubType())
111    {
112    }
113
114    virtual ~HelloWorldSubscriber()
115    {
116        if (reader_ != nullptr)
117        {
118            subscriber_->delete_datareader(reader_);
119        }
120        if (topic_ != nullptr)
121        {
122            participant_->delete_topic(topic_);
123        }
124        if (subscriber_ != nullptr)
125        {
126            participant_->delete_subscriber(subscriber_);
127        }
128        DomainParticipantFactory::get_instance()->delete_participant(participant_);
129    }
130
131    //!Initialize the subscriber
132    bool init()
133    {
134        DomainParticipantQos participantQos;
135        participantQos.name("Participant_subscriber");
136        participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
137
138        if (participant_ == nullptr)
139        {
140            return false;
141        }
142
143        // Register the Type
144        type_.register_type(participant_);
145
146        // Create the subscriptions Topic
147        topic_ = participant_->create_topic("HelloWorldTopic", "HelloWorld", TOPIC_QOS_DEFAULT);
148
149        if (topic_ == nullptr)
150        {
151            return false;
152        }
153
154        // Create the Subscriber
155        subscriber_ = participant_->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr);
156
157        if (subscriber_ == nullptr)
158        {
159            return false;
160        }
161
162        // Create the DataReader
163        reader_ = subscriber_->create_datareader(topic_, DATAREADER_QOS_DEFAULT, &listener_);
164
165        if (reader_ == nullptr)
166        {
167            return false;
168        }
169
170        return true;
171    }
172
173    //!Run the Subscriber
174    void run(
175        uint32_t samples)
176    {
177        while(listener_.samples_ < samples)
178        {
179            std::this_thread::sleep_for(std::chrono::milliseconds(100));
180        }
181    }
182};
183
184int main(
185        int argc,
186        char** argv)
187{
188    std::cout << "Starting subscriber." << std::endl;
189    uint32_t samples = 10;
190
191    HelloWorldSubscriber* mysub = new HelloWorldSubscriber();
192    if(mysub->init())
193    {
194        mysub->run(samples);
195    }
196
197    delete mysub;
198    return 0;
199}

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/DomainParticipant.hpp>
#include <fastdds/dds/domain/DomainParticipantFactory.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>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/topic/TypeSupport.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;
    uint32_t samples = 10;

    HelloWorldSubscriber* mysub = new HelloWorldSubscriber();
    if(mysub->init())
    {
        mysub->run(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.

1.3.11. Next steps

In the eProsima Fast DDS Github repository you will find more complex examples that implement DDS communication for a multitude of use cases and scenarios. You can find them here.