[#24061] Added filter functions to 'writers'

Danipiza · Danipiza · commit da25f7e6d554 · 2026-02-16T07:47:33.000+01:00
Signed-off-by: danipiza &lt;dpizarrogallego@gmail.com&gt;
diff --git a/ddspipe_core/include/ddspipe_core/interface/IWriter.hpp b/ddspipe_core/include/ddspipe_core/interface/IWriter.hpp
@@ -14,6 +14,8 @@
 
 #pragma once
 
+#include <set>
+
 #include <cpp_utils/ReturnCode.hpp>
 
 #include <ddspipe_core/interface/IRoutingData.hpp>
@@ -83,6 +85,16 @@ class IWriter
     virtual utils::ReturnCode write(
             IRoutingData& data) noexcept = 0;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_CORE_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) = 0;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_CORE_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) = 0;
+
     //! Update method to change the topic_partitions
     DDSPIPE_CORE_DllAPI
     virtual void update_topic_partitions(
diff --git a/ddspipe_participants/include/ddspipe_participants/testing/entities/mock_entities.hpp b/ddspipe_participants/include/ddspipe_participants/testing/entities/mock_entities.hpp
@@ -136,6 +136,17 @@ class MockWriter : public BaseWriter
     utils::ReturnCode write_nts_(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/BaseWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/BaseWriter.hpp
@@ -89,6 +89,16 @@ class BaseWriter : public core::IWriter
     virtual utils::ReturnCode write(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
     //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     void update_topic_partitions(
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/BlankWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/BlankWriter.hpp
@@ -43,6 +43,16 @@ class BlankWriter : public core::IWriter
     utils::ReturnCode write(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
     //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     void update_topic_partitions(
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/InternalWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/InternalWriter.hpp
@@ -40,6 +40,17 @@ class InternalWriter : public BaseWriter
     virtual utils::ReturnCode write_nts_(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
 
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/RtpsEchoWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/auxiliar/RtpsEchoWriter.hpp
@@ -52,6 +52,15 @@ class RtpsEchoWriter : public BlankWriter
     virtual utils::ReturnCode write(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
 
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/dds/CommonWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/dds/CommonWriter.hpp
@@ -74,7 +74,9 @@ class CommonWriter : public BaseWriter
      * @pre this method can only be called once.
      */
     DDSPIPE_PARTICIPANTS_DllAPI
-    void init();
+    void init(
+            const std::set<std::string>& partitions_set,
+            const std::string& content_topicfilter_expression);
 
     /////////////////////
     // STATIC ATTRIBUTES
@@ -164,7 +166,17 @@ class CommonWriter : public BaseWriter
             eprosima::fastdds::rtps::WriteParams& to_send_params,
             const core::types::RtpsPayloadData& data) const noexcept;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
 
+    //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
@@ -184,6 +196,7 @@ class CommonWriter : public BaseWriter
     std::shared_ptr<core::PayloadPoolMediator> payload_pool_;
 
     core::types::DdsTopic topic_;
+    eprosima::fastdds::dds::ContentFilteredTopic* filtered_topic_;
 
     fastdds::dds::Publisher* dds_publisher_;
     fastdds::dds::DataWriter* writer_;
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/dds/MultiWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/dds/MultiWriter.hpp
@@ -86,6 +86,15 @@ class MultiWriter : public BaseWriter
     virtual utils::ReturnCode write_nts_(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
 
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/dynamic_types/SchemaWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/dynamic_types/SchemaWriter.hpp
@@ -45,6 +45,16 @@ class SchemaWriter : public BaseWriter
             std::shared_ptr<core::PayloadPool> payload_pool,
             std::shared_ptr<ISchemaHandler> schema_handler);
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
     //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     void update_topic_partitions(
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/dynamic_types/TypeObjectWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/dynamic_types/TypeObjectWriter.hpp
@@ -54,6 +54,17 @@ class TypeObjectWriter : public BaseWriter
     utils::ReturnCode write_nts_(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
 
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/rtps/CommonWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/rtps/CommonWriter.hpp
@@ -96,7 +96,8 @@ class CommonWriter : public BaseWriter, public fastdds::rtps::WriterListener
      * @pre this method can only be called once.
      */
     DDSPIPE_PARTICIPANTS_DllAPI
-    void init();
+    void init(
+            const std::set<std::string>& partitions_set);
 
     /////////////////////////
     // RTPS LISTENER METHODS
@@ -205,6 +206,17 @@ class CommonWriter : public BaseWriter, public fastdds::rtps::WriterListener
             eprosima::fastdds::rtps::WriteParams& to_send_params,
             const core::types::RtpsPayloadData& data) const noexcept;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     DDSPIPE_PARTICIPANTS_DllAPI
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
@@ -236,7 +248,8 @@ class CommonWriter : public BaseWriter, public fastdds::rtps::WriterListener
             const fastdds::rtps::WriterAttributes& writer_attributes,
             const fastdds::rtps::TopicDescription& topic_description,
             const fastdds::dds::WriterQos& writer_qos,
-            const utils::PoolConfiguration& pool_configuration);
+            const utils::PoolConfiguration& pool_configuration,
+            const std::set<std::string>& partitions_set);
 
     /**
      * @brief History Attributes to create RTPS Writer History
diff --git a/ddspipe_participants/include/ddspipe_participants/writer/rtps/MultiWriter.hpp b/ddspipe_participants/include/ddspipe_participants/writer/rtps/MultiWriter.hpp
@@ -82,6 +82,17 @@ class MultiWriter : public BaseWriter
     virtual utils::ReturnCode write_nts_(
             core::IRoutingData& data) noexcept override;
 
+    //! Update method to change the partitions in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_partitions(
+            const std::set<std::string>& partitions_set) override;
+
+    //! Update method to change the expression in the content_topicfilter
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual void update_content_topic_filter(
+            const std::string& expression) override;
+
+    //! Update method to change the topic_partitions
     virtual void update_topic_partitions(
             const std::map<std::string, std::string>& partition_name) override;
 
diff --git a/ddspipe_participants/src/cpp/participant/dds/CommonParticipant.cpp b/ddspipe_participants/src/cpp/participant/dds/CommonParticipant.cpp
@@ -130,6 +130,13 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
     }
     const core::types::DdsTopic& dds_topic = *topic_ptr;
 
+    // contenttopicfilter
+    std::string content_topic_filter_expr = "";
+    if (topic_filter_dict_.find(dds_topic.m_topic_name) != topic_filter_dict_.end())
+    {
+        content_topic_filter_expr = topic_filter_dict_[dds_topic.m_topic_name];
+    }
+
     // Check that it is RTPS topic
     if (dds_topic.internal_type_discriminator() != core::types::INTERNAL_TOPIC_TYPE_RTPS)
     {
@@ -164,7 +171,7 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
             dds_participant_,
             fastdds_topic,
             configuration_->is_repeater);
-        writer->init();
+        writer->init(partition_filter_set_, content_topic_filter_expr);
 
         return writer;
     }
diff --git a/ddspipe_participants/src/cpp/participant/rtps/CommonParticipant.cpp b/ddspipe_participants/src/cpp/participant/rtps/CommonParticipant.cpp
@@ -425,7 +425,7 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
             this->payload_pool_,
             rtps_participant_,
             this->configuration_->is_repeater);
-        writer->init();
+        writer->init(partition_filter_set_);
 
         return writer;
     }
@@ -453,7 +453,7 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
                 this->payload_pool_,
                 rtps_participant_,
                 this->configuration_->is_repeater);
-            writer->init();
+            writer->init(partition_filter_set_);
 
             return writer;
         }
diff --git a/ddspipe_participants/src/cpp/reader/auxiliar/BaseReader.cpp b/ddspipe_participants/src/cpp/reader/auxiliar/BaseReader.cpp
@@ -207,13 +207,13 @@ core::types::DdsTopic BaseReader::topic() const
 }
 
 void BaseReader::update_partitions(
-        const std::set<std::string>& partitions_set)
+        const std::set<std::string>& /* partitions_set */)
 {
     // Nothing
 }
 
 void BaseReader::update_content_topic_filter(
-        const std::string& expression)
+        const std::string& /* expression */)
 {
     // Nothing
 }
diff --git a/ddspipe_participants/src/cpp/reader/auxiliar/BlankReader.cpp b/ddspipe_participants/src/cpp/reader/auxiliar/BlankReader.cpp
@@ -65,13 +65,13 @@ core::types::DdsTopic BlankReader::topic() const
 }
 
 void BlankReader::update_partitions(
-        const std::set<std::string>& partitions_set)
+        const std::set<std::string>& /* partitions_set */)
 {
     // Nothing
 }
 
 void BlankReader::update_content_topic_filter(
-        const std::string& expression)
+        const std::string& /* expression */)
 {
     // Nothing
     return;
diff --git a/ddspipe_participants/src/cpp/reader/auxiliar/InternalReader.cpp b/ddspipe_participants/src/cpp/reader/auxiliar/InternalReader.cpp
@@ -89,7 +89,7 @@ void InternalReader::update_partitions(
 }
 
 void InternalReader::update_content_topic_filter(
-        const std::string& expression)
+        const std::string& /* expression */)
 {
     // Nothing
 }
diff --git a/ddspipe_participants/src/cpp/testing/entities/mock_entities.cpp b/ddspipe_participants/src/cpp/testing/entities/mock_entities.cpp
@@ -184,6 +184,18 @@ utils::ReturnCode MockWriter::write_nts_(
     return utils::ReturnCode::RETCODE_OK;
 }
 
+void MockWriter::update_partitions(
+        const std::set<std::string>& /* partitions_set */)
+{
+    // Nothing
+}
+
+void MockWriter::update_content_topic_filter(
+        const std::string& /* expression */)
+{
+    // Nothing
+}
+
 void MockWriter::update_topic_partitions(
         const std::map<std::string, std::string>& partition_name)
 {
diff --git a/ddspipe_participants/src/cpp/writer/auxiliar/BaseWriter.cpp b/ddspipe_participants/src/cpp/writer/auxiliar/BaseWriter.cpp
@@ -119,6 +119,18 @@ bool BaseWriter::should_send_sample_() noexcept
     return true;
 }
 
+void BaseWriter::update_partitions(
+        const std::set<std::string>& /* partitions_set */)
+{
+    // Nothing
+}
+
+void BaseWriter::update_content_topic_filter(
+        const std::string& /* expression */)
+{
+    // Nothing
+}
+
 void BaseWriter::update_topic_partitions(
         const std::map<std::string, std::string>& /* partition_name */)
 {
diff --git a/ddspipe_participants/src/cpp/writer/auxiliar/BlankWriter.cpp b/ddspipe_participants/src/cpp/writer/auxiliar/BlankWriter.cpp
@@ -33,6 +33,18 @@ utils::ReturnCode BlankWriter::write(
     return utils::ReturnCode::RETCODE_OK;
 }
 
+void BlankWriter::update_partitions(
+        const std::set<std::string>& /* partitions_set */)
+{
+    // Nothing
+}
+
+void BlankWriter::update_content_topic_filter(
+        const std::string& /* expression */)
+{
+    // Nothing
+}
+
 void BlankWriter::update_topic_partitions(
         const std::map<std::string, std::string>& /* partition_name */)
 {
diff --git a/ddspipe_participants/src/cpp/writer/auxiliar/InternalWriter.cpp b/ddspipe_participants/src/cpp/writer/auxiliar/InternalWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/auxiliar/RtpsEchoWriter.cpp b/ddspipe_participants/src/cpp/writer/auxiliar/RtpsEchoWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/dds/CommonWriter.cpp b/ddspipe_participants/src/cpp/writer/dds/CommonWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/dds/MultiWriter.cpp b/ddspipe_participants/src/cpp/writer/dds/MultiWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/dynamic_types/SchemaWriter.cpp b/ddspipe_participants/src/cpp/writer/dynamic_types/SchemaWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/dynamic_types/TypeObjectWriter.cpp b/ddspipe_participants/src/cpp/writer/dynamic_types/TypeObjectWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/rtps/CommonWriter.cpp b/ddspipe_participants/src/cpp/writer/rtps/CommonWriter.cpp
diff --git a/ddspipe_participants/src/cpp/writer/rtps/MultiWriter.cpp b/ddspipe_participants/src/cpp/writer/rtps/MultiWriter.cpp
