1.3. Writing a simple publisher and subscriber application

This section details how to create an simple Fast DDS application with a publisher and a subscriber 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. The way we will make these accessible from the workspace depends on the installation procedure we have followed in the Installation Manual.

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

If you have followed the Colcon installation there are several ways to import the libraries. If you want these to be accessible only from the current shell session, run one of the following two commands.

source <path/to/Fast-DDS/workspace>/install/setup.bash

If you want these to be accessible from any session, you can add the Fast DDS installation directory to your $PATH variable in the shell configuration files running the following command.

echo 'source <path/to/Fast-DDS/workspace>/install/setup.bash' >> ~/.bashrc

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:

  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 your favorite 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

Now you have the publisher’s source code. The publisher is going to send 10 publications under the topic HelloWorld.

  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. Is the object responsible for the creation of DataReaders.

  • 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 allow us to execute routines in case of an event. The overridden callback on_publication_matched allows you to define 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 you to track 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;
    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 you can build, compile and run the publisher application. From the build directory in the workspace, run the following commands.

cmake ..
make
./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

Now you have the subscriber’s source code. 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

As you have noticed, the source code to implement the subscriber is practically identical to the source code implemented by the publisher. Therefore, we will focus on the main differences between them, without explaining all the code again.

Following the same structure as in the publisher explanation, we start with 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_);
}

Then we have 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 you can build, compile and run the subscriber application. From the build directory in the workspace, run the following commands.

cmake ..
make clean && make
./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.