Merge pull request #3926 from didier-wenzek/feat/flow-enabled-aws-mapper

feat: AWS mapper is configurable using flows

Author: didier-wenzek

Cargo.lock

@@ -1,25 +1,23 @@
 use crate::core::component::TEdgeComponent;
 use crate::core::mapper::start_basic_actors;
 use crate::core::mqtt::configure_proxy;
+use crate::core::mqtt::flows_status_topic;
 use anyhow::Context;
 use async_trait::async_trait;
-use aws_mapper_ext::converter::AwsConverter;
-use clock::WallClock;
-use mqtt_channel::TopicFilter;
-use tedge_actors::ConvertingActor;
-use tedge_actors::MessageSink;
-use tedge_actors::MessageSource;
-use tedge_actors::NoConfig;
+use aws_mapper_ext::AwsConverter;
+use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
 use tedge_api::service_health_topic;
 use tedge_config::models::TopicPrefix;
-use tedge_config::tedge_toml::mapper_config::AwsMapperConfig;
 use tedge_config::tedge_toml::mapper_config::AwsMapperSpecificConfig;
 use tedge_config::tedge_toml::ProfileName;
 use tedge_config::TEdgeConfig;
+use tedge_file_system_ext::FsWatchActorBuilder;
+use tedge_flows::FlowsMapperBuilder;
 use tedge_mqtt_bridge::rumqttc::Transport;
 use tedge_mqtt_bridge::BridgeConfig;
 use tedge_mqtt_bridge::MqttBridgeActorBuilder;
+use tedge_watch_ext::WatchActorBuilder;
 use tracing::warn;
 use yansi::Paint;
 
@@ -32,15 +30,16 @@ impl TEdgeComponent for AwsMapper {
     async fn start(
         &self,
         tedge_config: TEdgeConfig,
-        _config_dir: &tedge_config::Path,
+        config_dir: &tedge_config::Path,
     ) -> Result<(), anyhow::Error> {
         let aws_config = tedge_config.mapper_config::<AwsMapperSpecificConfig>(&self.profile)?;
         let prefix = &aws_config.bridge.topic_prefix;
         let aws_mapper_name = format!("tedge-mapper-{prefix}");
         let (mut runtime, mut mqtt_actor) =
             start_basic_actors(&aws_mapper_name, &tedge_config).await?;
-
         let mqtt_schema = MqttSchema::with_root(tedge_config.mqtt.topic_root.clone());
+        let service_topic_id = EntityTopicId::default_main_service(&aws_mapper_name)?;
+
         if tedge_config.mqtt.bridge.built_in {
             let device_id = aws_config.device.id()?;
             let device_topic_id = tedge_config.mqtt.device_topic_id.clone();
@@ -78,37 +77,36 @@ impl TEdgeComponent for AwsMapper {
         } else if tedge_config.proxy.address.or_none().is_some() {
             warn!("`proxy.address` is configured without the built-in bridge enabled. The bridge MQTT connection to the cloud will {} communicate via the configured proxy.", "not".bold())
         }
-        let clock = Box::new(WallClock);
         let aws_converter = AwsConverter::new(
             aws_config.cloud_specific.mapper.timestamp,
-            clock,
-            mqtt_schema,
+            &mqtt_schema,
             aws_config.cloud_specific.mapper.timestamp_format,
             prefix.value().clone(),
             aws_config.mapper.mqtt.max_payload_size.0,
+            aws_config.topics.to_string(),
         );
-        let mut aws_converting_actor = ConvertingActor::builder("AwsConverter", aws_converter);
-
-        aws_converting_actor.connect_source(get_topic_filter(&aws_config), &mut mqtt_actor);
-        aws_converting_actor.connect_sink(NoConfig, &mqtt_actor);
-
-        runtime.spawn(aws_converting_actor).await?;
+        let flows_dir =
+            tedge_flows::flows_dir(config_dir, "aws", self.profile.as_ref().map(|p| p.as_ref()));
+        let flows = aws_converter.flow_registry(flows_dir).await?;
+        let flows_status = flows_status_topic(&mqtt_schema, &service_topic_id);
+
+        let mut fs_actor = FsWatchActorBuilder::new();
+        let mut cmd_watcher_actor = WatchActorBuilder::new();
+
+        let mut flows_mapper = FlowsMapperBuilder::try_new(flows, flows_status).await?;
+        flows_mapper.connect(&mut mqtt_actor);
+        flows_mapper.connect_fs(&mut fs_actor);
+        flows_mapper.connect_cmd(&mut cmd_watcher_actor);
+
+        runtime.spawn(flows_mapper).await?;
+        runtime.spawn(fs_actor).await?;
+        runtime.spawn(cmd_watcher_actor).await?;
         runtime.spawn(mqtt_actor).await?;
         runtime.run_to_completion().await?;
         Ok(())
     }
 }
 
-fn get_topic_filter(aws_config: &AwsMapperConfig) -> TopicFilter {
-    let mut topics = TopicFilter::empty();
-    for topic in aws_config.topics.0.clone() {
-        if topics.try_add(&topic).is_err() {
-            warn!("The configured topic '{topic}' is invalid and ignored.");
-        }
-    }
-    topics
-}
-
 fn built_in_bridge_rules(
     remote_client_id: &str,
     topic_prefix: &TopicPrefix,

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

@@ -1,4 +1,3 @@
-use crate::az::SkipMosquittoHealthStatus;
 use crate::core::component::TEdgeComponent;
 use crate::core::mapper::start_basic_actors;
 use crate::core::mqtt::configure_proxy;
@@ -100,7 +99,6 @@ impl TEdgeComponent for AzureMapper {
         let flows_dir =
             tedge_flows::flows_dir(config_dir, "az", self.profile.as_ref().map(|p| p.as_ref()));
         let mut flows = ConnectedFlowRegistry::new(flows_dir);
-        flows.register_builtin(SkipMosquittoHealthStatus);
         flows
             .persist_builtin_flow("mea", az_converter.builtin_flow().as_str())
             .await?;

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

@@ -1,42 +1 @@
-use std::time::SystemTime;
-use tedge_flows::ConfigError;
-use tedge_flows::FlowError;
-use tedge_flows::JsonValue;
-use tedge_flows::Message;
-use tedge_flows::Transformer;
-
 pub mod mapper;
-
-#[derive(Clone, Default)]
-struct SkipMosquittoHealthStatus;
-
-impl Transformer for SkipMosquittoHealthStatus {
-    fn name(&self) -> &str {
-        "skip-mosquitto-health-status"
-    }
-
-    fn set_config(&mut self, _config: JsonValue) -> Result<(), ConfigError> {
-        Ok(())
-    }
-
-    fn on_message(
-        &self,
-        _timestamp: SystemTime,
-        message: &Message,
-    ) -> Result<Vec<Message>, FlowError> {
-        // don't convert mosquitto bridge notification topic
-        // https://github.com/thin-edge/thin-edge.io/issues/2236
-        if let [_, _, _, _, _, "status", "health"] =
-            message.topic.split('/').collect::<Vec<_>>()[..]
-        {
-            if message
-                .payload_str()
-                .map(|s| s == "0" || s == "1")
-                .unwrap_or(false)
-            {
-                return Ok(vec![]);
-            }
-        }
-        Ok(vec![message.clone()])
-    }
-}

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

@@ -10,24 +10,21 @@ homepage = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
-clock = { workspace = true }
-log = { workspace = true }
-serde = { workspace = true }
-serde_json = { workspace = true }
-tedge_actors = { workspace = true }
+camino = { workspace = true }
 tedge_api = { workspace = true }
 tedge_config = { workspace = true }
+tedge_flows = { workspace = true }
 tedge_mqtt_ext = { workspace = true }
 tedge_utils = { workspace = true }
-thiserror = { workspace = true }
-time = { workspace = true }
 
 [dev-dependencies]
 assert-json-diff = { workspace = true }
 assert_matches = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+tempfile = { workspace = true }
 time = { workspace = true, features = ["macros"] }
+tokio = { workspace = true, features = ["test-util"] }
 
 [lints]
 workspace = true