feat(services/oss): Implement Write Returns Metadata for oss (#5688)

* feat(services/oss): Implement Write Returns Metadata for oss

* add write_has_content_md5 to Capability

Author: meteorgan

core/src/services/oss/backend.rs

@@ -460,6 +460,8 @@ impl Builder for OssBuilder {
                                 Some(usize::MAX)
                             },
                             write_with_user_metadata: true,
+                            write_has_etag: true,
+                            write_has_version: self.config.enable_versioning,
 
                             delete: true,
                             delete_with_version: self.config.enable_versioning,
@@ -541,7 +543,7 @@ impl Access for OssBackend {
                 let headers = resp.headers();
                 let mut meta = self.core.parse_metadata(path, resp.headers())?;
 
-                if let Some(v) = parse_header_to_str(headers, "x-oss-version-id")? {
+                if let Some(v) = parse_header_to_str(headers, constants::X_OSS_VERSION_ID)? {
                     meta.set_version(v);
                 }
 

core/src/services/oss/core.rs

@@ -54,6 +54,8 @@ pub mod constants {
 
     pub const X_OSS_FORBID_OVERWRITE: &str = "x-oss-forbid-overwrite";
 
+    pub const X_OSS_VERSION_ID: &str = "x-oss-version-id";
+
     pub const RESPONSE_CONTENT_DISPOSITION: &str = "response-content-disposition";
 
     pub const OSS_QUERY_VERSION_ID: &str = "versionId";

core/src/services/oss/writer.rs

@@ -17,7 +17,7 @@
 
 use std::sync::Arc;
 
-use http::StatusCode;
+use http::{HeaderMap, HeaderValue, StatusCode};
 
 use super::core::*;
 use super::error::parse_error;
@@ -41,6 +41,21 @@ impl OssWriter {
             op,
         }
     }
+
+    fn parse_metadata(headers: &HeaderMap<HeaderValue>) -> Result<Metadata> {
+        let mut meta = Metadata::default();
+        if let Some(etag) = parse_etag(headers)? {
+            meta.set_etag(etag);
+        }
+        if let Some(md5) = parse_content_md5(headers)? {
+            meta.set_content_md5(md5);
+        }
+        if let Some(version) = parse_header_to_str(headers, constants::X_OSS_VERSION_ID)? {
+            meta.set_version(version);
+        }
+
+        Ok(meta)
+    }
 }
 
 impl oio::MultipartWrite for OssWriter {
@@ -53,10 +68,11 @@ impl oio::MultipartWrite for OssWriter {
 
         let resp = self.core.send(req).await?;
 
+        let meta = Self::parse_metadata(resp.headers())?;
         let status = resp.status();
 
         match status {
-            StatusCode::CREATED | StatusCode::OK => Ok(Metadata::default()),
+            StatusCode::CREATED | StatusCode::OK => Ok(meta),
             _ => Err(parse_error(resp)),
         }
     }
@@ -145,10 +161,11 @@ impl oio::MultipartWrite for OssWriter {
             .oss_complete_multipart_upload_request(&self.path, upload_id, false, parts)
             .await?;
 
+        let meta = Self::parse_metadata(resp.headers())?;
         let status = resp.status();
 
         match status {
-            StatusCode::OK => Ok(Metadata::default()),
+            StatusCode::OK => Ok(meta),
             _ => Err(parse_error(resp)),
         }
     }
@@ -198,10 +215,11 @@ impl oio::AppendWrite for OssWriter {
 
         let resp = self.core.send(req).await?;
 
+        let meta = Self::parse_metadata(resp.headers())?;
         let status = resp.status();
 
         match status {
-            StatusCode::OK => Ok(Metadata::default()),
+            StatusCode::OK => Ok(meta),
             _ => Err(parse_error(resp)),
         }
     }

core/src/types/capability.rs

@@ -166,6 +166,8 @@ pub struct Capability {
     pub write_has_etag: bool,
     /// Indicates whether version information is available in write response
     pub write_has_version: bool,
+    /// Indicates whether content MD5 checksum is available in write response
+    pub write_has_content_md5: bool,
 
     /// Indicates if directory creation is supported.
     pub create_dir: bool,

core/tests/behavior/async_write.rs

@@ -273,6 +273,9 @@ pub async fn test_write_returns_metadata(op: Operator) -> Result<()> {
     if cap.write_has_version {
         assert_eq!(stat_meta.version(), meta.version());
     }
+    if cap.write_has_content_md5 {
+        assert_eq!(stat_meta.content_md5(), meta.content_md5());
+    }
 
     Ok(())
 }
@@ -599,6 +602,9 @@ pub async fn test_writer_return_metadata(op: Operator) -> Result<()> {
     if cap.write_has_version {
         assert_eq!(stat_meta.version(), meta.version());
     }
+    if cap.write_has_content_md5 {
+        assert_eq!(stat_meta.content_md5(), meta.content_md5());
+    }
 
     Ok(())
 }
@@ -665,6 +671,9 @@ pub async fn test_write_with_append_returns_metadata(op: Operator) -> Result<()>
     if cap.write_has_version {
         assert_eq!(stat_meta.version(), meta.version());
     }
+    if cap.write_has_content_md5 {
+        assert_eq!(stat_meta.content_md5(), meta.content_md5());
+    }
 
     Ok(())
 }

core/tests/behavior/blocking_write.rs

@@ -109,6 +109,9 @@ pub fn test_blocking_write_returns_metadata(op: BlockingOperator) -> Result<()>
     if cap.write_has_version {
         assert_eq!(meta.version(), stat_meta.version());
     }
+    if cap.write_has_content_md5 {
+        assert_eq!(meta.content_md5(), stat_meta.content_md5());
+    }
 
     Ok(())
 }
@@ -166,6 +169,9 @@ pub fn test_blocking_write_with_append_returns_metadata(op: BlockingOperator) ->
     if cap.write_has_version {
         assert_eq!(meta.version(), stat_meta.version());
     }
+    if cap.write_has_content_md5 {
+        assert_eq!(meta.content_md5(), stat_meta.content_md5());
+    }
 
     Ok(())
 }