poc: tedge MEA DB

Cargo.toml

@@ -55,6 +55,7 @@ tedge_config_macros-impl = { path = "crates/common/tedge_config_macros/impl" }
 tedge_config_manager = { path = "crates/extensions/tedge_config_manager" }
 tedge_downloader_ext = { path = "crates/extensions/tedge_downloader_ext" }
 tedge_file_system_ext = { path = "crates/extensions/tedge_file_system_ext" }
+tedge_gen_mapper = { path = "crates/extensions/tedge_gen_mapper" }
 tedge_health_ext = { path = "crates/extensions/tedge_health_ext" }
 tedge_http_ext = { path = "crates/extensions/tedge_http_ext" }
 tedge_log_manager = { path = "crates/extensions/tedge_log_manager" }
@@ -109,6 +110,7 @@ env_logger = "0.10"
 fastrand = "2.0"
 figment = { version = "0.10" }
 filetime = "0.2"
+fjall = "2.11"
 flate2 = "1.1.1"
 freedesktop_entry_parser = "1.3.0"
 futures = "0.3"
@@ -163,6 +165,7 @@ regex = "1.4"
 reqwest = { version = "0.12", default-features = false }
 ron = "0.8"
 rpassword = "5.0"
+rquickjs = { version = "0.9", default-features = false }
 rstest = "0.16.0"
 rumqttc = { git = "https://github.com/jarhodes314/rumqtt", rev = "8c489faf6af910956c97b55587ff3ecb2ac4e96f" }
 rumqttd = "0.19"

ci/build_scripts/build.sh

@@ -101,7 +101,7 @@ BUILD_WITH="${BUILD_WITH:-zig}"
 COMMON_BUILD_OPTIONS=(
     "--release"
 )
-TOOLCHAIN="${TOOLCHAIN:-+1.78}"
+TOOLCHAIN="${TOOLCHAIN:-+1.82}"
 # Note: Minimum version that is supported with riscv64gc-unknown-linux-gnu is 2.27
 GLIBC_VERSION="${GLIBC_VERSION:-2.17}"
 RISCV_GLIBC_VERSION="${RISCV_GLIBC_VERSION:-2.27}"

crates/common/mqtt_channel/src/topics.rs

@@ -117,10 +117,15 @@ impl TopicFilter {
     }
 
     /// Check if the given topic matches this filter pattern.
-    pub fn accept_topic(&self, topic: &Topic) -> bool {
+    pub fn accept_topic_name(&self, topic: &str) -> bool {
         self.patterns
             .iter()
-            .any(|pattern| rumqttc::matches(&topic.name, pattern))
+            .any(|pattern| rumqttc::matches(topic, pattern))
+    }
+
+    /// Check if the given topic matches this filter pattern.
+    pub fn accept_topic(&self, topic: &Topic) -> bool {
+        self.accept_topic_name(&topic.name)
     }
 
     /// Check if the given message matches this filter pattern.

crates/core/plugin_sm/src/operation_logs.rs

@@ -152,7 +152,7 @@ mod tests {
         let unrelated_2 = create_file(log_dir.path(), "bar");
 
         // Open the log dir
-        let _operation_logs = OperationLogs::try_new(log_dir.into_path().try_into().unwrap())?;
+        let _operation_logs = OperationLogs::try_new(log_dir.keep().try_into().unwrap())?;
 
         // Outdated logs are removed
         assert!(!update_log_1.exists());

crates/core/tedge/Cargo.toml

@@ -54,6 +54,7 @@ tedge-watchdog = { workspace = true }
 tedge-write = { workspace = true }
 tedge_api = { workspace = true }
 tedge_config = { workspace = true }
+tedge_gen_mapper = { workspace = true }
 tedge_utils = { workspace = true }
 thiserror = { workspace = true }
 time = { workspace = true }

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

@@ -0,0 +1,126 @@
+use crate::cli::mapping::list::ListCommand;
+use crate::cli::mapping::test::TestCommand;
+use crate::command::BuildCommand;
+use crate::command::Command;
+use crate::ConfigError;
+use anyhow::anyhow;
+use anyhow::Context;
+use anyhow::Error;
+use std::path::PathBuf;
+use tedge_config::TEdgeConfig;
+use tedge_gen_mapper::pipeline::Message;
+use tedge_gen_mapper::MessageProcessor;
+
+#[derive(clap::Subcommand, Debug)]
+pub enum TEdgeMappingCli {
+    /// List pipelines and filters
+    List {
+        /// Path to pipeline and filter specs
+        ///
+        /// Default to /etc/tedge/gen-mapper
+        #[clap(long)]
+        mapping_dir: Option<PathBuf>,
+
+        /// List pipelines processing messages published on this topic
+        ///
+        /// If none is provided, lists all the pipelines
+        #[clap(long)]
+        topic: Option<String>,
+    },
+
+    /// Process message samples
+    Test {
+        /// Path to pipeline and filter specs
+        ///
+        /// Default to /etc/tedge/gen-mapper
+        #[clap(long)]
+        mapping_dir: Option<PathBuf>,
+
+        /// Path to the javascript filter or TOML pipeline definition
+        ///
+        /// If none is provided, applies all the matching pipelines
+        #[clap(long)]
+        filter: Option<PathBuf>,
+
+        /// Send a tick after all the message samples
+        #[clap(long = "final-tick")]
+        final_tick: bool,
+
+        /// Topic of the message sample
+        ///
+        /// If none is provided, messages are read from stdin expecting a line per message:
+        /// [topic] payload
+        topic: Option<String>,
+
+        /// Payload of the message sample
+        ///
+        /// If none is provided, payloads are read from stdin
+        payload: Option<String>,
+    },
+}
+
+impl BuildCommand for TEdgeMappingCli {
+    fn build_command(self, config: &TEdgeConfig) -> Result<Box<dyn Command>, ConfigError> {
+        match self {
+            TEdgeMappingCli::List { mapping_dir, topic } => {
+                let mapping_dir = mapping_dir.unwrap_or_else(|| Self::default_mapping_dir(config));
+                Ok(ListCommand { mapping_dir, topic }.into_boxed())
+            }
+
+            TEdgeMappingCli::Test {
+                mapping_dir,
+                filter,
+                final_tick,
+                topic,
+                payload,
+            } => {
+                let mapping_dir = mapping_dir.unwrap_or_else(|| Self::default_mapping_dir(config));
+                let message = match (topic, payload) {
+                    (Some(topic), Some(payload)) => Some(Message { topic, payload }),
+                    (Some(_), None) => Err(anyhow!("Missing sample payload"))?,
+                    (None, Some(_)) => Err(anyhow!("Missing sample topic"))?,
+                    (None, None) => None,
+                };
+                Ok(TestCommand {
+                    mapping_dir,
+                    filter,
+                    message,
+                    final_tick,
+                }
+                .into_boxed())
+            }
+        }
+    }
+}
+
+impl TEdgeMappingCli {
+    fn default_mapping_dir(config: &TEdgeConfig) -> PathBuf {
+        config.root_dir().join("gen-mapper").into()
+    }
+
+    pub async fn load_pipelines(mapping_dir: &PathBuf) -> Result<MessageProcessor, Error> {
+        MessageProcessor::try_new(mapping_dir)
+            .await
+            .with_context(|| {
+                format!(
+                    "loading pipelines and filters from {}",
+                    mapping_dir.display()
+                )
+            })
+    }
+
+    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/list.rs

@@ -0,0 +1,47 @@
+use crate::cli::mapping::TEdgeMappingCli;
+use crate::command::Command;
+use crate::log::MaybeFancy;
+use anyhow::Error;
+use std::path::PathBuf;
+use tedge_config::TEdgeConfig;
+use tedge_gen_mapper::pipeline::Pipeline;
+
+pub struct ListCommand {
+    pub mapping_dir: PathBuf,
+    pub topic: Option<String>,
+}
+
+#[async_trait::async_trait]
+impl Command for ListCommand {
+    fn description(&self) -> String {
+        format!(
+            "list pipelines and filters in {:}",
+            self.mapping_dir.display()
+        )
+    }
+
+    async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
+        let processor = TEdgeMappingCli::load_pipelines(&self.mapping_dir).await?;
+
+        match &self.topic {
+            Some(topic) => processor
+                .pipelines
+                .iter()
+                .filter(|(_, pipeline)| pipeline.topics().accept_topic_name(topic))
+                .for_each(Self::display),
+
+            None => processor.pipelines.iter().for_each(Self::display),
+        }
+
+        Ok(())
+    }
+}
+
+impl ListCommand {
+    fn display((pipeline_id, pipeline): (&String, &Pipeline)) {
+        println!("{pipeline_id}");
+        for stage in pipeline.stages.iter() {
+            println!("\t{}", stage.filter.path.display());
+        }
+    }
+}

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

@@ -0,0 +1,5 @@
+mod cli;
+mod list;
+mod test;
+
+pub use cli::TEdgeMappingCli;

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

@@ -0,0 +1,135 @@
+use crate::cli::mapping::TEdgeMappingCli;
+use crate::command::Command;
+use crate::log::MaybeFancy;
+use anyhow::Error;
+use std::path::PathBuf;
+use tedge_config::TEdgeConfig;
+use tedge_gen_mapper::pipeline::*;
+use tedge_gen_mapper::MessageProcessor;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::BufReader;
+use tokio::io::Stdin;
+
+pub struct TestCommand {
+    pub mapping_dir: PathBuf,
+    pub filter: Option<PathBuf>,
+    pub message: Option<Message>,
+    pub final_tick: bool,
+}
+
+#[async_trait::async_trait]
+impl Command for TestCommand {
+    fn description(&self) -> String {
+        format!(
+            "process message samples using pipelines and filters in {:}",
+            self.mapping_dir.display()
+        )
+    }
+
+    async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
+        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;
+        } else {
+            let mut stdin = BufReader::new(tokio::io::stdin());
+            while let Some(message) = next_message(&mut stdin).await {
+                let timestamp = DateTime::now();
+                self.process(&mut processor, &message, &timestamp).await;
+            }
+        }
+        if self.final_tick {
+            let timestamp = DateTime::now();
+            self.tick(&mut processor, &timestamp).await;
+        }
+        Ok(())
+    }
+}
+
+impl TestCommand {
+    async fn process(
+        &self,
+        processor: &mut MessageProcessor,
+        message: &Message,
+        timestamp: &DateTime,
+    ) {
+        processor
+            .process(MessageSource::MQTT, timestamp, message)
+            .await
+            .into_iter()
+            .map(|(_, v)| v)
+            .for_each(print)
+    }
+
+    async fn tick(&self, processor: &mut MessageProcessor, timestamp: &DateTime) {
+        processor
+            .tick(timestamp)
+            .await
+            .into_iter()
+            .map(|(_, v)| v)
+            .for_each(print)
+    }
+}
+
+fn print(messages: Result<Vec<Message>, FilterError>) {
+    match messages {
+        Ok(messages) => {
+            for message in messages {
+                println!("[{}] {}", message.topic, message.payload);
+            }
+        }
+        Err(err) => {
+            eprintln!("Error: {}", err)
+        }
+    }
+}
+
+fn parse(line: String) -> Result<Option<Message>, Error> {
+    let line = line.trim();
+    if line.is_empty() {
+        return Ok(None);
+    }
+    if !line.starts_with("[") {
+        return Err(anyhow::anyhow!("Missing opening bracket: {}", line));
+    }
+    let Some(closing_bracket) = line.find(']') else {
+        return Err(anyhow::anyhow!("Missing closing bracket: {}", line));
+    };
+
+    let topic = line[1..closing_bracket].to_string();
+    let payload = line[closing_bracket + 1..].to_string();
+
+    Ok(Some(Message { topic, payload }))
+}
+
+async fn next_line(input: &mut BufReader<Stdin>) -> Option<String> {
+    loop {
+        let mut line = String::new();
+        match input.read_line(&mut line).await {
+            Ok(0) => return None,
+            Ok(_) => {
+                let line = line.trim();
+                if !line.is_empty() {
+                    return Some(line.to_string());
+                }
+            }
+            Err(err) => {
+                eprintln!("Fail to read input stream {}", err);
+                return None;
+            }
+        }
+    }
+}
+async fn next_message(input: &mut BufReader<Stdin>) -> Option<Message> {
+    let line = next_line(input).await?;
+    match parse(line) {
+        Ok(message) => message,
+        Err(err) => {
+            eprintln!("Fail to parse input message {}", err);
+            None
+        }
+    }
+}