Add JsonValue support for binary fields

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

Author: didier-wenzek

crates/extensions/tedge_flows/src/js_script.rs

@@ -40,7 +40,7 @@ impl JsScript {
     pub fn with_config(self, config: Option<serde_json::Value>) -> Self {
         if let Some(config) = config {
             Self {
-                config: JsonValue(config),
+                config: JsonValue::from(config),
                 ..self
             }
         } else {

crates/extensions/tedge_flows/src/js_value.rs

@@ -1,88 +1,214 @@
 use crate::flow::DateTime;
 use crate::flow::FlowError;
 use crate::flow::Message;
-use anyhow::Context;
 use rquickjs::Ctx;
 use rquickjs::FromJs;
 use rquickjs::IntoJs;
 use rquickjs::Value;
+use std::collections::BTreeMap;
 
+/// Akin to serde_json::Value with an extra case for binary data
 #[derive(Clone, Debug)]
-pub struct JsonValue(pub serde_json::Value);
+pub enum JsonValue {
+    Null,
+    Bool(bool),
+    Number(serde_json::Number),
+    String(String),
+    Bytes(Vec<u8>), // <= This case motivates the use of JsonValue vs serde_json::Value
+    Array(Vec<JsonValue>),
+    Object(BTreeMap<String, JsonValue>),
+}
 
 impl Default for JsonValue {
     fn default() -> Self {
-        JsonValue(serde_json::Value::Object(Default::default()))
+        JsonValue::Object(Default::default())
+    }
+}
+
+impl JsonValue {
+    fn string(value: impl ToString) -> Self {
+        JsonValue::String(value.to_string())
+    }
+
+    fn number(value: impl Into<serde_json::Number>) -> Self {
+        JsonValue::Number(value.into())
+    }
+
+    fn option(value: Option<impl Into<JsonValue>>) -> Self {
+        value.map(|v| v.into()).unwrap_or(JsonValue::Null)
+    }
+
+    fn object<T, K>(values: T) -> Self
+    where
+        T: IntoIterator<Item = (K, JsonValue)>,
+        K: ToString,
+    {
+        let object = values
+            .into_iter()
+            .map(|(k, v)| (k.to_string(), v))
+            .collect();
+        JsonValue::Object(object)
     }
 }
 
 impl From<Message> for JsonValue {
     fn from(value: Message) -> Self {
-        JsonValue(value.json())
+        JsonValue::object([
+            ("topic", JsonValue::string(value.topic)),
+            ("payload", JsonValue::string(value.payload)),
+            ("timestamp", JsonValue::option(value.timestamp)),
+        ])
     }
 }
 
 impl From<DateTime> for JsonValue {
     fn from(value: DateTime) -> Self {
-        JsonValue(value.json())
+        JsonValue::object([
+            ("seconds", JsonValue::number(value.seconds)),
+            ("nanoseconds", JsonValue::number(value.nanoseconds)),
+        ])
+    }
+}
+
+impl From<serde_json::Value> for JsonValue {
+    fn from(value: serde_json::Value) -> Self {
+        match value {
+            serde_json::Value::Null => JsonValue::Null,
+            serde_json::Value::Bool(b) => JsonValue::Bool(b),
+            serde_json::Value::Number(n) => JsonValue::Number(n),
+            serde_json::Value::String(s) => JsonValue::String(s),
+            serde_json::Value::Array(a) => {
+                JsonValue::Array(a.into_iter().map(JsonValue::from).collect())
+            }
+            serde_json::Value::Object(o) => {
+                JsonValue::object(o.into_iter().map(|(k, v)| (k, JsonValue::from(v))))
+            }
+        }
+    }
+}
+
+impl From<JsonValue> for serde_json::Value {
+    fn from(value: JsonValue) -> Self {
+        match value {
+            JsonValue::Null => serde_json::Value::Null,
+            JsonValue::Bool(b) => serde_json::Value::Bool(b),
+            JsonValue::Number(n) => serde_json::Value::Number(n),
+            JsonValue::String(s) => serde_json::Value::String(s),
+            JsonValue::Bytes(b) => serde_json::Value::String(format!("0x {b:?}")),
+            JsonValue::Array(a) => {
+                serde_json::Value::Array(a.into_iter().map(serde_json::Value::from).collect())
+            }
+            JsonValue::Object(o) => serde_json::Value::Object(
+                o.into_iter()
+                    .map(|(k, v)| (k, serde_json::Value::from(v)))
+                    .collect(),
+            ),
+        }
     }
 }
 
-impl TryFrom<serde_json::Value> for Message {
+impl TryFrom<BTreeMap<String, JsonValue>> for Message {
     type Error = FlowError;
 
-    fn try_from(value: serde_json::Value) -> Result<Self, Self::Error> {
-        let message = serde_json::from_value(value)
-            .with_context(|| "Couldn't extract message payload and topic")?;
-        Ok(message)
+    fn try_from(value: BTreeMap<String, JsonValue>) -> Result<Self, Self::Error> {
+        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 timestamp = value
+            .get("timestamp")
+            .map(|t| DateTime::try_from(t.clone()))
+            .transpose()?;
+
+        Ok(Message {
+            topic: topic.to_owned(),
+            payload: payload.to_owned(),
+            timestamp,
+        })
     }
 }
 
 impl TryFrom<JsonValue> for Message {
     type Error = FlowError;
 
     fn try_from(value: JsonValue) -> Result<Self, Self::Error> {
-        Message::try_from(value.0)
+        let JsonValue::Object(object) = value else {
+            return Err(
+                anyhow::anyhow!("Expect a message object with a topic and a payload").into(),
+            );
+        };
+        Message::try_from(object)
+    }
+}
+
+impl TryFrom<JsonValue> for DateTime {
+    type Error = FlowError;
+
+    fn try_from(value: JsonValue) -> Result<Self, Self::Error> {
+        let JsonValue::Object(object) = value else {
+            return Err(
+                anyhow::anyhow!("Expect a timestamp object with seconds and nanoseconds").into(),
+            );
+        };
+        DateTime::try_from(object)
+    }
+}
+
+impl TryFrom<BTreeMap<String, JsonValue>> for DateTime {
+    type Error = FlowError;
+
+    fn try_from(value: BTreeMap<String, JsonValue>) -> Result<Self, Self::Error> {
+        let Some(JsonValue::Number(seconds)) = value.get("seconds") else {
+            return Err(anyhow::anyhow!("Missing timestamp seconds").into());
+        };
+        let Some(JsonValue::Number(nanoseconds)) = value.get("nanoseconds") else {
+            return Err(anyhow::anyhow!("Missing timestamp nanoseconds").into());
+        };
+
+        Ok(DateTime {
+            seconds: seconds.as_u64().unwrap_or_default(),
+            nanoseconds: nanoseconds.as_u64().unwrap_or_default() as u32,
+        })
     }
 }
 
 impl TryFrom<JsonValue> for Vec<Message> {
     type Error = FlowError;
 
     fn try_from(value: JsonValue) -> Result<Self, Self::Error> {
-        match value.0 {
-            serde_json::Value::Array(array) => array.into_iter().map(Message::try_from).collect(),
-            serde_json::Value::Object(map) => {
-                Message::try_from(serde_json::Value::Object(map)).map(|message| vec![message])
-            }
-            serde_json::Value::Null => Ok(vec![]),
+        match value {
+            JsonValue::Array(array) => array.into_iter().map(Message::try_from).collect(),
+            JsonValue::Object(object) => Message::try_from(object).map(|message| vec![message]),
+            JsonValue::Null => Ok(vec![]),
             _ => Err(
                 anyhow::anyhow!("Flow scripts are expected to return an array of messages").into(),
             ),
         }
     }
 }
 
-struct JsonValueRef<'a>(&'a serde_json::Value);
+struct JsonValueRef<'a>(&'a JsonValue);
 
 impl<'js> IntoJs<'js> for JsonValue {
     fn into_js(self, ctx: &Ctx<'js>) -> rquickjs::Result<Value<'js>> {
-        JsonValueRef(&self.0).into_js(ctx)
+        JsonValueRef(&self).into_js(ctx)
     }
 }
 
 impl<'js> IntoJs<'js> for &JsonValue {
     fn into_js(self, ctx: &Ctx<'js>) -> rquickjs::Result<Value<'js>> {
-        JsonValueRef(&self.0).into_js(ctx)
+        JsonValueRef(self).into_js(ctx)
     }
 }
 
 impl<'js> IntoJs<'js> for JsonValueRef<'_> {
     fn into_js(self, ctx: &Ctx<'js>) -> rquickjs::Result<Value<'js>> {
         match self.0 {
-            serde_json::Value::Null => Ok(Value::new_null(ctx.clone())),
-            serde_json::Value::Bool(value) => Ok(Value::new_bool(ctx.clone(), *value)),
-            serde_json::Value::Number(value) => {
+            JsonValue::Null => Ok(Value::new_null(ctx.clone())),
+            JsonValue::Bool(value) => Ok(Value::new_bool(ctx.clone(), *value)),
+            JsonValue::Number(value) => {
                 if let Some(n) = value.as_i64() {
                     if let Ok(n) = i32::try_from(n) {
                         return Ok(Value::new_int(ctx.clone(), n));
@@ -94,20 +220,24 @@ impl<'js> IntoJs<'js> for JsonValueRef<'_> {
                 let nan = rquickjs::String::from_str(ctx.clone(), "NaN")?;
                 Ok(nan.into_value())
             }
-            serde_json::Value::String(value) => {
+            JsonValue::String(value) => {
                 let string = rquickjs::String::from_str(ctx.clone(), value)?;
                 Ok(string.into_value())
             }
-            serde_json::Value::Array(values) => {
+            JsonValue::Bytes(value) => {
+                let string = rquickjs::TypedArray::new(ctx.clone(), value.clone())?;
+                Ok(string.into_value())
+            }
+            JsonValue::Array(values) => {
                 let array = rquickjs::Array::new(ctx.clone())?;
                 for (i, value) in values.iter().enumerate() {
                     array.set(i, JsonValueRef(value))?;
                 }
                 Ok(array.into_value())
             }
-            serde_json::Value::Object(values) => {
+            JsonValue::Object(values) => {
                 let object = rquickjs::Object::new(ctx.clone())?;
-                for (key, value) in values.into_iter() {
+                for (key, value) in values.iter() {
                     object.set(key, JsonValueRef(value))?;
                 }
                 Ok(object.into_value())
@@ -130,40 +260,42 @@ impl JsonValue {
             return promise.finish();
         }
         if let Some(b) = value.as_bool() {
-            return Ok(JsonValue(serde_json::Value::Bool(b)));
+            return Ok(JsonValue::Bool(b));
         }
         if let Some(n) = value.as_int() {
-            return Ok(JsonValue(serde_json::Value::Number(n.into())));
+            return Ok(JsonValue::Number(n.into()));
         }
         if let Some(n) = value.as_float() {
             let js_n = serde_json::Number::from_f64(n)
-                .map(serde_json::Value::Number)
-                .unwrap_or(serde_json::Value::Null);
-            return Ok(JsonValue(js_n));
+                .map(JsonValue::Number)
+                .unwrap_or(JsonValue::Null);
+            return Ok(js_n);
         }
         if let Some(string) = value.as_string() {
-            return Ok(JsonValue(serde_json::Value::String(string.to_string()?)));
+            return Ok(JsonValue::String(string.to_string()?));
         }
         if let Some(array) = value.as_array() {
-            let array: rquickjs::Result<Vec<JsonValue>> = array.iter().collect();
-            let array = array?.into_iter().map(|v| v.0).collect();
-            return Ok(JsonValue(serde_json::Value::Array(array)));
+            let mut js_array = Vec::new();
+            for v in array.iter() {
+                js_array.push(JsonValue::from_js_value(v?)?)
+            }
+            return Ok(JsonValue::Array(js_array));
         }
         if let Some(object) = value.as_object() {
-            let mut js_object = serde_json::Map::new();
+            let mut js_object = BTreeMap::new();
             for key in object.keys::<String>().flatten() {
-                if let Ok(JsonValue(v)) = object.get(&key) {
-                    js_object.insert(key, v.clone());
+                if let Ok(v) = object.get(&key) {
+                    js_object.insert(key, JsonValue::from_js_value(v)?);
                 }
             }
-            return Ok(JsonValue(serde_json::Value::Object(js_object)));
+            return Ok(JsonValue::Object(js_object));
         }
 
-        Ok(JsonValue(serde_json::Value::Null))
+        Ok(JsonValue::Null)
     }
 
     pub(crate) fn display(value: Value<'_>) -> String {
-        let json = JsonValue::from_js_value(value).unwrap_or_default();
-        serde_json::to_string_pretty(&json.0).unwrap()
+        let json = serde_json::Value::from(JsonValue::from_js_value(value).unwrap_or_default());
+        serde_json::to_string_pretty(&json).unwrap()
     }
 }