dbus: Use FutureSource instead of a channel

Cargo.lock

@@ -544,6 +544,7 @@ checksum = "4dbf9978365bac10f54d1d4b04f7ce4427e51f71d61f2fe15e3fed5166474df7"
 dependencies = [
  "async-task",
  "bitflags 2.11.0",
+ "futures-core",
  "polling",
  "rustix 1.1.4",
  "slab",

Cargo.toml

@@ -12,7 +12,7 @@ members = ["cosmic-comp-config"]
 [dependencies]
 anyhow = { version = "1.0.102", features = ["backtrace"] }
 bitflags = "2.11.0"
-calloop = { version = "0.14.4", features = ["executor"] }
+calloop = { version = "0.14.4", features = ["executor", "stream"] }
 cosmic-comp-config = { path = "cosmic-comp-config", features = [
     "libdisplay-info",
     "output",

src/dbus/mod.rs

@@ -3,10 +3,9 @@ use crate::{
     utils::prelude::OutputExt,
 };
 use anyhow::{Context, Result};
-use calloop::{InsertError, LoopHandle, RegistrationToken};
+use calloop::{InsertError, LoopHandle, RegistrationToken, stream::StreamSource};
 use cosmic_comp_config::output::comp::OutputState;
-use futures_executor::{ThreadPool, block_on};
+use futures_executor::block_on;
-use futures_util::stream::StreamExt;
 use std::collections::HashMap;
 use tracing::{error, warn};
 use zbus::blocking::{Connection, fdo::DBusProxy};
@@ -17,19 +16,15 @@ pub mod logind;
 mod name_owners;
 mod power;
 
-pub fn init(
+pub fn init(evlh: &LoopHandle<'static, State>) -> Result<Vec<RegistrationToken>> {
-    evlh: &LoopHandle<'static, State>,
-    executor: &ThreadPool,
-) -> Result<Vec<RegistrationToken>> {
     let mut tokens = Vec::new();
 
     match block_on(power::init()) {
         Ok(power_daemon) => {
-            let (tx, rx) = calloop::channel::channel();
+            if let Ok(stream) = block_on(power_daemon.receive_hot_plug_detect()) {
-
+                let source = StreamSource::new(stream).unwrap();
-            let token = evlh
+                let token = evlh
-                .insert_source(rx, |event, _, state| match event {
+                    .insert_source(source, |_, _, state| {
-                    calloop::channel::Event::Msg(_) => {
                         let nodes = match &mut state.backend {
                             BackendData::Kms(kms) => {
                                 kms.drm_devices.keys().cloned().collect::<Vec<_>>()
@@ -56,24 +51,12 @@ pub fn init(
                                 }
                             }
                         }
-                    }
+                    })
-                    calloop::channel::Event::Closed => (),
+                    .map_err(|InsertError { error, .. }| error)
-                })
+                    .with_context(|| "Failed to add channel to event_loop")?;
-                .map_err(|InsertError { error, .. }| error)
-                .with_context(|| "Failed to add channel to event_loop")?;
 
-            // start helper thread
+                tokens.push(token);
-            executor.spawn_ok(async move {
+            }
-                if let Ok(mut msg_iter) = power_daemon.receive_hot_plug_detect().await {
-                    while let Some(msg) = msg_iter.next().await {
-                        if tx.send(msg).is_err() {
-                            break;
-                        }
-                    }
-                }
-            });
-
-            tokens.push(token);
         }
         Err(err) => {
             tracing::info!(?err, "Failed to connect to com.system76.PowerDaemon");

src/state.rs

@@ -729,7 +729,7 @@ impl State {
 
         let async_executor = ThreadPool::builder().pool_size(1).create().unwrap();
 
-        if let Err(err) = crate::dbus::init(&handle, &async_executor) {
+        if let Err(err) = crate::dbus::init(&handle) {
             tracing::warn!(?err, "Failed to initialize dbus handlers");
         }