Fix JS engine with one module per filter instance

Now, each instance of a script is given its own static state

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

Author: didier-wenzek

crates/extensions/tedge_gen_mapper/src/actor.rs

@@ -61,9 +61,7 @@ impl Actor for GenMapper {
                             let Ok(path) = Utf8PathBuf::try_from(path) else {
                                 continue;
                             };
-                            if matches!(path.extension(), Some("js" | "ts")) {
-                                self.processor.add_filter(path).await;
-                            } else if path.extension() == Some("toml") {
+                            if matches!(path.extension(), Some("toml")) {
                                 self.processor.add_pipeline(path).await;
                                 self.send_updated_subscriptions().await?;
                             }

crates/extensions/tedge_gen_mapper/src/config.rs

@@ -3,9 +3,11 @@ use crate::js_runtime::JsRuntime;
 use crate::pipeline::Pipeline;
 use crate::pipeline::Stage;
 use crate::LoadError;
+use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use serde::Deserialize;
 use serde_json::Value;
+use std::fmt::Debug;
 use std::path::Path;
 use tedge_mqtt_ext::TopicFilter;
 
@@ -59,34 +61,43 @@ impl PipelineConfig {
         }
     }
 
-    pub fn compile(
+    pub async fn compile(
         self,
-        js_runtime: &JsRuntime,
+        js_runtime: &mut JsRuntime,
         config_dir: &Path,
         source: Utf8PathBuf,
     ) -> Result<Pipeline, ConfigError> {
-        let input = topic_filters(&self.input_topics)?;
-        let stages = self
-            .stages
-            .into_iter()
-            .map(|stage| stage.compile(js_runtime, config_dir))
-            .collect::<Result<Vec<_>, _>>()?;
+        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?;
+            stages.push(stage);
+        }
         Ok(Pipeline {
-            input_topics: input,
+            input_topics,
             stages,
             source,
         })
     }
 }
 
 impl StageConfig {
-    pub fn compile(self, _js_runtime: &JsRuntime, config_dir: &Path) -> Result<Stage, ConfigError> {
+    pub async fn compile(
+        self,
+        config_dir: &Path,
+        index: usize,
+        pipeline: &Utf8Path,
+    ) -> Result<Stage, ConfigError> {
         let path = match self.filter {
             FilterSpec::JavaScript(path) if path.is_absolute() => path.into(),
             FilterSpec::JavaScript(path) if path.starts_with(config_dir) => path.into(),
             FilterSpec::JavaScript(path) => config_dir.join(path),
         };
-        let filter = JsFilter::new(path)
+        let filter = JsFilter::new(pipeline.to_owned().into(), index, path)
             .with_config(self.config)
             .with_tick_every_seconds(self.tick_every_seconds);
         let config_topics = topic_filters(&self.meta_topics)?;

crates/extensions/tedge_gen_mapper/src/js_filter.rs

@@ -14,6 +14,7 @@ use tracing::debug;
 
 #[derive(Clone)]
 pub struct JsFilter {
+    pub module_name: String,
     pub path: PathBuf,
     pub config: JsonValue,
     pub tick_every_seconds: u64,
@@ -23,16 +24,18 @@ pub struct JsFilter {
 pub struct JsonValue(serde_json::Value);
 
 impl JsFilter {
-    pub fn new(path: PathBuf) -> Self {
+    pub fn new(pipeline: PathBuf, index: usize, path: PathBuf) -> Self {
+        let module_name = format!("{}|{}|{}", pipeline.display(), index, path.display());
         JsFilter {
+            module_name,
             path,
             config: JsonValue::default(),
             tick_every_seconds: 0,
         }
     }
 
     pub fn module_name(&self) -> String {
-        self.path.display().to_string()
+        self.module_name.to_owned()
     }
 
     pub fn with_config(self, config: Option<serde_json::Value>) -> Self {
@@ -77,7 +80,7 @@ impl JsFilter {
             message.clone().into(),
             self.config.clone(),
         ];
-        js.call_function(&self.path, "process", input)
+        js.call_function(&self.module_name(), "process", input)
             .await
             .map_err(pipeline::error_from_js)?
             .try_into()
@@ -98,7 +101,7 @@ impl JsFilter {
         debug!(target: "MAPPING", "{}: update_config({message:?})", self.module_name());
         let input = vec![message.clone().into(), self.config.clone()];
         let config = js
-            .call_function(&self.path, "update_config", input)
+            .call_function(&self.module_name(), "update_config", input)
             .await
             .map_err(pipeline::error_from_js)?;
         self.config = config;
@@ -122,7 +125,7 @@ impl JsFilter {
         }
         debug!(target: "MAPPING", "{}: tick({timestamp:?})", self.module_name());
         let input = vec![timestamp.clone().into(), self.config.clone()];
-        js.call_function(&self.path, "tick", input)
+        js.call_function(&self.module_name(), "tick", input)
             .await
             .map_err(pipeline::error_from_js)?
             .try_into()
@@ -281,8 +284,8 @@ mod tests {
     async fn identity_filter() {
         let script = "export function process(t,msg) { return [msg]; };";
         let mut runtime = JsRuntime::try_new().await.unwrap();
-        runtime.load_js("id.js", script).await.unwrap();
-        let filter = JsFilter::new("id.js".into());
+        let filter = JsFilter::new("id.toml".into(), 1, "id.js".into());
+        runtime.load_js(filter.module_name(), script).await.unwrap();
 
         let input = Message::new("te/main/device///m/", "hello world");
         let output = input.clone();
@@ -299,8 +302,8 @@ mod tests {
     async fn error_filter() {
         let script = r#"export function process(t,msg) { throw new Error("Cannot process that message"); };"#;
         let mut runtime = JsRuntime::try_new().await.unwrap();
-        runtime.load_js("err.js", script).await.unwrap();
-        let filter = JsFilter::new("err.js".into());
+        let filter = JsFilter::new("err.toml".into(), 1, "err.js".into());
+        runtime.load_js(filter.module_name(), script).await.unwrap();
 
         let input = Message::new("te/main/device///m/", "hello world");
         let error = filter
@@ -335,8 +338,8 @@ export function process (timestamp, message, config) {
 }
         "#;
         let mut runtime = JsRuntime::try_new().await.unwrap();
-        runtime.load_js("collectd.js", script).await.unwrap();
-        let filter = JsFilter::new("collectd.js".into());
+        let filter = JsFilter::new("collectd.toml".into(), 1, "collectd.js".into());
+        runtime.load_js(filter.module_name(), script).await.unwrap();
 
         let input = Message::new(
             "collectd/h/memory/percent-used",

crates/extensions/tedge_gen_mapper/src/js_runtime.rs

@@ -23,20 +23,22 @@ impl JsRuntime {
         Ok(JsRuntime { runtime, worker })
     }
 
-    pub async fn load_file(&mut self, path: impl AsRef<Path>) -> Result<(), LoadError> {
+    pub async fn load_file(
+        &mut self,
+        module_name: String,
+        path: impl AsRef<Path>,
+    ) -> Result<(), LoadError> {
         let path = path.as_ref();
         let source = tokio::fs::read_to_string(path).await?;
-        self.load_js(path, source).await
+        self.load_js(module_name, source).await
     }
 
     pub async fn load_js(
         &mut self,
-        path: impl AsRef<Path>,
+        name: String,
         source: impl Into<Vec<u8>>,
     ) -> Result<(), LoadError> {
         let (sender, receiver) = oneshot::channel();
-        let path = path.as_ref().to_path_buf();
-        let name = path.display().to_string();
         let source = source.into();
         self.worker
             .send(JsRequest::LoadModule {
@@ -51,14 +53,14 @@ impl JsRuntime {
 
     pub async fn call_function(
         &self,
-        module: &Path,
+        module: &str,
         function: &str,
         args: Vec<JsonValue>,
     ) -> Result<JsonValue, LoadError> {
         let (sender, receiver) = oneshot::channel();
         self.worker
             .send(JsRequest::CallFunction {
-                module: module.display().to_string(),
+                module: module.to_string(),
                 function: function.to_string(),
                 args,
                 sender,