fix(CompositeDevice): asyncronously fetch target capabilities instead of blocking

Author: ShadowApex

src/input/composite_device/mod.rs

@@ -334,16 +334,7 @@ impl CompositeDevice {
                         }
                     }
                     CompositeCommand::GetTargetCapabilities(sender) => {
-                        let target_caps = match self.targets.get_capabilities().await {
-                            Ok(caps) => caps,
-                            Err(e) => {
-                                log::error!("Failed to get target capabilities: {e:?}");
-                                continue;
-                            }
-                        };
-                        if let Err(e) = sender.send(target_caps).await {
-                            log::error!("Failed to send target capabilities: {:?}", e);
-                        }
+                        self.targets.send_capabilities(sender);
                     }
                     CompositeCommand::SetInterceptMode(mode) => self.set_intercept_mode(mode).await,
                     CompositeCommand::GetInterceptMode(sender) => {

src/input/composite_device/targets.rs

@@ -4,7 +4,10 @@ use std::{
     time::Duration,
 };
 
-use tokio::sync::mpsc::{self, Sender};
+use tokio::{
+    sync::mpsc::{self, Sender},
+    task::JoinSet,
+};
 use zbus::Connection;
 
 use crate::{
@@ -388,33 +391,41 @@ impl CompositeDeviceTargets {
         }
     }
 
-    // Get the capabilities of all target devices
-    pub async fn get_capabilities(&self) -> Result<HashSet<Capability>, Box<dyn Error>> {
-        let mut target_caps = HashSet::new();
+    /// Get the capabilities of all target devices and send the result to the
+    /// given channel.
+    pub fn send_capabilities(&self, sender: Sender<HashSet<Capability>>) {
+        // Use a JoinSet to query all targets simultaneously
+        let mut tasks = JoinSet::new();
         for target in self.target_devices.values() {
-            let caps = match target.get_capabilities().await {
-                Ok(caps) => caps,
-                Err(e) => {
-                    return Err(format!("Failed to get target capabilities: {e:?}").into());
-                }
-            };
-            for cap in caps {
-                target_caps.insert(cap);
-            }
+            let target = target.clone();
+            tasks.spawn(async move { target.get_capabilities().await });
         }
         for target in self.target_dbus_devices.values() {
-            let caps = match target.get_capabilities().await {
-                Ok(caps) => caps,
-                Err(e) => {
-                    return Err(format!("Failed to get target capabilities: {e:?}").into());
+            let target = target.clone();
+            tasks.spawn(async move { target.get_capabilities().await });
+        }
+
+        // Collect the results in a task and send them to the given channel
+        tokio::task::spawn(async move {
+            let mut target_caps = HashSet::new();
+            let results = tasks.join_all().await;
+            for result in results {
+                let caps = match result {
+                    Ok(caps) => caps,
+                    Err(e) => {
+                        log::warn!("Failed to get target capabilities: {e}");
+                        continue;
+                    }
+                };
+                for cap in caps {
+                    target_caps.insert(cap);
                 }
-            };
-            for cap in caps {
-                target_caps.insert(cap);
             }
-        }
 
-        Ok(target_caps)
+            if let Err(e) = sender.send(target_caps).await {
+                log::error!("Failed to send target capabilities: {e}");
+            }
+        });
     }
 
     /// Update the target capabilities of the given target device

src/input/target/client.rs

@@ -29,7 +29,7 @@ pub enum ClientError {
     #[error("failed to try to send command to device: {0}")]
     TrySendError(TrySendError<TargetCommand>),
     #[error("service encountered an error processing the request: {0}")]
-    ServiceError(Box<dyn std::error::Error>),
+    ServiceError(Box<dyn std::error::Error + Send + Sync>),
     #[error("device no longer exists")]
     ChannelClosed,
 }