moss: Add percentage callback for global download progress

By using an atomic total bytes to download, we can get a nice smooth
progress bar of all packages as they download despite being async.

This doesn't account for unpacking or db storing yet. We may also wish
to spit the individual progress reporting in the future as well.

Author: joebonrichie

moss/src/cli/sync.rs

@@ -118,7 +118,7 @@ pub fn handle(args: &ArgMatches, installation: Installation, debug: bool) -> Res
         return Err(Error::Cancelled);
     }
 
-    runtime::block_on(client.cache_packages(&synced))?;
+    runtime::block_on(client.cache_packages(&synced, None))?;
 
     // Map finalized state to a [`Selection`] by referencing
     // it's value from the previous state

moss/src/client/install.rs

@@ -96,7 +96,7 @@ pub fn install(client: &mut Client, pkgs: &[&str], yes: bool) -> Result<Timing,
     instant = Instant::now();
 
     // Cache packages
-    runtime::block_on(client.cache_packages(&missing))?;
+    runtime::block_on(client.cache_packages(&missing, None))?;
 
     timing.fetch = instant.elapsed();
     instant = Instant::now();

moss/src/client/mod.rs

@@ -14,7 +14,10 @@ use std::{
     fmt, io,
     os::{fd::RawFd, unix::fs::symlink},
     path::{Path, PathBuf},
-    sync::Arc,
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
     time::{Duration, Instant},
 };
 
@@ -86,6 +89,11 @@ pub enum ProgressStage {
     System = 2,
 }
 
+#[derive(Clone, Copy, Debug)]
+pub enum DownloadCallback {
+    DownloadOverall(f32),
+}
+
 impl Client {
     /// Construct a new Client for the given [`Installation`]
     pub fn new(client_name: impl ToString, installation: Installation) -> Result<Client, Error> {
@@ -490,7 +498,11 @@ impl Client {
     }
 
     /// Download & unpack the provided packages. Packages already cached will be validated & skipped.
-    pub async fn cache_packages<T>(&self, packages: &[T]) -> Result<(), Error>
+    pub async fn cache_packages<T>(
+        &self,
+        packages: &[T],
+        progress_callback: Option<Arc<dyn Fn(DownloadCallback) + Send + Sync>>,
+    ) -> Result<(), Error>
     where
         T: Borrow<Package>,
     {
@@ -509,6 +521,12 @@ impl Client {
 
         let unpacking_in_progress = cache::UnpackingInProgress::default();
 
+        let total_bytes = packages.iter().fold(0, |acc, p| {
+            let pkg: &Package = p.borrow();
+            acc + pkg.meta.download_size.unwrap_or_default()
+        });
+        let total_completed_bytes = &AtomicU64::new(0);
+
         // Download and unpack each package
         let cached = stream::iter(packages)
             .map(|package| async {
@@ -536,12 +554,24 @@ impl Client {
                 // Download and update progress
                 let download = cache::fetch(&package.meta, &self.installation, |progress| {
                     progress_bar.inc(progress.delta);
+                    if let Some(ref callback) = progress_callback {
+                        callback({
+                            DownloadCallback::DownloadOverall({
+                                let current_pkg_completed = progress.completed;
+                                let total_completed =
+                                    total_completed_bytes.load(Ordering::Relaxed) + current_pkg_completed;
+                                total_completed as f32 / total_bytes as f32
+                            })
+                        })
+                    }
                 })
                 .await?;
                 let is_cached = download.was_cached;
 
                 // Move rest of blocking code to threadpool
 
+                total_completed_bytes.fetch_add(package.meta.download_size.unwrap_or_default(), Ordering::Relaxed);
+
                 let multi_progress = multi_progress.clone();
                 let total_progress = total_progress.clone();
                 let unpacking_in_progress = unpacking_in_progress.clone();

moss/src/client/verify.rs

@@ -214,7 +214,7 @@ pub fn verify(client: &Client, yes: bool, verbose: bool) -> Result<(), client::E
         println!("Reinstalling packages");
 
         // And re-cache all packages that comprise the corrupt / missing asset
-        runtime::block_on(client.cache_packages(&issue_packages))?;
+        runtime::block_on(client.cache_packages(&issue_packages, None))?;
     }
 
     // Now we must fix any states that referenced these packages