futures::Stream for station state and network connectivity.

Allows clients to get asynchronously notified on connectivity changes.

`futures::Stream` for station state and network connectivity.

Allows clients to get asynchronously notified on connectivity changes.

Author: abezukor

Cargo.lock

@@ -14,6 +14,7 @@ zvariant = "5"
 uuid = { version = "1", features = ["v4"] }
 thiserror = "2.0.17"
 strum = {version = "0.27.2", features = ["derive"]}
+futures-lite = "2.6.1"
 
 [dev-dependencies]
 clap = {version = "4.5.48", features = ["derive"]}

Cargo.toml

@@ -1,3 +1,7 @@
+use futures_lite::{Stream, StreamExt};
+use zbus::Proxy;
+use zvariant::OwnedValue;
+
 pub mod session;
 
 pub mod station;
@@ -17,3 +21,15 @@ pub mod access_point;
 pub mod modes;
 
 pub mod error;
+
+async fn property_stream<'p, T: TryFrom<OwnedValue, Error = zvariant::Error> + Unpin>(
+    proxy: Proxy<'p>,
+    property_name: &'p str,
+) -> zbus::Result<impl Stream<Item = zbus::Result<T>> + Unpin> {
+    Ok(Box::pin(
+        proxy
+            .receive_property_changed(property_name)
+            .await
+            .then(|property_changed| async move { property_changed.get().await }),
+    ))
+}

src/lib.rs

@@ -1,5 +1,6 @@
 use std::{str::FromStr, sync::Arc};
 
+use futures_lite::{Stream, StreamExt};
 use strum::EnumString;
 use zbus::{Connection, Proxy, Result as ZbusResult};
 use zvariant::{OwnedObjectPath, OwnedValue};
@@ -50,9 +51,18 @@ impl Network {
     }
 
     pub async fn connected(&self) -> ZbusResult<bool> {
+        self.connected_stream()
+            .await?
+            .next()
+            .await
+            .ok_or_else(|| zbus::Error::Unsupported)?
+    }
+
+    pub async fn connected_stream(
+        &self,
+    ) -> zbus::Result<impl Stream<Item = zbus::Result<bool>> + Unpin> {
         let proxy = self.proxy().await?;
-        let is_connected: bool = proxy.get_property("Connected").await?;
-        Ok(is_connected)
+        crate::property_stream(proxy, "Connected").await
     }
 
     pub async fn device(&self) -> ZbusResult<Device> {

src/network.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 
+use futures_lite::{Stream, StreamExt};
 use strum::EnumString;
 use zvariant::{OwnedObjectPath, OwnedValue, Value};
 
@@ -50,8 +51,18 @@ impl Station {
     }
 
     pub async fn state(&self) -> zbus::Result<State> {
+        self.state_stream()
+            .await?
+            .next()
+            .await
+            .ok_or_else(|| zbus::Error::Unsupported)?
+    }
+
+    pub async fn state_stream(
+        &self,
+    ) -> zbus::Result<impl Stream<Item = zbus::Result<State>> + Unpin> {
         let proxy = self.proxy().await?;
-        proxy.get_property("State").await
+        crate::property_stream(proxy, "State").await
     }
 
     pub async fn connected_network(&self) -> zbus::Result<Option<Network>> {