Update to latest irpc

Author: rklaehn

Cargo.lock

@@ -33,8 +33,8 @@ self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
 iroh-base = "0.35.0"
 reflink-copy = "0.1.24"
-irpc = { version = "0.3.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream"], default-features = false }
-irpc-derive = { version = "0.3.0", default-features = false }
+irpc = { version = "0.4.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream"], default-features = false }
+irpc-derive = { version = "0.4.0", default-features = false }
 iroh-metrics = { version = "0.32.0" }
 hashlink = "0.10.0"
 futures-buffered = "0.2.11"

Cargo.toml

@@ -24,7 +24,7 @@ use bao_tree::{
 use bytes::Bytes;
 use genawaiter::sync::Gen;
 use iroh_io::{AsyncStreamReader, TokioStreamReader};
-use irpc::channel::{oneshot, spsc};
+use irpc::channel::{mpsc, oneshot};
 use n0_future::{Stream, StreamExt, future, stream};
 use quinn::SendStream;
 use range_collections::{RangeSet2, range_set::RangeSetRange};
@@ -225,7 +225,7 @@ impl Blobs {
         };
         let client = self.client.clone();
         let stream = Gen::new(|co| async move {
-            let (mut sender, mut receiver) = match client.bidi_streaming(inner, 32, 32).await {
+            let (sender, mut receiver) = match client.bidi_streaming(inner, 32, 32).await {
                 Ok(x) => x,
                 Err(cause) => {
                     co.yield_(AddProgressItem::Error(cause.into())).await;
@@ -338,7 +338,7 @@ impl Blobs {
         trace!("{:?}", options);
         if options.hash == Hash::EMPTY {
             return ObserveProgress::new(async move {
-                let (mut tx, rx) = spsc::channel(1);
+                let (tx, rx) = mpsc::channel(1);
                 tx.send(Bitfield::complete(0)).await.ok();
                 Ok(rx)
             });
@@ -405,7 +405,7 @@ impl Blobs {
         let tree = BaoTree::new(size.get(), IROH_BLOCK_SIZE);
         let mut decoder = ResponseDecoder::new(hash.into(), ranges, tree, reader);
         let options = ImportBaoOptions { hash, size };
-        let mut handle = self.import_bao_with_opts(options, 32).await?;
+        let handle = self.import_bao_with_opts(options, 32).await?;
         let driver = async move {
             let reader = loop {
                 match decoder.next().await {
@@ -509,7 +509,7 @@ impl<'a> BatchAddProgress<'a> {
 pub struct Batch<'a> {
     scope: Scope,
     blobs: &'a Blobs,
-    _tx: spsc::Sender<BatchResponse>,
+    _tx: mpsc::Sender<BatchResponse>,
 }
 
 impl<'a> Batch<'a> {
@@ -647,7 +647,7 @@ impl<'a> AddProgress<'a> {
 /// Calling [`ObserveProgress::aggregated`] will return a stream of states,
 /// where each state is the current state at the time of the update.
 pub struct ObserveProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<Bitfield>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<Bitfield>>>,
 }
 
 impl IntoFuture for ObserveProgress {
@@ -668,7 +668,7 @@ impl IntoFuture for ObserveProgress {
 
 impl ObserveProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<Bitfield>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<Bitfield>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
@@ -710,7 +710,7 @@ impl ObserveProgress {
 /// It also implements [`IntoFuture`], so you can await it to get the size of the
 /// exported blob.
 pub struct ExportProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<ExportProgressItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<ExportProgressItem>>>,
 }
 
 impl IntoFuture for ExportProgress {
@@ -725,7 +725,7 @@ impl IntoFuture for ExportProgress {
 
 impl ExportProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<ExportProgressItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<ExportProgressItem>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
@@ -768,15 +768,15 @@ impl ExportProgress {
 
 /// A handle for an ongoing bao import operation.
 pub struct ImportBaoHandle {
-    pub tx: spsc::Sender<BaoContentItem>,
+    pub tx: mpsc::Sender<BaoContentItem>,
     pub rx: oneshot::Receiver<super::Result<()>>,
 }
 
 impl ImportBaoHandle {
     pub(crate) async fn new(
         fut: impl Future<
             Output = irpc::Result<(
-                spsc::Sender<BaoContentItem>,
+                mpsc::Sender<BaoContentItem>,
                 oneshot::Receiver<super::Result<()>>,
             )>,
         > + Send
@@ -789,20 +789,20 @@ impl ImportBaoHandle {
 
 /// A progress handle for a blobs list operation.
 pub struct BlobsListProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<super::Result<Hash>>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<super::Result<Hash>>>>,
 }
 
 impl BlobsListProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<super::Result<Hash>>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<super::Result<Hash>>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
         }
     }
 
     pub async fn hashes(self) -> RequestResult<Vec<Hash>> {
-        let mut rx: spsc::Receiver<Result<Hash, super::Error>> = self.inner.await?;
+        let mut rx: mpsc::Receiver<Result<Hash, super::Error>> = self.inner.await?;
         let mut hashes = Vec::new();
         while let Some(item) = rx.recv().await? {
             hashes.push(item?);
@@ -829,13 +829,13 @@ impl BlobsListProgress {
 /// You can get access to the underlying stream using the [`ExportBaoResult::stream`] method.
 pub struct ExportRangesProgress {
     ranges: RangeSet2<u64>,
-    inner: future::Boxed<irpc::Result<spsc::Receiver<ExportRangesItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<ExportRangesItem>>>,
 }
 
 impl ExportRangesProgress {
     fn new(
         ranges: RangeSet2<u64>,
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<ExportRangesItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<ExportRangesItem>>> + Send + 'static,
     ) -> Self {
         Self {
             ranges,
@@ -909,12 +909,12 @@ impl ExportRangesProgress {
 ///
 /// You can get access to the underlying stream using the [`ExportBaoResult::stream`] method.
 pub struct ExportBaoProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<EncodedItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<EncodedItem>>>,
 }
 
 impl ExportBaoProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<EncodedItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<EncodedItem>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),

src/api/blobs.rs

@@ -11,15 +11,12 @@ use std::{
 use anyhow::bail;
 use genawaiter::sync::Gen;
 use iroh::{Endpoint, NodeId, endpoint::Connection};
-use irpc::channel::spsc;
+use irpc::channel::mpsc;
 use irpc_derive::rpc_requests;
 use n0_future::{BufferedStreamExt, Stream, StreamExt, future, stream};
 use rand::seq::SliceRandom;
 use serde::{Deserialize, Serialize, de::Error};
-use tokio::{
-    sync::{Mutex, mpsc},
-    task::JoinSet,
-};
+use tokio::{sync::Mutex, task::JoinSet};
 use tokio_util::time::FutureExt;
 use tracing::{info, instrument::Instrument, warn};
 
@@ -43,7 +40,7 @@ impl irpc::Service for DownloaderService {}
 #[rpc_requests(DownloaderService, message = SwarmMsg, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 enum SwarmProtocol {
-    #[rpc(tx = spsc::Sender<DownloadProgessItem>)]
+    #[rpc(tx = mpsc::Sender<DownloadProgessItem>)]
     Download(DownloadRequest),
 }
 
@@ -83,7 +80,7 @@ impl DownloaderActor {
         }
     }
 
-    async fn run(mut self, mut rx: mpsc::Receiver<SwarmMsg>) {
+    async fn run(mut self, mut rx: tokio::sync::mpsc::Receiver<SwarmMsg>) {
         while let Some(msg) = rx.recv().await {
             match msg {
                 SwarmMsg::Download(request) => {
@@ -115,7 +112,7 @@ async fn handle_download_impl(
     store: Store,
     pool: ConnectionPool,
     request: DownloadRequest,
-    tx: &mut spsc::Sender<DownloadProgessItem>,
+    tx: &mut mpsc::Sender<DownloadProgessItem>,
 ) -> anyhow::Result<()> {
     match request.strategy {
         SplitStrategy::Split => handle_download_split_impl(store, pool, request, tx).await?,
@@ -136,11 +133,11 @@ async fn handle_download_split_impl(
     store: Store,
     pool: ConnectionPool,
     request: DownloadRequest,
-    tx: &mut spsc::Sender<DownloadProgessItem>,
+    tx: &mut mpsc::Sender<DownloadProgessItem>,
 ) -> anyhow::Result<()> {
     let providers = request.providers;
     let requests = split_request(&request.request, &providers, &pool, &store, Drain).await?;
-    let (progress_tx, progress_rx) = mpsc::channel(32);
+    let (progress_tx, progress_rx) = tokio::sync::mpsc::channel(32);
     let mut futs = stream::iter(requests.into_iter().enumerate())
         .map(|(id, request)| {
             let pool = pool.clone();
@@ -149,7 +146,7 @@ async fn handle_download_split_impl(
             let progress_tx = progress_tx.clone();
             async move {
                 let hash = request.hash;
-                let (tx, rx) = mpsc::channel::<(usize, DownloadProgessItem)>(16);
+                let (tx, rx) = tokio::sync::mpsc::channel::<(usize, DownloadProgessItem)>(16);
                 progress_tx.send(rx).await.ok();
                 let sink = TokioMpscSenderSink(tx)
                     .with_map_err(io::Error::other)
@@ -199,7 +196,7 @@ async fn handle_download_split_impl(
     Ok(())
 }
 
-fn into_stream<T>(mut recv: mpsc::Receiver<T>) -> impl Stream<Item = T> {
+fn into_stream<T>(mut recv: tokio::sync::mpsc::Receiver<T>) -> impl Stream<Item = T> {
     Gen::new(|co| async move {
         while let Some(item) = recv.recv().await {
             co.yield_(item).await;
@@ -309,11 +306,11 @@ impl<'de> Deserialize<'de> for DownloadRequest {
 pub type DownloadOptions = DownloadRequest;
 
 pub struct DownloadProgress {
-    fut: future::Boxed<irpc::Result<spsc::Receiver<DownloadProgessItem>>>,
+    fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgessItem>>>,
 }
 
 impl DownloadProgress {
-    fn new(fut: future::Boxed<irpc::Result<spsc::Receiver<DownloadProgessItem>>>) -> Self {
+    fn new(fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgessItem>>>) -> Self {
         Self { fut }
     }
 
@@ -351,7 +348,7 @@ impl IntoFuture for DownloadProgress {
 
 impl Downloader {
     pub fn new(store: &Store, endpoint: &Endpoint) -> Self {
-        let (tx, rx) = mpsc::channel::<SwarmMsg>(32);
+        let (tx, rx) = tokio::sync::mpsc::channel::<SwarmMsg>(32);
         let actor = DownloaderActor::new(store.clone(), endpoint.clone());
         tokio::spawn(actor.run(rx));
         Self { client: tx.into() }

src/api/downloader.rs

@@ -28,7 +28,7 @@ use bao_tree::{
     io::{BaoContentItem, Leaf, mixed::EncodedItem},
 };
 use bytes::Bytes;
-use irpc::channel::{oneshot, spsc};
+use irpc::channel::{mpsc, oneshot};
 use irpc_derive::rpc_requests;
 use n0_future::Stream;
 use range_collections::RangeSet2;
@@ -91,29 +91,29 @@ impl irpc::Service for StoreService {}
 #[rpc_requests(StoreService, message = Command, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 pub enum Request {
-    #[rpc(tx = spsc::Sender<super::Result<Hash>>)]
+    #[rpc(tx = mpsc::Sender<super::Result<Hash>>)]
     ListBlobs(ListRequest),
-    #[rpc(tx = oneshot::Sender<Scope>, rx = spsc::Receiver<BatchResponse>)]
+    #[rpc(tx = oneshot::Sender<Scope>, rx = mpsc::Receiver<BatchResponse>)]
     Batch(BatchRequest),
     #[rpc(tx = oneshot::Sender<super::Result<()>>)]
     DeleteBlobs(BlobDeleteRequest),
-    #[rpc(rx = spsc::Receiver<BaoContentItem>, tx = oneshot::Sender<super::Result<()>>)]
+    #[rpc(rx = mpsc::Receiver<BaoContentItem>, tx = oneshot::Sender<super::Result<()>>)]
     ImportBao(ImportBaoRequest),
-    #[rpc(tx = spsc::Sender<EncodedItem>)]
+    #[rpc(tx = mpsc::Sender<EncodedItem>)]
     ExportBao(ExportBaoRequest),
-    #[rpc(tx = spsc::Sender<ExportRangesItem>)]
+    #[rpc(tx = mpsc::Sender<ExportRangesItem>)]
     ExportRanges(ExportRangesRequest),
-    #[rpc(tx = spsc::Sender<Bitfield>)]
+    #[rpc(tx = mpsc::Sender<Bitfield>)]
     Observe(ObserveRequest),
     #[rpc(tx = oneshot::Sender<BlobStatus>)]
     BlobStatus(BlobStatusRequest),
-    #[rpc(tx = spsc::Sender<AddProgressItem>)]
+    #[rpc(tx = mpsc::Sender<AddProgressItem>)]
     ImportBytes(ImportBytesRequest),
-    #[rpc(rx = spsc::Receiver<ImportByteStreamUpdate>, tx = spsc::Sender<AddProgressItem>)]
+    #[rpc(rx = mpsc::Receiver<ImportByteStreamUpdate>, tx = mpsc::Sender<AddProgressItem>)]
     ImportByteStream(ImportByteStreamRequest),
-    #[rpc(tx = spsc::Sender<AddProgressItem>)]
+    #[rpc(tx = mpsc::Sender<AddProgressItem>)]
     ImportPath(ImportPathRequest),
-    #[rpc(tx = spsc::Sender<ExportProgressItem>)]
+    #[rpc(tx = mpsc::Sender<ExportProgressItem>)]
     ExportPath(ExportPathRequest),
     #[rpc(tx = oneshot::Sender<Vec<super::Result<TagInfo>>>)]
     ListTags(ListTagsRequest),

src/api/proto.rs

@@ -860,7 +860,7 @@ async fn get_blob_ranges_impl(
     };
     let buffer_size = get_buffer_size(size);
     trace!(%size, %buffer_size, "get blob");
-    let mut handle = store
+    let handle = store
         .import_bao(hash, size, buffer_size)
         .await
         .map_err(|e| GetError::LocalFailure(e.into()))?;