Merge pull request #3714 from rina23q/refactor/remove-outdated-messages-from-availability-actor

refactor: cleanup code and outdated comments from the availability actor

Author: rina23q

crates/extensions/c8y_mapper_ext/src/availability/actor.rs

@@ -1,12 +1,11 @@
 use crate::availability::AvailabilityConfig;
 use crate::availability::AvailabilityInput;
 use crate::availability::AvailabilityOutput;
-use crate::availability::C8yJsonInventoryUpdate;
+use crate::availability::C8yJsonEmptyInventoryUpdate;
 use crate::availability::C8ySmartRestSetInterval117;
 use crate::availability::TimerStart;
 use async_trait::async_trait;
 use c8y_api::smartrest::topic::C8yTopic;
-use serde_json::json;
 use std::collections::HashMap;
 use tedge_actors::Actor;
 use tedge_actors::LoggingSender;
@@ -156,44 +155,40 @@ impl AvailabilityActor {
     /// Insert a <"device topic ID" - "health entity topic ID" and "external ID"> pair into the map.
     /// If @health is provided in the registration message, use the value as long as it's valid as entity topic ID.
     /// If @health is not provided, use the "tedge-agent" service topic ID as default.
-    /// @id is the only source to know the device's external ID. Hence, @id must be provided in the registration message.
     fn update_device_service_pair(
         &mut self,
         registration_message: &EntityRegistrationMessage,
     ) -> RegistrationResult {
         let source = &registration_message.topic_id;
 
-        let result = match registration_message.health_endpoint.clone() {
+        let maybe_health_topic_id = match registration_message.health_endpoint.clone() {
             None => registration_message
                 .topic_id
-                .default_service_for_device("tedge-agent")
-                .ok_or_else( || format!("The entity is not in the default topic scheme. Please specify '@health' to enable availability monitoring for the device '{source}'")),
-            Some(topic_id) => Ok(topic_id),
+                .default_service_for_device("tedge-agent"),
+            Some(topic_id) => Some(topic_id),
         };
 
-        match result {
-            Ok(health_topic_id) => match registration_message.external_id.clone() {
-                None => RegistrationResult::Skip(format!("Registration message is skipped since '@id' is missing in the payload. source: '{source}', {registration_message:?}")),
-                Some(external_id) => {
-                    if let Some(ids) = self.device_ids_map.get(source) {
-                        if ids.health_topic_id == health_topic_id {
-                            return RegistrationResult::Skip(format!("Registration message is skipped since no health endpoint change is detected. source: '{source}', {registration_message:?}"))
-                        }
+        match (maybe_health_topic_id, registration_message.external_id.clone()) {
+            (None, _) => RegistrationResult::Error(format!("The entity is not in the default topic scheme. Please specify '@health' to enable availability monitoring for the device '{source}'")),
+            (_, None) => RegistrationResult::Error(format!("External ID is missing from the received registration message. source: '{source}', {registration_message:?}")),
+            (Some(health_topic_id), Some(external_id)) => {
+                if let Some(ids) = self.device_ids_map.get(source) {
+                    if ids.health_topic_id == health_topic_id {
+                        return RegistrationResult::Skip(format!("Registration message is skipped since no health endpoint change is detected. source: '{source}', {registration_message:?}"))
                     }
+                }
 
-                    match self.device_ids_map.insert(
-                        source.clone(),
-                        DeviceIds {
-                            health_topic_id,
-                            external_id,
-                        },
-                    ) {
-                        None => RegistrationResult::New,
-                        Some(_) => RegistrationResult::Update,
-                    }
+                match self.device_ids_map.insert(
+                    source.clone(),
+                    DeviceIds {
+                        health_topic_id,
+                        external_id,
+                    },
+                ) {
+                    None => RegistrationResult::New,
+                    Some(_) => RegistrationResult::Update,
                 }
-            },
-            Err(err) => RegistrationResult::Error(err),
+            }
         }
     }
 
@@ -264,9 +259,8 @@ impl AvailabilityActor {
             if let Some(health_status) = self.health_status_map.get(service_topic_id) {
                 // Send an empty JSON over MQTT message if the target service status is "up"
                 if health_status.status == Status::Up {
-                    let json_over_mqtt = C8yJsonInventoryUpdate {
+                    let json_over_mqtt = C8yJsonEmptyInventoryUpdate {
                         external_id: external_id.into(),
-                        payload: json!({}),
                         prefix: self.config.c8y_prefix.clone(),
                     };
                     self.message_box.send(json_over_mqtt.into()).await?;

crates/extensions/c8y_mapper_ext/src/availability/builder.rs

@@ -86,7 +86,7 @@ impl AvailabilityBuilder {
     fn mqtt_message_builder() -> impl Fn(AvailabilityOutput) -> Option<MqttMessage> {
         move |res| match res {
             AvailabilityOutput::C8ySmartRestSetInterval117(value) => Some(value.into()),
-            AvailabilityOutput::C8yJsonInventoryUpdate(value) => Some(value.into()),
+            AvailabilityOutput::C8yJsonEmptyInventoryUpdate(value) => Some(value.into()),
         }
     }
 }

crates/extensions/c8y_mapper_ext/src/availability/mod.rs

@@ -1,6 +1,8 @@
 use crate::availability::actor::TimerPayload;
+use crate::inventory::inventory_update_topic;
 pub use builder::AvailabilityBuilder;
 use c8y_api::smartrest::inventory::C8ySmartRestSetInterval117;
+use serde_json::json;
 use std::time::Duration;
 use tedge_actors::fan_in_message_type;
 use tedge_api::entity::EntityExternalId;
@@ -12,7 +14,6 @@ use tedge_config::models::TopicPrefix;
 use tedge_config::tedge_toml::ReadError;
 use tedge_config::TEdgeConfig;
 use tedge_mqtt_ext::MqttMessage;
-use tedge_mqtt_ext::Topic;
 use tedge_timer_ext::SetTimeout;
 use tedge_timer_ext::Timeout;
 
@@ -26,27 +27,18 @@ pub type TimerComplete = Timeout<TimerPayload>;
 pub type SourceHealthStatus = (EntityTopicId, HealthStatus);
 
 fan_in_message_type!(AvailabilityInput[EntityRegistrationMessage, SourceHealthStatus, TimerComplete] : Debug);
-fan_in_message_type!(AvailabilityOutput[C8ySmartRestSetInterval117, C8yJsonInventoryUpdate] : Debug);
+fan_in_message_type!(AvailabilityOutput[C8ySmartRestSetInterval117, C8yJsonEmptyInventoryUpdate] : Debug);
 
-// TODO! Make it generic and move to c8y_api crate while refactoring c8y-mapper
 #[derive(Debug)]
-pub struct C8yJsonInventoryUpdate {
+pub struct C8yJsonEmptyInventoryUpdate {
     external_id: String,
-    payload: serde_json::Value,
-    pub prefix: TopicPrefix,
+    prefix: TopicPrefix,
 }
 
-impl From<C8yJsonInventoryUpdate> for MqttMessage {
-    fn from(value: C8yJsonInventoryUpdate) -> Self {
-        let json_over_mqtt_topic = format!(
-            "{prefix}/inventory/managedObjects/update/{external_id}",
-            prefix = value.prefix,
-            external_id = value.external_id
-        );
-        MqttMessage::new(
-            &Topic::new_unchecked(&json_over_mqtt_topic),
-            value.payload.to_string(),
-        )
+impl From<C8yJsonEmptyInventoryUpdate> for MqttMessage {
+    fn from(value: C8yJsonEmptyInventoryUpdate) -> Self {
+        let json_over_mqtt_topic = inventory_update_topic(&value.prefix, &value.external_id);
+        MqttMessage::new(&json_over_mqtt_topic, json!({}).to_string())
     }
 }
 

crates/extensions/c8y_mapper_ext/src/inventory.rs

@@ -11,6 +11,7 @@ use tedge_api::entity::EntityType;
 use tedge_api::entity_store::EntityTwinMessage;
 use tedge_api::mqtt_topics::Channel;
 use tedge_api::mqtt_topics::EntityTopicId;
+use tedge_config::models::TopicPrefix;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
 use tracing::info;
@@ -178,13 +179,18 @@ impl CumulocityConverter {
         source: &EntityTopicId,
     ) -> Result<Topic, ConversionError> {
         let entity_external_id = self.entity_cache.try_get(source)?.external_id.as_ref();
-        Ok(Topic::new_unchecked(&format!(
-            "{prefix}/{INVENTORY_MANAGED_OBJECTS_TOPIC}/{entity_external_id}",
-            prefix = self.config.bridge_config.c8y_prefix,
-        )))
+        let topic =
+            inventory_update_topic(&self.config.bridge_config.c8y_prefix, entity_external_id);
+        Ok(topic)
     }
 }
 
+pub fn inventory_update_topic(prefix: &TopicPrefix, external_id: &str) -> Topic {
+    Topic::new_unchecked(&format!(
+        "{prefix}/{INVENTORY_MANAGED_OBJECTS_TOPIC}/{external_id}",
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use crate::converter::tests::create_c8y_converter;