Merge pull request #4001 from didier-wenzek/refactor/deprecate-flow-message-string-payload

breaking change: message payloads are passed as binary to flow scripts

Author: didier-wenzek

crates/extensions/tedge_flows/src/js_script.rs

@@ -184,9 +184,10 @@ mod tests {
     async fn collectd_script() {
         let js = r#"
 export function onMessage(message, context) {
-    const { topic = "topic/not/set" } = context.config;
+    const { topic = "topic/not/set" } = context.config
+    const td = new globalThis.TextDecoder()
     let groups = message.topic.split( '/')
-    let data = message.payload.split(':')
+    let data = td.decode(message.payload).split(':')
 
     let group = groups[2]
 	let measurement = groups[3]
@@ -335,7 +336,7 @@ export function onMessage(message) {
         let js = r#"
 export async function onMessage(message) {
     const utf8decoder = new TextDecoder();
-    const encodedText = message.raw_payload;
+    const encodedText = message.payload;
     console.log(encodedText);
     const decodedText = utf8decoder.decode(encodedText);
     console.log(decodedText);
@@ -360,8 +361,10 @@ export async function onMessage(message) {
         let js = r#"
 export async function onMessage(message) {
     const utf8encoder = new TextEncoder();
-    console.log(message.payload);
-    const encodedText = utf8encoder.encode(message.payload);
+    const utf8decoder = new TextDecoder();
+    const payload = utf8decoder.decode(message.payload);
+    console.log(payload);
+    const encodedText = utf8encoder.encode(payload);
     console.log(encodedText);
     return [{topic:"encoded", payload: encodedText}];
 }
@@ -382,10 +385,11 @@ export async function onMessage(message) {
     #[tokio::test]
     async fn decode_utf8_with_bom_and_invalid_chars() {
         let js = r#"
+const utf8 = new TextDecoder();
+
 export async function onMessage(message) {
-    const utf8decoder = new TextDecoder();
-    const encodedText = message.raw_payload;
-    const decodedText = utf8decoder.decode(encodedText);
+    const encodedText = message.payload;
+    const decodedText = utf8.decode(encodedText);
     return [{topic:"decoded", payload: decodedText}];
 }
         "#;
@@ -408,10 +412,12 @@ export async function onMessage(message) {
         let js = r#"
 export async function onMessage(message) {
     const utf8encoder = new TextEncoder();
+    const utf8decoder = new TextDecoder();
+    const payload = utf8decoder.decode(message.payload);
     const u8array = new Uint8Array(8);
-    const result = utf8encoder.encodeInto(message.payload, u8array);
+    const result = utf8encoder.encodeInto(payload, u8array);
     console.log(result);
-    utf8encoder.encodeInto(message.payload, u8array.subarray(4));
+    utf8encoder.encodeInto(payload, u8array.subarray(4));
     return [{topic:"encoded", payload: u8array}];
 }
         "#;
@@ -435,7 +441,7 @@ export async function onMessage(message) {
     const te = new globalThis.TextEncoder();
     const td = new globalThis.TextDecoder();
 
-    const encodedText = message.raw_payload;
+    const encodedText = message.payload;
     const decodedText = td.decode(encodedText);
     const finalPayload = te.encode(decodedText + decodedText);
     return [{topic:"decoded", payload: finalPayload}];
@@ -458,7 +464,7 @@ export async function onMessage(message) {
     async fn reading_raw_integers() {
         let js = r#"
 export async function onMessage(message) {
-    const measurements = new Uint32Array(message.raw_payload.buffer);
+    const measurements = new Uint32Array(message.payload.buffer);
 
     const tedge_json = {
         time: measurements[0],

crates/extensions/tedge_flows/src/js_value.rs

@@ -115,14 +115,10 @@ 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", payload),
-            ("raw_payload", raw_payload),
+            ("payload", raw_payload),
             ("time", JsonValue::option(value.timestamp)),
         ])
     }

docs/src/references/mappers/flows.md

@@ -74,18 +74,30 @@ interface FlowStep {
 }
 ```
 
-A message contains the message topic and payload as well as a processing timestamp.
-The bytes of the raw message payload are also accessible as an array of unsigned bytes: 
+A message has three attributes: a `topic`, a `payload` and a processing timestamp.
 
 ```ts
 type Message = {
   topic: string,
-  payload: string,
-  raw_payload: Uint8Array,
+  payload: Uint8Array,
   time: Date
 }
 ```
 
+:::note
+The message `payload` is an array of unsigned bytes that has to be explicitly converted to a string when appropriate:
+
+```js
+const utf8 = new TextDecoder();
+
+export function onMessage(message) {
+    let string_payload = utf8.decode(message.payload)
+    let json_payload = JSON.parse(string_payload)
+    // ..
+}
+```
+:::
+
 ### Context
 
 The `context` object passed to `onMessage()` and `onInterval()` gives scripts and flows a way to share data.
@@ -126,9 +138,9 @@ The `context.config` is an object freely defined by the step module, to provide
 
 The `onMessage` function is called for each message to be transformed
   - The arguments passed to the function are:
-    - The message `{ topic: string, payload: string, raw_payload: Uint8Array, time: Date }`
+    - The message `{ topic: string, payload: Uint8Array, time: Date }`
     - A context object with the config and state
-  - The function is expected to return zero, one or many transformed messages `[{ topic: string, payload: string }]`
+  - The function is expected to return zero, one or many transformed messages `[{ topic: string, payload: Uint8Array | string }]`
   - An exception can be thrown if the input message cannot be transformed.
 
 A flow script can also export a `onInterval` function

tests/RobotFramework/tests/azure/az-registration.js

@@ -1,16 +1,18 @@
+const utf8 = new TextDecoder()
+
 export function onMessage(message, context) {
     let topic = message.topic.split( '/')
     if (topic.length < 5) {
         throw new Error("Not a thin-edge entity registration");
     }
     let entity_id = `${topic[1]}/${topic[2]}/${topic[3]}/${topic[4]}`
 
-    let entity = JSON.parse(message.payload)
+    let entity = JSON.parse(utf8.decode(message.payload))
     context.mapper.set(entity_id, entity)
 
     console.log("Entity metadata", entity_id, entity)
     return [{
         topic: "te/infos",
         payload: `New entity: ${entity.name}`
     }]
-}
+}

tests/RobotFramework/tests/azure/custom-az-v2.js

@@ -1,8 +1,10 @@
+const utf8 = new TextDecoder()
+
 export function onMessage(message, context) {
     let topic = message.topic.split('/')
     let mea_type = topic[6];
 
-    let mea = JSON.parse(message.payload)
+    let mea = JSON.parse(utf8.decode(message.payload))
     if (mea_type) {
         mea.type = mea_type
     }
@@ -21,4 +23,4 @@ export function onMessage(message, context) {
         topic: message.topic,
         payload: JSON.stringify(mea)
     }]
-}
+}

tests/RobotFramework/tests/azure/custom-az.js

@@ -1,8 +1,10 @@
+const utf8 = new TextDecoder()
+
 export function onMessage(message) {
     let topic = message.topic.split('/')
     let mea_type = topic[6];
 
-    let mea = JSON.parse(message.payload)
+    let mea = JSON.parse(utf8.decode(message.payload))
     if (mea_type) {
         mea.type = mea_type
     }
@@ -11,4 +13,4 @@ export function onMessage(message) {
         topic: message.topic,
         payload: JSON.stringify(mea)
     }]
-}
+}

tests/RobotFramework/tests/cumulocity/flows/collectd.js

@@ -1,7 +1,9 @@
+const utf8 = new TextDecoder()
+
 export function onMessage(message, context) {
     const { topic = "te/device/main///m/collectd" } = context.config;
     let groups = message.topic.split('/')
-    let data = message.payload.split(':')
+    let data = utf8.decode(message.payload).split(':')
 
     if (groups.length < 4) {
         throw new Error("Not a collectd topic");
@@ -20,4 +22,4 @@ export function onMessage(message, context) {
         topic: topic,
         payload: `{"time": ${time}, "${group}": {"${measurement}": ${value}}}`
     }]
-}
+}

tests/RobotFramework/tests/cumulocity/flows/custom-measurements.js

@@ -1,7 +1,9 @@
+const utf8 = new TextDecoder()
+
 export function onMessage(message, context) {
     const topic = message.topic.split('/');
     const name = topic[6];
-    const value = JSON.parse(message.payload);
+    const value = JSON.parse(utf8.decode(message.payload));
     if (!Number.isFinite(value)) {
         throw new Error("Invalid payload. Only numbers are accepted");
     }
@@ -31,4 +33,4 @@ export function onMessage(message, context) {
         topic: "c8y/measurement/measurements/create",
         payload: JSON.stringify(c8y_msg),
     }]
-}
+}

tests/RobotFramework/tests/tedge_flows/flows/average.js

@@ -7,9 +7,11 @@ class State {
     static agg_for_topic = {}
 }
 
+const utf8 = new TextDecoder();
+
 export function onMessage (message) {
     let topic = message.topic
-    let payload = JSON.parse(message.payload)
+    let payload = JSON.parse(utf8.decode(message.payload))
     let agg_payload = State.agg_for_topic[topic]
     if (agg_payload) {
         for (let [k, v] of Object.entries(payload)) {
@@ -105,4 +107,4 @@ export function onInterval() {
 
     State.agg_for_topic = {}
     return messages
-}
+}

tests/RobotFramework/tests/tedge_flows/flows/collectd-to-te.js

@@ -1,7 +1,9 @@
+const utf8 = new TextDecoder();
+
 export function onMessage(message, context) {
     const { topic = "te/device/main///m/collectd" } = context.config;
     let groups = message.topic.split('/')
-    let data = message.payload.split(':')
+    let data = utf8.decode(message.payload).split(':')
 
     if (groups.length < 4) {
         throw new Error("Not a collectd topic");
@@ -20,4 +22,4 @@ export function onMessage(message, context) {
         topic: topic,
         payload: `{"time": ${time}, "${group}": {"${measurement}": ${value}}}`
     }]
-}
+}