The default config of flow transformation is {}

The default config was previously null, forcing explicit checks in the scripts.

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/extensions/tedge_gen_mapper/pipelines/circuit-breaker.js

@@ -19,10 +19,10 @@ export function process (timestamp, message, config) {
     State.total += 1
     State.batch[0] += 1
     if (State.open) {
-        let back_to_normal = config?.back_to_normal || 100
+        let back_to_normal = config.back_to_normal || 100
         if (State.total < back_to_normal) {
             State.open = false
-            if (config?.message_on_back_to_normal) {
+            if (config.message_on_back_to_normal) {
                 return [config?.message_on_back_to_normal, message]
             } else {
                 return [message]
@@ -31,13 +31,13 @@ export function process (timestamp, message, config) {
             return []
         }
     } else {
-        let too_many = config?.too_many || 1000
+        let too_many = config.too_many || 1000
         if (State.total < too_many) {
             return [message]
         } else {
             State.open = true
-            if (config?.message_on_too_many) {
-                return [config?.message_on_too_many]
+            if (config.message_on_too_many) {
+                return [config.message_on_too_many]
             } else {
                 return []
             }
@@ -47,15 +47,15 @@ export function process (timestamp, message, config) {
 
 
 export function tick(timestamp, config) {
-    let max_batch_count = config?.tick_count || 10
+    let max_batch_count = config.tick_count || 10
     let new_batch_count = State.batch.unshift(0)
     if (new_batch_count > max_batch_count) {
         State.total -= State.batch.pop()
     }
 
-    if (config?.stats_topic) {
+    if (config.stats_topic) {
         return [{
-            topic: config?.stats_topic,
+            topic: config.stats_topic,
             payload: `{"circuit-breaker-open": ${State.open}, "total": ${State.total}, "batch": ${State.batch}}`
         }]
     } else {

crates/extensions/tedge_gen_mapper/pipelines/collectd-to-te.js

@@ -8,7 +8,7 @@ export function process(_timestamp, message, config) {
     let value = data[1]
 
     return [{
-        topic: config?.topic || "te/device/main///m/collectd",
+        topic: config.topic || "te/device/main///m/collectd",
         payload: `{"time": ${time}, "${group}": {"${measurement}": ${value}}}`
     }]
 }

crates/extensions/tedge_gen_mapper/pipelines/drop_stragglers.js

@@ -12,8 +12,8 @@ export function process (timestamp, message, config) {
     }
 
     let time = timestamp.seconds + (timestamp.nanoseconds / 1e9)
-    let max = time + (config?.max_advance || 1);
-    let min = time - (config?.max_delay || 10);
+    let max = time + (config.max_advance || 1);
+    let min = time - (config.max_delay || 10);
 
     if (min <= msg_timestamp && msg_timestamp <= max) {
         return [message]

crates/extensions/tedge_gen_mapper/pipelines/measurements.toml

@@ -2,6 +2,5 @@ input_topics = ["te/+/+/+/+/m/+"]
 
 stages = [
     { filter = "add_timestamp.js" },
-    { filter = "drop_stragglers.js", config = { max_delay = 60 } },
     { filter = "te_to_c8y.js", meta_topics = ["te/+/+/+/+/m/+/meta"] }
 ]

crates/extensions/tedge_gen_mapper/pipelines/set_topic.js

@@ -1,6 +1,6 @@
 export function process (timestamp, message, config) {
   return [{
-    topic: config?.topic || "te/error",
+    topic: config.topic || "te/error",
     payload: message.payload
   }]
 }

crates/extensions/tedge_gen_mapper/pipelines/te_to_c8y.js

@@ -51,10 +51,10 @@ export function process(t, message, config) {
     type: type
   }
 
-  let meta = (config || {})[`${message.topic}/meta`] || {}
+  let meta = config[`${message.topic}/meta`] || {}
 
   for (let [k, v] of Object.entries(payload)) {
-    let k_meta = (meta || {})[k] || {}
+    let k_meta = meta[k] || {}
     if (k === "time") {
       let t = v
       if (typeof(v) === "number") {
@@ -123,9 +123,6 @@ export function update_config(message, config) {
   let fragment = {
     [type]: metadata
   }
-  if (!config) {
-    config = {}
-  }
   Object.assign(config, fragment)
 
   return config

crates/extensions/tedge_gen_mapper/src/js_filter.rs

@@ -20,9 +20,15 @@ pub struct JsFilter {
     pub tick_every_seconds: u64,
 }
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct JsonValue(serde_json::Value);
 
+impl Default for JsonValue {
+    fn default() -> Self {
+        JsonValue(serde_json::Value::Object(Default::default()))
+    }
+}
+
 impl JsFilter {
     pub fn new(pipeline: PathBuf, index: usize, path: PathBuf) -> Self {
         let module_name = format!("{}|{}|{}", pipeline.display(), index, path.display());

tests/RobotFramework/tests/tedge_gen_mapper/pipelines/count-messages.js

@@ -13,7 +13,7 @@ export function process (timestamp, message) {
 
 export function tick(timestamp, config) {
     let message = {
-        topic: config?.topic || "te/error",
+        topic: config.topic || "te/error",
         payload: JSON.stringify(State.count_per_topic)
     }
 

tests/RobotFramework/tests/tedge_gen_mapper/pipelines/set_topic.js

@@ -1,6 +1,6 @@
 export function process (timestamp, message, config) {
   return [{
-    topic: config?.topic || "te/error",
+    topic: config.topic || "te/error",
     payload: message.payload
   }]
 }

tests/RobotFramework/tests/tedge_gen_mapper/pipelines/te_to_c8y.js

@@ -51,10 +51,10 @@ export function process(t, message, config) {
     type: type
   }
 
-  let meta = (config || {})[`${message.topic}/meta`] || {}
+  let meta = config[`${message.topic}/meta`] || {}
 
   for (let [k, v] of Object.entries(payload)) {
-    let k_meta = (meta || {})[k] || {}
+    let k_meta = meta[k] || {}
     if (k === "time") {
       let t = v
       if (typeof(v) === "number") {
@@ -123,9 +123,6 @@ export function update_config(message, config) {
   let fragment = {
     [type]: metadata
   }
-  if (!config) {
-    config = {}
-  }
   Object.assign(config, fragment)
 
   return config