use async_trait::async_trait;
use http::{
Method, Uri,
header::{HeaderMap, HeaderName, HeaderValue},
};
use std::{
sync::{Arc, Mutex},
time::Duration,
};
use url::Url;
use crate::error::{Error, Result};
#[cfg(feature = "checksum")]
use crate::helpers::encode_checksum;
use crate::runtime::{MaybeSend, MaybeSendSync};
use crate::transport::{Transport, TransportBody, TransportRequest};
mod handle;
mod protocol;
mod upload;
pub use handle::Upload;
pub use protocol::{NewUpload, ServerCapabilities, UploadInfo};
#[cfg(all(feature = "source-file", not(target_arch = "wasm32")))]
pub use upload::FileSource;
#[cfg(not(target_arch = "wasm32"))]
pub use upload::ParallelUpload;
pub use upload::{UploadProgress, UploadSource};
#[cfg(feature = "transport-reqwest")]
use crate::transport::ReqwestTransport;
#[derive(Clone)]
pub struct Client<
#[cfg(feature = "transport-reqwest")] T = ReqwestTransport,
#[cfg(not(feature = "transport-reqwest"))] T,
> {
endpoint: Url,
transport: T,
headers: HeaderMap,
max_retries: usize,
retry_delay: Duration,
max_chunk_size: usize,
max_initial_upload_size: usize,
#[cfg(feature = "checksum")]
checksum: Option<ChecksumMode>,
header_provider: Option<Arc<dyn HeaderProvider>>,
retry_hook: Option<Arc<dyn RetryHook>>,
restrict_to_endpoint_origin: bool,
capabilities: Arc<Mutex<Option<ServerCapabilities>>>,
}
#[cfg(feature = "checksum")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ChecksumMode {
Header(tus_protocol::ChecksumAlgorithm),
Trailer(tus_protocol::ChecksumAlgorithm),
}
#[cfg(feature = "checksum")]
impl ChecksumMode {
fn algorithm(self) -> tus_protocol::ChecksumAlgorithm {
match self {
ChecksumMode::Header(algorithm) | ChecksumMode::Trailer(algorithm) => algorithm,
}
}
}
#[cfg(feature = "checksum")]
impl From<tus_protocol::ChecksumAlgorithm> for ChecksumMode {
fn from(algorithm: tus_protocol::ChecksumAlgorithm) -> Self {
Self::Header(algorithm)
}
}
#[cfg(feature = "transport-reqwest")]
impl Client<ReqwestTransport> {
#[must_use]
pub fn new(endpoint: Url) -> Self {
Self::with_transport(endpoint, ReqwestTransport::new())
}
}
impl<T> std::fmt::Debug for Client<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut debug = f.debug_struct("Client");
debug
.field("endpoint", &self.endpoint)
.field("max_retries", &self.max_retries)
.field("retry_delay", &self.retry_delay)
.field("max_chunk_size", &self.max_chunk_size)
.field("max_initial_upload_size", &self.max_initial_upload_size);
#[cfg(feature = "checksum")]
debug.field("checksum", &self.checksum);
debug.finish()
}
}
impl<T> Client<T>
where
T: Transport,
{
#[must_use]
pub fn with_transport(endpoint: Url, transport: T) -> Self {
Self {
endpoint,
transport,
headers: HeaderMap::new(),
max_retries: 3,
retry_delay: Duration::from_millis(200),
max_chunk_size: 8 * 1024 * 1024,
max_initial_upload_size: 256 * 1024,
#[cfg(feature = "checksum")]
checksum: None,
header_provider: None,
retry_hook: None,
restrict_to_endpoint_origin: false,
capabilities: Arc::new(Mutex::new(None)),
}
}
#[must_use]
pub fn with_headers(mut self, headers: HeaderMap) -> Self {
self.headers = headers;
self
}
#[must_use]
pub fn with_endpoint_origin_restriction(mut self) -> Self {
self.restrict_to_endpoint_origin = true;
self
}
#[must_use]
pub fn with_max_retries(mut self, max_retries: usize) -> Self {
self.max_retries = max_retries;
self
}
#[must_use]
pub fn with_retry_delay(mut self, retry_delay: Duration) -> Self {
self.retry_delay = retry_delay;
self
}
#[must_use]
pub fn with_max_chunk_size(mut self, max_chunk_size: usize) -> Self {
self.max_chunk_size = max_chunk_size.max(1);
self
}
#[must_use]
pub fn with_max_initial_upload_size(mut self, max_initial_upload_size: usize) -> Self {
self.max_initial_upload_size = max_initial_upload_size;
self
}
#[cfg(feature = "checksum")]
#[must_use]
pub fn with_checksum(mut self, mode: impl Into<ChecksumMode>) -> Self {
self.checksum = Some(mode.into());
self
}
#[must_use]
pub fn with_header_provider<P>(mut self, provider: P) -> Self
where
P: HeaderProvider + 'static,
{
self.header_provider = Some(Arc::new(provider));
self
}
#[must_use]
pub fn with_retry_hook<H>(mut self, hook: H) -> Self
where
H: RetryHook + 'static,
{
self.retry_hook = Some(Arc::new(hook));
self
}
async fn request(&self, method: Method, url: &str) -> Result<TransportRequest> {
let url = Url::parse(url)?;
let uri: Uri = url.as_str().parse().map_err(|err| {
Error::transport_permanent(format!("invalid transport URI {}: {err}", url.as_str()))
})?;
let mut request = http::Request::builder()
.method(method)
.uri(uri)
.body(TransportBody::Empty)
.map_err(|err| Error::transport_permanent(format!("failed to build request: {err}")))?;
for (name, value) in &self.headers {
request.headers_mut().append(name.clone(), value.clone());
}
if let Some(provider) = &self.header_provider {
for (name, value) in provider.headers().await?.iter() {
request.headers_mut().insert(name.clone(), value.clone());
}
}
request.headers_mut().insert(
HeaderName::from_static("tus-resumable"),
HeaderValue::from_static(tus_protocol::TUS_RESUMABLE),
);
Ok(request)
}
pub(crate) fn known_capabilities(&self) -> Option<ServerCapabilities> {
self.capabilities
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone()
}
pub(crate) fn store_capabilities(&self, capabilities: &ServerCapabilities) {
*self
.capabilities
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(capabilities.clone());
}
#[cfg(feature = "checksum")]
fn apply_checksum(
&self,
mut request: TransportRequest,
body: Vec<u8>,
) -> Result<TransportRequest> {
let Some(mode) = self.checksum else {
*request.body_mut() = TransportBody::Bytes(body);
return Ok(request);
};
let checksum = encode_checksum(mode.algorithm(), &body);
match mode {
ChecksumMode::Header(algorithm) => {
request.headers_mut().insert(
HeaderName::from_static("upload-checksum"),
HeaderValue::from_str(&format!("{} {}", algorithm.as_str(), checksum))
.map_err(|_| Error::InvalidRequestHeader {
name: "Upload-Checksum".to_string(),
value: checksum.clone(),
})?,
);
*request.body_mut() = TransportBody::Bytes(body);
Ok(request)
}
#[cfg(target_arch = "wasm32")]
ChecksumMode::Trailer(_) => Err(Error::transport_permanent(
"checksum trailers are not supported on wasm32 (browser fetch cannot send \
request trailers); use ChecksumMode::Header instead",
)),
#[cfg(not(target_arch = "wasm32"))]
ChecksumMode::Trailer(algorithm) => {
request.headers_mut().insert(
HeaderName::from_static("trailer"),
HeaderValue::from_static("upload-checksum"),
);
*request.body_mut() = TransportBody::BytesWithTrailer {
body,
trailer_name: HeaderName::from_static("upload-checksum"),
trailer_value: format!("{} {}", algorithm.as_str(), checksum),
};
Ok(request)
}
}
}
#[cfg(not(feature = "checksum"))]
fn apply_checksum(
&self,
mut request: TransportRequest,
body: Vec<u8>,
) -> Result<TransportRequest> {
*request.body_mut() = TransportBody::Bytes(body);
Ok(request)
}
#[cfg(feature = "checksum")]
fn apply_creation_checksum(
&self,
mut request: TransportRequest,
body: Vec<u8>,
) -> Result<TransportRequest> {
let server_advertises_checksum = self
.known_capabilities()
.is_some_and(|capabilities| capabilities.has_extension("checksum"));
if server_advertises_checksum {
self.apply_checksum(request, body)
} else {
*request.body_mut() = TransportBody::Bytes(body);
Ok(request)
}
}
#[cfg(not(feature = "checksum"))]
fn apply_creation_checksum(
&self,
request: TransportRequest,
body: Vec<u8>,
) -> Result<TransportRequest> {
self.apply_checksum(request, body)
}
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait HeaderProvider: MaybeSendSync {
async fn headers(&self) -> Result<HeaderMap>;
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<F, Fut> HeaderProvider for F
where
F: Fn() -> Fut + MaybeSendSync,
Fut: std::future::Future<Output = Result<HeaderMap>> + MaybeSend,
{
async fn headers(&self) -> Result<HeaderMap> {
self().await
}
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait RetryHook: MaybeSendSync {
async fn before_retry(&self, attempt: usize, error: &Error) -> bool;
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<F, Fut> RetryHook for F
where
F: Fn(usize, &Error) -> Fut + MaybeSendSync,
Fut: std::future::Future<Output = bool> + MaybeSend,
{
async fn before_retry(&self, attempt: usize, error: &Error) -> bool {
self(attempt, error).await
}
}
fn insert_request_header(
request: &mut TransportRequest,
name: impl AsRef<str>,
value: impl ToString,
) -> Result<()> {
let name = name.as_ref();
let value = value.to_string();
let header_name =
HeaderName::from_bytes(name.as_bytes()).map_err(|_| Error::InvalidRequestHeader {
name: name.to_string(),
value: value.clone(),
})?;
let header_value = HeaderValue::from_str(&value).map_err(|_| Error::InvalidRequestHeader {
name: header_name.as_str().to_string(),
value: value.clone(),
})?;
request.headers_mut().insert(header_name, header_value);
Ok(())
}
#[cfg(test)]
mod test_support;
#[cfg(test)]
mod tests {
use super::*;
use crate::client::test_support::{MockTransport, endpoint_url, transport_response};
use http::header::{HeaderMap, HeaderName, HeaderValue, LOCATION};
use tus_protocol::UploadMetadata;
#[cfg(not(target_arch = "wasm32"))]
use tokio::test as async_test;
#[cfg(target_arch = "wasm32")]
use wasm_bindgen_test::wasm_bindgen_test as async_test;
#[async_test]
async fn headers_are_applied_to_transport_requests() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("x-tenant-id"),
HeaderValue::from_static("team-a"),
);
let client =
Client::with_transport(endpoint_url(), transport.clone()).with_headers(headers);
let (upload, _info) = client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
assert_eq!(upload.url().as_str(), "http://example.test/files/mock-id");
let requests = transport.requests.lock().unwrap();
let request = requests.first().unwrap();
assert_eq!(
request
.headers()
.get("x-tenant-id")
.and_then(|value| value.to_str().ok()),
Some("team-a")
);
}
#[async_test]
async fn repeated_default_header_values_are_all_sent() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let mut headers = HeaderMap::new();
headers.append(
HeaderName::from_static("x-trace-tag"),
HeaderValue::from_static("alpha"),
);
headers.append(
HeaderName::from_static("x-trace-tag"),
HeaderValue::from_static("beta"),
);
let client =
Client::with_transport(endpoint_url(), transport.clone()).with_headers(headers);
client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
let requests = transport.requests.lock().unwrap();
let values: Vec<_> = requests
.first()
.unwrap()
.headers()
.get_all("x-trace-tag")
.iter()
.filter_map(|value| value.to_str().ok())
.collect();
assert_eq!(values, vec!["alpha", "beta"]);
}
#[async_test]
async fn header_provider_is_applied_to_transport_requests() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let client = Client::with_transport(endpoint_url(), transport.clone())
.with_header_provider(|| {
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("x-tenant-id"),
HeaderValue::from_static("team-a"),
);
std::future::ready(Ok(headers))
});
let (upload, _info) = client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
assert_eq!(upload.url().as_str(), "http://example.test/files/mock-id");
let requests = transport.requests.lock().unwrap();
let request = requests.first().unwrap();
assert_eq!(
request
.headers()
.get("x-tenant-id")
.and_then(|value| value.to_str().ok()),
Some("team-a")
);
}
#[async_test]
async fn header_provider_replaces_configured_multi_values_of_same_name() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let mut headers = HeaderMap::new();
headers.append(
HeaderName::from_static("x-trace-tag"),
HeaderValue::from_static("alpha"),
);
headers.append(
HeaderName::from_static("x-trace-tag"),
HeaderValue::from_static("beta"),
);
headers.insert(
HeaderName::from_static("x-tenant-id"),
HeaderValue::from_static("team-a"),
);
let client = Client::with_transport(endpoint_url(), transport.clone())
.with_headers(headers)
.with_header_provider(|| {
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("x-trace-tag"),
HeaderValue::from_static("fresh"),
);
std::future::ready(Ok(headers))
});
client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
let requests = transport.requests.lock().unwrap();
let request = requests.first().unwrap();
let trace_tags: Vec<_> = request
.headers()
.get_all("x-trace-tag")
.iter()
.filter_map(|value| value.to_str().ok())
.collect();
assert_eq!(
trace_tags,
vec!["fresh"],
"provider value must replace every configured value of the same name"
);
assert_eq!(
request
.headers()
.get("x-tenant-id")
.and_then(|value| value.to_str().ok()),
Some("team-a"),
"configured headers with other names must survive"
);
}
#[async_test]
async fn box_transport_erases_the_concrete_transport_type() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let boxed = crate::BoxTransport::new(transport.clone());
let client: Client<crate::BoxTransport> = Client::with_transport(endpoint_url(), boxed);
let (upload, _info) = client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
assert_eq!(upload.url().as_str(), "http://example.test/files/mock-id");
assert_eq!(transport.requests.lock().unwrap().len(), 1);
}
#[async_test]
async fn arc_wrapped_transports_are_transports() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let client = Client::with_transport(endpoint_url(), Arc::new(transport.clone()));
let (upload, _info) = client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
assert_eq!(upload.url().as_str(), "http://example.test/files/mock-id");
assert_eq!(transport.requests.lock().unwrap().len(), 1);
}
#[async_test]
async fn protocol_headers_override_configured_headers() {
let transport = MockTransport::default();
transport
.responses
.lock()
.unwrap()
.push_back(Ok(transport_response(
201,
{
let mut headers = HeaderMap::new();
headers.insert(LOCATION, HeaderValue::from_static("/files/mock-id"));
headers
},
Vec::new(),
)));
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("tus-resumable"),
HeaderValue::from_static("0.0.0"),
);
let client =
Client::with_transport(endpoint_url(), transport.clone()).with_headers(headers);
client
.create_upload(NewUpload::new(5, UploadMetadata::new()))
.await
.unwrap();
let requests = transport.requests.lock().unwrap();
assert_eq!(
requests
.first()
.unwrap()
.headers()
.get("tus-resumable")
.and_then(|value| value.to_str().ok()),
Some(tus_protocol::TUS_RESUMABLE)
);
}
}