use futures::stream::{Stream, StreamExt};
use log::debug;
use std::pin::Pin;
use tokio::select;
use tokio::sync::watch;
use zbus::Connection;
use crate::Result;
use crate::api::models::ConnectionError;
use crate::dbus::{NMDeviceProxy, NMProxy};
pub async fn monitor_device_changes<F>(
conn: &Connection,
mut shutdown: watch::Receiver<()>,
callback: F,
) -> Result<()>
where
F: Fn() + 'static,
{
let nm = NMProxy::new(conn).await?;
let mut streams: Vec<Pin<Box<dyn Stream<Item = _>>>> = Vec::new();
let device_added_stream = nm.receive_device_added().await?;
let device_removed_stream = nm.receive_device_removed().await?;
let state_changed_stream = nm.receive_state_changed().await?;
streams.push(Box::pin(device_added_stream.map(|_| ())));
streams.push(Box::pin(device_removed_stream.map(|_| ())));
streams.push(Box::pin(state_changed_stream.map(|_| ())));
debug!("Subscribed to NetworkManager device signals");
let devices = nm.get_devices().await?;
for dev_path in devices {
if let Ok(dev) = NMDeviceProxy::builder(conn)
.path(dev_path.clone())?
.build()
.await
&& let Ok(state_stream) = dev.receive_device_state_changed().await
{
streams.push(Box::pin(state_stream.map(|_| ())));
debug!("Subscribed to state change signals on device: {dev_path}");
}
}
debug!(
"Monitoring {} signal streams for device changes",
streams.len()
);
let mut merged = futures::stream::select_all(streams);
loop {
select! {
_ = shutdown.changed() => {
debug!("Network monitoring shutdown requested");
break;
}
signal = merged.next() => {
match signal {
Some(_) => callback(),
None => break,
}
}
}
}
while let Some(_signal) = merged.next().await {
debug!("Device change detected");
callback();
}
Err(ConnectionError::Stuck("monitoring stream ended".into()))
}