Support repeater feature in DDS (XML) participants

Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>

Author: juanlofer-eprosima

ddspipe_participants/include/ddspipe_participants/writer/dds/CommonWriter.hpp

@@ -20,7 +20,9 @@
 #include <fastdds/dds/publisher/DataWriter.hpp>
 #include <fastdds/dds/publisher/Publisher.hpp>
 #include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
+#include <fastdds/dds/topic/IContentFilter.hpp>
 #include <fastdds/dds/topic/Topic.hpp>
+#include <fastdds/rtps/common/WriteParams.hpp>
 
 #include <ddspipe_core/efficiency/payload/PayloadPoolMediator.hpp>
 #include <ddspipe_core/types/participant/ParticipantId.hpp>
@@ -99,7 +101,8 @@ class CommonWriter : public BaseWriter
             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,
+            const bool repeater);
 
     /////////////////////////
     // IWRITER METHODS
@@ -128,20 +131,30 @@ class CommonWriter : public BaseWriter
     // INTERNAL METHODS
     /////////////////////////
 
-    virtual
-    fastdds::dds::PublisherQos
-    reckon_publisher_qos_() const noexcept;
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual fastdds::dds::PublisherQos reckon_publisher_qos_() const noexcept;
+
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual fastdds::dds::DataWriterQos reckon_writer_qos_() const noexcept;
 
-    virtual
-    fastdds::dds::DataWriterQos
-    reckon_writer_qos_() const noexcept;
+    /**
+     * @brief Auxiliary method used in \c write to fill the sample to send and write params.
+     *
+     * @param [out] to_send_params write params to be filled and sent.
+     * @param [in] data data received that must be sent.
+     */
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual utils::ReturnCode fill_to_send_data_(
+            eprosima::fastdds::rtps::WriteParams& to_send_params,
+            const core::types::RtpsPayloadData& data) const noexcept;
 
     /////////////////////////
     // EXTERNAL VARIABLES
     /////////////////////////
 
     fastdds::dds::DomainParticipant* dds_participant_;
     fastdds::dds::Topic* dds_topic_;
+    bool repeater_;
 
     /////////////////////////
     // INTERNAL VARIABLES
@@ -153,6 +166,7 @@ class CommonWriter : public BaseWriter
 
     fastdds::dds::Publisher* dds_publisher_;
     fastdds::dds::DataWriter* writer_;
+    std::shared_ptr<fastdds::dds::IContentFilter> data_filter_;
 };
 
 } /* namespace dds */

ddspipe_participants/include/ddspipe_participants/writer/dds/MultiWriter.hpp

@@ -48,6 +48,7 @@ class MultiWriter : public BaseWriter
      * @param payload_pool      Shared Payload Pool to received data and take it.
      * @param participant       DDS Participant pointer.
      * @param topic_entity      DDS Topic pointer.
+     * @param repeater          If this MultiWriter is a repeater or not.
      *
      * @throw \c InitializationException in case any creation has failed
      */
@@ -57,7 +58,8 @@ class MultiWriter : public BaseWriter
             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,
+            const bool repeater = false);
 
     /**
      * @brief Destroy the MultiWriter object
@@ -99,6 +101,7 @@ class MultiWriter : public BaseWriter
 
     fastdds::dds::DomainParticipant* dds_participant_;
     fastdds::dds::Topic* dds_topic_;
+    bool repeater_;
 
     /////////////////////////
     // INTERNAL VARIABLES

ddspipe_participants/include/ddspipe_participants/writer/dds/QoSSpecificWriter.hpp

@@ -39,8 +39,10 @@ class QoSSpecificWriter : public CommonWriter
      * @param participant_id    Router Id of the Participant that has created this QoSSpecificWriter.
      * @param topic             Topic that this QoSSpecificWriter subscribes to.
      * @param payload_pool      Shared Payload Pool to received data and take it.
+     * @param specific_qos      Specific QoS of the Endpoint.
      * @param participant       DDS Participant pointer.
      * @param topic_entity      DDS Topic pointer.
+     * @param repeater          If this QoSSpecificWriter is a repeater or not.
      *
      * @throw \c InitializationException in case any creation has failed
      */
@@ -51,7 +53,8 @@ class QoSSpecificWriter : public CommonWriter
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             const core::types::SpecificEndpointQoS& specific_qos,
             fastdds::dds::DomainParticipant* participant,
-            fastdds::dds::Topic* topic_entity);
+            fastdds::dds::Topic* topic_entity,
+            const bool repeater = false);
 
 protected:
 

ddspipe_participants/include/ddspipe_participants/writer/dds/SimpleWriter.hpp

@@ -43,6 +43,7 @@ class SimpleWriter : public CommonWriter
      * @param payload_pool      Shared Payload Pool to received data and take it.
      * @param participant       DDS Participant pointer.
      * @param topic_entity      DDS Topic pointer.
+     * @param repeater          If this SimpleWriter is a repeater or not.
      *
      * @throw \c InitializationException in case any creation has failed
      */
@@ -52,7 +53,8 @@ class SimpleWriter : public CommonWriter
             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,
+            const bool repeater = false);
 
 };
 

ddspipe_participants/include/ddspipe_participants/writer/dds/filter/RepeaterDataFilter.hpp

@@ -0,0 +1,53 @@
+// Copyright 2025 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <fastdds/rtps/common/Guid.hpp>
+#include <fastdds/rtps/common/WriteParams.hpp>
+
+#include <ddspipe_participants/library/library_dll.h>
+#include <ddspipe_participants/writer/dds/filter/SelfDataFilter.hpp>
+
+namespace eprosima {
+namespace ddspipe {
+namespace participants {
+namespace dds {
+
+struct RepeaterWriteData : public fastdds::rtps::WriteParams::UserWriteData
+{
+    fastdds::rtps::GuidPrefix_t last_writer_guid_prefix;  ///< GUID of the writer that sent this change.
+};
+
+/**
+ * This filter allows to not send messages from this Writer to the Readers belonging to the source Participant.
+ * It is used in "repeater" participants in order to propagate information to external participants
+ * (participants not belonging to the same DDS-Router instance),
+ * leaving out the participant from which this information was received.
+ */
+class RepeaterDataFilter : public SelfDataFilter
+{
+public:
+
+    DDSPIPE_PARTICIPANTS_DllAPI
+    bool evaluate(
+            const fastdds::rtps::SerializedPayload_t& payload,
+            const fastdds::dds::IContentFilter::FilterSampleInfo& sample_info,
+            const fastdds::rtps::GUID_t& reader_guid) const override;
+};
+
+} /* namespace dds */
+} /* namespace participants */
+} /* namespace ddspipe */
+} /* namespace eprosima */

ddspipe_participants/include/ddspipe_participants/writer/dds/filter/SelfDataFilter.hpp

@@ -0,0 +1,52 @@
+// Copyright 2025 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <fastdds/dds/topic/IContentFilter.hpp>
+
+#include <fastdds/rtps/common/Guid.hpp>
+#include <fastdds/rtps/common/SerializedPayload.hpp>
+
+#include <ddspipe_participants/library/library_dll.h>
+
+namespace eprosima {
+namespace ddspipe {
+namespace participants {
+namespace dds {
+
+/**
+ * This filter allows to not send messages from this Writer to the Readers in the same Participant.
+ */
+class SelfDataFilter : public fastdds::dds::IContentFilter
+{
+public:
+
+    /**
+     * @brief Whether incoming sample is relevant for this reader.
+     *
+     * @return true if the reader does not belong to same Participant.
+     * @return false otherwise.
+     */
+    DDSPIPE_PARTICIPANTS_DllAPI
+    bool evaluate(
+            const fastdds::rtps::SerializedPayload_t& payload,
+            const fastdds::dds::IContentFilter::FilterSampleInfo& sample_info,
+            const fastdds::rtps::GUID_t& reader_guid) const override;
+};
+
+} /* namespace dds */
+} /* namespace participants */
+} /* namespace ddspipe */
+} /* namespace eprosima */

ddspipe_participants/include/ddspipe_participants/writer/rtps/CommonWriter.hpp

@@ -193,7 +193,7 @@ class CommonWriter : public BaseWriter, public fastdds::rtps::WriterListener
             core::IRoutingData& data) noexcept override;
 
     /**
-     * @brief Auxiliary method used in \c write to fill the cache change to send.
+     * @brief Auxiliary method used in \c write to fill the cache change to send and write params.
      *
      * @param [out] to_send_change_to_fill cache change to be filled and sent.
      * @param [out] to_send_params write params to be filled and sent.

ddspipe_participants/src/cpp/participant/dds/CommonParticipant.cpp

@@ -109,7 +109,7 @@ bool CommonParticipant::is_rtps_kind() const noexcept
 
 bool CommonParticipant::is_repeater() const noexcept
 {
-    return false;
+    return configuration_->is_repeater;
 }
 
 core::types::TopicQoS CommonParticipant::topic_qos() const noexcept
@@ -147,7 +147,8 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
             dds_topic,
             this->payload_pool_,
             dds_participant_,
-            fastdds_topic);
+            fastdds_topic,
+            configuration_->is_repeater);
     }
     else
     {
@@ -156,7 +157,8 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
             dds_topic,
             this->payload_pool_,
             dds_participant_,
-            fastdds_topic);
+            fastdds_topic,
+            configuration_->is_repeater);
         writer->init();
 
         return writer;

ddspipe_participants/src/cpp/writer/dds/CommonWriter.cpp

@@ -13,17 +13,20 @@
 // limitations under the License.
 
 
-#include <fastdds/rtps/RTPSDomain.hpp>
-#include <fastdds/rtps/participant/RTPSParticipant.hpp>
 #include <fastdds/rtps/common/CacheChange.hpp>
+#include <fastdds/rtps/common/WriteParams.hpp>
+#include <fastdds/rtps/participant/RTPSParticipant.hpp>
+#include <fastdds/rtps/RTPSDomain.hpp>
 
 #include <cpp_utils/exception/InitializationException.hpp>
 #include <cpp_utils/Log.hpp>
 #include <cpp_utils/time/time_utils.hpp>
 
 #include <ddspipe_participants/efficiency/cache_change/CacheChangePool.hpp>
-#include <ddspipe_participants/writer/dds/CommonWriter.hpp>
 #include <ddspipe_participants/types/dds/RouterCacheChange.hpp>
+#include <ddspipe_participants/writer/dds/CommonWriter.hpp>
+#include <ddspipe_participants/writer/dds/filter/RepeaterDataFilter.hpp>
+#include <ddspipe_participants/writer/dds/filter/SelfDataFilter.hpp>
 
 namespace eprosima {
 namespace ddspipe {
@@ -83,19 +86,39 @@ void CommonWriter::init()
                   utils::Formatter() << "Error creating DataWriter for Participant " <<
                       participant_id_ << " in topic " << topic_ << ".");
     }
+
+    if (repeater_)
+    {
+        // Use filter writer of origin
+        data_filter_ = std::make_shared<RepeaterDataFilter>();
+    }
+    else
+    {
+        // Use default filter
+        data_filter_ = std::make_shared<SelfDataFilter>();
+    }
+
+    if (fastdds::dds::RETCODE_OK != writer_->set_sample_prefilter(data_filter_))
+    {
+        throw utils::InitializationException(
+                  utils::Formatter() << "Error setting DataWriter prefilter for Participant " <<
+                      participant_id_ << " in topic " << topic_ << ".");
+    }
 }
 
 CommonWriter::CommonWriter(
         const ParticipantId& participant_id,
         const DdsTopic& topic,
         const std::shared_ptr<core::PayloadPool>& payload_pool,
         fastdds::dds::DomainParticipant* participant,
-        fastdds::dds::Topic* topic_entity)
+        fastdds::dds::Topic* topic_entity,
+        const bool repeater)
     : BaseWriter(participant_id, topic.topic_qos.max_tx_rate)
     , dds_participant_(participant)
     , dds_topic_(topic_entity)
     , payload_pool_(new core::PayloadPoolMediator(payload_pool))
     , topic_(topic)
+    , repeater_(repeater)
     , dds_publisher_(nullptr)
     , writer_(nullptr)
 {
@@ -110,15 +133,23 @@ utils::ReturnCode CommonWriter::write_nts_(
 
     auto& rtps_data = dynamic_cast<core::types::RtpsPayloadData&>(data);
 
-    if (topic_.topic_qos.keyed)
-    {
-        // TODO check if in case of dispose it must be done something differently
-        return payload_pool_->write(writer_, &rtps_data, rtps_data.instanceHandle);
-    }
-    else
+    fastdds::rtps::WriteParams wparams;
+
+    if (fill_to_send_data_(wparams, rtps_data) != utils::ReturnCode::RETCODE_OK)
     {
-        return payload_pool_->write(writer_, &rtps_data);
+        EPROSIMA_LOG_ERROR(DDSPIPE_DDS_WRITER, "Error setting data to send.");
+        return utils::ReturnCode::RETCODE_ERROR;
     }
+
+    // WARNING: At the time of this writing, there is no DataWriter API to write both with parameters and instance handle.
+    // However, the current implementation of write without instance handle always computes its value via type support
+    // (if it corresponds to a keyed topic), so it is equivalent to write with instance handle (and can hence use the
+    // write with params overload to cover all cases). Future developers should be aware of this and might need to
+    // upload this method if the DataWriter implementation changes at some point.
+    return payload_pool_->write(writer_, &rtps_data, wparams);
+
+    // TODO: handle dipose case -> DataWriter::write will always send ALIVE changes, so this case must be handled
+    // with additional logic (e.g. by using unregister_instance instead of write).
 }
 
 fastdds::dds::PublisherQos CommonWriter::reckon_publisher_qos_() const noexcept
@@ -172,6 +203,23 @@ fastdds::dds::DataWriterQos CommonWriter::reckon_writer_qos_() const noexcept
     return qos;
 }
 
+utils::ReturnCode CommonWriter::fill_to_send_data_(
+        fastdds::rtps::WriteParams& to_send_params,
+        const RtpsPayloadData& data) const noexcept
+{
+    if (repeater_)
+    {
+        auto write_data = std::make_shared<RepeaterWriteData>();
+        write_data->last_writer_guid_prefix = data.source_guid.guidPrefix;
+        to_send_params.user_write_data(write_data);
+    }
+
+    // Set source time stamp to be the original one
+    to_send_params.source_timestamp(data.source_timestamp);
+
+    return utils::ReturnCode::RETCODE_OK;
+}
+
 } /* namespace dds */
 } /* namespace participants */
 } /* namespace ddspipe */