Naming fix: Filter -> Flow Step

In some context `Flow script` is more appropriate.
For instance, when two steps use the same script.

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

Author: didier-wenzek

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

@@ -13,9 +13,9 @@ use tedge_gen_mapper::MessageProcessor;
 
 #[derive(clap::Subcommand, Debug)]
 pub enum TEdgeMappingCli {
-    /// List flows and filters
+    /// List flows and steps
     List {
-        /// Path to flow and filter specs
+        /// Path to the directory of flows and steps
         ///
         /// Default to /etc/tedge/gen-mapper
         #[clap(long)]
@@ -30,17 +30,17 @@ pub enum TEdgeMappingCli {
 
     /// Process message samples
     Test {
-        /// Path to flow and filter specs
+        /// Path to the directory of flows and steps
         ///
         /// Default to /etc/tedge/gen-mapper
         #[clap(long)]
         mapping_dir: Option<PathBuf>,
 
-        /// Path to the javascript filter or TOML flow definition
+        /// Path to the flow step script or TOML flow definition
         ///
         /// If none is provided, applies all the matching flows
         #[clap(long)]
-        filter: Option<PathBuf>,
+        flow: Option<PathBuf>,
 
         /// Send a tick after all the message samples
         #[clap(long = "final-tick")]
@@ -69,7 +69,7 @@ impl BuildCommand for TEdgeMappingCli {
 
             TEdgeMappingCli::Test {
                 mapping_dir,
-                filter,
+                flow,
                 final_tick,
                 topic,
                 payload,
@@ -83,7 +83,7 @@ impl BuildCommand for TEdgeMappingCli {
                 };
                 Ok(TestCommand {
                     mapping_dir,
-                    filter,
+                    flow,
                     message,
                     final_tick,
                 }
@@ -101,10 +101,10 @@ impl TEdgeMappingCli {
     pub async fn load_flows(mapping_dir: &PathBuf) -> Result<MessageProcessor, Error> {
         MessageProcessor::try_new(mapping_dir)
             .await
-            .with_context(|| format!("loading flows and filters from {}", mapping_dir.display()))
+            .with_context(|| format!("loading flows and steps from {}", mapping_dir.display()))
     }
 
-    pub async fn load_filter(
+    pub async fn load_file(
         mapping_dir: &PathBuf,
         path: &PathBuf,
     ) -> Result<MessageProcessor, Error> {
@@ -113,9 +113,9 @@ impl TEdgeMappingCli {
                 .await
                 .with_context(|| format!("loading flow {flow}", flow = path.display()))
         } else {
-            MessageProcessor::try_new_single_filter(mapping_dir, path)
+            MessageProcessor::try_new_single_step_flow(mapping_dir, path)
                 .await
-                .with_context(|| format!("loading filter {filter}", filter = path.display()))
+                .with_context(|| format!("loading flow script {script}", script = path.display()))
         }
     }
 }

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

@@ -41,7 +41,7 @@ impl ListCommand {
     fn display((flow_id, flow): (&String, &Flow)) {
         println!("{flow_id}");
         for step in flow.steps.iter() {
-            println!("\t{}", step.filter.path.display());
+            println!("\t{}", step.script.path.display());
         }
     }
 }

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

@@ -12,7 +12,7 @@ use tokio::io::Stdin;
 
 pub struct TestCommand {
     pub mapping_dir: PathBuf,
-    pub filter: Option<PathBuf>,
+    pub flow: Option<PathBuf>,
     pub message: Option<Message>,
     pub final_tick: bool,
 }
@@ -21,15 +21,15 @@ pub struct TestCommand {
 impl Command for TestCommand {
     fn description(&self) -> String {
         format!(
-            "process message samples using flows and filters in {:}",
+            "process message samples using flows and steps in {:}",
             self.mapping_dir.display()
         )
     }
 
     async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
-        let mut processor = match &self.filter {
+        let mut processor = match &self.flow {
             None => TEdgeMappingCli::load_flows(&self.mapping_dir).await?,
-            Some(filter) => TEdgeMappingCli::load_filter(&self.mapping_dir, filter).await?,
+            Some(flow) => TEdgeMappingCli::load_file(&self.mapping_dir, flow).await?,
         };
         if let Some(message) = &self.message {
             let timestamp = DateTime::now();
@@ -74,7 +74,7 @@ impl TestCommand {
     }
 }
 
-fn print(messages: Result<Vec<Message>, FilterError>) {
+fn print(messages: Result<Vec<Message>, FlowError>) {
     match messages {
         Ok(messages) => {
             for message in messages {

crates/extensions/tedge_gen_mapper/flows/circuit-breaker.js

@@ -1,10 +1,10 @@
-// A filter that let messages go through, unless too many messages are received within a given period
+// A flow step that let messages go through, unless too many messages are received within a given period
 //
-// This filter is configured by the following settings:
+// This flow step is configured by the following settings:
 // - tick_every_seconds: the frequency at which the sliding window is moved
 // - tick_count: size of the time windows
 // - too_many: how many messages is too many (received during the last tick_count*tick_every_seconds seconds)
-// - back_to_normal: how many messages is okay to reactivate the filter if bellow
+// - back_to_normal: how many messages is okay to reactivate the flow step if bellow
 // - message_on_too_many: message sent when the upper threshold is crossed
 // - message_on_back_to_normal: message sent when the lower threshold is crossed
 // - stats_topic: topic for statistic messages

crates/extensions/tedge_gen_mapper/flows/collectd.toml

@@ -1,6 +1,6 @@
 input_topics = ["collectd/+/+/+"]
 
 steps = [
-    { filter = "collectd-to-te.js" },
-    { filter = "average.js", tick_every_seconds = 10 }
+    { script = "collectd-to-te.js" },
+    { script = "average.js", tick_every_seconds = 10 }
 ]

crates/extensions/tedge_gen_mapper/flows/loop.toml

@@ -2,6 +2,6 @@
 input_topics = ["loopback/#"]
 
 steps = [
-    { filter = "add_timestamp.js" },
-    { filter = "circuit-breaker.js", tick_every_seconds = 1, config = { stats_topic = "te/error", too_many = 10000, message_on_too_many = { topic = "te/device/main///a/too-many-messages", payload = "too many messages" }, message_on_back_to_normal = { topic = "te/device/main///a/too-many-messages", payload = "back to normal" } } }
+    { script = "add_timestamp.js" },
+    { script = "circuit-breaker.js", tick_every_seconds = 1, config = { stats_topic = "te/error", too_many = 10000, message_on_too_many = { topic = "te/device/main///a/too-many-messages", payload = "too many messages" }, message_on_back_to_normal = { topic = "te/device/main///a/too-many-messages", payload = "back to normal" } } }
 ]

crates/extensions/tedge_gen_mapper/flows/measurements.toml

@@ -1,6 +1,6 @@
 input_topics = ["te/+/+/+/+/m/+"]
 
 steps = [
-    { filter = "add_timestamp.js" },
-    { filter = "te_to_c8y.js", meta_topics = ["te/+/+/+/+/m/+/meta"] }
+    { script = "add_timestamp.js" },
+    { script = "te_to_c8y.js", meta_topics = ["te/+/+/+/+/m/+/meta"] }
 ]

crates/extensions/tedge_gen_mapper/src/actor.rs

@@ -41,7 +41,7 @@ impl Actor for GenMapper {
                 message = self.messages.recv() => {
                     match message {
                         Some(InputMessage::MqttMessage(message)) => match Message::try_from(message) {
-                            Ok(message) => self.filter(message).await?,
+                            Ok(message) => self.process(message).await?,
                             Err(err) => {
                                 error!(target: "gen-mapper", "Cannot process message: {err}");
                             }
@@ -51,7 +51,7 @@ impl Actor for GenMapper {
                                 continue;
                             };
                             if matches!(path.extension(), Some("js" | "ts")) {
-                                self.processor.reload_filter(path).await;
+                                self.processor.reload_script(path).await;
                             } else if path.extension() == Some("toml") {
                                 self.processor.reload_flow(path).await;
                                 self.send_updated_subscriptions().await?;
@@ -71,7 +71,7 @@ impl Actor for GenMapper {
                                 continue;
                             };
                             if matches!(path.extension(), Some("js" | "ts")) {
-                                self.processor.remove_filter(path).await;
+                                self.processor.remove_script(path).await;
                             } else if path.extension() == Some("toml") {
                                 self.processor.remove_flow(path).await;
                                 self.send_updated_subscriptions().await?;
@@ -102,7 +102,7 @@ impl GenMapper {
         diff
     }
 
-    async fn filter(&mut self, message: Message) -> Result<(), RuntimeError> {
+    async fn process(&mut self, message: Message) -> Result<(), RuntimeError> {
         let timestamp = DateTime::now();
         for (flow_id, flow_messages) in self.processor.process(&timestamp, &message).await {
             match flow_messages {

crates/extensions/tedge_gen_mapper/src/config.rs

@@ -1,7 +1,7 @@
 use crate::flow::Flow;
 use crate::flow::FlowStep;
-use crate::js_filter::JsFilter;
 use crate::js_runtime::JsRuntime;
+use crate::js_script::JsScript;
 use crate::LoadError;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
@@ -19,7 +19,7 @@ pub struct FlowConfig {
 
 #[derive(Deserialize)]
 pub struct StepConfig {
-    filter: FilterSpec,
+    script: ScriptSpec,
 
     #[serde(default)]
     config: Option<Value>,
@@ -33,7 +33,7 @@ pub struct StepConfig {
 
 #[derive(Deserialize)]
 #[serde(untagged)]
-pub enum FilterSpec {
+pub enum ScriptSpec {
     JavaScript(Utf8PathBuf),
 }
 
@@ -47,10 +47,10 @@ pub enum ConfigError {
 }
 
 impl FlowConfig {
-    pub fn from_filter(filter: Utf8PathBuf) -> Self {
+    pub fn from_step(script: Utf8PathBuf) -> Self {
         let input_topic = "#".to_string();
         let step = StepConfig {
-            filter: FilterSpec::JavaScript(filter),
+            script: ScriptSpec::JavaScript(script),
             config: None,
             tick_every_seconds: 0,
             meta_topics: vec![],
@@ -71,7 +71,7 @@ impl FlowConfig {
         let mut steps = vec![];
         for (i, step) in self.steps.into_iter().enumerate() {
             let mut step = step.compile(config_dir, i, &source).await?;
-            js_runtime.load_filter(&mut step.filter).await?;
+            js_runtime.load_script(&mut step.script).await?;
             step.check(&source);
             step.fix();
             steps.push(step);
@@ -91,17 +91,17 @@ impl StepConfig {
         index: usize,
         flow: &Utf8Path,
     ) -> Result<FlowStep, 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 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) => config_dir.join(path),
         };
-        let filter = JsFilter::new(flow.to_owned().into(), index, path)
+        let script = JsScript::new(flow.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)?;
         Ok(FlowStep {
-            filter,
+            script,
             config_topics,
         })
     }