Naming fix: Stage -> Step

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

Author: didier-wenzek

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

@@ -14,7 +14,10 @@ pub struct ListCommand {
 #[async_trait::async_trait]
 impl Command for ListCommand {
     fn description(&self) -> String {
-        format!("list flows and filters in {:}", self.mapping_dir.display())
+        format!(
+            "list flows and flow steps in {:}",
+            self.mapping_dir.display()
+        )
     }
 
     async fn execute(&self, _config: TEdgeConfig) -> Result<(), MaybeFancy<Error>> {
@@ -37,8 +40,8 @@ impl Command for ListCommand {
 impl ListCommand {
     fn display((flow_id, flow): (&String, &Flow)) {
         println!("{flow_id}");
-        for stage in flow.stages.iter() {
-            println!("\t{}", stage.filter.path.display());
+        for step in flow.steps.iter() {
+            println!("\t{}", step.filter.path.display());
         }
     }
 }

crates/extensions/tedge_gen_mapper/flows/collectd.toml

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

crates/extensions/tedge_gen_mapper/flows/loop.toml

@@ -1,7 +1,7 @@
 # This flow is on purpose looping: the messages are published to the same topic
 input_topics = ["loopback/#"]
 
-stages = [
+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" } } }
 ]

crates/extensions/tedge_gen_mapper/flows/measurements.toml

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

crates/extensions/tedge_gen_mapper/src/config.rs

@@ -1,5 +1,5 @@
 use crate::flow::Flow;
-use crate::flow::Stage;
+use crate::flow::FlowStep;
 use crate::js_filter::JsFilter;
 use crate::js_runtime::JsRuntime;
 use crate::LoadError;
@@ -14,11 +14,11 @@ use tedge_mqtt_ext::TopicFilter;
 #[derive(Deserialize)]
 pub struct FlowConfig {
     input_topics: Vec<String>,
-    stages: Vec<StageConfig>,
+    steps: Vec<StepConfig>,
 }
 
 #[derive(Deserialize)]
-pub struct StageConfig {
+pub struct StepConfig {
     filter: FilterSpec,
 
     #[serde(default)]
@@ -49,15 +49,15 @@ pub enum ConfigError {
 impl FlowConfig {
     pub fn from_filter(filter: Utf8PathBuf) -> Self {
         let input_topic = "#".to_string();
-        let stage = StageConfig {
+        let step = StepConfig {
             filter: FilterSpec::JavaScript(filter),
             config: None,
             tick_every_seconds: 0,
             meta_topics: vec![],
         };
         Self {
             input_topics: vec![input_topic],
-            stages: vec![stage],
+            steps: vec![step],
         }
     }
 
@@ -68,29 +68,29 @@ impl FlowConfig {
         source: Utf8PathBuf,
     ) -> Result<Flow, ConfigError> {
         let input_topics = topic_filters(&self.input_topics)?;
-        let mut stages = vec![];
-        for (i, stage) in self.stages.into_iter().enumerate() {
-            let mut stage = stage.compile(config_dir, i, &source).await?;
-            js_runtime.load_filter(&mut stage.filter).await?;
-            stage.check(&source);
-            stage.fix();
-            stages.push(stage);
+        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?;
+            step.check(&source);
+            step.fix();
+            steps.push(step);
         }
         Ok(Flow {
             input_topics,
-            stages,
+            steps,
             source,
         })
     }
 }
 
-impl StageConfig {
+impl StepConfig {
     pub async fn compile(
         self,
         config_dir: &Path,
         index: usize,
         flow: &Utf8Path,
-    ) -> Result<Stage, ConfigError> {
+    ) -> 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(),
@@ -100,7 +100,7 @@ impl StageConfig {
             .with_config(self.config)
             .with_tick_every_seconds(self.tick_every_seconds);
         let config_topics = topic_filters(&self.meta_topics)?;
-        Ok(Stage {
+        Ok(FlowStep {
             filter,
             config_topics,
         })

crates/extensions/tedge_gen_mapper/src/flow.rs

@@ -16,14 +16,14 @@ pub struct Flow {
     /// The source topics
     pub input_topics: TopicFilter,
 
-    /// Transformation stages to apply in order to the messages
-    pub stages: Vec<Stage>,
+    /// Transformation steps to apply in order to the messages
+    pub steps: Vec<FlowStep>,
 
     pub source: Utf8PathBuf,
 }
 
-/// A message transformation stage
-pub struct Stage {
+/// A message transformation step
+pub struct FlowStep {
     pub filter: JsFilter,
     pub config_topics: TopicFilter,
 }
@@ -55,8 +55,8 @@ pub enum FilterError {
 impl Flow {
     pub fn topics(&self) -> TopicFilter {
         let mut topics = self.input_topics.clone();
-        for stage in self.stages.iter() {
-            topics.add_all(stage.config_topics.clone())
+        for step in self.steps.iter() {
+            topics.add_all(step.config_topics.clone())
         }
         topics
     }
@@ -66,9 +66,9 @@ impl Flow {
         js_runtime: &JsRuntime,
         message: &Message,
     ) -> Result<(), FilterError> {
-        for stage in self.stages.iter_mut() {
-            if stage.config_topics.accept_topic_name(&message.topic) {
-                stage.filter.update_config(js_runtime, message).await?
+        for step in self.steps.iter_mut() {
+            if step.config_topics.accept_topic_name(&message.topic) {
+                step.filter.update_config(js_runtime, message).await?
             }
         }
         Ok(())
@@ -88,12 +88,12 @@ impl Flow {
 
         let stated_at = stats.flow_process_start(self.source.as_str());
         let mut messages = vec![message.clone()];
-        for stage in self.stages.iter() {
-            let js = stage.filter.source();
+        for step in self.steps.iter() {
+            let js = step.filter.source();
             let mut transformed_messages = vec![];
             for message in messages.iter() {
                 let filter_started_at = stats.filter_start(&js, "process");
-                let filter_output = stage.filter.process(js_runtime, timestamp, message).await;
+                let filter_output = step.filter.process(js_runtime, timestamp, message).await;
                 match &filter_output {
                     Ok(messages) => {
                         stats.filter_done(&js, "process", filter_started_at, messages.len())
@@ -117,13 +117,13 @@ impl Flow {
     ) -> Result<Vec<Message>, FilterError> {
         let stated_at = stats.flow_tick_start(self.source.as_str());
         let mut messages = vec![];
-        for stage in self.stages.iter() {
-            let js = stage.filter.source();
+        for step in self.steps.iter() {
+            let js = step.filter.source();
             // Process first the messages triggered upstream by the tick
             let mut transformed_messages = vec![];
             for message in messages.iter() {
                 let filter_started_at = stats.filter_start(&js, "process");
-                let filter_output = stage.filter.process(js_runtime, timestamp, message).await;
+                let filter_output = step.filter.process(js_runtime, timestamp, message).await;
                 match &filter_output {
                     Ok(messages) => {
                         stats.filter_done(&js, "process", filter_started_at, messages.len())
@@ -135,22 +135,22 @@ impl Flow {
 
             // Only then process the tick
             let filter_started_at = stats.filter_start(&js, "tick");
-            let tick_output = stage.filter.tick(js_runtime, timestamp).await;
+            let tick_output = step.filter.tick(js_runtime, timestamp).await;
             match &tick_output {
                 Ok(messages) => stats.filter_done(&js, "tick", filter_started_at, messages.len()),
                 Err(_) => stats.filter_failed(&js, "tick"),
             }
             transformed_messages.extend(tick_output?);
 
-            // Iterate with all the messages collected at this stage
+            // Iterate with all the messages collected at this step
             messages = transformed_messages;
         }
         stats.flow_tick_done(self.source.as_str(), stated_at, messages.len());
         Ok(messages)
     }
 }
 
-impl Stage {
+impl FlowStep {
     pub(crate) fn check(&self, flow: &Utf8Path) {
         let filter = &self.filter;
         if filter.no_js_process {

crates/extensions/tedge_gen_mapper/src/runtime.rs

@@ -142,9 +142,9 @@ impl MessageProcessor {
 
     pub async fn reload_filter(&mut self, path: Utf8PathBuf) {
         for flow in self.flows.values_mut() {
-            for stage in &mut flow.stages {
-                if stage.filter.path() == path {
-                    match self.js_runtime.load_filter(&mut stage.filter).await {
+            for step in &mut flow.steps {
+                if step.filter.path() == path {
+                    match self.js_runtime.load_filter(&mut step.filter).await {
                         Ok(()) => {
                             info!(target: "gen-mapper", "Reloaded filter {path}");
                         }
@@ -160,8 +160,8 @@ impl MessageProcessor {
 
     pub async fn remove_filter(&mut self, path: Utf8PathBuf) {
         for (flow_id, flow) in self.flows.iter() {
-            for stage in flow.stages.iter() {
-                if stage.filter.path() == path {
+            for step in flow.steps.iter() {
+                if step.filter.path() == path {
                     warn!(target: "gen-mapper", "Removing a filter used by {flow_id}: {path}");
                     return;
                 }

docs/src/references/mappers/gen-mapper.md

@@ -48,14 +48,14 @@ which is used to drive all MQTT message transformations.
 - A filter is defined by a Javascript file with `.js` extension.
 - The definition of flows must provide a list of MQTT topics to subscribe to.
   - The flow will be feed with all the messages received on these topics.
-- A flow definition also provides a list of stages.
-  - Each stage is built from a javascript and is possibly given a config (arbitrary json that will be passed to the script)
-  - Each stage can also subscribe to a list of MQTT topics (which messages will be passed to the script to update its config)
+- A flow definition provides a list of steps.
+  - Each step is built from a javascript and is possibly given a config (arbitrary json that will be passed to the script)
+  - Each step can also subscribe to a list of MQTT topics (which messages will be passed to the script to update its config)
 
 ```toml
 input_topics = ["te/+/+/+/+/m/+"]
 
-stages = [
+steps = [
     { filter = "add_timestamp.js" },
     { filter = "drop_stragglers.js", config = { max_delay = 60 } },
     { filter = "te_to_c8y.js", meta_topics = ["te/+/+/+/+/m/+/meta"] }

tests/RobotFramework/tests/tedge_gen_mapper/flows/count-events.toml

@@ -1,5 +1,5 @@
 input_topics = ["test/+/+/+/+/e/+"]
 
-stages = [
+steps = [
     { filter = "count-messages.js", config = { topic = "test/count/e" }, tick_every_seconds = 1 },
 ]

tests/RobotFramework/tests/tedge_gen_mapper/flows/count-measurements.toml

@@ -1,5 +1,5 @@
 input_topics = ["test/+/+/+/+/m/+"]
 
-stages = [
+steps = [
     { filter = "count-messages.js", config = { topic = "test/count/m" }, tick_every_seconds = 1 },
 ]