Merge pull request OpenDDS#5092 from mitza-oci/relay-control

Updated RtpsRelayControl and its supporting code in RtpsRelay

Author: jrw972

tools/rtpsrelay/RelayConfigControlListener.cpp

@@ -19,7 +19,7 @@ void RelayConfigControlListener::on_data_available(DDS::DataReader_ptr reader)
   RelayConfig control;
   DDS::SampleInfo info;
   while (control_reader->take_next_sample(control, info) == DDS::RETCODE_OK) {
-    if (info.valid_data) {
+    if (info.valid_data && info.instance_state == DDS::ALIVE_INSTANCE_STATE) {
       for (const auto& p : control.config()) {
         TheServiceParticipant->config_store()->set(p.first, p.second);
       }

tools/rtpsrelay/RtpsRelay.cpp

@@ -978,14 +978,24 @@ int run(int argc, ACE_TCHAR* argv[])
     return EXIT_FAILURE;
   }
 
+  DDS::SubscriberQos subscriber_partitioned_qos;
+  relay_participant->get_default_subscriber_qos(subscriber_partitioned_qos);
+  subscriber_partitioned_qos.partition.name = publisher_qos.partition.name;
+  DDS::Subscriber_var partition_subscriber = relay_participant->create_subscriber(subscriber_partitioned_qos, nullptr,
+                                                                                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
+  if (!partition_subscriber) {
+    ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: failed to create Relay partition subscriber\n"));
+    return EXIT_FAILURE;
+  }
+
   DDS::DataReaderQos relay_config_control_qos;
-  relay_subscriber->get_default_datareader_qos(relay_config_control_qos);
+  partition_subscriber->get_default_datareader_qos(relay_config_control_qos);
   relay_config_control_qos.reliability.kind = DDS::RELIABLE_RELIABILITY_QOS;
 
   DDS::DataReaderListener_var relay_config_control_listener(new RelayConfigControlListener);
   DDS::DataReader_var relay_config_control_reader =
-    relay_subscriber->create_datareader(cft_config_ctrl, relay_config_control_qos,
-                                        relay_config_control_listener, DDS::DATA_AVAILABLE_STATUS);
+    partition_subscriber->create_datareader(cft_config_ctrl, relay_config_control_qos,
+                                            relay_config_control_listener, DDS::DATA_AVAILABLE_STATUS);
   if (!relay_config_control_reader) {
     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: failed to create %C data reader\n", RELAY_CONFIG_CONTROL_TOPIC_NAME.c_str()));
     return EXIT_FAILURE;

tools/rtpsrelay/control/RtpsRelayControl.cpp

@@ -65,7 +65,8 @@ struct ShutdownHandler : ACE_Event_Handler {
 
 int write_if_ready(const RelayConfig& config,
                    RelayConfigDataWriter_var& relay_config_data_writer,
-                   const DDS::WaitSet_var& waiter)
+                   const DDS::WaitSet_var& waiter,
+                   bool& write_pending)
 {
   DDS::PublicationMatchedStatus matches{};
   if (relay_config_data_writer->get_publication_matched_status(matches) != ::DDS::RETCODE_OK) {
@@ -94,9 +95,7 @@ int write_if_ready(const RelayConfig& config,
         waiter->detach_condition(status_cond);
       }
     }
-    DDS::Publisher_var pub{relay_config_data_writer->get_publisher()};
-    pub->delete_datawriter(relay_config_data_writer);
-    relay_config_data_writer = nullptr;
+    write_pending = false;
   }
   return EXIT_SUCCESS;
 }
@@ -217,6 +216,7 @@ int run(int argc, ACE_TCHAR* argv[])
 
   DDS::WaitSet_var waiter = new DDS::WaitSet;
 
+  bool write_pending{false};
   RelayConfigDataWriter_var relay_config_data_writer;
   if (!config.config().empty()) {
     if (config.relay_id().empty()) {
@@ -252,6 +252,7 @@ int run(int argc, ACE_TCHAR* argv[])
     DDS::StatusCondition_var cond = relay_config_data_writer->get_statuscondition();
     cond->set_enabled_statuses(DDS::PUBLICATION_MATCHED_STATUS);
     waiter->attach_condition(cond);
+    write_pending = true;
   }
 
   DDS::Subscriber_var subscriber{participant->create_subscriber(subscriber_qos, nullptr, 0)};
@@ -292,8 +293,8 @@ int run(int argc, ACE_TCHAR* argv[])
   waiter->attach_condition(shutdown_guard);
 
   for (auto done{false}; !done;) {
-    if (relay_config_data_writer) {
-      if (write_if_ready(config, relay_config_data_writer, waiter) != EXIT_SUCCESS) {
+    if (relay_config_data_writer && write_pending) {
+      if (write_if_ready(config, relay_config_data_writer, waiter, write_pending) != EXIT_SUCCESS) {
         return EXIT_FAILURE;
       }
     }