perf: concurrent HA API calls

perf: only update HA API for states that have changed

Author: AntoineGS

src/home_assistant/api.rs

@@ -9,24 +9,31 @@ use home_assistant_rest::Client;
 use log::{error, info};
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
+use futures_util::future::try_join_all;
 
 pub struct HaApi {
-    client: Client,
     ha_configuration: HaConfiguration,
 }
 
 impl HaApi {
     pub fn new(ha_configuration: HaConfiguration) -> anyhow::Result<Self> {
-        let client = Client::new(&*ha_configuration.url, &*ha_configuration.long_live_token)?;
         Ok(Self {
-            client,
             ha_configuration,
         })
     }
 
-    /* friendly_name is needed as API calls wipe the configured name */
-    async fn update_ha(&self, state: &AtomicBool, ha_entity: &HaEntity) -> anyhow::Result<()> {
-        let api_status = self.client.get_api_status().await.unwrap();
+    // friendly_name is needed as API calls wipe the configured name
+    async fn update_ha(&self, state: &AtomicBool, prev_state: &AtomicBool, ha_entity: &HaEntity, force_update: bool) -> anyhow::Result<()> {
+        let state_bool = state.load(Ordering::Relaxed);
+        let prev_state_bool = prev_state.load(Ordering::Relaxed);
+
+        // we exit early if nothing has changed, and we are not forcing an update
+        if state_bool == prev_state_bool && !force_update {
+            return Ok(());
+        }
+
+        let client = Client::new(&*self.ha_configuration.url, &*self.ha_configuration.long_live_token)?;
+        let api_status = client.get_api_status().await.unwrap();
 
         if api_status.message != "API running." {
             error!("Home Assistant API cannot be reached");
@@ -39,8 +46,6 @@ impl HaApi {
             ha_entity.friendly_name.to_string(),
         );
 
-        let state_bool = state.load(Ordering::Relaxed);
-
         let icon = if state_bool {
             &ha_entity.icons.on
         } else {
@@ -58,67 +63,81 @@ impl HaApi {
 
         info!("Updating HA entity ({}) to '{}'", &ha_entity.id, &state_str);
 
-        let post_states_res = self.client.post_states(params).await;
+        let post_states_res = client.post_states(params).await;
 
         if post_states_res.is_err() {
             error!("{}", post_states_res.unwrap_err());
         };
 
+        prev_state.store(state_bool, Ordering::Relaxed);
+
         Ok(())
     }
 }
 
 #[async_trait]
 impl Listener for HaApi {
-    async fn notify_changed(&self, teams_states: &TeamsStates) -> anyhow::Result<()> {
+    async fn notify_changed(&self, teams_states: &TeamsStates, force_update: bool) -> anyhow::Result<()> {
         // Reflection would be nice here... Tried with bevy_reflect but ran into an issue with AtomicBool
-        self.update_ha(
+        let mut futures = Vec::new();
+
+        futures.push(self.update_ha(
             &teams_states.is_in_meeting,
+            &teams_states.prev_is_in_meeting,
             &self.ha_configuration.entities.is_in_meeting,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.is_video_on,
+            &teams_states.prev_is_video_on,
             &self.ha_configuration.entities.is_video_on,
-        )
-            .await?;
-        
-        self.update_ha(
+            force_update,
+        ));
+
+        futures.push(self.update_ha(
             &teams_states.is_muted,
+            &teams_states.prev_is_muted,
             &self.ha_configuration.entities.is_muted,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.is_hand_raised,
+            &teams_states.prev_is_hand_raised,
             &self.ha_configuration.entities.is_hand_raised,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.is_recording_on,
+            &teams_states.prev_is_recording_on,
             &self.ha_configuration.entities.is_recording_on,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.is_background_blurred,
+            &teams_states.prev_is_background_blurred,
             &self.ha_configuration.entities.is_background_blurred,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.is_sharing,
+            &teams_states.prev_is_sharing,
             &self.ha_configuration.entities.is_sharing,
-        )
-            .await?;
+            force_update,
+        ));
 
-        self.update_ha(
+        futures.push(self.update_ha(
             &teams_states.has_unread_messages,
+            &teams_states.prev_has_unread_messages,
             &self.ha_configuration.entities.has_unread_messages,
-        )
-            .await?;
+            force_update,
+        ));
+
+        try_join_all(futures).await?;
 
         Ok(())
     }

src/mqtt/api.rs

@@ -38,7 +38,7 @@ impl MqttApi {
 
 #[async_trait]
 impl Listener for MqttApi {
-    async fn notify_changed(&self, teams_states: &TeamsStates) -> anyhow::Result<()> {
+    async fn notify_changed(&self, teams_states: &TeamsStates, _: bool) -> anyhow::Result<()> {
         let muted = &*bool_to_str(teams_states.is_muted.load(Ordering::Relaxed));
         let video_on = &*bool_to_str(teams_states.is_video_on.load(Ordering::Relaxed));
         let hand_raised = &*bool_to_str(teams_states.is_hand_raised.load(Ordering::Relaxed));

src/teams_ws/api.rs

@@ -34,13 +34,21 @@ impl TeamsAPI {
     pub fn new(conf: &TeamsConfiguration) -> Self {
         let teams_states = Arc::new(TeamsStates {
             is_muted: AtomicBool::new(false),
+            prev_is_muted: AtomicBool::new(false),
             is_video_on: AtomicBool::new(false),
+            prev_is_video_on: AtomicBool::new(false),
             is_hand_raised: AtomicBool::new(false),
+            prev_is_hand_raised: AtomicBool::new(false),
             is_in_meeting: AtomicBool::new(false),
+            prev_is_in_meeting: AtomicBool::new(false),
             is_recording_on: AtomicBool::new(false),
+            prev_is_recording_on: AtomicBool::new(false),
             is_background_blurred: AtomicBool::new(false),
+            prev_is_background_blurred: AtomicBool::new(false),
             is_sharing: AtomicBool::new(false),
+            prev_is_sharing: AtomicBool::new(false),
             has_unread_messages: AtomicBool::new(false),
+            prev_has_unread_messages: AtomicBool::new(false),
         });
 
         let api_token = if !conf.api_token.is_empty() {
@@ -160,12 +168,14 @@ async fn parse_data_and_notify_listener(
         )
             .await;
 
-        if force_update.swap(false, Ordering::Relaxed) || has_changed {
+        let force_update = force_update.swap(false, Ordering::Relaxed);
+
+        if force_update || has_changed {
             // 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;
+                let result = listener.lock().unwrap().notify_changed(&teams_states, force_update).await;
 
                 if result.is_ok() || (i == MAX_RETRIES) {
                     result?;

src/teams_ws/states.rs

@@ -2,11 +2,19 @@ use std::sync::atomic::AtomicBool;
 
 pub struct TeamsStates {
     pub is_muted: AtomicBool,
+    pub prev_is_muted: AtomicBool,
     pub is_video_on: AtomicBool,
+    pub prev_is_video_on: AtomicBool,
     pub is_hand_raised: AtomicBool,
+    pub prev_is_hand_raised: AtomicBool,
     pub is_in_meeting: AtomicBool,
+    pub prev_is_in_meeting: AtomicBool,
     pub is_recording_on: AtomicBool,
+    pub prev_is_recording_on: AtomicBool,
     pub is_background_blurred: AtomicBool,
+    pub prev_is_background_blurred: AtomicBool,
     pub is_sharing: AtomicBool,
+    pub prev_is_sharing: AtomicBool,
     pub has_unread_messages: AtomicBool,
+    pub prev_has_unread_messages: AtomicBool,
 }

src/traits.rs

@@ -6,6 +6,6 @@ pub trait StopController {}
 // todo: convert to Rust built-in once 1.75 is released
 #[async_trait]
 pub trait Listener {
-    async fn notify_changed(&self, teams_states: &TeamsStates) -> anyhow::Result<()>;
+    async fn notify_changed(&self, teams_states: &TeamsStates, force_update: bool) -> anyhow::Result<()>;
     fn reconnect(&mut self);
 }