Update DDS-Pipe filters (#174)

* [#24061] Updated filters to update 'DDS' spy readers instead of reseting them

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Removed unused functions from the previous filter implementation + 'is_fastddsspy' bool

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Updated code for DDS-Record-Replay filter

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Changed 'enable()' callback after all filter updates

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Added RTPS filter partition + changed participant default to DDS

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Refactor of new filter functions in IParticipant

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Removed unused functions in IParticipant

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Cleaned code + added FastDD-Spy filter yaml configuration

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Added support partitions in replayer using DDS

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Uncrustify

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Removed unnecessary mutex in DdsPipe

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] First approach for 'update_topic_partition()' in writers

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Finished IWriter 'update_topic_partitions()' implementation

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* Pass by reference and const instead of value

Signed-off-by: Raül <raulojeda@eprosima.com>

* Uncrustify

Signed-off-by: Raül <raulojeda@eprosima.com>

* More uncrustify

Signed-off-by: Raül <raulojeda@eprosima.com>

* [#24061] Applied revision

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Added filter functions to 'writers'

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Removed lock in DdsPipe + separated 'update_filters()'

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Recovered lock in DdsPipe + solved error in Windows

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Solved TSAN race conditions + removed 'bridges_mutex_'

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Removed non-standard DDS writer content_filteredtopic functions

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

* [#24061] Applied revision

Signed-off-by: danipiza <dpizarrogallego@gmail.com>

---------

Signed-off-by: danipiza <dpizarrogallego@gmail.com>
Signed-off-by: Raül <raulojeda@eprosima.com>
Co-authored-by: Raül <raulojeda@eprosima.com>

Author: Danipiza

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

@@ -65,34 +65,6 @@ class DdsBridge : public Bridge
             const bool remove_unused_entities,
             const std::vector<core::types::ManualTopic>& manual_topics);
 
-    /**
-     * Bridge constructor with partition filter.
-     *
-     * In Bridge construction, the inside \c Tracks are created.
-     * In Bridge construction, a Writer and a Reader are created for each Participant.
-     *
-     * @param topic: Topic of which this Bridge manages communication
-     * @param participant_database: Collection of Participants to manage communication
-     * @param payload_pool: Payload Pool that handles the reservation/release of payloads throughout the DDS Router
-     * @param thread_pool: Shared pool of threads in charge of data transmission.
-     * @param routes_config: Configuration encapsulating the routes of a DdsPipe instance.
-     * @param remove_unused_entities: Flag for removing unused entitites in the Bridge.
-     * @param manual_topics: List of topics of the Bridge.
-     * @param filter_partition: Set of partitions used in the filter of the Bridge.
-     *
-     * @throw InitializationException in case \c IWriters or \c IReaders creation fails.
-     */
-    DDSPIPE_CORE_DllAPI
-    DdsBridge(
-            const utils::Heritable<types::DistributedTopic>& topic,
-            const std::shared_ptr<ParticipantsDatabase>& participants_database,
-            const std::shared_ptr<PayloadPool>& payload_pool,
-            const std::shared_ptr<utils::SlotThreadPool>& thread_pool,
-            const RoutesConfiguration& routes_config,
-            const bool remove_unused_entities,
-            const std::vector<core::types::ManualTopic>& manual_topics,
-            const std::set<std::string> filter_partition);
-
 
     DDSPIPE_CORE_DllAPI
     ~DdsBridge();
@@ -141,42 +113,46 @@ class DdsBridge : public Bridge
     void remove_writer(
             const types::ParticipantId& participant_id) noexcept;
 
-    DDSPIPE_CORE_DllAPI
-    void update_readers_track(
-            const std::set<std::string> filter_partition_set);
-
-    DDSPIPE_CORE_DllAPI
-    void add_filter_partition(
-            const std::set<std::string> filter_partition_);
-
+    /**
+     * Adds a partition with its respective guid.
+     *
+     * @param guid: Guid of the endpoint.
+     * @param partition: Partition that is going to be added.
+     */
     DDSPIPE_CORE_DllAPI
     void add_partition_to_topic(
             std::string guid,
             std::string partition);
 
-protected:
-
     /**
-     * Create the readers, writers, and tracks that are required by the routes.
+     * Function that updates the partition set of all bridges.
      *
-     * Thread safe
+     * @param partitions_set: Set of partitions
+     */
+    DDSPIPE_CORE_DllAPI
+    void update_partitions(
+            const std::set<std::string>& partitions_set);
+
+    /**
+     * Function that updates the content filtered topic of all bridges
      *
-     * @throw InitializationException in case \c IWriters or \c IReaders creation fails.
+     * @param expression: The expression of the filter.
      */
     DDSPIPE_CORE_DllAPI
-    void create_all_tracks_();
+    void update_topic_filter(
+            const std::string& expression);
+
+protected:
 
     /**
-     * Create the readers, writers, and tracks that are required by the routes and
-     * satisfy the filter.
+     * Create the readers, writers, and tracks that are required by the routes.
      *
      * Thread safe
      *
      * @throw InitializationException in case \c IWriters or \c IReaders creation fails.
      */
     DDSPIPE_CORE_DllAPI
-    void create_all_tracks_with_filter(
-            const std::set<std::string> filter_partition_set);
+    void create_all_tracks_();
 
     /**
      * Add each Participant's IWriters to its Track.
@@ -205,20 +181,6 @@ class DdsBridge : public Bridge
     void add_writers_to_tracks_nts_(
             std::map<types::ParticipantId, std::shared_ptr<IWriter>>& writers);
 
-    /**
-     * Add each Participant's IWriters to its Track with filter.
-     * If the Participant's IReader doesn't exist, create it.
-     * If the Participant's Track doesn't exist, create it.
-     *
-     * @param writers: The map of ids to writers that are required for the tracks.
-     *
-     * @throw InitializationException in case \c IReaders creation fails.
-     */
-    DDSPIPE_CORE_DllAPI
-    void add_writers_to_tracks_with_filter_nts_(
-            std::map<types::ParticipantId, std::shared_ptr<IWriter>>& writers,
-            const std::set<std::string> filter_partition_set);
-
     /**
      * @brief Impose the Topic QoS that have been pre-configured for a participant.
      *

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

@@ -137,6 +137,43 @@ class Track
     DDSPIPE_CORE_DllAPI
     void update_reader();
 
+    /**
+     * Update the writer's topic partitions
+     *
+     * @param partition_name Dictionary of <participant_id, partition name>.
+     *
+     * Tread safe
+     */
+    DDSPIPE_CORE_DllAPI
+    void update_writers_topic_partitions(
+            const std::map<std::string, std::string>& partition_name);
+
+    /**
+     * Update the partitions configured in the track reader.
+     *
+     * The reader is disabled before applying the new partitions
+     * and enabled again afterwards to ensure the reader receive data
+     * with the correct partitions.
+     *
+     * @param partitions_set Set of partitions to configure in the reader.
+     */
+    DDSPIPE_CORE_DllAPI
+    void update_reader_partitions(
+            const std::set<std::string>& partitions_set);
+
+    /**
+     * Update the content filter expression configured in the track reader.
+     *
+     * The reader is disabled before applying the new expression
+     * and enabled again afterwards to ensure the reader receive data
+     * with the correct topic filter.
+     *
+     * @param expression SQL-like content filter expression to apply.
+     */
+    DDSPIPE_CORE_DllAPI
+    void update_reader_content_filter(
+            const std::string& expression);
+
 protected:
 
     /*

ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp

@@ -98,20 +98,6 @@ class DdsPipe
     utils::ReturnCode reload_configuration(
             const DdsPipeConfiguration& new_configuration);
 
-    /**
-     * @brief Reload the filter partitions set.
-     *
-     * @param [in] new_filter_partition_set : new filter partition set.
-     *
-     * @return \c RETCODE_OK if the new std::set has been updated correctly.
-     * @return \c RETCODE_ERROR if any other error has occurred.
-     *
-     * @note This method calls \c update_readers_track() of all Topic bridges, to update the DataReaders.
-     */
-    DDSPIPE_CORE_DllAPI
-    void reload_filter_partition(
-            const std::set<std::string> new_filter_partition_set);
-
     /////////////////////////
     // ENABLING METHODS
     /////////////////////////
@@ -140,14 +126,38 @@ class DdsPipe
     DDSPIPE_CORE_DllAPI
     utils::ReturnCode disable() noexcept;
 
+    /**
+     * @brief Update the stored partitions filter.
+     *
+     * This method replaces the current partitions filter used by the pipe.
+     *
+     * @param [in] filter_partition_set : new partitions filter.
+     */
     DDSPIPE_CORE_DllAPI
-    void update_readers_track(
-            const std::string topic_name,
+    void update_filter(
             const std::set<std::string> filter_partition_set);
 
+    /**
+     * @brief Update the partitions in every DDS topic bridge.
+     *
+     * @param [in] partitions_set : set of partitions to apply.
+     */
+    DDSPIPE_CORE_DllAPI
+    void update_partitions(
+            const std::set<std::string>& partitions_set);
+
+    /**
+     * @brief Update the content filter expression for one topic.
+     *
+     * The expression is applied to bridges whose topic name matches \c topic_name.
+     *
+     * @param [in] topic_name : topic where the filter must be updated.
+     * @param [in] expression : new content filter expression.
+     */
     DDSPIPE_CORE_DllAPI
-    void update_filter(
-            const std::set<std::string> filter_partition_set);
+    void update_content_filter(
+            const std::string& topic_name,
+            const std::string& expression);
 
 protected:
 
@@ -369,6 +379,14 @@ class DdsPipe
      */
     void deactivate_all_topics_nts_() noexcept;
 
+    /**
+     * @brief Update partitions for all bridges.
+     *
+     * This method must be called with \c mutex_ locked.
+     */
+    void update_partitions_nts_(
+            const std::set<std::string>& partitions_set);
+
     //////////////////////////
     // CONFIGURATION VARIABLES
     //////////////////////////
@@ -409,9 +427,6 @@ class DdsPipe
     //! Thread Pool for tracks
     std::shared_ptr<utils::SlotThreadPool> thread_pool_;
 
-    //! Allowed partitions added in the filter
-    std::set<std::string> allowed_partition_list_;
-
     /////////////////////////
     // INTERNAL DATA STORAGE
     /////////////////////////
@@ -452,11 +467,6 @@ class DdsPipe
      * @brief Internal mutex for concurrent calls
      */
     mutable std::mutex mutex_;
-
-    /**
-     * @brief Internal mutex for bridges
-     */
-    mutable std::mutex bridges_mutex_;
 };
 
 } /* namespace core */

ddspipe_core/include/ddspipe_core/interface/IParticipant.hpp

@@ -96,24 +96,6 @@ class IParticipant
     virtual std::shared_ptr<IReader> create_reader(
             const ITopic& topic) = 0;
 
-    /**
-     * @brief Return a new Reader
-     *
-     * Each reader is associated with a \c Bridge with the topic \c topic .
-     * This reader will receive messages in this topic.
-     *
-     * @param [in] topic : Topic that this Reader will work with.
-     * @param [in] partitions: List of allowed partitions
-     *
-     * @return Reader in this Participant referring this topic
-     *
-     * @throw \c InitializationException in case the reader creation fails.
-     */
-    DDSPIPE_CORE_DllAPI
-    virtual std::shared_ptr<IReader> create_reader_with_filter(
-            const ITopic& topic,
-            const std::set<std::string> partitions) = 0;
-
     /**
      * Add a Partition in the Participant.
      *
@@ -164,6 +146,27 @@ class IParticipant
      */
     DDSPIPE_CORE_DllAPI
     virtual void clear_topic_partitions() = 0;
+
+    /**
+     * Update the partition filter data structure of the participant
+     *
+     * @param [in] partitions : Set of partitions for the new filter of partitions.
+     */
+    DDSPIPE_CORE_DllAPI
+    virtual void update_partitions(
+            std::set<std::string> partitions) = 0;
+
+    /**
+     * Update the content_topicfilter data structure of the participant
+     *
+     * @param [in] topic_name : Name of the topic.
+     * @param [in] expression : Expression for the Content Filtered Topic
+     */
+    DDSPIPE_CORE_DllAPI
+    virtual void update_content_topicfilter(
+            const std::string& topic_name,
+            const std::string& expression) = 0;
+
 };
 
 } /* namespace core */

ddspipe_core/include/ddspipe_core/interface/IReader.hpp

@@ -130,6 +130,16 @@ class IReader
     DDSPIPE_CORE_DllAPI
     virtual types::DdsTopic topic() const = 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;
+
     DDSPIPE_CORE_DllAPI
     virtual types::ParticipantId participant_id() const = 0;
     /////////////////////////

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>
@@ -82,6 +84,16 @@ class IWriter
     DDSPIPE_CORE_DllAPI
     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 topic_partitions
+    DDSPIPE_CORE_DllAPI
+    virtual void update_topic_partitions(
+            const std::map<std::string, std::string>& partition_name) = 0;
 };
 
 } /* namespace core */

ddspipe_core/include/ddspipe_core/types/topic/filter/WildcardDdsFilterTopic.hpp

@@ -80,6 +80,9 @@ struct WildcardDdsFilterTopic : public IFilterTopic
     //! The subset of manually configured Topic QoS.
     utils::Fuzzy<types::TopicQoS> topic_qos;
 
+    //! Topic filter. If not set all messages are sent/received.
+    utils::Fuzzy<std::string> content_topic_filter;
+
 protected:
 
     /////////////////////////