fixup! Add base64 encoding support to tedge flows test

Author: didier-wenzek

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

@@ -1,4 +1,3 @@
-use crate::cli::flows::decode_message;
 use crate::cli::flows::list::ListCommand;
 use crate::cli::flows::test::TestCommand;
 use crate::command::BuildCommand;
@@ -9,6 +8,7 @@ use anyhow::Context;
 use anyhow::Error;
 use std::path::PathBuf;
 use tedge_config::TEdgeConfig;
+use tedge_flows::flow::Message;
 use tedge_flows::MessageProcessor;
 
 #[derive(clap::Subcommand, Debug)]
@@ -86,9 +86,7 @@ impl BuildCommand for TEdgeFlowsCli {
             } => {
                 let flows_dir = flows_dir.unwrap_or_else(|| Self::default_flows_dir(config));
                 let message = match (topic, payload) {
-                    (Some(topic), Some(payload)) => {
-                        Some(decode_message(topic, payload, base64_input)?)
-                    }
+                    (Some(topic), Some(payload)) => Some(Message::new(topic, payload)),
                     (Some(_), None) => Err(anyhow!("Missing sample payload"))?,
                     (None, Some(_)) => Err(anyhow!("Missing sample topic"))?,
                     (None, None) => None,

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

@@ -2,27 +2,4 @@ mod cli;
 mod list;
 mod test;
 
-use base64::prelude::*;
 pub use cli::TEdgeFlowsCli;
-use tedge_flows::flow::Message;
-
-fn decode_message(
-    topic: String,
-    payload: String,
-    base64_payload: bool,
-) -> Result<Message, anyhow::Error> {
-    let payload = if base64_payload {
-        BASE64_STANDARD.decode(payload.as_bytes())?
-    } else {
-        payload.into_bytes()
-    };
-    Ok(Message::new(topic, payload))
-}
-
-fn encode_message(mut message: Message, base64_payload: bool) -> Message {
-    if base64_payload {
-        message.payload = BASE64_STANDARD.encode(message.payload).into_bytes();
-    };
-
-    message
-}

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

@@ -1,9 +1,9 @@
-use crate::cli::flows::decode_message;
-use crate::cli::flows::encode_message;
 use crate::cli::flows::TEdgeFlowsCli;
 use crate::command::Command;
 use crate::log::MaybeFancy;
 use anyhow::Error;
+use base64::prelude::BASE64_STANDARD;
+use base64::prelude::*;
 use std::path::PathBuf;
 use tedge_config::TEdgeConfig;
 use tedge_flows::flow::*;
@@ -37,12 +37,13 @@ impl Command for TestCommand {
         };
         if let Some(message) = &self.message {
             let timestamp = DateTime::now();
-            self.process(&mut processor, message, &timestamp).await;
+            self.process(&mut processor, message.clone(), &timestamp)
+                .await;
         } else {
             let mut stdin = BufReader::new(tokio::io::stdin());
-            while let Some(message) = next_message(self.base64_input, &mut stdin).await {
+            while let Some(message) = next_message(&mut stdin).await {
                 let timestamp = DateTime::now();
-                self.process(&mut processor, &message, &timestamp).await;
+                self.process(&mut processor, message, &timestamp).await;
             }
         }
         if self.final_on_interval {
@@ -57,41 +58,52 @@ impl TestCommand {
     async fn process(
         &self,
         processor: &mut MessageProcessor,
-        message: &Message,
+        mut message: Message,
         timestamp: &DateTime,
     ) {
+        if self.base64_input {
+            match BASE64_STANDARD.decode(message.payload) {
+                Ok(decoded) => message.payload = decoded,
+                Err(err) => {
+                    tracing::error!("Cannot decode message: {}", err);
+                    return;
+                }
+            }
+        };
         processor
-            .on_message(timestamp, message)
+            .on_message(timestamp, &message)
             .await
             .into_iter()
-            .map(|(_, v)| v)
-            .for_each(|msg| print(self.base64_output, msg))
+            .for_each(|msg| self.print_messages(msg))
     }
 
     async fn tick(&self, processor: &mut MessageProcessor, timestamp: &DateTime) {
         processor
             .on_interval(timestamp)
             .await
             .into_iter()
-            .map(|(_, v)| v)
-            .for_each(|msg| print(self.base64_output, msg))
+            .for_each(|msg| self.print_messages(msg))
     }
-}
 
-fn print(base64_output: bool, messages: Result<Vec<Message>, FlowError>) {
-    match messages {
-        Ok(messages) => {
-            for message in messages {
-                println!("{}", encode_message(message, base64_output));
+    fn print_messages(&self, (flow, messages): (String, Result<Vec<Message>, FlowError>)) {
+        match messages {
+            Ok(mut messages) => {
+                if self.base64_output {
+                    for message in messages.iter_mut() {
+                        message.payload = BASE64_STANDARD.encode(&message.payload).into_bytes();
+                    }
+                }
+                for message in messages {
+                    println!("{}", message);
+                }
+            }
+            Err(err) => {
+                tracing::error!("Error in {flow}: {}", err)
             }
-        }
-        Err(err) => {
-            eprintln!("Error: {}", err)
         }
     }
 }
-
-fn parse(base64_input: bool, line: String) -> Result<Option<Message>, Error> {
+fn parse_input(line: String) -> Result<Option<Message>, Error> {
     let line = line.trim();
     if line.is_empty() {
         return Ok(None);
@@ -105,7 +117,7 @@ fn parse(base64_input: bool, line: String) -> Result<Option<Message>, Error> {
 
     let topic = line[1..closing_bracket].to_string();
     let payload = line[closing_bracket + 1..].to_string();
-    let message = decode_message(topic, payload, base64_input)?;
+    let message = Message::new(topic, payload);
 
     Ok(Some(message))
 }
@@ -128,9 +140,10 @@ async fn next_line(input: &mut BufReader<Stdin>) -> Option<String> {
         }
     }
 }
-async fn next_message(base64_input: bool, input: &mut BufReader<Stdin>) -> Option<Message> {
+
+async fn next_message(input: &mut BufReader<Stdin>) -> Option<Message> {
     let line = next_line(input).await?;
-    match parse(base64_input, line) {
+    match parse_input(line) {
         Ok(message) => message,
         Err(err) => {
             eprintln!("Fail to parse input message {}", err);