Add test to ensure that cloning MQTT configurations doesn't share subscriptions

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

Author: jarhodes314

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

@@ -255,8 +255,14 @@ async fn dynamic_subscribers_receive_retain_messages() {
     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();
-    broker.publish_with_opts("b/c", "retain", QoS::AtLeastOnce, true).await.unwrap();
+    broker
+        .publish_with_opts("a/b", "retain", QoS::AtLeastOnce, true)
+        .await
+        .unwrap();
+    broker
+        .publish_with_opts("b/c", "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);
@@ -275,14 +281,27 @@ async fn dynamic_subscribers_receive_retain_messages() {
     // 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/b".into(), "b/c".into()].into(), unsubscribe: [].into() }, client_id: client_id_1 })).await.unwrap();
+    client_1
+        .send(PublishOrSubscribe::Subscribe(SubscriptionRequest {
+            diff: SubscriptionDiff {
+                subscribe: ["a/b".into(), "b/c".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();
     let recv2 = timeout(client_1.recv()).await.unwrap();
 
     // Retain message should not be redelivered to client_0
-    assert!(tokio::time::timeout(Duration::from_millis(200), client_0.recv()).await.is_err());
+    assert!(
+        tokio::time::timeout(Duration::from_millis(200), client_0.recv())
+            .await
+            .is_err()
+    );
 
     if recv.topic.name == "a/b" {
         assert_eq!(recv, msg);