feat: add cancel token to http client upload

core/Cargo.lock

@@ -1263,6 +1263,7 @@ dependencies = [
  "tokio-rustls",
  "tokio-stream",
  "tokio-tungstenite",
+ "tokio-util",
  "tracing",
  "tracing-subscriber",
  "tungstenite",

core/Cargo.toml

@@ -30,6 +30,7 @@ tokio = { version = "1.49.0", features = ["full"] }
 tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"], optional = true }
 tokio-stream = "0.1.18"
 tokio-tungstenite = { version = "0.28.0", features = ["rustls-tls-webpki-roots"], optional = true }
+tokio-util = { version = "0.7.16", optional = true }
 tracing = "0.1.44"
 tracing-subscriber = { version = "0.3.20" }
 tungstenite = "0.28.0"
@@ -40,7 +41,7 @@ x509-parser = { version = "0.18.0", features = ["verify"], optional = true }
 [features]
 default = []
 crypto = ["ed25519-dalek", "rsa", "sha2"]
-http = ["crypto", "form_urlencoded", "http-body-util", "hyper", "hyper-util", "pem", "percent-encoding", "reqwest", "rustls", "socket2", "tokio-rustls", "x509-parser"]
+http = ["crypto", "form_urlencoded", "http-body-util", "hyper", "hyper-util", "pem", "percent-encoding", "reqwest", "rustls", "socket2", "tokio-rustls", "tokio-util", "x509-parser"]
 webrtc-signaling = ["tokio-tungstenite"]
 webrtc = ["crypto", "flate2", "dep:webrtc", "webrtc-signaling", "x509-parser"]
 full = ["crypto", "http", "webrtc"]

core/src/http/client/mod.rs

@@ -37,6 +37,9 @@ pub enum ClientError {
 
     #[error(transparent)]
     Other(#[from] anyhow::Error),
+
+    #[error("Upload cancelled")]
+    Cancelled,
 }
 
 impl LsHttpClient {
@@ -110,19 +113,20 @@ impl LsHttpClient {
         file_id: &str,
         token: &str,
         binary: tokio::sync::mpsc::Receiver<Vec<u8>>,
+        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,
+                        protocol, ip, port, public_key, session_id, file_id, token, binary, cancel,
                     )
                     .await
             }
             LsHttpClient::V3(client) => {
                 client
                     .upload(
-                        protocol, ip, port, public_key, session_id, file_id, token, binary,
+                        protocol, ip, port, public_key, session_id, file_id, token, binary, cancel,
                     )
                     .await
             }

core/src/http/client/v2.rs

@@ -9,6 +9,7 @@ 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;
 
 /// HTTP client for LocalSend Protocol v2.1.
@@ -193,6 +194,7 @@ impl LsHttpClientV2 {
     /// * `file_id` - File ID to upload
     /// * `token` - File-specific token from prepare_upload
     /// * `binary` - Channel receiving file chunks
+    /// * `cancel` - Cancellation token; cancelling it aborts the upload with [`ClientError::Cancelled`]
     ///
     /// # Errors
     /// * 400 - Missing parameters
@@ -209,6 +211,7 @@ impl LsHttpClientV2 {
         file_id: &str,
         token: &str,
         binary: mpsc::Receiver<Vec<u8>>,
+        cancel: CancellationToken,
     ) -> Result<(), ClientError> {
         let url = TargetUrl {
             version: ApiVersion::V2,
@@ -227,7 +230,10 @@ impl LsHttpClientV2 {
         let stream = ReceiverStream::new(binary).map(Ok::<Vec<u8>, anyhow::Error>);
         let body = reqwest::Body::wrap_stream(stream);
 
-        let res = self.client.post(&url).body(body).send().await?;
+        let res = tokio::select! {
+            res = self.client.post(&url).body(body).send() => res?,
+            _ = cancel.cancelled() => return Err(ClientError::Cancelled),
+        };
 
         if protocol == ProtocolType::Https {
             super::verify_cert_from_res(&res, public_key)?;

core/src/http/client/v3.rs

@@ -10,6 +10,7 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use tokio::sync::{mpsc, Mutex};
 use tokio_stream::wrappers::ReceiverStream;
+use tokio_util::sync::CancellationToken;
 
 pub struct LsHttpClientV3 {
     client: reqwest::Client,
@@ -180,6 +181,9 @@ impl LsHttpClientV3 {
     }
 
     /// Uploads a file to the server.
+    ///
+    /// `cancel` is a cancellation token; cancelling it aborts the upload with
+    /// [`ClientError::Cancelled`].
     pub async fn upload(
         &self,
         protocol: ProtocolType,
@@ -190,8 +194,9 @@ impl LsHttpClientV3 {
         file_id: &str,
         token: &str,
         binary: mpsc::Receiver<Vec<u8>>,
+        cancel: CancellationToken,
     ) -> Result<(), ClientError> {
-        let res = self
+        let send = self
             .client
             .post(
                 TargetUrl {
@@ -212,8 +217,12 @@ impl LsHttpClientV3 {
                 let stream = ReceiverStream::new(binary).map(Ok::<Vec<u8>, anyhow::Error>);
                 reqwest::Body::wrap_stream(stream)
             })
-            .send()
-            .await?;
+            .send();
+
+        let res = tokio::select! {
+            res = send => res?,
+            _ = cancel.cancelled() => return Err(ClientError::Cancelled),
+        };
 
         if protocol == ProtocolType::Https {
             super::verify_cert_from_res(&res, public_key)?;