Remove most uses of flume in iroh itself

The rest requires the docs flume purge PR to be merged.

iroh-blobs/src/downloader/progress.rs

@@ -11,13 +11,13 @@ use parking_lot::Mutex;
 
 use crate::{
     get::{db::DownloadProgress, progress::TransferState},
-    util::progress::{FlumeProgressSender, IdGenerator, ProgressSendError, ProgressSender},
+    util::progress::{AsyncChannelProgressSender, IdGenerator, ProgressSendError, ProgressSender},
 };
 
 use super::DownloadKind;
 
 /// The channel that can be used to subscribe to progress updates.
-pub type ProgressSubscriber = FlumeProgressSender<DownloadProgress>;
+pub type ProgressSubscriber = AsyncChannelProgressSender<DownloadProgress>;
 
 /// Track the progress of downloads.
 ///

iroh-blobs/src/downloader/test.rs

@@ -12,7 +12,7 @@ use crate::{
     get::{db::BlobId, progress::TransferState},
     util::{
         local_pool::LocalPool,
-        progress::{FlumeProgressSender, IdGenerator},
+        progress::{AsyncChannelProgressSender, IdGenerator},
     },
 };
 
@@ -276,13 +276,13 @@ async fn concurrent_progress() {
     let hash = Hash::new([0u8; 32]);
     let kind_1 = HashAndFormat::raw(hash);
 
-    let (prog_a_tx, prog_a_rx) = flume::bounded(64);
-    let prog_a_tx = FlumeProgressSender::new(prog_a_tx);
+    let (prog_a_tx, prog_a_rx) = async_channel::bounded(64);
+    let prog_a_tx = AsyncChannelProgressSender::new(prog_a_tx);
     let req = DownloadRequest::new(kind_1, vec![peer]).progress_sender(prog_a_tx);
     let handle_a = downloader.queue(req).await;
 
-    let (prog_b_tx, prog_b_rx) = flume::bounded(64);
-    let prog_b_tx = FlumeProgressSender::new(prog_b_tx);
+    let (prog_b_tx, prog_b_rx) = async_channel::bounded(64);
+    let prog_b_tx = AsyncChannelProgressSender::new(prog_b_tx);
     let req = DownloadRequest::new(kind_1, vec![peer]).progress_sender(prog_b_tx);
     let handle_b = downloader.queue(req).await;
 
@@ -292,21 +292,21 @@ async fn concurrent_progress() {
     let mut state_b = TransferState::new(hash);
     let mut state_c = TransferState::new(hash);
 
-    let prog1_a = prog_a_rx.recv_async().await.unwrap();
-    let prog1_b = prog_b_rx.recv_async().await.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, size: 100, ..} if hash == hash));
     assert!(matches!(prog1_b, DownloadProgress::Found { hash, size: 100, ..} if hash == hash));
 
     state_a.on_progress(prog1_a);
     state_b.on_progress(prog1_b);
     assert_eq!(state_a, state_b);
 
-    let (prog_c_tx, prog_c_rx) = flume::bounded(64);
-    let prog_c_tx = FlumeProgressSender::new(prog_c_tx);
+    let (prog_c_tx, prog_c_rx) = async_channel::bounded(64);
+    let prog_c_tx = AsyncChannelProgressSender::new(prog_c_tx);
     let req = DownloadRequest::new(kind_1, vec![peer]).progress_sender(prog_c_tx);
     let handle_c = downloader.queue(req).await;
 
-    let prog1_c = prog_c_rx.recv_async().await.unwrap();
+    let prog1_c = prog_c_rx.recv().await.unwrap();
     assert!(matches!(&prog1_c, DownloadProgress::InitialState(state) if state == &state_a));
     state_c.on_progress(prog1_c);
 
@@ -317,9 +317,9 @@ async fn concurrent_progress() {
     res_b.unwrap();
     res_c.unwrap();
 
-    let prog_a: Vec<_> = prog_a_rx.into_stream().collect().await;
-    let prog_b: Vec<_> = prog_b_rx.into_stream().collect().await;
-    let prog_c: Vec<_> = prog_c_rx.into_stream().collect().await;
+    let prog_a: Vec<_> = prog_a_rx.collect().await;
+    let prog_b: Vec<_> = prog_b_rx.collect().await;
+    let prog_c: Vec<_> = prog_c_rx.collect().await;
 
     assert_eq!(prog_a.len(), 1);
     assert_eq!(prog_b.len(), 1);

iroh/Cargo.toml

@@ -53,6 +53,7 @@ tracing = "0.1"
 walkdir = "2"
 
 # Examples
+async-channel = "2.3.1"
 clap = { version = "4", features = ["derive"], optional = true }
 indicatif = { version = "0.17", features = ["tokio"], optional = true }
 ref-cast = "1.0.23"