feat: content-type configuration

Cargo.toml

@@ -107,6 +107,7 @@ http = "0.2"
 log = "0.4"
 md-5 = "0.10"
 metrics = { version = "0.20", optional = true }
+mime = "0.3"
 moka = { version = "0.9", optional = true, features = ["future"] }
 once_cell = "1"
 parking_lot = "0.12"

src/object/object.rs

@@ -871,6 +871,95 @@ impl Object {
         Ok(())
     }
 
+    /// Write data with option described in OpenDAL [rfc-0661](../../docs/rfcs/0661-path-in-accessor.md)
+    ///
+    /// # Notes
+    ///
+    /// - Write will make sure all bytes has been written, or an error will be returned.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use opendal::services::s3;
+    /// # use opendal::ops::OpWrite;
+    /// # use std::io::Result;
+    /// # use opendal::Operator;
+    /// # use futures::StreamExt;
+    /// # use futures::SinkExt;
+    /// # use opendal::Scheme;
+    /// use bytes::Bytes;
+    /// use mime::TEXT_PLAIN_UTF_8;
+    ///
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<()> {
+    /// let op = Operator::from_env(Scheme::S3)?;
+    /// let o = op.object("path/to/file");
+    /// let bs = b"hello, world!".to_vec();
+    /// let args = OpWrite::new(bs.len() as u64).with_content_type(TEXT_PLAIN_UTF_8);
+    /// let _ = o.write_with(args, bs).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn write_with(&self, args: OpWrite, bs: impl Into<Vec<u8>>) -> Result<()> {
+        if !validate_path(self.path(), ObjectMode::FILE) {
+            return Err(new_other_object_error(
+                Operation::Write,
+                self.path(),
+                anyhow!("Is a directory"),
+            ));
+        }
+
+        let bs = bs.into();
+        let r = Cursor::new(bs);
+        let _ = self.acc.write(self.path(), args, Box::new(r)).await?;
+
+        Ok(())
+    }
+
+    /// Write data with option described in OpenDAL [rfc-0661](../../docs/rfcs/0661-path-in-accessor.md)
+    ///
+    /// # Notes
+    ///
+    /// - Write will make sure all bytes has been written, or an error will be returned.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use opendal::services::s3;
+    /// # use opendal::ops::OpWrite;
+    /// # use std::io::Result;
+    /// # use opendal::Operator;
+    /// # use futures::StreamExt;
+    /// # use futures::SinkExt;
+    /// # use opendal::Scheme;
+    /// use bytes::Bytes;
+    /// use mime::TEXT_PLAIN_UTF_8;
+    ///
+    /// # fn main() -> Result<()> {
+    /// # let op = Operator::from_env(Scheme::S3)?;
+    /// let o = op.object("hello.txt");
+    /// let bs = b"hello, world!".to_vec();
+    /// let ow = OpWrite::new(bs.len() as u64).with_content_type(TEXT_PLAIN_UTF_8);
+    /// let _ = o.blocking_write_with(ow, bs)?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn blocking_write_with(&self, args: OpWrite, bs: impl Into<Vec<u8>>) -> Result<()> {
+        if !validate_path(self.path(), ObjectMode::FILE) {
+            return Err(new_other_object_error(
+                Operation::Write,
+                self.path(),
+                anyhow!("Is a directory"),
+            ));
+        }
+
+        let bs = bs.into();
+        let r = std::io::Cursor::new(bs);
+
+        let _ = self.acc.blocking_write(self.path(), args, Box::new(r))?;
+        Ok(())
+    }
+
     /// Delete object.
     ///
     /// # Notes

src/ops/op_write.rs

@@ -12,22 +12,61 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use mime::Mime;
+
 /// Args for `write` operation.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone)]
 pub struct OpWrite {
     size: u64,
+    content_type: Mime,
+}
+
+impl Default for OpWrite {
+    fn default() -> Self {
+        Self {
+            size: 0,
+            /// MIME type of content
+            ///
+            /// # NOTE
+            /// > Generic binary data (or binary data whose true type is unknown) is application/octet-stream
+            /// --- [MDN Reference](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
+            content_type: mime::APPLICATION_OCTET_STREAM,
+        }
+    }
 }
 
 impl OpWrite {
     /// Create a new `OpWrite`.
     ///
     /// If input path is not a file path, an error will be returned.
     pub fn new(size: u64) -> Self {
-        Self { size }
+        Self {
+            size,
+            /// MIME type of content
+            ///
+            /// # NOTE
+            /// > Generic binary data (or binary data whose true type is unknown) is application/octet-stream
+            /// --- [MDN Reference](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
+            content_type: mime::APPLICATION_OCTET_STREAM,
+        }
     }
 
+    /// Set the content type of option
+    pub fn with_content_type(self, content_type: Mime) -> Self {
+        Self {
+            size: self.size(),
+            content_type,
+        }
+    }
+}
+
+impl OpWrite {
     /// Get size from option.
     pub fn size(&self) -> u64 {
         self.size
     }
+    /// Get the content type from option
+    pub fn content_type(&self) -> Mime {
+        self.content_type.clone()
+    }
 }

src/services/azblob/backend.rs

@@ -24,11 +24,13 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use http::header::HeaderName;
 use http::header::CONTENT_LENGTH;
+use http::header::CONTENT_TYPE;
 use http::Request;
 use http::Response;
 use http::StatusCode;
 use log::debug;
 use log::info;
+use mime::Mime;
 use reqsign::AzureStorageSigner;
 
 use super::dir_stream::DirStream;
@@ -255,7 +257,7 @@ impl Accessor for Backend {
     }
 
     async fn create(&self, path: &str, _: OpCreate) -> Result<()> {
-        let mut req = self.azblob_put_blob_request(path, Some(0), AsyncBody::Empty)?;
+        let mut req = self.azblob_put_blob_request(path, Some(0), None, AsyncBody::Empty)?;
 
         self.signer
             .sign(&mut req)
@@ -303,8 +305,12 @@ impl Accessor for Backend {
     }
 
     async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
-        let mut req =
-            self.azblob_put_blob_request(path, Some(args.size()), AsyncBody::Reader(r))?;
+        let mut req = self.azblob_put_blob_request(
+            path,
+            Some(args.size()),
+            Some(args.content_type()),
+            AsyncBody::Reader(r),
+        )?;
 
         self.signer
             .sign(&mut req)
@@ -457,6 +463,7 @@ impl Backend {
         &self,
         path: &str,
         size: Option<u64>,
+        content_type: Option<Mime>,
         body: AsyncBody,
     ) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
@@ -474,6 +481,10 @@ impl Backend {
             req = req.header(CONTENT_LENGTH, size)
         }
 
+        if let Some(ty) = content_type {
+            req = req.header(CONTENT_TYPE, ty.to_string())
+        }
+
         req = req.header(HeaderName::from_static(X_MS_BLOB_TYPE), "BlockBlob");
 
         // Set body

src/services/gcs/backend.rs

@@ -22,11 +22,13 @@ use std::sync::Arc;
 use anyhow::anyhow;
 use async_trait::async_trait;
 use http::header::CONTENT_LENGTH;
+use http::header::CONTENT_TYPE;
 use http::Request;
 use http::Response;
 use http::StatusCode;
 use log::debug;
 use log::info;
+use mime::Mime;
 use reqsign::GoogleSigner;
 use serde::Deserialize;
 use serde_json;
@@ -240,7 +242,7 @@ impl Accessor for Backend {
     }
 
     async fn create(&self, path: &str, _: OpCreate) -> Result<()> {
-        let mut req = self.gcs_insert_object_request(path, Some(0), AsyncBody::Empty)?;
+        let mut req = self.gcs_insert_object_request(path, Some(0), None, AsyncBody::Empty)?;
 
         self.signer
             .sign(&mut req)
@@ -280,8 +282,12 @@ impl Accessor for Backend {
     }
 
     async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
-        let mut req =
-            self.gcs_insert_object_request(path, Some(args.size()), AsyncBody::Reader(r))?;
+        let mut req = self.gcs_insert_object_request(
+            path,
+            Some(args.size()),
+            Some(args.content_type()),
+            AsyncBody::Reader(r),
+        )?;
 
         self.signer
             .sign(&mut req)
@@ -439,6 +445,7 @@ impl Backend {
         &self,
         path: &str,
         size: Option<u64>,
+        content_type: Option<Mime>,
         body: AsyncBody,
     ) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
@@ -456,6 +463,10 @@ impl Backend {
             req = req.header(CONTENT_LENGTH, size)
         }
 
+        if let Some(mime) = content_type {
+            req = req.header(CONTENT_TYPE, mime.to_string())
+        }
+
         // Set body
         let req = req
             .body(body)

src/services/obs/backend.rs

@@ -21,12 +21,14 @@ use std::sync::Arc;
 use anyhow::anyhow;
 use async_trait::async_trait;
 use http::header::CONTENT_LENGTH;
+use http::header::CONTENT_TYPE;
 use http::Request;
 use http::Response;
 use http::StatusCode;
 use http::Uri;
 use log::debug;
 use log::info;
+use mime::Mime;
 use reqsign::HuaweicloudObsSigner;
 
 use super::error::parse_error;
@@ -279,7 +281,7 @@ impl Accessor for Backend {
     }
 
     async fn create(&self, path: &str, _: OpCreate) -> Result<()> {
-        let mut req = self.obs_put_object_request(path, Some(0), AsyncBody::Empty)?;
+        let mut req = self.obs_put_object_request(path, Some(0), None, AsyncBody::Empty)?;
 
         self.signer
             .sign(&mut req)
@@ -327,7 +329,12 @@ impl Accessor for Backend {
     }
 
     async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
-        let mut req = self.obs_put_object_request(path, Some(args.size()), AsyncBody::Reader(r))?;
+        let mut req = self.obs_put_object_request(
+            path,
+            Some(args.size()),
+            Some(args.content_type()),
+            AsyncBody::Reader(r),
+        )?;
 
         self.signer
             .sign(&mut req)
@@ -471,6 +478,7 @@ impl Backend {
         &self,
         path: &str,
         size: Option<u64>,
+        content_type: Option<Mime>,
         body: AsyncBody,
     ) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
@@ -483,6 +491,10 @@ impl Backend {
             req = req.header(CONTENT_LENGTH, size)
         }
 
+        if let Some(mime) = content_type {
+            req = req.header(CONTENT_TYPE, mime.to_string())
+        }
+
         let req = req
             .body(body)
             .map_err(|e| new_request_build_error(Operation::Write, path, e))?;