Merge pull request #3805 from didier-wenzek/feat/flow-source-types

feat: tedge flows consuming and producing file content

Author: didier-wenzek

Cargo.lock

@@ -69,6 +69,7 @@ tedge_test_utils = { path = "crates/tests/tedge_test_utils" }
 tedge_timer_ext = { path = "crates/extensions/tedge_timer_ext" }
 tedge_uploader_ext = { path = "crates/extensions/tedge_uploader_ext" }
 tedge_utils = { path = "crates/common/tedge_utils" }
+tedge_watch_ext = { path = "crates/extensions/tedge_watch_ext" }
 upload = { path = "crates/common/upload" }
 
 # external dependencies

Cargo.toml

@@ -74,8 +74,9 @@ impl TestCommand {
                 }
             }
         };
+        let source = SourceTag::Mqtt;
         processor
-            .on_message(timestamp, &message)
+            .on_message(timestamp, &source, &message)
             .await
             .into_iter()
             .for_each(|msg| self.print_messages(msg))
@@ -89,9 +90,9 @@ impl TestCommand {
             .for_each(|msg| self.print_messages(msg))
     }
 
-    fn print_messages(&self, (flow, messages): (String, Result<Vec<Message>, FlowError>)) {
-        match messages {
-            Ok(mut messages) => {
+    fn print_messages(&self, result: FlowResult) {
+        match result {
+            FlowResult::Ok { mut messages, .. } => {
                 if self.base64_output {
                     for message in messages.iter_mut() {
                         message.payload = BASE64_STANDARD.encode(&message.payload).into_bytes();
@@ -101,8 +102,8 @@ impl TestCommand {
                     println!("{}", message);
                 }
             }
-            Err(err) => {
-                tracing::error!("Error in {flow}: {}", err)
+            FlowResult::Err { flow, error, .. } => {
+                tracing::error!("Error in {flow}: {}", error)
             }
         }
     }

crates/core/tedge/src/cli/flows/test.rs

@@ -36,6 +36,7 @@ tedge_mqtt_ext = { workspace = true }
 tedge_signal_ext = { workspace = true }
 tedge_timer_ext = { workspace = true }
 tedge_uploader_ext = { workspace = true }
+tedge_watch_ext = { workspace = true }
 tracing = { workspace = true }
 yansi = { workspace = true }
 

crates/core/tedge_mapper/Cargo.toml

@@ -3,6 +3,7 @@ use crate::TEdgeComponent;
 use tedge_config::TEdgeConfig;
 use tedge_file_system_ext::FsWatchActorBuilder;
 use tedge_flows::FlowsMapperBuilder;
+use tedge_watch_ext::WatchActorBuilder;
 
 pub struct GenMapper;
 
@@ -21,9 +22,13 @@ impl TEdgeComponent for GenMapper {
         flows_mapper.connect(&mut mqtt_actor);
         flows_mapper.connect_fs(&mut fs_actor);
 
+        let mut cmd_watcher_actor = WatchActorBuilder::new();
+        cmd_watcher_actor.connect(&mut flows_mapper);
+
         runtime.spawn(flows_mapper).await?;
         runtime.spawn(mqtt_actor).await?;
         runtime.spawn(fs_actor).await?;
+        runtime.spawn(cmd_watcher_actor).await?;
         runtime.run_to_completion().await?;
         Ok(())
     }

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

@@ -27,6 +27,7 @@ serde_json = { workspace = true }
 tedge_actors = { workspace = true }
 tedge_file_system_ext = { workspace = true }
 tedge_mqtt_ext = { workspace = true }
+tedge_watch_ext = { workspace = true }
 thiserror = { workspace = true }
 time = { workspace = true }
 tokio = { workspace = true, features = ["fs", "macros", "time", "sync"] }