How to get cdr Stream directly?

5 posts / 0 new
Last post
Offline
Last seen: 2 months 6 days ago
Joined: 04/20/2022
Posts: 2
How to get cdr Stream directly?

Hello All,

I am using rti_connext_dds-6.0.0 with modern C++ API and going to develop an application doing things similar to the Recorder service. Which means I don't need to deserialize cdr stream to sample. For now, I use to_cdr_buffer() function to serialize samples I get. In , they have callback OnRawDataAvailable() which is called when apps get stream data. Are there some ways to do the same thing in RTI DDS? Please help. Thanks.

Keywords:
Howard's picture
Offline
Last seen: 2 days 7 hours ago
Joined: 11/29/2012
Posts: 383

Are you asking if you can get the data for a DataReader before it's deserialized by DDS?  No, there is no way to intercept the data.  DDS will take a data packet, deserialize it into a typed-data structure (aka data Sample) and store it in the DataReader's receive cache before any user code is called.

Offline
Last seen: 2 months 6 days ago
Joined: 04/20/2022
Posts: 2

No any callback function or listener between "take a data packet" and "store it in the DataReader's receive cache"? I found an article https://community.rti.com/sites/default/files/knowledge_base/Sol00005071.txt. Is that possible to get cdr stream by modify Plugin_deserialize()?

 
Howard's picture
Offline
Last seen: 2 days 7 hours ago
Joined: 11/29/2012
Posts: 383

No explicit callback just to do that.  And even if there was, DDS would expect it to work...even if you modify it to do other things.

This function:

xxxPlugin_deserialize_from_cdr_buffer();

is called by

void topic_type_support< xxx >::from_cdr_buffer(xxx& sample,
        const std::vector<char>& buffer);

 

and while it is used in the code to store a received CDR buffer in the receive cache...it may be called for other reasons by DDS.   And these functions are generated on a per-data type basis.  That includes datatypes that are used as members of other data types.

And fundamentally, they still have to do the deserialization of the buffer.

The question is why do you need to store the serialized buffer?   Have you determined that the amount of CPU/time for deserialization of your data stream exceeds the performance requirements of your system?

Why are you writing your own version of RTI Recording Service? If you don't like how RTI Recording Service stores data, you can modify RTI Recording Service can be modified to transform the data before it's stored or use a different storage mechanism than the builtin SQLite database file.

See: https://community.rti.com/static/documentation/connext-dds/6.1.1/doc/manuals/connext_dds_professional/services/recording_service/sdk.html

 

Offline
Last seen: 1 month 3 weeks ago
Joined: 02/20/2014
Posts: 10

I've done a similar thing since RTI Recording Service is unavailable/unusable in some use cases. The closest I could figure out to avoid the deserialization step was to switch to using DynamicData for the recorder's subscriber, which you could then fairly easily dump to a serialized buffer. I didn't end up doing it that way though since, at least at the time, Connext wouldn't let you register a normal autogen type and a dynamic data type which was a requirement for the implementation. If I really cared to I could've instantiated a new participant for the recorder, but that's more system overhead to then deal with that.

I've recently considered too that potentially you could implement a custom recorder transport that might let you pick off serialized data that way, but seems like it's a fair amount of work given that custom transports seem to be questionably supported in more recent Connext versions.

In the mean-time I'm basically doing it the way discussed above. Normal subscriber receives a deserialized message, it gets reserialized into a new CDR buffer, saves that serialized buffer out, and then post-process gets deserialized (which could be done in a number of different ways). It's more overhead for sure, but for our message sizes and quantities it wasn't a problem.