diff --git a/src/aws/client.rs b/src/aws/client.rs index 199f4a81..d6b2534d 100644 --- a/src/aws/client.rs +++ b/src/aws/client.rs @@ -36,8 +36,8 @@ use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse}; use crate::list::{PaginatedListOptions, PaginatedListResult}; use crate::multipart::PartId; use crate::{ - Attribute, Attributes, ClientOptions, GetOptions, MultipartId, Path, PutMultipartOptions, - PutPayload, PutResult, Result, RetryConfig, TagSet, + Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, Path, + PutMultipartOptions, PutPayload, PutResult, Result, RetryConfig, TagSet, }; use async_trait::async_trait; use base64::Engine; @@ -470,8 +470,10 @@ impl Request<'_> { pub(crate) async fn do_put(self) -> Result { let response = self.send().await?; - Ok(get_put_result(response.headers(), VERSION_HEADER) - .map_err(|source| Error::Metadata { source })?) + Ok( + get_put_result(response, VERSION_HEADER) + .map_err(|source| Error::Metadata { source })?, + ) } } @@ -857,11 +859,12 @@ impl S3Client { path: location.as_ref().to_string(), })?; - let version = get_version(response.headers(), VERSION_HEADER) + let (parts, body) = response.into_parts(); + + let version = get_version(&parts.headers, VERSION_HEADER) .map_err(|source| Error::Metadata { source })?; - let data = response - .into_body() + let data = body .bytes() .await .map_err(|source| Error::CompleteMultipartResponseBody { source })?; @@ -872,6 +875,7 @@ impl S3Client { Ok(PutResult { e_tag: Some(response.e_tag), version, + extensions: parts.extensions, }) } @@ -993,8 +997,11 @@ impl ListClient for Arc { .with_aws_sigv4(credential.authorizer(), None) .send_retry(&self.config.retry_config) .await - .map_err(|source| Error::ListRequest { source })? - .into_body() + .map_err(|source| Error::ListRequest { source })?; + + let (parts, body) = response.into_parts(); + + let response = body .bytes() .await .map_err(|source| Error::ListResponseBody { source })?; @@ -1004,8 +1011,11 @@ impl ListClient for Arc { let token = response.next_continuation_token.take(); + let mut result: ListResult = response.try_into()?; + result.extensions = parts.extensions; + Ok(PaginatedListResult { - result: response.try_into()?, + result, page_token: token, }) } diff --git a/src/aws/mod.rs b/src/aws/mod.rs index e1cdb065..765871a4 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -679,7 +679,10 @@ mod tests { #[tokio::test] async fn s3_test() { maybe_skip_integration!(); - let config = AmazonS3Builder::from_env(); + // tag the extensions of every HTTP response with a marker, + // allowing response_extensions to verify their propagation + let config = + AmazonS3Builder::from_env().with_http_connector(MarkerHttpConnector::default()); let integration = config.build().unwrap(); let config = &integration.client.config; @@ -701,6 +704,7 @@ mod tests { s3_encryption(&integration).await; put_get_attributes(&integration).await; list_paginated(&integration, &integration).await; + response_extensions(&integration, true).await; // Object tagging is not supported by S3 Express One Zone if config.session_provider.is_none() { diff --git a/src/azure/client.rs b/src/azure/client.rs index 6f1ad238..fd3e923b 100644 --- a/src/azure/client.rs +++ b/src/azure/client.rs @@ -564,8 +564,10 @@ impl AzureClient { }; let response = builder.header(&BLOB_TYPE, "BlockBlob").send().await?; - Ok(get_put_result(response.headers(), VERSION_HEADER) - .map_err(|source| Error::Metadata { source })?) + Ok( + get_put_result(response, VERSION_HEADER) + .map_err(|source| Error::Metadata { source })?, + ) } /// PUT a block @@ -617,8 +619,10 @@ impl AzureClient { .send() .await?; - Ok(get_put_result(response.headers(), VERSION_HEADER) - .map_err(|source| Error::Metadata { source })?) + Ok( + get_put_result(response, VERSION_HEADER) + .map_err(|source| Error::Metadata { source })?, + ) } fn build_bulk_delete_body( @@ -983,8 +987,11 @@ impl ListClient for Arc { .sensitive(sensitive) .send() .await - .map_err(|source| Error::ListRequest { source })? - .into_body() + .map_err(|source| Error::ListRequest { source })?; + + let (parts, body) = response.into_parts(); + + let response = body .bytes() .await .map_err(|source| Error::ListResponseBody { source })?; @@ -1007,8 +1014,11 @@ impl ListClient for Arc { } } + let mut result = to_list_result(response, prefix)?; + result.extensions = parts.extensions; + Ok(PaginatedListResult { - result: to_list_result(response, prefix)?, + result, page_token: token, }) } @@ -1051,6 +1061,7 @@ fn to_list_result(value: ListResultInternal, prefix: Option<&str>) -> Result GetContext { meta, range, attributes, + extensions: parts.extensions, }) } diff --git a/src/client/header.rs b/src/client/header.rs index 4c9470c3..d0dd27a5 100644 --- a/src/client/header.rs +++ b/src/client/header.rs @@ -70,15 +70,22 @@ pub(crate) enum Error { }, } -/// Extracts a PutResult from the provided [`HeaderMap`] +/// Extracts a PutResult from the provided response +/// +/// Propagates the extensions of the response into the [`crate::PutResult`] #[cfg(any(feature = "aws", feature = "gcp", feature = "azure"))] pub(crate) fn get_put_result( - headers: &HeaderMap, + response: crate::client::HttpResponse, version: &str, ) -> Result { - let e_tag = Some(get_etag(headers)?); - let version = get_version(headers, version)?; - Ok(crate::PutResult { e_tag, version }) + let (parts, _) = response.into_parts(); + let e_tag = Some(get_etag(&parts.headers)?); + let version = get_version(&parts.headers, version)?; + Ok(crate::PutResult { + e_tag, + version, + extensions: parts.extensions, + }) } /// Extracts a optional version from the provided [`HeaderMap`] diff --git a/src/client/list.rs b/src/client/list.rs index c2123183..e0d684d1 100644 --- a/src/client/list.rs +++ b/src/client/list.rs @@ -114,16 +114,19 @@ impl ListClientExt for T { let mut common_prefixes = BTreeSet::new(); let mut objects = Vec::new(); + let mut extensions = http::Extensions::new(); while let Some(result) = stream.next().await { let response = result?; common_prefixes.extend(response.common_prefixes); objects.extend(response.objects); + extensions.extend(response.extensions); } Ok(ListResult { common_prefixes: common_prefixes.into_iter().collect(), objects, + extensions, }) } } diff --git a/src/client/s3.rs b/src/client/s3.rs index 2beec612..2c8fb63c 100644 --- a/src/client/s3.rs +++ b/src/client/s3.rs @@ -52,6 +52,7 @@ impl TryFrom for ListResult { Ok(Self { common_prefixes, objects, + extensions: Default::default(), }) } } diff --git a/src/gcp/client.rs b/src/gcp/client.rs index 4a246d21..5356cc45 100644 --- a/src/gcp/client.rs +++ b/src/gcp/client.rs @@ -32,7 +32,7 @@ use crate::multipart::PartId; use crate::path::Path; use crate::util::hex_encode; use crate::{ - Attribute, Attributes, ClientOptions, CopyMode, GetOptions, MultipartId, PutMode, + Attribute, Attributes, ClientOptions, CopyMode, GetOptions, ListResult, MultipartId, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result, RetryConfig, }; use async_trait::async_trait; @@ -250,8 +250,10 @@ impl Request<'_> { async fn do_put(self) -> Result { let response = self.send().await?; - Ok(get_put_result(response.headers(), VERSION_HEADER) - .map_err(|source| Error::Metadata { source })?) + Ok( + get_put_result(response, VERSION_HEADER) + .map_err(|source| Error::Metadata { source })?, + ) } } @@ -541,11 +543,12 @@ impl GoogleCloudStorageClient { .await .map_err(|source| Error::CompleteMultipartRequest { source })?; - let version = get_version(response.headers(), VERSION_HEADER) + let (parts, body) = response.into_parts(); + + let version = get_version(&parts.headers, VERSION_HEADER) .map_err(|source| Error::Metadata { source })?; - let data = response - .into_body() + let data = body .bytes() .await .map_err(|source| Error::CompleteMultipartResponseBody { source })?; @@ -556,6 +559,7 @@ impl GoogleCloudStorageClient { Ok(PutResult { e_tag: Some(response.e_tag), version, + extensions: parts.extensions, }) } @@ -705,8 +709,11 @@ impl ListClient for Arc { .with_bearer_auth(credential.as_deref()) .send_retry(&self.config.retry_config) .await - .map_err(|source| Error::ListRequest { source })? - .into_body() + .map_err(|source| Error::ListRequest { source })?; + + let (parts, body) = response.into_parts(); + + let response = body .bytes() .await .map_err(|source| Error::ListResponseBody { source })?; @@ -715,8 +722,12 @@ impl ListClient for Arc { .map_err(|source| Error::InvalidListResponse { source })?; let token = response.next_continuation_token.take(); + + let mut result: ListResult = response.try_into()?; + result.extensions = parts.extensions; + Ok(PaginatedListResult { - result: response.try_into()?, + result, page_token: token, }) } diff --git a/src/gcp/mod.rs b/src/gcp/mod.rs index 51e85ae6..7dc22ad6 100644 --- a/src/gcp/mod.rs +++ b/src/gcp/mod.rs @@ -311,7 +311,12 @@ mod test { #[tokio::test] async fn gcs_test() { maybe_skip_integration!(); - let integration = GoogleCloudStorageBuilder::from_env().build().unwrap(); + // tag the extensions of every HTTP response with a marker, + // allowing response_extensions to verify their propagation + let integration = GoogleCloudStorageBuilder::from_env() + .with_http_connector(MarkerHttpConnector::default()) + .build() + .unwrap(); put_get_delete_list(&integration).await; list_with_offset_exclusivity(&integration).await; @@ -336,6 +341,10 @@ mod test { // Fake GCS server doesn't currently support attributes put_get_attributes(&integration).await; } + + // Fake GCS server does not yet implement XML Multipart uploads + let test_multipart = integration.client.config().base_url == DEFAULT_GCS_BASE_URL; + response_extensions(&integration, test_multipart).await; } #[tokio::test] diff --git a/src/http/client.rs b/src/http/client.rs index 5b2d3434..4375a794 100644 --- a/src/http/client.rs +++ b/src/http/client.rs @@ -256,12 +256,15 @@ impl Client { .send() .await; - let response = match result { - Ok(result) => result - .into_body() - .bytes() - .await - .map_err(|source| Error::Reqwest { source })?, + let (response, extensions) = match result { + Ok(result) => { + let (parts, body) = result.into_parts(); + let body = body + .bytes() + .await + .map_err(|source| Error::Reqwest { source })?; + (body, parts.extensions) + } Err(e) if matches!(e.status(), Some(StatusCode::NOT_FOUND)) => { return match depth { "0" => { @@ -286,8 +289,9 @@ impl Client { } }; - let status = quick_xml::de::from_reader(response.reader()) + let mut status: MultiStatus = quick_xml::de::from_reader(response.reader()) .map_err(|source| Error::InvalidPropFind { source })?; + status.extensions = extensions; Ok(status) } @@ -415,6 +419,9 @@ impl GetClient for Client { #[derive(Deserialize, Default)] pub(crate) struct MultiStatus { pub response: Vec, + /// The extensions of the HTTP response this was parsed from + #[serde(skip)] + pub extensions: ::http::Extensions, } #[derive(Deserialize)] diff --git a/src/http/mod.rs b/src/http/mod.rs index 4d4081b8..3fbf162f 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -111,7 +111,8 @@ impl ObjectStore for HttpStore { } let response = self.client.put(location, payload, opts.attributes).await?; - let e_tag = match get_etag(response.headers()) { + let (parts, _) = response.into_parts(); + let e_tag = match get_etag(&parts.headers) { Ok(e_tag) => Some(e_tag), Err(crate::client::header::Error::MissingEtag) => None, Err(source) => return Err(Error::Metadata { source }.into()), @@ -120,6 +121,7 @@ impl ObjectStore for HttpStore { Ok(PutResult { e_tag, version: None, + extensions: parts.extensions, }) } @@ -209,6 +211,7 @@ impl ObjectStore for HttpStore { Ok(ListResult { common_prefixes, objects, + extensions: status.extensions, }) } @@ -302,9 +305,12 @@ mod tests { maybe_skip_integration!(); let url = std::env::var("HTTP_URL").expect("HTTP_URL must be set"); let options = ClientOptions::new().with_allow_http(true); + // tag the extensions of every HTTP response with a marker, + // allowing response_extensions to verify their propagation let integration = HttpBuilder::new() .with_url(url) .with_client_options(options) + .with_http_connector(MarkerHttpConnector::default()) .build() .unwrap(); @@ -313,5 +319,6 @@ mod tests { list_with_delimiter(&integration).await; rename_and_copy(&integration).await; copy_if_not_exists(&integration).await; + response_extensions(&integration, false).await; } } diff --git a/src/integration.rs b/src/integration.rs index e68837c5..8b435d35 100644 --- a/src/integration.rs +++ b/src/integration.rs @@ -1436,3 +1436,92 @@ pub async fn list_with_offset_exclusivity(storage: &DynObjectStore) { // Clean up delete_fixtures(storage).await; } + +#[cfg(all( + feature = "cloud", + not(all(target_arch = "wasm32", target_os = "wasi")) +))] +mod marker { + use super::*; + use crate::ClientOptions; + use crate::client::{ + HttpClient, HttpConnector, HttpError, HttpRequest, HttpResponse, HttpService, + ReqwestConnector, + }; + use async_trait::async_trait; + + /// A marker inserted into response extensions by [`MarkerHttpConnector`] + #[derive(Clone, Debug, PartialEq)] + struct Marker; + + /// [`HttpService`] middleware that tags every response with a [`Marker`] + #[derive(Debug)] + struct MarkerService(HttpClient); + + #[async_trait] + impl HttpService for MarkerService { + async fn call(&self, req: HttpRequest) -> Result { + let mut response = self.0.execute(req).await?; + response.extensions_mut().insert(Marker); + Ok(response) + } + } + + /// An [`HttpConnector`] that tags the extensions of every HTTP response with a + /// marker, allowing [`response_extensions`] to verify their propagation + #[derive(Debug, Default)] + pub struct MarkerHttpConnector(ReqwestConnector); + + impl HttpConnector for MarkerHttpConnector { + fn connect(&self, options: &ClientOptions) -> crate::Result { + let client = self.0.connect(options)?; + Ok(HttpClient::new(MarkerService(client))) + } + } + + /// Tests that the extensions of HTTP responses are propagated to returned results + /// + /// The provided store must have been built with [`MarkerHttpConnector`] + pub async fn response_extensions(storage: &DynObjectStore, test_multipart: bool) { + delete_fixtures(storage).await; + + let location = Path::from("test_response_extensions/file.txt"); + let data = Bytes::from("arbitrary data"); + + let put = storage.put(&location, data.clone().into()).await.unwrap(); + assert!( + put.extensions.get::().is_some(), + "PutResult should contain the response extensions" + ); + + let get = storage.get(&location).await.unwrap(); + assert!( + get.extensions.get::().is_some(), + "GetResult should contain the response extensions" + ); + + let list = storage.list_with_delimiter(None).await.unwrap(); + assert!( + list.extensions.get::().is_some(), + "ListResult should contain the response extensions" + ); + + if test_multipart { + let mut upload = storage.put_multipart(&location).await.unwrap(); + upload.put_part(data.into()).await.unwrap(); + let complete = upload.complete().await.unwrap(); + assert!( + complete.extensions.get::().is_some(), + "PutResult of a completed multipart upload should contain the response extensions" + ); + } + + delete_fixtures(storage).await; + } +} + +#[cfg(all( + feature = "cloud", + not(all(target_arch = "wasm32", target_os = "wasi")) +))] +pub use marker::{MarkerHttpConnector, response_extensions}; diff --git a/src/lib.rs b/src/lib.rs index d3ea9ee2..cb286f0c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1413,6 +1413,14 @@ pub struct ListResult { pub common_prefixes: Vec, /// Object metadata for the listing pub objects: Vec, + /// Implementation-specific extensions. Intended for use by [`ObjectStore`] implementations + /// that need to return context-specific information (like cache status) from trait methods. + /// + /// HTTP-backed stores in this crate populate this with the extensions of the HTTP + /// response, allowing custom HTTP middleware to propagate information to callers. + /// Where a result is assembled from multiple paginated requests, the extensions of + /// each response are merged, with those of later responses taking precedence. + pub extensions: Extensions, } /// The metadata that describes an object. @@ -1627,6 +1635,12 @@ pub struct GetResult { pub range: Range, /// Additional object attributes pub attributes: Attributes, + /// Implementation-specific extensions. Intended for use by [`ObjectStore`] implementations + /// that need to return context-specific information (like cache status) from trait methods. + /// + /// HTTP-backed stores in this crate populate this with the extensions of the HTTP + /// response, allowing custom HTTP middleware to propagate information to callers. + pub extensions: Extensions, } /// The kind of a [`GetResult`] @@ -1865,7 +1879,7 @@ impl From for PutMultipartOptions { } /// Result for a put request -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct PutResult { /// The unique identifier for the newly created object /// @@ -1873,8 +1887,34 @@ pub struct PutResult { pub e_tag: Option, /// A version indicator for the newly created object pub version: Option, + /// Implementation-specific extensions. Intended for use by [`ObjectStore`] implementations + /// that need to return context-specific information (like cache status) from trait methods. + /// + /// HTTP-backed stores in this crate populate this with the extensions of the HTTP + /// response, allowing custom HTTP middleware to propagate information to callers. + /// + /// These extensions are excluded from [`PartialEq`] and [`Eq`]. + pub extensions: Extensions, } +impl PartialEq for PutResult { + fn eq(&self, other: &Self) -> bool { + let Self { + e_tag, + version, + extensions: _, + } = self; + let Self { + e_tag: other_e_tag, + version: other_version, + extensions: _, + } = other; + (e_tag == other_e_tag) && (version == other_version) + } +} + +impl Eq for PutResult {} + /// Configure preconditions for the copy operation #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum CopyMode { diff --git a/src/local.rs b/src/local.rs index 2e71ce63..b3f3f7f7 100644 --- a/src/local.rs +++ b/src/local.rs @@ -423,6 +423,7 @@ impl ObjectStore for LocalFileSystem { Ok(PutResult { e_tag, version: None, + extensions: Default::default(), }) }) .await @@ -466,6 +467,7 @@ impl ObjectStore for LocalFileSystem { attributes: Attributes::default(), range, meta, + extensions: Default::default(), }) }) .await @@ -563,6 +565,7 @@ impl ObjectStore for LocalFileSystem { Ok(ListResult { common_prefixes: common_prefixes.into_iter().collect(), objects, + extensions: Default::default(), }) }) .await @@ -933,6 +936,7 @@ impl MultipartUpload for LocalUpload { Ok(PutResult { e_tag: Some(get_etag(&metadata)), version: None, + extensions: Default::default(), }) }) .await diff --git a/src/memory.rs b/src/memory.rs index 383c553e..abc08d55 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -218,6 +218,7 @@ impl ObjectStore for InMemory { Ok(PutResult { e_tag: Some(etag.to_string()), version: None, + extensions: Default::default(), }) } @@ -266,6 +267,7 @@ impl ObjectStore for InMemory { attributes: entry.attributes, meta, range, + extensions: Default::default(), }) } @@ -383,6 +385,7 @@ impl ObjectStore for InMemory { Ok(ListResult { objects, common_prefixes: common_prefixes.into_iter().collect(), + extensions: Default::default(), }) } @@ -463,6 +466,7 @@ impl MultipartStore for InMemory { Ok(PutResult { e_tag: Some(etag.to_string()), version: None, + extensions: Default::default(), }) } @@ -529,6 +533,7 @@ impl MultipartUpload for InMemoryUpload { Ok(PutResult { e_tag: Some(etag.to_string()), version: None, + extensions: Default::default(), }) } diff --git a/src/prefix.rs b/src/prefix.rs index 46cd6816..9b3cf19e 100644 --- a/src/prefix.rs +++ b/src/prefix.rs @@ -186,6 +186,7 @@ impl ObjectStore for PrefixStore { .into_iter() .map(|meta| self.strip_meta(meta)) .collect(), + extensions: lst.extensions, }) } diff --git a/src/upload.rs b/src/upload.rs index 01e6b28b..b5bc940f 100644 --- a/src/upload.rs +++ b/src/upload.rs @@ -294,6 +294,7 @@ mod tests { Ok(PutResult { e_tag: None, version: None, + extensions: Default::default(), }) }