Add import_byte_stream_mid that takes a stream

so we don't have to go item -> request -> stream

Author: rklaehn

Cargo.lock

@@ -33,8 +33,8 @@ self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
 iroh-base = "0.34.1"
 reflink-copy = "0.1.24"
-irpc = { version = "0.2.3", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream"], default-features = false, path = "../irpc" }
-irpc-derive = { version = "0.2.3", default-features = false }
+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 }
 iroh-metrics = { version = "0.32.0" }
 hashlink = "0.10.0"
 futures-buffered = "0.2.11"

Cargo.toml

@@ -15,7 +15,7 @@ use std::{
     fs::{self, File, OpenOptions},
     io::{self, Seek, Write},
     path::PathBuf,
-    sync::{Arc, mpsc::RecvError},
+    sync::Arc,
 };
 
 use bao_tree::{
@@ -163,21 +163,12 @@ pub async fn import_bytes(cmd: ImportBytesMsg, ctx: Arc<TaskContext>) {
     if ctx.options.is_inlined_all(size) {
         import_bytes_tiny_outer(cmd, ctx).await;
     } else {
-        let (mut tx, rx) = spsc::channel(2);
-        tx.send(ImportByteStreamUpdate::Bytes(cmd.data.clone()))
-            .await
-            .unwrap();
-        tx.send(ImportByteStreamUpdate::Done).await.unwrap();
-        let cmd = ImportByteStreamMsg {
-            inner: ImportByteStreamRequest {
-                format: cmd.format,
-                scope: cmd.scope,
-            },
-            tx: cmd.tx,
-            rx,
-            span: tracing::Span::current(),
+        let request = ImportByteStreamRequest {
+            format: cmd.format,
+            scope: cmd.scope,
         };
-        import_byte_stream_outer(cmd, ctx).await;
+        let stream = stream::iter(Some(Ok(cmd.data.clone())));
+        import_byte_stream_mid(request, cmd.tx, cmd.span, stream, ctx).await;
     }
 }
 
@@ -235,7 +226,8 @@ async fn import_bytes_tiny_impl(
 
 #[instrument(skip_all)]
 pub async fn import_byte_stream(cmd: ImportByteStreamMsg, ctx: Arc<TaskContext>) {
-    import_byte_stream_outer(cmd, ctx).await;
+    let stream = into_stream(cmd.rx);
+    import_byte_stream_mid(cmd.inner, cmd.tx, cmd.span, stream, ctx).await
 }
 
 fn into_stream(
@@ -263,20 +255,25 @@ fn into_stream(
     })
 }
 
-pub async fn import_byte_stream_outer(mut cmd: ImportByteStreamMsg, ctx: Arc<TaskContext>) {
-    let stream = into_stream(cmd.rx);
-    match import_byte_stream_impl(cmd.inner, &mut cmd.tx, stream, ctx.options.clone()).await {
+async fn import_byte_stream_mid(
+    request: ImportByteStreamRequest,
+    mut tx: spsc::Sender<AddProgressItem>,
+    span: tracing::Span,
+    stream: impl Stream<Item = io::Result<Bytes>> + Unpin,
+    ctx: Arc<TaskContext>,
+) {
+    match import_byte_stream_impl(request, &mut tx, stream, ctx.options.clone()).await {
         Ok(entry) => {
             let entry = ImportEntryMsg {
                 inner: entry,
-                tx: cmd.tx,
+                tx,
                 rx: NoReceiver,
-                span: cmd.span,
+                span,
             };
             ctx.internal_cmd_tx.send(entry.into()).await.ok();
         }
         Err(cause) => {
-            cmd.tx.send(cause.into()).await.ok();
+            tx.send(cause.into()).await.ok();
         }
     }
 }

src/store/fs/import.rs

@@ -27,7 +27,7 @@ use bao_tree::{
 };
 use bytes::Bytes;
 use irpc::channel::spsc;
-use n0_future::{StreamExt, future::yield_now};
+use n0_future::future::yield_now;
 use range_collections::range_set::RangeSetRange;
 use tokio::{
     io::AsyncReadExt,
@@ -43,15 +43,14 @@ use crate::{
         self, ApiClient,
         blobs::{AddProgressItem, Bitfield, BlobStatus, ExportProgressItem},
         proto::{
-            BatchMsg, BatchResponse, BlobDeleteRequest, BlobStatusMsg, BlobStatusRequest,
-            BoxedByteStream, Command, CreateTagMsg, CreateTagRequest, CreateTempTagMsg,
-            DeleteBlobsMsg, DeleteTagsMsg, DeleteTagsRequest, ExportBaoMsg, ExportBaoRequest,
-            ExportPathMsg, ExportPathRequest, ExportRangesItem, ExportRangesMsg,
-            ExportRangesRequest, ImportBaoMsg, ImportBaoRequest, ImportByteStreamMsg,
-            ImportByteStreamUpdate, ImportBytesMsg, ImportBytesRequest, ImportPathMsg,
-            ImportPathRequest, ListBlobsMsg, ListTagsMsg, ListTagsRequest, ObserveMsg,
-            ObserveRequest, RenameTagMsg, RenameTagRequest, Scope, SetTagMsg, SetTagRequest,
-            ShutdownMsg, SyncDbMsg,
+            BatchMsg, BatchResponse, BlobDeleteRequest, BlobStatusMsg, BlobStatusRequest, Command,
+            CreateTagMsg, CreateTagRequest, CreateTempTagMsg, DeleteBlobsMsg, DeleteTagsMsg,
+            DeleteTagsRequest, ExportBaoMsg, ExportBaoRequest, ExportPathMsg, ExportPathRequest,
+            ExportRangesItem, ExportRangesMsg, ExportRangesRequest, ImportBaoMsg, ImportBaoRequest,
+            ImportByteStreamMsg, ImportByteStreamUpdate, ImportBytesMsg, ImportBytesRequest,
+            ImportPathMsg, ImportPathRequest, ListBlobsMsg, ListTagsMsg, ListTagsRequest,
+            ObserveMsg, ObserveRequest, RenameTagMsg, RenameTagRequest, Scope, SetTagMsg,
+            SetTagRequest, ShutdownMsg, SyncDbMsg,
         },
         tags::TagInfo,
     },