Skip to content

Commit d48fd08

Browse files
DanipizaDanipiza
authored
Support "partitions" in DDS-Record-Replay & Fast-DDS-spy
* [#23594] First approach Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Record Partition in SQL with new database Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Replay with Partitions (SQL) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] SQL last changes in Recorder and Replayer Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Cleaned and optimizated code Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Added new yaml partition filter in recorder and replayer Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Optimized and cleaned code Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated recorder partition filter (empty partitions) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Removed unnecessary listener constructor Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23686] Updated code for fastddsspy partitions filter Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Deleted unnecessary partition_allowed_list in DDSPipe class Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23686] Added code for reseting the DataReader with filter command Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23686] Updated DDS-Pipe and Fast-DDS-spy to spy information with partitions filter Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23686] Updated and cleaned code for Spy Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Added partitions filter reload with .yaml configuration update Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated code (uncrustify) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated code (uncrustify) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated code (uncrustify) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated code (uncrustify) Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated code with review suggestions Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Recovered 'CommonParticipant* parent_class_' pointer first implementation Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Recovered 'CommonParticipant* parent_class_' pointer first implementation Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Updated Multiwriter creation Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#23594] Add final revision suggestions Signed-off-by: danipiza <dpizarrogallego@gmail.com> --------- Signed-off-by: danipiza <dpizarrogallego@gmail.com>
1 parent 1dce5c9 commit d48fd08

36 files changed

Lines changed: 1376 additions & 30 deletions

‎ddspipe_core/include/ddspipe_core/communication/dds/DdsBridge.hpp‎

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,9 @@ class DdsBridge : public Bridge
4949
* @param participant_database: Collection of Participants to manage communication
5050
* @param payload_pool: Payload Pool that handles the reservation/release of payloads throughout the DDS Router
5151
* @param thread_pool: Shared pool of threads in charge of data transmission.
52-
* @param enable: Whether the Bridge should be initialized as enabled
52+
* @param routes_config: Configuration encapsulating the routes of a DdsPipe instance.
53+
* @param remove_unused_entities: Flag for removing unused entitites in the Bridge.
54+
* @param manual_topics: List of topics of the Bridge.
5355
*
5456
* @throw InitializationException in case \c IWriters or \c IReaders creation fails.
5557
*/
@@ -63,6 +65,35 @@ class DdsBridge : public Bridge
6365
const bool remove_unused_entities,
6466
const std::vector<core::types::ManualTopic>& manual_topics);
6567

68+
/**
69+
* Bridge constructor with partition filter.
70+
*
71+
* In Bridge construction, the inside \c Tracks are created.
72+
* In Bridge construction, a Writer and a Reader are created for each Participant.
73+
*
74+
* @param topic: Topic of which this Bridge manages communication
75+
* @param participant_database: Collection of Participants to manage communication
76+
* @param payload_pool: Payload Pool that handles the reservation/release of payloads throughout the DDS Router
77+
* @param thread_pool: Shared pool of threads in charge of data transmission.
78+
* @param routes_config: Configuration encapsulating the routes of a DdsPipe instance.
79+
* @param remove_unused_entities: Flag for removing unused entitites in the Bridge.
80+
* @param manual_topics: List of topics of the Bridge.
81+
* @param filter_partition: Set of partitions used in the filter of the Bridge.
82+
*
83+
* @throw InitializationException in case \c IWriters or \c IReaders creation fails.
84+
*/
85+
DDSPIPE_CORE_DllAPI
86+
DdsBridge(
87+
const utils::Heritable<types::DistributedTopic>& topic,
88+
const std::shared_ptr<ParticipantsDatabase>& participants_database,
89+
const std::shared_ptr<PayloadPool>& payload_pool,
90+
const std::shared_ptr<utils::SlotThreadPool>& thread_pool,
91+
const RoutesConfiguration& routes_config,
92+
const bool remove_unused_entities,
93+
const std::vector<core::types::ManualTopic>& manual_topics,
94+
const std::set<std::string> filter_partition);
95+
96+
6697
DDSPIPE_CORE_DllAPI
6798
~DdsBridge();
6899

@@ -110,6 +141,19 @@ class DdsBridge : public Bridge
110141
void remove_writer(
111142
const types::ParticipantId& participant_id) noexcept;
112143

144+
DDSPIPE_CORE_DllAPI
145+
void update_readers_track(
146+
const std::set<std::string> filter_partition_set);
147+
148+
DDSPIPE_CORE_DllAPI
149+
void add_filter_partition(
150+
const std::set<std::string> filter_partition_);
151+
152+
DDSPIPE_CORE_DllAPI
153+
void add_partition_to_topic(
154+
std::string guid,
155+
std::string partition);
156+
113157
protected:
114158

115159
/**
@@ -122,6 +166,18 @@ class DdsBridge : public Bridge
122166
DDSPIPE_CORE_DllAPI
123167
void create_all_tracks_();
124168

169+
/**
170+
* Create the readers, writers, and tracks that are required by the routes and
171+
* satisfy the filter.
172+
*
173+
* Thread safe
174+
*
175+
* @throw InitializationException in case \c IWriters or \c IReaders creation fails.
176+
*/
177+
DDSPIPE_CORE_DllAPI
178+
void create_all_tracks_with_filter(
179+
const std::set<std::string> filter_partition_set);
180+
125181
/**
126182
* Add each Participant's IWriters to its Track.
127183
* If the Participant's IReader doesn't exist, create it.
@@ -149,6 +205,20 @@ class DdsBridge : public Bridge
149205
void add_writers_to_tracks_nts_(
150206
std::map<types::ParticipantId, std::shared_ptr<IWriter>>& writers);
151207

208+
/**
209+
* Add each Participant's IWriters to its Track with filter.
210+
* If the Participant's IReader doesn't exist, create it.
211+
* If the Participant's Track doesn't exist, create it.
212+
*
213+
* @param writers: The map of ids to writers that are required for the tracks.
214+
*
215+
* @throw InitializationException in case \c IReaders creation fails.
216+
*/
217+
DDSPIPE_CORE_DllAPI
218+
void add_writers_to_tracks_with_filter_nts_(
219+
std::map<types::ParticipantId, std::shared_ptr<IWriter>>& writers,
220+
const std::set<std::string> filter_partition_set);
221+
152222
/**
153223
* @brief Impose the Topic QoS that have been pre-configured for a participant.
154224
*
@@ -171,6 +241,9 @@ class DdsBridge : public Bridge
171241
//! Topics that explicitally set a QoS attribute for this participant.
172242
std::vector<types::ManualTopic> manual_topics_;
173243

244+
//! Allowed partitions list added in the filter.
245+
std::set<std::string> filter_partition_;
246+
174247
/**
175248
* Inside \c Tracks
176249
* They are indexed by the Id of the participant that is source

‎ddspipe_core/include/ddspipe_core/communication/dds/Track.hpp‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,9 @@ class Track
134134
DDSPIPE_CORE_DllAPI
135135
bool has_writers() noexcept;
136136

137+
DDSPIPE_CORE_DllAPI
138+
void update_reader();
139+
137140
protected:
138141

139142
/*

‎ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,20 @@ class DdsPipe
9898
utils::ReturnCode reload_configuration(
9999
const DdsPipeConfiguration& new_configuration);
100100

101+
/**
102+
* @brief Reload the filter partitions set.
103+
*
104+
* @param [in] new_filter_partition_set : new filter partition set.
105+
*
106+
* @return \c RETCODE_OK if the new std::set has been updated correctly.
107+
* @return \c RETCODE_ERROR if any other error has occurred.
108+
*
109+
* @note This method calls \c update_readers_track() of all Topic bridges, to update the DataReaders.
110+
*/
111+
DDSPIPE_CORE_DllAPI
112+
void reload_filter_partition(
113+
const std::set<std::string> new_filter_partition_set);
114+
101115
/////////////////////////
102116
// ENABLING METHODS
103117
/////////////////////////
@@ -126,6 +140,15 @@ class DdsPipe
126140
DDSPIPE_CORE_DllAPI
127141
utils::ReturnCode disable() noexcept;
128142

143+
DDSPIPE_CORE_DllAPI
144+
void update_readers_track(
145+
const std::string topic_name,
146+
const std::set<std::string> filter_partition_set);
147+
148+
DDSPIPE_CORE_DllAPI
149+
void update_filter(
150+
const std::set<std::string> filter_partition_set);
151+
129152
protected:
130153

131154
/////////////////////////
@@ -386,6 +409,9 @@ class DdsPipe
386409
//! Thread Pool for tracks
387410
std::shared_ptr<utils::SlotThreadPool> thread_pool_;
388411

412+
//! Allowed partitions added in the filter
413+
std::set<std::string> allowed_partition_list_;
414+
389415
/////////////////////////
390416
// INTERNAL DATA STORAGE
391417
/////////////////////////
@@ -419,6 +445,9 @@ class DdsPipe
419445
//! Whether the DdsPipe is currently communicating data or not
420446
bool enabled_;
421447

448+
//! Allowed partitions list added in the filter.
449+
std::set<std::string> filter_partition_;
450+
422451
/**
423452
* @brief Internal mutex for concurrent calls
424453
*/

‎ddspipe_core/include/ddspipe_core/dynamic/DiscoveryDatabase.hpp‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,22 @@ class DiscoveryDatabase
102102
bool endpoint_exists(
103103
const types::Guid& guid) const noexcept;
104104

105+
/**
106+
* @brief Insert guid to the filter set
107+
*
108+
* This method stores the guids that dont pass the partition filter.
109+
*
110+
* @param [in] new_endpoint: new endpoint to store
111+
*/
112+
DDSPIPE_CORE_DllAPI
113+
void add_filtered_endpoint(
114+
const types::Guid guid);
115+
116+
DDSPIPE_CORE_DllAPI
117+
bool exists_filtered_endpoint(
118+
const types::Guid endpoint_guid);
119+
120+
105121
/**
106122
* @brief Insert endpoint to the database
107123
*
@@ -247,6 +263,9 @@ class DiscoveryDatabase
247263
//! Database of endpoints indexed by guid
248264
std::map<types::Guid, types::Endpoint> entities_;
249265

266+
//! Database of endpoints indexed by guid
267+
std::set<types::Guid> entities_filter_;
268+
250269
//! Mutex to guard queries to the database
251270
mutable std::shared_timed_mutex mutex_;
252271

‎ddspipe_core/include/ddspipe_core/interface/IParticipant.hpp‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,10 @@ class IParticipant
6060
DDSPIPE_CORE_DllAPI
6161
virtual types::TopicQoS topic_qos() const noexcept = 0;
6262

63+
//! The Participant's Topic Partitions.
64+
DDSPIPE_CORE_DllAPI
65+
virtual std::map<std::string, std::map<std::string, std::string>> topic_partitions() const noexcept = 0;
66+
6367
/**
6468
* @brief Return a new Writer
6569
*
@@ -91,6 +95,75 @@ class IParticipant
9195
DDSPIPE_CORE_DllAPI
9296
virtual std::shared_ptr<IReader> create_reader(
9397
const ITopic& topic) = 0;
98+
99+
/**
100+
* @brief Return a new Reader
101+
*
102+
* Each reader is associated with a \c Bridge with the topic \c topic .
103+
* This reader will receive messages in this topic.
104+
*
105+
* @param [in] topic : Topic that this Reader will work with.
106+
* @param [in] partitions: List of allowed partitions
107+
*
108+
* @return Reader in this Participant referring this topic
109+
*
110+
* @throw \c InitializationException in case the reader creation fails.
111+
*/
112+
DDSPIPE_CORE_DllAPI
113+
virtual std::shared_ptr<IReader> create_reader_with_filter(
114+
const ITopic& topic,
115+
const std::set<std::string> partitions) = 0;
116+
117+
/**
118+
* Add a Partition in the Participant.
119+
*
120+
* @param [in] topic_name : The topic.
121+
* @param [in] writer_guid : The writer guid.
122+
* @param [in] partition : Partition name that will be added.
123+
*
124+
* @return bool if the function adds the partition.
125+
*/
126+
DDSPIPE_CORE_DllAPI
127+
virtual bool add_topic_partition(
128+
const std::string& topic_name,
129+
const std::string& writer_guid,
130+
const std::string& partition) = 0;
131+
132+
/**
133+
* Update a Partition in the Participant.
134+
*
135+
* @param [in] topic_name : The topic.
136+
* @param [in] writer_guid : The writer guid.
137+
* @param [in] partition : Partition name that will be added.
138+
*
139+
* @return bool if the function updates the partition.
140+
*/
141+
DDSPIPE_CORE_DllAPI
142+
virtual bool update_topic_partition(
143+
const std::string& topic_name,
144+
const std::string& writer_guid,
145+
const std::string& partition) = 0;
146+
147+
/**
148+
* Remove a Partition of the Participant.
149+
*
150+
* @param [in] topic_name : The topic.
151+
* @param [in] writer_guid : The writer guid.
152+
* @param [in] partition : Partition name that will be added.
153+
*
154+
* @return bool if the function deletes the partition.
155+
*/
156+
DDSPIPE_CORE_DllAPI
157+
virtual bool delete_topic_partition(
158+
const std::string& topic_name,
159+
const std::string& writer_guid,
160+
const std::string& partition) = 0;
161+
162+
/**
163+
* Clear the Partitions of the Participant.
164+
*/
165+
DDSPIPE_CORE_DllAPI
166+
virtual void clear_topic_partitions() = 0;
94167
};
95168

96169
} /* namespace core */

‎ddspipe_core/include/ddspipe_core/types/dds/Endpoint.hpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,10 @@ struct Endpoint
5858
DDSPIPE_CORE_DllAPI
5959
TopicQoS topic_qos() const noexcept;
6060

61+
//! Partition of the topic
62+
DDSPIPE_CORE_DllAPI
63+
std::map<std::string, std::string> topic_partitions() const noexcept;
64+
6165
//! Whether the endpoint is a writer
6266
DDSPIPE_CORE_DllAPI
6367
bool is_writer() const noexcept;
@@ -87,6 +91,9 @@ struct Endpoint
8791
//! Specific QoS of the entity
8892
SpecificEndpointQoS specific_qos {};
8993

94+
//! Specific Partitions of the endpoint
95+
std::map<std::string, std::string> specific_partitions;
96+
9097
//! Whether the endpoint is currently active
9198
bool active {true};
9299

‎ddspipe_core/include/ddspipe_core/types/topic/Topic.hpp‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,19 @@ struct Topic : public ITopic, public IConfiguration
134134
* If the Topic has manually configured Topic QoS, the Topic QoS that are manually configured get overriden.
135135
*/
136136
types::TopicQoS topic_qos{};
137+
138+
/**
139+
* @brief The partitions set of the Topic.
140+
*
141+
* The data is stored with the following pair: <writer_guid, partition_set>
142+
*
143+
* e.g.: In ShapeDemos there are the following partitions ("", "A", "B", "C", "D", "*")
144+
* and a valid partitions sets could be:
145+
* - "A"
146+
* - "A|B" # the writer writes in the two partitions.
147+
* - "*"
148+
*/
149+
std::map<std::string, std::string> partition_name{};
137150
};
138151

139152
/**

0 commit comments

Comments
 (0)