feat(power): implement periodic update for connected devices

Cargo.lock

@@ -460,6 +460,16 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "async-fn-stream"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e71711442f1016c768c259bec59300a10efe753bc3e686ec19e2c6a54a97c29b"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "async-fs"
 version = "2.1.2"
@@ -1624,6 +1634,7 @@ dependencies = [
  "as-result",
  "ashpd 0.9.2",
  "async-channel",
+ "async-fn-stream",
  "bluez-zbus",
  "chrono",
  "clap",

cosmic-settings/Cargo.toml

@@ -74,6 +74,7 @@ rustix = "0.38.41"
 gettext-rs = { version = "0.7.2", features = [
     "gettext-system",
 ], optional = true }
+async-fn-stream = "0.2.2"
 
 [dependencies.cosmic-settings-subscriptions]
 git = "https://github.com/pop-os/cosmic-settings-subscriptions"

cosmic-settings/src/pages/power/backend/mod.rs

@@ -1,8 +1,7 @@
 use chrono::{Duration, TimeDelta};
-use futures::future::join_all;
+use futures::{future::join_all, FutureExt, Stream, StreamExt};
-use futures::FutureExt;
 use upower_dbus::{BatteryState, BatteryType, DeviceProxy};
-use zbus::Connection;
+use zbus::{zvariant::ObjectPath, Connection};
 
 mod ppdaemon;
 mod s76powerdaemon;
@@ -239,6 +238,7 @@ pub struct ConnectedDevice {
     pub model: String,
     pub device_icon: &'static str,
     pub battery: Battery,
+    pub device_path: String,
 }
 
 async fn get_device_proxy<'a>() -> Result<upower_dbus::DeviceProxy<'a>, zbus::Error> {
@@ -406,6 +406,7 @@ impl Battery {
 impl ConnectedDevice {
     async fn from_device_maybe(proxy: DeviceProxy<'_>) -> Option<Self> {
         let device_type = proxy.type_().await.unwrap_or(BatteryType::Unknown);
+        let device_path = proxy.clone().into_inner().path().to_string();
         if matches!(
             device_type,
             BatteryType::Unknown | BatteryType::LinePower | BatteryType::Battery
@@ -445,9 +446,56 @@ impl ConnectedDevice {
             model,
             device_icon,
             battery,
+            device_path,
         })
     }
 
+    pub async fn device_removed_stream(
+        connection: &'_ Connection,
+    ) -> Result<impl Stream<Item = String> + '_, zbus::Error> {
+        let proxy = upower_dbus::UPowerProxy::new(connection).await?;
+        let stream = proxy.receive_device_removed().await?;
+
+        let transformed_stream = stream.filter_map(move |device_removed| async move {
+            let device_path: ObjectPath<'static> = match device_removed.args() {
+                Ok(args) => args.device().to_owned(),
+                Err(e) => {
+                    tracing::error!("Failed to get DeviceRemoved arguments: {e}");
+                    return None;
+                }
+            };
+            Some(device_path.to_string())
+        });
+
+        Ok(transformed_stream)
+    }
+
+    pub async fn device_added_stream(
+        connection: &'_ Connection,
+    ) -> Result<impl futures::Stream<Item = ConnectedDevice> + '_, zbus::Error> {
+        let proxy = upower_dbus::UPowerProxy::new(connection).await?;
+        let stream = proxy.receive_device_added().await?;
+
+        let transformed_stream = stream.filter_map(move |device_added| async move {
+            let device_path: ObjectPath<'static> = match device_added.args() {
+                Ok(args) => args.device().to_owned(),
+                Err(e) => {
+                    tracing::error!("Failed to get DeviceAdded arguments: {e}");
+                    return None;
+                }
+            };
+            match DeviceProxy::new(connection, &device_path).await {
+                Ok(device) => ConnectedDevice::from_device_maybe(device).await,
+                Err(e) => {
+                    tracing::error!("Failed to create DeviceProxy from {device_path}: {e}");
+                    None
+                }
+            }
+        });
+
+        Ok(transformed_stream)
+    }
+
     pub async fn update_connected_devices() -> Vec<Self> {
         let proxy = enumerate_devices().await;
 

cosmic-settings/src/pages/power/mod.rs

@@ -12,6 +12,7 @@ use cosmic::Task;
 use cosmic_config::{Config, CosmicConfigEntry};
 use cosmic_idle_config::CosmicIdleConfig;
 use cosmic_settings_page::{self as page, section, Section};
+use futures::StreamExt;
 use itertools::Itertools;
 use slab::Slab;
 use slotmap::SlotMap;
@@ -123,6 +124,52 @@ impl page::Page<crate::pages::Message> for Page {
                 let devices = ConnectedDevice::update_connected_devices().await;
                 Message::UpdateConnectedDevices(devices)
             }),
+            cosmic::Task::run(
+                async_fn_stream::fn_stream(|emitter| async move {
+                    let span = tracing::span!(tracing::Level::INFO, "power::device_stream task");
+                    let _span_handle = span.enter();
+
+                    let Ok(connection) = zbus::Connection::system().await else {
+                        tracing::error!("could not established zbus connection to system");
+                        return;
+                    };
+
+                    let added_stream = ConnectedDevice::device_added_stream(&connection).await;
+                    let removed_stream = ConnectedDevice::device_removed_stream(&connection).await;
+
+                    let added_future = async {
+                        match added_stream {
+                            Ok(stream) => {
+                                futures::pin_mut!(stream);
+                                while let Some(device) = stream.next().await {
+                                    tracing::info!(device = device.model, "device added");
+                                    emitter.emit(Message::DeviceConnect(device)).await;
+                                }
+                            }
+                            Err(err) => tracing::error!(?err, "cannot establish added stream"),
+                        }
+                    };
+
+                    let removed_future = async {
+                        match removed_stream {
+                            Ok(stream) => {
+                                futures::pin_mut!(stream);
+                                while let Some(device_path) = stream.next().await {
+                                    tracing::info!(device_path, "device removed");
+                                    emitter.emit(Message::DeviceDisconnect(device_path)).await;
+                                }
+                            }
+                            Err(err) => tracing::error!(?err, "cannot establish removed stream"),
+                        }
+                    };
+
+                    futures::pin_mut!(added_future);
+                    futures::pin_mut!(removed_future);
+
+                    futures::future::select(added_future, removed_future).await;
+                }),
+                |msg| msg,
+            ),
         ];
 
         let (task, handle) = cosmic::Task::batch(futures)
@@ -147,6 +194,8 @@ pub enum Message {
     PowerProfileChange(PowerProfile),
     UpdateBattery(Battery),
     UpdateConnectedDevices(Vec<ConnectedDevice>),
+    DeviceDisconnect(String),
+    DeviceConnect(ConnectedDevice),
     ScreenOffTimeChange(Option<Duration>),
     SuspendOnAcTimeChange(Option<Duration>),
     SuspendOnBatteryTimeChange(Option<Duration>),
@@ -192,6 +241,12 @@ impl Page {
                     tracing::error!("failed to set suspend on battery time: {}", err)
                 }
             }
+            Message::DeviceDisconnect(device_path) => self
+                .connected_devices
+                .retain(|device| device.device_path != device_path),
+            Message::DeviceConnect(connected_device) => {
+                self.connected_devices.push(connected_device)
+            }
         };
     }
 }