Add base64 encoding support to tedge mqtt

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

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

@@ -30,6 +30,9 @@ pub enum TEdgeMqttCli {
         /// Retain flag
         #[clap(short, long = "retain")]
         retain: bool,
+        /// Decode the payload before publishing
+        #[clap(long)]
+        base64: bool,
         /// Repeat the message
         #[clap(long)]
         repeat: Option<u32>,
@@ -51,6 +54,9 @@ pub enum TEdgeMqttCli {
         /// Avoid printing the message topics on the console
         #[clap(long = "no-topic")]
         hide_topic: bool,
+        /// Encode the received payloads
+        #[clap(long)]
+        base64: bool,
         /// Disconnect and exit after the specified timeout (e.g., 60s, 1h)
         #[clap(long, short = 'W')]
         duration: Option<SecondsOrHumanTime>,
@@ -75,6 +81,7 @@ impl BuildCommand for TEdgeMqttCli {
                     message,
                     qos,
                     retain,
+                    base64,
                     repeat,
                     sleep,
                 } => MqttPublishCommand {
@@ -85,6 +92,7 @@ impl BuildCommand for TEdgeMqttCli {
                     qos,
                     client_id: format!("{}-{}", PUB_CLIENT_PREFIX, std::process::id()),
                     retain,
+                    base64,
                     ca_file: auth_config.ca_file.clone(),
                     ca_dir: auth_config.ca_dir,
                     client_auth_config: auth_config.client,
@@ -96,6 +104,7 @@ impl BuildCommand for TEdgeMqttCli {
                     topic,
                     qos,
                     hide_topic,
+                    base64,
                     duration,
                     count,
                     retained_only,
@@ -105,6 +114,7 @@ impl BuildCommand for TEdgeMqttCli {
                     topic,
                     qos,
                     hide_topic,
+                    base64,
                     client_id: format!("{}-{}", SUB_CLIENT_PREFIX, std::process::id()),
                     ca_file: auth_config.ca_file,
                     ca_dir: auth_config.ca_dir,

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

@@ -1,5 +1,6 @@
 use crate::command::Command;
 use crate::log::MaybeFancy;
+use base64::prelude::*;
 use camino::Utf8PathBuf;
 use mqtt_channel::MqttMessage;
 use mqtt_channel::PubChannel;
@@ -19,6 +20,7 @@ pub struct MqttPublishCommand {
     pub qos: mqtt_channel::QoS,
     pub client_id: String,
     pub retain: bool,
+    pub base64: bool,
     pub ca_file: Option<Utf8PathBuf>,
     pub ca_dir: Option<Utf8PathBuf>,
     pub client_auth_config: Option<MqttAuthClientConfig>,
@@ -72,7 +74,13 @@ async fn publish(cmd: &MqttPublishCommand) -> Result<(), anyhow::Error> {
     let mut mqtt = mqtt_channel::Connection::new(&config).await?;
     let mut signals = tedge_utils::signals::TermSignals::new(None);
 
-    let message = MqttMessage::new(&cmd.topic, cmd.message.clone())
+    let payload = if cmd.base64 {
+        BASE64_STANDARD.decode(cmd.message.as_bytes())?
+    } else {
+        cmd.message.clone().into_bytes()
+    };
+
+    let message = MqttMessage::new(&cmd.topic, payload)
         .with_qos(cmd.qos)
         .with_retain_flag(cmd.retain);
 

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

@@ -1,6 +1,8 @@
 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;
@@ -20,6 +22,7 @@ pub struct MqttSubscribeCommand {
     pub topic: SimpleTopicFilter,
     pub qos: QoS,
     pub hide_topic: bool,
+    pub base64: bool,
     pub client_id: String,
     pub ca_file: Option<Utf8PathBuf>,
     pub ca_dir: Option<Utf8PathBuf>,
@@ -83,27 +86,38 @@ async fn subscribe(cmd: &MqttSubscribeCommand) -> Result<(), anyhow::Error> {
             }
         };
 
+        if cmd.retained_only && !message.retain {
+            info!(target: "MQTT", "Received first non-retained message. topic={}", message.topic.name);
+            break;
+        }
+
+        let message = if cmd.base64 {
+            MqttMessage::new(
+                &message.topic,
+                BASE64_STANDARD.encode(message.payload_bytes()),
+            )
+        } else {
+            message
+        };
+
         match message.payload_str() {
             Ok(payload) => {
-                if cmd.retained_only && !message.retain {
-                    info!(target: "MQTT", "Received first non-retained message. topic={}", message.topic.name);
-                    break;
-                }
                 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) {
-                    info!(target: "MQTT", "Received {n_messages} message/s");
-                    break;
-                }
             }
             Err(err) => error!(target: "MQTT", "{err}"),
         }
+
+        n_messages += 1;
+        if matches!(cmd.count, Some(count) if count > 0 && n_messages >= count) {
+            info!(target: "MQTT", "Received {n_messages} message/s");
+            break;
+        }
     }
 
     mqtt.published.close_channel();

tests/RobotFramework/tests/tedge/tedge_mqtt.robot

@@ -45,6 +45,18 @@ tedge mqtt sub breaks after receiving first non-retailed message
     Should Contain    ${messages[0]}    [foo/1] bar1
     Should Contain    ${messages[1]}    [foo/2] bar2
 
+sending binary payload with tedge mqtt
+    Execute Command
+    ...    tedge mqtt pub -r --base64 te/device/main///m/ eyAidGVtcGVyYXR1cmUiOiAzMC4yLCAiaHVtaWRpdHkiOiAyMyB9
+    ${messages}=    Execute Command    tedge mqtt sub te/device/main///m/ --count 1
+    Should Contain    ${messages}    { "temperature": 30.2, "humidity": 23 }
+
+receiving binary payload with tedge mqtt
+    Execute Command
+    ...    tedge mqtt pub -r te/device/main///m/ '{ "temperature": 30.2, "humidity": 23 }'
+    ${messages}=    Execute Command    tedge mqtt sub --base64 te/device/main///m/ --count 1
+    Should Contain    ${messages}    eyAidGVtcGVyYXR1cmUiOiAzMC4yLCAiaHVtaWRpdHkiOiAyMyB9
+
 
 *** Keywords ***
 Validate duration