opensource/localsend
1
0
Fork 0
mirror of https://github.com/localsend/localsend.git synced 2026-08-07 07:14:52 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

feat: allow Rust to read file content directly

Browse Source
This commit is contained in:
Tien Do Nam
2026-07-13 20:31:27 +02:00
parent 91ac619029
commit 1c6728f71c
15 changed files with 410 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/crypto/cert.rs
+1 -2
View File
@@ -180,8 +180,7 @@ nidU/qXQvBJ7NPUkXXgbcgqxK735iijOqQHmKts=
-----END CERTIFICATE-----"
.to_string();
assert_eq!(
verify_cert_from_pem(cert_expired, Some(PUBLIC_KEY))
.map_err(|e| e.to_string()),
verify_cert_from_pem(cert_expired, Some(PUBLIC_KEY)).map_err(|e| e.to_string()),
Err("Time validity error".to_string())
);
}
core/src/http/client/mod.rs
+4 -4
View File
@@ -6,7 +6,7 @@ pub use v2::LsHttpClientV2;
pub use v3::LsHttpClientV3;
use crate::http::StatusCodeError;
use crate::{crypto, http};
use crate::{crypto, http, model};
use reqwest::Response;
use serde::{Deserialize, Serialize};
use thiserror::Error;
@@ -112,21 +112,21 @@ impl LsHttpClient {
session_id: &str,
file_id: &str,
token: &str,
binary: tokio::sync::mpsc::Receiver<Vec<u8>>,
content: model::transfer::FileContent,
cancel: tokio_util::sync::CancellationToken,
) -> Result<(), ClientError> {
match self {
LsHttpClient::V2(client) => {
client
.upload(
protocol, ip, port, public_key, session_id, file_id, token, binary, cancel,
protocol, ip, port, public_key, session_id, file_id, token, content, cancel,
)
.await
}
LsHttpClient::V3(client) => {
client
.upload(
protocol, ip, port, public_key, session_id, file_id, token, binary, cancel,
protocol, ip, port, public_key, session_id, file_id, token, content, cancel,
)
.await
}
core/src/http/client/v2.rs
+5 -5
View File
@@ -5,12 +5,12 @@ use crate::http::dto_v2::{
InfoResponseDtoV2, PrepareDownloadResponseDtoV2, PrepareUploadRequestDtoV2,
PrepareUploadResponseDtoV2, PrepareUploadResultV2, RegisterDtoV2, RegisterResponseDtoV2,
};
use crate::model;
use futures_util::StreamExt;
use reqwest::{Response, StatusCode};
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;
/// HTTP client for LocalSend Protocol v2.1.
pub struct LsHttpClientV2 {
@@ -193,7 +193,7 @@ impl LsHttpClientV2 {
/// * `session_id` - Session ID from prepare_upload
/// * `file_id` - File ID to upload
/// * `token` - File-specific token from prepare_upload
/// * `binary` - Channel receiving file chunks
/// * `content` - The file content to upload (a chunk stream or a raw file descriptor)
/// * `cancel` - Cancellation token; cancelling it aborts the upload with [`ClientError::Cancelled`]
///
/// # Errors
@@ -210,7 +210,7 @@ impl LsHttpClientV2 {
session_id: &str,
file_id: &str,
token: &str,
binary: mpsc::Receiver<Vec<u8>>,
content: model::transfer::FileContent,
cancel: CancellationToken,
) -> Result<(), ClientError> {
let url = TargetUrl {
@@ -227,7 +227,7 @@ impl LsHttpClientV2 {
}
.to_string();
let stream = ReceiverStream::new(binary).map(Ok::<Vec<u8>, anyhow::Error>);
let stream = ReceiverStream::new(content.into_receiver()).map(Ok::<Vec<u8>, anyhow::Error>);
let body = reqwest::Body::wrap_stream(stream);
let res = tokio::select! {
core/src/http/client/v3.rs
+5 -4
View File
@@ -1,14 +1,14 @@
use super::{ClientError, ResponseExt, ResultWithPublicKey};
use crate::http;
use crate::http::client::url::{ApiVersion, TargetUrl};
use crate::http::dto::ProtocolType;
use crate::{crypto, util};
use crate::{http, model};
use futures_util::StreamExt;
use lru::LruCache;
use reqwest::{Response, StatusCode};
use std::num::NonZeroUsize;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tokio::sync::Mutex;
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::sync::CancellationToken;
@@ -193,7 +193,7 @@ impl LsHttpClientV3 {
session_id: &str,
file_id: &str,
token: &str,
binary: mpsc::Receiver<Vec<u8>>,
content: model::transfer::FileContent,
cancel: CancellationToken,
) -> Result<(), ClientError> {
let send = self
@@ -214,7 +214,8 @@ impl LsHttpClientV3 {
.to_string(),
)
.body({
let stream = ReceiverStream::new(binary).map(Ok::<Vec<u8>, anyhow::Error>);
let stream =
ReceiverStream::new(content.into_receiver()).map(Ok::<Vec<u8>, anyhow::Error>);
reqwest::Body::wrap_stream(stream)
})
.send();
core/src/http/server/client_cert_verifier.rs
+4 -4
View File
@@ -1,11 +1,11 @@
use std::fmt::{Debug, Formatter};
use std::sync::Arc;
use rustls::pki_types::pem::PemObject;
use rustls::{DigitallySignedStruct, DistinguishedName, Error, RootCertStore, SignatureScheme};
use rustls::client::danger::HandshakeSignatureValid;
use rustls::pki_types::pem::PemObject;
use rustls::pki_types::{CertificateDer, UnixTime};
use rustls::server::danger::{ClientCertVerified, ClientCertVerifier};
use rustls::server::WebPkiClientVerifier;
use rustls::{DigitallySignedStruct, DistinguishedName, Error, RootCertStore, SignatureScheme};
use std::fmt::{Debug, Formatter};
use std::sync::Arc;
use x509_parser::nom::AsBytes;
/// Enables client certificate verification.
core/src/http/server/controller/v2.rs
+145 -17
View File
@@ -15,7 +15,9 @@ 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 uuid::Uuid;
@@ -52,10 +54,9 @@ pub enum ServerEventV2 {
/// An accepted file is being uploaded via `POST /api/localsend/v2/upload`.
///
/// Binary chunks arrive on `binary_rx` until the channel is closed.
/// The application should compare the number of received bytes with `file.size`
/// and report the result on `result_tx` which determines the HTTP response
/// (200 on `Ok`, 500 on `Err` or when `result_tx` is dropped).
/// The application must answer on `target_tx` with where the file content
/// should go (a stream to consume itself, a path, or a file descriptor).
/// Dropping `target_tx` results in a 500 response.
FileUpload {
/// The session ID of the upload session.
session_id: String,
@@ -66,11 +67,8 @@ pub enum ServerEventV2 {
/// The metadata of the file being uploaded.
file: FileDto,
/// Channel receiving the binary chunks of the file.
binary_rx: mpsc::Receiver<Bytes>,
/// Channel to report whether the file was processed successfully.
result_tx: oneshot::Sender<Result<(), String>>,
/// Channel to send the target the file content should be written to.
target_tx: oneshot::Sender<FileUploadTargetV2>,
},
/// An upload session ended.
@@ -83,6 +81,47 @@ 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<Bytes>,
/// Channel on which the application reports whether the file was
/// processed successfully.
result_rx: oneshot::Receiver<Result<(), String>>,
},
/// 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<Result<(), String>>,
},
/// 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<Result<(), String>>,
},
}
/// The application's decision for a prepare-upload request.
#[derive(Debug)]
pub enum PrepareUploadDecisionV2 {
@@ -308,22 +347,56 @@ pub(crate) async fn upload(
// Marks the file as failed if this request is aborted mid-transfer.
let mut upload_guard = UploadGuard::new(v2.clone(), session_id.clone(), file_id.clone());
let (binary_tx, binary_rx) = mpsc::channel::<Bytes>(UPLOAD_CHANNEL_CAPACITY);
let (result_tx, result_rx) = oneshot::channel::<Result<(), String>>();
let file_size = file_dto.size;
let (target_tx, target_rx) = oneshot::channel::<FileUploadTargetV2>();
let event = ServerEventV2::FileUpload {
session_id: session_id.clone(),
file_id: file_id.clone(),
file: file_dto,
binary_rx,
result_tx,
target_tx,
};
if v2.event_tx.send(event).await.is_err() {
upload_guard.finish(false).await;
return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR));
}
// Forward the request body to the application.
let Ok(target) = target_rx.await else {
upload_guard.finish(false).await;
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 {
@@ -336,7 +409,8 @@ pub(crate) async fn upload(
continue;
}
if binary_tx.send(data).await.is_err() {
// The application dropped the receiver.
// The receiver is gone (dropped by the application or
// closed by the file writer after an error).
stream_error = true;
break;
}
@@ -349,7 +423,7 @@ pub(crate) async fn upload(
}
}
// Signal end of file to the application.
// Signal end of file to the receiving side.
drop(binary_tx);
let success = match stream_error {
@@ -357,7 +431,7 @@ pub(crate) async fn upload(
false => match result_rx.await {
Ok(Ok(())) => true,
Ok(Err(err)) => {
tracing::warn!("Application failed to process file {file_id}: {err}");
tracing::warn!("Failed to process file {file_id}: {err}");
false
}
Err(_) => false,
@@ -372,6 +446,60 @@ 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<Output = Result<tokio::fs::File, String>> + Send + 'static,
expected_size: u64,
result_tx: oneshot::Sender<Result<(), String>>,
) -> (mpsc::Sender<Bytes>, oneshot::Receiver<Result<(), String>>) {
let (binary_tx, mut binary_rx) = mpsc::channel::<Bytes>(UPLOAD_CHANNEL_CAPACITY);
let (internal_tx, internal_rx) = oneshot::channel::<Result<(), String>>();
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<Output = Result<tokio::fs::File, String>>,
expected_size: u64,
rx: &mut mpsc::Receiver<Bytes>,
) -> 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<Incoming>,
state: AppState,
core/src/http/server/controller/web.rs
+14 -16
View File
@@ -1,12 +1,10 @@
use crate::http::dto_v2::{
InfoResponseDtoV2, PrepareDownloadResponseDtoV2, PROTOCOL_VERSION_V2,
};
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::{AppState, RequestClientInfo};
use crate::model::transfer::FileDto;
use crate::model::transfer::{FileContent, FileDto};
use bytes::Bytes;
use http_body_util::{BodyExt, StreamBody};
use hyper::body::{Frame, Incoming};
@@ -47,10 +45,10 @@ pub enum WebSendEvent {
/// An accepted web client downloads a file via `GET /api/localsend/v2/download`.
///
/// The application must respond on `content_tx` with a channel receiving
/// the binary chunks of the file. The response body advertises `file.size`
/// bytes, so the application should send exactly that many bytes before
/// closing the channel (closing it earlier aborts the download).
/// The application must respond on `content_tx` with the file content. The
/// response body advertises `file.size` bytes, so the application should
/// provide exactly that many bytes before closing the stream (closing it
/// earlier aborts the download).
/// Dropping `content_tx` results in a 500 response.
FileDownload {
/// The ID of the download session.
@@ -62,8 +60,8 @@ pub enum WebSendEvent {
/// The metadata of the file being downloaded.
file: FileDto,
/// Channel to provide the receiver on which the file content arrives.
content_tx: oneshot::Sender<mpsc::Receiver<Bytes>>,
/// Channel to provide the content of the file being downloaded.
content_tx: oneshot::Sender<FileContent>,
},
}
@@ -324,7 +322,7 @@ pub(crate) async fn download(
};
// The application provides the file content as a stream of bytes.
let (content_tx, content_rx) = oneshot::channel();
let (content_tx, content_rx) = oneshot::channel::<FileContent>();
let event = WebSendEvent::FileDownload {
session_id: session_id.clone(),
file_id: file_id.clone(),
@@ -335,12 +333,12 @@ pub(crate) async fn download(
return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR));
}
let binary_rx = content_rx
let content = content_rx
.await
.map_err(|_| AppError::Status(StatusCode::INTERNAL_SERVER_ERROR))?;
let size = file.size;
let body = receiver_stream_body(binary_rx);
let body = receiver_stream_body(content.into_receiver());
// The file name may be inside directories.
let file_name = file.file_name.replace('/', "-");
@@ -417,9 +415,9 @@ async fn file_list_response(
}
/// Streams application-provided chunks as a response body.
fn receiver_stream_body(binary_rx: mpsc::Receiver<Bytes>) -> BoxedBody {
let stream =
ReceiverStream::new(binary_rx).map(|chunk| Ok::<_, std::io::Error>(Frame::data(chunk)));
fn receiver_stream_body(binary_rx: mpsc::Receiver<Vec<u8>>) -> BoxedBody {
let stream = ReceiverStream::new(binary_rx)
.map(|chunk| Ok::<_, std::io::Error>(Frame::data(Bytes::from(chunk))));
StreamBody::new(stream).boxed()
}
core/src/http/server/mod.rs
+2 -4
View File
@@ -7,7 +7,7 @@ mod response;
mod session;
pub use crate::http::server::controller::v2::{
PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2,
FileUploadTargetV2, PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2,
};
pub use crate::http::server::controller::web::{WebSendConfig, WebSendEvent, WebSendI18n};
@@ -312,9 +312,7 @@ async fn handle_request(req: Request<Incoming>) -> Result<Response<BoxedBody>, h
}))
}
async fn handle_request_inner(
mut req: Request<Incoming>,
) -> Result<Response<BoxedBody>, AppError> {
async fn handle_request_inner(mut req: Request<Incoming>) -> Result<Response<BoxedBody>, AppError> {
let Some(state) = req.extensions_mut().remove::<AppState>() else {
return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR));
};
core/src/http/server/query.rs
+1 -1
View File
@@ -42,4 +42,4 @@ mod tests {
assert_eq!(query.get("a").unwrap(), "%zz");
assert_eq!(query.get("b").unwrap(), "%4");
}
}
}
core/src/http/server/response.rs
+2 -3
View File
@@ -34,9 +34,8 @@ impl<T: Serialize> JsonResponse<T> {
http::HeaderValue::from_static("application/json"),
);
*response.body_mut() = full_body(
serde_json::to_string(&self.body).unwrap_or_else(|_| "{}".to_string()),
);
*response.body_mut() =
full_body(serde_json::to_string(&self.body).unwrap_or_else(|_| "{}".to_string()));
response
}
core/src/http/state.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use serde::{Deserialize, Serialize};
use crate::model::discovery::DeviceType;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)]
#[serde(rename_all = "camelCase")]
core/src/main.rs
+8 -5
View File
@@ -7,7 +7,7 @@ mod webrtc;
use crate::crypto::token;
use crate::http::client::LsHttpClientV3;
use crate::http::dto::{PrepareUploadRequestDto, ProtocolType, RegisterDto};
use crate::http::server::{PrepareUploadDecisionV2, ServerEventV2};
use crate::http::server::{FileUploadTargetV2, PrepareUploadDecisionV2, ServerEventV2};
use crate::http::server::{ServerConfigV2, TlsConfig};
use crate::model::discovery::DeviceType;
use crate::webrtc::signaling::{ClientInfo, WsServerMessage};
@@ -165,11 +165,14 @@ async fn server_test() -> Result<()> {
));
}
ServerEventV2::FileUpload {
file,
mut binary_rx,
result_tx,
..
file, target_tx, ..
} => {
let (binary_tx, mut binary_rx) = mpsc::channel::<Bytes>(16);
let (result_tx, result_rx) = oneshot::channel::<Result<(), String>>();
let _ = target_tx.send(FileUploadTargetV2::Stream {
binary_tx,
result_rx,
});
tokio::spawn(async move {
let mut received: u64 = 0;
while let Some(chunk) = binary_rx.recv().await {
core/src/model/transfer.rs
+84
View File
@@ -1,4 +1,88 @@
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use tokio::sync::mpsc;
/// Channel capacity used when normalizing a file-backed [`FileContent`] into a stream.
const FILE_CHANNEL_CAPACITY: usize = 16;
/// The binary content of a file provided by the application for a transfer.
///
/// Shared by the HTTP client (upload) and server (download API) so both can
/// obtain a file's content as an in-memory stream of chunks, from a regular
/// file path, or, on Android, directly from a raw file descriptor.
#[derive(Debug)]
pub enum FileContent {
/// A stream of binary chunks. The channel is closed once the file has been
/// fully provided.
Stream(mpsc::Receiver<Vec<u8>>),
/// A path to a regular file the content is read from.
Path(PathBuf),
/// A raw file descriptor the content is read from (Android only).
#[cfg(target_os = "android")]
Fd(std::os::fd::RawFd),
}
impl FileContent {
/// Normalizes the content into a stream of binary chunks.
///
/// [`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<Vec<u8>> {
match self {
FileContent::Stream(rx) => rx,
FileContent::Path(path) => {
let (tx, rx) = mpsc::channel(FILE_CHANNEL_CAPACITY);
tokio::spawn(async move {
match tokio::fs::File::open(&path).await {
Ok(file) => read_file_into_sender(file, tx).await,
Err(e) => {
tracing::error!("Failed to open {}: {e}", path.display());
}
}
});
rx
}
#[cfg(target_os = "android")]
FileContent::Fd(fd) => {
use std::os::fd::FromRawFd;
let (tx, rx) = mpsc::channel(FILE_CHANNEL_CAPACITY);
// SAFETY: the descriptor is owned by this transfer; wrapping it in
// a File transfers that ownership so it is closed once reading finishes.
let std_file = unsafe { std::fs::File::from_raw_fd(fd) };
let file = tokio::fs::File::from_std(std_file);
tokio::spawn(read_file_into_sender(file, tx));
rx
}
}
}
}
/// 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<Vec<u8>>) {
use tokio::io::AsyncReadExt;
let mut buffer = vec![0u8; 64 * 1024];
loop {
match file.read(&mut buffer).await {
Ok(0) => break,
Ok(n) => {
if tx.send(buffer[..n].to_vec()).await.is_err() {
break;
}
}
Err(e) => {
tracing::error!("Failed to read file content: {e}");
break;
}
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
core/tests/v2_server.rs
+124 -28
View File
@@ -3,15 +3,19 @@
use localsend::http::client::{ClientError, LsHttpClientV2};
use localsend::http::dto::ProtocolType;
use localsend::http::dto_v2::{PrepareUploadRequestDtoV2, ProtocolTypeV2, RegisterDtoV2};
use localsend::http::server::{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::FileDto;
use localsend::model::transfer::{FileContent, FileDto};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU16, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio_util::sync::CancellationToken;
struct TestServer {
port: u16,
@@ -22,7 +26,15 @@ struct TestServer {
_stop_tx: oneshot::Sender<()>,
}
async fn start_test_server(pin: Option<String>, accept: bool) -> TestServer {
/// Starts a test server.
///
/// Uploads are received as a stream, or written by the server into `save_dir`
/// when given. Either way, the content ends up in [`TestServer::received`].
async fn start_test_server(
pin: Option<String>,
accept: bool,
save_dir: Option<PathBuf>,
) -> TestServer {
let _ = tracing_subscriber::fmt().with_test_writer().try_init();
let port = free_port();
let received: Arc<Mutex<HashMap<String, Vec<u8>>>> = Arc::new(Mutex::new(HashMap::new()));
@@ -50,20 +62,41 @@ async fn start_test_server(pin: Option<String>, accept: bool) -> TestServer {
let _ = decision_tx.send(decision);
}
ServerEventV2::FileUpload {
file_id,
mut binary_rx,
result_tx,
..
file_id, target_tx, ..
} => {
let received = received.clone();
tokio::spawn(async move {
let mut bytes = Vec::new();
while let Some(chunk) = binary_rx.recv().await {
bytes.extend_from_slice(&chunk);
match &save_dir {
None => {
let (binary_tx, mut binary_rx) = mpsc::channel(16);
let (result_tx, result_rx) = oneshot::channel();
let _ = target_tx.send(FileUploadTargetV2::Stream {
binary_tx,
result_rx,
});
tokio::spawn(async move {
let mut bytes = Vec::new();
while let Some(chunk) = binary_rx.recv().await {
bytes.extend_from_slice(&chunk);
}
received.lock().await.insert(file_id, bytes);
let _ = result_tx.send(Ok(()));
});
}
received.lock().await.insert(file_id, bytes);
let _ = result_tx.send(Ok(()));
});
Some(dir) => {
let path = dir.join(&file_id);
let (result_tx, result_rx) = oneshot::channel();
let _ = target_tx.send(FileUploadTargetV2::Path {
path: path.clone(),
result_tx,
});
tokio::spawn(async move {
if let Ok(Ok(())) = result_rx.await {
let bytes = tokio::fs::read(&path).await.unwrap();
received.lock().await.insert(file_id, bytes);
}
});
}
}
}
ServerEventV2::SessionEnd { session_id, reason } => {
session_ends.lock().await.push((session_id, reason));
@@ -85,10 +118,7 @@ async fn start_test_server(pin: Option<String>, accept: bool) -> TestServer {
device_type: None,
token: "server-fingerprint".to_string(),
},
Some(ServerConfigV2 {
pin,
event_tx,
}),
Some(ServerConfigV2 { pin, event_tx }),
None,
stop_rx,
)
@@ -195,7 +225,8 @@ async fn upload_bytes(
session_id,
file_id,
token,
rx,
FileContent::Stream(rx),
CancellationToken::new(),
)
.await
}
@@ -210,7 +241,7 @@ fn assert_status(result: Result<impl Sized, ClientError>, expected_status: u16)
#[tokio::test]
async fn test_register_and_info() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let response = client
@@ -231,7 +262,7 @@ async fn test_register_and_info() {
#[tokio::test]
async fn test_register_over_ipv6() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let response = client
@@ -243,7 +274,7 @@ async fn test_register_over_ipv6() {
#[tokio::test]
async fn test_full_upload_flow() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file_a = file_dto("file-a", "a.bin", 100_000);
@@ -316,9 +347,74 @@ async fn test_full_upload_flow() {
assert_status(result, 403);
}
#[tokio::test]
async fn test_upload_saved_to_path_by_server() {
let save_dir = std::env::temp_dir().join(format!("localsend-test-{}", uuid::Uuid::new_v4()));
tokio::fs::create_dir_all(&save_dir).await.unwrap();
let server = start_test_server(None, true, Some(save_dir.clone())).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file_a = file_dto("file-a", "a.bin", 100_000);
let file_b = file_dto("file-b", "b.bin", 5);
let result = client
.prepare_upload(
ProtocolType::Http,
"127.0.0.1",
server.port,
None,
prepare_upload_request(&[file_a.clone(), file_b.clone()]),
None,
)
.await
.unwrap();
let response = result.response.unwrap();
let bytes_a: Vec<u8> = (0..100_000u32).map(|i| i as u8).collect();
let bytes_b = b"hello".to_vec();
upload_bytes(
&client,
server.port,
&response.session_id,
"file-a",
&response.files["file-a"],
&bytes_a,
)
.await
.unwrap();
upload_bytes(
&client,
server.port,
&response.session_id,
"file-b",
&response.files["file-b"],
&bytes_b,
)
.await
.unwrap();
// The test harness reads the files back after the server reported the result.
tokio::time::sleep(Duration::from_millis(100)).await;
let received = server.received.lock().await;
assert_eq!(received["file-a"], bytes_a);
assert_eq!(received["file-b"], bytes_b);
drop(received);
let session_ends = server.session_ends.lock().await;
assert_eq!(
*session_ends,
vec![(response.session_id.clone(), SessionEndReasonV2::Finished)]
);
let _ = tokio::fs::remove_dir_all(&save_dir).await;
}
#[tokio::test]
async fn test_upload_with_invalid_token() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file = file_dto("file-a", "a.bin", 5);
@@ -362,7 +458,7 @@ async fn test_upload_with_invalid_token() {
#[tokio::test]
async fn test_upload_missing_parameters() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let response = localsend::reqwest::Client::new()
.post(format!(
@@ -378,7 +474,7 @@ async fn test_upload_missing_parameters() {
#[tokio::test]
async fn test_second_session_blocked_and_cancel() {
let server = start_test_server(None, true).await;
let server = start_test_server(None, true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file = file_dto("file-a", "a.bin", 5);
@@ -442,7 +538,7 @@ async fn test_second_session_blocked_and_cancel() {
#[tokio::test]
async fn test_prepare_upload_declined() {
let server = start_test_server(None, false).await;
let server = start_test_server(None, false, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file = file_dto("file-a", "a.bin", 5);
@@ -474,7 +570,7 @@ async fn test_prepare_upload_declined() {
#[tokio::test]
async fn test_pin() {
let server = start_test_server(Some("123456".to_string()), true).await;
let server = start_test_server(Some("123456".to_string()), true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file = file_dto("file-a", "a.bin", 5);
@@ -521,7 +617,7 @@ async fn test_pin() {
#[tokio::test]
async fn test_pin_too_many_attempts() {
let server = start_test_server(Some("123456".to_string()), true).await;
let server = start_test_server(Some("123456".to_string()), true, None).await;
let client = LsHttpClientV2::try_new_without_cert().unwrap();
let file = file_dto("file-a", "a.bin", 5);
core/tests/v2_web_send.rs
+10 -16
View File
@@ -3,10 +3,10 @@
use bytes::Bytes;
use localsend::http::client::{ClientError, LsHttpClientV2};
use localsend::http::dto::ProtocolType;
use localsend::http::server::{ServerEventV2, WebSendEvent};
use localsend::http::server::{start_with_port, ServerConfigV2, WebSendConfig, WebSendI18n};
use localsend::http::server::{ServerEventV2, WebSendEvent};
use localsend::http::state::ClientInfo;
use localsend::model::transfer::FileDto;
use localsend::model::transfer::{FileContent, FileDto};
use std::collections::HashMap;
use std::net::IpAddr;
use std::path::PathBuf;
@@ -34,10 +34,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<Bytes>) {
async fn stream(self, tx: mpsc::Sender<Vec<u8>>) {
match self {
TestFileContent::Bytes(bytes) => {
let _ = tx.send(bytes).await;
let _ = tx.send(bytes.to_vec()).await;
}
TestFileContent::Path(path) => {
let mut file = tokio::fs::File::open(&path)
@@ -49,11 +49,7 @@ impl TestFileContent {
if bytes_read == 0 {
break; // EOF
}
if tx
.send(Bytes::copy_from_slice(&buffer[..bytes_read]))
.await
.is_err()
{
if tx.send(buffer[..bytes_read].to_vec()).await.is_err() {
break; // client disconnected
}
}
@@ -97,8 +93,8 @@ async fn start_test_server(
.expect("FileDownload for unknown file")
.clone();
tokio::spawn(async move {
let (tx, rx) = mpsc::channel::<Bytes>(16);
if content_tx.send(rx).is_err() {
let (tx, rx) = mpsc::channel::<Vec<u8>>(16);
if content_tx.send(FileContent::Stream(rx)).is_err() {
return;
}
content.stream(tx).await;
@@ -203,7 +199,8 @@ fn web_send_config(
Vec<u8>,
) {
let disk_content: Vec<u8> = (0..100_000u32).map(|i| i as u8).collect();
let disk_path = std::env::temp_dir().join(format!("localsend-web-send-{}", uuid::Uuid::new_v4()));
let disk_path =
std::env::temp_dir().join(format!("localsend-web-send-{}", uuid::Uuid::new_v4()));
std::fs::write(&disk_path, &disk_content).expect("Failed to write test file");
// The config only carries the metadata; the content is streamed by the
@@ -283,10 +280,7 @@ async fn test_web_page() {
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let i18n = response
.json::<HashMap<String, String>>()
.await
.unwrap();
let i18n = response.json::<HashMap<String, String>>().await.unwrap();
assert_eq!(i18n["enterPin"], "Enter PIN");
assert!(i18n.contains_key("waiting"));