Rename DownloadProgress to DownloadProgressEvent and BlobProgress to BlobProgressEvent

Author: rklaehn

examples/hello-world-fetch.rs

@@ -66,13 +66,13 @@ async fn main() -> Result<()> {
         "'Hello World' example expects to fetch a single blob, but the ticket indicates a collection.",
     );
 
-    // `download` returns a stream of `DownloadProgress` events. You can iterate through these updates to get progress
+    // `download` returns a stream of `DownloadProgressEvent`. You can iterate through these updates to get progress
     // on the state of your download.
     let download_stream = blobs_client
         .download(ticket.hash(), ticket.node_addr().clone())
         .await?;
 
-    // You can also just `await` the stream, which will poll the `DownloadProgress` stream for you.
+    // You can also just `await` the stream, which will poll the `DownloadProgressEvent` stream for you.
     let outcome = download_stream.await.context("unable to download hash")?;
 
     println!(

examples/local-swarm-discovery.rs

@@ -140,16 +140,14 @@ mod progress {
         ProgressStyle,
     };
     use iroh_blobs::{
-        get::{
-            progress::{BlobProgress, DownloadProgress},
-            Stats,
-        },
+        get::Stats,
+        rpc::client::blobs::{BlobProgressEvent, DownloadProgressEvent},
         Hash,
     };
 
     pub async fn show_download_progress(
         hash: Hash,
-        mut stream: impl Stream<Item = Result<DownloadProgress>> + Unpin,
+        mut stream: impl Stream<Item = Result<DownloadProgressEvent>> + Unpin,
     ) -> Result<()> {
         eprintln!("Fetching: {}", hash);
         let mp = MultiProgress::new();
@@ -160,7 +158,7 @@ mod progress {
         let mut seq = false;
         while let Some(x) = stream.next().await {
             match x? {
-                DownloadProgress::InitialState(state) => {
+                DownloadProgressEvent::InitialState(state) => {
                     if state.connected {
                         op.set_message(format!("{} Requesting ...\n", style("[2/3]").bold().dim()));
                     }
@@ -180,21 +178,21 @@ mod progress {
                             ip.set_length(size.value());
                             ip.reset();
                             match blob.progress {
-                                BlobProgress::Pending => {}
-                                BlobProgress::Progressing(offset) => ip.set_position(offset),
-                                BlobProgress::Done => ip.finish_and_clear(),
+                                BlobProgressEvent::Pending => {}
+                                BlobProgressEvent::Progressing(offset) => ip.set_position(offset),
+                                BlobProgressEvent::Done => ip.finish_and_clear(),
                             }
                             if !seq {
                                 op.finish_and_clear();
                             }
                         }
                     }
                 }
-                DownloadProgress::FoundLocal { .. } => {}
-                DownloadProgress::Connected => {
+                DownloadProgressEvent::FoundLocal { .. } => {}
+                DownloadProgressEvent::Connected => {
                     op.set_message(format!("{} Requesting ...\n", style("[2/3]").bold().dim()));
                 }
-                DownloadProgress::FoundHashSeq { children, .. } => {
+                DownloadProgressEvent::FoundHashSeq { children, .. } => {
                     op.set_message(format!(
                         "{} Downloading {} blob(s)\n",
                         style("[3/3]").bold().dim(),
@@ -204,7 +202,7 @@ mod progress {
                     op.reset();
                     seq = true;
                 }
-                DownloadProgress::Found { size, child, .. } => {
+                DownloadProgressEvent::Found { size, child, .. } => {
                     if seq {
                         op.set_position(child.into());
                     } else {
@@ -213,13 +211,13 @@ mod progress {
                     ip.set_length(size);
                     ip.reset();
                 }
-                DownloadProgress::Progress { offset, .. } => {
+                DownloadProgressEvent::Progress { offset, .. } => {
                     ip.set_position(offset);
                 }
-                DownloadProgress::Done { .. } => {
+                DownloadProgressEvent::Done { .. } => {
                     ip.finish_and_clear();
                 }
-                DownloadProgress::AllDone(Stats {
+                DownloadProgressEvent::AllDone(Stats {
                     bytes_read,
                     elapsed,
                     ..
@@ -233,7 +231,7 @@ mod progress {
                     );
                     break;
                 }
-                DownloadProgress::Abort(e) => {
+                DownloadProgressEvent::Abort(e) => {
                     bail!("download aborted: {}", e);
                 }
             }

src/cli.rs

@@ -19,14 +19,10 @@ use iroh::{NodeAddr, PublicKey, RelayUrl};
 use tokio::io::AsyncWriteExt;
 
 use crate::{
-    get::{
-        progress::{BlobProgress, DownloadProgress},
-        Stats,
-    },
-    provider::AddProgress,
+    get::Stats,
     rpc::client::blobs::{
-        self, BlobInfo, BlobStatus, CollectionInfo, DownloadMode, DownloadOptions,
-        IncompleteBlobInfo, WrapOption,
+        self, AddProgressEvent, BlobInfo, BlobProgressEvent, BlobStatus, CollectionInfo,
+        DownloadMode, DownloadOptions, DownloadProgressEvent, IncompleteBlobInfo, WrapOption,
     },
     store::{ConsistencyCheckProgress, ExportFormat, ExportMode, ReportLevel, ValidateProgress},
     ticket::BlobTicket,
@@ -898,29 +894,29 @@ pub struct ProvideResponseEntry {
     pub hash: Hash,
 }
 
-/// Combines the [`AddProgress`] outputs from a [`Stream`] into a single tuple.
+/// Combines the [`AddProgressEvent`] outputs from a [`Stream`] into a single tuple.
 pub async fn aggregate_add_response(
-    mut stream: impl Stream<Item = Result<AddProgress>> + Unpin,
+    mut stream: impl Stream<Item = Result<AddProgressEvent>> + Unpin,
 ) -> Result<(Hash, BlobFormat, Vec<ProvideResponseEntry>)> {
     let mut hash_and_format = None;
     let mut collections = BTreeMap::<u64, (String, u64, Option<Hash>)>::new();
     let mut mp = Some(ProvideProgressState::new());
     while let Some(item) = stream.next().await {
         match item? {
-            AddProgress::Found { name, id, size } => {
+            AddProgressEvent::Found { name, id, size } => {
                 tracing::trace!("Found({id},{name},{size})");
                 if let Some(mp) = mp.as_mut() {
                     mp.found(name.clone(), id, size);
                 }
                 collections.insert(id, (name, size, None));
             }
-            AddProgress::Progress { id, offset } => {
+            AddProgressEvent::Progress { id, offset } => {
                 tracing::trace!("Progress({id}, {offset})");
                 if let Some(mp) = mp.as_mut() {
                     mp.progress(id, offset);
                 }
             }
-            AddProgress::Done { hash, id } => {
+            AddProgressEvent::Done { hash, id } => {
                 tracing::trace!("Done({id},{hash:?})");
                 if let Some(mp) = mp.as_mut() {
                     mp.done(id, hash);
@@ -934,15 +930,15 @@ pub async fn aggregate_add_response(
                     }
                 }
             }
-            AddProgress::AllDone { hash, format, .. } => {
+            AddProgressEvent::AllDone { hash, format, .. } => {
                 tracing::trace!("AllDone({hash:?})");
                 if let Some(mp) = mp.take() {
                     mp.all_done();
                 }
                 hash_and_format = Some(HashAndFormat { hash, format });
                 break;
             }
-            AddProgress::Abort(e) => {
+            AddProgressEvent::Abort(e) => {
                 if let Some(mp) = mp.take() {
                     mp.error();
                 }
@@ -1035,7 +1031,7 @@ impl ProvideProgressState {
 /// Displays the download progress for a given stream.
 pub async fn show_download_progress(
     hash: Hash,
-    mut stream: impl Stream<Item = Result<DownloadProgress>> + Unpin,
+    mut stream: impl Stream<Item = Result<DownloadProgressEvent>> + Unpin,
 ) -> Result<()> {
     eprintln!("Fetching: {}", hash);
     let mp = MultiProgress::new();
@@ -1046,7 +1042,7 @@ pub async fn show_download_progress(
     let mut seq = false;
     while let Some(x) = stream.next().await {
         match x? {
-            DownloadProgress::InitialState(state) => {
+            DownloadProgressEvent::InitialState(state) => {
                 if state.connected {
                     op.set_message(format!("{} Requesting ...\n", style("[2/3]").bold().dim()));
                 }
@@ -1066,21 +1062,21 @@ pub async fn show_download_progress(
                         ip.set_length(size.value());
                         ip.reset();
                         match blob.progress {
-                            BlobProgress::Pending => {}
-                            BlobProgress::Progressing(offset) => ip.set_position(offset),
-                            BlobProgress::Done => ip.finish_and_clear(),
+                            BlobProgressEvent::Pending => {}
+                            BlobProgressEvent::Progressing(offset) => ip.set_position(offset),
+                            BlobProgressEvent::Done => ip.finish_and_clear(),
                         }
                         if !seq {
                             op.finish_and_clear();
                         }
                     }
                 }
             }
-            DownloadProgress::FoundLocal { .. } => {}
-            DownloadProgress::Connected => {
+            DownloadProgressEvent::FoundLocal { .. } => {}
+            DownloadProgressEvent::Connected => {
                 op.set_message(format!("{} Requesting ...\n", style("[2/3]").bold().dim()));
             }
-            DownloadProgress::FoundHashSeq { children, .. } => {
+            DownloadProgressEvent::FoundHashSeq { children, .. } => {
                 op.set_message(format!(
                     "{} Downloading {} blob(s)\n",
                     style("[3/3]").bold().dim(),
@@ -1090,7 +1086,7 @@ pub async fn show_download_progress(
                 op.reset();
                 seq = true;
             }
-            DownloadProgress::Found { size, child, .. } => {
+            DownloadProgressEvent::Found { size, child, .. } => {
                 if seq {
                     op.set_position(child.into());
                 } else {
@@ -1099,13 +1095,13 @@ pub async fn show_download_progress(
                 ip.set_length(size);
                 ip.reset();
             }
-            DownloadProgress::Progress { offset, .. } => {
+            DownloadProgressEvent::Progress { offset, .. } => {
                 ip.set_position(offset);
             }
-            DownloadProgress::Done { .. } => {
+            DownloadProgressEvent::Done { .. } => {
                 ip.finish_and_clear();
             }
-            DownloadProgress::AllDone(Stats {
+            DownloadProgressEvent::AllDone(Stats {
                 bytes_read,
                 elapsed,
                 ..
@@ -1119,7 +1115,7 @@ pub async fn show_download_progress(
                 );
                 break;
             }
-            DownloadProgress::Abort(e) => {
+            DownloadProgressEvent::Abort(e) => {
                 bail!("download aborted: {}", e);
             }
         }

src/downloader.rs

@@ -55,7 +55,7 @@ use tokio_util::{either::Either, sync::CancellationToken, time::delay_queue};
 use tracing::{debug, error, error_span, trace, warn, Instrument};
 
 use crate::{
-    get::{progress::DownloadProgress, Stats},
+    get::{progress::DownloadProgressEvent, Stats},
     metrics::Metrics,
     store::Store,
     util::{local_pool::LocalPoolHandle, progress::ProgressSender},
@@ -797,7 +797,7 @@ impl<G: Getter<Connection = D::Connection>, D: DialerT> Service<G, D> {
             if let Some(sender) = handlers.on_progress {
                 self.progress_tracker.unsubscribe(&kind, &sender);
                 sender
-                    .send(DownloadProgress::Abort(serde_error::Error::new(
+                    .send(DownloadProgressEvent::Abort(serde_error::Error::new(
                         &*anyhow::Error::from(DownloadError::Cancelled),
                     )))
                     .await

src/downloader/progress.rs

@@ -11,12 +11,12 @@ use parking_lot::Mutex;
 
 use super::DownloadKind;
 use crate::{
-    get::progress::{DownloadProgress, TransferState},
+    get::progress::{DownloadProgressEvent, TransferState},
     util::progress::{AsyncChannelProgressSender, IdGenerator, ProgressSendError, ProgressSender},
 };
 
 /// The channel that can be used to subscribe to progress updates.
-pub type ProgressSubscriber = AsyncChannelProgressSender<DownloadProgress>;
+pub type ProgressSubscriber = AsyncChannelProgressSender<DownloadProgressEvent>;
 
 /// Track the progress of downloads.
 ///
@@ -101,8 +101,8 @@ struct Inner {
 }
 
 impl Inner {
-    fn subscribe(&mut self, subscriber: ProgressSubscriber) -> DownloadProgress {
-        let msg = DownloadProgress::InitialState(self.state.clone());
+    fn subscribe(&mut self, subscriber: ProgressSubscriber) -> DownloadProgressEvent {
+        let msg = DownloadProgressEvent::InitialState(self.state.clone());
         self.subscribers.push(subscriber);
         msg
     }
@@ -111,7 +111,7 @@ impl Inner {
         self.subscribers.retain(|s| !s.same_channel(sender));
     }
 
-    fn on_progress(&mut self, progress: DownloadProgress) {
+    fn on_progress(&mut self, progress: DownloadProgressEvent) {
         self.state.on_progress(progress);
     }
 }
@@ -129,7 +129,7 @@ impl IdGenerator for BroadcastProgressSender {
 }
 
 impl ProgressSender for BroadcastProgressSender {
-    type Msg = DownloadProgress;
+    type Msg = DownloadProgressEvent;
 
     async fn send(&self, msg: Self::Msg) -> Result<(), ProgressSendError> {
         // making sure that the lock is not held across an await point.

src/downloader/test.rs

@@ -10,7 +10,7 @@ use iroh::SecretKey;
 
 use super::*;
 use crate::{
-    get::progress::{BlobId, BlobProgress, TransferState},
+    get::progress::{BlobId, BlobProgressEvent, DownloadProgressEvent, TransferState},
     util::{
         local_pool::LocalPool,
         progress::{AsyncChannelProgressSender, IdGenerator},
@@ -255,7 +255,7 @@ async fn concurrent_progress() {
             start_rx.await.unwrap();
             let id = progress.new_id();
             progress
-                .send(DownloadProgress::Found {
+                .send(DownloadProgressEvent::Found {
                     id,
                     child: BlobId::Root,
                     hash,
@@ -264,7 +264,10 @@ async fn concurrent_progress() {
                 .await
                 .unwrap();
             done_rx.await.unwrap();
-            progress.send(DownloadProgress::Done { id }).await.unwrap();
+            progress
+                .send(DownloadProgressEvent::Done { id })
+                .await
+                .unwrap();
             Ok(Stats::default())
         }
         .boxed()
@@ -293,18 +296,18 @@ async fn concurrent_progress() {
     let prog0_b = prog_b_rx.recv().await.unwrap();
     assert!(matches!(
         prog0_b,
-        DownloadProgress::InitialState(state) if state.root.hash == hash && state.root.progress == BlobProgress::Pending,
+        DownloadProgressEvent::InitialState(state) if state.root.hash == hash && state.root.progress == BlobProgressEvent::Pending,
     ));
 
     start_tx.send(()).unwrap();
 
     let prog1_a = prog_a_rx.recv().await.unwrap();
     let prog1_b = prog_b_rx.recv().await.unwrap();
     assert!(
-        matches!(prog1_a, DownloadProgress::Found { hash: found_hash, size: 100, ..} if found_hash == hash)
+        matches!(prog1_a, DownloadProgressEvent::Found { hash: found_hash, size: 100, ..} if found_hash == hash)
     );
     assert!(
-        matches!(prog1_b, DownloadProgress::Found { hash: found_hash, size: 100, ..} if found_hash == hash)
+        matches!(prog1_b, DownloadProgressEvent::Found { hash: found_hash, size: 100, ..} if found_hash == hash)
     );
 
     state_a.on_progress(prog1_a);
@@ -317,7 +320,7 @@ async fn concurrent_progress() {
     let handle_c = downloader.queue(req).await;
 
     let prog1_c = prog_c_rx.recv().await.unwrap();
-    assert!(matches!(&prog1_c, DownloadProgress::InitialState(state) if state == &state_a));
+    assert!(matches!(&prog1_c, DownloadProgressEvent::InitialState(state) if state == &state_a));
     state_c.on_progress(prog1_c);
 
     done_tx.send(()).unwrap();
@@ -335,9 +338,9 @@ async fn concurrent_progress() {
     assert_eq!(prog_b.len(), 1);
     assert_eq!(prog_c.len(), 1);
 
-    assert!(matches!(prog_a[0], DownloadProgress::Done { .. }));
-    assert!(matches!(prog_b[0], DownloadProgress::Done { .. }));
-    assert!(matches!(prog_c[0], DownloadProgress::Done { .. }));
+    assert!(matches!(prog_a[0], DownloadProgressEvent::Done { .. }));
+    assert!(matches!(prog_b[0], DownloadProgressEvent::Done { .. }));
+    assert!(matches!(prog_c[0], DownloadProgressEvent::Done { .. }));
 
     for p in prog_a {
         state_a.on_progress(p);