fix: ensure MQTT is reconnected upon connection failure, fixes #19

Author: AntoineGS

src/home_assistant/api.rs

@@ -73,50 +73,54 @@ impl Listener for HaApi {
             &teams_states.is_muted,
             &self.ha_configuration.entities.is_muted,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_video_on,
             &self.ha_configuration.entities.is_video_on,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_hand_raised,
             &self.ha_configuration.entities.is_hand_raised,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_in_meeting,
             &self.ha_configuration.entities.is_in_meeting,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_recording_on,
             &self.ha_configuration.entities.is_recording_on,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_background_blurred,
             &self.ha_configuration.entities.is_background_blurred,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.is_sharing,
             &self.ha_configuration.entities.is_sharing,
         )
-        .await?;
+            .await?;
 
         self.update_ha(
             &teams_states.has_unread_messages,
             &self.ha_configuration.entities.has_unread_messages,
         )
-        .await?;
+            .await?;
 
         Ok(())
     }
+
+    fn reconnect(&mut self) {
+        // considered not needed for now, as I believe the API will reconnect upon failure (not tested)
+    }
 }

src/main.rs

@@ -1,4 +1,5 @@
 #![windows_subsystem = "windows"]
+
 mod configuration;
 mod home_assistant;
 mod logging;
@@ -10,7 +11,7 @@ mod utils;
 
 use std::process::exit;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time;
 
 use crate::configuration::get_configuration;
@@ -72,7 +73,7 @@ async fn run_apis(
     };
 
     teams_api
-        .start_listening(Arc::new(listener), is_running.clone(), toggle_mute.clone())
+        .start_listening(Arc::new(Mutex::new(listener)), is_running.clone(), toggle_mute.clone())
         .await?;
 
     Ok(())

src/mqtt/api.rs

@@ -76,4 +76,20 @@ impl Listener for MqttApi {
 
         Ok(())
     }
+
+    fn reconnect(&mut self) {
+        let mut mqtt_options = MqttOptions::new(
+            "teams-status",
+            self.mqtt_configuration.url(),
+            self.mqtt_configuration.port,
+        );
+
+        mqtt_options.set_credentials(&self.mqtt_configuration.username, &self.mqtt_configuration.password);
+        mqtt_options.set_keep_alive(Duration::from_secs(5));
+        let (client, mut event_loop) = AsyncClient::new(mqtt_options, 10);
+
+        self.client = client;
+        // mqttc requires this to work
+        task::spawn(async move { while let Ok(_) = event_loop.poll().await {} });
+    }
 }

src/teams_ws/api.rs

@@ -8,8 +8,9 @@ use futures_util::{future, pin_mut, SinkExt, StreamExt};
 use json::JsonValue;
 use log::{error, info};
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time;
+use std::time::Duration;
 use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
 
 const JSON_MEETING_UPDATE: &str = "meetingUpdate";
@@ -23,6 +24,7 @@ const JSON_IS_BACKGROUND_BLURRED: &str = "isBackgroundBlurred";
 const JSON_IS_SHARING: &str = "isSharing";
 const JSON_HAS_UNREAD_MESSAGES: &str = "hasUnreadMessages";
 const JSON_TOKEN_REFRESH: &str = "tokenRefresh";
+
 pub struct TeamsAPI {
     pub teams_states: Arc<TeamsStates>,
     pub url: String,
@@ -56,7 +58,7 @@ impl TeamsAPI {
 
     pub async fn start_listening(
         &self,
-        listener: Arc<Box<dyn Listener>>,
+        listener: Arc<Mutex<Box<dyn Listener>>>,
         is_running: Arc<AtomicBool>,
         toggle_mute: Arc<AtomicBool>,
     ) -> anyhow::Result<()> {
@@ -72,16 +74,17 @@ impl TeamsAPI {
                     let data = &message.unwrap().into_data();
                     let json = String::from_utf8_lossy(data);
                     info!("{}", json);
-                    let parse_result = parse_data(
+
+                    let parse_result = parse_data_and_notify_listener(
                         &json,
                         listener.clone(),
                         self.teams_states.clone(),
                         force_update.clone(),
                     )
-                    .await;
+                        .await;
 
                     if parse_result.is_err() {
-                        error!("{}", parse_result.unwrap_err())
+                        error!("Unable to parse or notify listener, abandoning: {}", parse_result.unwrap_err());
                     }
                 }
             })
@@ -127,9 +130,9 @@ async fn update_value(
     teams_state_value.swap(new_value, Ordering::Relaxed) != new_value
 }
 
-async fn parse_data(
+async fn parse_data_and_notify_listener(
     json: &str,
-    listener: Arc<Box<dyn Listener>>,
+    listener: Arc<Mutex<Box<dyn Listener>>>,
     teams_states: Arc<TeamsStates>,
     force_update: Arc<AtomicBool>,
 ) -> anyhow::Result<()> {
@@ -148,17 +151,32 @@ async fn parse_data(
             &answer,
             JSON_IS_BACKGROUND_BLURRED,
         )
-        .await;
+            .await;
         has_changed |= update_value(&teams_states.is_sharing, &answer, JSON_IS_SHARING).await;
         has_changed |= update_value(
             &teams_states.has_unread_messages,
             &answer,
             JSON_HAS_UNREAD_MESSAGES,
         )
-        .await;
+            .await;
 
         if force_update.swap(false, Ordering::Relaxed) || has_changed {
-            listener.notify_changed(&teams_states).await?;
+            // Issue!: This will only run once regardless of MAX_RETRIES
+            // for some reason after a reconnect the notify_changed will get a pass no matter what
+            const MAX_RETRIES: i32 = 3;
+            for i in 1..MAX_RETRIES {
+                let result = listener.lock().unwrap().notify_changed(&teams_states).await;
+
+                if result.is_ok() || (i == MAX_RETRIES) {
+                    result?
+                }
+                // we will try to reconnect if the connection failed
+                else if i < MAX_RETRIES {
+                    error!("{}: Reconnecting and retrying...", result.unwrap_err());
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+                    listener.lock().unwrap().reconnect();
+                }
+            }
         }
     } else if answer.has_key(JSON_TOKEN_REFRESH) && !answer[JSON_TOKEN_REFRESH].is_empty() {
         change_teams_configuration(

src/traits.rs

@@ -7,4 +7,5 @@ pub trait StopController {}
 #[async_trait]
 pub trait Listener {
     async fn notify_changed(&self, teams_states: &TeamsStates) -> anyhow::Result<()>;
+    fn reconnect(&mut self);
 }