Skip to content

Commit 2d4e84c

Browse files
juanlofer-eprosimajuanlofer-eprosima
authored
Support repeater feature in DDS (XML) participants (#151)
* Support repeater feature in DDS (XML) participants Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com> * Apply suggestions Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com> * Apply more suggestions Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com> --------- Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com>
1 parent 5b319e9 commit 2d4e84c

15 files changed

Lines changed: 360 additions & 33 deletions

File tree

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

Lines changed: 37 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,9 @@
2020
#include <fastdds/dds/publisher/DataWriter.hpp>
2121
#include <fastdds/dds/publisher/Publisher.hpp>
2222
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
23+
#include <fastdds/dds/topic/IContentFilter.hpp>
2324
#include <fastdds/dds/topic/Topic.hpp>
25+
#include <fastdds/rtps/common/WriteParams.hpp>
2426

2527
#include <ddspipe_core/efficiency/payload/PayloadPoolMediator.hpp>
2628
#include <ddspipe_core/types/participant/ParticipantId.hpp>
@@ -99,7 +101,8 @@ class CommonWriter : public BaseWriter
99101
const core::types::DdsTopic& topic,
100102
const std::shared_ptr<core::PayloadPool>& payload_pool,
101103
fastdds::dds::DomainParticipant* participant,
102-
fastdds::dds::Topic* topic_entity);
104+
fastdds::dds::Topic* topic_entity,
105+
const bool repeater);
103106

104107
/////////////////////////
105108
// IWRITER METHODS
@@ -128,20 +131,46 @@ class CommonWriter : public BaseWriter
128131
// INTERNAL METHODS
129132
/////////////////////////
130133

131-
virtual
132-
fastdds::dds::PublisherQos
133-
reckon_publisher_qos_() const noexcept;
134+
/**
135+
* @brief Get the specific Publisher QoS for the \c DdsTopic associated to this writer.
136+
*
137+
* This method is used to set the QoS of the Publisher that will be created for this writer.
138+
* It takes into account the default publisher QoS and the specific \c DdsTopic QoS of this writer.
139+
*
140+
* @return The Publisher QoS to be used for this writer.
141+
*/
142+
DDSPIPE_PARTICIPANTS_DllAPI
143+
virtual fastdds::dds::PublisherQos reckon_publisher_qos_() const noexcept;
134144

135-
virtual
136-
fastdds::dds::DataWriterQos
137-
reckon_writer_qos_() const noexcept;
145+
/**
146+
* @brief Get the specific DataWriter QoS for the \c DdsTopic associated to this writer.
147+
*
148+
* This method is used to set the QoS of the DataWriter that will be created for this writer.
149+
* It takes into account the default data writer QoS and the specific \c DdsTopic QoS of this writer.
150+
*
151+
* @return The DataWriter QoS to be used for this writer.
152+
*/
153+
DDSPIPE_PARTICIPANTS_DllAPI
154+
virtual fastdds::dds::DataWriterQos reckon_writer_qos_() const noexcept;
155+
156+
/**
157+
* @brief Auxiliary method used in \c write to fill the sample to send and write params.
158+
*
159+
* @param [out] to_send_params write params to be filled and sent.
160+
* @param [in] data data received that must be sent.
161+
*/
162+
DDSPIPE_PARTICIPANTS_DllAPI
163+
virtual utils::ReturnCode fill_to_send_data_(
164+
eprosima::fastdds::rtps::WriteParams& to_send_params,
165+
const core::types::RtpsPayloadData& data) const noexcept;
138166

139167
/////////////////////////
140168
// EXTERNAL VARIABLES
141169
/////////////////////////
142170

143171
fastdds::dds::DomainParticipant* dds_participant_;
144172
fastdds::dds::Topic* dds_topic_;
173+
bool repeater_;
145174

146175
/////////////////////////
147176
// INTERNAL VARIABLES
@@ -153,6 +182,7 @@ class CommonWriter : public BaseWriter
153182

154183
fastdds::dds::Publisher* dds_publisher_;
155184
fastdds::dds::DataWriter* writer_;
185+
std::shared_ptr<fastdds::dds::IContentFilter> data_filter_;
156186
};
157187

158188
} /* namespace dds */

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ class MultiWriter : public BaseWriter
4848
* @param payload_pool Shared Payload Pool to received data and take it.
4949
* @param participant DDS Participant pointer.
5050
* @param topic_entity DDS Topic pointer.
51+
* @param repeater If this MultiWriter is a repeater or not.
5152
*
5253
* @throw \c InitializationException in case any creation has failed
5354
*/
@@ -57,7 +58,8 @@ class MultiWriter : public BaseWriter
5758
const core::types::DdsTopic& topic,
5859
const std::shared_ptr<core::PayloadPool>& payload_pool,
5960
fastdds::dds::DomainParticipant* participant,
60-
fastdds::dds::Topic* topic_entity);
61+
fastdds::dds::Topic* topic_entity,
62+
const bool repeater = false);
6163

6264
/**
6365
* @brief Destroy the MultiWriter object
@@ -99,6 +101,7 @@ class MultiWriter : public BaseWriter
99101

100102
fastdds::dds::DomainParticipant* dds_participant_;
101103
fastdds::dds::Topic* dds_topic_;
104+
bool repeater_;
102105

103106
/////////////////////////
104107
// INTERNAL VARIABLES

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,10 @@ class QoSSpecificWriter : public CommonWriter
3939
* @param participant_id Router Id of the Participant that has created this QoSSpecificWriter.
4040
* @param topic Topic that this QoSSpecificWriter subscribes to.
4141
* @param payload_pool Shared Payload Pool to received data and take it.
42+
* @param specific_qos Specific QoS of the Endpoint.
4243
* @param participant DDS Participant pointer.
4344
* @param topic_entity DDS Topic pointer.
45+
* @param repeater If this QoSSpecificWriter is a repeater or not.
4446
*
4547
* @throw \c InitializationException in case any creation has failed
4648
*/
@@ -51,7 +53,8 @@ class QoSSpecificWriter : public CommonWriter
5153
const std::shared_ptr<core::PayloadPool>& payload_pool,
5254
const core::types::SpecificEndpointQoS& specific_qos,
5355
fastdds::dds::DomainParticipant* participant,
54-
fastdds::dds::Topic* topic_entity);
56+
fastdds::dds::Topic* topic_entity,
57+
const bool repeater = false);
5558

5659
protected:
5760

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ class SimpleWriter : public CommonWriter
4343
* @param payload_pool Shared Payload Pool to received data and take it.
4444
* @param participant DDS Participant pointer.
4545
* @param topic_entity DDS Topic pointer.
46+
* @param repeater If this SimpleWriter is a repeater or not.
4647
*
4748
* @throw \c InitializationException in case any creation has failed
4849
*/
@@ -52,7 +53,8 @@ class SimpleWriter : public CommonWriter
5253
const core::types::DdsTopic& topic,
5354
const std::shared_ptr<core::PayloadPool>& payload_pool,
5455
fastdds::dds::DomainParticipant* participant,
55-
fastdds::dds::Topic* topic_entity);
56+
fastdds::dds::Topic* topic_entity,
57+
const bool repeater = false);
5658

5759
};
5860

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

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
// Copyright 2025 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
/**
16+
* @file RepeaterDataFilter.hpp
17+
*/
18+
19+
#pragma once
20+
21+
#include <fastdds/rtps/common/Guid.hpp>
22+
#include <fastdds/rtps/common/WriteParams.hpp>
23+
24+
#include <ddspipe_participants/library/library_dll.h>
25+
#include <ddspipe_participants/writer/dds/filter/SelfDataFilter.hpp>
26+
27+
namespace eprosima {
28+
namespace ddspipe {
29+
namespace participants {
30+
namespace dds {
31+
32+
/**
33+
* @brief Struct containing all relevant information for the \c RepeaterDataFilter.
34+
*
35+
* This structure extends the \c UserWriteData to include the GUID prefix of the last writer
36+
* that sent a change. It is used to track the origin of a message in repeater scenarios.
37+
*/
38+
struct RepeaterWriteData : public fastdds::rtps::WriteParams::UserWriteData
39+
{
40+
/// GUID of the writer that sent this change.
41+
fastdds::rtps::GuidPrefix_t last_writer_guid_prefix;
42+
};
43+
44+
/**
45+
* @brief Data filter to prevent sending messages back to their source participant.
46+
*
47+
* This filter is used by "repeater" participants to propagate information only to external participants,
48+
* i.e., participants not belonging to the same DDS-Router instance. It ensures that messages are not
49+
* sent to readers that belong to the participant from which the information was originally received.
50+
*/
51+
class RepeaterDataFilter : public SelfDataFilter
52+
{
53+
public:
54+
55+
/**
56+
* @brief Evaluates whether a message should be sent to a given reader.
57+
*
58+
* This method checks whether the reader belongs to the same participant from which the message originated.
59+
* I also checks whether the reader belongs to the same participant as the writer.
60+
*
61+
* @param payload The serialized payload of the message.
62+
* @param sample_info Information about the sample for content filtering.
63+
* @param reader_guid The GUID of the reader to which the message may be sent.
64+
* @return true if the reader does not belong to the participant that sent the message
65+
* and also does not belong to the same participant as the writer.
66+
* @return false otherwise.
67+
*/
68+
DDSPIPE_PARTICIPANTS_DllAPI
69+
bool evaluate(
70+
const fastdds::rtps::SerializedPayload_t& payload,
71+
const fastdds::dds::IContentFilter::FilterSampleInfo& sample_info,
72+
const fastdds::rtps::GUID_t& reader_guid) const override;
73+
};
74+
75+
} /* namespace dds */
76+
} /* namespace participants */
77+
} /* namespace ddspipe */
78+
} /* namespace eprosima */

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

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
// Copyright 2025 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
/**
16+
* @file SelfDataFilter.hpp
17+
*/
18+
19+
#pragma once
20+
21+
#include <fastdds/dds/topic/IContentFilter.hpp>
22+
23+
#include <fastdds/rtps/common/Guid.hpp>
24+
#include <fastdds/rtps/common/SerializedPayload.hpp>
25+
26+
#include <ddspipe_participants/library/library_dll.h>
27+
28+
namespace eprosima {
29+
namespace ddspipe {
30+
namespace participants {
31+
namespace dds {
32+
33+
/**
34+
* @brief Filter to prevent sending messages from this writer to readers in the same participant.
35+
*
36+
* This filter ensures that messages originating from this writer are not delivered
37+
* to readers that belong to the same participant, avoiding self-data delivery.
38+
*/
39+
class SelfDataFilter : public fastdds::dds::IContentFilter
40+
{
41+
public:
42+
43+
/**
44+
* @brief Evaluates whether a message should be sent to a given reader.
45+
*
46+
* This method checks whether the reader belongs to the same participant as the writer.
47+
*
48+
* @param payload The serialized payload of the message.
49+
* @param sample_info Information about the sample for content filtering.
50+
* @param reader_guid The GUID of the reader to which the message may be sent.
51+
* @return true if the reader does not belong to the same participant.
52+
* @return false otherwise.
53+
*/
54+
DDSPIPE_PARTICIPANTS_DllAPI
55+
bool evaluate(
56+
const fastdds::rtps::SerializedPayload_t& payload,
57+
const fastdds::dds::IContentFilter::FilterSampleInfo& sample_info,
58+
const fastdds::rtps::GUID_t& reader_guid) const override;
59+
};
60+
61+
} /* namespace dds */
62+
} /* namespace participants */
63+
} /* namespace ddspipe */
64+
} /* namespace eprosima */

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,7 +193,7 @@ class CommonWriter : public BaseWriter, public fastdds::rtps::WriterListener
193193
core::IRoutingData& data) noexcept override;
194194

195195
/**
196-
* @brief Auxiliary method used in \c write to fill the cache change to send.
196+
* @brief Auxiliary method used in \c write to fill the cache change to send and write params.
197197
*
198198
* @param [out] to_send_change_to_fill cache change to be filled and sent.
199199
* @param [out] to_send_params write params to be filled and sent.

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ bool CommonParticipant::is_rtps_kind() const noexcept
105105

106106
bool CommonParticipant::is_repeater() const noexcept
107107
{
108-
return false;
108+
return configuration_->is_repeater;
109109
}
110110

111111
core::types::TopicQoS CommonParticipant::topic_qos() const noexcept
@@ -143,7 +143,8 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
143143
dds_topic,
144144
this->payload_pool_,
145145
dds_participant_,
146-
fastdds_topic);
146+
fastdds_topic,
147+
configuration_->is_repeater);
147148
}
148149
else
149150
{
@@ -152,7 +153,8 @@ std::shared_ptr<core::IWriter> CommonParticipant::create_writer(
152153
dds_topic,
153154
this->payload_pool_,
154155
dds_participant_,
155-
fastdds_topic);
156+
fastdds_topic,
157+
configuration_->is_repeater);
156158
writer->init();
157159

158160
return writer;

0 commit comments

Comments
 (0)