Move subscription mutex from config to connection

This fixes the handling of subscriptions when we clone mqtt_channel::Config. With the previous behaviour, any `Config` clones shared all their subscriptions.

Author: jarhodes314

crates/common/mqtt_channel/src/config.rs

@@ -11,7 +11,6 @@ use std::fmt::Formatter;
 use std::ops::Deref;
 use std::path::Path;
 use std::sync::Arc;
-use std::sync::Mutex;
 use zeroize::Zeroizing;
 
 pub const MAX_PACKET_SIZE: usize = 268435455;
@@ -33,7 +32,7 @@ pub struct Config {
     /// The list of topics to subscribe to on connect
     ///
     /// Default: An empty topic list
-    pub subscriptions: Arc<Mutex<TopicFilter>>,
+    pub subscriptions: TopicFilter,
 
     /// Clean the MQTT session upon connect if set to `true`.
     ///
@@ -164,7 +163,7 @@ impl Default for Config {
                 authentication: None,
             },
             session_name: None,
-            subscriptions: Arc::new(Mutex::new(TopicFilter::empty())),
+            subscriptions: TopicFilter::empty(),
             clean_session: false,
             queue_capacity: 1024,
             max_packet_size: 16 * 1024 * 1024,
@@ -214,8 +213,8 @@ impl Config {
     /// Add a list of topics to subscribe to on connect
     ///
     /// Can be called several times to subscribe to many topics.
-    pub fn with_subscriptions(self, topics: TopicFilter) -> Self {
-        self.subscriptions.lock().unwrap().add_all(topics);
+    pub fn with_subscriptions(mut self, topics: TopicFilter) -> Self {
+        self.subscriptions.add_all(topics);
         self
     }
 

crates/common/mqtt_channel/src/connection.rs

@@ -164,9 +164,15 @@ impl Connection {
         let (published_sender, published_receiver) = mpsc::unbounded();
         let (error_sender, error_receiver) = mpsc::unbounded();
         let (pub_done_sender, pub_done_receiver) = oneshot::channel();
+        let subscriptions = Arc::new(Mutex::new(config.subscriptions.clone()));
 
-        let (mqtt_client, event_loop) =
-            Connection::open(config, received_sender.clone(), error_sender.clone()).await?;
+        let (mqtt_client, event_loop) = Connection::open(
+            config,
+            received_sender.clone(),
+            error_sender.clone(),
+            subscriptions.clone(),
+        )
+        .await?;
         let permits = Arc::new(Semaphore::new(1));
         let permit = permits.clone().acquire_owned().await.unwrap();
         let pub_count = Arc::new(AtomicUsize::new(0));
@@ -179,6 +185,7 @@ impl Connection {
             pub_done_sender,
             permits,
             pub_count.clone(),
+            subscriptions.clone(),
         ));
         tokio::spawn(Connection::sender_loop(
             mqtt_client.clone(),
@@ -194,7 +201,7 @@ impl Connection {
             published: published_sender,
             errors: error_receiver,
             pub_done: pub_done_receiver,
-            subscriptions: SubscriberHandle::new(mqtt_client, config.subscriptions.clone()),
+            subscriptions: SubscriberHandle::new(mqtt_client, subscriptions),
         })
     }
 
@@ -207,6 +214,7 @@ impl Connection {
         config: &Config,
         mut message_sender: mpsc::UnboundedSender<MqttMessage>,
         mut error_sender: mpsc::UnboundedSender<MqttError>,
+        subscriptions: Arc<Mutex<TopicFilter>>,
     ) -> Result<(AsyncClient, EventLoop), MqttError> {
         const INSECURE_MQTT_PORT: u16 = 1883;
         const SECURE_MQTT_PORT: u16 = 8883;
@@ -234,7 +242,7 @@ impl Connection {
                     };
                     info!(target: "MQTT", "Connection established");
 
-                    let subscriptions = config.subscriptions.lock().unwrap().filters();
+                    let subscriptions = subscriptions.lock().unwrap().filters();
 
                     // Need check here otherwise it will hang waiting for a SubAck, and none will come when there is no subscription.
                     if subscriptions.is_empty() {
@@ -291,6 +299,7 @@ impl Connection {
         done: oneshot::Sender<()>,
         permits: Arc<Semaphore>,
         pub_count: Arc<AtomicUsize>,
+        subscriptions: Arc<Mutex<TopicFilter>>,
     ) -> Result<(), MqttError> {
         let mut triggered_disconnect = false;
         let mut disconnect_permit = None;
@@ -363,7 +372,7 @@ impl Connection {
                             // If session_name is not provided or if the broker session persistence
                             // is not enabled or working, then re-subscribe
 
-                            let subscriptions = config.subscriptions.lock().unwrap().filters();
+                            let subscriptions = subscriptions.lock().unwrap().filters();
                             // Need check here otherwise it will hang waiting for a SubAck, and none will come when there is no subscription.
                             if subscriptions.is_empty() {
                                 break;