Passing binary payloads to flows

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

Author: didier-wenzek

crates/common/mqtt_channel/src/messages.rs

@@ -78,6 +78,18 @@ impl From<String> for DebugPayload {
     }
 }
 
+impl From<DebugPayload> for Vec<u8> {
+    fn from(value: DebugPayload) -> Self {
+        value.0
+    }
+}
+
+impl From<Vec<u8>> for DebugPayload {
+    fn from(value: Vec<u8>) -> Self {
+        DebugPayload(value)
+    }
+}
+
 impl AsRef<Payload> for DebugPayload {
     fn as_ref(&self) -> &Payload {
         &self.0
@@ -202,6 +214,11 @@ impl MqttMessage {
     pub fn payload_bytes(&self) -> &[u8] {
         self.payload.as_bytes()
     }
+
+    /// Split the message into a (topic, payload) pair
+    pub fn split(self) -> (String, Payload) {
+        (self.topic.name, self.payload.0)
+    }
 }
 
 impl From<MqttMessage> for Publish {

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

@@ -78,7 +78,7 @@ impl BuildCommand for TEdgeFlowsCli {
                 let message = match (topic, payload) {
                     (Some(topic), Some(payload)) => Some(Message {
                         topic,
-                        payload,
+                        payload: payload.into_bytes(),
                         timestamp: None,
                     }),
                     (Some(_), None) => Err(anyhow!("Missing sample payload"))?,

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

@@ -78,7 +78,7 @@ fn print(messages: Result<Vec<Message>, FlowError>) {
     match messages {
         Ok(messages) => {
             for message in messages {
-                println!("[{}] {}", message.topic, message.payload);
+                println!("{message}");
             }
         }
         Err(err) => {
@@ -104,7 +104,7 @@ fn parse(line: String) -> Result<Option<Message>, Error> {
 
     Ok(Some(Message {
         topic,
-        payload,
+        payload: payload.into_bytes(),
         timestamp: None,
     }))
 }

crates/extensions/tedge_flows/src/flow.rs

@@ -38,10 +38,10 @@ pub struct DateTime {
     pub nanoseconds: u32,
 }
 
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
+#[derive(Clone, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
 pub struct Message {
     pub topic: String,
-    pub payload: String,
+    pub payload: Vec<u8>,
     pub timestamp: Option<DateTime>,
 }
 
@@ -222,7 +222,16 @@ impl Message {
     pub(crate) fn new(topic: &str, payload: &str) -> Self {
         Message {
             topic: topic.to_string(),
-            payload: payload.to_string(),
+            payload: payload.to_string().into_bytes(),
+            timestamp: Some(DateTime::now()),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn new_binary(topic: &str, payload: impl Into<Vec<u8>>) -> Self {
+        Message {
+            topic: topic.to_string(),
+            payload: payload.into(),
             timestamp: Some(DateTime::now()),
         }
     }
@@ -234,17 +243,35 @@ impl Message {
             json!({"topic": self.topic, "payload": self.payload, "timestamp": null})
         }
     }
+
+    pub fn payload_str(&self) -> Option<&str> {
+        std::str::from_utf8(&self.payload).ok()
+    }
+}
+
+impl std::fmt::Display for Message {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        '['.fmt(f)?;
+        self.topic.fmt(f)?;
+        "] ".fmt(f)?;
+        match &self.payload_str() {
+            Some(str) => str.fmt(f),
+            None => format!("{:?}", self.payload).fmt(f),
+        }
+    }
+}
+
+impl std::fmt::Debug for Message {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(self, f)
+    }
 }
 
 impl TryFrom<MqttMessage> for Message {
     type Error = FlowError;
 
     fn try_from(message: MqttMessage) -> Result<Self, Self::Error> {
-        let topic = message.topic.to_string();
-        let payload = message
-            .payload_str()
-            .map_err(|_| FlowError::UnsupportedMessage("Not an UTF8 payload".to_string()))?
-            .to_string();
+        let (topic, payload) = message.split();
         Ok(Message {
             topic,
             payload,

crates/extensions/tedge_flows/src/js_lib/text_encoder.rs

@@ -29,7 +29,14 @@ impl<'js> TextEncoder {
 
     pub fn encode(&self, ctx: Ctx<'js>, text: Value<'js>) -> Result<TypedArray<'js, u8>> {
         let string = match text.as_string() {
-            None => "".to_string(),
+            None => {
+                if let Some(object) = text.as_object() {
+                    if let Some(bytes) = object.as_typed_array::<u8>() {
+                        return Ok(bytes.clone());
+                    }
+                }
+                "".to_string()
+            }
             Some(js_string) => js_string.to_string()?,
         };
         TypedArray::new(ctx.clone(), string.as_bytes())

crates/extensions/tedge_flows/src/js_script.rs

@@ -77,7 +77,7 @@ impl JsScript {
         timestamp: &DateTime,
         message: &Message,
     ) -> Result<Vec<Message>, FlowError> {
-        debug!(target: "flows", "{}: onMessage({timestamp:?}, {message:?})", self.module_name());
+        debug!(target: "flows", "{}: onMessage({timestamp:?}, {message})", self.module_name());
         if self.no_js_on_message_fun {
             return Ok(vec![message.clone()]);
         }
@@ -105,7 +105,7 @@ impl JsScript {
         js: &JsRuntime,
         message: &Message,
     ) -> Result<(), FlowError> {
-        debug!(target: "flows", "{}: onConfigUpdate({message:?})", self.module_name());
+        debug!(target: "flows", "{}: onConfigUpdate({message})", self.module_name());
         if self.no_js_on_config_update_fun {
             return Ok(());
         }
@@ -354,13 +354,16 @@ export function onMessage(message) {
         let js = r#"
 export async function onMessage(message, config) {
     const utf8decoder = new TextDecoder();
-    const encodedText = new Uint8Array([240, 159, 146, 150]);
-    return [{topic:"decoded", payload: utf8decoder.decode(encodedText)}];
+    const encodedText = message.raw_payload;
+    console.log(encodedText);
+    const decodedText = utf8decoder.decode(encodedText);
+    console.log(decodedText);
+    return [{topic:"decoded", payload: decodedText}];
 }
         "#;
         let (runtime, script) = runtime_with(js).await;
 
-        let input = Message::new("encoded", "content");
+        let input = Message::new_binary("encoded", [240, 159, 146, 150]);
         let mut output = Message::new("decoded", "💖");
         output.timestamp = None;
         assert_eq!(
@@ -377,14 +380,16 @@ export async function onMessage(message, config) {
         let js = r#"
 export async function onMessage(message, config) {
     const utf8encoder = new TextEncoder();
+    console.log(message.payload);
     const encodedText = utf8encoder.encode(message.payload);
-    return [{topic:"encoded", payload: `[${encodedText}]`}];
+    console.log(encodedText);
+    return [{topic:"encoded", payload: encodedText}];
 }
         "#;
         let (runtime, script) = runtime_with(js).await;
 
         let input = Message::new("decoded", "💖");
-        let mut output = Message::new("encoded", "[240,159,146,150]");
+        let mut output = Message::new_binary("encoded", [240, 159, 146, 150]);
         output.timestamp = None;
         assert_eq!(
             script

crates/extensions/tedge_flows/src/js_value.rs

@@ -53,9 +53,15 @@ impl JsonValue {
 
 impl From<Message> for JsonValue {
     fn from(value: Message) -> Self {
+        let raw_payload = JsonValue::Bytes(value.payload.clone());
+        let payload = match String::from_utf8(value.payload) {
+            Ok(utf8) => JsonValue::string(utf8),
+            Err(_) => JsonValue::Null,
+        };
         JsonValue::object([
             ("topic", JsonValue::string(value.topic)),
-            ("payload", JsonValue::string(value.payload)),
+            ("payload", payload),
+            ("raw_payload", raw_payload),
             ("timestamp", JsonValue::option(value.timestamp)),
         ])
     }
@@ -114,8 +120,10 @@ impl TryFrom<BTreeMap<String, JsonValue>> for Message {
         let Some(JsonValue::String(topic)) = value.get("topic") else {
             return Err(anyhow::anyhow!("Missing message topic").into());
         };
-        let Some(JsonValue::String(payload)) = value.get("payload") else {
-            return Err(anyhow::anyhow!("Missing message payload").into());
+        let payload = match value.get("payload") {
+            Some(JsonValue::String(payload)) => payload.to_owned().into_bytes(),
+            Some(JsonValue::Bytes(payload)) => payload.to_owned(),
+            _ => return Err(anyhow::anyhow!("Missing message payload").into()),
         };
         let timestamp = value
             .get("timestamp")
@@ -124,7 +132,7 @@ impl TryFrom<BTreeMap<String, JsonValue>> for Message {
 
         Ok(Message {
             topic: topic.to_owned(),
-            payload: payload.to_owned(),
+            payload,
             timestamp,
         })
     }
@@ -225,8 +233,8 @@ impl<'js> IntoJs<'js> for JsonValueRef<'_> {
                 Ok(string.into_value())
             }
             JsonValue::Bytes(value) => {
-                let string = rquickjs::TypedArray::new(ctx.clone(), value.clone())?;
-                Ok(string.into_value())
+                let bytes = rquickjs::TypedArray::new(ctx.clone(), value.clone())?;
+                Ok(bytes.into_value())
             }
             JsonValue::Array(values) => {
                 let array = rquickjs::Array::new(ctx.clone())?;
@@ -271,6 +279,12 @@ impl JsonValue {
                 .unwrap_or(JsonValue::Null);
             return Ok(js_n);
         }
+        if let Some(object) = value.as_object() {
+            if let Some(bytes) = object.as_typed_array::<u8>() {
+                let bytes = bytes.as_bytes().unwrap_or_default().to_vec();
+                return Ok(JsonValue::Bytes(bytes));
+            }
+        }
         if let Some(string) = value.as_string() {
             return Ok(JsonValue::String(string.to_string()?));
         }