Request-Reply Examples

Examples on how to use the request-reply API .

Examples on how to use the request-reply API .

Request-Reply code examples.

Request-Reply Examples

Requesters and Repliers provide a way to use the Request-Reply communication pattern on top of the DDS entities. An application uses a Requester to send requests to a Replier; another application using a Replier receives a request and can send one or more replies for that request. The Requester that sent the request (and only that one) will receive the reply (or replies).

DDS Types

RTI Connext uses DDS data types for sending and receiving requests and replies. Valid types are those generated by the rtiddsgen code generator, the DDS built-in types, and DynamicData. Refer to the Core Libraries User's Manual and the following links for more information:

• Code Generator User's Manual,
• Using the DDS built-in types,

• Using DynamicData

Set up

• Create a DomainParticipant
• Create a requester
• Create a requester with parameters
• Create a replier

Requester: sending requests and receiving replies

• Basic Requester example
• Taking loaned samples
  
• Taking samples by copy
  
• Correlation between requests and replies

Replier: receiving requests and sending replies

• Basic Replier example
• SimpleReplier example

• Taking loaned samples

• Taking samples by copy

Error handling

• Error handling example

Note

To use Request-Reply you need to include the additional rticonnextmsg.jar to build your Java application.

Creating a Requester

• Setting up a participant
• Creating a Requester

          
          // Note: error checking omitted for simplicity
          
          // Create a DomainParticipant
          DomainParticipant participant = 
                  DomainParticipantFactory.get_instance().create_participant(
                      domainId, DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, 
                      null, StatusKind.STATUS_MASK_NONE);
                 
          // Create a Requester (if creation fails an exception is thrown)
          // Foo is the request type and FooTypeSupport is its DDS TypeSupport
          // Bar is the reply type and BarTypeSupport is its DDS TypeSupport
          Requester<Foo, Bar> requester = 
                  new Requester<Foo, Bar>(
                          participant, "TestService",
                          FooTypeSupport.get_instance(),
                          BarTypeSupport.get_instance());
          

• Finalizing a Requester

          requester.close();

See also

Creating a Requester with optional parameters

Configuring Request-Reply QoS profiles

Creating a Requester with optional parameters

• Setting up a participant
• Creating a Requester with optional parameters

          
          // Note: error checking omitted for simplicity
          
          // Create a DomainParticipant
          DomainParticipant participant = 
                  DomainParticipantFactory.get_instance().create_participant(
                          0, DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, 
                          null, StatusKind.STATUS_MASK_NONE);
          
          // Create a Requester with a QoS profile (located for example in
          // USER_QOS_PROFILES.xml, in the current working directory)
          Requester<Foo, Bar> requester = 
                  new Requester<Foo, Bar>(
                      new RequesterParams(
                          participant,
                          FooTypeSupport.get_instance(),
                          BarTypeSupport.get_instance())
                      .setServiceName("TestService")
                      .setQosProfile("RequestReplyExampleProfiles", 
                                     "RequesterExampleProfile"));
          

  See also

  Creating a Requester

Basic Requester example

• Creating a Requester
• Creating a Requester with optional parameters
• Basic Requester example

          
          // Send request
          Foo request = new Foo();
          request.message = "A request";
          requester.sendRequest(request);
          
          // Create reply Sample
          Sample<Bar> reply = requester.createReplySample();
          
          // Receive reply (wait for it and get the sample)
          boolean received = requester.receiveReply(reply , MAX_WAIT);
          
          if (received) {
              if (reply.getInfo().valid_data) {
                  System.out.println("Received reply: " + reply.getData().message);
              } else {
                  System.out.println("Received invalid reply");
              }
          } else {
              System.out.println("Reply not received");
          }
          

• Basic Requester example using the String built-in type

          
          // Create a requester using the String 
          // type with the DDS StringTypeSupport
          Requester<String, String> requester = 
                  new Requester<String, String>(
                      new RequesterParams(
                          participant,
                          StringTypeSupport.get_instance(),
                          StringTypeSupport.get_instance())
                      .setServiceName("TestServiceString")
                      .setQosProfile("MyLibrary", "RequesterProfile"));
          
          // Send request
          requester.sendRequest("A request");
          
          // Create reply Sample
          Sample<String> reply = requester.createReplySample();
          
          // Receive reply (wait for it and get the sample)
          boolean received = requester.receiveReply(reply , MAX_WAIT);
          
          if (received && reply.getInfo().valid_data) {
              System.out.println("Received reply: " + reply.getData());
          }
          

• Basic Requester example using DynamicData.

          
          // Instantiate DynamicData type support. For simplicity we create them
          // from existing, generated TypeCode objects (Foo and Bar)
          DynamicDataTypeSupport fooDynamicTypeSupport = new DynamicDataTypeSupport(
                  FooTypeCode.VALUE, new DynamicDataTypeProperty_t());
          DynamicDataTypeSupport barDynamicTypeSupport = new DynamicDataTypeSupport(
                  BarTypeCode.VALUE, new DynamicDataTypeProperty_t());
          
          // Create a requester using a DynamicData type support
          Requester<DynamicData, DynamicData> requester = 
                  new Requester<DynamicData, DynamicData>(
                          participant, "TestService",
                          fooDynamicTypeSupport,
                          barDynamicTypeSupport);
          
          // Create a Foo instance
          DynamicData foo = (DynamicData) fooDynamicTypeSupport.create_data();
          foo.set_string("message", DynamicData.MEMBER_ID_UNSPECIFIED, 
                  "A dynamic request");
          
          // Send request
          requester.sendRequest(foo);
          
          // Create reply Sample
          Sample<DynamicData> reply = requester.createReplySample();
          
          // Receive reply (wait for it and get the sample)
          boolean received = requester.receiveReply(reply , MAX_WAIT);
          
          if (received && reply.getInfo().valid_data) {
              System.out.println("Received reply: " + 
                  reply.getData().get_string("message", DynamicData.MEMBER_ID_UNSPECIFIED));
          }
          

  See also

  Basic Replier example

Taking loaned samples

• Get iterator to loaned replies (no copies)

          
          // Take loaned samples
          // Unlike receiveReplies(), takeReplies() doesn't block. It returns an
          // empty collection if there are no replies at that moment.
          // (before that we can call requester.wait_for_replies)
          Sample.Iterator<Bar> replies = requester.takeReplies();
   
          try {
              while (replies.hasNext()) {
                  Sample<Bar> reply = replies.next();
                  System.out.println("Received: " + reply.getData().message);
              }
          } finally {
              replies.close();
          }
          

  A more compact version using a try-with-resources block (since Java 7):

  try (Sample.Iterator<String> replies = requester.getReplies()) {
      while (replies.hasNext()) {
          Sample<String> reply = replies.next();
          System.out.println("Received: " + reply.getData());
      }
  }

See also

Basic Requester example

Basic Replier example

Taking samples by copy

Taking samples by copy

• Get copies of the replies:

          
          ArrayList<Sample<Bar>> replies = new ArrayList<Sample<Bar>>();
          
          // Add to existing list, reusing and overwriting any existing elements
          requester.takeReplies(replies, 10);
          for (Sample<Bar> reply : replies) {
              System.out.println("Received: " + reply.getData().message);
          }
                  
          // Get replies in a newly created list
          List<Sample<Bar>> newList = requester.takeReplies(null, 10);
          for (Sample<Bar> reply : newList) {
              System.out.println("Received: " + reply.getData().message);
          }
          

See also

Basic Replier example

Taking loaned samples

Correlating requests and replies

• Creating a Requester
• Example 1) Waiting for a reply for a specific request

          
          // Create requests
          WriteSample<Foo> request1 = requester.createRequestSample();
          request1.getData().message = "Request 1";
          WriteSample<Foo> request2 = requester.createRequestSample();
          request2.getData().message = "Request 2";
          
          // Send requests using WriteSample
          requester.sendRequest(request1);
          requester.sendRequest(request2);
   
          // Wait for a reply to the second request
          Sample<Bar> reply = requester.createReplySample();
          requester.waitForReplies(1, MAX_WAIT, request2.getIdentity());
          requester.takeReply(reply, request2.getIdentity());
          
          assert reply.getRelatedIdentity().equals(request2.getIdentity());
          System.out.println("Received reply: " + reply.getData().message);
   
          // Wait for a reply to the first request
          requester.waitForReplies(1, MAX_WAIT, request1.getIdentity());
          requester.takeReply(reply, request1.getIdentity());
          
          assert reply.getRelatedIdentity().equals(request1.getIdentity());
          System.out.println("Received reply: " + reply.getData().message);
          

• Example 2) Correlate reply after receiving it

          
          // Create requests
          WriteSample<Foo> request1 = requester.createRequestSample();
          request1.getData().message = "Request 1";
          WriteSample<Foo> request2 = requester.createRequestSample();
          request2.getData().message = "Request 2";
          
          // Send requests using WriteSample
          requester.sendRequest(request1);
          requester.sendRequest(request2);
   
          // Wait for two replies
          boolean received = requester.waitForReplies(2, MAX_WAIT);
          if (!received) {
              System.out.println("Replies not received");
              return;
          }
          
          Sample.Iterator<Bar> replies = requester.takeReplies();
          try {
              while(replies.hasNext()) {
                  Sample<Bar> reply = replies.next();
                  if (reply.getRelatedIdentity().equals(
                          request1.getIdentity())) {
                      System.out.println(
                          "Received reply for request 1: " + reply.getData().message);
                  } else if (reply.getRelatedIdentity().equals(
                          request2.getIdentity())) {
                      System.out.println(
                          "Received reply for request 2: " + reply.getData().message);
                  } else {
                      // Should not happen
                      System.out.println("Received unexpected reply");
                  }
              }
          } finally {
              // Return the loan
              replies.close();
          }
          

See also

Basic Requester example

Basic Replier example

Creating a Replier

• Setting up a participant
• Creating a Replier

          
          // Create a DomainParticipant
          DomainParticipant participant = 
                  DomainParticipantFactory.get_instance().create_participant(
                          domain_id, DomainParticipantFactory.PARTICIPANT_QOS_DEFAULT, 
                          null, StatusKind.STATUS_MASK_NONE);
          
          // Create a Replier
          Replier<Foo, Bar> replier = 
                  new Replier<Foo, Bar>(
                          participant, "TestService",
                          FooTypeSupport.get_instance(),
                          BarTypeSupport.get_instance());
          

• Finalizing a Replier

          replier.close();

Basic Replier example

• Creating a Replier
• Basic Replier example

          
          // Create a Sample for reading
          Sample<Foo> request = replier.createRequestSample();
          
          // Receive one request
          boolean received = replier.receiveRequest(request, MAX_WAIT);
          
          if (!received) {
              System.out.println("Request not received");
              return false;
          }
          
          if (request.getInfo().valid_data) {
              // Send a reply for that request
              Bar reply = new Bar();
              reply.message = "Reply for " + request.getData().message;
              replier.sendReply(reply, request.getIdentity());
          }
          
          // Note: a replier can send more than one reply for the same request
          

See also

Basic Requester example

SimpleReplier example

• Implement a listener

      class MySimpleReplierListener implements SimpleReplierListener<Foo, Bar> {
   
          private Bar reply = new Bar();
          
          public Bar onRequestAvailable(Sample<Foo> request) {
              if (request.getInfo().valid_data) {
                  // Return a reply
                  reply.message = "Simple reply for " + request.getData().message; 
                  return reply;
              } else {
                  // Do not reply to dispose/unregister samples
                  return null;
              }
          }
   
          public void returnLoan(Bar reply) {
              // nothing to do
          }
          
      }

• And create a SimpleReplier with the listener:

          
          // Create a SimpleReplier using a DomainParticipant, service name and
          // the a SimpleReplierListener implementation
          SimpleReplier<Foo, Bar> replier = 
                  new SimpleReplier<Foo, Bar>(
                          participant, "TestService",
                          new MySimpleReplierListener(),
                          FooTypeSupport.get_instance(), 
                          BarTypeSupport.get_instance());
          
          // After creation the SimpleReplier is already active and may call
          // the listener
          

See also

Basic Requester example

Error handling example

• Catching an exception from the request-reply API

   
          // All the operations in the request-reply API
          // throw exceptions in case of error.
   
          // You can also set the verbosity level to warning for
          // additional information
          Logger.get_instance().set_verbosity(
                  LogVerbosity.NDDS_CONFIG_LOG_VERBOSITY_WARNING);
   
   
          //
          // For example, this replier is sending a reply for a nonexistent request
          //
          WriteSample<Bar> reply = replier.createReplySample();
          SampleIdentity_t invalidRequestId = new SampleIdentity_t();
          
          
          try {
              replier.sendReply(reply, invalidRequestId);
          } catch (RETCODE_BAD_PARAMETER ex) { 
              // You can also catch RETCODE_ERROR, which is more generic
              System.out.println("Exception while sending reply: " + ex);
          }
   

Configuring Request-Reply QoS profiles

If you do not specify your own quality of service parameters (in com.rti.connext.requestreply.RequesterParams and com.rti.connext.requestreply.ReplierParams<TReq,TRep>), a com.rti.connext.requestreply.Requester<TReq,TRep> and com.rti.connext.requestreply.Replier<TReq,TRep> are created using a default configuration defined in the built-in profile com.rti.dds.infrastructure.BuiltinQosProfiles.PROFILE_PATTERN_RPC. You use this built-in profile as the base for your own profiles:

<?xml version="1.0" encoding="UTF-8"?>
<!--
Description
Example XML QoS Profile for HelloWorld
 
 
(c) Copyright, Real-Time Innovations, 2012.  All rights reserved.
RTI grants Licensee a license to use, modify, compile, and create derivative
works of the software solely for use with RTI Connext DDS. Licensee may
redistribute copies of the software provided that all such copies are
subject to this license. The software is provided "as is", with no warranty
of any type, including any warranty for fitness for any purpose. RTI is
under no obligation to maintain or support the software. RTI shall not be
liable for any incidental or consequential damages arising out of the use
or inability to use the software.
 
 
The QoS configuration of the DDS entities in the generated example is loaded
from this file.
 
 
This file is used only when it is in the current working directory, the
environment variable NDDS_QOS_PROFILES is defined and points to this file, or
if the filename is specified in code.
 
 
QoS profiles are a tool for pre-specifying behavior of DDS entities (e.g.
DomainParticipants, DataReaders, and DataWriters).  For Domain Participants, this
means primarily configuring platform resources (e.g. UDP buffers sizes). For
DataReaders and DataWriters, it is aligning their behavior with the required
design patterns and ensuring the QoS for DataWriters and DataReaders are
compatible at runtime.
 
 
The profiles in this file use inheritance from built-in QoS Profiles and
composition from "QoS Snippets", which are provided in this product, to create
Four example profiles.  One for a DomainParticipant, and three additional profiles
which specify DataReader and DataWriter behavior for common communication design
Patterns. 
 
 
The profiles will require optimization for networks that are highly variable in
throughput, latency, and/or sample loss, or have other unique characteristics.
Applications with special resource, latency or throughput requirements will also 
require optimized profiles.
 
 
A collection of built-in profiles can be found in the
BuiltinProfiles.documentationONLY.xml file located in the
$NDDSHOME/resource/xml/ directory.
 
 
Any of these "base" QoS configurations may be employed by referring to them by
the name shown in the BuiltinProfiles.documentationONLY.xml file.
 
 
There is also a library that contains a collection of
QoS Snippets that set some specific features or configurations. They are 
also located in the BuiltinProfiles.documentationONLY.xml file.
https://community.rti.com/best-practices/qos-profile-inheritance-and-composition-guidance
 
 
The file BuiltinProfiles.documentationONLY.xml is not intended to 
be changed by users. To modify any of the values in a built-in profile, 
the example in this file which inherits from a built-in profile should be
used, with changes applied as needed.
 
 
The "Configuring QoS with XML" chapter in the RTI Connext DDS Core Libraries User's Manual 
has more information about managing QoS using XML QoS Profiles.
-->
 
<dds xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:noNamespaceSchemaLocation="file:////rti/jenkins/workspace/connextdds_htmldocs_support_connextdds-doc_7.7.0/build/nddsgen.2.0/resource/schema/rti_dds_qos_profiles.xsd">
    <!-- A QoS library is a named set of QoS profiles.  This QoS Library contains
         the QoS profiles which may be used in the generated example
    -->
    <qos_library name="HelloWorld_Library">
 
        <!-- A QoS profile groups a set of related QoS for one or more DDS Entities -->
 
        <!-- The following design-pattern oriented profiles are intended only for DataReader and DataWriter DDS Entities -->
 
        <qos_profile name="Aperiodic_Reliable_Profile" base_name="BuiltinQosLib::Generic.Common">
            <!-- Aperiodic Reliable behavior is suitable for Command, Response, Event or Alarm use-cases that require reliable data delivery. Reliability is the protocol by which samples that are lost for any reason are retransmitted to reliable DataReaders. In this profile, samples are not retained for late joining DataReaders. Reliability begins when a DataWriter matches a DataReader at discovery.
An average DataWriter send frequency of > 500 ms is appropriate for this profile.
            -->
            <base_name>
                <!-- Makes DataReader and DataWriter reliable (BEST_EFFORT DataReaders allowed).
                     DataWriter will block up to 5 seconds if cache becomes full of unacknowledged
                     samples, after which a write() call will fail if cache is still full -->
                <element>BuiltinQosSnippetLib::QosPolicy.Reliability.Reliable</element>
                <!-- DataWriters will use all available cache for reliability retransmission,
                     DataReaders use all available cache to save samples until removed -->
                <element>BuiltinQosSnippetLib::BuiltinQosSnippetLib::QosPolicy.History.KeepAll</element>
                <!-- Optimize reliability timing for a stable, "normal" network,
                     allowing DataWriter cache to grow from default of 32 samples to 40,
                     DataReader cache to grow from 32 samples to 256. Datawriter issues heartbeats for
                     acknowledgements every 200 ms while there are unacknowledged samples
                     and will issue up to 500 periodic retries before reliability fails -->
                <element>BuiltinQosSnippetLib::Optimization.ReliabilityProtocol.Common</element>
            </base_name>
 
 
            <!-- Name Datawriters and Datareaders by use-case -->
            <datawriter_qos>
                <publication_name>
                    <name>HelloWorld_AR_DataWriter</name>
                </publication_name>
            </datawriter_qos>
            <datareader_qos>
                <subscription_name>
                    <name>HelloWorld_AR_DataReader</name>
                </subscription_name>
            </datareader_qos>
            <domain_participant_qos>
                <!--
                The participant name, if it is set, will be displayed in the
                RTI tools, making it easier for you to tell one
                application from another when you're debugging.
                -->
                <participant_name>
                    <name>HelloWorld_AR_Participant</name>
                    <role_name>AR_HelloWorld_AR_ParticipantRole</role_name>
                </participant_name>
            </domain_participant_qos>
        </qos_profile>
 
 
        <qos_profile name="Periodic_Best_Effort_Profile" base_name="BuiltinQosLib::Generic.Common">
            <!-- Periodic BEST_EFFORT behavior is suitable for samples that update frequently enough that recovery
                 of lost or out of order samples is not required. -->
            <!-- Datawriters will not allow Reliable DataReaders. DataWriter only needs history of 1 since
                 there are no retransmissions with BEST_EFFORT -->
            <base_name>
                <element>BuiltinQosSnippetLib::QosPolicy.Reliability.BestEffort</element>
                <element>BuiltinQosSnippetLib::QosPolicy.History.KeepLast_1</element>
            </base_name>
 
 
            <!-- Name Datawriters and Datareaders by use-case -->
            <datawriter_qos>
                <publication_name>
                    <name>HelloWorld_PBE_DataWriter</name>
                </publication_name>
            </datawriter_qos>
            <datareader_qos>
                <subscription_name>
                    <name>HelloWorld_PBE_DataReader</name>
                </subscription_name>
                <!-- Increase DataReader cache to 100 samples per instance for incoming data to prevent overwriting -->
                <history>
                    <kind>KEEP_LAST_HISTORY_QOS</kind>
                    <depth>100</depth>
                </history>
            </datareader_qos>
            <domain_participant_qos>
                <!--
                The participant name, if it is set, will be displayed in the
                RTI tools, making it easier for you to tell one
                application from another when you're debugging.
                -->
                <participant_name>
                    <name>HelloWorld_PBE_Participant</name>
                    <role_name>HelloWorld_PBE_ParticipantRole</role_name>
                </participant_name>
            </domain_participant_qos>
        </qos_profile>
 
 
        <qos_profile name="Last_Value_Profile" base_name="Aperiodic_Reliable_Profile">
            <!-- Last Value behavior is suitable for state data updates or resolving race conditions at discovery.  
               This profile extends the Aperiodic_Reliable_Profile to provide the last update of each instance to late joining DataReaders when it is discovered,
               without having to query the DataWriter   -->
 
 
            <!-- DataWriter retains samples (per instance) for late joining DataReaders,
                 DataReaders request last values upon discovery of DataWriters -->
            <base_name>
                <element>BuiltinQosSnippetLib::QosPolicy.Durability.TransientLocal</element>
            </base_name>
 
 
            <!-- Name Datawriters and Datareaders by use-case -->
            <datawriter_qos>
                <publication_name>
                    <name>HelloWorld_LV_DataWriter</name>
                </publication_name>
                <!-- Have DataWriters keep only the most recently updated sample for delivery
                     to late joining DataReaders, still use all available cache for reliability -->
                <durability>
                    <writer_depth>1</writer_depth>
                </durability>
            </datawriter_qos>
 
 
            <datareader_qos>
                <subscription_name>
                    <name>HelloWorld_LV_DataReader</name>
                </subscription_name>
                <!-- Keep up to 256 samples per instance for incoming samples before overwriting the oldest -->
                <history>
                    <kind>KEEP_LAST_HISTORY_QOS</kind>
                    <depth>256</depth>
                </history>
            </datareader_qos>
        </qos_profile>
       
        <!-- DDS DomainParticipant QoS specifies discovery behavior and platform resource
             values. These settings are appropriate for Windows or Linux server platforms and
             are intended only for DDS DomainParticipant entities -->
        <qos_profile name="Participant_Profile" base_name="BuiltinQosLib::Generic.Common">
            <domain_participant_qos>
                <!-- To have a unique name for each DomainParticipant, the name may be set in code for the DomainParticipant QoS, or in an environment variable referenced in the name tag.  --> 
                <participant_name>
                    <name>HelloWorld_LV_Participant</name>
                </participant_name>
                <transport_builtin>
                    <udpv4>
                        <!-- For high throughput implementations or cases where there are
                             many DDS Entities to be discovered, larger UDP buffer sizes
                             are essential for discovery, latency, and/or throughput performance,
                             especially for non-RTOS operating systems like Linux or Windows.
                             UDP receive socket buffers are especially important. Linux, for
                             example, has small default values and should be configured to support buffers equal to or greater than these QoS values. Reference                           https://community.rti.com/howto/improve-rti-connext-dds-network-performance-linux for additional              details.  Windows does not usually require special configuration.
                        <send_socket_buffer_size>10485760</send_socket_buffer_size>
                        <recv_socket_buffer_size>2097152</recv_socket_buffer_size> -->
                        <!-- Connext will try to use all available networks on a given host.  
                             An allow list (white list) helps Connext use only the specified
                             interface(s) and prevents duplicate sending of data. For more information:
                             https://community.rti.com/kb/how-do-i-restrict-and-prioritize-interfaces-rti-connext-use
                        <allow_interfaces_list>  
                            <element>192.168.0.*</element>
                        </allow_interfaces_list> 
                        -->
                    </udpv4>
                </transport_builtin>
            </domain_participant_qos>
        </qos_profile>
 
 
      <!--  This profile is provided for reference and can be uncommented
      for debug purposes. Should additional logging information be required
      or requested by RTI Support the values should be changed from the
      defaults specified here 
 
 
        <qos_profile name="Domain_Participant_Factory_Qos">
            <participant_factory_qos is_default_participant_factory_qos="true">
                <logging>
            <category>ALL</category>
            <output_file/>
            <print_format>DEFAULT</print_format>
            <verbosity>ERROR</verbosity>
        </logging>
           </participant_factory_qos>
        </qos_profile>
-->
 
 
    </qos_library>
</dds>
 
 
 

The example Creating a Requester with optional parameters shows how to create a com.rti.connext.requestreply.Requester<TReq,TRep> using this profile.

See also

Creating a Requester with optional parameters

Configuring QoS Profiles with XML