feat: add override_content_type (#2734)

* feat: add override_content_type

* refactor: replace s3_get_object/s3_get_object_request params with OpRead

---------

Co-authored-by: GXD <[email protected]>

core/src/raw/ops.rs

@@ -242,6 +242,7 @@ pub struct OpRead {
     br: BytesRange,
     if_match: Option<String>,
     if_none_match: Option<String>,
+    override_content_type: Option<String>,
     override_cache_control: Option<String>,
     override_content_disposition: Option<String>,
     version: Option<String>,
@@ -287,6 +288,17 @@ impl OpRead {
         self.override_cache_control.as_deref()
     }
 
+    /// Sets the content-type header that should be send back by the remote read operation.
+    pub fn with_override_content_type(mut self, content_type: &str) -> Self {
+        self.override_content_type = Some(content_type.into());
+        self
+    }
+
+    /// Returns the content-type header that should be send back by the remote read operation.
+    pub fn override_content_type(&self) -> Option<&str> {
+        self.override_content_type.as_deref()
+    }
+
     /// Set the If-Match of the option
     pub fn with_if_match(mut self, if_match: &str) -> Self {
         self.if_match = Some(if_match.to_string());

core/src/services/s3/backend.rs

@@ -921,6 +921,7 @@ impl Accessor for S3Backend {
                 read_with_if_none_match: true,
                 read_with_override_cache_control: true,
                 read_with_override_content_disposition: true,
+                read_with_override_content_type: true,
 
                 write: true,
                 write_can_sink: true,
@@ -972,16 +973,7 @@ impl Accessor for S3Backend {
     }
 
     async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
-        let resp = self
-            .core
-            .s3_get_object(
-                path,
-                args.range(),
-                args.if_none_match(),
-                args.if_match(),
-                args.override_content_disposition(),
-            )
-            .await?;
+        let resp = self.core.s3_get_object(path, args).await?;
 
         let status = resp.status();
 
@@ -1071,14 +1063,7 @@ impl Accessor for S3Backend {
                 self.core
                     .s3_head_object_request(path, v.if_none_match(), v.if_match())?
             }
-            PresignOperation::Read(v) => self.core.s3_get_object_request(
-                path,
-                v.range(),
-                v.override_content_disposition(),
-                v.override_cache_control(),
-                v.if_none_match(),
-                v.if_match(),
-            )?,
+            PresignOperation::Read(v) => self.core.s3_get_object_request(path, v.clone())?,
             PresignOperation::Write(_) => {
                 self.core
                     .s3_put_object_request(path, None, None, None, None, AsyncBody::Empty)?

core/src/services/s3/core.rs

@@ -63,6 +63,7 @@ mod constants {
         "x-amz-copy-source-server-side-encryption-customer-key-md5";
 
     pub const RESPONSE_CONTENT_DISPOSITION: &str = "response-content-disposition";
+    pub const RESPONSE_CONTENT_TYPE: &str = "response-content-type";
     pub const RESPONSE_CACHE_CONTROL: &str = "response-cache-control";
 }
 
@@ -238,30 +239,29 @@ impl S3Core {
         Ok(req)
     }
 
-    pub fn s3_get_object_request(
-        &self,
-        path: &str,
-        range: BytesRange,
-        override_content_disposition: Option<&str>,
-        override_cache_control: Option<&str>,
-        if_none_match: Option<&str>,
-        if_match: Option<&str>,
-    ) -> Result<Request<AsyncBody>> {
+    pub fn s3_get_object_request(&self, path: &str, args: OpRead) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
 
         // Construct headers to add to the request
         let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
 
         // Add query arguments to the URL based on response overrides
         let mut query_args = Vec::new();
-        if let Some(override_content_disposition) = override_content_disposition {
+        if let Some(override_content_disposition) = args.override_content_disposition() {
             query_args.push(format!(
                 "{}={}",
                 constants::RESPONSE_CONTENT_DISPOSITION,
                 percent_encode_path(override_content_disposition)
             ))
         }
-        if let Some(override_cache_control) = override_cache_control {
+        if let Some(override_content_type) = args.override_content_type() {
+            query_args.push(format!(
+                "{}={}",
+                constants::RESPONSE_CONTENT_TYPE,
+                percent_encode_path(override_content_type)
+            ))
+        }
+        if let Some(override_cache_control) = args.override_cache_control() {
             query_args.push(format!(
                 "{}={}",
                 constants::RESPONSE_CACHE_CONTROL,
@@ -274,15 +274,16 @@ impl S3Core {
 
         let mut req = Request::get(&url);
 
+        let range = args.range();
         if !range.is_full() {
             req = req.header(http::header::RANGE, range.to_header());
         }
 
-        if let Some(if_none_match) = if_none_match {
+        if let Some(if_none_match) = args.if_none_match() {
             req = req.header(IF_NONE_MATCH, if_none_match);
         }
 
-        if let Some(if_match) = if_match {
+        if let Some(if_match) = args.if_match() {
             req = req.header(IF_MATCH, if_match);
         }
         // Set SSE headers.
@@ -299,19 +300,9 @@ impl S3Core {
     pub async fn s3_get_object(
         &self,
         path: &str,
-        range: BytesRange,
-        if_none_match: Option<&str>,
-        if_match: Option<&str>,
-        override_content_disposition: Option<&str>,
+        args: OpRead,
     ) -> Result<Response<IncomingAsyncBody>> {
-        let mut req = self.s3_get_object_request(
-            path,
-            range,
-            override_content_disposition,
-            None,
-            if_none_match,
-            if_match,
-        )?;
+        let mut req = self.s3_get_object_request(path, args)?;
 
         self.sign(&mut req).await?;
 

core/src/types/capability.rs

@@ -71,6 +71,8 @@ pub struct Capability {
     pub read_with_override_cache_control: bool,
     /// if operator supports read with override content disposition natively, it will be true.
     pub read_with_override_content_disposition: bool,
+    /// if operator supports read with override content type natively, it will be true.
+    pub read_with_override_content_type: bool,
 
     /// If operator supports write natively, it will be true.
     pub write: bool,

core/src/types/operator/operator_futures.rs

@@ -251,6 +251,14 @@ impl FuturePresignRead {
         self
     }
 
+    /// Sets the content-type header that should be send back by the remote read operation.
+    pub fn override_content_type(mut self, v: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|(args, dur)| (args.with_override_content_type(v), dur));
+        self
+    }
+
     /// Set the If-Match of the option
     pub fn if_match(mut self, v: &str) -> Self {
         self.0 = self.0.map_args(|(args, dur)| (args.with_if_match(v), dur));
@@ -352,6 +360,14 @@ impl FutureRead {
         self
     }
 
+    /// Sets the content-type header that should be send back by the remote read operation.
+    pub fn override_content_type(mut self, content_type: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|args| args.with_override_content_type(content_type));
+        self
+    }
+
     /// Set the If-Match for this operation.
     pub fn if_match(mut self, v: &str) -> Self {
         self.0 = self.0.map_args(|args| args.with_if_match(v));
@@ -407,6 +423,14 @@ impl FutureReader {
         self
     }
 
+    /// Sets the content-type header that should be send back by the remote read operation.
+    pub fn override_content_type(mut self, content_type: &str) -> Self {
+        self.0 = self
+            .0
+            .map_args(|args| args.with_override_content_type(content_type));
+        self
+    }
+
     /// Set the If-Match for this operation.
     pub fn if_match(mut self, v: &str) -> Self {
         self.0 = self.0.map_args(|args| args.with_if_match(v));

core/tests/behavior/write.rs

@@ -75,6 +75,7 @@ pub fn behavior_write_tests(op: &Operator) -> Vec<Trial> {
         test_read_with_special_chars,
         test_read_with_override_cache_control,
         test_read_with_override_content_disposition,
+        test_read_with_override_content_type,
         test_delete_file,
         test_delete_empty_dir,
         test_delete_with_special_chars,
@@ -927,6 +928,54 @@ pub async fn test_read_with_override_content_disposition(op: Operator) -> Result
     Ok(())
 }
 
+/// Read file with override_content_type should succeed.
+pub async fn test_read_with_override_content_type(op: Operator) -> Result<()> {
+    if !(op.info().capability().read_with_override_content_type && op.info().can_presign()) {
+        return Ok(());
+    }
+
+    let path = uuid::Uuid::new_v4().to_string();
+    let (content, _) = gen_bytes();
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+
+    let target_content_type = "application/opendal";
+
+    let signed_req = op
+        .presign_read_with(&path, Duration::from_secs(60))
+        .override_content_type(target_content_type)
+        .await
+        .expect("presign must succeed");
+
+    let client = reqwest::Client::new();
+    let mut req = client.request(
+        signed_req.method().clone(),
+        Url::from_str(&signed_req.uri().to_string()).expect("must be valid url"),
+    );
+    for (k, v) in signed_req.header() {
+        req = req.header(k, v);
+    }
+
+    let resp = req.send().await.expect("send must succeed");
+
+    assert_eq!(resp.status(), StatusCode::OK);
+    assert_eq!(
+        resp.headers()
+            .get(http::header::CONTENT_TYPE)
+            .expect("content-type header must exist")
+            .to_str()
+            .expect("content-type header must be string"),
+        target_content_type
+    );
+    assert_eq!(resp.bytes().await?, content);
+
+    op.delete(&path).await.expect("delete must succeed");
+
+    Ok(())
+}
+
 /// Delete existing file should succeed.
 pub async fn test_writer_abort(op: Operator) -> Result<()> {
     let path = uuid::Uuid::new_v4().to_string();