This blog post aims to give you an overview over what RTI Connext DDS is, how it works and in which context it is used here at the ICCLab. Furthermore it shows you how to set it up and implement a simple HelloWorld-like application. It is not intended to explain details or specific concepts. For this you are kindly referred to the very abundant and well written documentation which comes with the product or can be found online.

The source code of the example can be found here –> HelloWorld_Source

What is RTI Connext DDS?

RTI Connext DDS is a network middleware built for real-time distributed applications. It implements the Data-Centric Publish-Subscribe (DCPS) API within OMG’s Data Distribution Service (DDS). As can be deduced by this name, the employed communications-model of this middleware is publish-subscribe. In this model applications “publish” data they want to share and “subscribe” to data they need. These messages of data flow directly between publishers and subscribers and hence eliminate the need of having a centralized node (server/broker) which has to organize the data-flow. A real life example of this model would be television or newspapers. The “data-centric” part of DCPS describes the fundamental concept supported by the design of the API. In data-centric communications, the focus lies on the distribution of data (known data types in named streams) between communicating applications. In contrast, in object-centric communications the fundamental concept is the interface (know methods of known types) between applications. Another strength of RTI Connext DDS is its advanced and easy to use quality of service (QoS) support.

 How does RTI Connext DDS work?

The most important objects and entities of RTI Connext DDS are:

• Domain: 
  Is a concept to bind individual applications together for communication.
  It can be seen as the “channel” through which data flows.
• DomainParticipant: 
  Enables an application to exchange messages within a certain domain.
  Is used to create Publishers, DataWriters, Subscribers, DataReaders and Topics.
• Publisher and DataWriter:
  Are used to send messages/data. A DataWriter writes the actual message/data and publishes it into a domain. A Publisher is used to group individual DataWriters together and to share common properties (e.g. QoS settings) among them.
• Subscriber and DataReader: 
  Are used to receive messages/data. A DataReader reads the actual message it has received. A Subscriber is used to group individual DataReaders together and to share common properties (e.g. QoS settings) among them.
• Topic: 
  Connects a DataWriter to a DataReader. Consists of a name and a data-type

To establish a communication-channel, DataWriters and DataReaders must have the same Topic of the same data-type and be members of the same Domain. There is no need to define an address of a subscriber or publisher, RTI Connext DDS works with an auto-discovery mechanism to match publishers with subscribers.

The following picture (Source: RTI Connext Core Libraries and Utilities Getting Started Guide) shows you the relationship of these objects and entities.

 

In what Context is RTI Connext DDS used at the ICCLab?

The ICCLab is currently one of the members of the FI-WARE open call project “Middleware for efficient and QoS/Security-aware invocation of services and exchange of messages” named KIARA. RTI Connext DDS middleware serves as the basis on which KIARA will build upon. KIARA’s goal is to add the following capabilities to RTI Connext DDS:

• Multi-Domain Model
• Runtime Negotiation and Transport Configuration
• Embedded Compiler to simplify data access by applications
• Additional communication protocols and mechanisms
• Support for SOAP and RESTfull marshaling
• Security by Design

KIARA will use the plugin architecture of RTI Connext DDS to implement these extensions, but they should also be easily adoptable for alternative DDS implementations like OpenDDS.

Setting up RTI Connext DDS

Remark: As this is a middleware, it is possible to send data from one platform (language/compiler/operating system/processor) to many others. Therefore there are multiple operating systems this middleware could be installed on and multiple programming languages an application could be written in. In this example the used operating system is Windows 7 and the programming language is Java (JDK1.7) with the Eclipse IDE (JUNO).

To set up RTI Connext DDS follow these steps:

1) Install RTI Connext DDS

RTI Connext DDS can easily be installed by using the installer.

2) Install License File

Copy the license file (as received by RTI) into the C:…RTI and C:…RTIndds.x.x.x folders.

3) Set Environment Variables

RTI_LICENSE_FILE	C:…RTI
NDDSHOME	C:…RTIndds.x.x.x
PATH (append)	C:…RTIndds.x.x.x lib<architecture>jdk

4) Eclipse: Add Libraries

In order to be able to build applications, add C:…RTIndds.5.0.0classnddsjava.jar to your project libraries (Project–>Properties–>Java Build Path–>Libraries–>Add External JARs).

 

Getting started with RTI Connext DDS: HelloWorld

As mentioned before, the documentation of RTI Connext DDS is abundant – the user manual alone is 780 pages strong – and very well and clearly written. The “Getting Started” guide shows you basics concepts and familiarizes with the API.  There are also numerous code examples. And every example is available in C, C++, C# and Java. A big compliment has to be given here to RTI for this outstanding documentation of the product.

I will now walk you through the HelloSimple example which also comes with the product. This application has two classes, a publisher who allows a user to write an arbitrary line of text and a subscriber who will receive this text as soon as it becomes available.

Implementing the Publisher

Let’s start with the Publisher class which in this case is named HelloPublisher.

public class HelloPublisher {
public static final void main(String[] args)

Gist: 4025165

1) Creating the DomainParticipant

First a DomainParticipant has to be created. The DomainParticipant defines in which Domain an application belongs and will be used to create all other needed entities like the Publisher, DataWriter and Topic. The create_participant() method takes four arguments:

domainID	An integer value which defines the Domain the DomainParticipant is in.
QoS	Quality of service settings to be used. In this case the default QoS settings are being used.
listener	An optional listener (callback routine) to be invoked when specific events with respect to the DomainParticipant occur.
mask	Defines the type of events that trigger the callback routine (listener).

// Create the DDS Domain participant on domain ID 0
DomainParticipant participant = DomainParticipantFactory.get_instance().create_participant(
0,                                                // Domain ID = 0
DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, // QoS Settings = Default
null,                                             // listener
StatusKind.STATUS_MASK_NONE);                     // mask
if (participant == null) {
    System.err.println("Unable to create domain participant");
    return;
}

Gist: 4025343

2) Creating the Topic

The DomainParticipant is now used to create a Topic with the name “Hello, World” and the built-in String data type. Remember, a Topic always consists of a name and a data type. The create_topic() method takes five arguments:

topic_name	Name of the Topic (String)
type_name	The name of the user data type. In this case, the built-in String data type is used.
QoS	Quality of service settings to be used. In this case the default QoS settings are being used.
listener	An optional listener (callback routine) to be invoked when specific events with respect to the Topic occur. If this is null, the DomainParticipant‘s listener (if existing) will be used.
mask	Defines the type of events that trigger the callback routine (listener).

// Create the topic "Hello World" with the built-in String type
Topic helloWorldTopic = participant.create_topic(
    "Hello, World",                      //topic_name
    StringTypeSupport.get_type_name(),   //type_name
    DomainParticipant.TOPIC_QOS_DEFAULT, //QoS
    null,                                //listener
    StatusKind.STATUS_MASK_NONE);        //mask
if (topic == null) {
    System.err.println("Unable to create topic.");
    return;
}

Gist: 4025420

3) Creating the DataWriter (with default Publisher)

The DomainParticipant can now instantiate a DataWriter. This object will then later be used to write the messages to be sent to subscribers. The create_datawriter() method takes four arguments:

topic	The topic for which this DataWriter will write messages/data.
QoS	The name of the user data type. In this case, the built-in String data type is used.
listener	An optional listener (callback routine) to be invoked when specific events with respect to the DataWriter occur. If this is null, the Publisher‘s listener (if existing) will be used.
mask	Defines the type of events that trigger the callback routine (listener).

If you’re wondering why there was no Publisher instantiated. The creation of a Publisher is optional and if not explicitly created, a default publisher is used (this is an extension of RTI Connext DDS and not part of the DDS standard). Otherwise the Publisher could be used to create the DataWriter.

// Create the data writer using the default publisher
StringDataWriter dataWriter = (StringDataWriter)participant.create_datawriter(
    helloWorldTopic,                  // Topic
    Publisher.DATAWRITER_QOS_DEFAULT, // QoS
    null,                             // listener
    StatusKind.STATUS_MASK_NONE);     // mask
if (dataWriter == null) {
    System.err.println("Unable to create data writer\n");
    return;
}

Gist: 4030283

4) Writing messages with DataWriter

Finally we can use the DataWriter to write – and automatically publish through the default Publisher – messages that we in this case receive as user input. The write() method takes two arguments.

Instance	An instance of the data type to be sent. In this case a String.
InstanceHandle	An InstanceHandle. This is used with keyed data types, a subject not covered in this blog.

Before terminating the program, all entities of the DomainParticipant and the DomainParticipant itself are deleted.

System.out.println("Ready to write data.");
System.out.println("When the subscriber is ready, you can start writing.");
System.out.print("Press CTRL+C to terminate or enter an empty line to do a clean shutdown.\n\n"); 

BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try
{
while (true) {
System.out.print("Please type a message> ");
    String toWrite = reader.readLine();
        dataWriter.write(toWrite, InstanceHandle_t.HANDLE_NIL); 

        if (toWrite.equals("")) break;
    }
}
catch (IOException e)
{
    e.printStackTrace();
} catch (RETCODE_ERROR e) {
    // This exception can be thrown from DDS write operation
    e.printStackTrace();
} 

System.out.println("Exiting..."); 

// Deleting entities of DomainParticipant and DomainParticipant
participant.delete_contained_entities();
DomainParticipantFactory.get_instance().delete_participant(participant);
}}

Gist: 4030299

Implementing the Subscriber

The Subscriber is very similar to the Publisher.  There also needs to be a DomainParticipant and a Topic. The creation of those will not be explained again. Instead of a DataWriter, there will now be a DataReader. A listener will be implemented to automatically run when new messages are available. For the Subscriber to be able to receive messages sent by the HelloPublisher, the DomainParticipant must be in the same Domain and the Topic must have the same name and data type.

public class HelloSubscriber extends DataReaderAdapter { 

// For clean shutdown sequence
private static boolean shutdown_flag = false;

public static final void main(String[] args) {

// Create the DDS Domain participant on domain ID 0
DomainParticipant participant = DomainParticipantFactory.get_instance().create_participant(
    0,                                                // Domain ID = 0
    DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, // QoS
    null,                                             // listener
    StatusKind.STATUS_MASK_NONE);                     // mask
if (participant == null) {
System.err.println("Unable to create domain participant");
return;
} 

// Create the topic "Hello World" for the String type
Topic topic = participant.create_topic(
    "Hello, World",                      // Topic Name
    StringTypeSupport.get_type_name(),   // Topic Data Type
    DomainParticipant.TOPIC_QOS_DEFAULT, // QoS
    null,                                // listener
    StatusKind.STATUS_MASK_NONE);        // mask
if (topic == null) {
    System.err.println("Unable to create topic.");
    return;
}

Gist: 4030332

1 ) Creating the DataReader

After the DomainParticipant and Topic, the DataReader will be created. As can be seen, it has to be bound to a topic at instantiation time and in this case – for the first time in our example – uses a listener as well as a mask. The listener is, in this case, an instance of HelloSubscriber. That is also the reason why the HelloSubscriber class extends from the DataReaderAdapter class and overrides the on_data_available(DataReader reader) method which is the method that will be triggered once data becomes available. The mask indicates that this listener should be trigger as soon as new messages/data is available.

// Create the data reader using the default publisher
StringDataReader dataReader = (StringDataReader) participant.create_datareader(
    topic, //Topic
    Subscriber.DATAREADER_QOS_DEFAULT, // QoS
    new HelloSubscriber(), // Listener
    StatusKind.DATA_AVAILABLE_STATUS); // mask

if (dataReader == null) {
    System.err.println("Unable to create DDS Data Reader");
    return;
}

// Reading User-Input
System.out.println("Ready to read data.");
System.out.println("Press CTRL+C to terminate.");

for (;;) {
try {
    Thread.sleep(2000);
    if(shutdown_flag) break;
    }catch (InterruptedException e) {
    // Nothing to do...
    }
}

System.out.println("Shutting down...");

// Deleting entities and DomainParticipant
participant.delete_contained_entities();
DomainParticipantFactory.get_instance().delete_participant(participant);
}

Gist: 4030389

2) Reading the Data

Finally in the on_data_available() Method the logic to read data is implemented. Every received message (called sample)  is actually composed of two parts. First of the message itself (here a String) and second of a metafile of type SampleInfo which  contains additional information about that message.

The created stringReader then tries to read the next sample with the take_next_sample() method into which it passes the info file. If the sample contains actual data, the valid_data property of the info object will equal true and the sample will be printed onto the console.

public void on_data_available(DataReader reader) {

// Method Parameter of Type DataReader must be cast to Data Type ofDataReader for this Topic
StringDataReader stringReader = (StringDataReader) reader;
SampleInfo info = new SampleInfo();

for (;;) {
try {
    String sample = stringReader.take_next_sample(info);
    if (info.valid_data) {
        System.out.println(sample);

        if (sample.equals("")) {
        shutdown_flag = true;
        }
    }
}
catch (RETCODE_NO_DATA noData) {
    // No more data to read
    break;
}
catch (RETCODE_ERROR e) {
    // An error occurred
    e.printStackTrace();
}
}}}

Gist: 4030451

Conclusion

This should have given you a small insight into RTI Connext DDS. There is still a vast amount of additional functionalities and configurations that this product has to offer. But after exploring this middleware for some weeks it has clearly shown to be an excellent starting point for the KIARA project.