fixup! fixup! fixup! fixup! fixup! fixup! fixup! Entity-cache preparation step for mea transformers

Author: didier-wenzek

crates/extensions/c8y_mapper_ext/src/flows.rs

@@ -21,7 +21,10 @@ impl C8yMapperBuilder {
         };
         let mut flows = ConnectedFlowRegistry::new(flows_dir);
 
-        flows.register_builtin(crate::mea::message_cache::MessageCache::default());
+        let mapper_topic_id = self.config.service_topic_id.clone();
+        flows.register_builtin(crate::mea::message_cache::MessageCache::new(
+            mapper_topic_id,
+        ));
         flows.register_builtin(crate::mea::measurements::MeasurementConverter::default());
         flows.register_builtin(crate::mea::events::EventConverter::default());
         flows.register_builtin(crate::mea::alarms::AlarmConverter::default());
@@ -109,7 +112,7 @@ topic = "{errors_topic}"
 
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "unix", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}", mapper_topic_id = "{mapper_topic_id}" }} }},
+    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
     {{ builtin = "into-c8y-measurements", config = {{ topic_root = "{topic_prefix}" }} }},
     {{ builtin = "limit-payload-size", config = {{ max_size = {max_size} }} }},
 ]
@@ -140,7 +143,7 @@ topic = "{errors_topic}"
 
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "rfc3339", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}", mapper_topic_id = "{mapper_topic_id}" }} }},
+    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
     {{ builtin = "into-c8y-events", config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}", max_mqtt_payload_size = {max_mqtt_payload_size} }} }},
 ]
 
@@ -171,7 +174,7 @@ topic = "{errors_topic}"
 
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "rfc3339", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}", mapper_topic_id = "{mapper_topic_id}" }} }},
+    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
     {{ builtin = "into-c8y-alarms", interval = "3s", config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}" }} }},
     {{ builtin = "limit-payload-size", config = {{ max_size = {max_size} }} }},
 ]
@@ -201,7 +204,7 @@ topic = "{errors_topic}"
             r#"input.mqtt.topics = {input_topics:?}
 
 steps = [
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}", mapper_topic_id = "{mapper_topic_id}" }} }},
+    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
     {{ builtin = "into-c8y-health-status", config = {{ topic_root = "{topic_prefix}", main_device = "{main_device}", c8y_prefix = "{c8y_prefix}" }} }},
 ]
 

crates/extensions/c8y_mapper_ext/src/mea/message_cache.rs

@@ -1,7 +1,6 @@
 use crate::mea::entities::C8yEntityBirth;
 use crate::mea::get_entity_metadata;
 use std::collections::HashMap;
-use std::str::FromStr;
 use std::time::SystemTime;
 use tedge_api::mqtt_topics::Channel;
 use tedge_api::mqtt_topics::EntityTopicId;
@@ -24,11 +23,11 @@ pub struct MessageCache {
     cache: HashMap<EntityTopicId, RingBuffer<Message>>,
 }
 
-impl Default for MessageCache {
-    fn default() -> Self {
+impl MessageCache {
+    pub fn new(mapper_topic_id: EntityTopicId) -> Self {
         MessageCache {
             mqtt_schema: MqttSchema::default(),
-            mapper_topic_id: EntityTopicId::default_main_service("tedge-mapper-c8y").unwrap(),
+            mapper_topic_id,
             cache: HashMap::default(),
         }
     }
@@ -43,11 +42,6 @@ impl tedge_flows::Transformer for MessageCache {
         if let Some(root) = config.string_property("topic_root") {
             self.mqtt_schema = MqttSchema::with_root(root.to_string())
         }
-        if let Some(mapper_topic_id) = config.string_property("mapper_topic_id") {
-            self.mapper_topic_id = EntityTopicId::from_str(mapper_topic_id).map_err(|err| {
-                ConfigError::IncorrectSetting(format!("Not a valid entity topic id: {}", err))
-            })?;
-        }
         Ok(())
     }