Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: content-type configuration #878

Merged
merged 5 commits into from
Oct 18, 2022
Merged

feat: content-type configuration #878

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
89f7fea
feat: implement sync/async `write_with`
ClSlaid Oct 17, 2022
39b5e8e
refactor: directly move
ClSlaid Oct 17, 2022
7046000
refactor: make users config mime types by themselves
ClSlaid Oct 18, 2022
20f74f1
Merge remote-tracking branch 'opendal/main' into write-path-in-access…
ClSlaid Oct 18, 2022
2e74692
refactor: use Option for OpWrite
ClSlaid Oct 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
75 changes: 71 additions & 4 deletions src/object/object.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -727,6 +727,40 @@ impl Object {
/// # }
/// ```
pub async fn write(&self, bs: impl Into<Vec<u8>>) -> Result<()> {
let bs: Vec<u8> = bs.into();
let op = OpWrite::new(bs.len() as u64);
self.write_with(op, bs).await
}

/// 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;
///
/// # #[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");
/// 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,
Expand All @@ -736,9 +770,9 @@ impl Object {
}

let bs = bs.into();
let op = OpWrite::new(bs.len() as u64);
let r = Cursor::new(bs);
let _ = self.acc.write(self.path(), op, Box::new(r)).await?;
let _ = self.acc.write(self.path(), args, Box::new(r)).await?;

Ok(())
}

Expand Down Expand Up @@ -767,6 +801,39 @@ impl Object {
/// # }
/// ```
pub fn blocking_write(&self, bs: impl Into<Vec<u8>>) -> Result<()> {
let bs: Vec<u8> = bs.into();
let op = OpWrite::new(bs.len() as u64);
self.blocking_write_with(op, bs)
}

/// 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;
///
/// # 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");
/// 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,
Expand All @@ -776,9 +843,9 @@ impl Object {
}

let bs = bs.into();
let op = OpWrite::new(bs.len() as u64);
let r = std::io::Cursor::new(bs);
let _ = self.acc.blocking_write(self.path(), op, Box::new(r))?;

let _ = self.acc.blocking_write(self.path(), args, Box::new(r))?;
Ok(())
}

Expand Down
20 changes: 19 additions & 1 deletion src/ops/op_write.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,18 +16,36 @@
#[derive(Debug, Clone, Default)]
pub struct OpWrite {
size: u64,
content_type: String,
Xuanwo marked this conversation as resolved.
Show resolved Hide resolved
}

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,
content_type: Default::default(),
Xuanwo marked this conversation as resolved.
Show resolved Hide resolved
}
}

/// Set the content type of option
pub fn with_content_type(self, content_type: &str) -> Self {
Self {
size: self.size(),
content_type: content_type.to_string(),
}
}
}

impl OpWrite {
/// Get size from option.
pub fn size(&self) -> u64 {
self.size
}
/// Get the content type from option
pub fn content_type(&self) -> String {
Xuanwo marked this conversation as resolved.
Show resolved Hide resolved
self.content_type.clone()
}
}
18 changes: 15 additions & 3 deletions src/services/azblob/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ 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;
Expand Down Expand Up @@ -255,7 +256,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)?;
Xuanwo marked this conversation as resolved.
Show resolved Hide resolved

self.signer
.sign(&mut req)
Expand Down Expand Up @@ -303,8 +304,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()),
ClSlaid marked this conversation as resolved.
Show resolved Hide resolved
Xuanwo marked this conversation as resolved.
Show resolved Hide resolved
AsyncBody::Reader(r),
)?;

self.signer
.sign(&mut req)
Expand Down Expand Up @@ -457,6 +462,7 @@ impl Backend {
&self,
path: &str,
size: Option<u64>,
content_type: Option<String>,
body: AsyncBody,
) -> Result<Request<AsyncBody>> {
let p = build_abs_path(&self.root, path);
Expand All @@ -474,6 +480,12 @@ impl Backend {
req = req.header(CONTENT_LENGTH, size)
}

if let Some(ty) = content_type {
if !ty.is_empty() {
req = req.header(CONTENT_TYPE, ty)
}
}

req = req.header(HeaderName::from_static(X_MS_BLOB_TYPE), "BlockBlob");

// Set body
Expand Down
18 changes: 15 additions & 3 deletions src/services/gcs/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ 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;
Expand Down Expand Up @@ -240,7 +241,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)
Expand Down Expand Up @@ -280,8 +281,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)
Expand Down Expand Up @@ -439,6 +444,7 @@ impl Backend {
&self,
path: &str,
size: Option<u64>,
content_type: Option<String>,
body: AsyncBody,
) -> Result<Request<AsyncBody>> {
let p = build_abs_path(&self.root, path);
Expand All @@ -456,6 +462,12 @@ impl Backend {
req = req.header(CONTENT_LENGTH, size)
}

if let Some(mime) = content_type {
if !mime.is_empty() {
req = req.header(CONTENT_TYPE, mime)
}
}

// Set body
let req = req
.body(body)
Expand Down
17 changes: 15 additions & 2 deletions src/services/obs/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ 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;
Expand Down Expand Up @@ -279,7 +280,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)
Expand Down Expand Up @@ -327,7 +328,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)
Expand Down Expand Up @@ -471,6 +477,7 @@ impl Backend {
&self,
path: &str,
size: Option<u64>,
content_type: Option<String>,
body: AsyncBody,
) -> Result<Request<AsyncBody>> {
let p = build_abs_path(&self.root, path);
Expand All @@ -483,6 +490,12 @@ impl Backend {
req = req.header(CONTENT_LENGTH, size)
}

if let Some(mime) = content_type {
if !mime.is_empty() {
req = req.header(CONTENT_TYPE, mime)
}
}

let req = req
.body(body)
.map_err(|e| new_request_build_error(Operation::Write, path, e))?;
Expand Down
21 changes: 18 additions & 3 deletions src/services/oss/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,7 @@ impl Backend {
&self,
path: &str,
size: Option<u64>,
content_type: Option<String>,
body: AsyncBody,
) -> Result<Request<AsyncBody>> {
let p = build_abs_path(&self.root, path);
Expand All @@ -304,6 +305,12 @@ impl Backend {
.header(HOST, &self.host)
.header(CONTENT_LENGTH, size.unwrap_or_default());

if let Some(mime) = content_type {
if !mime.is_empty() {
req = req.header(CONTENT_TYPE, mime)
}
}

let req = req
.body(body)
.map_err(|e| new_request_build_error(Operation::Write, path, e))?;
Expand Down Expand Up @@ -422,9 +429,10 @@ impl Backend {
&self,
path: &str,
size: Option<u64>,
content_type: Option<String>,
body: AsyncBody,
) -> Result<Response<IncomingAsyncBody>> {
let mut req = self.oss_put_object_request(path, size, body)?;
let mut req = self.oss_put_object_request(path, size, content_type, body)?;

self.signer
.sign(&mut req)
Expand Down Expand Up @@ -477,7 +485,9 @@ impl Accessor for Backend {
}

async fn create(&self, path: &str, _: OpCreate) -> Result<()> {
let resp = self.oss_put_object(path, None, AsyncBody::Empty).await?;
let resp = self
.oss_put_object(path, None, None, AsyncBody::Empty)
.await?;
let status = resp.status();

match status {
Expand Down Expand Up @@ -515,7 +525,12 @@ impl Accessor for Backend {

async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
let resp = self
.oss_put_object(path, Some(args.size()), AsyncBody::Reader(r))
.oss_put_object(
path,
Some(args.size()),
Some(args.content_type()),
AsyncBody::Reader(r),
)
.await?;

let status = resp.status();
Expand Down