[21232] Fix flaky latency tests on mac

src/cpp/rtps/writer/ReaderLocator.cpp

@@ -207,7 +207,7 @@ bool ReaderLocator::send(
     return true;
 }
 
-RTPSReader* ReaderLocator::local_reader()
+BaseReader* ReaderLocator::local_reader()
 {
     if (!local_reader_)
     {

src/cpp/rtps/writer/ReaderLocator.hpp

@@ -31,7 +31,7 @@ namespace rtps {
 
 class RTPSParticipantImpl;
 class RTPSWriter;
-class RTPSReader;
+class BaseReader;
 class IDataSharingNotifier;
 
 /**
@@ -67,10 +67,10 @@ class ReaderLocator : public RTPSMessageSenderInterface
         return is_local_reader_;
     }
 
-    RTPSReader* local_reader();
+    BaseReader* local_reader();
 
     void local_reader(
-            RTPSReader* local_reader)
+            BaseReader* local_reader)
     {
         local_reader_ = local_reader;
     }
@@ -260,7 +260,7 @@ class ReaderLocator : public RTPSMessageSenderInterface
     LocatorSelectorEntry async_locator_info_;
     bool expects_inline_qos_;
     bool is_local_reader_;
-    RTPSReader* local_reader_;
+    BaseReader* local_reader_;
     std::vector<GuidPrefix_t> guid_prefix_as_vector_;
     std::vector<GUID_t> guid_as_vector_;
     IDataSharingNotifier* datasharing_notifier_;

src/cpp/rtps/writer/ReaderProxy.hpp

@@ -40,6 +40,7 @@ namespace eprosima {
 namespace fastdds {
 namespace rtps {
 
+class BaseReader;
 class StatefulWriter;
 class TimedEvent;
 class RTPSReader;
@@ -289,7 +290,7 @@ class ReaderProxy
      * Get the local reader on the same process (if any).
      * @return The local reader on the same process.
      */
-    inline RTPSReader* local_reader()
+    inline BaseReader* local_reader()
     {
         return locator_info_.local_reader();
     }

src/cpp/rtps/writer/StatefulWriter.cpp

@@ -473,14 +473,14 @@ bool StatefulWriter::intraprocess_delivery(
         CacheChange_t* change,
         ReaderProxy* reader_proxy)
 {
-    RTPSReader* reader = reader_proxy->local_reader();
+    BaseReader* reader = reader_proxy->local_reader();
     if (reader)
     {
         if (change->write_params.related_sample_identity() != SampleIdentity::unknown())
         {
             change->write_params.sample_identity(change->write_params.related_sample_identity());
         }
-        return BaseReader::downcast(reader)->process_data_msg(change);
+        return reader->process_data_msg(change);
     }
     return false;
 }

test/mock/rtps/ReaderLocator/rtps/writer/ReaderLocator.hpp

@@ -33,7 +33,7 @@ namespace rtps {
 
 class RTPSParticipantImpl;
 class RTPSWriter;
-class RTPSReader;
+class BaseReader;
 class IDataSharingNotifier;
 
 /**
@@ -202,7 +202,7 @@ class ReaderLocator : public RTPSMessageSenderInterface
         return false;
     }
 
-    RTPSReader* local_reader()
+    BaseReader* local_reader()
     {
         return nullptr;
     }

test/performance/latency/LatencyTestPublisher.cpp

@@ -57,29 +57,6 @@ LatencyTestPublisher::LatencyTestPublisher()
 
 LatencyTestPublisher::~LatencyTestPublisher()
 {
-    // Static type endpoints shpuld have been removed for each payload iteration
-    if (dynamic_types_)
-    {
-        destroy_data_endpoints();
-    }
-    else if (nullptr != data_writer_
-            || nullptr != data_reader_
-            || nullptr != latency_data_pub_topic_
-            || nullptr != latency_data_sub_topic_
-            || !latency_data_type_)
-    {
-        EPROSIMA_LOG_ERROR(LATENCYPUBLISHER, "ERROR unregistering the DATA type and/or removing the endpoints");
-    }
-
-    subscriber_->delete_datareader(command_reader_);
-    participant_->delete_subscriber(subscriber_);
-
-    publisher_->delete_datawriter(command_writer_);
-    participant_->delete_publisher(publisher_);
-
-    participant_->delete_topic(latency_command_sub_topic_);
-    participant_->delete_topic(latency_command_pub_topic_);
-
     std::string TestCommandType("TestCommandType");
     participant_->unregister_type(TestCommandType);
 
@@ -682,6 +659,32 @@ void LatencyTestPublisher::run()
     }
 }
 
+void LatencyTestPublisher::destroy_user_entities()
+{
+    // Static type endpoints shpuld have been removed for each payload iteration
+    if (dynamic_types_)
+    {
+        destroy_data_endpoints();
+    }
+    else if (nullptr != data_writer_
+            || nullptr != data_reader_
+            || nullptr != latency_data_pub_topic_
+            || nullptr != latency_data_sub_topic_
+            || !latency_data_type_)
+    {
+        EPROSIMA_LOG_ERROR(LATENCYPUBLISHER, "ERROR unregistering the DATA type and/or removing the endpoints");
+    }
+
+    subscriber_->delete_datareader(command_reader_);
+    participant_->delete_subscriber(subscriber_);
+
+    publisher_->delete_datawriter(command_writer_);
+    participant_->delete_publisher(publisher_);
+
+    participant_->delete_topic(latency_command_sub_topic_);
+    participant_->delete_topic(latency_command_pub_topic_);
+}
+
 void LatencyTestPublisher::export_csv(
         const std::string& data_name,
         const std::string& str_reliable,

test/performance/latency/LatencyTestPublisher.hpp

@@ -100,6 +100,8 @@ class LatencyTestPublisher
 
     void run();
 
+    void destroy_user_entities();
+
 private:
 
     bool init_dynamic_types();

test/performance/latency/LatencyTestSubscriber.cpp

@@ -51,29 +51,6 @@ LatencyTestSubscriber::LatencyTestSubscriber()
 
 LatencyTestSubscriber::~LatencyTestSubscriber()
 {
-    // Static type endpoints should have been remove for each payload iteration
-    if (dynamic_types_)
-    {
-        destroy_data_endpoints();
-    }
-    else if (nullptr != data_writer_
-            || nullptr != data_reader_
-            || nullptr != latency_data_pub_topic_
-            || nullptr != latency_data_sub_topic_
-            || !latency_data_type_)
-    {
-        EPROSIMA_LOG_ERROR(LATENCYSUBSCRIBER, "ERROR unregistering the DATA type and/or removing the endpoints");
-    }
-
-    subscriber_->delete_datareader(command_reader_);
-    participant_->delete_subscriber(subscriber_);
-
-    publisher_->delete_datawriter(command_writer_);
-    participant_->delete_publisher(publisher_);
-
-    participant_->delete_topic(latency_command_sub_topic_);
-    participant_->delete_topic(latency_command_pub_topic_);
-
     std::string TestCommandType("TestCommandType");
     participant_->unregister_type(TestCommandType);
 
@@ -641,6 +618,32 @@ void LatencyTestSubscriber::run()
     }
 }
 
+void LatencyTestSubscriber::destroy_user_entities()
+{
+    // Static type endpoints should have been remove for each payload iteration
+    if (dynamic_types_)
+    {
+        destroy_data_endpoints();
+    }
+    else if (nullptr != data_writer_
+            || nullptr != data_reader_
+            || nullptr != latency_data_pub_topic_
+            || nullptr != latency_data_sub_topic_
+            || !latency_data_type_)
+    {
+        EPROSIMA_LOG_ERROR(LATENCYSUBSCRIBER, "ERROR unregistering the DATA type and/or removing the endpoints");
+    }
+
+    subscriber_->delete_datareader(command_reader_);
+    participant_->delete_subscriber(subscriber_);
+
+    publisher_->delete_datawriter(command_writer_);
+    participant_->delete_publisher(publisher_);
+
+    participant_->delete_topic(latency_command_sub_topic_);
+    participant_->delete_topic(latency_command_pub_topic_);
+}
+
 bool LatencyTestSubscriber::test(
         uint32_t datasize)
 {

test/performance/latency/LatencyTestSubscriber.hpp

@@ -60,6 +60,8 @@ class LatencyTestSubscriber
 
     void run();
 
+    void destroy_user_entities();
+
     bool test(
             uint32_t datasize);
 

test/performance/latency/main_LatencyTest.cpp

@@ -503,6 +503,7 @@ int main(
                 dynamic_types, data_sharing, data_loans, shared_memory, forced_domain, data_sizes))
         {
             latency_publisher.run();
+            latency_publisher.destroy_user_entities();
         }
         else
         {
@@ -518,6 +519,7 @@ int main(
                 xml_config_file, dynamic_types, data_sharing, data_loans, shared_memory, forced_domain, data_sizes))
         {
             latency_subscriber.run();
+            latency_subscriber.destroy_user_entities();
         }
         else
         {
@@ -568,6 +570,13 @@ int main(
             {
                 sub.join();
             }
+
+            for (auto& sub : latency_subscribers)
+            {
+                sub->destroy_user_entities();
+            }
+
+            latency_publisher.destroy_user_entities();
         }
         else
         {