wayle_network/wired/
monitoring.rs1use 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}