Progress without channel

rust/agama-dbus-server/src/dbus/interfaces/progress.rs

@@ -1,29 +1,80 @@
-use async_std::channel::Receiver;
-use zbus::{dbus_interface, Connection, SignalContext};
+use zbus::{dbus_interface, Connection, SignalContext, InterfaceRef};
 use crate::error::Error;
 
 pub struct Progress {
+    iface_ref: Option<InterfaceRef<Progress>>,
     total_steps: u8,
+    current_step: Option<Step>,
+}
+
+#[derive(Clone)]
+struct Step(u8, String);
+
+impl Default for Step {
+    fn default() -> Self {
+        Step(0, String::new())
+    }
 }
 
 impl Progress {
     pub fn new() -> Self {
         Self {
+            iface_ref: None,
             total_steps: 0,
+            current_step: None
         }
     }
 
-    // pub async fn update(&mut self, reference: InterfaceRef<Progress>, value: u8) -> Result<(), Error> {
-    pub async fn set_total_steps(
-        &mut self,
-        value: u8,
-        signal_context: &SignalContext<'_>
-    ) -> Result<(), Error> {
+    pub fn set_iface_ref(&mut self, iface_ref: InterfaceRef<Self>) {
+        self.iface_ref = Some(iface_ref);
+    }
+
+    pub async fn start(&mut self, total_steps: u8) -> Result<(), Error> {
+        self.set_total_steps(total_steps).await?;
+        self.set_current_step(None).await?;
+        Ok(())
+    }
+
+    pub async fn step(&mut self, description: String) -> Result<(), Error> {
+        let Step(index, _) = self.current_step.clone().unwrap_or_default();
+        self.set_current_step(Some(Step(index + 1, description))).await
+    }
+
+    pub async fn finish(&mut self) -> Result<(), Error> {
+        self.set_total_steps(0).await?;
+        self.set_current_step(None).await?;
+        Ok(())
+    }
+
+    async fn set_total_steps(&mut self, value: u8) -> Result<(), Error> {
         self.total_steps = value;
-        self.total_steps_changed(signal_context).await?;
+        self.total_steps_signal().await
+    }
+
+    async fn set_current_step(&mut self, step: Option<Step>) -> Result<(), Error> {
+        self.current_step = step;
+        self.current_step_signal().await
+    }
+
+    async fn total_steps_signal(&self) -> Result<(), Error> {
+        if let Some(signal_context) = self.signal_context() {
+            self.total_steps_changed(signal_context).await?;
+        }
+        Ok(())
+    }
 
+    async fn current_step_signal(&self) -> Result<(), Error> {
+        if let Some(signal_context) = self.signal_context() {
+            self.current_step_changed(signal_context).await?;
+        }
         Ok(())
     }
+
+    fn signal_context(&self) -> Option<&SignalContext> {
+        self.iface_ref
+            .as_ref()
+            .and_then(|iref| Some(iref.signal_context()))
+    }
 }
 
 #[dbus_interface(name = "org.opensuse.Agama1.Progress")]
@@ -32,25 +83,24 @@ impl Progress {
     fn total_steps(&self) -> u8 {
         self.total_steps
     }
+
+    #[dbus_interface(property)]
+    fn current_step(&self) -> (u8, String) {
+        let Step(index, description) = self.current_step.clone().unwrap_or_default();
+        (index, description)
+    }
 }
 
-pub async fn add_interface(
-    path: &str,
+pub async fn export_interface(
     connection: &Connection,
-    rx: Receiver<u8>,
-) -> Result<(), Box<dyn std::error::Error>> {
+    path: &str,
+) -> Result<InterfaceRef<Progress>, Box<dyn std::error::Error>> {
     let progress = Progress::new();
     connection.object_server().at(path, progress).await?;
 
     let iface_ref = connection.object_server().interface::<_, Progress>(path).await?;
+    let mut iface = iface_ref.get_mut().await;
+    iface.set_iface_ref(iface_ref.clone());
 
-    async_std::task::spawn(async move {
-        let mut iface = iface_ref.get_mut().await;
-
-        while let Ok(value) = rx.recv().await {
-            iface.set_total_steps(value, iface_ref.signal_context()).await.unwrap();
-        }
-    });
-
-    Ok(())
+    Ok(iface_ref.clone())
 }

rust/agama-dbus-server/src/manager.rs

@@ -1,30 +1,30 @@
-use crate::dbus::interfaces::progress::add_interface;
-use crate::progress::Progress;
-use zbus::{dbus_interface, Connection};
+use crate::dbus::interfaces::progress::{export_interface, Progress};
+use zbus::{dbus_interface, Connection, InterfaceRef};
 use crate::error::Error;
 
 pub struct Manager {
-    progress: Progress,
+    progress_ref: InterfaceRef<Progress>,
 }
 
 impl Manager {
-    pub fn new() -> Self {
+    pub fn new(progress_ref: InterfaceRef<Progress>) -> Self {
         Self {
-            progress: Progress::new(),
+            progress_ref,
         }
     }
 }
 
-// impl Default for Manager {
-//     fn default() -> Self {
-//         Self::new()
-//     }
-// }
-
 #[dbus_interface(name = "org.opensuse.Agama1.Manager")]
 impl Manager {
     pub async fn probe(&mut self) -> Result<(), Error> {
-        self.progress.start(2).await
+        let mut progress = self.progress_ref.get_mut().await;
+
+        progress.start(2).await?;
+        progress.step("step 1".to_string()).await?;
+        progress.step("step 2".to_string()).await?;
+        progress.finish().await?;
+
+        Ok(())
     }
 }
 
@@ -33,12 +33,9 @@ pub async fn export_dbus_objects(
 ) -> Result<(), Box<dyn std::error::Error>> {
     const PATH: &str = "/org/opensuse/Agama1";
 
-    // When serving, request the service name _after_ exposing the main object
-    let manager = Manager::new();
-    let rx = manager.progress.rx();
+    let progress_ref = export_interface(connection, PATH).await?;
+    let manager = Manager::new(progress_ref);
     connection.object_server().at(PATH, manager).await?;
 
-    add_interface(PATH, connection, rx).await?;
-
     Ok(())
 }