[#24061] Updated code for filter

Danipiza · Danipiza · commit d0935daa34e3 · 2026-01-20T08:10:52.000+01:00
Signed-off-by: danipiza &lt;dpizarrogallego@gmail.com&gt;
diff --git a/ddspipe_core/include/ddspipe_core/communication/dds/DdsBridge.hpp b/ddspipe_core/include/ddspipe_core/communication/dds/DdsBridge.hpp
@@ -113,6 +113,11 @@ class DdsBridge : public Bridge
     void remove_writer(
             const types::ParticipantId& participant_id) noexcept;
 
+    DDSPIPE_CORE_DllAPI
+    void add_partition_to_topic(
+            std::string guid,
+            std::string partition);
+
     /**
      * TODO
      */
diff --git a/ddspipe_core/include/ddspipe_core/communication/dds/Track.hpp b/ddspipe_core/include/ddspipe_core/communication/dds/Track.hpp
@@ -251,7 +251,7 @@ class Track
     /**
      * Boolean use to update fastddsspy filter
      */
-    std::atomic<bool> is_fastddsspy;
+    std::atomic<bool> has_filter_;
 
     /**
      * Mutex to guard while the Track is sending a message so it could not be disabled.
diff --git a/ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp b/ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp
@@ -98,6 +98,20 @@ class DdsPipe
     utils::ReturnCode reload_configuration(
             const DdsPipeConfiguration& new_configuration);
 
+     /** TODO. danip
+     * @brief Reload the filter partitions set.
+     *
+     * @param [in] new_filter_partition_set : new filter partition set.
+     *
+     * @return \c RETCODE_OK if the new std::set has been updated correctly.
+     * @return \c RETCODE_ERROR if any other error has occurred.
+     *
+     * @note This method calls \c update_readers_track() of all Topic bridges, to update the DataReaders.
+     */
+//     DDSPIPE_CORE_DllAPI
+//     void reload_filter_partition(
+//             const std::set<std::string> new_filter_partition_set);
+
     /////////////////////////
     // ENABLING METHODS
     /////////////////////////
@@ -128,7 +142,7 @@ class DdsPipe
 
     DDSPIPE_CORE_DllAPI
     void update_partitions(
-        std::set<std::string> partitions_set);
+            std::set<std::string> partitions_set);
 
     DDSPIPE_CORE_DllAPI
     void update_content_filter(
diff --git a/ddspipe_core/src/cpp/communication/dds/DdsBridge.cpp b/ddspipe_core/src/cpp/communication/dds/DdsBridge.cpp
@@ -353,6 +353,13 @@ std::ostream& operator <<(
     return os;
 }
 
+void DdsBridge::add_partition_to_topic(
+        std::string guid,
+        std::string partition)
+{
+    topic_->partition_name[guid] = partition;
+}
+
 } /* namespace core */
 } /* namespace ddspipe */
 } /* namespace eprosima */
diff --git a/ddspipe_core/src/cpp/communication/dds/Track.cpp b/ddspipe_core/src/cpp/communication/dds/Track.cpp
@@ -63,7 +63,7 @@ Track::Track(
 
     // Initialized to false, only fastddspy
     // will change this value
-    is_fastddsspy = false;
+    has_filter_ = false;
 
     logDebug(DDSPIPE_TRACK, "Track " << *this << " created.");
 }
@@ -110,7 +110,7 @@ void Track::enable() noexcept
             writer_it.second->enable();
         }
 
-        if(is_fastddsspy)
+        if(has_filter_)
         {
             reader_->update_content_topic_filter("");
             reader_->update_partitions(std::set<std::string>());
diff --git a/ddspipe_core/src/cpp/core/DdsPipe.cpp b/ddspipe_core/src/cpp/core/DdsPipe.cpp
@@ -131,6 +131,30 @@ utils::ReturnCode DdsPipe::reload_configuration(
     return reload_allowed_topics_(allowed_topics);
 }
 
+// void DdsPipe::reload_filter_partition(
+//         const std::set<std::string> filter_partition_set)
+// {
+//     update_filter(filter_partition_set);
+
+//     std::vector<DdsBridge*> targets;
+//     {
+//         std::lock_guard<std::mutex> lock(bridges_mutex_);
+
+//         for (const auto& bridge : bridges_)
+//         {
+//             if (bridge.second)
+//             {
+//                 targets.push_back(bridge.second.get());
+//             }
+//         }
+//     }
+
+//     for (auto* target : targets)
+//     {
+//         target->update_readers_track(filter_partition_set);
+//     }
+// }
+
 utils::ReturnCode DdsPipe::enable() noexcept
 {
     std::lock_guard<std::mutex> lock(mutex_);
@@ -292,7 +316,40 @@ void DdsPipe::discovered_endpoint_nts_(
     }
     else if (is_endpoint_relevant_(endpoint))
     {
-        discovered_topic_nts_(utils::Heritable<DdsTopic>::make_heritable(endpoint.topic));
+        DdsTopic topic_with_partitions = endpoint.topic;
+        topic_with_partitions.partition_name = endpoint.specific_partitions;
+
+        discovered_topic_nts_(utils::Heritable<DdsTopic>::make_heritable(topic_with_partitions));
+    }
+    else
+    {
+        // there is a filter.
+        // update the track of the topic to
+        // add the partition in the reader if it is in the filter
+
+        // update partitions under bridges_mutex_
+        {
+            std::lock_guard<std::mutex> lock(bridges_mutex_);
+
+            const auto bridge_it = bridges_.find(utils::Heritable<DdsTopic>::make_heritable(endpoint.topic));
+            // add the specific partition of the endpoint in the bridges topic.
+            if (bridge_it != bridges_.end())
+            {
+                std::ostringstream guid_ss;
+                guid_ss << endpoint.guid;
+
+                const auto part_it = endpoint.specific_partitions.find(guid_ss.str());
+                if (part_it != endpoint.specific_partitions.end())
+                {
+                    bridge_it->second->add_partition_to_topic(guid_ss.str(), part_it->second);
+                }
+            }
+        }
+        // update readers outside the lock
+        // if (!filter_partition_.empty()) // TODO. danip
+        // {
+        //     update_readers_track(endpoint.topic.m_topic_name, filter_partition_);
+        // }
     }
 }
 
diff --git a/ddspipe_participants/include/ddspipe_participants/participant/dds/CommonParticipant.hpp b/ddspipe_participants/include/ddspipe_participants/participant/dds/CommonParticipant.hpp
@@ -198,7 +198,7 @@ class CommonParticipant : public core::IParticipant
             const std::shared_ptr<SimpleParticipantConfiguration>& participant_configuration,
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             const std::shared_ptr<core::DiscoveryDatabase>& discovery_database,
-            bool is_fastddsspy);
+            bool has_filter);
 
     /////////////////////////
     // VIRTUAL METHODS
@@ -272,7 +272,7 @@ class CommonParticipant : public core::IParticipant
     //! <Topics <Writer_guid, Partitions set>>
     std::map<std::string, std::map<std::string, std::string>> partition_names;
 
-    bool is_fastddsspy_;
+    bool has_filter_;
 };
 
 } /* namespace dds */
diff --git a/ddspipe_participants/include/ddspipe_participants/participant/dds/XmlParticipant.hpp b/ddspipe_participants/include/ddspipe_participants/participant/dds/XmlParticipant.hpp
@@ -37,7 +37,7 @@ class XmlParticipant
             const std::shared_ptr<XmlParticipantConfiguration>& participant_configuration,
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             const std::shared_ptr<core::DiscoveryDatabase>& discovery_database,
-            bool is_fastddsspy = false);
+            bool has_filter = false);
 
     /**
      * Specialized parent call so if it fails returns a blank one.
@@ -55,6 +55,13 @@ class XmlParticipant
     std::shared_ptr<core::IReader> create_reader(
             const core::ITopic& topic) override;
 
+    DDSPIPE_PARTICIPANTS_DllAPI
+    void update_filters(
+        const int flag,
+        std::set<std::string> partitions = std::set<std::string>(),
+        const std::string& topic_name = "",
+        const std::string& expression = "");
+
 protected:
 
     /////////////////////////
@@ -72,6 +79,13 @@ class XmlParticipant
 
     //! Participant configuration
     const XmlParticipantConfiguration& xml_specific_configuration_;
+
+    // Boolean to check if the created participant has filter
+    bool has_filter_;
+    // Filter partitions set
+    std::set<std::string> partition_filter_set_;
+    // Filter content_topicfilter dict
+    std::map<std::string, std::string> topic_filter_dict_;
 };
 
 } /* namespace dds */
diff --git a/ddspipe_participants/include/ddspipe_participants/participant/dynamic_types/XmlDynTypesParticipant.hpp b/ddspipe_participants/include/ddspipe_participants/participant/dynamic_types/XmlDynTypesParticipant.hpp
@@ -62,7 +62,7 @@ class XmlDynTypesParticipant : public dds::XmlParticipant
             std::shared_ptr<XmlParticipantConfiguration> participant_configuration,
             std::shared_ptr<core::PayloadPool> payload_pool,
             std::shared_ptr<core::DiscoveryDatabase> discovery_database,
-            bool is_fastddsspy = false);
+            bool has_filter = false);
 
     /**
      * @brief Creates a writer object based on the topic QoS.
diff --git a/ddspipe_participants/include/ddspipe_participants/reader/dds/CommonReader.hpp b/ddspipe_participants/include/ddspipe_participants/reader/dds/CommonReader.hpp
@@ -118,7 +118,7 @@ class CommonReader : public BaseReader, public fastdds::dds::DataReaderListener,
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             fastdds::dds::DomainParticipant* participant,
             fastdds::dds::Topic* topic_entity,
-            bool is_fastddsspy);
+            bool has_filter);
 
     // Specific enable/disable do not need to be implemented
 
@@ -185,7 +185,7 @@ class CommonReader : public BaseReader, public fastdds::dds::DataReaderListener,
     core::types::DdsTopic topic_;
     eprosima::fastdds::dds::ContentFilteredTopic* filtered_topic_;
     //std::string filter;
-    bool is_fastddsspy_;
+    bool has_filter_;
 
     fastdds::dds::Subscriber* dds_subscriber_;
     fastdds::dds::DataReader* reader_;
diff --git a/ddspipe_participants/include/ddspipe_participants/reader/dds/SimpleReader.hpp b/ddspipe_participants/include/ddspipe_participants/reader/dds/SimpleReader.hpp
@@ -48,7 +48,8 @@ class SimpleReader : public CommonReader
             const core::types::DdsTopic& topic,
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             fastdds::dds::DomainParticipant* participant,
-            fastdds::dds::Topic* topic_entity);
+            fastdds::dds::Topic* topic_entity,
+            bool has_filter = false);
 };
 
 } /* namespace dds */
diff --git a/ddspipe_participants/include/ddspipe_participants/reader/dds/SpecificQoSReader.hpp b/ddspipe_participants/include/ddspipe_participants/reader/dds/SpecificQoSReader.hpp
@@ -55,7 +55,7 @@ class SpecificQoSReader : public CommonReader
             fastdds::dds::DomainParticipant* participant,
             fastdds::dds::Topic* topic_entity,
             const std::shared_ptr<core::DiscoveryDatabase>& discovery_database,
-            bool is_fastddsspy);
+            bool has_filter);
 
 protected:
 
diff --git a/ddspipe_participants/include/ddspipe_participants/reader/rtps/CommonReader.hpp b/ddspipe_participants/include/ddspipe_participants/reader/rtps/CommonReader.hpp
@@ -207,7 +207,7 @@ class CommonReader : public BaseReader, public fastdds::rtps::ReaderListener
             const fastdds::rtps::TopicDescription& topic_desc,
             const fastdds::dds::ReaderQos& reader_qos,
             const std::set<std::string> allowed_partition_list,
-            bool is_fastddsspy);
+            bool has_filter);
 
     // Specific enable/disable do not need to be implemented
 
diff --git a/ddspipe_participants/src/cpp/participant/dds/CommonParticipant.cpp b/ddspipe_participants/src/cpp/participant/dds/CommonParticipant.cpp
@@ -201,7 +201,7 @@ std::shared_ptr<core::IReader> CommonParticipant::create_reader(
             dds_participant_,
             fastdds_topic,
             discovery_database_,
-            is_fastddsspy_); // TODO. danip HEREEEE
+            has_filter_); // TODO. danip HEREEEE
         reader->init();
 
         return reader;
@@ -213,7 +213,8 @@ std::shared_ptr<core::IReader> CommonParticipant::create_reader(
             dds_topic,
             this->payload_pool_,
             dds_participant_,
-            fastdds_topic);
+            fastdds_topic,
+            has_filter_); // TODO. danip HEREEEE
         reader->init();
 
         return reader;
@@ -370,11 +371,11 @@ CommonParticipant::CommonParticipant(
         const std::shared_ptr<SimpleParticipantConfiguration>& participant_configuration,
         const std::shared_ptr<core::PayloadPool>& payload_pool,
         const std::shared_ptr<core::DiscoveryDatabase>& discovery_database,
-        bool is_fastddsspy)
+        bool has_filter)
     : configuration_(participant_configuration)
     , payload_pool_(payload_pool)
     , discovery_database_(discovery_database)
-    , is_fastddsspy_(is_fastddsspy)
+    , has_filter_(has_filter)
 {
     // Do nothing
 }
diff --git a/ddspipe_participants/src/cpp/participant/dds/XmlParticipant.cpp b/ddspipe_participants/src/cpp/participant/dds/XmlParticipant.cpp
@@ -36,9 +36,10 @@ XmlParticipant::XmlParticipant(
         const std::shared_ptr<XmlParticipantConfiguration>& participant_configuration,
         const std::shared_ptr<core::PayloadPool>& payload_pool,
         const std::shared_ptr<core::DiscoveryDatabase>& discovery_database,
-        bool is_fastddsspy)
-    : CommonParticipant(participant_configuration, payload_pool, discovery_database, is_fastddsspy) // TODO. danip
+        bool has_filter)
+    : CommonParticipant(participant_configuration, payload_pool, discovery_database, has_filter) // TODO. danip
     , xml_specific_configuration_(*reinterpret_cast<XmlParticipantConfiguration*>(configuration_.get()))
+    , has_filter_(has_filter)
 {
     fastdds::dds::DomainParticipantExtendedQos extended_qos;
     if (xml_specific_configuration_.participant_profile.is_set() &&
@@ -73,7 +74,28 @@ std::shared_ptr<core::IReader> XmlParticipant::create_reader(
 {
     try
     {
-        return CommonParticipant::create_reader(topic);
+        std::shared_ptr<core::IReader> ret =
+            CommonParticipant::create_reader(topic);
+
+        if (has_filter_)
+        {
+            // Update the partitions
+            ret->update_partitions(partition_filter_set_);
+
+            // Check if the topic has a 'content_topicfilter' filter active
+            auto content_topicfilter_it = topic_filter_dict_.find(topic.topic_name());
+            std::string expression = "";
+            if(content_topicfilter_it != topic_filter_dict_.end())
+            {
+                expression = content_topicfilter_it->second;
+            }
+            // Update content_topicfilter expression
+            //  in this function, the reader is enabled
+            //  (to ensure safety updates of filters)
+            ret->update_content_topic_filter(expression);
+        }
+
+        return ret;
     }
     catch (const utils::InitializationException& e)
     {
@@ -108,6 +130,26 @@ fastdds::dds::DomainParticipantQos XmlParticipant::reckon_participant_qos_() con
     return qos;
 }
 
+// TODO. danip
+void XmlParticipant::update_filters(
+        const int flag,
+        std::set<std::string> partitions,
+        const std::string& topic_name,
+        const std::string& expression)
+{
+
+    if (flag == 0)
+    {
+        // partitions
+        partition_filter_set_ = partitions;
+    }
+    else
+    {
+        // content_topicfilter
+        topic_filter_dict_[topic_name] = expression;
+    }
+}
+
 } /* namespace dds */
 } /* namespace participants */
 } /* namespace ddspipe */
diff --git a/ddspipe_participants/src/cpp/participant/dynamic_types/XmlDynTypesParticipant.cpp b/ddspipe_participants/src/cpp/participant/dynamic_types/XmlDynTypesParticipant.cpp
@@ -45,12 +45,12 @@ XmlDynTypesParticipant::XmlDynTypesParticipant(
         std::shared_ptr<XmlParticipantConfiguration> participant_configuration,
         std::shared_ptr<PayloadPool> payload_pool,
         std::shared_ptr<DiscoveryDatabase> discovery_database,
-        bool is_fastddsspy)
+        bool has_filter)
     : XmlParticipant(
         participant_configuration,
         payload_pool,
         discovery_database,
-        is_fastddsspy)
+        has_filter)
     , type_object_reader_(std::make_shared<InternalReader>(
                 this->id()))
 {
diff --git a/ddspipe_participants/src/cpp/reader/dds/CommonReader.cpp b/ddspipe_participants/src/cpp/reader/dds/CommonReader.cpp
diff --git a/ddspipe_participants/src/cpp/reader/dds/SimpleReader.cpp b/ddspipe_participants/src/cpp/reader/dds/SimpleReader.cpp
diff --git a/ddspipe_participants/src/cpp/reader/dds/SpecificQoSReader.cpp b/ddspipe_participants/src/cpp/reader/dds/SpecificQoSReader.cpp
diff --git a/ddspipe_participants/src/cpp/reader/rtps/CommonReader.cpp b/ddspipe_participants/src/cpp/reader/rtps/CommonReader.cpp

Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ Track::Track(`
`63`	`63`
`64`	`64`	`// Initialized to false, only fastddspy`
`65`	`65`	`// will change this value`
`66`		`- is_fastddsspy = false;`
	`66`	`+ has_filter_ = false;`
`67`	`67`
`68`	`68`	`logDebug(DDSPIPE_TRACK, "Track " << *this << " created.");`
`69`	`69`	`}`
`@@ -110,7 +110,7 @@ void Track::enable() noexcept`
`110`	`110`	`writer_it.second->enable();`
`111`	`111`	`}`
`112`	`112`
`113`		`- if(is_fastddsspy)`
	`113`	`+ if(has_filter_)`
`114`	`114`	`{`
`115`	`115`	`reader_->update_content_topic_filter("");`
`116`	`116`	`reader_->update_partitions(std::set<std::string>());`