Warn when javascript module and config are not consistent

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

Author: didier-wenzek

crates/common/mqtt_channel/src/topics.rs

@@ -116,6 +116,10 @@ impl TopicFilter {
         }
     }
 
+    pub fn is_empty(&self) -> bool {
+        self.patterns.is_empty()
+    }
+
     /// Check if the given topic matches this filter pattern.
     pub fn accept_topic_name(&self, topic: &str) -> bool {
         self.patterns

crates/extensions/tedge_gen_mapper/src/config.rs

@@ -70,11 +70,9 @@ impl PipelineConfig {
         let input_topics = topic_filters(&self.input_topics)?;
         let mut stages = vec![];
         for (i, stage) in self.stages.into_iter().enumerate() {
-            let stage = stage.compile(config_dir, i, &source).await?;
-            let filter = &stage.filter;
-            js_runtime
-                .load_file(filter.module_name(), filter.path())
-                .await?;
+            let mut stage = stage.compile(config_dir, i, &source).await?;
+            js_runtime.load_filter(&mut stage.filter).await?;
+            stage.check(&source);
             stages.push(stage);
         }
         Ok(Pipeline {

crates/extensions/tedge_gen_mapper/src/js_filter.rs

@@ -18,6 +18,9 @@ pub struct JsFilter {
     pub path: PathBuf,
     pub config: JsonValue,
     pub tick_every_seconds: u64,
+    pub no_js_process: bool,
+    pub no_js_update_config: bool,
+    pub no_js_tick: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -37,6 +40,9 @@ impl JsFilter {
             path,
             config: JsonValue::default(),
             tick_every_seconds: 0,
+            no_js_process: true,
+            no_js_update_config: true,
+            no_js_tick: true,
         }
     }
 
@@ -81,6 +87,10 @@ impl JsFilter {
         message: &Message,
     ) -> Result<Vec<Message>, FilterError> {
         debug!(target: "MAPPING", "{}: process({timestamp:?}, {message:?})", self.module_name());
+        if self.no_js_process {
+            return Ok(vec![message.clone()]);
+        }
+
         let input = vec![
             timestamp.clone().into(),
             message.clone().into(),
@@ -105,6 +115,10 @@ impl JsFilter {
         message: &Message,
     ) -> Result<(), FilterError> {
         debug!(target: "MAPPING", "{}: update_config({message:?})", self.module_name());
+        if self.no_js_update_config {
+            return Ok(());
+        }
+
         let input = vec![message.clone().into(), self.config.clone()];
         let config = js
             .call_function(&self.module_name(), "update_config", input)
@@ -126,6 +140,9 @@ impl JsFilter {
         js: &JsRuntime,
         timestamp: &DateTime,
     ) -> Result<Vec<Message>, FilterError> {
+        if self.no_js_tick {
+            return Ok(vec![]);
+        }
         if !timestamp.tick_now(self.tick_every_seconds) {
             return Ok(vec![]);
         }

crates/extensions/tedge_gen_mapper/src/js_runtime.rs

@@ -1,3 +1,4 @@
+use crate::js_filter::JsFilter;
 use crate::js_filter::JsonValue;
 use crate::LoadError;
 use anyhow::anyhow;
@@ -23,11 +24,24 @@ impl JsRuntime {
         Ok(JsRuntime { runtime, worker })
     }
 
+    pub async fn load_filter(&mut self, filter: &mut JsFilter) -> Result<(), LoadError> {
+        let exports = self.load_file(filter.module_name(), filter.path()).await?;
+        for export in exports {
+            match export {
+                "process" => filter.no_js_process = false,
+                "update_config" => filter.no_js_update_config = false,
+                "tick" => filter.no_js_tick = false,
+                _ => (),
+            }
+        }
+        Ok(())
+    }
+
     pub async fn load_file(
         &mut self,
         module_name: String,
         path: impl AsRef<Path>,
-    ) -> Result<(), LoadError> {
+    ) -> Result<Vec<&'static str>, LoadError> {
         let path = path.as_ref();
         let source = tokio::fs::read_to_string(path).await?;
         self.load_js(module_name, source).await
@@ -37,13 +51,15 @@ impl JsRuntime {
         &mut self,
         name: String,
         source: impl Into<Vec<u8>>,
-    ) -> Result<(), LoadError> {
+    ) -> Result<Vec<&'static str>, LoadError> {
         let (sender, receiver) = oneshot::channel();
         let source = source.into();
+        let imports = vec!["process", "update_config", "tick"];
         self.worker
             .send(JsRequest::LoadModule {
                 name,
                 source,
+                imports,
                 sender,
             })
             .await
@@ -87,7 +103,8 @@ enum JsRequest {
     LoadModule {
         name: String,
         source: Vec<u8>,
-        sender: oneshot::Sender<Result<(), LoadError>>,
+        imports: Vec<&'static str>,
+        sender: oneshot::Sender<Result<Vec<&'static str>, LoadError>>,
     },
     CallFunction {
         module: String,
@@ -118,8 +135,8 @@ impl JsWorker {
             let mut modules = JsModules::new();
             while let Some(request) = self.requests.recv().await {
                 match request {
-                    JsRequest::LoadModule{name, source, sender} => {
-                        let result = modules.load_module(ctx.clone(), name, source).await;
+                    JsRequest::LoadModule{name, source, sender, imports} => {
+                        let result = modules.load_module(ctx.clone(), name, source, imports).await;
                         let _ = sender.send(result);
                     }
                     JsRequest::CallFunction{module, function, args, sender} => {
@@ -149,13 +166,24 @@ impl<'js> JsModules<'js> {
         ctx: Ctx<'js>,
         name: String,
         source: Vec<u8>,
-    ) -> Result<(), LoadError> {
+        imports: Vec<&'static str>,
+    ) -> Result<Vec<&'static str>, LoadError> {
         debug!(target: "MAPPING", "compile({name})");
         let module = Module::declare(ctx, name.clone(), source)?;
         let (module, p) = module.eval()?;
         let () = p.finish()?;
+
+        let mut exports = vec![];
+        for import in imports {
+            if let Ok(Some(v)) = module.get(import) {
+                if rquickjs::Function::from_value(v).is_ok() {
+                    exports.push(import);
+                }
+            }
+        }
+
         self.modules.insert(name, module);
-        Ok(())
+        Ok(exports)
     }
 
     async fn call_function(
@@ -178,7 +206,10 @@ impl<'js> JsModules<'js> {
                 module_name: module_name.clone(),
                 function: function.clone(),
             })?;
-        let f = rquickjs::Function::from_value(f)?;
+        let f = rquickjs::Function::from_value(f).map_err(|_| LoadError::UnknownFunction {
+            module_name: module_name.clone(),
+            function: function.clone(),
+        })?;
 
         let r = match &args[..] {
             [] => f.call(()),

crates/extensions/tedge_gen_mapper/src/pipeline.rs

@@ -1,12 +1,14 @@
 use crate::js_filter::JsFilter;
 use crate::js_runtime::JsRuntime;
 use crate::LoadError;
+use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use serde_json::json;
 use serde_json::Value;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::TopicFilter;
 use time::OffsetDateTime;
+use tracing::warn;
 
 /// A chain of transformation of MQTT messages
 pub struct Pipeline {
@@ -118,6 +120,21 @@ impl Pipeline {
     }
 }
 
+impl Stage {
+    pub(crate) fn check(&self, pipeline: &Utf8Path) {
+        let filter = &self.filter;
+        if filter.no_js_process {
+            warn!(target: "MAPPING", "Filter with no 'process' function: {}", filter.path.display());
+        }
+        if filter.no_js_update_config && !self.config_topics.is_empty() {
+            warn!(target: "MAPPING", "Filter with no 'config_update' function: {}; but configured with 'config_topics' in {pipeline}", filter.path.display());
+        }
+        if filter.no_js_tick && filter.tick_every_seconds != 0 {
+            warn!(target: "MAPPING", "Filter with no 'tick' function: {}; but configured with 'tick_every_seconds' in {pipeline}", filter.path.display());
+        }
+    }
+}
+
 impl DateTime {
     pub fn now() -> Self {
         DateTime::try_from(OffsetDateTime::now_utc()).unwrap()

crates/extensions/tedge_gen_mapper/src/runtime.rs

@@ -116,11 +116,7 @@ impl MessageProcessor {
         for pipeline in self.pipelines.values_mut() {
             for stage in &mut pipeline.stages {
                 if stage.filter.path() == path {
-                    match self
-                        .js_runtime
-                        .load_file(stage.filter.module_name(), &path)
-                        .await
-                    {
+                    match self.js_runtime.load_filter(&mut stage.filter).await {
                         Ok(()) => {
                             info!(target: "gen-mapper", "Reloaded filter {path}");
                         }