refactor: move file content conversion to mod.rs

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

@@ -7,9 +7,12 @@ pub use v3::LsHttpClientV3;
 
 use crate::http::StatusCodeError;
 use crate::{crypto, http, model};
+use bytes::Bytes;
+use futures_util::StreamExt;
 use reqwest::Response;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
+use tokio_stream::wrappers::ReceiverStream;
 
 pub enum LsHttpClient {
     V2(LsHttpClientV2),
@@ -117,20 +120,19 @@ impl LsHttpClient {
         progress: impl Fn(u64) + Send + 'static,
         cancel: tokio_util::sync::CancellationToken,
     ) -> Result<(), ClientError> {
+        let body = upload_body(content, progress);
         match self {
             LsHttpClient::V2(client) => {
                 client
                     .upload(
-                        protocol, ip, port, public_key, session_id, file_id, token, content,
-                        progress, cancel,
+                        protocol, ip, port, public_key, session_id, file_id, token, body, cancel,
                     )
                     .await
             }
             LsHttpClient::V3(client) => {
                 client
                     .upload(
-                        protocol, ip, port, public_key, session_id, file_id, token, content,
-                        progress, cancel,
+                        protocol, ip, port, public_key, session_id, file_id, token, body, cancel,
                     )
                     .await
             }
@@ -151,6 +153,21 @@ impl LsHttpClient {
     }
 }
 
+/// Builds a streaming request body from the file content, invoking `progress`
+/// with the cumulative number of bytes read as each chunk is sent.
+pub(super) fn upload_body(
+    content: model::transfer::FileContent,
+    progress: impl Fn(u64) + Send + 'static,
+) -> reqwest::Body {
+    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)
+}
+
 pub(super) fn create_reqwest_client(
     private_key: &str,
     cert: &str,

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

@@ -5,12 +5,9 @@ use crate::http::dto_v2::{
     InfoResponseDtoV2, PrepareDownloadResponseDtoV2, PrepareUploadRequestDtoV2,
     PrepareUploadResponseDtoV2, PrepareUploadResultV2, RegisterDtoV2, RegisterResponseDtoV2,
 };
-use crate::model;
-use bytes::Bytes;
 use futures_util::StreamExt;
 use reqwest::{Response, StatusCode};
 use tokio::io::AsyncWriteExt;
-use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::sync::CancellationToken;
 
 /// HTTP client for LocalSend Protocol v2.1.
@@ -199,8 +196,7 @@ impl LsHttpClientV2 {
     /// * `session_id` - Session ID from prepare_upload
     /// * `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
+    /// * `body` - The streaming request body carrying the file content
     /// * `cancel` - Cancellation token; cancelling it aborts the upload with [`ClientError::Cancelled`]
     ///
     /// # Errors
@@ -217,8 +213,7 @@ impl LsHttpClientV2 {
         session_id: &str,
         file_id: &str,
         token: &str,
-        content: model::transfer::FileContent,
-        progress: impl Fn(u64) + Send + 'static,
+        body: reqwest::Body,
         cancel: CancellationToken,
     ) -> Result<(), ClientError> {
         let url = TargetUrl {
@@ -235,14 +230,6 @@ impl LsHttpClientV2 {
         }
         .to_string();
 
-        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! {
             res = self.client.post(&url).body(body).send() => res?,
             _ = cancel.cancelled() => return Err(ClientError::Cancelled),

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

@@ -1,16 +1,13 @@
 use super::{ClientError, ResponseExt, ResultWithPublicKey};
 use crate::http::client::url::{ApiVersion, TargetUrl};
 use crate::http::dto::ProtocolType;
+use crate::http;
 use crate::{crypto, util};
-use crate::{http, model};
-use bytes::Bytes;
-use futures_util::StreamExt;
 use lru::LruCache;
 use reqwest::{Response, StatusCode};
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use tokio::sync::Mutex;
-use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::sync::CancellationToken;
 
 pub struct LsHttpClientV3 {
@@ -187,7 +184,7 @@ impl LsHttpClientV3 {
 
     /// Uploads a file to the server.
     ///
-    /// `progress` is called with the cumulative number of bytes read for the upload.
+    /// `body` is the streaming request body carrying the file content.
     ///
     /// `cancel` is a cancellation token; cancelling it aborts the upload with
     /// [`ClientError::Cancelled`].
@@ -200,8 +197,7 @@ impl LsHttpClientV3 {
         session_id: &str,
         file_id: &str,
         token: &str,
-        content: model::transfer::FileContent,
-        progress: impl Fn(u64) + Send + 'static,
+        body: reqwest::Body,
         cancel: CancellationToken,
     ) -> Result<(), ClientError> {
         let send = self
@@ -221,15 +217,7 @@ impl LsHttpClientV3 {
                 }
                 .to_string(),
             )
-            .body({
-                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)
-            })
+            .body(body)
             .send();
 
         let res = tokio::select! {