wayle_network/wired/
monitoring.rs

1use std::sync::{Arc, Weak};
2
3use tokio_stream::StreamExt;
4use tokio_util::sync::CancellationToken;
5use tracing::debug;
6use wayle_traits::ModelMonitoring;
7
8use super::Wired;
9use crate::{
10    core::config::ip4_config::Ip4Config,
11    error::Error,
12    proxy::devices::DeviceProxy,
13    types::states::{NMDeviceState, NetworkStatus},
14};
15
16impl ModelMonitoring for Wired {
17    type Error = Error;
18
19    async fn start_monitoring(self: Arc<Self>) -> Result<(), Self::Error> {
20        let device_arc = Arc::new(self.device.clone());
21        device_arc.start_monitoring().await?;
22
23        let Some(ref cancellation_token) = self.device.core.cancellation_token else {
24            return Err(Error::MissingCancellationToken);
25        };
26
27        let cancel_token = cancellation_token.clone();
28        let weak_self = Arc::downgrade(&self);
29        let device_proxy = DeviceProxy::new(
30            &self.device.core.connection,
31            self.device.core.object_path.clone(),
32        )
33        .await
34        .map_err(Error::DbusError)?;
35
36        tokio::spawn(async move {
37            let _ = monitor_wired_connectivity(weak_self, device_proxy, cancel_token).await;
38        });
39
40        Ok(())
41    }
42}
43
44async fn monitor_wired_connectivity(
45    weak_wired: Weak<Wired>,
46    proxy: DeviceProxy<'static>,
47    cancellation_token: CancellationToken,
48) -> Result<(), Error> {
49    let mut connectivity_changed = proxy.receive_state_changed().await;
50    let mut ip4_config_changed = proxy.receive_ip4_config_changed().await;
51
52    loop {
53        let Some(wired) = weak_wired.upgrade() else {
54            return Ok(());
55        };
56
57        tokio::select! {
58            _ = cancellation_token.cancelled() => {
59                debug!("Wired monitoring cancelled for {}", wired.device.core.object_path);
60                return Ok(());
61            }
62            Some(change) = connectivity_changed.next() => {
63                if let Ok(new_connectivity) = change.get().await {
64                    let device_state = NMDeviceState::from_u32(new_connectivity);
65                    let status = NetworkStatus::from_device_state(device_state);
66                    wired.connectivity.set(status);
67
68                    if status == NetworkStatus::Connected {
69                        let ip = Ip4Config::resolve_address(
70                            &wired.device.core.connection,
71                            wired.device.core.ip4_config.get(),
72                        ).await;
73                        wired.ip4_address.set(ip);
74                    }
75                }
76            }
77            Some(change) = ip4_config_changed.next() => {
78                if let Ok(ip4_path) = change.get().await {
79                    let ip = Ip4Config::resolve_address(
80                        &wired.device.core.connection,
81                        ip4_path,
82                    ).await;
83                    wired.ip4_address.set(ip);
84                }
85            }
86            else => {
87                break;
88            }
89        }
90    }
91
92    Ok(())
93}
wayle_network/wired/monitoring.rs

wayle_network/wired/
monitoring.rs
