Your Historical Data, Now On-Demand

It is not always practical, or even possible, for an application to know what subset of data it will need access to as that data is being published. For example, imagine an application that subscribes to one subset of data during normal operating conditions, but then after an alarm is triggered all data from the past 24 hours is required. Does it make sense for the subscribing application to receive all data at all times just in case the alarm sounds, taking up bandwidth and resources in the subscribing application? Or does it make more sense for the subscribing application to be able to request the data it needs at the moment it needs it, saving bandwidth and resources in the common case? The second option is the most obvious choice, right? We thought so too, which is why with the introduction of Topic Queries in RTI Connext DDS 5.3.0, we allow DataReaders to do just that – send a request to all matching DataWriters for historical data that matches a given filter expression.

Topic Queries introduce another way for DDS developers to intelligently distribute data within their systems. Content filtered topics (CFTs) allow a DataReader to subscribe to a subset of live data from a Topic. They enable DataWriters to only publish data that is currently of interest to its matching DataReaders – saving bandwidth and complicated application logic that would have to otherwise handle filtering after reception. Topic Queries complement CFTs by adding the ability to request previously published, and even previously filtered, samples allowing applications to scalably respond to dynamic and changing requirements within a system.

DataReaders create Topic Queries with an associated filter and filter expression which selects the subset of data that the DataReader is requesting from its matching DataWriters. For example, you would do the following to a create a Topic Query that will request all samples that are available in matching DataWriters’ caches where x=5:

struct DDS_TopicQuerySelection selection = 
       DDS_TopicQuerySelection_INITIALIZER;
struct DDS_TopicQuery *topic_query = NULL;

selection.filter_expression = “x = 5”; 
topic_query = DDS_DataReader_create_topic_query(reader, &selection);

The above snippet will trigger a request that is sent out on a new built-in ServiceRequest channel. All matching DataWriters that have data in their caches where x=5 will respond to the request by republishing those samples. Samples that are sent in response to Topic Queries are sent in parallel to live data, meaning that Topic Queries will not interrupt the flow of live data. Samples associated with a topic query can also be identified and read separately from live data using the new topic_query_guid field in the SampleInfo and the stream_kinds field in the ReadConditionParams. For example, to read only samples that are in response to a Topic Query, an application could do the following:

struct DDS_ReadConditionParams read_cond_params =
       DDS_READCONDITIONPARAMS_DEFAULT;

/* Read only samples that are in response to a Topic Query */
read_cond_params.stream_kinds = DDS_TOPIC_QUERY_STREAM;
read_cond = DDS_DataReader_create_readcondition_w_params(
       reader,
       &read_cond_params);

retCode = FooDataReader_read_w_condition(
       FooDataReader_narrow(reader),
       &dataSeq,
       &infoSeq,
       DDS_LENGTH_UNLIMITED,
       (DDS_ReadCondition *)read_cond);

As an added bonus, Topic Queries work with Routing Service enabling the propagation of historical data through Routing Service for late-joining DataReaders, which was a previously unsupported use-case. Before the introduction of Topic Queries, a late-joining DataReader that was matched with an existing Routing Service stream would not trigger the transmission of historical data from the original DataWriter on the other side of the Routing Service. Now, the DataReader can create a Topic Query and request all historical data from the original DataWriter.

To learn more about Topic Queries and the associated APIs and QoS Policies, refer to section 6.5.22 ‘TOPIC_QUERY_DISPATCH_QosPolicy (DDS Extension)’ and Chapter 22 ‘Topic Queries’ in the RTI Connext DDS Core Libraries User’s Manual (HTML|PDF) and Chapter 10 ‘Topic Query Support’ in the Routing Service User’s Manual (PDF).