wayle_network/core/settings/
monitoring.rs

1use std::sync::{Arc, Weak};
2
3use futures::StreamExt;
4use tokio_util::sync::CancellationToken;
5use tracing::{debug, warn};
6use wayle_core::remove_and_cancel;
7use wayle_traits::{ModelMonitoring, Reactive};
8use zbus::zvariant::OwnedObjectPath;
9
10use super::Settings;
11use crate::{
12    core::settings_connection::{ConnectionSettings, ConnectionSettingsParams},
13    error::Error,
14    proxy::settings::SettingsProxy,
15};
16
17impl ModelMonitoring for Settings {
18    type Error = Error;
19
20    async fn start_monitoring(self: Arc<Self>) -> Result<(), Self::Error> {
21        let Some(ref cancellation_token) = self.cancellation_token else {
22            return Err(Error::MissingCancellationToken);
23        };
24
25        let settings_proxy = SettingsProxy::new(&self.zbus_connection).await?;
26        let cancel_token = cancellation_token.clone();
27        let weak_self = Arc::downgrade(&self);
28
29        tokio::spawn(async move {
30            if let Err(e) = monitor(weak_self, settings_proxy, cancel_token).await {
31                warn!(error = %e, "cannot start settings monitor");
32            }
33        });
34
35        Ok(())
36    }
37}
38
39#[allow(clippy::cognitive_complexity)]
40async fn monitor(
41    weak_settings: Weak<Settings>,
42    settings_proxy: SettingsProxy<'_>,
43    cancellation_token: CancellationToken,
44) -> Result<(), Error> {
45    let mut connection_removed = settings_proxy.receive_connection_removed().await;
46    let mut connection_added = settings_proxy.receive_new_connection().await;
47    let mut hostname_changed = settings_proxy.receive_hostname_changed().await;
48    let mut can_modify_changed = settings_proxy.receive_can_modify_changed().await;
49    let mut version_id_changed = settings_proxy.receive_version_id_changed().await;
50
51    loop {
52        let Some(settings) = weak_settings.upgrade() else {
53            return Ok(());
54        };
55
56        tokio::select! {
57            _ = cancellation_token.cancelled() => {
58                debug!("SettingsMonitor cancelled");
59                return Ok(());
60            }
61            Some(event) = async { connection_added.as_mut().ok()?.next().await }, if
62                connection_added.is_ok() => {
63                    if let Ok(args) = event.args() {
64                        let _ = add_connection(args.connection, &settings).await;
65                    }
66                }
67            Some(event) = async { connection_removed.as_mut().ok()?.next().await }, if
68                connection_removed.is_ok() => {
69                    if let Ok(args) = event.args() {
70                        let _ = remove_connection(args.connection, &settings).await;
71                    }
72            }
73            Some(change) = hostname_changed.next() => {
74                if let Ok(new_hostname) = change.get().await {
75                    settings.hostname.set(new_hostname);
76                }
77            }
78            Some(change) = can_modify_changed.next() => {
79                if let Ok(new_can_modify) = change.get().await {
80                    settings.can_modify.set(new_can_modify);
81                }
82
83            }
84            Some(change) = version_id_changed.next() => {
85                if let Ok(new_version_id) = change.get().await {
86                    settings.version_id.set(new_version_id);
87                }
88            }
89            else => {
90                warn!("All property streams ended for Settings");
91                break;
92            }
93        }
94    }
95
96    Ok(())
97}
98
99async fn add_connection(
100    connection_path: OwnedObjectPath,
101    settings: &Arc<Settings>,
102) -> Result<(), Error> {
103    let new_connection = ConnectionSettings::get(ConnectionSettingsParams {
104        connection: &settings.zbus_connection,
105        path: connection_path.clone(),
106    })
107    .await?;
108
109    let mut current_connections = settings.connections.get();
110
111    let found_connection = current_connections
112        .iter()
113        .find(|connection| connection.object_path == connection_path);
114
115    if found_connection.is_none() {
116        current_connections.push(new_connection);
117        settings.connections.set(current_connections);
118    }
119
120    Ok(())
121}
122
123async fn remove_connection(
124    connection_path: OwnedObjectPath,
125    settings: &Arc<Settings>,
126) -> Result<(), Error> {
127    remove_and_cancel!(settings.connections.clone(), connection_path);
128    Ok(())
129}
wayle_network/core/settings/monitoring.rs

wayle_network/core/settings/
monitoring.rs
