Ensure dynamic connection is only created when needed

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Author: jarhodes314

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/extensions/tedge_mqtt_ext/src/lib.rs

@@ -326,7 +326,7 @@ impl FromPeers {
             match message {
                 PublishOrSubscribe::Publish(message) => {
                     tracing::debug!(target: "MQTT pub", "{message}");
-                    SinkExt::send(outgoing_mqtt,  message)
+                    SinkExt::send(outgoing_mqtt, message)
                         .await
                         .map_err(Box::new)?;
                 }
@@ -365,18 +365,29 @@ impl FromPeers {
                             client.unsubscribe_many(diff.unsubscribe).await.unwrap();
                         }
                     });
-                    let dynamic_connection_config = self.base_config.clone().with_subscriptions(tf);
-                    let mut sender = tx_to_peers.clone();
-                    tokio::spawn(async move {
-                        let mut conn = mqtt_channel::Connection::new(&dynamic_connection_config).await.unwrap();
-                        while let Ok(msg) = tokio::time::timeout(Duration::from_secs(10), conn.received.next()).await {
-                            if let Some(msg) = msg {
-                                if msg.retain {
-                                    SinkExt::send(&mut sender, (request.client_id, msg)).await.unwrap();
+                    if !tf.is_empty() {
+                        let dynamic_connection_config =
+                            self.base_config.clone().with_subscriptions(tf);
+                        let mut sender = tx_to_peers.clone();
+                        tokio::spawn(async move {
+                            let mut conn =
+                                mqtt_channel::Connection::new(&dynamic_connection_config)
+                                    .await
+                                    .unwrap();
+                            while let Ok(msg) =
+                                tokio::time::timeout(Duration::from_secs(10), conn.received.next())
+                                    .await
+                            {
+                                if let Some(msg) = msg {
+                                    if msg.retain {
+                                        SinkExt::send(&mut sender, (request.client_id, msg))
+                                            .await
+                                            .unwrap();
+                                    }
                                 }
                             }
-                        }
-                    });
+                        });
+                    }
                 }
             }
         }
@@ -510,9 +521,13 @@ impl Actor for MqttActor {
         tokio::spawn(async move { self.trie_service.run().await });
 
         tedge_utils::futures::select(
-            self.from_peers
-                .relay_messages_to(&mut mqtt_client.published, &mut to_peer, mqtt_client.subscriptions),
-            self.to_peers.relay_messages_from(&mut mqtt_client.received, &mut from_peer),
+            self.from_peers.relay_messages_to(
+                &mut mqtt_client.published,
+                &mut to_peer,
+                mqtt_client.subscriptions,
+            ),
+            self.to_peers
+                .relay_messages_from(&mut mqtt_client.received, &mut from_peer),
         )
         .await
     }
@@ -624,12 +639,12 @@ mod unit_tests {
                 id: 0
             ))
             .await;
-        
+
         actor
             .subscribe_client
             .assert_unsubscribed_from(["a/b".into()])
             .await;
-    
+
         actor.close().await;
     }
 
@@ -752,9 +767,15 @@ mod unit_tests {
             let subscribe_client = MockSubscriberOps::default();
             let from_peers = {
                 let client = subscribe_client.clone();
-                tokio::spawn(async move { fp.relay_messages_to(&mut outgoing_mqtt, &mut tx, client).await })
+                tokio::spawn(async move {
+                    fp.relay_messages_to(&mut outgoing_mqtt, &mut tx, client)
+                        .await
+                })
             };
-            let to_peers = tokio::spawn(async move { tp.relay_messages_from(&mut incoming_messages, &mut rx).await });
+            let to_peers = tokio::spawn(async move {
+                tp.relay_messages_from(&mut incoming_messages, &mut rx)
+                    .await
+            });
 
             Self {
                 subscribe_client,
@@ -771,7 +792,7 @@ mod unit_tests {
 
         /// Closes the channels associated with this actor and waits for both
         /// loops to finish executing
-        /// 
+        ///
         /// This allows the `SubscriberOps::drop` implementation to reliably
         /// flag any unasserted communication
         pub async fn close(mut self) {

crates/extensions/tedge_mqtt_ext/src/tests.rs

@@ -312,6 +312,52 @@ async fn dynamic_subscribers_receive_retain_messages() {
     }
 }
 
+#[tokio::test]
+async fn dynamic_subscribers_receive_retain_messages_when_upgrading_topic() {
+    let broker = mqtt_tests::test_mqtt_broker();
+    let mqtt_config = MqttConfig::default().with_port(broker.port);
+    let mut mqtt = MqttActorBuilder::new(mqtt_config);
+
+    broker
+        .publish_with_opts("a/b", "retain", QoS::AtLeastOnce, true)
+        .await
+        .unwrap();
+
+    let mut client_0 = SimpleMessageBoxBuilder::<_, PublishOrSubscribe>::new("dyn-subscriber", 16);
+    let mut client_1 = SimpleMessageBoxBuilder::<_, PublishOrSubscribe>::new("dyn-subscriber1", 16);
+    let _client_id_0 = mqtt.connect_id_sink(TopicFilter::new_unchecked("a/b"), &client_0);
+    let client_id_1 = mqtt.connect_id_sink(TopicFilter::empty(), &client_1);
+    client_0.connect_sink(NoConfig, &mqtt);
+    client_1.connect_sink(NoConfig, &mqtt);
+    let mqtt = mqtt.build();
+    tokio::spawn(async move { mqtt.run().await.unwrap() });
+    let mut client_0 = client_0.build();
+    let mut client_1 = client_1.build();
+
+    let msg = MqttMessage::new(&Topic::new_unchecked("a/b"), "retain").with_retain();
+
+    // client_0 receives retain message upon subscribing to "a/b"
+    assert_eq!(timeout(client_0.recv()).await.unwrap(), msg);
+
+    client_1
+        .send(PublishOrSubscribe::Subscribe(SubscriptionRequest {
+            diff: SubscriptionDiff {
+                subscribe: ["a/+".into()].into(),
+                unsubscribe: [].into(),
+            },
+            client_id: client_id_1,
+        }))
+        .await
+        .unwrap();
+
+    // client_1 should receive both "a/b" and "b/c" retain messages upon subscribing
+    let recv = timeout(client_1.recv()).await.unwrap();
+    assert_eq!(recv, msg);
+
+    // Retain message might be redelivered to client_0, but this is
+    // implementation dependent, don't assert either way
+}
+
 async fn timeout<T>(fut: impl Future<Output = T>) -> T {
     tokio::time::timeout(Duration::from_secs(1), fut)
         .await