Manually fix a set of Topic QoS, and implement the max-tx-rate, max-rx-rate, and downsampling

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

@@ -20,6 +20,8 @@
 #include <ddspipe_core/communication/dds/Track.hpp>
 #include <ddspipe_core/configuration/RoutesConfiguration.hpp>
 #include <ddspipe_core/types/topic/dds/DistributedTopic.hpp>
+#include <ddspipe_core/types/topic/filter/ManualTopic.hpp>
+#include <ddspipe_core/types/topic/filter/WildcardDdsFilterTopic.hpp>
 
 namespace eprosima {
 namespace ddspipe {
@@ -58,7 +60,8 @@ class DdsBridge : public Bridge
             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 bool remove_unused_entities,
+            const std::vector<core::types::ManualTopic>& manual_topics);
 
     DDSPIPE_CORE_DllAPI
     ~DdsBridge();
@@ -124,8 +127,6 @@ class DdsBridge : public Bridge
      * If the Participant's IReader doesn't exist, create it.
      * If the Participant's Track doesn't exist, create it.
      *
-     * Thread safe
-     *
      * @param writers: The map of ids to writers that are required for the tracks.
      *
      * @throw InitializationException in case \c IReaders creation fails.
@@ -140,8 +141,6 @@ class DdsBridge : public Bridge
      * If the Participant's IReader doesn't exist, create it.
      * If the Participant's Track doesn't exist, create it.
      *
-     * Thread safe
-     *
      * @param writers: The map of ids to writers that are required for the tracks.
      *
      * @throw InitializationException in case \c IReaders creation fails.
@@ -150,10 +149,28 @@ class DdsBridge : public Bridge
     void add_writers_to_tracks_nts_(
             std::map<types::ParticipantId, std::shared_ptr<IWriter>>& writers);
 
+    /**
+     * @brief Impose the Topic QoS that have been pre-configured for a participant.
+     *
+     * First, it imposes the Topic QoS configured at \c manual_topics and then the ones configured at \c participants.
+     */
+    DDSPIPE_CORE_DllAPI
+    utils::Heritable<types::DistributedTopic> create_topic_for_participant_nts_(
+            const std::shared_ptr<IParticipant>& participant) noexcept;
+
+    /////////////////////////
+    // VARIABLES
+    /////////////////////////
+
+    //! Topic associated to the DdsBridge.
     utils::Heritable<types::DistributedTopic> topic_;
 
+    //! Routes associated to the Topic.
     RoutesConfiguration::RoutesMap routes_;
 
+    //! Topics that explicitally set a QoS attribute for this participant.
+    std::vector<types::ManualTopic> manual_topics_;
+
     /**
      * Inside \c Tracks
      * They are indexed by the Id of the participant that is source

ddspipe_core/include/ddspipe_core/configuration/DdsPipeConfiguration.hpp

@@ -24,6 +24,7 @@
 #include <ddspipe_core/configuration/TopicRoutesConfiguration.hpp>
 #include <ddspipe_core/types/participant/ParticipantId.hpp>
 #include <ddspipe_core/types/topic/dds/DistributedTopic.hpp>
+#include <ddspipe_core/types/topic/filter/ManualTopic.hpp>
 
 #include <ddspipe_core/library/library_dll.h>
 
@@ -73,10 +74,29 @@ struct DdsPipeConfiguration : public IConfiguration
     RoutesConfiguration get_routes_config(
             const utils::Heritable<types::DistributedTopic>& topic) const noexcept;
 
+    /**
+     * @brief Select the \c manual_topics for a topic.
+     *
+     * @return The manual topics for a specific topic.
+     */
+    DDSPIPE_CORE_DllAPI
+    std::vector<core::types::ManualTopic> get_manual_topics(
+            const core::ITopic& topic) const noexcept;
+
     /////////////////////////
     // VARIABLES
     /////////////////////////
 
+    //! Topic lists to build the AllowedTopics
+    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> allowlist{};
+    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> blocklist{};
+
+    //! Builtin topics to create at the beggining of the execution
+    std::set<utils::Heritable<ddspipe::core::types::DistributedTopic>> builtin_topics{};
+
+    //! Set of manually configured Topic QoS
+    std::vector<ddspipe::core::types::ManualTopic> manual_topics{};
+
     //! Configuration of the generic routes.
     RoutesConfiguration routes{};
 
@@ -85,6 +105,9 @@ struct DdsPipeConfiguration : public IConfiguration
 
     //! Whether entities should be removed when they have no writers connected to them.
     bool remove_unused_entities = false;
+
+    //! Whether the DDS Pipe should be initialized enabled.
+    bool init_enabled = false;
 };
 
 } /* namespace core */

ddspipe_core/include/ddspipe_core/core/DdsPipe.hpp

@@ -61,14 +61,11 @@ class DdsPipe
      */
     DDSPIPE_CORE_DllAPI
     DdsPipe(
-            const std::shared_ptr<AllowedTopicList>& allowed_topics,
+            const DdsPipeConfiguration& configuration,
             const std::shared_ptr<DiscoveryDatabase>& discovery_database,
             const std::shared_ptr<PayloadPool>& payload_pool,
             const std::shared_ptr<ParticipantsDatabase>& participants_database,
-            const std::shared_ptr<utils::SlotThreadPool>& thread_pool,
-            const std::set<utils::Heritable<types::DistributedTopic>>& builtin_topics = {},
-            bool start_enable = false,
-            const DdsPipeConfiguration& configuration = {});
+            const std::shared_ptr<utils::SlotThreadPool>& thread_pool);
 
     /**
      * @brief Destroy the DdsPipe object
@@ -85,19 +82,21 @@ class DdsPipe
     /////////////////////////
 
     /**
-     * @brief Reload the allowed topic configuration
+     * @brief Reload the DdsPipe configuration.
      *
-     * @param [in] configuration : new configuration
+     * @param [in] new_configuration : new configuration.
      *
-     * @return \c RETCODE_OK if configuration has been updated correctly
-     * @return \c RETCODE_NO_DATA if new configuration has not changed
-     * @return \c RETCODE_ERROR if any other error has occurred
+     * @return \c RETCODE_OK if the configuration has been updated correctly.
+     * @return \c RETCODE_NO_DATA if the new configuration has not changed.
+     * @return \c RETCODE_ERROR if any other error has occurred.
      *
-     * @throw \c ConfigurationException in case the new yaml is not well-formed
+     * @note This method checks that the new configuration file is valid and calls \c reload_allowed_topics_
+     *
+     * @throw \c ConfigurationException in case the new yaml is not well-formed.
      */
     DDSPIPE_CORE_DllAPI
-    utils::ReturnCode reload_allowed_topics(
-            const std::shared_ptr<AllowedTopicList>& allowed_topics);
+    utils::ReturnCode reload_configuration(
+            const DdsPipeConfiguration& new_configuration);
 
     /////////////////////////
     // ENABLING METHODS
@@ -129,6 +128,28 @@ class DdsPipe
 
 protected:
 
+    /////////////////////////
+    // INTERACTION METHODS
+    /////////////////////////
+    /**
+     * @brief Load allowed topics from configuration
+     *
+     * @throw \c ConfigurationException in case the yaml inside allowlist is not well-formed
+     */
+    void init_allowed_topics_();
+
+    /**
+     * @brief Reload the allowed topics configuration.
+     *
+     * @param [in] allowed_topics : new allowed topics.
+     *
+     * @return \c RETCODE_OK if the allowed topics have been updated correctly.
+     * @return \c RETCODE_NO_DATA if the new allowed topics have not changed.
+     * @return \c RETCODE_ERROR if any other error has occurred.
+     */
+    utils::ReturnCode reload_allowed_topics_(
+            const std::shared_ptr<AllowedTopicList>& allowed_topics);
+
     /////////////////////////
     // CALLBACK METHODS
     /////////////////////////
@@ -308,6 +329,13 @@ class DdsPipe
      */
     void deactivate_all_topics_nts_() noexcept;
 
+    //////////////////////////
+    // CONFIGURATION VARIABLES
+    //////////////////////////
+
+    //! Configuration of the DDS Pipe
+    DdsPipeConfiguration configuration_;
+
     /////////////////////////
     // SHARED DATA STORAGE
     /////////////////////////
@@ -378,13 +406,6 @@ class DdsPipe
      * @brief Internal mutex for concurrent calls
      */
     mutable std::mutex mutex_;
-
-    //////////////////////////
-    // CONFIGURATION VARIABLES
-    //////////////////////////
-
-    //! Configuration of the DDS Pipe
-    DdsPipeConfiguration configuration_;
 };
 
 } /* namespace core */

ddspipe_core/include/ddspipe_core/interface/IParticipant.hpp

@@ -16,6 +16,7 @@
 
 #include <ddspipe_core/interface/IReader.hpp>
 #include <ddspipe_core/interface/IWriter.hpp>
+#include <ddspipe_core/types/dds/TopicQoS.hpp>
 #include <ddspipe_core/types/participant/ParticipantId.hpp>
 #include <ddspipe_core/types/topic/Topic.hpp>
 
@@ -43,24 +44,22 @@ class IParticipant
     DDSPIPE_CORE_DllAPI
     virtual ~IParticipant() = default;
 
-    /**
-     * @brief Return the unique identifier of this Participant.
-     *
-     * @return This Participant id
-     */
+    //! The Participant's unique identifier.
     DDSPIPE_CORE_DllAPI
     virtual types::ParticipantId id() const noexcept = 0;
 
-    //! Whether this participant is RTPS
+    //! Whether the Participant is RTPS.
     DDSPIPE_CORE_DllAPI
     virtual bool is_rtps_kind() const noexcept = 0;
 
-    /**
-     * @brief Whether this Participant requires to connect ist own readers with its own writers.
-     */
+    //! Whether the Participant requires to connect its own readers with its own writers.
     DDSPIPE_CORE_DllAPI
     virtual bool is_repeater() const noexcept = 0;
 
+    //! The Participant's Topic QoS.
+    DDSPIPE_CORE_DllAPI
+    virtual types::TopicQoS topic_qos() const noexcept = 0;
+
     /**
      * @brief Return a new Writer
      *

ddspipe_core/include/ddspipe_core/interface/ITopic.hpp

@@ -18,6 +18,7 @@
 #include <string>
 
 #include <cpp_utils/Formatter.hpp>
+#include <cpp_utils/memory/Heritable.hpp>
 
 #include <ddspipe_core/library/library_dll.h>
 #include <ddspipe_core/types/topic/TopicInternalTypeDiscriminator.hpp>
@@ -60,6 +61,10 @@ class ITopic
     DDSPIPE_CORE_DllAPI
     virtual std::string serialize() const noexcept = 0;
 
+    //! Make a copy of the ITopic
+    DDSPIPE_CORE_DllAPI
+    virtual utils::Heritable<ITopic> copy() const noexcept = 0;
+
     /**
      * This refers to an internal used identifier that declares which kind of data type is going to be
      * transmitted in this Itopic inside the core.