simultaneous connection and better erroring

Cargo.lock

@@ -1143,7 +1143,7 @@ dependencies = [
 
 [[package]]
 name = "homekit-exporter"
-version = "0.4.1"
+version = "0.5.0"
 dependencies = [
  "clap",
  "env_logger",

homekit-controller/src/lib.rs

@@ -12,11 +12,10 @@ use x25519_dalek::{EphemeralSecret, PublicKey};
 
 pub use crate::{
     discovery::{spawn_discover_thread, MdnsDiscoveredList},
-    pairing_data::{CharacteristicType, Data, ServiceType},
+    pairing_data::{Accessory, CharacteristicType, Data, PythonPairingData, ServiceType},
 };
 use crate::{
     homekit_http::{AccessorySocket, DiscoveryError},
-    pairing_data::{Accessory, PythonPairingData},
     tlv8::{decode, HomekitState, TlvEncodableData, TlvEncode, TlvError, TlvType},
 };
 
@@ -43,7 +42,7 @@ impl PythonPairingData {
         Ok(connected_device)
     }
 
-    fn to_connection(&self, discovered: MdnsDiscoveredList) -> DeviceConnection {
+    pub fn to_connection(&self, discovered: MdnsDiscoveredList) -> DeviceConnection {
         DeviceConnection {
             accessories: Default::default(),
             discovered,
@@ -79,7 +78,7 @@ pub struct DeviceConnection {
 }
 
 impl DeviceConnection {
-    async fn connect(&mut self) -> Result<(), HomekitError> {
+    pub async fn connect(&mut self) -> Result<(), HomekitError> {
         if let Some(mut socket) = self.socket.take() {
             socket.disconnect().await?;
         }

homekit-exporter/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "homekit-exporter"
-version = "0.4.1"
+version = "0.5.0"
 edition = "2021"
 license = "Apache-2.0"
 description = "Prometheus exporter for HomeKit sensors"

homekit-exporter/packaging/PKGBUILD

@@ -1,7 +1,7 @@
 # Maintainer: Alex Janka <alex@alexjanka.com>
 
 pkgname=homekit-logger
-pkgver=0.4.1
+pkgver=0.5.0
 pkgrel=1
 pkgdesc="Prometheus exporter for HomeKit sensors"
 arch=('x86_64' 'aarch64')

homekit-exporter/src/main.rs

@@ -9,6 +9,7 @@ use homekit_controller::{
     spawn_discover_thread, CharacteristicType, DeviceConnection, HomekitError,
 };
 use server::launch;
+use tokio::task::JoinSet;
 
 mod server;
 
@@ -45,6 +46,7 @@ const MONITORED_CHARACTERISTICS: [CharacteristicType; 7] = [
 async fn rocket() -> rocket::Rocket<rocket::Build> {
     env_logger::init();
     let args = Args::parse();
+
     match args.command {
         Commands::Watch => match init(args.pairing_data).await {
             Ok(paired) => launch(paired, args.port),
@@ -76,27 +78,25 @@ async fn init(pairing_data: PathBuf) -> Result<HashMap<String, DeviceConnection>
     tokio::time::sleep(Duration::from_secs(2)).await;
     if pairing_data.is_file() {
         let devices = homekit_controller::load(pairing_data)?;
-        let mut connected_devices = HashMap::new();
+        let mut set = JoinSet::new();
         for (k, v) in devices {
-            let mut num = 0;
-            loop {
-                match v.connect(&discovered).await {
-                    Ok(v) => {
+            let discovered = discovered.clone();
+            set.spawn(async move {
+                let mut connection = v.to_connection(discovered.clone());
+
+                match tokio::time::timeout(Duration::from_secs(2), connection.connect()).await {
+                    Ok(Err(e)) => log::error!("error connecting to {k}: {e:?}"),
+                    Err(_) => log::error!("timeout connecting to {k}"),
+                    _ => {}
+                }
+                (k, connection)
+            });
+        }
+        let mut connected_devices = HashMap::new();
+        while let Some(Ok((k, v))) = set.join_next().await {
             connected_devices.insert(k, v);
-                        break;
-                    }
-                    Err(e) => {
-                        num += 1;
-                        log::error!("error connecting to {k}: {e:?}");
-                        if num > 10 {
-                            log::error!("\t...not connecting");
-                            break;
-                        }
-                    }
-                }
-                tokio::time::sleep(Duration::from_millis(1000)).await;
-            }
         }
+
         if connected_devices.is_empty() {
             Err(HomekitError::NoSuccessfulConnections)
         } else {

homekit-exporter/src/server.rs

@@ -1,7 +1,7 @@
 use homekit_controller::{Data, DeviceConnection};
 use rocket::State;
-use std::{collections::HashMap, ops::DerefMut};
-use tokio::sync::Mutex;
+use std::{collections::HashMap, time::Duration};
+use tokio::{sync::Mutex, task::JoinSet};
 
 use crate::MONITORED_CHARACTERISTICS;
 
@@ -15,13 +15,23 @@ pub fn launch(
         .mount("/", routes![metrics])
 }
 
+const METRIC_GATHER_TIMEOUT_SECS: u64 = 3;
+
 #[get("/metrics")]
 pub async fn metrics(state: &State<Mutex<HashMap<String, DeviceConnection>>>) -> Option<String> {
-    let mut s = String::new();
     let mut state = state.lock().await;
-    let mut shown_types = Vec::new();
-    for (name, connected) in state.deref_mut() {
-        connected.update_characteristics().await.ok()?;
+    let mut set = JoinSet::new();
+    for (name, mut connected) in state.drain() {
+        set.spawn(
+            async move {
+                let mut return_string = String::new();
+                let mut types_seen = Vec::new();
+
+                match tokio::time::timeout(Duration::from_secs(METRIC_GATHER_TIMEOUT_SECS), connected.update_characteristics()).await{
+                    Ok(r) => r.ok()?,
+                    Err(_) => log::info!("failed to update characteristics for {name}"),
+                }
+
 
                 for (aid, accessory) in &connected.accessories {
                     for service in accessory.services.values() {
@@ -30,15 +40,11 @@ pub async fn metrics(state: &State<Mutex<HashMap<String, DeviceConnection>>>) ->
                                 continue;
                             }
                             if let Some(value) = &characteristic.value {
-                        if !shown_types.contains(&characteristic.characteristic_type) {
-                            s.push_str(
-                                format!("# TYPE {} gauge\n", characteristic.characteristic_type)
-                                    .as_str(),
-                            );
-                            shown_types.push(characteristic.characteristic_type);
+                                if !types_seen.contains(&characteristic.characteristic_type) {
+                                    types_seen.push(characteristic.characteristic_type);
                                 }
-                        s.push_str(
-                            format!(
+
+                                return_string.push_str(format!(
                                         "{}{{hub=\"{}\",service_type=\"{}\",accessory_name=\"{}\",accessory=\"{}.{}\",service=\"{}.{}.{}\"}} {}\n",
                                         characteristic.characteristic_type,
                                         name,
@@ -61,13 +67,29 @@ pub async fn metrics(state: &State<Mutex<HashMap<String, DeviceConnection>>>) ->
                                             Data::Tlv8(v) => v.to_string(),
                                             Data::Data(v) => v.to_string(),
                                         }
-                            )
-                            .as_str(),
-                        )
+                                    ).as_str());
                             }
                         }
                     }
                 }
+
+                Some((name,connected,return_string,types_seen))
             }
+        );
+    }
+
+    let mut s = String::new();
+    let mut shown_types = Vec::new();
+    while let Some(Ok(Some((k, v, metrics, types)))) = set.join_next().await {
+        state.insert(k, v);
+        for c in types {
+            if !shown_types.contains(&c) {
+                s.push_str(format!("# TYPE {} gauge\n", c).as_str());
+                shown_types.push(c);
+            }
+        }
+        s.push_str(&metrics);
+    }
+
     Some(s)
 }