Manually fix a set of Topic QoS, and implement the max-tx-rate (#86)

* Make the DdsPipeConfiguration required Signed-off-by: tempate <[email protected]> * Rename max-reception-rate to max-rx-rate Signed-off-by: tempate <[email protected]> * Support max-tx-rate in the replayer Signed-off-by: tempate <[email protected]> * Support manual topics configuration Signed-off-by: tempate <[email protected]> * Move the allowed & builtin topics to the pipe Signed-off-by: tempate <[email protected]> * Add Topic QoS to the McapReaderParticipant Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Support the new Topic QoS Signed-off-by: tempate <[email protected]> * Minor fixes to the documentation Signed-off-by: tempate <[email protected]> * Add missing quotes to wildcard entries Signed-off-by: tempate <[email protected]> * Fix downsampling test Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Apply DdsPipe suggestions Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Remove the built-in topics from the DdsReplayer Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Apply more suggestions Signed-off-by: tempate <[email protected]> * Apply the manual topics to the dynamic writer Signed-off-by: tempate <[email protected]> * Fix compilation error Signed-off-by: tempate <[email protected]> * Remove Topic Filtering outdated note Signed-off-by: tempate <[email protected]> * Fix Windows tag Signed-off-by: tempate <[email protected]> * Fix forthcoming notes Signed-off-by: tempate <[email protected]> * Uncrustify Signed-off-by: tempate <[email protected]> --------- Signed-off-by: tempate <[email protected]>
eProsima · Nov 8, 2023 · 0d0c59a · 0d0c59a
1 parent 9270ac5
commit 0d0c59a
Show file tree

Hide file tree

Showing 28 changed files with 412 additions and 402 deletions.
diff --git a/ddsrecorder/src/cpp/main.cpp b/ddsrecorder/src/cpp/main.cpp
@@ -67,11 +67,7 @@ std::unique_ptr<eprosima::utils::event::FileWatcherHandler> create_filewatcher(
                 try
                 {
                     eprosima::ddsrecorder::yaml::RecorderConfiguration new_configuration(file_path);
-                    // Create new allowed topics list
-                    auto new_allowed_topics = std::make_shared<core::AllowedTopicList>(
-                        new_configuration.allowlist,
-                        new_configuration.blocklist);
-                    recorder->reload_allowed_topics(new_allowed_topics);
+                    recorder->reload_configuration(new_configuration);
                 }
                 catch (const std::exception& e)
                 {
@@ -102,11 +98,7 @@ std::unique_ptr<eprosima::utils::event::PeriodicEventHandler> create_periodic_ha
                 try
                 {
                     eprosima::ddsrecorder::yaml::RecorderConfiguration new_configuration(file_path);
-                    // Create new allowed topics list
-                    auto new_allowed_topics = std::make_shared<core::AllowedTopicList>(
-                        new_configuration.allowlist,
-                        new_configuration.blocklist);
-                    recorder->reload_allowed_topics(new_allowed_topics);
+                    recorder->reload_configuration(new_configuration);
                 }
                 catch (const std::exception& e)
                 {

diff --git a/ddsrecorder/src/cpp/tool/DdsRecorder.cpp b/ddsrecorder/src/cpp/tool/DdsRecorder.cpp
@@ -33,11 +33,6 @@ DdsRecorder::DdsRecorder(
         const DdsRecorderStateCode& init_state,
         const std::string& file_name)
 {
-    // Create allowed topics list
-    auto allowed_topics = std::make_shared<AllowedTopicList>(
-        configuration.allowlist,
-        configuration.blocklist);
-
     // Create Discovery Database
     discovery_database_ =
             std::make_shared<DiscoveryDatabase>();
@@ -115,19 +110,17 @@ DdsRecorder::DdsRecorder(
 
     // Create DDS Pipe
     pipe_ = std::make_unique<DdsPipe>(
-        allowed_topics,
+        configuration.ddspipe_configuration,
         discovery_database_,
         payload_pool_,
         participants_database_,
-        thread_pool_,
-        configuration.builtin_topics,
-        true);
+        thread_pool_);
 }
 
-utils::ReturnCode DdsRecorder::reload_allowed_topics(
-        const std::shared_ptr<AllowedTopicList>& allowed_topics)
+utils::ReturnCode DdsRecorder::reload_configuration(
+        const yaml::RecorderConfiguration& new_configuration)
 {
-    return pipe_->reload_allowed_topics(allowed_topics);
+    return pipe_->reload_configuration(new_configuration.ddspipe_configuration);
 }
 
 void DdsRecorder::start()

diff --git a/ddsrecorder/src/cpp/tool/DdsRecorder.hpp b/ddsrecorder/src/cpp/tool/DdsRecorder.hpp
@@ -77,8 +77,8 @@ class DdsRecorder
      * @return \c RETCODE_OK if allowed topics list has been updated correctly
      * @return \c RETCODE_NO_DATA if new allowed topics list is the same as the previous one
      */
-    utils::ReturnCode reload_allowed_topics(
-            const std::shared_ptr<ddspipe::core::AllowedTopicList>& allowed_topics);
+    utils::ReturnCode reload_configuration(
+            const yaml::RecorderConfiguration& new_configuration);
 
     //! Start recorder (\c mcap_handler_)
     void start();

diff --git a/ddsrecorder/test/blackbox/mcap/McapFileCreationTest.cpp b/ddsrecorder/test/blackbox/mcap/McapFileCreationTest.cpp
@@ -88,10 +88,10 @@ std::unique_ptr<DdsRecorder> create_recorder(
     YAML::Node yml;
 
     eprosima::ddsrecorder::yaml::RecorderConfiguration configuration(yml);
-    configuration.downsampling = downsampling;
+    configuration.topic_qos.downsampling = downsampling;
     // Set default value for downsampling
     // TODO: Change mechanism setting topic qos' default values from specs
-    eprosima::ddspipe::core::types::TopicQoS::default_downsampling.store(downsampling);
+    eprosima::ddspipe::core::types::TopicQoS::default_topic_qos.set_value(configuration.topic_qos);
     configuration.event_window = event_window;
     eprosima::ddspipe::core::types::DomainId domainId;
     domainId.domain_id = test::DOMAIN;

diff --git a/ddsrecorder_participants/include/ddsrecorder_participants/replayer/McapReaderParticipant.hpp b/ddsrecorder_participants/include/ddsrecorder_participants/replayer/McapReaderParticipant.hpp
@@ -70,6 +70,10 @@ class McapReaderParticipant : public ddspipe::core::IParticipant
     DDSRECORDER_PARTICIPANTS_DllAPI
     bool is_rtps_kind() const noexcept override;
 
+    //! Override topic_qos() IParticipant method
+    DDSRECORDER_PARTICIPANTS_DllAPI
+    ddspipe::core::types::TopicQoS topic_qos() const noexcept override;
+
     //! Override create_writer_() IParticipant method
     DDSRECORDER_PARTICIPANTS_DllAPI
     std::shared_ptr<ddspipe::core::IWriter> create_writer(

diff --git a/ddsrecorder_participants/src/cpp/replayer/McapReaderParticipant.cpp b/ddsrecorder_participants/src/cpp/replayer/McapReaderParticipant.cpp
@@ -68,6 +68,11 @@ bool McapReaderParticipant::is_rtps_kind() const noexcept
     return false;
 }
 
+TopicQoS McapReaderParticipant::topic_qos() const noexcept
+{
+    return configuration_->topic_qos;
+}
+
 std::shared_ptr<IWriter> McapReaderParticipant::create_writer(
         const ITopic& /* topic */)
 {

diff --git a/ddsrecorder_yaml/include/ddsrecorder_yaml/recorder/YamlReaderConfiguration.hpp b/ddsrecorder_yaml/include/ddsrecorder_yaml/recorder/YamlReaderConfiguration.hpp
@@ -22,6 +22,8 @@
 
 #include <cpp_utils/memory/Heritable.hpp>
 
+#include <ddspipe_core/configuration/DdsPipeConfiguration.hpp>
+#include <ddspipe_core/types/dds/TopicQoS.hpp>
 #include <ddspipe_core/types/topic/dds/DistributedTopic.hpp>
 #include <ddspipe_core/types/topic/filter/IFilterTopic.hpp>
 
@@ -52,15 +54,13 @@ class DDSRECORDER_YAML_DllAPI RecorderConfiguration
     RecorderConfiguration(
             const std::string& file_path);
 
+    // DDS Pipe Configuration
+    ddspipe::core::DdsPipeConfiguration ddspipe_configuration;
+
     // Participants configurations
     std::shared_ptr<ddspipe::participants::SimpleParticipantConfiguration> simple_configuration;
     std::shared_ptr<ddspipe::participants::ParticipantConfiguration> recorder_configuration;
 
-    // Topic filtering
-    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> allowlist{};
-    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> blocklist{};
-    std::set<utils::Heritable<ddspipe::core::types::DistributedTopic>> builtin_topics{};
-
     // Output file params
     std::string output_filepath = ".";
     std::string output_filename = "output";
@@ -71,8 +71,6 @@ class DDSRECORDER_YAML_DllAPI RecorderConfiguration
     unsigned int buffer_size = 100;
     unsigned int event_window = 20;
     bool log_publish_time = false;
-    unsigned int downsampling = 1;
-    float max_reception_rate = 0;
     bool only_with_type = false;
     mcap::McapWriterOptions mcap_writer_options{"ros2"};
     bool record_types = true;
@@ -86,9 +84,9 @@ class DDSRECORDER_YAML_DllAPI RecorderConfiguration
 
     // Specs
     unsigned int n_threads = 12;
-    unsigned int max_history_depth = 5000;
     int max_pending_samples = 5000;  // -1 <-> no limit || 0 <-> no pending samples
     unsigned int cleanup_period;
+    ddspipe::core::types::TopicQoS topic_qos{};
 
 protected:
 

diff --git a/ddsrecorder_yaml/include/ddsrecorder_yaml/replayer/YamlReaderConfiguration.hpp b/ddsrecorder_yaml/include/ddsrecorder_yaml/replayer/YamlReaderConfiguration.hpp
@@ -22,6 +22,8 @@
 #include <cpp_utils/time/time_utils.hpp>
 #include <cpp_utils/types/Fuzzy.hpp>
 
+#include <ddspipe_core/configuration/DdsPipeConfiguration.hpp>
+#include <ddspipe_core/types/dds/TopicQoS.hpp>
 #include <ddspipe_core/types/topic/dds/DistributedTopic.hpp>
 #include <ddspipe_core/types/topic/filter/IFilterTopic.hpp>
 
@@ -52,15 +54,13 @@ class DDSRECORDER_YAML_DllAPI ReplayerConfiguration
     ReplayerConfiguration(
             const std::string& file_path);
 
+    // DDS Pipe Configuration
+    ddspipe::core::DdsPipeConfiguration ddspipe_configuration;
+
     // Participants configurations
     std::shared_ptr<ddsrecorder::participants::McapReaderParticipantConfiguration> mcap_reader_configuration;
     std::shared_ptr<ddspipe::participants::SimpleParticipantConfiguration> replayer_configuration;
 
-    // Topic filtering
-    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> allowlist{};
-    std::set<utils::Heritable<ddspipe::core::types::IFilterTopic>> blocklist{};
-    std::set<utils::Heritable<ddspipe::core::types::DistributedTopic>> builtin_topics{};
-
     // Replay params
     std::string input_file;
     utils::Fuzzy<utils::Timestamp> begin_time{};
@@ -71,7 +71,7 @@ class DDSRECORDER_YAML_DllAPI ReplayerConfiguration
 
     // Specs
     unsigned int n_threads = 12;
-    unsigned int max_history_depth = 5000;
+    ddspipe::core::types::TopicQoS topic_qos{};
 
 protected:
 

diff --git a/ddsrecorder_yaml/src/cpp/recorder/YamlReaderConfiguration.cpp b/ddsrecorder_yaml/src/cpp/recorder/YamlReaderConfiguration.cpp
@@ -20,6 +20,7 @@
 #include <cpp_utils/utils.hpp>
 
 #include <ddspipe_core/types/dynamic_types/types.hpp>
+#include <ddspipe_core/types/topic/filter/ManualTopic.hpp>
 #include <ddspipe_core/types/topic/filter/WildcardDdsFilterTopic.hpp>
 #include <ddspipe_participants/types/address/Address.hpp>
 
@@ -115,11 +116,16 @@ void RecorderConfiguration::load_ddsrecorder_configuration_(
         WildcardDdsFilterTopic rpc_request_topic, rpc_response_topic;
         rpc_request_topic.topic_name.set_value("rq/*");
         rpc_response_topic.topic_name.set_value("rr/*");
-        blocklist.insert(
+
+        ddspipe_configuration.blocklist.insert(
             utils::Heritable<WildcardDdsFilterTopic>::make_heritable(rpc_request_topic));
-        blocklist.insert(
+
+        ddspipe_configuration.blocklist.insert(
             utils::Heritable<WildcardDdsFilterTopic>::make_heritable(rpc_response_topic));
 
+        // The DDS Pipe should be enabled on start up.
+        ddspipe_configuration.init_enabled = true;
+
         // Initialize controller domain with the same as the one being recorded
         // WARNING: dds tag must have been parsed beforehand
         controller_domain = simple_configuration->domain;
@@ -198,24 +204,6 @@ void RecorderConfiguration::load_recorder_configuration_(
         log_publish_time = YamlReader::get<bool>(yml, RECORDER_LOG_PUBLISH_TIME_TAG, version);
     }
 
-    /////
-    // Get optional downsampling
-    if (YamlReader::is_tag_present(yml, DOWNSAMPLING_TAG))
-    {
-        downsampling = YamlReader::get_positive_int(yml, DOWNSAMPLING_TAG);
-        // Set default value for downsampling
-        TopicQoS::default_downsampling.store(downsampling);
-    }
-
-    /////
-    // Get optional max reception rate
-    if (YamlReader::is_tag_present(yml, MAX_RECEPTION_RATE_TAG))
-    {
-        // Set default value for max reception rate
-        TopicQoS::default_max_reception_rate.store(YamlReader::get_nonnegative_float(yml,
-                MAX_RECEPTION_RATE_TAG));
-    }
-
     /////
     // Get optional only_with_type
     if (YamlReader::is_tag_present(yml, RECORDER_ONLY_WITH_TYPE_TAG))
@@ -290,12 +278,12 @@ void RecorderConfiguration::load_specs_configuration_(
         n_threads = YamlReader::get_positive_int(yml, NUMBER_THREADS_TAG);
     }
 
-    // Get maximum history depth
-    if (YamlReader::is_tag_present(yml, MAX_HISTORY_DEPTH_TAG))
+    /////
+    // Get optional Topic QoS
+    if (YamlReader::is_tag_present(yml, SPECS_QOS_TAG))
     {
-        max_history_depth = YamlReader::get_positive_int(yml, MAX_HISTORY_DEPTH_TAG);
-        // Set default value for history
-        TopicQoS::default_history_depth.store(max_history_depth);
+        YamlReader::fill<TopicQoS>(topic_qos, YamlReader::get_value_in_tag(yml, SPECS_QOS_TAG), version);
+        TopicQoS::default_topic_qos.set_value(topic_qos);
     }
 
     // Get max pending samples
@@ -363,28 +351,39 @@ void RecorderConfiguration::load_dds_configuration_(
     // Get optional allowlist
     if (YamlReader::is_tag_present(yml, ALLOWLIST_TAG))
     {
-        allowlist = YamlReader::get_set<utils::Heritable<IFilterTopic>>(yml, ALLOWLIST_TAG, version);
+        ddspipe_configuration.allowlist = YamlReader::get_set<utils::Heritable<IFilterTopic>>(yml, ALLOWLIST_TAG,
+                        version);
 
         // Add to allowlist always the type object topic
         WildcardDdsFilterTopic internal_topic;
         internal_topic.topic_name.set_value(TYPE_OBJECT_TOPIC_NAME);
-        allowlist.insert(
+        ddspipe_configuration.allowlist.insert(
             utils::Heritable<WildcardDdsFilterTopic>::make_heritable(internal_topic));
     }
 
     /////
     // Get optional blocklist
     if (YamlReader::is_tag_present(yml, BLOCKLIST_TAG))
     {
-        blocklist = YamlReader::get_set<utils::Heritable<IFilterTopic>>(yml, BLOCKLIST_TAG, version);
+        ddspipe_configuration.blocklist = YamlReader::get_set<utils::Heritable<IFilterTopic>>(yml, BLOCKLIST_TAG,
+                        version);
+    }
+
+    /////
+    // Get optional topics
+    if (YamlReader::is_tag_present(yml, TOPICS_TAG))
+    {
+        const auto& manual_topics = YamlReader::get_list<ManualTopic>(yml, TOPICS_TAG, version);
+        ddspipe_configuration.manual_topics =
+                std::vector<ManualTopic>(manual_topics.begin(), manual_topics.end());
     }
 
     /////
     // Get optional builtin topics
     if (YamlReader::is_tag_present(yml, BUILTIN_TAG))
     {
         // WARNING: Parse builtin topics AFTER specs and recorder, as some topic-specific default values are set there
-        builtin_topics = YamlReader::get_set<utils::Heritable<DistributedTopic>>(yml, BUILTIN_TAG,
+        ddspipe_configuration.builtin_topics = YamlReader::get_set<utils::Heritable<DistributedTopic>>(yml, BUILTIN_TAG,
                         version);
     }
 }