Update to use distributed condition

sonndinh · sonndinh · commit ed6220639c9c · 2025-05-08T11:59:56.000-05:00
diff --git a/tests/DCPS/RecorderReplayer/Args.h b/tests/DCPS/RecorderReplayer/Args.h
@@ -1,6 +1,4 @@
 /*
- *
- *
  * Distributed under the OpenDDS License.
  * See: http://www.opendds.org/license.html
  */
@@ -9,7 +7,6 @@
 #define MESSENGER_TEST_ARGS_H
 
 #include <dds/DCPS/transport/framework/TransportRegistry.h>
-
 #include <dds/DCPS/transport/framework/TransportConfig.h>
 #include <dds/DCPS/transport/framework/TransportInst.h>
 
@@ -18,6 +15,14 @@
 #include <ace/Log_Msg.h>
 #include <ace/OS_NS_stdlib.h>
 
+const char* ACTOR_PUBLISHER = "Publisher";
+const char* ACTOR_RECORDER = "Recorder";
+const char* ACTOR_REPLAYER = "Replayer";
+const char* ACTOR_SUBSCRIBER = "Subscriber";
+
+const char* EVENT_RECORDER_JOINED = "RecorderJoined";
+const char* EVENT_SUBSCRIBER_JOINED = "SubscriberJoined";
+
 inline int
 parse_args(int argc, ACE_TCHAR *argv[])
 {
@@ -29,18 +34,13 @@ parse_args(int argc, ACE_TCHAR *argv[])
   while ((c = get_opts()) != -1) {
     switch (c) {
     case 't':
-
       if (ACE_OS::strcmp(get_opts.opt_arg(), ACE_TEXT("udp")) == 0) {
         transport_type = "udp";
-
       } else if (ACE_OS::strcmp(get_opts.opt_arg(), ACE_TEXT("multicast")) == 0) {
         transport_type = "multicast";
-
       } else if (ACE_OS::strcmp(get_opts.opt_arg(), ACE_TEXT("tcp")) == 0) {
         transport_type = "tcp";
-
       }
-
       break;
     case 'p':
       thread_per_connection = true;
@@ -49,7 +49,7 @@ parse_args(int argc, ACE_TCHAR *argv[])
     default:
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("usage: %s [-t transport]\n"), argv[0]),
-                       -1);
+                        -1);
     }
   }
 
@@ -61,19 +61,16 @@ parse_args(int argc, ACE_TCHAR *argv[])
   }
 
   if (thread_per_connection) {
-    OpenDDS::DCPS::TransportConfig_rch config =
-      TheTransportRegistry->fix_empty_default();
+    OpenDDS::DCPS::TransportConfig_rch config = TheTransportRegistry->fix_empty_default();
     if (config.in() == 0) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("no default config\n"), argv[0]),
-                       -1);
-    }
-    else if (config->instances_.size() < 1) {
+                        -1);
+    } else if (config->instances_.size() < 1) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("no instances on default config\n"), argv[0]),
-                       -1);
-    }
-    else if (config->instances_.size() > 1) {
+                        -1);
+    } else if (config->instances_.size() > 1) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("too many instances on default config, using first\n"), argv[0]));
     }
@@ -84,5 +81,4 @@ parse_args(int argc, ACE_TCHAR *argv[])
   return 0;
 }
 
-
 #endif
diff --git a/tests/DCPS/RecorderReplayer/Publisher.cpp b/tests/DCPS/RecorderReplayer/Publisher.cpp
@@ -1,186 +1,149 @@
 /*
- *
- *
  * Distributed under the OpenDDS License.
  * See: http://www.opendds.org/license.html
  */
 
-#include <ace/Get_Opt.h>
-#include <ace/Log_Msg.h>
-#include <ace/OS_NS_stdlib.h>
+#include "MessengerTypeSupportImpl.h"
+#include "Args.h"
+
+#include <tests/Utils/DistributedConditionSet.h>
 
 #include <dds/DCPS/Marked_Default_Qos.h>
 #include <dds/DCPS/PublisherImpl.h>
 #include <dds/DCPS/Service_Participant.h>
 #include <dds/DCPS/WaitSet.h>
-
-#include "dds/DCPS/StaticIncludes.h"
+#include <dds/DCPS/StaticIncludes.h>
 #if defined ACE_AS_STATIC_LIBS && !defined OPENDDS_SAFETY_PROFILE
-#include <dds/DCPS/transport/udp/Udp.h>
-#include <dds/DCPS/transport/multicast/Multicast.h>
-#include <dds/DCPS/RTPS/RtpsDiscovery.h>
-#include <dds/DCPS/transport/rtps_udp/RtpsUdp.h>
-#include <dds/DCPS/transport/shmem/Shmem.h>
+# include <dds/DCPS/transport/udp/Udp.h>
+# include <dds/DCPS/transport/multicast/Multicast.h>
+# include <dds/DCPS/RTPS/RtpsDiscovery.h>
+# include <dds/DCPS/transport/rtps_udp/RtpsUdp.h>
+# include <dds/DCPS/transport/shmem/Shmem.h>
 #endif
 
-#include "MessengerTypeSupportImpl.h"
-#include "Args.h"
+#include <ace/Get_Opt.h>
+#include <ace/Log_Msg.h>
+#include <ace/OS_NS_stdlib.h>
 
 int ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 {
- try {
-     // Initialize DomainParticipantFactory
-     DDS::DomainParticipantFactory_var dpf =
-       TheParticipantFactoryWithArgs(argc, argv);
+  try {
+    // Initialize DomainParticipantFactory
+    DDS::DomainParticipantFactory_var dpf = TheParticipantFactoryWithArgs(argc, argv);
 
-     int error;
-     if ((error = parse_args(argc, argv)) != 0) {
-       return error;
-     }
+    int error;
+    if ((error = parse_args(argc, argv)) != 0) {
+      return error;
+    }
 
     // Create DomainParticipant
-    DDS::DomainParticipant_var participant =
-      dpf->create_participant(4,
-                              PARTICIPANT_QOS_DEFAULT,
-                              0,
-                              OpenDDS::DCPS::DEFAULT_STATUS_MASK);
+    DDS::DomainParticipant_var participant = dpf->create_participant(4,
+                                                                     PARTICIPANT_QOS_DEFAULT,
+                                                                     0,
+                                                                     OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
     if (!participant) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("ERROR: %N:%l: main() -")
                         ACE_TEXT(" create_participant failed!\n")),
-                       -1);
+                        -1);
     }
 
     ACE_DEBUG((LM_DEBUG, "(%P|%t) Start publisher\n"));
 
-    {
-      // Register TypeSupport (Messenger::Message)
-      Messenger::MessageTypeSupport_var ts =
-        new Messenger::MessageTypeSupportImpl;
+    // Register TypeSupport (Messenger::Message)
+    Messenger::MessageTypeSupport_var ts = new Messenger::MessageTypeSupportImpl;
 
-      if (ts->register_type(participant, "Messenger") != DDS::RETCODE_OK) {
-        ACE_ERROR_RETURN((LM_ERROR,
-                          ACE_TEXT("ERROR: %N:%l: main() -")
-                          ACE_TEXT(" register_type failed!\n")),
-                         -1);
-      }
+    if (ts->register_type(participant, "Messenger") != DDS::RETCODE_OK) {
+      ACE_ERROR_RETURN((LM_ERROR,
+                        ACE_TEXT("ERROR: %N:%l: main() -")
+                        ACE_TEXT(" register_type failed!\n")),
+                        -1);
+    }
 
-      // Create Topic (Movie Discussion List)
-      DDS::Topic_var topic =
-        participant->create_topic("Movie Discussion List",
-                                  "Messenger",
-                                  TOPIC_QOS_DEFAULT,
-                                  0,
-                                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
-
-      if (!topic) {
-        ACE_ERROR_RETURN((LM_ERROR,
-                          ACE_TEXT("ERROR: %N:%l: main() -")
-                          ACE_TEXT(" create_topic failed!\n")),
-                         -1);
-      }
+    // Create Topic (Movie Discussion List)
+    DDS::Topic_var topic = participant->create_topic("Movie Discussion List",
+                                                     "Messenger",
+                                                     TOPIC_QOS_DEFAULT,
+                                                     0,
+                                                     OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
-      // setup partition
-      DDS::PublisherQos pub_qos;
-      participant->get_default_publisher_qos(pub_qos);
-
-      DDS::StringSeq my_partition;
-      my_partition.length(1);
-      my_partition[0] = "One";
-      pub_qos.partition.name = my_partition;
-
-      // Create Publisher
-      DDS::Publisher_var publisher =
-        participant->create_publisher(pub_qos,
-                                      0,
-                                      OpenDDS::DCPS::DEFAULT_STATUS_MASK);
-
-      if (!publisher) {
-        ACE_ERROR_RETURN((LM_ERROR,
-                          ACE_TEXT("ERROR: %N:%l: main() -")
-                          ACE_TEXT(" create_publisher failed!\n")),
-                         -1);
-      }
+    if (!topic) {
+      ACE_ERROR_RETURN((LM_ERROR,
+                        ACE_TEXT("ERROR: %N:%l: main() -")
+                        ACE_TEXT(" create_topic failed!\n")),
+                        -1);
+    }
 
-      // Create DataWriter
-      DDS::DataWriter_var writer =
-        publisher->create_datawriter(topic,
-                                     DATAWRITER_QOS_DEFAULT,
-                                     0,
-                                     OpenDDS::DCPS::DEFAULT_STATUS_MASK);
-
-      if (!writer) {
-        ACE_ERROR_RETURN((LM_ERROR,
-                          ACE_TEXT("ERROR: %N:%l: main() -")
-                          ACE_TEXT(" create_datawriter failed!\n")),
-                         -1);
-      }
+    // Setup partition
+    DDS::PublisherQos pub_qos;
+    participant->get_default_publisher_qos(pub_qos);
 
-      Messenger::MessageDataWriter_var message_writer =
-        Messenger::MessageDataWriter::_narrow(writer);
+    DDS::StringSeq my_partition;
+    my_partition.length(1);
+    my_partition[0] = "One";
+    pub_qos.partition.name = my_partition;
 
-      if (!message_writer) {
-        ACE_ERROR_RETURN((LM_ERROR,
-                          ACE_TEXT("ERROR: %N:%l: main() -")
-                          ACE_TEXT(" _narrow failed!\n")),
-                         -1);
-      }
+    // Create Publisher
+    DDS::Publisher_var publisher = participant->create_publisher(pub_qos,
+                                                                 0,
+                                                                 OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
-      // Block until Subscriber is available
-      DDS::StatusCondition_var condition = writer->get_statuscondition();
-      condition->set_enabled_statuses(DDS::PUBLICATION_MATCHED_STATUS);
-
-      DDS::WaitSet_var ws = new DDS::WaitSet;
-      ws->attach_condition(condition);
-
-      while (true) {
-        DDS::PublicationMatchedStatus matches;
-        if (writer->get_publication_matched_status(matches) != ::DDS::RETCODE_OK) {
-          ACE_ERROR_RETURN((LM_ERROR,
-                            ACE_TEXT("ERROR: %N:%l: main() -")
-                            ACE_TEXT(" get_publication_matched_status failed!\n")),
-                           -1);
-        }
-
-        if (matches.current_count >= 1) {
-          break;
-        }
-
-        DDS::ConditionSeq conditions;
-        DDS::Duration_t timeout = { 60, 0 };
-        if (ws->wait(conditions, timeout) != DDS::RETCODE_OK) {
-          ACE_ERROR_RETURN((LM_ERROR,
-                            ACE_TEXT("ERROR: %N:%l: main() -")
-                            ACE_TEXT(" wait failed!\n")),
-                           -1);
-        }
-      }
+    if (!publisher) {
+      ACE_ERROR_RETURN((LM_ERROR,
+                        ACE_TEXT("ERROR: %N:%l: main() -")
+                        ACE_TEXT(" create_publisher failed!\n")),
+                        -1);
+    }
 
-      ws->detach_condition(condition);
+    // Create DataWriter
+    DDS::DataWriter_var writer = publisher->create_datawriter(topic,
+                                                              DATAWRITER_QOS_DEFAULT,
+                                                              0,
+                                                              OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
-      // Write samples
-      Messenger::Message message;
-      message.subject_id = 99;
+    if (!writer) {
+      ACE_ERROR_RETURN((LM_ERROR,
+                        ACE_TEXT("ERROR: %N:%l: main() -")
+                        ACE_TEXT(" create_datawriter failed!\n")),
+                        -1);
+    }
 
-      message.from       = "Comic Book Guy";
-      message.subject    = "Review";
-      message.text       = "Worst. Movie. Ever.";
-      message.count      = 0;
+    Messenger::MessageDataWriter_var message_writer = Messenger::MessageDataWriter::_narrow(writer);
 
-      for (int i = 0; i < 10; ++i) {
-        const DDS::ReturnCode_t error = message_writer->write(message, DDS::HANDLE_NIL);
-        ++message.count;
-        ++message.subject_id;
+    if (!message_writer) {
+      ACE_ERROR_RETURN((LM_ERROR,
+                        ACE_TEXT("ERROR: %N:%l: main() -")
+                        ACE_TEXT(" _narrow failed!\n")),
+                        -1);
+    }
 
-        if (error != DDS::RETCODE_OK) {
-          ACE_ERROR((LM_ERROR,
-                     ACE_TEXT("ERROR: %N:%l: main() -")
-                     ACE_TEXT(" write returned %d!\n"), error));
-        }
+    // Block until Recorder joins
+    DistributedConditionSet_rch dcs = OpenDDS::DCPS::make_rch<FileBasedDistributedConditionSet>();
+    dcs->wait_for(ACTOR_PUBLISHER, ACTOR_RECORDER, EVENT_RECORDER_JOINED);
+
+    // Write samples
+    Messenger::Message message;
+    message.subject_id = 99;
+    message.from       = "Comic Book Guy";
+    message.subject    = "Review";
+    message.text       = "Worst. Movie. Ever.";
+    message.count      = 0;
+
+    for (int i = 0; i < 10; ++i) {
+      const DDS::ReturnCode_t error = message_writer->write(message, DDS::HANDLE_NIL);
+      ++message.count;
+      ++message.subject_id;
+
+      if (error != DDS::RETCODE_OK) {
+        ACE_ERROR((LM_ERROR,
+                   ACE_TEXT("ERROR: %N:%l: main() -")
+                   ACE_TEXT(" write returned %d!\n"), error));
       }
     }
 
+    // TODO(sonndinh): May wait until all samples are received by the Recorder
+
     ACE_DEBUG((LM_DEBUG, "(%P|%t) Stop publisher\n"));
 
     // Clean-up!
diff --git a/tests/DCPS/RecorderReplayer/Relay.cpp b/tests/DCPS/RecorderReplayer/Relay.cpp
