fixup! fixup! Add base64 encoding support to tedge mqtt

Author: didier-wenzek

crates/core/tedge/src/cli/mqtt/subscribe.rs

@@ -2,15 +2,13 @@ use crate::command::Command;
 use crate::log::MaybeFancy;
 use base64::prelude::*;
 use camino::Utf8PathBuf;
-use mqtt_channel::MqttMessage;
 use mqtt_channel::QoS;
 use mqtt_channel::StreamExt;
 use mqtt_channel::TopicFilter;
 use std::time::Duration;
 use tedge_config::tedge_toml::MqttAuthClientConfig;
 use tedge_config::TEdgeConfig;
 use tokio::io::AsyncWriteExt;
-use tracing::error;
 use tracing::info;
 
 const DEFAULT_QUEUE_CAPACITY: usize = 10;
@@ -91,27 +89,25 @@ async fn subscribe(cmd: &MqttSubscribeCommand) -> Result<(), anyhow::Error> {
             break;
         }
 
-        let message = if cmd.base64 {
-            MqttMessage::new(
-                &message.topic,
-                BASE64_STANDARD.encode(message.payload_bytes()),
-            )
+        let payload = if cmd.base64 {
+            BASE64_STANDARD.encode(message.payload_bytes())
         } else {
-            message
+            match message.payload_str() {
+                Ok(payload_str) => payload_str.to_string(),
+                Err(_) => format!(
+                    "<ERR=NON-UTF8> {}",
+                    BASE64_STANDARD.encode(message.payload_bytes())
+                ),
+            }
         };
 
-        match message.payload_str() {
-            Ok(payload) => {
-                let line = if cmd.hide_topic {
-                    format!("{payload}\n")
-                } else {
-                    format!("[{}] {payload}\n", &message.topic)
-                };
-                let _ = stdout.write_all(line.as_bytes()).await;
-                let _ = stdout.flush().await;
-            }
-            Err(err) => error!(target: "MQTT", "{err}"),
-        }
+        let line = if cmd.hide_topic {
+            format!("{payload}\n")
+        } else {
+            format!("[{}] {payload}\n", &message.topic)
+        };
+        let _ = stdout.write_all(line.as_bytes()).await;
+        let _ = stdout.flush().await;
 
         n_messages += 1;
         if matches!(cmd.count, Some(count) if count > 0 && n_messages >= count) {