Recommended way to process received data?

[ploe24]

Hi there,

a more "DDS-app-design"-related question:

We have included a Discovery Server, a Publisher and a Subscriber in our library.
Each class of them is built similar to the classes used in the DynamicHelloWorld and DiscoveryServer example(s).

The Subscriber, for instance, as separate class, with its own DataReader/DataReader-array and a private SubListener.
So far so good.
However, since we want to provide the received IDL type data (let's call it “IDL_DATA”) via the public API of our library, we wonder what the recommended way is to do so?!

Something alone these lines...
(but open for other suggestions on how to deal with this use case)

IDL_DATA data; // IDL data type
MyDDS myDDS; // Library interface
myDDS.readData(&data); // starts internal subscriber in library and waits for "on_data_available" callback?!

In contrast to the idea above, the example utilize a Subscriber::run() function to run a subscriber in a thread. There is no "callback" to get the data to the caller... it's only printed inside the private Listener callback implementation.

How to get the data from the Listener::on_data_available callback and how to "sync" the callback with our library API? Are there any recommendation or a "standard" way which works well?
(As the examples only show how to receive the data, which is completely fine and working in our case, but pretty "basic" and unusable in production code, since only printing the data inside the listener callback is obviously not what you want to do with the data) 😄

And so far I've not found any public examples (open source code) of "real" applications where FastDDS is in use... so one can get an idea how to actually continue with the data you have received in your Listener::on_data_available async callback.
If you know any real world example which is worth to have a look at, please let me know :)

Any help, suggestions and hints appreaciated :)

Thanks in advance,

[JesusPoderoso]

Hi @ploe24, thanks for using Fast DDS.
If I understood well, you want to expose the Listener::on_data_available callback to your library API but the Listener is a private member.
What do you think about the use of std::function, as we have done in the examples main.cpp with the signal handler? You can define a std::function<void(DataReader*)> data_received callback in the Listener, which is called when on_data_available is triggered, and assigned in construction or public setter:

private:
std::function<void(DataReader*)> data_received;

public:
YourListener::YourListener(std::function<bool(DataReader*)> data_received_callback)
    : data_received(data_received_callback) {}
// or
YourListener::YourListener(){}
void YourListener::set_custom_callback(std::function<bool(DataReader*)> data_received_callback) {
    data_received = data_received_callback;
}

// Make the on_data_available callback run the given std::function
void YourListener::on_data_available(DataReader* reader) override {
    data_received(reader);
}

In that way, assuming you propagate the std::function to the listener creation method, you can define the callback behavior in your library API or one step above:

IDL_DATA data; // IDL data type
// define the callback
std::function<void(DataReader*)> data_received_callback = [&](DataReader* reader)
                {
                    IDL_DATA data_;
                    SampleInfo info_;
                    // Add custom condition in the loop for listening to data, i.e. while true
                    while (true && (RETCODE_OK == reader->take_next_sample(&data_, &info_))) {
                        if ((info_.instance_state == ALIVE_INSTANCE_STATE) && info_.valid_data) {
                            // Use safely your IDL_DATA data_ here
                            // ... data_.method()
                        }
                    }
                };
// create the library interface with that callback, so it is (internally) propagated to the listener constructor
MyDDS myDDS(&data_received_callback); // Library interface propagates the callback to the listener constructor
myDDS.readData(&data); // starts internal subscriber in library and runs your custom callback

[ploe24]

@JesusPoderoso

In fact, this can be avoided too if you wrap the use of the data type in your custom callback and leave the take method in the on_data_available function.

You mean something like:

std::function< void (IDL_DATA &) > data_received;
// vs
std::function< void (IDL_DATA *) > data_received;

Does a pointer to the IDL data type make any sense here or better pass by reference as above?
The reader will discard the data when read in the callback? So it will be invalid to access by the program via API?

[JesusPoderoso]

@ploe24

Does a pointer to the IDL data type make any sense here or better pass by reference as above?

It depends on what you are going to do with the variable. In general terms, references are simpler and safer than pointers. For instance, if you are going to access the data (as it is done in the examples), then the use of references would be recommended. On the other hand, if you have dynamic arrays of data or want to point to another data sample, you should use pointers.

I think that the subscription behavior should be clarified. In case a reactive methodology is required, then we should go ahead with the previous snippets using listener callbacks:
Listener.hpp

private:
std::function<void(const IDL_DATA&)> data_received;

public:
YourListener::YourListener(
    std::function<void(const IDL_DATA&)> data_received_callback)
    : data_received(data_received_callback) {}
// or
YourListener::YourListener() {}
void YourListener::set_custom_callback(std::function<void(const IDL_DATA&)> data_received_callback) {
    data_received = data_received_callback;
}

// Make the on_data_available callback run the given std::function
void YourListener::on_data_available(DataReader* reader) override {
    IDL_DATA data_;
    SampleInfo info_;
    // Add custom condition in the loop for listening to data, i.e. while true
    while (true && (RETCODE_OK == reader->take_next_sample(&data_, &info_))) {
        if ((info_.instance_state == ALIVE_INSTANCE_STATE) && info_.valid_data) {
            // Run your callback method here
            data_received(&data_);
        }
    }
}

youcode.cpp

// define the callback
std::function<void(const IDL_DATA&)> data_received_callback = [&](const IDL_DATA& data)
                {
                    // Use safely your IDL_DATA data here
                    // ... data.method()
                };
// create the library interface with that callback, so it is (internally) propagated to the listener constructor
MyDDS myDDS(&data_received_callback); // Library interface propagates the callback to the listener constructor
myDDS.readData(); // starts internal subscriber in library which will run your custom callback on data available

The reader will discard the data when read in the callback? So it will be invalid to access by the program via API?

No, data is not discarded. The data is written in the given buffer, so the lifecycle of the data is the IDL_DATA variable lifecycle itself. The use of take method(s) would remove the sample from the DataReader, but using read method(s) would allow you to retrieve the sample in future queries. Refer to the Accessing received data section of the documentation for more information.

In this case, if the IDL_DATA data_ is defined in the on_data_available callback, the data_ lifecycle will persist during your data_received(const IDL_DATA&) callback and will be removed once it finishes (it will be reassigned in the next iteration of the on_data_available callback).
You would need to perform your API calls in the std::function (// Use safely your IDL_DATA data here). In any other case, a copy of the data should be done.

On the other hand, in case a blocking behavior is required (you want to block the data reception with busy waiting in the myDDS.readData(&data); method until it has been processed to start processing next sample), you should change the code to use DataReader::wait_for_unread_message(). Please refer to this documentation section.

myDDS.cpp

myDDS::readData(const IDL_DATA& data)
{
    // ...

    SampleInfo info_;
    // Add custom condition in the loop for listening to data, i.e. while true
    while (true) {
        // private data_reader_ declared performs the wait_for_unread_message call with a 2 seconds timeout
        if (data_reader_->wait_for_unread_message(eprosima::fastdds::dds::Duration_t(2, 0))){
            if (RETCODE_OK == data_reader_->take_next_sample(&data_, &info_){
                 if ((info_.instance_state == ALIVE_INSTANCE_STATE) && info_.valid_data) {
                    // Use safely your IDL_DATA data here
                    // ... data.method()
                }
            }
        }
        else {
            std::cout << "No data received in 2 seconds" << std::endl;
        }
    }
}

youcode.cpp

IDL_DATA data; // IDL data type
MyDDS myDDS; // Library interface
myDDS.readData(&data);  // busy wait reading 

In this case, the IDL_DATA data_ has been defined out of the myDDS class, so the data lifecycle will persist the myDDS.readData(&data) call but will be overwritten in each take_next_sample call.
You would need to perform your API calls in the loop (// Use safely your IDL_DATA data here). In any other case, a copy of the data should be done.

[ploe24]

@JesusPoderoso

Thanks a lot for your detailed answer :)

One quick question before I close this topic as resolved:
I've noticed when you (forcefully) terminate a subscriber, doesn't matter in which way or how it's configured (standalone or in our library), the publisher keeps publishing for exactly 20 seconds, before their readers and writers unmatch and the connection stops.
Is this "timeout" documented somewhere? I've checked the eProsima FastDDS Documentation, but could not find anything related?!
Is it because of RTPS standard? Is it possible to change this timeout?

Thanks again for your effort, really appreciate it :)

[JesusPoderoso]

@ploe24

First of all, you should manage possible crashes or handle signals in your application and delete the Fast DDS entities before the application ends (see hello world example). The correct way to do so is using the participant's delete_contained_entities method and the DomainParticipantFactory's delete_participant method (see hello world example too).

Is this "timeout" documented somewhere?

Yes, you are describing the Participant's Discovery Lease Duration. The participant waits those 20 seconds by default before considering a remote participant not alive.

Is it possible to change this timeout?

Yes, this code snippet shows how to configure the participant's discovery lease duration.

[ploe24]

First of all, you should manage possible crashes or handle signals in your application and delete the Fast DDS entities before the application ends (see hello world example). The correct way to do so is using the participant's delete_contained_entities method and the DomainParticipantFactory's delete_participant method (see hello world example too).

@JesusPoderoso Yes I thought so, but since our subscribers and publishers were built upon the "old" examples, if you just terminate a subscriber test app which in turn uses our library, there is no grateful exit. The publisher continues to publish for the 20s lease duration until it stops.
We will change that later...

Thank you for the helpful links

Have a nice day :)

[JesusPoderoso]

Hi @ploe24, other options are accessing data with a waiting thread using waitsets (as it is done in the hello world example), or using non-blocking calls with the DataReade method DataReader::wait_for_unread_message().
These options are interesting for releasing the Fast DDS data reception thread and managing the reception and data processing in the application thread.

[ploe24]

@JesusPoderoso

Had to re-open this again 😞

The std::function callback approach works well in our C++ clients.
However when using the SWIG wrapper for Python we haven't found a way to pass a function to the wrapped C++ Lib.
The signature of the function inside the library is:

void setCallback(std::function<void(const IDL_TYPE&)> callback);

and instead of

std::function<void(const IDL_TYPE&)> callback = [&](const IDL_TYPE& data) { // ... };

we use typing.Callable as some kind of replacement for std::function, as some state that the type conversion should work, which is not the case :(

Is there an easy way to use the custom callback approach in a SWIG Python client?

Do we have to re-configure the SWIG interface in order to be able to deal with std::function?!

[JesusPoderoso]

Hi @ploe24
Check the Fast DDS Python HelloWorld example for an equivalent listeners example. In particular, in this section you can replicate the discussed C++ scenario using a lambda function.
You may also consider contacting commercials support for premium support options.

I think that the changes would be similar to:

class ReaderListener(fastdds.DataReaderListener):
    def __init__(self, custom_callback):
        super().__init__()
        self.custom_callback =  custom_callback

    def on_data_available(self, reader):
        info = fastdds.SampleInfo()
        data = HelloWorld.HelloWorld()
        reader.take_next_sample(data, info)

        self.custom_callback(data)

# ...

class Reader():
  def __init__(self, domain, custom_callback):
    # ...
    self.listener = ReaderListener(custom_callback)

# ...

def my_custom_callback(data):
    # Use your IDL_DATA data here
    # ... data.method()

# __main__
# ...

reader = Reader(args.domain, my_custom_callback)
reader.run()

Hope it helps!

[ploe24]

@JesusPoderoso

Hi thanks for your reply.

Unfortunately this is not what I meant.
We are still using the C++ library "in the back", built the FastDDS_Python module and wrapped it using SWIG.
The python "client" only accesses the C++ library via its public (C++) functions.
One of these functions is void setCallback(std::function<void(const IDL_TYPE&)> callback);, which sets a std::function to the library's internal reader.

Calling this function from the main Python *.py file results in a parameter mismatch and other errors, because it seems that SWIG does not know how to deal with std::function out-of-the-box. We tried to pass a Python Callable object as well as a regular Python function to setCallback( ... ).
Both did not work :(

I've found this on StackOverflow, but it looks pretty complicated.
How to use SWIG to wrap std::function objects?
Is it needed, though or is there another way? Do we have to add something to the SWIG.i files of FastDDS_Python for this?!

[JesusPoderoso]

Hi @ploe24, sorry for the inconvenience.
I am afraid there is no fast-forward way to perform that task.
We have reached it in this project, you may take a look and see how it is implemented. I would say it is even more complicated than the StackOverlow thread you have shared.
Do not hesitate to contact support for premium support options.