use futures::stream::{Stream, StreamExt};
use log::{debug, warn};
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, NMWirelessProxy};
use crate::types::constants::device_type;
pub async fn monitor_network_changes<F>(
conn: &Connection,
mut shutdown: watch::Receiver<()>,
callback: F,
) -> Result<()>
where
F: Fn() + 'static,
{
let nm = NMProxy::new(conn).await?;
let devices = nm.get_devices().await?;
let mut streams: Vec<Pin<Box<dyn Stream<Item = _>>>> = Vec::new();
for dev_path in devices {
let dev = NMDeviceProxy::builder(conn)
.path(dev_path.clone())?
.build()
.await?;
if dev.device_type().await? != device_type::WIFI {
continue;
}
let wifi = NMWirelessProxy::builder(conn)
.path(dev_path.clone())?
.build()
.await?;
let added_stream = wifi.receive_access_point_added().await?;
let removed_stream = wifi.receive_access_point_removed().await?;
streams.push(Box::pin(added_stream.map(|_| ())));
streams.push(Box::pin(removed_stream.map(|_| ())));
debug!("Subscribed to AP signals on device: {dev_path}");
}
if streams.is_empty() {
warn!("No Wi-Fi devices found to monitor");
return Err(ConnectionError::NoWifiDevice);
}
debug!(
"Monitoring {} signal streams for network 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!("Network change detected");
callback();
}
Err(ConnectionError::Stuck("monitoring stream ended".into()))
}