[23264] Feature: DataWriter sample prefilter

include/fastdds/dds/publisher/DataWriter.hpp

@@ -55,6 +55,8 @@ class DataWriterListener;
 class DataWriterQos;
 class Topic;
 
+struct IContentFilter;
+
 /**
  * Class DataWriter, contains the actual implementation of the behaviour of the DataWriter.
  *
@@ -596,6 +598,22 @@ class DataWriter : public DomainEntity
     FASTDDS_EXPORTED_API ReturnCode_t get_publication_builtin_topic_data(
             PublicationBuiltinTopicData& publication_data) const;
 
+    /**
+     *  @brief Set a sample prefilter to be used. This filter is always
+     *  evaluated before sending the sample to any DataReader and prior to
+     *  any content filtering.
+     *  Reader filters should be enabled in the DataWriter.
+     *
+     * @param prefilter The prefilter to be set.
+     *
+     * @return RETCODE_OK if the prefilter is set correctly,
+     * @return RETCODE_PRECONDITION_NOT_MET if the reader filters are not enabled.
+     *
+     * @note The prefilter is currently incompatible with DataSharing.
+     */
+    FASTDDS_EXPORTED_API ReturnCode_t set_sample_prefilter(
+            std::shared_ptr<IContentFilter> prefilter);
+
 protected:
 
     DataWriterImpl* impl_;

include/fastdds/dds/topic/IContentFilter.hpp

@@ -24,6 +24,7 @@
 #include <fastdds/rtps/common/Guid.hpp>
 #include <fastdds/rtps/common/SampleIdentity.hpp>
 #include <fastdds/rtps/common/SerializedPayload.hpp>
+#include <fastdds/rtps/common/WriteParams.hpp>
 
 namespace eprosima {
 namespace fastdds {
@@ -34,6 +35,8 @@ namespace dds {
  */
 struct IContentFilter
 {
+    virtual ~IContentFilter() = default;
+
     using SerializedPayload = eprosima::fastdds::rtps::SerializedPayload_t;
     using GUID_t = fastdds::rtps::GUID_t;
 
@@ -45,10 +48,22 @@ struct IContentFilter
     {
         using SampleIdentity = eprosima::fastdds::rtps::SampleIdentity;
 
+        FilterSampleInfo() = default;
+
+        FilterSampleInfo(
+                const rtps::WriteParams& wparams)
+            : sample_identity(wparams.sample_identity())
+            , related_sample_identity(wparams.related_sample_identity())
+            , user_write_data(wparams.user_write_data())
+        {
+        }
+
         /// Identity of the sample being filtered.
         SampleIdentity sample_identity;
         /// Identity of a sample related to the one being filtered.
         SampleIdentity related_sample_identity;
+        /// Extra write information that can be used by the prefilter.
+        std::shared_ptr<fastdds::rtps::WriteParams::UserWriteData> user_write_data;
     };
 
     /**

include/fastdds/rtps/common/WriteParams.hpp

@@ -18,6 +18,8 @@
 #ifndef FASTDDS_RTPS_COMMON__WRITEPARAMS_HPP
 #define FASTDDS_RTPS_COMMON__WRITEPARAMS_HPP
 
+#include <memory>
+
 #include <fastdds/rtps/common/SampleIdentity.hpp>
 #include <fastdds/rtps/common/Time_t.hpp>
 
@@ -34,6 +36,21 @@ class FASTDDS_EXPORTED_API WriteParams
 {
 public:
 
+    /**
+     * @brief Base class storing custom information in the
+     * WriteParams structure for later filtering.
+     *
+     * This struct serves as a base class that allows derived
+     * classes to be deleted safely through a pointer to this base type.
+     * It is intended to be user-extensible.
+     */
+    struct FASTDDS_EXPORTED_API UserWriteData
+    {
+        UserWriteData() = default;
+
+        virtual ~UserWriteData() = default;
+    };
+
     /**
      * Set the value of the sample_identity member.
      *
@@ -178,6 +195,30 @@ class FASTDDS_EXPORTED_API WriteParams
         return *this;
     }
 
+    /**
+     * @brief Retrieves the user write data.
+     *
+     * @return Shared pointer to the user write data.
+     */
+    std::shared_ptr<UserWriteData> user_write_data() const
+    {
+        return user_write_data_;
+    }
+
+    /**
+     * Set the user write data.
+     *
+     * @param write_data  New value for the user_write_data member.
+     *
+     * @return Reference to the modified object in order to allow daisy chaining.
+     */
+    WriteParams& user_write_data(
+            std::shared_ptr<UserWriteData> write_data)
+    {
+        user_write_data_ = write_data;
+        return *this;
+    }
+
     static WriteParams WRITE_PARAM_DEFAULT;
 
     /**
@@ -198,12 +239,26 @@ class FASTDDS_EXPORTED_API WriteParams
 
 private:
 
+    class FASTDDS_EXPORTED_API UserWriteDataPtr : public std::shared_ptr<UserWriteData>
+    {
+    public:
+
+        UserWriteDataPtr(
+                std::shared_ptr<UserWriteData> ptr)
+            : std::shared_ptr<UserWriteData>(ptr)
+        {
+        }
+
+    };
+
     /// Attribute that holds sample_identity member value
     SampleIdentity sample_identity_;
     /// Attribute that holds related_sample_identity member value
     SampleIdentity related_sample_identity_;
     /// Attribute that holds source_timestamp member value
     Time_t source_timestamp_{ -1, TIME_T_INFINITE_NANOSECONDS };
+    /// User write data
+    UserWriteDataPtr user_write_data_{nullptr};
 };
 
 }  // namespace rtps

src/cpp/fastdds/publisher/DataWriter.cpp

@@ -305,6 +305,12 @@ ReturnCode_t DataWriter::get_publication_builtin_topic_data(
     return impl_->get_publication_builtin_topic_data(publication_data);
 }
 
+ReturnCode_t DataWriter::set_sample_prefilter(
+        std::shared_ptr<IContentFilter> prefilter)
+{
+    return impl_->set_sample_prefilter(prefilter);
+}
+
 } // namespace dds
 } // namespace fastdds
 } // namespace eprosima

src/cpp/fastdds/publisher/DataWriterImpl.cpp

@@ -1500,6 +1500,24 @@ ReturnCode_t DataWriterImpl::get_publication_matched_status(
     return RETCODE_OK;
 }
 
+ReturnCode_t DataWriterImpl::set_sample_prefilter(
+        std::shared_ptr<IContentFilter> prefilter)
+{
+    ReturnCode_t ret_code = RETCODE_OK;
+    if (!reader_filters_)
+    {
+        EPROSIMA_LOG_ERROR(DATA_WRITER, "Filtering is not enabled for this DataWriter");
+        ret_code = RETCODE_PRECONDITION_NOT_MET;
+    }
+    else
+    {
+        std::lock_guard<std::mutex> lock(sample_prefilter_mutex_);
+        sample_prefilter_ = prefilter;
+    }
+
+    return ret_code;
+}
+
 bool DataWriterImpl::deadline_timer_reschedule()
 {
     assert(qos_.deadline().period != dds::c_TimeInfinite);
@@ -2372,9 +2390,27 @@ bool DataWriterImpl::is_relevant(
         const fastdds::rtps::CacheChange_t& change,
         const fastdds::rtps::GUID_t& reader_guid) const
 {
-    assert(reader_filters_);
+    assert(reader_filters_ || sample_prefilter_);
+    bool is_relevant_for_reader = true;
     const DataWriterFilteredChange& writer_change = static_cast<const DataWriterFilteredChange&>(change);
-    return writer_change.is_relevant_for(reader_guid);
+
+    {
+        std::lock_guard<std::mutex> lock(sample_prefilter_mutex_);
+        if (sample_prefilter_)
+        {
+            IContentFilter::FilterSampleInfo filter_sample_info(writer_change.write_params);
+            is_relevant_for_reader = sample_prefilter_->evaluate(writer_change.serializedPayload,
+                            filter_sample_info,
+                            reader_guid);
+        }
+    }
+
+    if (is_relevant_for_reader && reader_filters_)
+    {
+        is_relevant_for_reader = writer_change.is_relevant_for(reader_guid);
+    }
+
+    return is_relevant_for_reader;
 }
 
 } // namespace dds

src/cpp/fastdds/publisher/DataWriterImpl.hpp

@@ -417,6 +417,20 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
     ReturnCode_t get_publication_builtin_topic_data(
             PublicationBuiltinTopicData& publication_data) const;
 
+    /**
+     *  @brief Set a sample prefilter to be used. This filter is always
+     *  evaluated before sending the sample to any DataReader and prior to
+     *  any content filtering.
+     *  Reader filters should be enabled in the DataWriter.
+     *
+     * @param prefilter The prefilter to be set.
+     *
+     * @return RETCODE_OK if the prefilter is set correctly,
+     * @return RETCODE_PRECONDITION_NOT_MET if the reader filters are not enabled.
+     */
+    ReturnCode_t set_sample_prefilter(
+            std::shared_ptr<IContentFilter> prefilter);
+
 protected:
 
     using IChangePool = eprosima::fastdds::rtps::IChangePool;
@@ -537,6 +551,9 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
 
     DataRepresentationId_t data_representation_ {DEFAULT_DATA_REPRESENTATION};
 
+    mutable std::mutex sample_prefilter_mutex_;
+    std::shared_ptr<IContentFilter> sample_prefilter_;
+
     ReturnCode_t check_write_preconditions(
             const void* const data,
             const InstanceHandle_t& handle,

test/blackbox/api/dds-pim/PubSubWriter.hpp

@@ -514,13 +514,25 @@ class PubSubWriter
 
     void send(
             std::list<type>& msgs,
-            uint32_t milliseconds = 0)
+            uint32_t milliseconds = 0,
+            bool send_with_params = false)
     {
         auto it = msgs.begin();
 
         while (it != msgs.end())
         {
-            if (eprosima::fastdds::dds::RETCODE_OK == datawriter_->write((void*)&(*it)))
+            auto ret = eprosima::fastdds::dds::RETCODE_OK;
+
+            if (send_with_params)
+            {
+                ret = datawriter_->write((void*)&(*it), write_params_);
+            }
+            else
+            {
+                ret = datawriter_->write((void*)&(*it));
+            }
+
+            if (eprosima::fastdds::dds::RETCODE_OK == ret)
             {
                 default_send_print<type>(*it);
                 it = msgs.erase(it);
@@ -1858,6 +1870,19 @@ class PubSubWriter
         return type_;
     }
 
+    eprosima::fastdds::dds::ReturnCode_t set_sample_prefilter(
+            std::shared_ptr<eprosima::fastdds::dds::IContentFilter> prefilter)
+    {
+        return datawriter_->set_sample_prefilter(prefilter);
+    }
+
+    PubSubWriter& write_params(
+            const eprosima::fastdds::rtps::WriteParams& params)
+    {
+        write_params_ = params;
+        return *this;
+    }
+
 protected:
 
     void participant_matched()
@@ -2158,6 +2183,7 @@ class PubSubWriter
     //! Preferred domain ID
     bool use_preferred_domain_id_;
     uint32_t preferred_domain_id_;
+    eprosima::fastdds::rtps::WriteParams write_params_;
 
 #if HAVE_SECURITY
     std::mutex mutexAuthentication_;