An Introduction to Connext DDS
Connext DDS is network middleware for real-time distributed applications. It provides the communications service that programmers need to distribute time-critical data between embedded and/or enterprise devices or nodes. Connext DDS uses a publish-subscribe communications model to make data-distribution efficient and robust. Connext DDS also supports the Request-Reply communication pattern1Request-reply communication is available when using the Connext DDS Professional, Evaluation, and Basic package types. It is not available in the Core package type..
Connext DDS implements the Data-Centric Publish-Subscribe (DCPS) API of the OMG’s specification, Data Distribution Service (DDS) for Real-Time Systems. DDS is the first standard developed for the needs of real-time systems, providing an efficient way to transfer data in a distributed system. With Connext DDS, you begin your development with a fault-tolerant and flexible communications infrastructure that will work over a wide variety of computer hardware, operating systems, languages, and networking transport protocols. Connext DDS is highly configurable, so programmers can adapt it to meet an application’s specific communication requirements.
An Overview of Connext DDS Objects
The primary objects in Connext DDS are:
- DomainParticipants
- Publishers and DataWriters
- Subscribers and DataReaders
- Topics
- Keys and DDS Samples
- Requesters and Repliers
Figure 1 Connext DDS Components
DomainParticipants
A DDS domain is a concept used to bind individual applications together for communication. To communicate with each other, DataWriters and DataReaders must have the same Topic of the same DDS data type and be members of the same DDS domain. Each DDS domain has a unique integer domain ID.
Applications in one DDS domain cannot subscribe to data published in a different DDS domain. Multiple DDS domains allow you to have multiple virtual distributed systems on the same physical network. This can be very useful if you want to run multiple independent tests of the same applications. You can run them at the same time on the same network as long as each test runs in a different DDS domain. Another typical configuration is to isolate your development team from your test and production teams: you can assign each team or even each developer a unique DDS domain.
DomainParticipant objects enable an application to exchange messages within DDS domains. An application must have a DomainParticipant for every DDS domain in which the application will communicate. (Unless your application is a bridging application, it will typically participate in only one DDS domain and have only one DomainParticipant.)
A DomainParticipant is analogous to a JMS Connection.
DomainParticipants are used to create Topics, Publishers, DataWriters, Subscribers, and DataReaders in the corresponding DDS domain.
Figure 2 Segregating Applications with Domains
The following code shows how to instantiate a DomainParticipant. You can find more information about all of the APIs in the API Reference HTML documentation.
To create a DomainParticipant:
In Traditional C++:
participant =
DDSDomainParticipantFactory::get_instance()->create_participant(
0, /* Domain ID */
DDS_PARTICIPANT_QOS_DEFAULT, /* QoS */
NULL, /* Listener */
DDS_STATUS_MASK_NONE);
dds::domain::DomainParticipant participant(0);
In Java:
participant =
DomainParticipantFactory.get_instance().create_participant(
0, // Domain ID
DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT,
null, // Listener
StatusKind.STATUS_MASK_NONE);
participant :=
DDS.DomainParticipantFactory.Get_Instance.Create_Participant( 0, -- Domain ID DDS.DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, -- QoS null, -- Listener DDS.STATUS_MASK_NONE);
As you can see, there are four pieces of information you supply when creating a new DomainParticipant:
- The ID of the DDS domain to which it belongs.
- Its qualities of service (QoS). The discussion on What is QoS? gives a brief introduction to the concept of QoS; you will learn more in Capabilities and Performance.
- Its listener and listener mask, which indicate the events that will generate callbacks to the DomainParticipant. You will see a brief example of a listener callback when we discuss DataReaders below. For a more comprehensive discussion of the Connext DDS status and notification system, see Chapter 4 in the User’s Manual.
(In the Modern C++ API the QoS and listener and mask are optional parameters.)
What is QoS?
Fine control over Quality of Service (QoS) is perhaps the most important feature of Connext DDS. Each data producer-consumer pair can establish independent quality of service agreements—even in many-to-many topologies. This allows applications to support extremely complex and flexible data-flow requirements.
QoS policies control virtually every aspect of Connext DDS and the underlying communications mechanisms. Many QoS policies are implemented as "contracts" between data producers (DataWriters) and consumers (DataReaders); producers offer and consumers request levels of service. The middleware is responsible for determining if the offer can satisfy the request, thereby establishing the communication or indicating an incompatibility error. Ensuring that participants meet the level-of-service contracts guarantees predictable operation. For example:
Periodic producers can indicate the speed at which they can publish by offering guaranteed update deadlines. By setting a deadline, a producer promises to send updates at a minimum rate. Consumers may then request data at that or any slower rate. If a consumer requests a higher data rate than the producer offers, the middleware will flag that pair as incompatible and notify both the publishing and subscribing applications.
Producers may offer different levels of reliability, characterized in part by the number of past DDS data samples they store for retransmission. Consumers may then request differing levels of reliable delivery, ranging from fast-but-unreliable "best effort" to highly reliable in-order delivery. This provides per-data-stream reliability control. A single producer can support consumers with different levels of reliability simultaneously.
Other QoS policies control when the middleware detects nodes that have failed, set delivery order, attach user data, prioritize messages, set resource utilization limits, partition the system into namespaces, control durability (for fault tolerance) and much more. The Connext DDS QoS policies offer unprecedented flexible communications control. The User’s Manual contains details about all available QoS policies.
Publishers and DataWriters
An application uses a DataWriter to publish data into a DDS domain. Once a DataWriter is created and configured with the correct QoS settings, an application only needs to use the DataWriter’s “write” operation to publish data.
A DataWriter is analogous to a JMS TopicPublisher. A Publisher is analogous to the producing aspect of a JMS TopicSession.
A Publisher is used to group individual DataWriters. You can specify default QoS behavior for a Publisher and have it apply to all the DataWriters in that Publisher’s group.
Figure 3 Entities Associated with Publications
In Traditional C++:
data_writer = participant->create_datawriter(
topic,
DDS_DATAWRITER_QOS_DEFAULT, /* QoS */
NULL, /* Listener */
DDS_STATUS_MASK_NONE);
In Modern C++:
dds::pub::DataWriter<dds::core::StringTopicType> writer(
rti::pub::implicit_publisher(participant), topic);
In Ada:
data_writer := participant.Create_DataWriter(
topic,
DDS.Publisher.DATAWRITER_QOS_DEFAULT, -- QoS
null, -- Listener
DDS.STATUS_MASK_NONE);
As you can see, each DataWriter is tied to a single topic. All data published by that DataWriter will be directed to that Topic.
As you will learn in Capabilities and Performance, each Topic—and therefore all DataWriters for that Topic—is associated with a particular concrete DDS data type. The write operation, which publishes data, is type safe, which means that before you can write data, you must perform a type cast, like this (in Modern C++ the DataWriter is already typed and there is no cast to perform):
In Traditional C++:
string_writer = DDSStringDataWriter::narrow(data_writer);
In Java:
StringDataWriter dataWriter =
(StringDataWriter) participant.create_datawriter(
topic,
Publisher.DATAWRITER_QOS_DEFAULT,
null, // listener
StatusKind.STATUS_MASK_NONE);
In Ada:
string_writer := DDS.Builtin_String_DataWriter.Narrow(data_writer);
Note that in this particular code example, you will not find any reference to the Publisher class. In fact, creating the Publisher object explicitly is optional, because many applications do not have the need to customize any behavior at that level. If you choose not to create a Publisher, the middleware will implicitly choose an internal Publisher object. If you do want to create a Publisher explicitly, create it with a call to participant.create_publisher() (you can find more about this method in the API Reference HTML documentation) and then simply replace the call to participant.create_datawriter() with a call to publisher.create_datawriter().
Subscribers and DataReaders
A DataReader is the point through which a subscribing application accesses the data that it has received over the network.
A DataReader is analogous to a JMS TopicSubscriber. A Subscriber is analogous to the consuming aspect of a JMS TopicSession.
Just as Publishers are used to group DataWriters, Subscribers are used to group DataReaders. Again, this allows you to configure a default set of QoS parameters and event handling routines that will apply to all DataReaders in the Subscriber's group.
Figure 4 Entities Associated with Subscriptions
Each DataReader is tied to a single topic. A DataReader will only receive data that was published on its Topic.
In Traditional C++:
data_reader = participant->create_datareader(
topic,
DDS_DATAREADER_QOS_DEFAULT, /* QoS */
&listener, /* Listener */
DDS_DATA_AVAILABLE_STATUS);
In Modern C++:
// Without a listener:
dds::sub::DataReader<dds::core::StringTopicType> data_reader(
rti::sub::implicit_subscriber(participant), topic);
// With a listener
dds::sub::DataReader<dds::core::StringTopicType> data_reader(
rti::sub::implicit_subscriber(participant),
topic,
participant->default_datareader_qos(),
&listener,
dds::core::status::StatusMask::data_available());
In Java:
StringDataReader dataReader =
(StringDataReader) participant.create_datareader(
topic,
Subscriber.DATAREADER_QOS_DEFAULT, // QoS
new HelloSubscriber(), // Listener
StatusKind.DATA_AVAILABLE_STATUS);
In Ada:
dataReader := DDS. Builtin_String_DataReader.Narrow(
participant.Create_DataReader(
topic.As_TopicDescription,
DDS.Subscriber.DATAREADER_QOS_DEFAULT, -- QoS
readerListener'Unchecked_Access, -- Listener
DDS.DATA_AVAILABLE_STATUS));
Connext DDS provides multiple ways for you to access your data: you can receive it asynchronously in a listener, you can block your own thread waiting for it to arrive using a helper object called a WaitSet, or you can poll in a non-blocking fashion. This example uses the former mechanism, and you will see that it passes a non-NULL listener to the create_datareader() method. The listener mask (DATA_AVAILABLE_STATUS) indicates that the application is only interested in receiving notifications of newly arriving data.
Let’s look at the callback implementation in the following example code.
In Traditional C++:
retcode = string_reader->take_next_sample(ptr_sample,info);
if (retcode == DDS_RETCODE_NO_DATA) {
/* No more samples */
break;
} else if (retcode != DDS_RETCODE_OK) {
cerr << "Unable to take data from data reader, error "
<< retcode << endl;
return;
}
In Modern C++:
dds::sub::LoanedSamples<dds::core::StringTopicType> samples = reader.take();
if (samples.length() == 0) {
std::cout << "No samples\n"
}
// In case of error an exception is thrown
In Ada:
begin
data_reader.Read_Next_Sample (ptr_sample, sample_info'Access);
if sample_info.Valid_Data then
Ada.Text_IO.Put_Line (DDS.To_Standard_String (ptr_sample));
end if;
exception
when DDS.NO_DATA =>
-- No more samples
exit;
when others =>
Self.receiving := FALSE;
exit;
end;
The take_next_sample() method retrieves a single DDS data sample (i.e., a message) from the DataReader, one at a time without blocking. If it was able to retrieve a DDS sample, it will return DDS_RETCODE_OK. If there was no data to take, it will return DDS_RETCODE_NO_DATA. Finally, if it tried to take data but failed to do so because it encountered a problem, it will return DDS_RETCODE_ERROR or another DDS_ReturnCode_t value (see the API Reference HTML documentation for a full list of error codes).
Connext DDS can publish not only actual data to a Topic, but also meta-data indicating, for example, that an object about which the DataReader has been receiving data no longer exists. In the latter case, the info argument to take_next_sample() will have its valid_data flag set to false.
This simple example is interested only in DDS data samples, not meta-data, so it only processes “valid” data.
In Modern C++ we are retrieving all the samples at once.
In Traditional C++:
if (info.valid_data) {
// Valid (this isn't just a lifecycle sample): print it
cout << ptr_sample << endl;
}
In Modern C++:
for (auto sample : samples) {
if (sample.info().valid()) {
std::cout << sample.data() << std::endl;
}
}
In Java:
try {
String sample = stringReader.take_next_sample(info);
if (info.valid_data) {
System.out.println(sample);
}
} catch (RETCODE_NO_DATA noData) {
// No more data to read
break;
} catch (RETCODE_ERROR e) {
// An error occurred
e.printStackTrace();
}
In Ada:
if sample_info.Valid_Data then -- Valid, print it Ada.Text_IO.Put_Line (DDS.To_Standard_String (ptr_sample)); end if;
Note that in this particular code example, you will not find any reference to the Subscriber class. In fact, as with Publishers, creating the Subscriber object explicitly is optional, because many applications do not have the need to customize any behavior at that level. If you, like this example, choose not to create a Subscriber, the middleware will implicitly choose an internal Subscriber object. If you do want to create a Subscriber explicitly, create it with a call to participant.create_subscriber (you can find more about this method in the API Reference HTML documentation) and then simply replace the call to participant.create_datareader with a call to subscriber.create_datareader.
Topics
Topics provide the basic connection points between DataWriters and DataReaders. To communicate, the Topic of a DataWriter on one node must match the Topic of a DataReader on any other node.
A Topic is comprised of a name and a DDS type. The name is a string that uniquely identifies the Topic within a DDS domain. The DDS type is the structural definition of the data contained within the Topic; this capability is described in Capabilities and Performance.
You can create a Topic with the following code:
In Traditional C++:
topic = participant->create_topic(
"Hello, World", /* Topic name*/
DDSStringTypeSupport::get_type_name(), /* Type name */
DDS_TOPIC_QOS_DEFAULT, /* Topic QoS */
NULL, /* Listener */
DDS_STATUS_MASK_NONE);
In Modern C++:
dds::topic::Topic<dds::core::StringTopicType> topic(
participant, "Hello, World");
In Java:
Topic topic = participant.create_topic(
"Hello, World", // Topic name
StringTypeSupport.get_type_name(), // Type name
DomainParticipant.TOPIC_QOS_DEFAULT, // QoS
null, // Listener
StatusKind.STATUS_MASK_NONE);
In Ada:
topic := participant.Create_Topic(
DDS.To_DDS_Builtin_String ("Hello, World"), -- Topic name
DDS.Builtin_String_TypeSupport.Get_Type_Name, -- Type name
DDS.DomainParticipant.TOPIC_QOS_DEFAULT, -- Topic QoS
null, -- Listener
DDS.STATUS_MASK_NONE);
Besides the new Topic’s name and DDS type, an application specifies three things:
- A suggested set of QoS for DataReaders and DataWriters for this Topic.
- A listener and listener mask that indicate which events the application wishes to be notified of, if any.
In this case, the Topic’s DDS type is a simple string, a type that is built into the middleware.
Keys and DDS Samples
The data values associated with a Topic can change over time. The different values of the Topic passed between applications are called DDS samples. A DDS sample is analogous to a message in other publish-subscribe middleware.
An application may use a single Topic to carry data about many objects. For example, a stock-trading application may have a single topic, "Stock Price," that it uses to communicate information about Apple, Google, Microsoft, and many other companies. Similarly, a radar track management application may have a single topic, "Aircraft Position," that carries data about many different airplanes and other vehicles. These objects within a Topic are called instances. For a specific DDS data type, you can select one or more fields within the DDS data type to form a key. A key is used to uniquely identify one instance of a Topic from another instance of the same Topic, very much like how the primary key in a database table identifies one record or another. DDS samples of different instances have different values for the key. DDS samples of the same instance of a Topic have the same key. Note that not all Topics have keys. For Topics without keys, there is only a single instance of that Topic.
Requesters and Repliers
This section describes the Request-Reply communication pattern, which is only available with the RTI Connext DDS Professional, Evaluation, and Basic package types. It is not available with the Core package type.
Requesters and Repliers provide a way to use the Request-Reply communication pattern on top of the previously described Connext DDS entities.
An application uses a Requester to send requests to a Replier; another application using a Replier receives a request and can send one or more replies for that request. The Requester that sent the request (and only that one) will receive the reply (or replies).
A Requester uses an existing DomainParticipant to communicate through a domain. It owns a DataWriter for writing requests and a DataReader for receiving replies.
Similarly, a Replier uses an existing DomainParticipant to communicate through a domain and owns a DataReader for receiving requests and a DataWriter for writing replies.
Figure 5 Request-Reply Overview
The Reply Topic filters samples so replies are received by exactly one Requester—the one that wrote the related request sample.
You can specify the QoS for the DataWriters and DataReaders that Requesters and Repliers create.
The following code shows how to create a Requester:
In C++:
Requester<Foo, Bar> * requester =
new Requester<Foo, Bar>(participant, "MyService");
In Java:
Requester<Foo, Bar> requester = new Requester<Foo, Bar>(
participant, "MyService",
FooTypeSupport.get_instance(),
BarTypeSupport.get_instance());
As you can see, we are passing an existing DomainParticipant to the constructor.
Foo is the request type and Bar is the reply type. In Compact, Type-Safe Data: Programming with DDS Data Types, you will learn what types you can use and how to create them.
The constructor also receives a string "MyService." This is the service name, and is used to create the Request Topic and the Reply Topic. In this example, the Requester will create a Request Topic called "MyServiceRequest" and a Reply Topic called "MyServiceReply."
Creating a Replier is very similar. The following code shows how to create a Replier:
In C++:
Replier<Foo, Bar> * replier =
new Replier<Foo, Bar>(participant, "TestService");
In Java:
Replier<Foo, Bar> replier = new Replier<Foo, Bar>(
participant, "TestService",
FooTypeSupport.get_instance(),
BarTypeSupport.get_instance());
This Replier will communicate with the Requester we created before, because they use the same service name (hence the topics are the same) and they use compatible QoS (the default).
More example code is available for C++, C, Java and C# as part of the API Reference HTML documentation, under Modules, Programming How-To’s, Request-Reply Examples.
© 2015 RTI