refactor: handle retain messages with dynamic subscriptions

crates/common/mqtt_channel/src/tests.rs

@@ -3,6 +3,7 @@ use futures::SinkExt;
 use futures::StreamExt;
 use std::convert::TryInto;
 use std::time::Duration;
+use tokio::time::timeout;
 
 const TIMEOUT: Duration = Duration::from_millis(1000);
 
@@ -568,3 +569,49 @@ async fn assert_payload_received(con: &mut Connection, payload: &'static str) {
         not_msg => panic!("Expected message to be received, got {not_msg:?}"),
     }
 }
+
+#[tokio::test]
+async fn connections_from_cloned_configs_are_independent() -> Result<(), anyhow::Error> {
+    // This test arose from an issue with dynamic subscriptions where
+    // subscriptions were shared between different MQTT channel instances
+    let broker = mqtt_tests::test_mqtt_broker();
+    let mqtt_config = Config::default().with_port(broker.port);
+    let mqtt_config_cloned = mqtt_config.clone();
+
+    let topic = uniquify!("a/test/topic");
+    let other_topic = uniquify!("different/test/topic");
+    let mqtt_config = mqtt_config.with_subscriptions(TopicFilter::new_unchecked(topic));
+    let mqtt_config_cloned =
+        mqtt_config_cloned.with_subscriptions(TopicFilter::new_unchecked(other_topic));
+
+    let mut con = Connection::new(&mqtt_config).await?;
+    let mut other_con = Connection::new(&mqtt_config_cloned).await?;
+
+    // Any messages published on that topic ...
+    broker.publish(topic, "original topic message").await?;
+    broker.publish(other_topic, "other topic message").await?;
+
+    // ... must be received by the client
+    assert_eq!(
+        MaybeMessage::Next(message(topic, "original topic message")),
+        next_message(&mut con.received).await
+    );
+    assert_eq!(
+        MaybeMessage::Next(message(other_topic, "other topic message")),
+        next_message(&mut other_con.received).await
+    );
+
+    assert!(
+        timeout(Duration::from_millis(200), next_message(&mut con.received))
+            .await
+            .is_err()
+    );
+    assert!(timeout(
+        Duration::from_millis(200),
+        next_message(&mut other_con.received)
+    )
+    .await
+    .is_err());
+
+    Ok(())
+}

crates/common/mqtt_channel/src/topics.rs

@@ -191,6 +191,10 @@ impl TopicFilter {
             None
         }
     }
+
+    pub fn is_empty(&self) -> bool {
+        self.patterns.is_empty()
+    }
 }
 
 impl TryInto<Topic> for &str {

crates/core/tedge_actors/src/test_helpers.rs

@@ -259,7 +259,6 @@ where
         }
     }
 
-    #[allow(clippy::needless_collect)] // To avoid issues with Send constraints
     async fn assert_received<Samples>(&mut self, expected: Samples)
     where
         Samples: IntoIterator + Send,

crates/core/tedge_agent/src/agent.rs

@@ -58,7 +58,6 @@ use tedge_log_manager::LogManagerConfig;
 use tedge_log_manager::LogManagerOptions;
 use tedge_mqtt_ext::MqttActorBuilder;
 use tedge_mqtt_ext::MqttConfig;
-use tedge_mqtt_ext::MqttDynamicConnector;
 use tedge_mqtt_ext::TopicFilter;
 use tedge_script_ext::ScriptActor;
 use tedge_signal_ext::SignalActor;
@@ -388,7 +387,6 @@ impl Agent {
             let entity_store_server = EntityStoreServer::new(
                 entity_store,
                 mqtt_schema.clone(),
-                Box::new(MqttDynamicConnector::new(self.config.mqtt_config)),
                 &mut mqtt_actor_builder,
                 self.config.entity_auto_register,
             );

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

@@ -1,10 +1,22 @@
+use std::convert::Infallible;
+
 use async_trait::async_trait;
 use serde_json::Map;
 use serde_json::Value;
+use tedge_actors::Actor;
+use tedge_actors::Builder;
 use tedge_actors::LoggingSender;
+use tedge_actors::MappingSender;
+use tedge_actors::MessageReceiver;
 use tedge_actors::MessageSink;
+use tedge_actors::MessageSource;
+use tedge_actors::NoConfig;
+use tedge_actors::NoMessage;
+use tedge_actors::RuntimeError;
 use tedge_actors::Sender;
 use tedge_actors::Server;
+use tedge_actors::SimpleMessageBox;
+use tedge_actors::SimpleMessageBoxBuilder;
 use tedge_api::entity::EntityMetadata;
 use tedge_api::entity_store;
 use tedge_api::entity_store::EntityRegistrationMessage;
@@ -18,11 +30,10 @@ use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
 use tedge_api::pending_entity_store::RegisteredEntityData;
 use tedge_api::EntityStore;
-use tedge_mqtt_ext::MqttConnector;
+use tedge_mqtt_ext::DynSubscriptionsInner;
 use tedge_mqtt_ext::MqttMessage;
+use tedge_mqtt_ext::MqttRequest;
 use tedge_mqtt_ext::TopicFilter;
-use tokio::time::timeout;
-use tokio::time::Duration;
 use tracing::error;
 
 #[derive(Debug)]
@@ -56,27 +67,94 @@ pub enum EntityStoreResponse {
 pub struct EntityStoreServer {
     entity_store: EntityStore,
     mqtt_schema: MqttSchema,
-    mqtt_connector: Box<dyn MqttConnector>,
     mqtt_publisher: LoggingSender<MqttMessage>,
     entity_auto_register: bool,
+    retain_requests: SimpleMessageBox<NoMessage, TopicFilter>,
+}
+
+struct DeregistrationActorBuilder {
+    mqtt_publish: LoggingSender<MqttMessage>,
+    messages: SimpleMessageBoxBuilder<MqttMessage, NoMessage>,
+}
+
+impl Builder<DeregistrationActor> for DeregistrationActorBuilder {
+    type Error = Infallible;
+
+    fn try_build(self) -> Result<DeregistrationActor, Self::Error> {
+        Ok(DeregistrationActor {
+            mqtt_publish: self.mqtt_publish,
+            messages: self.messages.build(),
+        })
+    }
+}
+
+struct DeregistrationActor {
+    messages: SimpleMessageBox<MqttMessage, NoMessage>,
+    mqtt_publish: LoggingSender<MqttMessage>,
+}
+
+#[async_trait::async_trait]
+impl Actor for DeregistrationActor {
+    fn name(&self) -> &str {
+        todo!()
+    }
+
+    async fn run(mut self) -> Result<(), RuntimeError> {
+        while let Ok(Some(msg)) = self.messages.try_recv().await {
+            if msg.retain && !msg.payload.as_bytes().is_empty() {
+                let clear_msg = MqttMessage::new(&msg.topic, "").with_retain();
+                self.mqtt_publish.send(clear_msg).await.unwrap();
+            }
+        }
+        Ok(())
+    }
+}
+
+impl MessageSink<MqttMessage> for DeregistrationActorBuilder {
+    fn get_sender(&self) -> tedge_actors::DynSender<MqttMessage> {
+        self.messages.get_sender()
+    }
 }
 
 impl EntityStoreServer {
-    pub fn new(
+    pub fn new<M>(
         entity_store: EntityStore,
         mqtt_schema: MqttSchema,
-        mqtt_connector: Box<dyn MqttConnector>,
-        mqtt_actor: &mut impl MessageSink<MqttMessage>,
+        mqtt_actor: &mut M,
         entity_auto_register: bool,
-    ) -> Self {
-        let mqtt_publisher = LoggingSender::new("MqttPublisher".into(), mqtt_actor.get_sender());
+    ) -> Self
+    where
+        M: MessageSink<MqttRequest>
+            + for<'a> MessageSource<MqttMessage, &'a mut DynSubscriptionsInner>,
+    {
+        let mqtt_publisher = LoggingSender::new(
+            "MqttPublisher".into(),
+            Box::new(MappingSender::new(mqtt_actor.get_sender(), |msg| {
+                [MqttRequest::Publish(msg)]
+            })),
+        );
+        let mut retain_requests = SimpleMessageBoxBuilder::new("DeregistrationClient", 16);
+        let mut dyn_subs = DynSubscriptionsInner::new(TopicFilter::empty());
+        let messages = SimpleMessageBoxBuilder::new("DeregistrationActor", 16);
+        let dereg_actor_builder = DeregistrationActorBuilder {
+            mqtt_publish: mqtt_publisher.clone(),
+            messages,
+        };
+        mqtt_actor.connect_sink(&mut dyn_subs, &dereg_actor_builder);
+        let client_id = dyn_subs.client_id();
+        mqtt_actor.connect_mapped_source(NoConfig, &mut retain_requests, move |topics| {
+            [MqttRequest::RetrieveRetain(client_id, topics)]
+        });
+
+        // TODO - no don't do this!
+        tokio::spawn(dereg_actor_builder.build().run());
 
         Self {
             entity_store,
             mqtt_schema,
-            mqtt_connector,
             mqtt_publisher,
             entity_auto_register,
+            retain_requests: retain_requests.build(),
         }
     }
 
@@ -321,6 +399,10 @@ impl EntityStoreServer {
             return deleted;
         }
 
+        if deleted.is_empty() {
+            return vec![];
+        }
+
         let mut topics = TopicFilter::empty();
         for entity in deleted.iter() {
             for channel_filter in [
@@ -340,33 +422,7 @@ impl EntityStoreServer {
             }
         }
 
-        // A single connection to retrieve all retained metadata messages for all deleted entities
-        match self.mqtt_connector.connect(topics.clone()).await {
-            Ok(mut connection) => {
-                while let Ok(Some(message)) =
-                    timeout(Duration::from_secs(1), connection.next_message()).await
-                {
-                    if message.retain
-                        && !message.payload_bytes().is_empty()
-                        && topics.accept(&message)
-                    {
-                        let clear_msg = MqttMessage::new(&message.topic, "").with_retain();
-                        if let Err(err) = self.mqtt_publisher.send(clear_msg).await {
-                            error!(
-                                    "Failed to clear retained message on topic {} while de-registering {} due to: {err}",
-                                    topic_id,
-                                    message.topic
-                                );
-                        }
-                    }
-                }
-
-                connection.disconnect().await;
-            }
-            Err(err) => {
-                error!("Failed to create MQTT connection for clearing entity data: {err}");
-            }
-        }
+        self.retain_requests.send(topics).await.unwrap();
 
         // Clear the entity metadata of all deleted entities bottom up
         for entity in deleted.iter().rev() {