Fix XML profiles in DDS-Record-Replay + filter reader partitions (#177)

* [#24061] Solved 'Could not compute key for data'

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Fixed missing dependency types

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Preserve reader partition filter on endpoint updates

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Applied revision

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Applied revision. Retry dynamic type initialization for keyed topics

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

---------

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

Author: Danipiza

ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp

@@ -463,6 +463,9 @@ class DdsPipe
     //! Allowed partitions list added in the filter.
     std::set<std::string> filter_partition_;
 
+    //! Current partitions configured in readers through \c update_partitions.
+    std::set<std::string> reader_partitions_;
+
     /**
      * @brief Internal mutex for concurrent calls
      */

ddspipe_core/src/cpp/core/DdsPipe.cpp

@@ -299,12 +299,11 @@ void DdsPipe::discovered_endpoint_nts_(
     }
     else
     {
-        // there is a filter.
-        // update the track of the topic to
-        // add the partition in the reader if it is in the filter
+        // Another endpoint from an already tracked topic/participant was discovered
+        // Refresh bridge partition metadata and reader partition QoS
 
         const auto bridge_it = bridges_.find(utils::Heritable<DdsTopic>::make_heritable(endpoint.topic));
-        // add the specific partition of the endpoint in the bridge topic.
+        // Add the specific partition of the endpoint in the bridge topic
         if (bridge_it != bridges_.end())
         {
             std::ostringstream guid_ss;
@@ -315,12 +314,14 @@ void DdsPipe::discovered_endpoint_nts_(
             {
                 bridge_it->second->add_partition_to_topic(guid_ss.str(), part_it->second);
             }
-        }
 
-        // update readers outside the lock
-        if (!filter_partition_.empty())
-        {
-            update_partitions_nts_(filter_partition_);
+            // Refresh this bridge so writers receive the updated guid->partition map
+            // Also refresh reader partitions using
+            // filter (if set), otherwise
+            // the current reader partition configuration
+            const auto& partitions_to_apply =
+                    filter_partition_.empty() ? reader_partitions_ : filter_partition_;
+            bridge_it->second->update_partitions(partitions_to_apply);
         }
     }
 }
@@ -624,6 +625,7 @@ void DdsPipe::update_partitions(
 {
     std::lock_guard<std::mutex> lock(mutex_);
 
+    reader_partitions_ = partitions_set;
     update_partitions_nts_(partitions_set);
 }
 

ddspipe_participants/include/ddspipe_participants/types/dds/TopicDataType.hpp

@@ -15,12 +15,14 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
 
 #include <fastdds/dds/core/policy/QosPolicies.hpp>
 #include <fastdds/dds/core/ReturnCode.hpp>
 #include <fastdds/dds/domain/DomainParticipantFactory.hpp>
 #include <fastdds/dds/topic/TopicDataType.hpp>
+#include <fastdds/dds/xtypes/dynamic_types/DynamicType.hpp>
 #include <fastdds/dds/xtypes/type_representation/detail/dds_xtypes_typeobject.hpp>
 #include <fastdds/rtps/common/InstanceHandle.hpp>
 #include <fastdds/rtps/common/SerializedPayload.hpp>
@@ -103,6 +105,16 @@ class TopicDataType : public fastdds::dds::TopicDataType
 
     std::shared_ptr<core::PayloadPool> payload_pool_;
 
+    //! Lazily created dynamic type used to compute key from serialized payload.
+    mutable fastdds::dds::DynamicType::_ref_type dynamic_type_{};
+    mutable std::mutex dynamic_type_mtx_;
+    //! Set to true if dynamic type initialization permanently failed due to structural errors.
+    //! Missing dependencies are treated as transient and retried.
+    mutable bool dynamic_type_init_failed_{false};
+
+    DDSPIPE_PARTICIPANTS_DllAPI
+    bool initialize_dynamic_type_() const;
+
 };
 
 } /* namespace dds */

ddspipe_participants/src/cpp/reader/dds/CommonReader.cpp

@@ -35,6 +35,25 @@ namespace dds {
 
 using namespace eprosima::ddspipe::core::types;
 
+static void apply_reader_partitions_(
+        fastdds::dds::PartitionQosPolicy& partition_qos,
+        const std::set<std::string>& partitions_filter)
+{
+    partition_qos.clear();
+
+    // Empty filter means no filter: subscribe to all partitions
+    if (partitions_filter.empty())
+    {
+        partition_qos.push_back("*");
+        return;
+    }
+
+    for (const std::string& partition : partitions_filter)
+    {
+        partition_qos.push_back(partition.c_str());
+    }
+}
+
 CommonReader::~CommonReader()
 {
     // This variables should be set, otherwise the creation should have fail
@@ -76,22 +95,11 @@ void CommonReader::init(
                                      << participant_id_ << " in topic " << topic_ << ".");
     }
 
-    // If the reader has an active filter
-    // change the qos partition before creating the datareader.
-    if (partitions_set.size() > 0)
-    {
-        // Get the current subscriber qos
-        fastdds::dds::SubscriberQos sub_qos = dds_subscriber_->get_qos();
-        // Remove all partitions from the qos
-        sub_qos.partition().clear();
-        // Add the filter partitions
-        for (const std::string& partition: partitions_set)
-        {
-            sub_qos.partition().push_back(partition.c_str());
-        }
-        // Update the subscriber qos
-        dds_subscriber_->set_qos(sub_qos);
-    }
+    // Update subscriber partitions before creating the datareader
+    // Empty filter means subscribe to all partitions ("*")
+    fastdds::dds::SubscriberQos sub_qos = dds_subscriber_->get_qos();
+    apply_reader_partitions_(sub_qos.partition(), partitions_set);
+    dds_subscriber_->set_qos(sub_qos);
 
     auto topic_tmp = dds_participant_->find_topic(topic_.topic_name(), 10);
 
@@ -263,10 +271,6 @@ void CommonReader::enable_nts_() noexcept
 fastdds::dds::SubscriberQos CommonReader::reckon_subscriber_qos_() const
 {
     fastdds::dds::SubscriberQos qos = dds_participant_->get_default_subscriber_qos();
-    if (topic_.topic_qos.has_partitions())
-    {
-        qos.partition().push_back("*");
-    }
     qos.entity_factory().autoenable_created_entities = false;
     return qos;
 }
@@ -393,12 +397,7 @@ void CommonReader::update_partitions(
         const std::set<std::string>& partitions_set)
 {
     fastdds::dds::SubscriberQos sub_qos = dds_subscriber_->get_qos();
-    sub_qos.partition().clear();
-    for (const std::string& partition: partitions_set)
-    {
-        sub_qos.partition().push_back(partition.c_str());
-    }
-
+    apply_reader_partitions_(sub_qos.partition(), partitions_set);
     dds_subscriber_->set_qos(sub_qos);
 }
 

ddspipe_participants/src/cpp/reader/rtps/CommonReader.cpp

@@ -37,6 +37,25 @@ namespace rtps {
 using namespace eprosima::ddspipe::core::types;
 using eprosima::ddspipe::core::types::operator <<;
 
+static void apply_reader_partitions_(
+        fastdds::dds::PartitionQosPolicy& partition_qos,
+        const std::set<std::string>& partitions_filter)
+{
+    partition_qos.clear();
+
+    // Empty filter means no filter: subscribe to all partitions
+    if (partitions_filter.empty())
+    {
+        partition_qos.push_back("*");
+        return;
+    }
+
+    for (const std::string& partition : partitions_filter)
+    {
+        partition_qos.push_back(partition.c_str());
+    }
+}
+
 CommonReader::CommonReader(
         const ParticipantId& participant_id,
         const DdsTopic& topic,
@@ -125,19 +144,9 @@ void CommonReader::internal_entities_creation_(
                                      << participant_id_ << " in topic " << topic_ << ".");
     }
 
-    if (partitions_set.size() > 0)
-    {
-        auto& sub_part_qos = reader_qos_.m_partition;
-        // Clear the partitions
-        sub_part_qos.clear();
-        // Add the partitions from the filter
-        for (const auto& partition : partitions_set)
-        {
-            sub_part_qos.push_back(partition.c_str());
-        }
-
-        // no content topic filter
-    }
+    // Update reader partitions before registration
+    // Empty filter means subscribe to all partitions ("*")
+    apply_reader_partitions_(reader_qos_.m_partition, partitions_set);
 
     // Register reader with topic
     if (!rtps_participant_->register_reader(rtps_reader_, topic_description, reader_qos_))
@@ -177,15 +186,7 @@ core::types::DdsTopic CommonReader::topic() const noexcept
 void CommonReader::update_partitions(
         const std::set<std::string>& partitions_set)
 {
-    // Get the partitions from the reader qos
-    auto& sub_part_qos = reader_qos_.m_partition;
-    // Clear the partitions
-    sub_part_qos.clear();
-    // Add the partitions from the filter
-    for (const auto& partition : partitions_set)
-    {
-        sub_part_qos.push_back(partition.c_str());
-    }
+    apply_reader_partitions_(reader_qos_.m_partition, partitions_set);
     // Update the reader with the new partitions
     rtps_participant_->update_reader(rtps_reader_, reader_qos_);
 }