Merge pull request #3799 from didier-wenzek/refactor/flows-paths

refactor: use more specific types

Author: jarhodes314

Cargo.lock

@@ -6,7 +6,7 @@ use crate::ConfigError;
 use anyhow::anyhow;
 use anyhow::Context;
 use anyhow::Error;
-use std::path::PathBuf;
+use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
 use tedge_flows::flow::Message;
 use tedge_flows::MessageProcessor;
@@ -19,7 +19,7 @@ pub enum TEdgeFlowsCli {
         ///
         /// Default to /etc/tedge/flows
         #[clap(long)]
-        flows_dir: Option<PathBuf>,
+        flows_dir: Option<Utf8PathBuf>,
 
         /// List flows processing messages published on this topic
         ///
@@ -34,13 +34,13 @@ pub enum TEdgeFlowsCli {
         ///
         /// Default to /etc/tedge/flows
         #[clap(long)]
-        flows_dir: Option<PathBuf>,
+        flows_dir: Option<Utf8PathBuf>,
 
         /// Path to the flow step script or TOML flow definition
         ///
         /// If none is provided, applies all the matching flows
         #[clap(long)]
-        flow: Option<PathBuf>,
+        flow: Option<Utf8PathBuf>,
 
         /// Trigger onInterval after all the message samples
         #[clap(long = "final-on-interval")]
@@ -106,25 +106,28 @@ impl BuildCommand for TEdgeFlowsCli {
 }
 
 impl TEdgeFlowsCli {
-    fn default_flows_dir(config: &TEdgeConfig) -> PathBuf {
-        config.root_dir().join("flows").into()
+    fn default_flows_dir(config: &TEdgeConfig) -> Utf8PathBuf {
+        config.root_dir().join("flows")
     }
 
-    pub async fn load_flows(flows_dir: &PathBuf) -> Result<MessageProcessor, Error> {
+    pub async fn load_flows(flows_dir: &Utf8PathBuf) -> Result<MessageProcessor, Error> {
         MessageProcessor::try_new(flows_dir)
             .await
-            .with_context(|| format!("loading flows and steps from {}", flows_dir.display()))
+            .with_context(|| format!("loading flows and steps from {flows_dir}"))
     }
 
-    pub async fn load_file(flows_dir: &PathBuf, path: &PathBuf) -> Result<MessageProcessor, Error> {
-        if let Some("toml") = path.extension().and_then(|s| s.to_str()) {
+    pub async fn load_file(
+        flows_dir: &Utf8PathBuf,
+        path: &Utf8PathBuf,
+    ) -> Result<MessageProcessor, Error> {
+        if let Some("toml") = path.extension() {
             MessageProcessor::try_new_single_flow(flows_dir, path)
                 .await
-                .with_context(|| format!("loading flow {flow}", flow = path.display()))
+                .with_context(|| format!("loading flow {path}"))
         } else {
             MessageProcessor::try_new_single_step_flow(flows_dir, path)
                 .await
-                .with_context(|| format!("loading flow script {script}", script = path.display()))
+                .with_context(|| format!("loading flow script {path}"))
         }
     }
 }

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

@@ -2,19 +2,19 @@ use crate::cli::flows::TEdgeFlowsCli;
 use crate::command::Command;
 use crate::log::MaybeFancy;
 use anyhow::Error;
-use std::path::PathBuf;
+use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
 use tedge_flows::flow::Flow;
 
 pub struct ListCommand {
-    pub flows_dir: PathBuf,
+    pub flows_dir: Utf8PathBuf,
     pub topic: Option<String>,
 }
 
 #[async_trait::async_trait]
 impl Command for ListCommand {
     fn description(&self) -> String {
-        format!("list flows and flow steps in {}", self.flows_dir.display())
+        format!("list flows and flow steps in {}", self.flows_dir)
     }
 
     async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
@@ -38,7 +38,7 @@ impl ListCommand {
     fn display((flow_id, flow): (&String, &Flow)) {
         println!("{flow_id}");
         for step in flow.steps.iter() {
-            println!("\t{}", step.script.path.display());
+            println!("\t{}", step.script.path);
         }
     }
 }

crates/core/tedge/src/cli/flows/list.rs

@@ -4,7 +4,7 @@ use crate::log::MaybeFancy;
 use anyhow::Error;
 use base64::prelude::BASE64_STANDARD;
 use base64::prelude::*;
-use std::path::PathBuf;
+use camino::Utf8PathBuf;
 use tedge_config::TEdgeConfig;
 use tedge_flows::flow::*;
 use tedge_flows::MessageProcessor;
@@ -13,8 +13,8 @@ use tokio::io::BufReader;
 use tokio::io::Stdin;
 
 pub struct TestCommand {
-    pub flows_dir: PathBuf,
-    pub flow: Option<PathBuf>,
+    pub flows_dir: Utf8PathBuf,
+    pub flow: Option<Utf8PathBuf>,
     pub message: Option<Message>,
     pub final_on_interval: bool,
     pub base64_input: bool,
@@ -25,8 +25,8 @@ pub struct TestCommand {
 impl Command for TestCommand {
     fn description(&self) -> String {
         format!(
-            "process message samples using flows and steps in {:}",
-            self.flows_dir.display()
+            "process message samples using flows and steps in {}",
+            self.flows_dir
         )
     }
 

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

@@ -32,5 +32,8 @@ tokio = { workspace = true, features = ["fs", "macros", "time", "sync"] }
 toml = { workspace = true, features = ["parse"] }
 tracing = { workspace = true }
 
+[dev-dependencies]
+tempfile = { workspace = true }
+
 [lints]
 workspace = true

crates/extensions/tedge_flows/Cargo.toml

@@ -9,7 +9,6 @@ use camino::Utf8PathBuf;
 use serde::Deserialize;
 use serde_json::Value;
 use std::fmt::Debug;
-use std::path::Path;
 use std::time::Duration;
 use tedge_mqtt_ext::TopicFilter;
 
@@ -75,7 +74,7 @@ impl FlowConfig {
     pub async fn compile(
         self,
         js_runtime: &mut JsRuntime,
-        config_dir: &Path,
+        config_dir: &Utf8Path,
         source: Utf8PathBuf,
     ) -> Result<Flow, ConfigError> {
         let input = self.input.try_into()?;
@@ -98,18 +97,18 @@ impl FlowConfig {
 impl StepConfig {
     pub async fn compile(
         self,
-        config_dir: &Path,
+        config_dir: &Utf8Path,
         index: usize,
         flow: &Utf8Path,
     ) -> Result<FlowStep, ConfigError> {
         let path = match self.script {
-            ScriptSpec::JavaScript(path) if path.is_absolute() => path.into(),
-            ScriptSpec::JavaScript(path) if path.starts_with(config_dir) => path.into(),
+            ScriptSpec::JavaScript(path) if path.is_absolute() => path,
+            ScriptSpec::JavaScript(path) if path.starts_with(config_dir) => path,
             ScriptSpec::JavaScript(path) => config_dir.join(path),
         };
-        let script = JsScript::new(flow.to_owned().into(), index, path)
+        let script = JsScript::new(flow.to_owned(), index, path)
             .with_config(self.config)
-            .with_interval_secs(self.interval.as_secs());
+            .with_interval(self.interval);
         let config_topics = topic_filters(self.meta_topics)?;
         Ok(FlowStep {
             script,

crates/extensions/tedge_flows/src/config.rs

@@ -161,21 +161,21 @@ impl FlowStep {
     pub(crate) fn check(&self, flow: &Utf8Path) {
         let script = &self.script;
         if script.no_js_on_message_fun {
-            warn!(target: "flows", "Flow script with no 'onMessage' function: {}", script.path.display());
+            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.display());
+            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_secs != 0 {
-            warn!(target: "flows", "Flow script with no 'onInterval' function: {}; but configured with an 'interval' in {flow}", script.path.display());
+        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);
         }
     }
 
     pub(crate) fn fix(&mut self) {
         let script = &mut self.script;
-        if !script.no_js_on_interval_fun && script.interval_secs == 0 {
-            // 0 as a default is not appropriate for a script with an onInterval handler
-            script.interval_secs = 1;
+        if !script.no_js_on_interval_fun && script.interval.is_zero() {
+            // Zero as a default is not appropriate for a script with an onInterval handler
+            script.interval = std::time::Duration::from_secs(1);
         }
     }
 }
@@ -193,8 +193,9 @@ impl DateTime {
         DateTime::try_from(OffsetDateTime::now_utc()).unwrap()
     }
 
-    pub fn tick_now(&self, tick_every_seconds: u64) -> bool {
-        tick_every_seconds != 0 && (self.seconds % tick_every_seconds == 0)
+    pub fn tick_now(&self, tick_every: std::time::Duration) -> bool {
+        let tick_every_secs = tick_every.as_secs();
+        tick_every_secs != 0 && (self.seconds % tick_every_secs == 0)
     }
 
     pub fn json(&self) -> Value {

crates/extensions/tedge_flows/src/flow.rs

@@ -4,29 +4,29 @@ use crate::flow::FlowError;
 use crate::flow::Message;
 use crate::js_runtime::JsRuntime;
 use crate::js_value::JsonValue;
-use std::path::Path;
-use std::path::PathBuf;
+use camino::Utf8Path;
+use camino::Utf8PathBuf;
 use tracing::debug;
 
 #[derive(Clone)]
 pub struct JsScript {
     pub module_name: String,
-    pub path: PathBuf,
+    pub path: Utf8PathBuf,
     pub config: JsonValue,
-    pub interval_secs: u64,
+    pub interval: std::time::Duration,
     pub no_js_on_message_fun: bool,
     pub no_js_on_config_update_fun: bool,
     pub no_js_on_interval_fun: bool,
 }
 
 impl JsScript {
-    pub fn new(flow: PathBuf, index: usize, path: PathBuf) -> Self {
-        let module_name = format!("{}|{}|{}", flow.display(), index, path.display());
+    pub fn new(flow: Utf8PathBuf, index: usize, path: Utf8PathBuf) -> Self {
+        let module_name = format!("{flow}|{index}|{path}");
         JsScript {
             module_name,
             path,
             config: JsonValue::default(),
-            interval_secs: 0,
+            interval: std::time::Duration::ZERO,
             no_js_on_message_fun: true,
             no_js_on_config_update_fun: true,
             no_js_on_interval_fun: true,
@@ -48,19 +48,16 @@ impl JsScript {
         }
     }
 
-    pub fn with_interval_secs(self, interval_secs: u64) -> Self {
-        Self {
-            interval_secs,
-            ..self
-        }
+    pub fn with_interval(self, interval: std::time::Duration) -> Self {
+        Self { interval, ..self }
     }
 
-    pub fn path(&self) -> &Path {
+    pub fn path(&self) -> &Utf8Path {
         &self.path
     }
 
     pub fn source(&self) -> String {
-        format!("{}", self.path.display())
+        self.path.to_string()
     }
 
     /// Transform an input message into zero, one or more output messages
@@ -134,7 +131,7 @@ impl JsScript {
         if self.no_js_on_interval_fun {
             return Ok(vec![]);
         }
-        if !timestamp.tick_now(self.interval_secs) {
+        if !timestamp.tick_now(self.interval) {
             return Ok(vec![]);
         }
         debug!(target: "flows", "{}: onInterval({timestamp:?})", self.module_name());

crates/extensions/tedge_flows/src/js_script.rs

@@ -10,8 +10,8 @@ mod stats;
 
 use crate::actor::FlowsMapper;
 pub use crate::runtime::MessageProcessor;
+use camino::Utf8Path;
 use std::convert::Infallible;
-use std::path::Path;
 use std::path::PathBuf;
 use tedge_actors::fan_in_message_type;
 use tedge_actors::Builder;
@@ -39,7 +39,7 @@ pub struct FlowsMapperBuilder {
 }
 
 impl FlowsMapperBuilder {
-    pub async fn try_new(config_dir: impl AsRef<Path>) -> Result<Self, LoadError> {
+    pub async fn try_new(config_dir: impl AsRef<Utf8Path>) -> Result<Self, LoadError> {
         let processor = MessageProcessor::try_new(config_dir).await?;
         Ok(FlowsMapperBuilder {
             message_box: SimpleMessageBoxBuilder::new("GenMapper", 16),
@@ -68,7 +68,7 @@ impl FlowsMapperBuilder {
 
     pub fn connect_fs(&mut self, fs: &mut impl MessageSource<FsWatchEvent, PathBuf>) {
         fs.connect_mapped_sink(
-            self.processor.config_dir.clone(),
+            self.processor.config_dir.clone().into(),
             &self.message_box,
             |msg| Some(InputMessage::FsWatchEvent(msg)),
         );