Merge pull request #3699 from albinsuresh/fix/clear-entity-twin-data-with-empty-message

fix: clear entity twin data in entity store with empty retained message

Author: albinsuresh

crates/core/tedge_agent/src/entity_manager/server.rs

@@ -149,7 +149,7 @@ impl Server for EntityStoreServer {
 }
 
 impl EntityStoreServer {
-    async fn process_mqtt_message(&mut self, message: MqttMessage) {
+    pub(crate) async fn process_mqtt_message(&mut self, message: MqttMessage) {
         if let Ok((topic_id, channel)) = self.mqtt_schema.entity_channel_of(&message.topic) {
             if let Channel::EntityMetadata = channel {
                 self.process_entity_registration(topic_id, message.payload_bytes())
@@ -219,7 +219,11 @@ impl EntityStoreServer {
         }
 
         if let Channel::EntityTwinData { fragment_key } = channel {
-            let fragment_value = serde_json::from_slice(message.payload_bytes())?;
+            let fragment_value = if message.payload().is_empty() {
+                Value::Null
+            } else {
+                serde_json::from_slice(message.payload_bytes())?
+            };
             let twin_message = EntityTwinMessage::new(topic_id, fragment_key, fragment_value);
             self.entity_store.update_twin_fragment(twin_message)?;
         }

crates/core/tedge_agent/src/entity_manager/tests.rs

@@ -18,6 +18,7 @@ use tedge_actors::Sender;
 use tedge_actors::Server;
 use tedge_api::entity::EntityMetadata;
 use tedge_api::entity::EntityType;
+use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_mqtt_ext::test_helpers::assert_received_contains_str;
 use tedge_mqtt_ext::MqttConnection;
 use tedge_mqtt_ext::MqttConnector;
@@ -191,6 +192,27 @@ async fn delete_entity_tree_clears_entities_bottom_up() {
         .await;
 }
 
+#[tokio::test]
+async fn clear_entity_twin_data() {
+    let (mut entity_store, _mqtt_box, _mqtt_sender) = entity::server("device-under-test");
+
+    entity_store
+        .process_mqtt_message(MqttMessage::from(("te/device/main///twin/x", "9")).with_retain())
+        .await;
+    let entity = entity_store
+        .get(&"device/main//".parse::<EntityTopicId>().unwrap())
+        .unwrap();
+    assert_eq!(entity.twin_data.get("x"), Some(&9.into()));
+
+    entity_store
+        .process_mqtt_message(MqttMessage::from(("te/device/main///twin/x", "")).with_retain())
+        .await;
+    let entity = entity_store
+        .get(&"device/main//".parse::<EntityTopicId>().unwrap())
+        .unwrap();
+    assert_eq!(entity.twin_data.get("x"), None);
+}
+
 proptest! {
     //#![proptest_config(proptest::prelude::ProptestConfig::with_cases(1000))]
     #[test]