Fancy progress bar!

Author: jackTabsCode

src/sync/collect.rs

@@ -3,7 +3,7 @@ use std::{
     collections::{HashMap, HashSet},
     path::PathBuf,
 };
-use tokio::sync::mpsc::Receiver;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 use crate::{
     asset::AssetRef,
@@ -16,11 +16,11 @@ use crate::{
 pub struct CollectResults {
     pub new_lockfile: Lockfile,
     pub input_sources: HashMap<String, NodeSource>,
-    pub new_count: u32,
+    pub new_count: u64,
 }
 
 pub async fn collect_events(
-    mut rx: Receiver<super::Event>,
+    mut rx: UnboundedReceiver<super::Event>,
     target: SyncTarget,
     inputs: InputMap,
     mp: MultiProgress,
@@ -43,6 +43,16 @@ pub async fn collect_events(
 
     while let Some(event) = rx.recv().await {
         match event {
+            super::Event::Discovered(path) => {
+                if !seen_paths.contains(&path) {
+                    progress.discovered += 1;
+                }
+            }
+            super::Event::InFlight(path) => {
+                if !seen_paths.contains(&path) {
+                    progress.in_flight.insert(path.clone());
+                }
+            }
             super::Event::Finished {
                 state,
                 input_name,
@@ -81,18 +91,13 @@ pub async fn collect_events(
 
                 progress.in_flight.remove(&path);
             }
-            super::Event::InFlight(path) => {
-                if !seen_paths.contains(&path) {
-                    progress.in_flight.insert(path.clone());
-                }
-            }
             super::Event::Failed(path) => {
                 progress.failed += 1;
                 progress.in_flight.remove(&path);
             }
         }
 
-        progress.update_msg();
+        progress.update();
     }
 
     progress.finish();
@@ -105,30 +110,39 @@ pub async fn collect_events(
 }
 
 struct Progress {
-    spinner: ProgressBar,
+    inner: ProgressBar,
     target: SyncTarget,
     in_flight: HashSet<PathBuf>,
-    synced: u32,
-    new: u32,
-    dupes: u32,
-    failed: u32,
+    discovered: u64,
+    synced: u64,
+    new: u64,
+    dupes: u64,
+    failed: u64,
 }
 
 impl Progress {
+    fn get_style(finished: bool) -> ProgressStyle {
+        ProgressStyle::default_bar()
+            .template(&format!(
+                "{{prefix:.{prefix_color}.bold}}{bar} {{pos}}/{{len}} assets: ({{msg}})",
+                prefix_color = if finished { "green" } else { "cyan" },
+                bar = if finished { "" } else { " [{bar:40}]" },
+            ))
+            .unwrap()
+            .progress_chars("=> ")
+    }
+
     fn new(mp: MultiProgress, target: SyncTarget) -> Self {
         let spinner = mp.add(ProgressBar::new_spinner());
-        spinner.set_style(
-            ProgressStyle::default_spinner()
-                .template("{spinner:.cyan} {msg}")
-                .unwrap(),
-        );
+        spinner.set_style(Progress::get_style(false));
+        spinner.set_prefix("Syncing");
         spinner.enable_steady_tick(std::time::Duration::from_millis(100));
-        spinner.set_message("Starting sync...");
 
         Self {
-            spinner,
+            inner: spinner,
             target,
             in_flight: HashSet::new(),
+            discovered: 0,
             synced: 0,
             new: 0,
             dupes: 0,
@@ -137,16 +151,13 @@ impl Progress {
     }
 
     fn get_msg(&self) -> String {
-        let mut str = format!("Synced {} files", self.synced + self.dupes);
-
         let mut parts = Vec::new();
 
         if self.new > 0 {
             let target_msg = match self.target {
                 SyncTarget::Cloud { dry_run: true } => "checked",
                 SyncTarget::Cloud { dry_run: false } => "uploaded",
-                SyncTarget::Studio => "written to content folder",
-                SyncTarget::Debug => "written to debug folder",
+                SyncTarget::Studio | SyncTarget::Debug => "written",
             };
             parts.push(format!("{} {}", self.new, target_msg));
         }
@@ -160,29 +171,26 @@ impl Progress {
 
         let in_flight = self.in_flight.len();
         if in_flight > 0 {
-            parts.push(format!("{} in-flight", in_flight));
+            parts.push(format!("{} processing", in_flight));
         }
 
         let failed = self.failed;
         if failed > 0 {
             parts.push(format!("{} failed", failed));
         }
 
-        if parts.is_empty() {
-            return str;
-        }
-
-        str.push_str(" (");
-        str.push_str(&parts.join(", "));
-        str.push(')');
-        str
+        parts.join(", ")
     }
 
-    fn update_msg(&self) {
-        self.spinner.set_message(self.get_msg());
+    fn update(&self) {
+        self.inner.set_position(self.synced + self.dupes);
+        self.inner.set_length(self.discovered);
+        self.inner.set_message(self.get_msg());
     }
 
     fn finish(&self) {
-        self.spinner.finish_with_message(self.get_msg());
+        self.inner.set_prefix("Synced");
+        self.inner.set_style(Progress::get_style(true));
+        self.inner.finish();
     }
 }

src/sync/mod.rs

@@ -41,6 +41,8 @@ impl TargetBackend {
 
 #[derive(Debug)]
 enum Event {
+    Discovered(PathBuf),
+    InFlight(PathBuf),
     Finished {
         state: EventState,
         input_name: String,
@@ -49,7 +51,6 @@ enum Event {
         hash: String,
         asset_ref: Option<AssetRef>,
     },
-    InFlight(PathBuf),
     Failed(PathBuf),
 }
 
@@ -71,7 +72,7 @@ pub async fn sync(args: SyncArgs, mp: MultiProgress) -> anyhow::Result<()> {
         db
     });
 
-    let (event_tx, event_rx) = mpsc::channel::<Event>(100);
+    let (event_tx, event_rx) = mpsc::unbounded_channel::<Event>();
 
     let collector_handle = tokio::spawn({
         let inputs = config.inputs.clone();

src/sync/walk.rs

@@ -19,7 +19,7 @@ use std::{
     sync::Arc,
 };
 use tokio::{
-    sync::{Mutex, Semaphore, mpsc::Sender},
+    sync::{Mutex, Semaphore, mpsc::UnboundedSender},
     task::JoinSet,
 };
 use walkdir::WalkDir;
@@ -39,7 +39,7 @@ struct InputState {
     bleed: bool,
 }
 
-pub async fn walk(params: Params, config: &Config, tx: &Sender<super::Event>) {
+pub async fn walk(params: Params, config: &Config, tx: &UnboundedSender<super::Event>) {
     let params = Arc::new(params);
 
     for (input_name, input) in &config.inputs {
@@ -78,14 +78,16 @@ pub async fn walk(params: Params, config: &Config, tx: &Sender<super::Event>) {
             let semaphore = semaphore.clone();
             let tx = tx.clone();
 
+            tx.send(super::Event::Discovered(path.clone())).unwrap();
+
             join_set.spawn(async move {
                 let _permit = semaphore.acquire_owned().await.unwrap();
 
-                tx.send(super::Event::InFlight(path.clone())).await.unwrap();
+                tx.send(super::Event::InFlight(path.clone())).unwrap();
 
                 if let Err(e) = process_entry(state.clone(), &path, &tx).await {
                     warn!("Failed to process file {}: {e:?}", path.display());
-                    tx.send(super::Event::Failed(path.clone())).await.unwrap();
+                    tx.send(super::Event::Failed(path.clone())).unwrap();
                 }
             });
         }
@@ -97,7 +99,7 @@ pub async fn walk(params: Params, config: &Config, tx: &Sender<super::Event>) {
 async fn process_entry(
     state: Arc<InputState>,
     path: &Path,
-    tx: &Sender<super::Event>,
+    tx: &UnboundedSender<super::Event>,
 ) -> anyhow::Result<()> {
     debug!("Handling entry: {}", path.display());
 
@@ -137,7 +139,7 @@ async fn process_entry(
                     asset_ref: lockfile_entry.map(Into::into),
                     hash: asset.hash.clone(),
                 };
-                tx.send(event).await.unwrap();
+                tx.send(event).unwrap();
 
                 return Ok(());
             }
@@ -163,7 +165,7 @@ async fn process_entry(
         hash: asset.hash.clone(),
         asset_ref,
     };
-    tx.send(event).await.unwrap();
+    tx.send(event).unwrap();
 
     Ok(())
 }