Deprecated flow step onConfigUpdate method

Now flow script have to use the shared KV store to share data such as entity metadata or measurement units

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

Author: didier-wenzek

crates/extensions/tedge_flows/src/config.rs

@@ -40,9 +40,6 @@ pub struct StepConfig {
     #[serde(default)]
     #[serde(deserialize_with = "parse_human_duration")]
     interval: Duration,
-
-    #[serde(default)]
-    meta_topics: Vec<String>,
 }
 
 #[derive(Deserialize)]
@@ -159,7 +156,6 @@ impl FlowConfig {
             script: ScriptSpec::JavaScript(script),
             config: None,
             interval: Duration::default(),
-            meta_topics: vec![],
         };
         Self {
             input: InputConfig::Mqtt {
@@ -214,11 +210,7 @@ impl StepConfig {
         let script = JsScript::new(flow.to_owned(), index, path)
             .with_config(self.config)
             .with_interval(self.interval);
-        let config_topics = topic_filters(self.meta_topics)?;
-        Ok(FlowStep {
-            script,
-            config_topics,
-        })
+        Ok(FlowStep { script })
     }
 }
 

crates/extensions/tedge_flows/src/flow.rs

@@ -44,7 +44,6 @@ pub struct Flow {
 /// A message transformation step
 pub struct FlowStep {
     pub script: JsScript,
-    pub config_topics: TopicFilter,
 }
 
 pub enum SourceTag {
@@ -156,33 +155,7 @@ impl Flow {
     }
 
     pub fn topics(&self) -> TopicFilter {
-        let mut topics = self.input.topics();
-        for step in self.steps.iter() {
-            topics.add_all(step.config_topics.clone())
-        }
-        topics
-    }
-
-    pub async fn on_config_update(
-        &mut self,
-        js_runtime: &JsRuntime,
-        message: &Message,
-    ) -> FlowResult {
-        let result = self.on_config_update_steps(js_runtime, message).await;
-        self.publish(result)
-    }
-
-    async fn on_config_update_steps(
-        &mut self,
-        js_runtime: &JsRuntime,
-        message: &Message,
-    ) -> Result<Vec<Message>, FlowError> {
-        for step in self.steps.iter_mut() {
-            if step.config_topics.accept_topic_name(&message.topic) {
-                step.script.on_config_update(js_runtime, message).await?
-            }
-        }
-        Ok(vec![])
+        self.input.topics()
     }
 
     pub fn accept_message(&self, _source: &SourceTag, message: &Message) -> bool {
@@ -377,9 +350,6 @@ impl FlowStep {
         if script.no_js_on_message_fun {
             warn!(target: "flows", "Flow script with no 'onMessage' function: {}", script.path);
         }
-        if script.no_js_on_config_update_fun && !self.config_topics.is_empty() {
-            warn!(target: "flows", "Flow script with no 'onConfigUpdate' function: {}; but configured with 'config_topics' in {flow}", script.path);
-        }
         if script.no_js_on_interval_fun && !script.interval.is_zero() {
             warn!(target: "flows", "Flow script with no 'onInterval' function: {}; but configured with an 'interval' in {flow}", script.path);
         }

crates/extensions/tedge_flows/src/js_runtime.rs

@@ -53,7 +53,6 @@ impl JsRuntime {
         for export in exports {
             match export {
                 "onMessage" => script.no_js_on_message_fun = false,
-                "onConfigUpdate" => script.no_js_on_config_update_fun = false,
                 "onInterval" => script.no_js_on_interval_fun = false,
                 _ => (),
             }
@@ -78,7 +77,7 @@ impl JsRuntime {
     ) -> Result<Vec<&'static str>, LoadError> {
         let (sender, receiver) = oneshot::channel();
         let source = source.into();
-        let imports = vec!["onMessage", "onConfigUpdate", "onInterval"];
+        let imports = vec!["onMessage", "onInterval"];
         TIME_CREDITS.store(100000, std::sync::atomic::Ordering::Relaxed);
         self.send(
             receiver,

crates/extensions/tedge_flows/src/js_script.rs

@@ -18,7 +18,6 @@ pub struct JsScript {
     pub interval: Duration,
     pub next_execution: Option<Instant>,
     pub no_js_on_message_fun: bool,
-    pub no_js_on_config_update_fun: bool,
     pub no_js_on_interval_fun: bool,
 }
 
@@ -32,7 +31,6 @@ impl JsScript {
             interval: Duration::ZERO,
             next_execution: None,
             no_js_on_message_fun: true,
-            no_js_on_config_update_fun: true,
             no_js_on_interval_fun: true,
         }
     }
@@ -69,7 +67,7 @@ impl JsScript {
     /// The "onMessage" function of the JS module is passed 3 arguments
     /// - the current timestamp
     /// - the message to be transformed
-    /// - the flow step config (as configured for the flow step, possibly updated by onConfigUpdate messages)
+    /// - the flow step config (as configured in the flow toml)
     ///
     /// The returned value is expected to be an array of messages.
     pub async fn on_message(
@@ -94,32 +92,6 @@ impl JsScript {
             .try_into()
     }
 
-    /// Update the flow step config using a metadata message
-    ///
-    /// The "onConfigUpdate" function of the JS module is passed 2 arguments
-    /// - the message
-    /// - the current flow step config
-    ///
-    /// The value returned by this function is used as the updated flow step config
-    pub async fn on_config_update(
-        &mut self,
-        js: &JsRuntime,
-        message: &Message,
-    ) -> Result<(), FlowError> {
-        debug!(target: "flows", "{}: onConfigUpdate({message})", self.module_name());
-        if self.no_js_on_config_update_fun {
-            return Ok(());
-        }
-
-        let input = vec![message.clone().into(), self.config.clone()];
-        let config = js
-            .call_function(&self.module_name(), "onConfigUpdate", input)
-            .await
-            .map_err(flow::error_from_js)?;
-        self.config = config;
-        Ok(())
-    }
-
     /// Initialize the next execution time for this script's interval
     /// Should be called after the script is loaded and interval is set
     pub fn init_next_execution(&mut self) {

crates/extensions/tedge_flows/src/runtime.rs

@@ -104,14 +104,6 @@ impl<Registry: FlowRegistryExt + Send> MessageProcessor<Registry> {
 
         let mut out_messages = vec![];
         for flow in self.registry.flows_mut() {
-            let config_result = flow
-                .as_mut()
-                .on_config_update(&self.js_runtime, message)
-                .await;
-            if config_result.is_err() {
-                out_messages.push(config_result);
-                continue;
-            }
             if flow.as_ref().accept_message(source, message) {
                 let flow_output = flow
                     .as_mut()

crates/extensions/tedge_flows/src/stats.rs

@@ -14,7 +14,6 @@ pub enum Dimension {
     Flow(String),
     OnMessage(String),
     OnInterval(String),
-    OnConfigUpdate(String),
 }
 
 pub enum Sample {
@@ -177,7 +176,6 @@ impl Display for Dimension {
             Dimension::Flow(toml) => write!(f, "flow {toml}"),
             Dimension::OnMessage(js) => write!(f, "onMessage step {js}"),
             Dimension::OnInterval(js) => write!(f, "onInterval step {js}"),
-            Dimension::OnConfigUpdate(js) => write!(f, "onConfigUpdate step {js}"),
         }
     }
 }
@@ -187,7 +185,6 @@ impl Dimension {
         match f {
             "onMessage" => Some(Dimension::OnMessage(js.to_owned())),
             "onInterval" => Some(Dimension::OnInterval(js.to_owned())),
-            "onConfigUpdate" => Some(Dimension::OnConfigUpdate(js.to_owned())),
             _ => None,
         }
     }

docs/src/references/mappers/flows.md

@@ -56,18 +56,14 @@ A transformation *script* is a JavaScript or TypeScript module that exports:
 
 - at least, a function `onMessage()`, aimed to transform one input message into zero, one or more output messages,
 - possibly, a function `onInterval()`, called at regular intervals to produce aggregated messages,
-- possibly, a function `onConfigUpdate()`, used to update the step config.
 
 ```ts
 interface FlowStep {
     // transform one input message into zero, one or more output messages
     onMessage(message: Message, config: object): null | Message | Message[],
 
     // called at regular intervals to produce aggregated messages
-    onInterval(time: Date, config: object): null | Message | Message[],
-  
-    // update the step config given a config update message
-    onConfigUpdate(message: Message, config: object): object
+    onInterval(time: Date, config: object): null | Message | Message[]
 }
 ```
 
@@ -92,11 +88,6 @@ The `onMessage` function is called for each message to be transformed
     - The config as read from the flow config or updated by the script
   - The function is expected to return zero, one or many transformed messages `[{ topic: string, payload: string }]`
   - An exception can be thrown if the input message cannot be transformed.
-- If defined and associated in the step config with `meta_topics`, the `onConfigUpdate` function is called on each message received on these `meta_topics`.
-  - The arguments are:
-    - The message to be interpreted as a config update `{ topic: string, payload: string }`
-    - The current config
-  - The returned value (an arbitrary JSON value) is then used as the new config for the flow script.
 - A flow script can also export a `onInterval` function
   - This function is called at a regular pace with the current time and config.
   - The flow script can then return zero, one or many transformed messages
@@ -112,18 +103,15 @@ The `onMessage` function is called for each message to be transformed
 - A step is defined by a JavaScript file with an `.mjs` or `.js` extension.
   - This can also be a TypeScript module with a `.ts` extension.
 - The definition of flow defines its input, output and error sink as well as a list of transformation steps.
-  - Each step is built from a javascript and is possibly given a config (arbitrary json that will be passed to the script)
-  - Each step can also subscribe to a list of MQTT meta topics where the metadata about the actual data message is stored
-    (e.g, meta topic of a measurement type where its units threshold values are defined).
-    The messages received on these topics will be passed to the `onConfigUpdate` letting the script update its config.
+- Each step is built from a javascript and is possibly given a config (arbitrary json that will be passed to the script)
 
 ```toml
 input.mqtt.topics = ["te/+/+/+/+/m/+"]
 
 steps = [
     { script = "add_timestamp.js" },
     { script = "drop_stragglers.js", config = { max_delay = 60 } },
-    { script = "te_to_c8y.js", meta_topics = ["te/+/+/+/+/m/+/meta"] }
+    { script = "te_to_c8y.js" }
 ]
 ```
 
@@ -216,7 +204,6 @@ This mapper:
 - loads all the flows defined in `/etc/tedge/flows`
 - reloads any flow or script that is created, updated or deleted while the mapper is running
 - subscribes to each flow `input.mqtt.topics`, dispatching the messages to the `onMessage` functions
-- subscribes to each step `meta_topics`, dispatching the messages to the `onConfigUpdate` functions
 - triggers at the configured pace the `onInterval` functions
 - publishes memory usage statistics
 - publishes flows and steps usage statistics