Tedge mapping test a single single filter

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

Author: didier-wenzek

crates/core/tedge/src/cli/mapping/cli.rs

@@ -108,4 +108,19 @@ impl TEdgeMappingCli {
                 )
             })
     }
+
+    pub async fn load_filter(
+        mapping_dir: &PathBuf,
+        path: &PathBuf,
+    ) -> Result<MessageProcessor, Error> {
+        if let Some("toml") = path.extension().and_then(|s| s.to_str()) {
+            MessageProcessor::try_new_single_pipeline(mapping_dir, path)
+                .await
+                .with_context(|| format!("loading pipeline {pipeline}", pipeline = path.display()))
+        } else {
+            MessageProcessor::try_new_single_filter(mapping_dir, path)
+                .await
+                .with_context(|| format!("loading filter {filter}", filter = path.display()))
+        }
+    }
 }

crates/core/tedge/src/cli/mapping/test.rs

@@ -27,7 +27,10 @@ impl Command for TestCommand {
     }
 
     async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
-        let mut processor = TEdgeMappingCli::load_pipelines(&self.mapping_dir).await?;
+        let mut processor = match &self.filter {
+            None => TEdgeMappingCli::load_pipelines(&self.mapping_dir).await?,
+            Some(filter) => TEdgeMappingCli::load_filter(&self.mapping_dir, filter).await?,
+        };
         if let Some(message) = &self.message {
             let timestamp = DateTime::now();
             self.process(&mut processor, message, &timestamp).await;
@@ -53,37 +56,21 @@ impl TestCommand {
         message: &Message,
         timestamp: &DateTime,
     ) {
-        match &self.filter {
-            Some(filter) => {
-                let filter_name = filter.display().to_string();
-                print(
-                    processor
-                        .process_with_pipeline(&filter_name, timestamp, message)
-                        .await,
-                )
-            }
-            None => processor
-                .process(timestamp, message)
-                .await
-                .into_iter()
-                .map(|(_, v)| v)
-                .for_each(print),
-        }
+        processor
+            .process(timestamp, message)
+            .await
+            .into_iter()
+            .map(|(_, v)| v)
+            .for_each(print)
     }
 
     async fn tick(&self, processor: &mut MessageProcessor, timestamp: &DateTime) {
-        match &self.filter {
-            Some(filter) => {
-                let filter_name = filter.display().to_string();
-                print(processor.tick_with_pipeline(&filter_name, timestamp).await)
-            }
-            None => processor
-                .tick(timestamp)
-                .await
-                .into_iter()
-                .map(|(_, v)| v)
-                .for_each(print),
-        }
+        processor
+            .tick(timestamp)
+            .await
+            .into_iter()
+            .map(|(_, v)| v)
+            .for_each(print)
     }
 }
 

crates/extensions/tedge_gen_mapper/src/config.rs

@@ -45,6 +45,20 @@ pub enum ConfigError {
 }
 
 impl PipelineConfig {
+    pub fn from_filter(filter: Utf8PathBuf) -> Self {
+        let input_topic = "#".to_string();
+        let stage = StageConfig {
+            filter: FilterSpec::JavaScript(filter),
+            config: None,
+            tick_every_seconds: 1,
+            meta_topics: vec![],
+        };
+        Self {
+            input_topics: vec![input_topic],
+            stages: vec![stage],
+        }
+    }
+
     pub fn compile(
         self,
         js_runtime: &JsRuntime,
@@ -69,6 +83,7 @@ impl StageConfig {
     pub fn compile(self, _js_runtime: &JsRuntime, config_dir: &Path) -> 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)

crates/extensions/tedge_gen_mapper/src/runtime.rs

@@ -42,6 +42,43 @@ impl MessageProcessor {
         })
     }
 
+    pub async fn try_new_single_pipeline(
+        config_dir: impl AsRef<Path>,
+        pipeline: impl AsRef<Path>,
+    ) -> Result<Self, LoadError> {
+        let config_dir = config_dir.as_ref().to_owned();
+        let pipeline = pipeline.as_ref().to_owned();
+        let mut js_runtime = JsRuntime::try_new().await?;
+        let mut pipeline_specs = PipelineSpecs::default();
+        pipeline_specs
+            .load_single_pipeline(&mut js_runtime, &config_dir, &pipeline)
+            .await;
+        let pipelines = pipeline_specs.compile(&js_runtime, &config_dir);
+        Ok(MessageProcessor {
+            config_dir,
+            pipelines,
+            js_runtime,
+        })
+    }
+
+    pub async fn try_new_single_filter(
+        config_dir: impl AsRef<Path>,
+        filter: impl AsRef<Path>,
+    ) -> Result<Self, LoadError> {
+        let config_dir = config_dir.as_ref().to_owned();
+        let mut js_runtime = JsRuntime::try_new().await?;
+        let mut pipeline_specs = PipelineSpecs::default();
+        pipeline_specs
+            .load_single_filter(&mut js_runtime, &filter)
+            .await;
+        let pipelines = pipeline_specs.compile(&js_runtime, &config_dir);
+        Ok(MessageProcessor {
+            config_dir,
+            pipelines,
+            js_runtime,
+        })
+    }
+
     pub fn subscriptions(&self) -> TopicFilter {
         let mut topics = TopicFilter::empty();
         for pipeline in self.pipelines.values() {
@@ -63,19 +100,6 @@ impl MessageProcessor {
         out_messages
     }
 
-    pub async fn process_with_pipeline(
-        &mut self,
-        pipeline_id: &String,
-        timestamp: &DateTime,
-        message: &Message,
-    ) -> Result<Vec<Message>, FilterError> {
-        let pipeline = self
-            .pipelines
-            .get_mut(pipeline_id)
-            .ok_or_else(|| anyhow::anyhow!("No such pipeline: {pipeline_id}"))?;
-        pipeline.process(&self.js_runtime, timestamp, message).await
-    }
-
     pub async fn tick(
         &mut self,
         timestamp: &DateTime,
@@ -88,18 +112,6 @@ impl MessageProcessor {
         out_messages
     }
 
-    pub async fn tick_with_pipeline(
-        &mut self,
-        pipeline_id: &String,
-        timestamp: &DateTime,
-    ) -> Result<Vec<Message>, FilterError> {
-        let pipeline = self
-            .pipelines
-            .get_mut(pipeline_id)
-            .ok_or_else(|| anyhow::anyhow!("No such pipeline: {pipeline_id}"))?;
-        pipeline.tick(&self.js_runtime, timestamp).await
-    }
-
     pub async fn dump_memory_stats(&self) {
         self.js_runtime.dump_memory_stats().await;
     }
@@ -235,10 +247,71 @@ impl PipelineSpecs {
         }
     }
 
+    pub async fn load_single_pipeline(
+        &mut self,
+        js_runtime: &mut JsRuntime,
+        config_dir: &PathBuf,
+        pipeline: &Path,
+    ) {
+        let Some(path) = Utf8Path::from_path(pipeline).map(|p| p.to_path_buf()) else {
+            error!(target: "MAPPING", "Skipping non UTF8 path: {}", pipeline.display());
+            return;
+        };
+        if let Err(err) = self.load_pipeline(&path).await {
+            error!(target: "MAPPING", "Failed to load pipeline {path}: {err}");
+            return;
+        }
+
+        let Ok(mut entries) = read_dir(config_dir).await.map_err(|err|
+            error!(target: "MAPPING", "Failed to read filters from {}: {err}", config_dir.display())
+        ) else {
+            return;
+        };
+
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            let Some(path) = Utf8Path::from_path(&entry.path()).map(|p| p.to_path_buf()) else {
+                error!(target: "MAPPING", "Skipping non UTF8 path: {}", entry.path().display());
+                continue;
+            };
+            if let Ok(file_type) = entry.file_type().await {
+                if file_type.is_file() {
+                    match path.extension() {
+                        Some("js") | Some("ts") => {
+                            info!(target: "MAPPING", "Loading filter: {path}");
+                            if let Err(err) = self.load_filter(js_runtime, path).await {
+                                error!(target: "MAPPING", "Failed to load filter: {err}");
+                            }
+                        }
+                        _ => {}
+                    }
+                }
+            }
+        }
+    }
+
+    pub async fn load_single_filter(
+        &mut self,
+        js_runtime: &mut JsRuntime,
+        filter: impl AsRef<Path>,
+    ) {
+        let filter = filter.as_ref();
+        let Some(path) = Utf8Path::from_path(filter).map(|p| p.to_path_buf()) else {
+            error!(target: "MAPPING", "Skipping non UTF8 path: {}", filter.display());
+            return;
+        };
+        if let Err(err) = js_runtime.load_file(&path).await {
+            error!(target: "MAPPING", "Failed to load filter {path}: {err}");
+        }
+        let pipeline_id = MessageProcessor::pipeline_id(&path);
+        let pipeline = PipelineConfig::from_filter(path.to_owned());
+        self.pipeline_specs
+            .insert(pipeline_id, (path.to_owned(), pipeline));
+    }
+
     async fn load_pipeline(&mut self, file: impl AsRef<Utf8Path>) -> Result<(), LoadError> {
         let path = file.as_ref();
         let pipeline_id = MessageProcessor::pipeline_id(path);
-        let specs = read_to_string(file.as_ref()).await?;
+        let specs = read_to_string(path).await?;
         let pipeline: PipelineConfig = toml::from_str(&specs)?;
         self.pipeline_specs
             .insert(pipeline_id, (path.to_owned(), pipeline));
@@ -249,9 +322,9 @@ impl PipelineSpecs {
     async fn load_filter(
         &mut self,
         js_runtime: &mut JsRuntime,
-        file: impl AsRef<Utf8Path>,
+        file: impl AsRef<Path>,
     ) -> Result<(), LoadError> {
-        js_runtime.load_file(file.as_ref()).await?;
+        js_runtime.load_file(file).await?;
         Ok(())
     }
 

tests/RobotFramework/tests/tedge_gen_mapper/pipelines/average.toml

@@ -50,7 +50,7 @@ Units are configured using topic metadata
 
 Computing average over a time window
     ${transformed_msg}    Execute Command
-    ...    cat /etc/tedge/gen-mapper/average.samples | awk '{ print $2 }' FS\='INPUT:' | tedge mapping test --final-tick
+    ...    cat /etc/tedge/gen-mapper/average.samples | awk '{ print $2 }' FS\='INPUT:' | tedge mapping test --final-tick --filter /etc/tedge/gen-mapper/average.js
     ...    strip=True
     ${expected_msg}    Execute Command
     ...    cat /etc/tedge/gen-mapper/average.samples | awk '{ if ($2) print $2 }' FS\='OUTPUT: '