feat: add batched Rust-backed upload task

packages/core/src/http/client/mod.rs

@@ -114,20 +114,23 @@ impl LsHttpClient {
         file_id: &str,
         token: &str,
         content: model::transfer::FileContent,
+        progress: impl Fn(u64) + Send + 'static,
         cancel: tokio_util::sync::CancellationToken,
     ) -> Result<(), ClientError> {
         match self {
             LsHttpClient::V2(client) => {
                 client
                     .upload(
-                        protocol, ip, port, public_key, session_id, file_id, token, content, cancel,
+                        protocol, ip, port, public_key, session_id, file_id, token, content,
+                        progress, cancel,
                     )
                     .await
             }
             LsHttpClient::V3(client) => {
                 client
                     .upload(
-                        protocol, ip, port, public_key, session_id, file_id, token, content, cancel,
+                        protocol, ip, port, public_key, session_id, file_id, token, content,
+                        progress, cancel,
                     )
                     .await
             }

packages/core/src/http/client/v2.rs

@@ -200,6 +200,7 @@ impl LsHttpClientV2 {
     /// * `file_id` - File ID to upload
     /// * `token` - File-specific token from prepare_upload
     /// * `content` - The file content to upload (a chunk stream or a raw file descriptor)
+    /// * `progress` - Called with the cumulative number of bytes read for the upload
     /// * `cancel` - Cancellation token; cancelling it aborts the upload with [`ClientError::Cancelled`]
     ///
     /// # Errors
@@ -217,6 +218,7 @@ impl LsHttpClientV2 {
         file_id: &str,
         token: &str,
         content: model::transfer::FileContent,
+        progress: impl Fn(u64) + Send + 'static,
         cancel: CancellationToken,
     ) -> Result<(), ClientError> {
         let url = TargetUrl {
@@ -233,7 +235,12 @@ impl LsHttpClientV2 {
         }
         .to_string();
 
-        let stream = ReceiverStream::new(content.into_receiver()).map(Ok::<Bytes, anyhow::Error>);
+        let mut sent = 0_u64;
+        let stream = ReceiverStream::new(content.into_receiver()).map(move |chunk| {
+            sent += chunk.len() as u64;
+            progress(sent);
+            Ok::<Bytes, anyhow::Error>(chunk)
+        });
         let body = reqwest::Body::wrap_stream(stream);
 
         let res = tokio::select! {

packages/core/src/http/client/v3.rs

@@ -187,6 +187,8 @@ impl LsHttpClientV3 {
 
     /// Uploads a file to the server.
     ///
+    /// `progress` is called with the cumulative number of bytes read for the upload.
+    ///
     /// `cancel` is a cancellation token; cancelling it aborts the upload with
     /// [`ClientError::Cancelled`].
     pub async fn upload(
@@ -199,6 +201,7 @@ impl LsHttpClientV3 {
         file_id: &str,
         token: &str,
         content: model::transfer::FileContent,
+        progress: impl Fn(u64) + Send + 'static,
         cancel: CancellationToken,
     ) -> Result<(), ClientError> {
         let send = self
@@ -219,8 +222,12 @@ impl LsHttpClientV3 {
                 .to_string(),
             )
             .body({
-                let stream =
-                    ReceiverStream::new(content.into_receiver()).map(Ok::<Bytes, anyhow::Error>);
+                let mut sent = 0_u64;
+                let stream = ReceiverStream::new(content.into_receiver()).map(move |chunk| {
+                    sent += chunk.len() as u64;
+                    progress(sent);
+                    Ok::<Bytes, anyhow::Error>(chunk)
+                });
                 reqwest::Body::wrap_stream(stream)
             })
             .send();

packages/core/tests/v2_server.rs

@@ -11,7 +11,7 @@ use localsend::http::state::ClientInfo;
 use localsend::model::transfer::{FileContent, FileDto};
 use std::collections::HashMap;
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicU16, Ordering};
+use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{mpsc, oneshot, Mutex};
@@ -209,6 +209,7 @@ async fn upload_bytes(
 ) -> Result<(), ClientError> {
     let (tx, rx) = mpsc::channel::<Bytes>(4);
     let chunks: Vec<Vec<u8>> = bytes.chunks(1024).map(|chunk| chunk.to_vec()).collect();
+    let sent = Arc::new(AtomicU64::new(0));
     tokio::spawn(async move {
         for chunk in chunks {
             if tx.send(Bytes::from(chunk)).await.is_err() {
@@ -217,7 +218,8 @@ async fn upload_bytes(
         }
     });
 
-    client
+    let progress = sent.clone();
+    let result = client
         .upload(
             ProtocolType::Http,
             "127.0.0.1",
@@ -227,9 +229,18 @@ async fn upload_bytes(
             file_id,
             token,
             FileContent::Stream(rx),
+            move |bytes_sent| {
+                progress.store(bytes_sent, Ordering::Relaxed);
+            },
             CancellationToken::new(),
         )
-        .await
+        .await;
+
+    if result.is_ok() {
+        assert_eq!(sent.load(Ordering::Relaxed), bytes.len() as u64);
+    }
+
+    result
 }
 
 fn assert_status(result: Result<impl Sized, ClientError>, expected_status: u16) {