From 3aedb1a67ff1e689f46145bd86ca650fe1299bd1 Mon Sep 17 00:00:00 2001 From: Tien Do Nam Date: Mon, 13 Jul 2026 21:47:32 +0200 Subject: [PATCH] refactor: restructure http server --- core/src/http/client/v2.rs | 3 +- core/src/http/client/v3.rs | 3 +- .../{ => common}/client_cert_verifier.rs | 0 .../server/{ => common}/collect_to_json.rs | 2 +- core/src/http/server/{ => common}/error.rs | 2 +- core/src/http/server/common/mod.rs | 8 + .../{controller/mod.rs => common/pin.rs} | 6 +- core/src/http/server/{ => common}/query.rs | 0 core/src/http/server/{ => common}/response.rs | 0 core/src/http/server/common/save.rs | 183 ++++++++++++++++ core/src/http/server/{ => common}/session.rs | 0 core/src/http/server/mod.rs | 75 +++---- core/src/http/server/{controller => }/v2.rs | 197 ++---------------- core/src/http/server/{controller => }/v3.rs | 6 +- core/src/http/server/{controller => }/web.rs | 16 +- core/src/main.rs | 5 +- core/src/model/transfer.rs | 15 +- core/tests/v2_server.rs | 14 +- core/tests/v2_web_send.rs | 18 +- 19 files changed, 285 insertions(+), 268 deletions(-) rename core/src/http/server/{ => common}/client_cert_verifier.rs (100%) rename core/src/http/server/{ => common}/collect_to_json.rs (92%) rename core/src/http/server/{ => common}/error.rs (94%) create mode 100644 core/src/http/server/common/mod.rs rename core/src/http/server/{controller/mod.rs => common/pin.rs} (93%) rename core/src/http/server/{ => common}/query.rs (100%) rename core/src/http/server/{ => common}/response.rs (100%) create mode 100644 core/src/http/server/common/save.rs rename core/src/http/server/{ => common}/session.rs (100%) rename core/src/http/server/{controller => }/v2.rs (68%) rename core/src/http/server/{controller => }/v3.rs (92%) rename core/src/http/server/{controller => }/web.rs (96%) diff --git a/core/src/http/client/v2.rs b/core/src/http/client/v2.rs index 99585a46..73b9a665 100644 --- a/core/src/http/client/v2.rs +++ b/core/src/http/client/v2.rs @@ -6,6 +6,7 @@ use crate::http::dto_v2::{ PrepareUploadResponseDtoV2, PrepareUploadResultV2, RegisterDtoV2, RegisterResponseDtoV2, }; use crate::model; +use bytes::Bytes; use futures_util::StreamExt; use reqwest::{Response, StatusCode}; use tokio::io::AsyncWriteExt; @@ -227,7 +228,7 @@ impl LsHttpClientV2 { } .to_string(); - let stream = ReceiverStream::new(content.into_receiver()).map(Ok::, anyhow::Error>); + let stream = ReceiverStream::new(content.into_receiver()).map(Ok::); let body = reqwest::Body::wrap_stream(stream); let res = tokio::select! { diff --git a/core/src/http/client/v3.rs b/core/src/http/client/v3.rs index 68bfb86c..70f9ace0 100644 --- a/core/src/http/client/v3.rs +++ b/core/src/http/client/v3.rs @@ -3,6 +3,7 @@ use crate::http::client::url::{ApiVersion, TargetUrl}; use crate::http::dto::ProtocolType; use crate::{crypto, util}; use crate::{http, model}; +use bytes::Bytes; use futures_util::StreamExt; use lru::LruCache; use reqwest::{Response, StatusCode}; @@ -215,7 +216,7 @@ impl LsHttpClientV3 { ) .body({ let stream = - ReceiverStream::new(content.into_receiver()).map(Ok::, anyhow::Error>); + ReceiverStream::new(content.into_receiver()).map(Ok::); reqwest::Body::wrap_stream(stream) }) .send(); diff --git a/core/src/http/server/client_cert_verifier.rs b/core/src/http/server/common/client_cert_verifier.rs similarity index 100% rename from core/src/http/server/client_cert_verifier.rs rename to core/src/http/server/common/client_cert_verifier.rs diff --git a/core/src/http/server/collect_to_json.rs b/core/src/http/server/common/collect_to_json.rs similarity index 92% rename from core/src/http/server/collect_to_json.rs rename to core/src/http/server/common/collect_to_json.rs index 24fdc35e..9a99f9c7 100644 --- a/core/src/http/server/collect_to_json.rs +++ b/core/src/http/server/common/collect_to_json.rs @@ -1,4 +1,4 @@ -use crate::http::server::error::AppError; +use crate::http::server::common::error::AppError; use http_body_util::BodyExt; use hyper::body::Incoming; use serde::de::DeserializeOwned; diff --git a/core/src/http/server/error.rs b/core/src/http/server/common/error.rs similarity index 94% rename from core/src/http/server/error.rs rename to core/src/http/server/common/error.rs index 18227d3d..e4264416 100644 --- a/core/src/http/server/error.rs +++ b/core/src/http/server/common/error.rs @@ -1,5 +1,5 @@ use crate::http::dto::ErrorResponse; -use crate::http::server::response::{BoxedBody, JsonResponse}; +use crate::http::server::common::response::{BoxedBody, JsonResponse}; use hyper::{Response, StatusCode}; #[derive(Debug, thiserror::Error)] diff --git a/core/src/http/server/common/mod.rs b/core/src/http/server/common/mod.rs new file mode 100644 index 00000000..2a0d9946 --- /dev/null +++ b/core/src/http/server/common/mod.rs @@ -0,0 +1,8 @@ +pub mod client_cert_verifier; +pub mod collect_to_json; +pub mod error; +pub mod pin; +pub mod query; +pub mod response; +pub mod save; +pub mod session; diff --git a/core/src/http/server/controller/mod.rs b/core/src/http/server/common/pin.rs similarity index 93% rename from core/src/http/server/controller/mod.rs rename to core/src/http/server/common/pin.rs index f34074ee..2353cf77 100644 --- a/core/src/http/server/controller/mod.rs +++ b/core/src/http/server/common/pin.rs @@ -1,8 +1,4 @@ -pub(crate) mod v2; -pub(crate) mod v3; -pub(crate) mod web; - -use crate::http::server::error::AppError; +use crate::http::server::common::error::AppError; use hyper::StatusCode; use lru::LruCache; use std::collections::HashMap; diff --git a/core/src/http/server/query.rs b/core/src/http/server/common/query.rs similarity index 100% rename from core/src/http/server/query.rs rename to core/src/http/server/common/query.rs diff --git a/core/src/http/server/response.rs b/core/src/http/server/common/response.rs similarity index 100% rename from core/src/http/server/response.rs rename to core/src/http/server/common/response.rs diff --git a/core/src/http/server/common/save.rs b/core/src/http/server/common/save.rs new file mode 100644 index 00000000..dffbcefc --- /dev/null +++ b/core/src/http/server/common/save.rs @@ -0,0 +1,183 @@ +use bytes::Bytes; +use http_body_util::BodyExt; +use hyper::body::Incoming; +use hyper::Request; +use std::future::Future; +use std::path::PathBuf; +use tokio::sync::{mpsc, oneshot}; + +/// Channel capacity for file upload chunks (provides backpressure). +const UPLOAD_CHANNEL_CAPACITY: usize = 16; + +/// Where the content of an uploaded file should go, decided by the application. +#[derive(Debug)] +pub enum FileUploadTarget { + /// The application consumes the binary chunks itself. + /// + /// The server forwards chunks into `binary_tx` and closes it at end of file. + /// The application should compare the number of received bytes with `file.size` + /// and report the result on the sender side of `result_rx` which determines + /// the HTTP response (200 on `Ok`, 500 on `Err` or when the sender is dropped). + Stream { + /// Channel the server sends the binary chunks of the file into. + binary_tx: mpsc::Sender, + + /// Channel on which the application reports whether the file was + /// processed successfully. + result_rx: oneshot::Receiver>, + }, + + /// The server writes the file to this path (created or truncated) + /// and reports the result on `result_tx`. + Path { + /// The path to write the file to. + path: PathBuf, + + /// Channel on which the server reports whether the file was saved successfully. + result_tx: oneshot::Sender>, + }, + + /// The server writes the file to this raw file descriptor (Android only) + /// and reports the result on `result_tx`. + #[cfg(target_os = "android")] + Fd { + /// The raw file descriptor to write the file to. + /// Ownership is transferred; the descriptor is closed after writing. + fd: std::os::fd::RawFd, + + /// Channel on which the server reports whether the file was saved successfully. + result_tx: oneshot::Sender>, + }, +} + +pub(crate) async fn save_req_to_target( + req: Request, + target: FileUploadTarget, + file_size: u64, +) -> bool { + // Resolve the target into a chunk sender and a result receiver. + let (binary_tx, result_rx) = match target { + FileUploadTarget::Stream { + binary_tx, + result_rx, + } => (binary_tx, result_rx), + FileUploadTarget::Path { path, result_tx } => spawn_file_writer( + async move { + tokio::fs::File::create(&path) + .await + .map_err(|e| format!("Failed to create {}: {e}", path.display())) + }, + file_size, + result_tx, + ), + #[cfg(target_os = "android")] + FileUploadTarget::Fd { fd, result_tx } => spawn_file_writer( + async move { + use std::os::fd::FromRawFd; + + // SAFETY: the descriptor is owned by this transfer; wrapping it in + // a File transfers that ownership so it is closed once writing finishes. + let std_file = unsafe { std::fs::File::from_raw_fd(fd) }; + Ok(tokio::fs::File::from_std(std_file)) + }, + file_size, + result_tx, + ), + }; + + // Forward the request body to the target. + let mut body = req.into_body(); + let mut stream_error = false; + while let Some(frame) = body.frame().await { + match frame { + Ok(frame) => { + let Ok(data) = frame.into_data() else { + continue; // ignore non-data frames (e.g. trailers) + }; + if data.is_empty() { + continue; + } + if binary_tx.send(data).await.is_err() { + // The receiver is gone (dropped by the application or + // closed by the file writer after an error). + stream_error = true; + break; + } + } + Err(err) => { + tracing::warn!("Error reading upload body of file: {err:#}"); + stream_error = true; + break; + } + } + } + + // Signal end of file to the receiving side. + drop(binary_tx); + + match stream_error { + true => false, + false => match result_rx.await { + Ok(Ok(())) => true, + Ok(Err(err)) => { + tracing::warn!("Failed to process file: {err}"); + false + } + Err(_) => false, + }, + } +} + +/// Spawns a task that writes incoming chunks to a file provided by `open`. +/// +/// Returns the sender for the binary chunks and a receiver for the final result. +/// The result is additionally reported to the application on `result_tx`. +fn spawn_file_writer( + open: impl Future> + Send + 'static, + expected_size: u64, + result_tx: oneshot::Sender>, +) -> (mpsc::Sender, oneshot::Receiver>) { + let (binary_tx, mut binary_rx) = mpsc::channel::(UPLOAD_CHANNEL_CAPACITY); + let (internal_tx, internal_rx) = oneshot::channel::>(); + + tokio::spawn(async move { + let result = write_file_from_receiver(open, expected_size, &mut binary_rx).await; + // Unblock the request handler if it is still sending chunks. + binary_rx.close(); + let _ = result_tx.send(result.clone()); + let _ = internal_tx.send(result); + }); + + (binary_tx, internal_rx) +} + +/// Writes all chunks received on `rx` to the file provided by `open`. +/// +/// Fails if the total number of written bytes does not match `expected_size` +/// (e.g. the sender disconnected mid-transfer). +async fn write_file_from_receiver( + open: impl Future>, + expected_size: u64, + rx: &mut mpsc::Receiver, +) -> Result<(), String> { + use tokio::io::AsyncWriteExt; + + let mut file = open.await?; + let mut written: u64 = 0; + while let Some(chunk) = rx.recv().await { + file.write_all(&chunk) + .await + .map_err(|e| format!("Failed to write file: {e}"))?; + written += chunk.len() as u64; + } + file.flush() + .await + .map_err(|e| format!("Failed to flush file: {e}"))?; + + if written != expected_size { + return Err(format!( + "Expected {expected_size} bytes, received {written}" + )); + } + Ok(()) +} diff --git a/core/src/http/server/session.rs b/core/src/http/server/common/session.rs similarity index 100% rename from core/src/http/server/session.rs rename to core/src/http/server/common/session.rs diff --git a/core/src/http/server/mod.rs b/core/src/http/server/mod.rs index 2e09021f..68a4236a 100644 --- a/core/src/http/server/mod.rs +++ b/core/src/http/server/mod.rs @@ -1,23 +1,17 @@ -mod client_cert_verifier; -mod collect_to_json; -mod controller; -mod error; -mod query; -mod response; -mod session; - -pub use crate::http::server::controller::v2::{ - FileUploadTargetV2, PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2, -}; -pub use crate::http::server::controller::web::{WebSendConfig, WebSendEvent, WebSendI18n}; +pub mod common; +pub mod v2; +pub mod v3; +pub mod web; use crate::crypto::cert::public_key_from_cert_der; -use crate::http::server::client_cert_verifier::CustomClientCertVerifier; -use crate::http::server::controller::web::WebPageState; -use crate::http::server::error::AppError; -use crate::http::server::response::BoxedBody; -use crate::http::server::session::SessionStateV2; +use crate::http::server::v2::ServerEventV2; +use crate::http::server::web::WebSendConfig; use crate::http::state::ClientInfo; +use common::client_cert_verifier::CustomClientCertVerifier; +use common::error::AppError; +use common::response; +use common::response::BoxedBody; +use common::session::SessionStateV2; use hyper::body::Incoming; use hyper::{Method, Request, Response, StatusCode}; use hyper_util::rt::{TokioExecutor, TokioIo}; @@ -31,6 +25,7 @@ use std::num::NonZeroUsize; use std::ops::Deref; use std::sync::Arc; use tokio::sync::{mpsc, oneshot, Mutex}; +use web::WebPageState; /// Configuration for the v2 (legacy) protocol endpoints. pub struct ServerConfigV2 { @@ -57,7 +52,7 @@ pub(crate) struct V2State { } #[derive(Clone)] -struct AppState { +pub struct AppState { /// Information about server's device. info: Arc>, @@ -277,7 +272,7 @@ fn create_tls_config(tls_config: &TlsConfig) -> anyhow::Result) -> Result Ok(controller::web::index(&state)), - (&Method::GET, "/main.js") => Ok(controller::web::main_js(&state)), - (&Method::GET, "/i18n.json") => controller::web::i18n(&state), + (&Method::GET, "/") => Ok(web::index(&state)), + (&Method::GET, "/main.js") => Ok(web::main_js(&state)), + (&Method::GET, "/i18n.json") => web::i18n(&state), (&Method::POST, "/api/localsend/v2/prepare-download") => { - controller::web::prepare_download(req, state, client_info).await + web::prepare_download(req, state, client_info).await } (&Method::GET, "/api/localsend/v2/download") => { - controller::web::download(req, state, client_info).await + web::download(req, state, client_info).await } (&Method::POST, "/api/localsend/v2/register") => { if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - Ok( - controller::v2::register(req.into_body(), state, client_info) - .await? - .into_response(), - ) + Ok(v2::register(req.into_body(), state, client_info) + .await? + .into_response()) } (&Method::GET, "/api/localsend/v2/info") => { if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - Ok(controller::v2::info(state).await?.into_response()) + Ok(v2::info(state).await?.into_response()) } (&Method::POST, "/api/localsend/v2/prepare-upload") => { if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - controller::v2::prepare_upload(req, state, client_info).await + v2::prepare_upload(req, state, client_info).await } (&Method::POST, "/api/localsend/v2/upload") => { if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - controller::v2::upload(req, state, client_info).await + v2::upload(req, state, client_info).await } (&Method::POST, "/api/localsend/v2/cancel") => { if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - controller::v2::cancel(req, state).await + v2::cancel(req, state).await } (&Method::POST, "/api/localsend/v3/nonce") => { - Ok( - controller::v3::nonce_exchange(req.into_body(), state, client_info) - .await? - .into_response(), - ) + Ok(v3::nonce_exchange(req.into_body(), state, client_info) + .await? + .into_response()) } (&Method::POST, "/api/localsend/v3/register") => { - Ok( - controller::v3::register(req.into_body(), state, client_info) - .await? - .into_response(), - ) + Ok(v3::register(req.into_body(), state, client_info) + .await? + .into_response()) } _ => { let mut res = Response::new(response::empty_body()); diff --git a/core/src/http/server/controller/v2.rs b/core/src/http/server/v2.rs similarity index 68% rename from core/src/http/server/controller/v2.rs rename to core/src/http/server/v2.rs index 4eb5b1fa..cbb36844 100644 --- a/core/src/http/server/controller/v2.rs +++ b/core/src/http/server/v2.rs @@ -2,24 +2,23 @@ use crate::http::dto_v2::{ InfoResponseDtoV2, PrepareUploadRequestDtoV2, PrepareUploadResponseDtoV2, RegisterDtoV2, RegisterResponseDtoV2, PROTOCOL_VERSION_V2, }; -use crate::http::server::collect_to_json::CollectToJson; -use crate::http::server::controller::check_pin; -use crate::http::server::error::AppError; -use crate::http::server::query::parse_query; -use crate::http::server::response::{empty_body, BoxedBody, JsonResponse}; -use crate::http::server::session::{FileStatusV2, SessionFileV2, SessionStateV2, UploadSessionV2}; -use crate::http::server::{AppState, RequestClientInfo, V2State}; +use crate::http::server::common::collect_to_json::CollectToJson; +use crate::http::server::common::error::AppError; +use crate::http::server::common::pin::check_pin; +use crate::http::server::common::query::parse_query; +use crate::http::server::common::response::{empty_body, BoxedBody, JsonResponse}; +use crate::http::server::common::save::FileUploadTarget; +use crate::http::server::common::session::{ + FileStatusV2, SessionFileV2, SessionStateV2, UploadSessionV2, +}; +use crate::http::server::{common, AppState, RequestClientInfo, V2State}; use crate::model::transfer::FileDto; -use bytes::Bytes; -use http_body_util::BodyExt; use hyper::body::Incoming; use hyper::{Request, Response, StatusCode}; use std::collections::{HashMap, HashSet}; -use std::future::Future; use std::net::IpAddr; -use std::path::PathBuf; use std::sync::Arc; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::oneshot; use uuid::Uuid; /// Events emitted by the v2 HTTP server that must be handled by the application. @@ -68,7 +67,7 @@ pub enum ServerEventV2 { file: FileDto, /// Channel to send the target the file content should be written to. - target_tx: oneshot::Sender, + target_tx: oneshot::Sender, }, /// An upload session ended. @@ -81,47 +80,6 @@ pub enum ServerEventV2 { }, } -/// Where the content of an uploaded file should go, decided by the application. -#[derive(Debug)] -pub enum FileUploadTargetV2 { - /// The application consumes the binary chunks itself. - /// - /// The server forwards chunks into `binary_tx` and closes it at end of file. - /// The application should compare the number of received bytes with `file.size` - /// and report the result on the sender side of `result_rx` which determines - /// the HTTP response (200 on `Ok`, 500 on `Err` or when the sender is dropped). - Stream { - /// Channel the server sends the binary chunks of the file into. - binary_tx: mpsc::Sender, - - /// Channel on which the application reports whether the file was - /// processed successfully. - result_rx: oneshot::Receiver>, - }, - - /// The server writes the file to this path (created or truncated) - /// and reports the result on `result_tx`. - Path { - /// The path to write the file to. - path: PathBuf, - - /// Channel on which the server reports whether the file was saved successfully. - result_tx: oneshot::Sender>, - }, - - /// The server writes the file to this raw file descriptor (Android only) - /// and reports the result on `result_tx`. - #[cfg(target_os = "android")] - Fd { - /// The raw file descriptor to write the file to. - /// Ownership is transferred; the descriptor is closed after writing. - fd: std::os::fd::RawFd, - - /// Channel on which the server reports whether the file was saved successfully. - result_tx: oneshot::Sender>, - }, -} - /// The application's decision for a prepare-upload request. #[derive(Debug)] pub enum PrepareUploadDecisionV2 { @@ -143,9 +101,6 @@ pub enum SessionEndReasonV2 { Cancelled, } -/// Channel capacity for file upload chunks (provides backpressure). -const UPLOAD_CHANNEL_CAPACITY: usize = 16; - pub(crate) async fn register( body: Incoming, state: AppState, @@ -348,7 +303,7 @@ pub(crate) async fn upload( let mut upload_guard = UploadGuard::new(v2.clone(), session_id.clone(), file_id.clone()); let file_size = file_dto.size; - let (target_tx, target_rx) = oneshot::channel::(); + let (target_tx, target_rx) = oneshot::channel::(); let event = ServerEventV2::FileUpload { session_id: session_id.clone(), @@ -366,77 +321,7 @@ pub(crate) async fn upload( return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)); }; - // Resolve the target into a chunk sender and a result receiver. - let (binary_tx, result_rx) = match target { - FileUploadTargetV2::Stream { - binary_tx, - result_rx, - } => (binary_tx, result_rx), - FileUploadTargetV2::Path { path, result_tx } => spawn_file_writer( - async move { - tokio::fs::File::create(&path) - .await - .map_err(|e| format!("Failed to create {}: {e}", path.display())) - }, - file_size, - result_tx, - ), - #[cfg(target_os = "android")] - FileUploadTargetV2::Fd { fd, result_tx } => spawn_file_writer( - async move { - use std::os::fd::FromRawFd; - - // SAFETY: the descriptor is owned by this transfer; wrapping it in - // a File transfers that ownership so it is closed once writing finishes. - let std_file = unsafe { std::fs::File::from_raw_fd(fd) }; - Ok(tokio::fs::File::from_std(std_file)) - }, - file_size, - result_tx, - ), - }; - - // Forward the request body to the target. - let mut body = req.into_body(); - let mut stream_error = false; - while let Some(frame) = body.frame().await { - match frame { - Ok(frame) => { - let Ok(data) = frame.into_data() else { - continue; // ignore non-data frames (e.g. trailers) - }; - if data.is_empty() { - continue; - } - if binary_tx.send(data).await.is_err() { - // The receiver is gone (dropped by the application or - // closed by the file writer after an error). - stream_error = true; - break; - } - } - Err(err) => { - tracing::warn!("Error reading upload body of file {file_id}: {err:#}"); - stream_error = true; - break; - } - } - } - - // Signal end of file to the receiving side. - drop(binary_tx); - - let success = match stream_error { - true => false, - false => match result_rx.await { - Ok(Ok(())) => true, - Ok(Err(err)) => { - tracing::warn!("Failed to process file {file_id}: {err}"); - false - } - Err(_) => false, - }, - }; + let success = common::save::save_req_to_target(req, target, file_size).await; upload_guard.finish(success).await; @@ -446,60 +331,6 @@ pub(crate) async fn upload( } } -/// Spawns a task that writes incoming chunks to a file provided by `open`. -/// -/// Returns the sender for the binary chunks and a receiver for the final result. -/// The result is additionally reported to the application on `result_tx`. -fn spawn_file_writer( - open: impl Future> + Send + 'static, - expected_size: u64, - result_tx: oneshot::Sender>, -) -> (mpsc::Sender, oneshot::Receiver>) { - let (binary_tx, mut binary_rx) = mpsc::channel::(UPLOAD_CHANNEL_CAPACITY); - let (internal_tx, internal_rx) = oneshot::channel::>(); - - tokio::spawn(async move { - let result = write_file_from_receiver(open, expected_size, &mut binary_rx).await; - // Unblock the request handler if it is still sending chunks. - binary_rx.close(); - let _ = result_tx.send(result.clone()); - let _ = internal_tx.send(result); - }); - - (binary_tx, internal_rx) -} - -/// Writes all chunks received on `rx` to the file provided by `open`. -/// -/// Fails if the total number of written bytes does not match `expected_size` -/// (e.g. the sender disconnected mid-transfer). -async fn write_file_from_receiver( - open: impl Future>, - expected_size: u64, - rx: &mut mpsc::Receiver, -) -> Result<(), String> { - use tokio::io::AsyncWriteExt; - - let mut file = open.await?; - let mut written: u64 = 0; - while let Some(chunk) = rx.recv().await { - file.write_all(&chunk) - .await - .map_err(|e| format!("Failed to write file: {e}"))?; - written += chunk.len() as u64; - } - file.flush() - .await - .map_err(|e| format!("Failed to flush file: {e}"))?; - - if written != expected_size { - return Err(format!( - "Expected {expected_size} bytes, received {written}" - )); - } - Ok(()) -} - pub(crate) async fn cancel( req: Request, state: AppState, diff --git a/core/src/http/server/controller/v3.rs b/core/src/http/server/v3.rs similarity index 92% rename from core/src/http/server/controller/v3.rs rename to core/src/http/server/v3.rs index cd9125cd..408e5930 100644 --- a/core/src/http/server/controller/v3.rs +++ b/core/src/http/server/v3.rs @@ -1,7 +1,7 @@ use crate::http::dto::{NonceRequest, NonceResponse, RegisterDto, RegisterResponseDto}; -use crate::http::server::collect_to_json::CollectToJson; -use crate::http::server::error::AppError; -use crate::http::server::response::JsonResponse; +use crate::http::server::common::collect_to_json::CollectToJson; +use crate::http::server::common::error::AppError; +use crate::http::server::common::response::JsonResponse; use crate::http::server::{AppState, RequestClientInfo}; use crate::{crypto, util}; use hyper::body::Incoming; diff --git a/core/src/http/server/controller/web.rs b/core/src/http/server/web.rs similarity index 96% rename from core/src/http/server/controller/web.rs rename to core/src/http/server/web.rs index 6a33aeea..24d0a2bc 100644 --- a/core/src/http/server/controller/web.rs +++ b/core/src/http/server/web.rs @@ -1,8 +1,8 @@ use crate::http::dto_v2::{InfoResponseDtoV2, PrepareDownloadResponseDtoV2, PROTOCOL_VERSION_V2}; -use crate::http::server::controller::check_pin; -use crate::http::server::error::AppError; -use crate::http::server::query::parse_query; -use crate::http::server::response::{full_body, BoxedBody, JsonResponse}; +use crate::http::server::common::error::AppError; +use crate::http::server::common::pin::check_pin; +use crate::http::server::common::query::parse_query; +use crate::http::server::common::response::{full_body, BoxedBody, JsonResponse}; use crate::http::server::{AppState, RequestClientInfo}; use crate::model::transfer::{FileContent, FileDto}; use bytes::Bytes; @@ -65,9 +65,9 @@ pub enum WebSendEvent { }, } -const INDEX_HTML: &str = include_str!("../../../../assets/web/index.html"); -const MAIN_JS: &str = include_str!("../../../../assets/web/main.js"); -const ERROR_403_HTML: &str = include_str!("../../../../assets/web/error-403.html"); +const INDEX_HTML: &str = include_str!("../../../assets/web/index.html"); +const MAIN_JS: &str = include_str!("../../../assets/web/main.js"); +const ERROR_403_HTML: &str = include_str!("../../../assets/web/error-403.html"); /// Characters that are percent-encoded in the content-disposition file name. /// Matches the component encoding of RFC 2396 (letters, digits and marks are kept). @@ -415,7 +415,7 @@ async fn file_list_response( } /// Streams application-provided chunks as a response body. -fn receiver_stream_body(binary_rx: mpsc::Receiver>) -> BoxedBody { +fn receiver_stream_body(binary_rx: mpsc::Receiver) -> BoxedBody { let stream = ReceiverStream::new(binary_rx) .map(|chunk| Ok::<_, std::io::Error>(Frame::data(Bytes::from(chunk)))); StreamBody::new(stream).boxed() diff --git a/core/src/main.rs b/core/src/main.rs index 1be9bdb2..8366065c 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -7,7 +7,8 @@ mod webrtc; use crate::crypto::token; use crate::http::client::LsHttpClientV3; use crate::http::dto::{PrepareUploadRequestDto, ProtocolType, RegisterDto}; -use crate::http::server::{FileUploadTargetV2, PrepareUploadDecisionV2, ServerEventV2}; +use crate::http::server::common::save::FileUploadTarget; +use crate::http::server::v2::{PrepareUploadDecisionV2, ServerEventV2}; use crate::http::server::{ServerConfigV2, TlsConfig}; use crate::model::discovery::DeviceType; use crate::webrtc::signaling::{ClientInfo, WsServerMessage}; @@ -169,7 +170,7 @@ async fn server_test() -> Result<()> { } => { let (binary_tx, mut binary_rx) = mpsc::channel::(16); let (result_tx, result_rx) = oneshot::channel::>(); - let _ = target_tx.send(FileUploadTargetV2::Stream { + let _ = target_tx.send(FileUploadTarget::Stream { binary_tx, result_rx, }); diff --git a/core/src/model/transfer.rs b/core/src/model/transfer.rs index 976f5f3a..b3f1be20 100644 --- a/core/src/model/transfer.rs +++ b/core/src/model/transfer.rs @@ -1,3 +1,4 @@ +use bytes::Bytes; use serde::{Deserialize, Serialize}; use std::path::PathBuf; use tokio::sync::mpsc; @@ -14,7 +15,7 @@ const FILE_CHANNEL_CAPACITY: usize = 16; pub enum FileContent { /// A stream of binary chunks. The channel is closed once the file has been /// fully provided. - Stream(mpsc::Receiver>), + Stream(mpsc::Receiver), /// A path to a regular file the content is read from. Path(PathBuf), @@ -30,7 +31,7 @@ impl FileContent { /// [`FileContent::Stream`] is returned as-is. For [`FileContent::Path`] and /// [`FileContent::Fd`], a background task reads the file and forwards the /// chunks; the channel is closed on EOF or on an I/O error. - pub fn into_receiver(self) -> mpsc::Receiver> { + pub fn into_receiver(self) -> mpsc::Receiver { match self { FileContent::Stream(rx) => rx, FileContent::Path(path) => { @@ -64,15 +65,15 @@ impl FileContent { /// Reads `file` to EOF, forwarding chunks on `tx`. /// /// Stops early if the receiver is gone or a read error occurs. -async fn read_file_into_sender(mut file: tokio::fs::File, tx: mpsc::Sender>) { +async fn read_file_into_sender(mut file: tokio::fs::File, tx: mpsc::Sender) { use tokio::io::AsyncReadExt; - let mut buffer = vec![0u8; 64 * 1024]; + let mut buffer = bytes::BytesMut::with_capacity(64 * 1024); loop { - match file.read(&mut buffer).await { + match file.read_buf(&mut buffer).await { Ok(0) => break, - Ok(n) => { - if tx.send(buffer[..n].to_vec()).await.is_err() { + Ok(_) => { + if tx.send(buffer.split().freeze()).await.is_err() { break; } } diff --git a/core/tests/v2_server.rs b/core/tests/v2_server.rs index e10f89e5..5fb249df 100644 --- a/core/tests/v2_server.rs +++ b/core/tests/v2_server.rs @@ -1,12 +1,12 @@ #![cfg(feature = "http")] +use bytes::Bytes; use localsend::http::client::{ClientError, LsHttpClientV2}; use localsend::http::dto::ProtocolType; use localsend::http::dto_v2::{PrepareUploadRequestDtoV2, ProtocolTypeV2, RegisterDtoV2}; +use localsend::http::server::common::save::FileUploadTarget; +use localsend::http::server::v2::{PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2}; use localsend::http::server::{start_with_port, ServerConfigV2}; -use localsend::http::server::{ - FileUploadTargetV2, PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2, -}; use localsend::http::state::ClientInfo; use localsend::model::transfer::{FileContent, FileDto}; use std::collections::HashMap; @@ -69,7 +69,7 @@ async fn start_test_server( None => { let (binary_tx, mut binary_rx) = mpsc::channel(16); let (result_tx, result_rx) = oneshot::channel(); - let _ = target_tx.send(FileUploadTargetV2::Stream { + let _ = target_tx.send(FileUploadTarget::Stream { binary_tx, result_rx, }); @@ -85,7 +85,7 @@ async fn start_test_server( Some(dir) => { let path = dir.join(&file_id); let (result_tx, result_rx) = oneshot::channel(); - let _ = target_tx.send(FileUploadTargetV2::Path { + let _ = target_tx.send(FileUploadTarget::Path { path: path.clone(), result_tx, }); @@ -206,11 +206,11 @@ async fn upload_bytes( token: &str, bytes: &[u8], ) -> Result<(), ClientError> { - let (tx, rx) = mpsc::channel::>(4); + let (tx, rx) = mpsc::channel::(4); let chunks: Vec> = bytes.chunks(1024).map(|chunk| chunk.to_vec()).collect(); tokio::spawn(async move { for chunk in chunks { - if tx.send(chunk).await.is_err() { + if tx.send(Bytes::from(chunk)).await.is_err() { break; } } diff --git a/core/tests/v2_web_send.rs b/core/tests/v2_web_send.rs index adc8013d..e2b05f90 100644 --- a/core/tests/v2_web_send.rs +++ b/core/tests/v2_web_send.rs @@ -3,8 +3,10 @@ use bytes::Bytes; use localsend::http::client::{ClientError, LsHttpClientV2}; use localsend::http::dto::ProtocolType; -use localsend::http::server::{start_with_port, ServerConfigV2, WebSendConfig, WebSendI18n}; -use localsend::http::server::{ServerEventV2, WebSendEvent}; +use localsend::http::server::v2::ServerEventV2; +use localsend::http::server::web::WebSendConfig; +use localsend::http::server::web::{WebSendEvent, WebSendI18n}; +use localsend::http::server::{start_with_port, ServerConfigV2}; use localsend::http::state::ClientInfo; use localsend::model::transfer::{FileContent, FileDto}; use std::collections::HashMap; @@ -34,10 +36,10 @@ enum TestFileContent { impl TestFileContent { /// Streams the content into `tx`, mimicking how an application would /// serve in-memory content or a file from disk. - async fn stream(self, tx: mpsc::Sender>) { + async fn stream(self, tx: mpsc::Sender) { match self { TestFileContent::Bytes(bytes) => { - let _ = tx.send(bytes.to_vec()).await; + let _ = tx.send(bytes).await; } TestFileContent::Path(path) => { let mut file = tokio::fs::File::open(&path) @@ -49,7 +51,11 @@ impl TestFileContent { if bytes_read == 0 { break; // EOF } - if tx.send(buffer[..bytes_read].to_vec()).await.is_err() { + if tx + .send(Bytes::copy_from_slice(&buffer[..bytes_read])) + .await + .is_err() + { break; // client disconnected } } @@ -93,7 +99,7 @@ async fn start_test_server( .expect("FileDownload for unknown file") .clone(); tokio::spawn(async move { - let (tx, rx) = mpsc::channel::>(16); + let (tx, rx) = mpsc::channel::(16); if content_tx.send(FileContent::Stream(rx)).is_err() { return; }