DataReader Use Cases

The following #includes are needed for the examples on this page

#include <dds/sub/ddssub.hpp>
#include <dds/domain/ddsdomain.hpp>
#include <dds/core/ddscore.hpp>
#include <dds/topic/ddstopic.hpp>
#include <dds/pub/ddspub.hpp>
#include <rti/core/ListenerBinder.hpp>

#include "Foo.hpp"

Setting up a DataReader

• Set up a Subscriber

• Set up a Topic

• Create a DataReader of user-defined type Foo with each of the available constructors

      // Create the Participant
      dds::domain::DomainParticipant participant(0);
  
      // Create a Topic
      dds::topic::Topic<Foo> topic(participant, "MyTopicName");
  
      // When not explicitly provided to the constructor, the 
      // created DataReader's QoS is default, has a NULL listener, and a status mask
      // dds::core::status::StatusMask::all()
  
      // In all of the following constructors 'topic' could also be an instance 
      // of dds::topic::ContentFilteredTopic 
       
      // Creating a DataReader, creating the subscriber inline
      dds::sub::DataReader<Foo> 
          reader1(dds::sub::Subscriber(participant), topic);
  
      // Create a Subscriber
      dds::sub::Subscriber subscriber(participant);
  
      // Creating a DataReader with modified QoS
      dds::sub::qos::DataReaderQos qos;
      qos << dds::core::policy::History::KeepAll();
      dds::sub::DataReader<Foo> reader2(subscriber, topic, qos);
  
      // Creating a DataReader with modified QoS and a listener
      ExampleDataReaderListener *listener = new ExampleDataReaderListener;
      dds::sub::DataReader<Foo> reader3(
         subscriber, 
         topic, 
         qos, 
         listener);
  
      // Creating a DataReader with modified QoS, a listener, and a liveliness_changed() 
      // status mask
      dds::sub::DataReader<Foo> reader4(
         subscriber, 
         topic, 
         qos, 
         listener, 
         dds::core::status::StatusMask::liveliness_changed());
  
      // Creating an entity with a listener automatically retains the entity, so 
      // we must either explicitly close the DataReader...
      reader3.close();
      
      // ... or set the listener to null
      reader4.listener(NULL, dds::core::status::StatusMask::none());

Managing instances

• Getting an instance key value of user data type Foo

      // Instead of a sample of type Foo, key_value() could also take a 
      // TopicInstance<Foo> object. key_value only populates the key field in the 
      // sample
      Foo retrieved_sample_key;
      reader.key_value(retrieved_sample_key, handle);

Set up a DataReader to access received data

• Set up a DataReader

• Set up the DataReader to handle the DDS_DATA_AVAILABLE_STATUS status, in one or both of the following two ways.

• Enable DDS_DATA_AVAILABLE_STATUS for the DDSDataReaderListener associated with the data reader
  • The processing to handle the status change is done in the DataReaderListener on_data_available() method of the attached listener.
  • Typical processing will access the received data.

• Enable DDS_DATA_AVAILABLE_STATUS for the DDSStatusCondition associated with the data reader
  • The processing to handle the status change is done when the data reader's attached status condition is triggered and the DDS_DATA_AVAILABLE_STATUS status on the data reader is changed.
  • Typical processing will access the received data.

Accessing Received Data

Reading data samples

You can read data samples from a DataReader by receiving a loan, inspecting the samples and then returning the loan, or you can directly copy them.

• To iterate through loaned data samples in your application you will use a LoanedSamples container. The following code shows different ways to use this container

      // Take all the samples in the reader queue (if you don't want to remove
      // them from the queue and read them later again, use reader.read() instead)
      dds::sub::LoanedSamples<Foo> samples = reader.take();
  
      // LoanedSamples<Foo> is a container of LoanedSample<Foo> elements, which
      // contain the actual data (Foo) and the SampleInfo
      for (dds::sub::LoanedSamples<Foo>::iterator sample_it = samples.begin();
           sample_it != samples.end();
           ++sample_it) {
  
          if (sample_it->info().valid()) {
              std::cout << "Received: " << sample_it->data() << std::endl;
          }
      }
  
      // The previous code is simpler with C++11's range-based for-loop
      for (const auto& sample : samples) {
          if (sample.info().valid()) {
              std::cout << "Received: " << sample.data() << std::endl;
          }
      }
  
      // You can also access samples using the subscript operator
      if (samples.length() >= 2) {
          if (samples[1].info().valid()) {
              std::cout << "Second sample is " << samples[1].data() << std::endl;
          } else {
              std::cout << "Second sample is invalid\n";
          }
      }
  
      // You can copy the samples using std::copy:
      std::vector<rti::sub::LoanedSample<Foo> > samples_copy;
      std::copy(samples.begin(), samples.end(), std::back_inserter(samples_copy));
  
      // You can also access the data directly. A sample can be implicitly
      // converted into a reference to its data type as long as it contains
      // valid data.
      //
      // The iterator adapter valid_samples will skip samples containing
      // invalid data
      std::vector<Foo> data_copy;
      std::copy(
          rti::sub::valid_samples(samples.begin()),
          rti::sub::valid_samples(samples.end()),
          std::back_inserter(data_copy));
  
      // Attempting to access invalid data will result in an exception.
      try {
          // The following will throw if !samples[0].info().valid()
          Foo data = samples[0].data();
          // The following will throw if any sample contains invalid data
          std::copy(samples.begin(), samples.end(), std::back_inserter(data_copy));
      } catch (const dds::core::PreconditionNotMetError& ex) {
          std::cerr << "Attempted to access sample containing invalid data: "
                    << ex.what() << std::endl;
      }
  
      // You can also apply other iterator-based algorithms,
      // like this example that prints just one field of the data type
      // (assuming that Foo contains an integer field called 'x')
      std::transform(
          rti::sub::valid_samples(samples.begin()),
          rti::sub::valid_samples(samples.end()),
          std::ostream_iterator<int>(std::cout, ", "),
          [](const Foo& data) { return data.x(); });
  
      // Once a LoanedSamples object goes out of scope its destructor
      // automatically returns the loan to the middleware. If you need to
      // explicitly return it, call return_loan()
      //
      // samples.return_loan()
  

• Although the LoanedSamples is the most generic way to access data in a DataReader, for convinience other overloaded versions of read/take allow to directly copy the data

  
      // The first version of read/take receives a back-insert iterator
      std::vector<dds::sub::Sample<Foo> > my_vector1;
      reader.take(std::back_inserter(my_vector1));
  
      // The second version receives a forward iterator and the maximum size
      std::vector<dds::sub::Sample<Foo> > my_vector2(10);
      reader.take(my_vector2.begin(), 10);
  

• See also Working with IDL types

Selecting what samples to read

The functions read() and take() select all samples in the DataReader queue. The function select() allows specifying different criteria about the data to read. A equivalent API, using the streaming >> operator is available too.

• select() example

  
      // To select what samples you want to read (or take) from a DataReader
      // queue, you can use a Selector. When you don't specify a parameter in the
      // Selector, the default applies.
      dds::sub::LoanedSamples<Foo> samples =
          // Call select() to create a Selector and then specify its parameters
          reader.select()
              // Specifies the maximum number of samples to read/take
              // in this function call. Default: set by Qos.
              .max_samples(20)
              // Specifies the sample, view and instance states - default: all.
              .state(dds::sub::status::DataState::new_instance())
              // Specifies a query on the data sample - default: all
              .content(dds::sub::Query(reader, "x > 10"))
              // Specifies what instance to read - default: all instances
              .instance(my_instance_handle)
              // And finally call read() or take()
              .take();
  
      // You can also specify state and content by using a Condition. Reusing
      // a Condition for multiple reads is more efficient than specifying the
      // content() every time.
      dds::sub::cond::QueryCondition query_condition(
          dds::sub::Query(reader, "x > 10"),
          dds::sub::status::DataState(
              dds::sub::status::SampleState::not_read(),
              dds::sub::status::ViewState::new_view(),
              dds::sub::status::InstanceState::alive()));
  
      samples = reader.select()
          .condition(query_condition)
          .take();
  
      // And you can read instance by instance using next_instance().
      //
      // To read the first instance, use InstanceHandle::nil()
      dds::core::InstanceHandle previous_handle = dds::core::InstanceHandle::nil();
      do {
          samples = reader.select()
              .next_instance(previous_handle)
              .take();
  
          if (samples.length() > 0) {
              // do something with samples ...
  
              // Update previous_handle to read the instance following this one
              previous_handle = samples[0].info().instance_handle();
          }
      } while (samples.length() > 0);
  
      // select() returns an object you can reuse
      dds::sub::DataReader<Foo>::Selector selector = reader.select().max_samples(20);
      samples = selector.read();
      // ...
      selector.state(dds::sub::status::DataState::new_instance());
      samples = selector.take();
  

• Streaming example

  
      using namespace dds::sub;
  
      // The overloaded shift operator defines a domain-specific language to
      // specify what samples to read. The following code has the same effect as
      // the first select() example from the previous code snippet
      LoanedSamples<Foo> samples;
      reader >> take // first indicate if you want to read or take;
             >> max_samples(20) // then specify the parameters you want...
             >> state(status::DataState::new_instance())
             >> content(Query(reader, "x > 10"))
             >> instance(my_instance_handle)
             >> samples; // and finally, the destination LoanedSamples container.
  

Reading samples using coherent access

If a Subscriber's PresentationQosPolicy access_scope is GROUP and ordered_access is true, the application can access the samples in order across DataWriters of the same group (Publisher with access_scope GROUP). The queue is ordered/sorted per topic across all instances belonging to DataWriters (or DataReaders) within the same Publisher (or Subscriber).

• Read samples using CoherentAccess

  void howto_read_coherently()
  {
      dds::domain::DomainParticipant participant(0);
      dds::topic::Topic<Foo> topic1(participant, "Topic1");
      dds::topic::Topic<Foo> topic2(participant, "Topic2");
  
      // Create a Subscriber with group presentation. The Publisher must also
      // have group presentation set in its QoS
      dds::sub::qos::SubscriberQos subscriber_qos;
      subscriber_qos << dds::core::policy::Presentation::GroupAccessScope(false, true);
      dds::sub::Subscriber subscriber(participant, subscriber_qos);
  
      dds::sub::DataReader<Foo> reader1(subscriber, topic1);
      dds::sub::DataReader<Foo> reader2(subscriber, topic2);
  
      // Assuming that a writer of "Topic1" wrote before the writer of "Topic2"
      // reader2 will only be able to access it's samples after reader1 takes 
      // its sample as long as CoherentAccess to the samples has been started
  
      int sample_count = read_coherently(subscriber, reader2);
      std::cout << sample_count << std::endl; // will be 0
  
      // The CoherentAccess destructor called when read_coherently exited 
      // will end coherent access to the samples. 
      // You can also explicitly call coherent_access.end() to end coherent access
  
      // reader2 can now access its sample 
      // ordered_samples.size() will = 1 
      dds::sub::LoanedSamples<Foo> ordered_samples = reader2.take();
  }
  
  template <typename T>
  int read_coherently(dds::sub::Subscriber& subscriber, dds::sub::DataReader<T>& reader)
  {
      dds::sub::CoherentAccess coherent_access(subscriber);
  
      dds::sub::LoanedSamples<T> ordered_samples = reader.take();
  
      return ordered_samples.length();
  }

DataReader Listeners

• Implementing a DataReaderListener class

  // To create a DataReaderListener, inherit from the NoOpDataReaderListener class
  // and then override any of the listener callback functions 
  class ExampleDataReaderListener : public dds::sub::NoOpDataReaderListener<Foo>
  {
  public:
      ExampleDataReaderListener() {}
  
  public:
      void on_sample_lost(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::SampleLostStatus &status)
      {
          std::cout << "on_sample_lost callback" << std::endl;
  
          // Access information from the status
          std::cout << "total_count = " << status.total_count() << std::endl;
          std::cout << "total_count_change = " << status.total_count_change() << std::endl;
      }
  
      void on_subscription_matched(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::SubscriptionMatchedStatus &status)
      {
          std::cout << "on_subscription_matched callback" << std::endl;
  
          // Access information from the status
          std::cout << "total_count = " << status.total_count() << std::endl;
          std::cout << "total_count_change = " << status.total_count_change() << std::endl;
          std::cout << "current_count = " << status.current_count() << std::endl;
          std::cout << "current_count_change = " << status.current_count_change() << std::endl;
  
          dds::core::InstanceHandle handle = status.last_publication_handle();
  
          // Extension
          std::cout << "current_count_peak = " << status->current_count_peak() << std::endl;
      }
  
      void on_sample_rejected(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::SampleRejectedStatus &status)
      {
          std::cout << "on_sample_rejected callback" << std::endl;
  
          // Access information from the status
          std::cout << "total_count = " << status.total_count() << std::endl;
          std::cout << "total_count_change = " << status.total_count_change() << std::endl;
          std::cout << "last_reason = " << status.last_reason() << std::endl;
  
          dds::core::InstanceHandle handle = status.last_instance_handle();
      }
  
      void on_requested_incompatible_qos(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::RequestedIncompatibleQosStatus &status)
      {
          std::cout << "on_requested_incompatible_qos callback" << std::endl;
  
          // Access information from the status
          std::cout << "total_count = " << status.total_count() << std::endl;
          std::cout << "total_count_change = " << status.total_count_change() << std::endl;
          std::cout << "last_policy_id = " << status.last_policy_id() << std::endl;
  
          dds::core::policy::QosPolicyCountSeq qos_seq = status.policies();
          if (qos_seq.size() > 0) {
              std::cout << "policy_id of one first incompatible Qos policy = " 
                  << qos_seq[0].policy_id() << std::endl;
          }
      }
  
      void on_requested_deadline_missed(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::RequestedDeadlineMissedStatus &status)
      {
          std::cout << "on_requested_deadline_missed callback" << std::endl;
  
          // Access information from the status
          std::cout << "total_count = " << status.total_count() << std::endl;
          std::cout << "total_count_change = " << status.total_count_change() << std::endl;
  
          dds::core::InstanceHandle handle = status.last_instance_handle();
      }
  
      void on_liveliness_changed(
         dds::sub::DataReader<Foo> &reader, 
         const dds::core::status::LivelinessChangedStatus &status)
      {
          std::cout << "on_liveliness_changed callback" << std::endl;
  
          // Access information from the status
          std::cout << "alive_count = " << status.alive_count() << std::endl;
          std::cout << "not_alive_count = " << status.not_alive_count() << std::endl;
          std::cout << "alive_count_change = " << status.alive_count_change() << std::endl;
          std::cout << "not_alive_count_change = " << status.not_alive_count_change() << std::endl;
  
          dds::core::InstanceHandle handle = status.last_publication_handle();
      }
  
      void on_data_available(dds::sub::DataReader<Foo> &reader)
      {
          std::cout << "on_data_available callback" << std::endl;
  
          // Read the available data
          dds::sub::LoanedSamples<Foo> samples = reader.read();
  
          // Print samples by copying to std::cout
          std::copy(
              samples.begin(),
              samples.end(),
              dds::sub::LoanedSamples<Foo>::ostream_iterator(std::cout, "\n"));
      }
  
  };

  
  
• Setting a listener on a DataReader

      dds::domain::DomainParticipant participant(0);
      dds::topic::Topic<Foo> topic(participant, "Topic1");
  
      ExampleDataReaderListener *listener = new ExampleDataReaderListener;
  
      // Create a DataReader with a listener and default 
      // dds::core::status::StatusMask::all()
      dds::sub::DataReader<Foo> reader1(
         dds::sub::Subscriber(participant), 
         topic,
         dds::sub::qos::DataReaderQos(), 
         listener);
  
      // Create a DataReader with a null listener
      dds::sub::DataReader<Foo> reader2(
         dds::sub::Subscriber(participant), topic);
  
      // Set the listener for the DataReader after creation
      reader2.listener(
         listener, 
         dds::core::status::StatusMask::sample_rejected());
  
      // Setting a listener on an entity automatically retains the entity so 
      // we must explicitly call close() in order for the DataReader to be
      // destroyed 
      reader1.close();
  
      // Instead of calling close, you can set the DataReader's listener to null 
      // and this will implicitly unretain the DataReader so that it will be 
      // destructed once all references to it go out of scope
      reader2.listener(NULL, dds::core::status::StatusMask::none());

  
  
• The ListenerBinder class will take care of setting the listener and mask for an entity during creation and detaching the listener from the entity when the ListenerBinder is destructed. This is helpful because you will not have to explcitly close the entity or detach the listener yourself in order to teardown the entity

      dds::domain::DomainParticipant participant(0);
      dds::topic::Topic<Foo> topic(participant, "Topic1");
  
      ExampleDataReaderListener *listener = new ExampleDataReaderListener;
  
      // Create a DataReader with a null listener
      dds::sub::DataReader<Foo> reader(
         dds::sub::Subscriber(participant), topic);
  
      // Attach the listener to the DataReader after creation using a 
      // ListenerBinder (Note: rti/core/ListenerBinder.hpp needs to be included)
      rti::core::ListenerBinder<dds::sub::DataReader<Foo> > scoped_listener =
          rti::core::bind_listener(
             reader,
             listener,
             dds::core::status::StatusMask::sample_lost());
  
      // Now there is no need to explicitly call close on the DataReader or set
      // the listener to null

Looking up DataReaders

• Lookup DataReaders by topic name

      dds::domain::DomainParticipant participant(0);
      dds::sub::Subscriber subscriber(participant);
  
      dds::topic::Topic<Foo> topic1(participant, "Topic1");
      dds::topic::Topic<Foo> topic2(participant, "Topic2");
  
      // Create three DataReaders for topic1 and one for topic2 and retain them
      create_and_retain_reader(subscriber, topic1);
      create_and_retain_reader(subscriber, topic1);
      create_and_retain_reader(subscriber, topic2);
  
      // We don't have any references to the created DataReaders but we can look 
      // them up using the subscriber and topic name that they were created with
      std::vector<dds::sub::DataReader<Foo> > readers;
  
      // reader_count will equal 1. If more than one DataReader belongs to a 
      // subscriber with the same topic name then find may return any one of them
      int reader_count = 
          dds::sub::find<dds::sub::DataReader<Foo> >(
             subscriber, 
             "Topic1", 
             std::back_inserter(readers));
  
      reader_count = 
          dds::sub::find<dds::sub::DataReader<Foo> >(
             subscriber, 
             "Topic2", 
             std::back_inserter(readers));
  

Getting Matched Publications

• Lookup a DataReader's matched publications

      dds::domain::DomainParticipant participant(0);
      dds::topic::Topic<Foo> topic(participant, "Topic1");
      dds::sub::DataReader<Foo> reader(dds::sub::Subscriber(participant), topic);
  
      // Create two writers that will match with the above reader
      dds::pub::DataWriter<Foo> writer1(dds::pub::Publisher(participant), topic);
      dds::pub::DataWriter<Foo> writer2(dds::pub::Publisher(participant), topic);
  
      // Get a list of the InstanceHandles corresponding to the DataReader's 
      // matched publications.
      dds::core::InstanceHandleSeq publication_handles1 = 
          dds::sub::matched_publications(reader);
  
      // matched_publications also takes in an iterator range and will return
      // an InstanceHandle iterator to the end of the InstanceHandleSeq
      dds::core::InstanceHandleSeq publication_handles2(2, dds::core::null);
  
      dds::core::InstanceHandleSeq::iterator last = 
          dds::sub::matched_publications(
             reader, 
             publication_handles2.begin(), 
             publication_handles2.end());
  
      // now last == publication_handles2.end()

  
  
• Use the InstanceHandles obtained from matched_publications() to get the PublicationBuiltinTopicData for the publications

      dds::topic::PublicationBuiltinTopicData publication_data =
          dds::sub::matched_publication_data(reader, publication_handles1[0]);

Accessing the Built-in Subscriber and DataReaders

• The next two examples assume the following setup

      // It is common to create entities disbaled when accessing and setting 
      // listeners on the built-in DataReaders so that information about
      // discovery traffic is missed
      dds::domain::qos::DomainParticipantFactoryQos qos;
      qos << dds::core::policy::EntityFactory(false);
      dds::domain::DomainParticipant::participant_factory_qos(qos);
  
      dds::domain::DomainParticipant participant(0);

  
  
• Access the built-in Subscriber

      dds::sub::Subscriber builtin_subscriber = dds::sub::builtin_subscriber(participant);

  
  
• Access the built-in DataReaders

      // There are constants for each of the built-in topics accessible through
      // the dds::topic namespace--see below for their use
  
      // Access the built-in DataReader for ParticipantBuiltinTopicData 
      std::vector<dds::sub::DataReader<dds::topic::ParticipantBuiltinTopicData> > participant_reader;
      int reader_count = 
          dds::sub::find<dds::sub::DataReader<dds::topic::ParticipantBuiltinTopicData> >(
             builtin_subscriber, 
             dds::topic::participant_topic_name(),
             std::back_inserter(participant_reader));
  
      // Access the built-in DataReader for PublicationBuiltinTopicData 
      std::vector<dds::sub::DataReader<dds::topic::PublicationBuiltinTopicData> > publication_reader;
      reader_count = 
          dds::sub::find<dds::sub::DataReader<dds::topic::PublicationBuiltinTopicData> >(
             builtin_subscriber, 
             dds::topic::publication_topic_name(),
             std::back_inserter(publication_reader));
  
      // Access the built-in DataReader for SubscriptionBuiltinTopicData 
      std::vector<dds::sub::DataReader<dds::topic::SubscriptionBuiltinTopicData> > subscription_reader;
      reader_count = 
          dds::sub::find<dds::sub::DataReader<dds::topic::SubscriptionBuiltinTopicData> >(
             builtin_subscriber, 
             dds::topic::subscription_topic_name(),
             std::back_inserter(subscription_reader));
  
      // participant_reader[0], publication_reader[0] and subscription_reader[0]
      // will now hold reference to the built-in DataReaders. 

Using untyped DataReaders

When you want to hold a reference to a DataReader and the type T doesn't matter, you can use the class dds::sub::AnyDataReader, which provides type-independent methods.

• For example, this function receives any DataReader

  // This function can receive any DataReader, independently of the template types
  void example_untyped(const dds::sub::AnyDataReader& untyped_reader)
  {
      // Use type-independent operations
      std::cout << "The topic name is: " << untyped_reader.topic_name() << std::endl;
  
      if (untyped_reader.type_name() == "Foo") {
          // Get back the typed DataReader.
          //
          // If the reader's type is not Foo, this operation throws
          // dds::core::InvalidDowncastError.
          //
          // typed_reader and untyped_reader share references to the same object.
          dds::sub::DataReader<Foo> typed_reader = untyped_reader.get<Foo>();
      }
  }

  
  
• And this is how you can call it

      dds::sub::DataReader<Foo> reader(subscriber, topic);
      example_untyped(reader); // automatic conversion to AnyDataReader

Tearing down a DataReader

• See Tearing Down An Entity for examples of the various ways to teardown a DataReader