feat: non-blocking and centralized progress reporting

rust/agama-cli/src/progress.rs

@@ -26,8 +26,8 @@ use console::style;
 use indicatif::{ProgressBar, ProgressStyle};
 use std::time::Duration;
 
-const MANAGER_SERVICE: &str = "org.opensuse.Agama.Manager1";
-const SOFTWARE_SERVICE: &str = "org.opensuse.Agama.Software1";
+const MANAGER_PROGRESS_OBJECT_PATH: &str = "/org/opensuse/Agama/Manager1";
+const SOFTWARE_PROGRESS_OBJECT_PATH: &str = "/org/opensuse/Agama/Software1";
 
 /// Displays the progress on the terminal.
 pub struct ProgressMonitor {
@@ -77,22 +77,22 @@ impl ProgressMonitor {
     ///
     /// It returns true if the monitor should continue.
     async fn update(&mut self, status: MonitorStatus) -> bool {
-        if status.progress.get(MANAGER_SERVICE).is_none() && self.running {
+        if status.progress.get(MANAGER_PROGRESS_OBJECT_PATH).is_none() && self.running {
             self.finish();
             if self.stop_on_idle {
                 return false;
             }
         }
 
-        if let Some(progress) = status.progress.get(MANAGER_SERVICE) {
+        if let Some(progress) = status.progress.get(MANAGER_PROGRESS_OBJECT_PATH) {
             self.running = true;
             if self.current_step != progress.current_step {
                 self.update_main(&progress).await;
                 self.current_step = progress.current_step;
             }
         }
 
-        match status.progress.get(SOFTWARE_SERVICE) {
+        match status.progress.get(SOFTWARE_PROGRESS_OBJECT_PATH) {
             Some(progress) => self.update_bar(progress),
             None => self.remove_bar(),
         }

rust/agama-lib/src/http/event.rs

@@ -49,8 +49,8 @@ pub enum Event {
     DevicesDirty {
         dirty: bool,
     },
-    Progress {
-        service: String,
+    ProgressChanged {
+        path: String,
         #[serde(flatten)]
         progress: Progress,
     },

rust/agama-lib/src/monitor.rs

@@ -60,8 +60,8 @@ use crate::{
     progress::Progress,
 };
 
-const MANAGER_SERVICE: &str = "org.opensuse.Agama.Manager1";
-// const SOFTWARE_SERVICE: &str = "org.opensuse.Agama.Software1";
+const MANAGER_PROGRESS_OBJECT_PATH: &str = "/org/opensuse/Agama/Manager1";
+const SOFTWARE_PROGRESS_OBJECT_PATH: &str = "/org/opensuse/Agama/Software1";
 
 #[derive(thiserror::Error, Debug)]
 pub enum MonitorError {
@@ -82,7 +82,7 @@ pub struct MonitorStatus {
     ///
     /// FIXME: do not hold the full status (some elements are not updated)
     pub installer_status: InstallerStatus,
-    /// Progress for each service using the name as the key. If the progress is
+    /// Progress for each service using the D-Bus object path as the key. If the progress is
     /// finished, the entry is removed from the map.
     pub progress: HashMap<String, Progress>,
 }
@@ -92,13 +92,13 @@ impl MonitorStatus {
     ///
     /// The entry is removed if the progress is finished.
     ///
-    /// * `service`: service name.
+    /// * `service`: D-Bus object path.
     /// * `progress`: updated progress.
-    fn update_progress(&mut self, service: String, progress: Progress) {
+    fn update_progress(&mut self, path: String, progress: Progress) {
         if progress.finished {
-            _ = self.progress.remove_entry(&service);
+            _ = self.progress.remove_entry(&path);
         } else {
-            _ = self.progress.insert(service, progress);
+            _ = self.progress.insert(path, progress);
         }
     }
 
@@ -224,11 +224,11 @@ impl Monitor {
     /// * `event`: Agama event.
     fn handle_event(&mut self, event: Event) {
         match event {
-            Event::Progress { service, progress } => {
-                self.status.update_progress(service, progress);
+            Event::ProgressChanged { path, progress } => {
+                self.status.update_progress(path, progress);
             }
             Event::ServiceStatusChanged { service, status } => {
-                if service.as_str() == MANAGER_SERVICE {
+                if service.as_str() == MANAGER_PROGRESS_OBJECT_PATH {
                     self.status.set_is_busy(status == 1);
                 }
             }
@@ -258,26 +258,32 @@ impl MonitorStatusReader {
             ..Default::default()
         };
 
-        self.add_service_progress(&mut status, MANAGER_SERVICE, "/manager/progress")
-            .await?;
-        // FIXME: do not read the software status yet because it might block
-        // the progress. Enable this line when the software service does not block
-        // self.add_service_progress(&mut status, SOFTWARE_SERVICE, "/software/progress")
-        //     .await?;
+        self.add_service_progress(
+            &mut status,
+            MANAGER_PROGRESS_OBJECT_PATH,
+            "/manager/progress",
+        )
+        .await?;
+        self.add_service_progress(
+            &mut status,
+            SOFTWARE_PROGRESS_OBJECT_PATH,
+            "/software/progress",
+        )
+        .await?;
         Ok(status)
     }
 
     async fn add_service_progress(
         &self,
         status: &mut MonitorStatus,
-        service: &str,
+        dbus_path: &str,
         path: &str,
     ) -> Result<(), MonitorError> {
         let progress: Progress = self.http.get(path).await?;
         if progress.finished {
             return Ok(());
         }
-        status.progress.insert(service.to_string(), progress);
+        status.progress.insert(dbus_path.to_string(), progress);
         Ok(())
     }
 }

rust/agama-lib/src/progress.rs

@@ -63,3 +63,13 @@ impl Progress {
         })
     }
 }
+
+/// Information about the current progress sequence.
+#[derive(Clone, Debug, Default, Serialize, utoipa::ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct ProgressSequence {
+    /// Sequence steps if known in advance
+    pub steps: Vec<String>,
+    #[serde(flatten)]
+    pub progress: Progress,
+}

rust/agama-lib/src/proxies.rs

@@ -19,7 +19,7 @@
 // find current contact information at www.suse.com.
 
 mod progress;
-pub use progress::ProgressProxy;
+pub use progress::{ProgressChanged, ProgressChangedArgs, ProgressChangedStream, ProgressProxy};
 
 mod service_status;
 pub use service_status::ServiceStatusProxy;

rust/agama-lib/src/proxies/progress.rs

@@ -22,6 +22,16 @@
 use zbus::proxy;
 #[proxy(interface = "org.opensuse.Agama1.Progress", assume_defaults = true)]
 pub trait Progress {
+    /// ProgressChanged signal
+    #[zbus(signal)]
+    fn progress_changed(
+        &self,
+        total_steps: u32,
+        current_step: (u32, &str),
+        finished: bool,
+        steps: Vec<&str>,
+    ) -> zbus::Result<()>;
+
     /// CurrentStep property
     #[zbus(property)]
     fn current_step(&self) -> zbus::Result<(u32, String)>;

rust/agama-server/src/error.rs

@@ -26,7 +26,11 @@ use axum::{
 };
 use serde_json::json;
 
-use crate::{l10n::LocaleError, questions::QuestionsError, web::common::IssuesServiceError};
+use crate::{
+    l10n::LocaleError,
+    questions::QuestionsError,
+    web::common::{IssuesServiceError, ProgressServiceError},
+};
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -42,6 +46,8 @@ pub enum Error {
     Locale(#[from] LocaleError),
     #[error("Issues service error: {0}")]
     Issues(#[from] IssuesServiceError),
+    #[error("Progress service error: {0}")]
+    Progress(#[from] ProgressServiceError),
 }
 
 // This would be nice, but using it for a return type

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

@@ -31,6 +31,7 @@ use agama_lib::{
     manager::{FinishMethod, InstallationPhase, InstallerStatus, ManagerClient},
     proxies::Manager1Proxy,
 };
+use anyhow::Context;
 use axum::{
     body::Body,
     extract::State,
@@ -45,7 +46,7 @@ use tokio_util::io::ReaderStream;
 
 use crate::{
     error::Error,
-    web::common::{progress_router, service_status_router},
+    web::common::{service_status_router, ProgressClient, ProgressRouterBuilder},
 };
 use agama_lib::http::Event;
 
@@ -85,12 +86,19 @@ pub async fn manager_stream(
 }
 
 /// Sets up and returns the axum service for the manager module
-pub async fn manager_service(dbus: zbus::Connection) -> Result<Router, ServiceError> {
+pub async fn manager_service(
+    dbus: zbus::Connection,
+    progress: ProgressClient,
+) -> Result<Router, ServiceError> {
     const DBUS_SERVICE: &str = "org.opensuse.Agama.Manager1";
     const DBUS_PATH: &str = "/org/opensuse/Agama/Manager1";
 
     let status_router = service_status_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
-    let progress_router = progress_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
+    // FIXME: use anyhow temporarily until we adapt all these methods to return
+    // the crate::error::Error instead of ServiceError.
+    let progress_router = ProgressRouterBuilder::new(DBUS_SERVICE, DBUS_PATH, progress)
+        .build()
+        .context("Could not build the progress router")?;
     let manager = ManagerClient::new(dbus.clone()).await?;
     let state = ManagerState { manager, dbus };
     Ok(Router::new()

rust/agama-server/src/software/web.rs

@@ -29,7 +29,8 @@ use crate::{
     error::Error,
     web::{
         common::{
-            progress_router, service_status_router, EventStreams, IssuesClient, IssuesRouterBuilder,
+            service_status_router, EventStreams, IssuesClient, IssuesRouterBuilder, ProgressClient,
+            ProgressRouterBuilder,
         },
         EventsReceiver,
     },
@@ -238,13 +239,13 @@ pub async fn software_service(
     dbus: zbus::Connection,
     events: EventsReceiver,
     issues: IssuesClient,
+    progress: ProgressClient,
 ) -> Result<Router, ServiceError> {
     const DBUS_SERVICE: &str = "org.opensuse.Agama.Software1";
     const DBUS_PATH: &str = "/org/opensuse/Agama/Software1";
     const DBUS_PRODUCT_PATH: &str = "/org/opensuse/Agama/Software1/Product";
 
     let status_router = service_status_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
-    let progress_router = progress_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
 
     // FIXME: use anyhow temporarily until we adapt all these methods to return
     // the crate::error::Error instead of ServiceError.
@@ -254,6 +255,9 @@ pub async fn software_service(
     let product_issues = IssuesRouterBuilder::new(DBUS_SERVICE, DBUS_PRODUCT_PATH, issues)
         .build()
         .context("Could not build an issues router")?;
+    let progress_router = ProgressRouterBuilder::new(DBUS_SERVICE, DBUS_PATH, progress)
+        .build()
+        .context("Could not build the progress router")?;
 
     let mut licenses_repo = LicensesRepo::default();
     if let Err(error) = licenses_repo.read() {

rust/agama-server/src/storage/web.rs

@@ -56,8 +56,8 @@ use crate::{
         iscsi::iscsi_stream,
     },
     web::common::{
-        jobs_service, progress_router, service_status_router, EventStreams, IssuesClient,
-        IssuesRouterBuilder,
+        jobs_service, service_status_router, EventStreams, IssuesClient, IssuesRouterBuilder,
+        ProgressClient, ProgressRouterBuilder,
     },
 };
 use agama_lib::http::Event;
@@ -101,16 +101,21 @@ struct StorageState<'a> {
 pub async fn storage_service(
     dbus: zbus::Connection,
     issues: IssuesClient,
+    progress: ProgressClient,
 ) -> Result<Router, ServiceError> {
     const DBUS_SERVICE: &str = "org.opensuse.Agama.Storage1";
     const DBUS_PATH: &str = "/org/opensuse/Agama/Storage1";
     const DBUS_DESTINATION: &str = "org.opensuse.Agama.Storage1";
 
     let status_router = service_status_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
-    let progress_router = progress_router(&dbus, DBUS_SERVICE, DBUS_PATH).await?;
+    // FIXME: use anyhow temporarily until we adapt all these methods to return
+    // the crate::error::Error instead of ServiceError.
     let issues_router = IssuesRouterBuilder::new(DBUS_SERVICE, DBUS_PATH, issues.clone())
         .build()
         .context("Could not build an issues router")?;
+    let progress_router = ProgressRouterBuilder::new(DBUS_SERVICE, DBUS_PATH, progress)
+        .build()
+        .context("Could not build the progress router")?;
     let iscsi_router = storage_iscsi_service(&dbus).await?;
     let dasd_router = dasd_service(&dbus).await?;
     let zfcp_router = zfcp_service(&dbus).await?;