Merge pull request #3836 from albinsuresh/refactor/add-topic-filters

refactor: add topic filters to combine them

Author: albinsuresh

crates/common/mqtt_channel/src/connection.rs

@@ -83,7 +83,7 @@ impl SubscriberOps for SubscriberHandle {
         {
             let mut subs = self.subscriptions.lock().unwrap();
             for topic in &topics {
-                subs.add(topic)?;
+                subs.try_add(topic)?;
             }
         }
         self.client

crates/common/mqtt_channel/src/topics.rs

@@ -8,6 +8,8 @@ use std::collections::HashSet;
 use std::convert::TryInto;
 use std::fmt::Display;
 use std::fmt::Formatter;
+use std::ops::Add;
+use std::ops::AddAssign;
 
 /// An MQTT topic
 #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize)]
@@ -93,14 +95,10 @@ impl TopicFilter {
     }
 
     /// Check if the pattern is valid and add it to this topic filter.
-    pub fn add(&mut self, pattern: &str) -> Result<(), MqttError> {
-        let pattern = String::from(pattern);
-        if rumqttc::valid_filter(&pattern) {
-            self.patterns.push(pattern);
-            Ok(())
-        } else {
-            Err(MqttError::InvalidFilter { pattern })
-        }
+    pub fn try_add(&mut self, pattern: &str) -> Result<(), MqttError> {
+        let other = TopicFilter::new(pattern)?;
+        *self += other;
+        Ok(())
     }
 
     /// Assuming the pattern is valid and add it to this topic filter.
@@ -240,7 +238,7 @@ impl TryInto<TopicFilter> for Vec<&str> {
     fn try_into(self) -> Result<TopicFilter, Self::Error> {
         let mut filter = TopicFilter::empty();
         for pattern in self.into_iter() {
-            filter.add(pattern)?
+            filter.try_add(pattern)?
         }
         Ok(filter)
     }
@@ -258,7 +256,7 @@ impl TryInto<TopicFilter> for Vec<String> {
     fn try_into(self) -> Result<TopicFilter, Self::Error> {
         let mut filter = TopicFilter::empty();
         for pattern in self.into_iter() {
-            filter.add(pattern.as_str())?
+            filter.try_add(pattern.as_str())?
         }
         Ok(filter)
     }
@@ -270,13 +268,28 @@ impl AsRef<str> for Topic {
     }
 }
 
+impl Add for TopicFilter {
+    type Output = TopicFilter;
+
+    fn add(mut self, other: TopicFilter) -> TopicFilter {
+        self += other;
+        self
+    }
+}
+
+impl AddAssign for TopicFilter {
+    fn add_assign(&mut self, rhs: Self) {
+        self.add_all(rhs);
+    }
+}
+
 impl TryInto<TopicFilter> for HashSet<String> {
     type Error = MqttError;
 
     fn try_into(self) -> Result<TopicFilter, Self::Error> {
         let mut filter = TopicFilter::empty();
         for pattern in self.into_iter() {
-            filter.add(pattern.as_str())?
+            filter.try_add(pattern.as_str())?
         }
         Ok(filter)
     }
@@ -345,4 +358,23 @@ mod tests {
         let removed = topics.remove_overlapping_patterns();
         assert!(removed.is_empty());
     }
+
+    #[test]
+    fn test_adding_topic_filters() {
+        let filter1 = TopicFilter::new_unchecked("a/b/c");
+        let filter2 = TopicFilter::new_unchecked("d/e/f");
+
+        let combined = filter1.clone() + filter2.clone();
+        assert!(combined.accept_topic_name("a/b/c"));
+        assert!(combined.accept_topic_name("d/e/f"));
+    }
+
+    #[test]
+    fn test_add_assign_topic_filters() {
+        let mut filter = TopicFilter::new_unchecked("a/b/c");
+        filter += TopicFilter::new_unchecked("d/e/f");
+
+        assert!(filter.accept_topic_name("a/b/c"));
+        assert!(filter.accept_topic_name("d/e/f"));
+    }
 }

crates/core/tedge_mapper/src/aws/mapper.rs

@@ -103,7 +103,7 @@ impl TEdgeComponent for AwsMapper {
 fn get_topic_filter(aws_config: &TEdgeConfigReaderAws) -> TopicFilter {
     let mut topics = TopicFilter::empty();
     for topic in aws_config.topics.0.clone() {
-        if topics.add(&topic).is_err() {
+        if topics.try_add(&topic).is_err() {
             warn!("The configured topic '{topic}' is invalid and ignored.");
         }
     }

crates/core/tedge_mapper/src/az/mapper.rs

@@ -111,7 +111,7 @@ impl TEdgeComponent for AzureMapper {
 fn get_topic_filter(az_config: &TEdgeConfigReaderAz) -> TopicFilter {
     let mut topics = TopicFilter::empty();
     for topic in az_config.topics.0.clone() {
-        if topics.add(&topic).is_err() {
+        if topics.try_add(&topic).is_err() {
             warn!("The configured topic '{topic}' is invalid and ignored.");
         }
     }

crates/extensions/c8y_mapper_ext/src/config.rs

@@ -221,7 +221,7 @@ impl C8yMapperConfig {
 
         // Add user configurable external topic filters
         for topic in c8y_config.topics.0.clone() {
-            if topics.add(&topic).is_err() {
+            if topics.try_add(&topic).is_err() {
                 warn!("The configured topic '{topic}' is invalid and ignored.");
             }
         }
@@ -293,7 +293,7 @@ impl C8yMapperConfig {
             bridge_config,
         ) {
             for topic in operations.topics_for_operations() {
-                topic_filter.add(&topic)?;
+                topic_filter.try_add(&topic)?;
             }
         }
 

crates/extensions/tedge_config_manager/src/lib.rs

@@ -148,10 +148,9 @@ impl ConfigManagerBuilder {
 
     /// List of MQTT topic filters the log actor has to subscribe to
     fn subscriptions(config: &ConfigManagerConfig) -> TopicFilter {
-        let mut topic_filter = TopicFilter::empty();
-        topic_filter.add_all(config.config_snapshot_topic.clone());
+        let mut topic_filter = config.config_snapshot_topic.clone();
         if config.config_update_enabled {
-            topic_filter.add_all(config.config_update_topic.clone());
+            topic_filter += config.config_update_topic.clone();
         }
         topic_filter
     }

crates/extensions/tedge_flows/src/config.rs

@@ -281,7 +281,7 @@ fn topic_filters(patterns: Vec<String>) -> Result<TopicFilter, ConfigError> {
     let mut topics = TopicFilter::empty();
     for pattern in patterns {
         topics
-            .add(pattern.as_str())
+            .try_add(pattern.as_str())
             .map_err(|_| ConfigError::IncorrectTopicFilter(pattern.clone()))?;
     }
     Ok(topics)

crates/extensions/tedge_log_manager/src/config.rs

@@ -70,14 +70,13 @@ impl LogManagerConfig {
             ChannelFilter::Command(OperationType::LogUpload),
         );
 
-        let mut log_metadata_sync_topics = mqtt_schema.topics(
+        let log_metadata_sync_topics = mqtt_schema.topics(
             EntityFilter::Entity(&mqtt_device_topic_id),
             ChannelFilter::Command(OperationType::SoftwareUpdate),
-        );
-        log_metadata_sync_topics.add_all(mqtt_schema.topics(
+        ) + mqtt_schema.topics(
             EntityFilter::Entity(&mqtt_device_topic_id),
             ChannelFilter::Command(OperationType::ConfigUpdate),
-        ));
+        );
 
         Ok(Self {
             mqtt_schema,

crates/extensions/tedge_log_manager/src/lib.rs

@@ -149,9 +149,7 @@ impl LogManagerBuilder {
 
     /// List of MQTT topic filters the log actor has to subscribe to
     fn subscriptions(config: &LogManagerConfig) -> TopicFilter {
-        let mut topics = config.logfile_request_topic.clone();
-        topics.add_all(config.log_metadata_sync_topics.clone());
-        topics
+        config.logfile_request_topic.clone() + config.log_metadata_sync_topics.clone()
     }
 
     /// Extract a log actor request from an MQTT message

crates/extensions/tedge_mqtt_ext/src/lib.rs

@@ -272,7 +272,7 @@ impl MqttActorBuilder {
     pub(crate) fn build_actor(self) -> MqttActor {
         let mut topic_filter = TopicFilter::empty();
         for pattern in &self.subscription_diff.subscribe {
-            topic_filter.add(pattern).unwrap();
+            topic_filter.try_add(pattern).unwrap();
             tracing::info!(target: "MQTT sub", "{pattern}");
         }