4. RTPS Layer

The lower level RTPS Layer of eprosima Fast DDS serves an implementation of the protocol defined in the RTPS standard. This layer provides more control over the internals of the communication protocol than the DDS Layer, so advanced users have finer control over the library’s functionalities.

4.1. Relation to the DDS Layer

Elements of this layer map one-to-one with elements from the DDS Layer, with a few additions. This correspondence is shown in the following table:

DDS Layer

RTPS Layer

Domain

RTPSDomain

DomainParticipant

RTPSParticipant

DataWriter

RTPSWriter

DataReader

RTPSReader

4.2. How to use the RTPS Layer

We will now go over the use of the RTPS Layer like we did with the DDS Layer one, explaining the new features it presents.

We recommend you to look at the two examples of how to use this layer the distribution comes with while reading this section. They are located in examples/RTPSTest_as_socket and in examples/RTPSTest_registered

4.2.1. Managing the Participant

Creating a RTPSParticipant is done with RTPSDomain::createParticipant(). RTPSParticipantAttributes structure is used to configure the RTPSParticipant upon creation.

RTPSParticipantAttributes participant_attr;
participant_attr.setName("participant");
RTPSParticipant* participant = RTPSDomain::createParticipant(0, participant_attr);

4.2.2. Managing the Writers and Readers

As the RTPS standard specifies, Writers and Readers are always associated with a History element. In the DDS Layer, its creation and management is hidden, but in the RTPS Layer, you have full control over its creation and configuration.

Writers are created with RTPSDomain::createRTPSWriter() and configured with a WriterAttributes structure. They also need a WriterHistory which is configured with a HistoryAttributes structure.

HistoryAttributes history_attr;
WriterHistory* history = new WriterHistory(history_attr);
WriterAttributes writer_attr;
RTPSWriter* writer = RTPSDomain::createRTPSWriter(participant, writer_attr, history);

The creation of a Reader is similar to that of the Writers. Note that in this case, you can provide a specialization of ReaderListener class that implements your callbacks:

class MyReaderListener : public ReaderListener
{
    // Callbacks override
};
MyReaderListener listener;
HistoryAttributes history_attr;
ReaderHistory* history = new ReaderHistory(history_attr);
ReaderAttributes reader_attr;
RTPSReader* reader = RTPSDomain::createRTPSReader(participant, reader_attr, history, &listener);

4.2.3. Using the History to Send and Receive Data

In the RTPS Protocol, Readers and Writers save the data about a topic in their associated History. Each piece of data is represented by a Change, which eprosima Fast DDS implements as CacheChange_t. Changes are always managed by the History. As a user, the procedure for interacting with the History is always the same:

  1. Request a CacheChange_t from the History

  2. Use it

  3. Release it

You can interact with the History of the Writer to send data. A callback that returns the maximum number of payload bytes is required:

//Request a change from the history
CacheChange_t* change = writer->new_change([]() -> uint32_t {
    return 255;
}, ALIVE);
//Write serialized data into the change
change->serializedPayload.length = sprintf((char*) change->serializedPayload.data, "My example string %d", 2) + 1;
//Insert change back into the history. The Writer takes care of the rest.
history->add_change(change);

If your topic data type has several fields, you will have to provide functions to serialize and deserialize your data in and out of the CacheChange_t. Fast DDS-Gen does this for you.

You can receive data from within a ReaderListener callback method as we did in the DDS Layer:

class MyReaderListener : public ReaderListener
{
public:

    MyReaderListener()
    {
    }

    ~MyReaderListener()
    {
    }

    void onNewCacheChangeAdded(
            RTPSReader* reader,
            const CacheChange_t* const change)
    {
        // The incoming message is enclosed within the `change` in the function parameters
        printf("%s\n", change->serializedPayload.data);
        // Once done, remove the change
        reader->getHistory()->remove_change((CacheChange_t*)change);
    }

};

4.3. Configuring Readers and Writers

One of the benefits of using the RTPS Layer is that it provides new configuration possibilities while maintaining the options from the DDS layer. For example, you can set a Writer or a Reader as a Reliable or Best-Effort endpoint as previously:

writer_attr.endpoint.reliabilityKind = BEST_EFFORT;

4.3.1. Setting the data durability kind

The Durability parameter defines the behavior of the Writer regarding samples already sent when a new Reader matches. eProsima Fast DDS offers three Durability options:

  • VOLATILE (default): Messages are discarded as they are sent. If a new Reader matches after message n, it will start received from message n+1.

  • TRANSIENT_LOCAL: The Writer saves a record of the last k messages it has sent. If a new reader matches after message n, it will start receiving from message n-k

  • TRANSIENT: As TRANSIENT_LOCAL, but the record of messages will be saved to persistent storage, so it will be available if the writer is destroyed and recreated, or in case of an application crash.

To choose your preferred option:

writer_attr.endpoint.durabilityKind = TRANSIENT_LOCAL;

Because in the RTPS Layer you have control over the History, in TRANSIENT_LOCAL and TRANSIENT modes the Writer sends all changes you have not explicitly released from the History.

4.4. Configuring the History

The History has its own configuration structure, the HistoryAttributes.

4.4.1. Changing the maximum size of the payload

You can choose the maximum size of the Payload that can go into a CacheChange_t. Be sure to choose a size that allows it to hold the biggest possible piece of data:

history_attr.payloadMaxSize  = 250;//Defaults to 500 bytes

4.4.2. Changing the size of the History

You can specify a maximum amount of changes for the History to hold and an initial amount of allocated changes:

history_attr.initialReservedCaches = 250; //Defaults to 500
history_attr.maximumReservedCaches = 500; //Defaults to 0 = Unlimited Changes

When the initial amount of reserved changes is lower than the maximum, the History will allocate more changes as they are needed until it reaches the maximum size.