Skip to content

Commit

Permalink
feat(services/wasabi): Rewrite the method signatures using OpRead,OpW… (
Browse files Browse the repository at this point in the history
  • Loading branch information
acehinnnqru authored Sep 17, 2023
1 parent adb1bc4 commit 82a865f
Show file tree
Hide file tree
Showing 3 changed files with 22 additions and 53 deletions.
17 changes: 4 additions & 13 deletions core/src/services/wasabi/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -715,7 +715,7 @@ impl Accessor for WasabiBackend {
async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
let mut req =
self.core
.put_object_request(path, Some(0), None, None, None, AsyncBody::Empty)?;
.put_object_request(path, Some(0), &OpWrite::default(), AsyncBody::Empty)?;

self.core.sign(&mut req).await?;

Expand All @@ -733,10 +733,7 @@ impl Accessor for WasabiBackend {
}

async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self
.core
.get_object(path, args.range(), args.if_none_match())
.await?;
let resp = self.core.get_object(path, &args).await?;

let status = resp.status();

Expand Down Expand Up @@ -814,16 +811,10 @@ impl Accessor for WasabiBackend {
// We will not send this request out, just for signing.
let mut req = match args.operation() {
PresignOperation::Stat(v) => self.core.head_object_request(path, v)?,
PresignOperation::Read(v) => self.core.get_object_request(
path,
v.range(),
v.override_content_disposition(),
v.override_cache_control(),
v.if_none_match(),
)?,
PresignOperation::Read(v) => self.core.get_object_request(path, v)?,
PresignOperation::Write(_) => {
self.core
.put_object_request(path, None, None, None, None, AsyncBody::Empty)?
.put_object_request(path, None, &OpWrite::default(), AsyncBody::Empty)?
}
};

Expand Down
54 changes: 17 additions & 37 deletions core/src/services/wasabi/core.rs
Original file line number Diff line number Diff line change
Expand Up @@ -231,29 +231,22 @@ impl WasabiCore {
Ok(req)
}

pub fn get_object_request(
&self,
path: &str,
range: BytesRange,
override_content_disposition: Option<&str>,
override_cache_control: Option<&str>,
if_none_match: Option<&str>,
) -> Result<Request<AsyncBody>> {
pub fn 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_cache_control) = args.override_cache_control() {
query_args.push(format!(
"{}={}",
constants::RESPONSE_CACHE_CONTROL,
Expand All @@ -266,11 +259,12 @@ impl WasabiCore {

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(http::header::IF_NONE_MATCH, if_none_match);
}

Expand All @@ -288,10 +282,9 @@ impl WasabiCore {
pub async fn get_object(
&self,
path: &str,
range: BytesRange,
if_none_match: Option<&str>,
args: &OpRead,
) -> Result<Response<IncomingAsyncBody>> {
let mut req = self.get_object_request(path, range, None, None, if_none_match)?;
let mut req = self.get_object_request(path, args)?;

self.sign(&mut req).await?;

Expand All @@ -302,9 +295,7 @@ impl WasabiCore {
&self,
path: &str,
size: Option<usize>,
content_type: Option<&str>,
content_disposition: Option<&str>,
cache_control: Option<&str>,
args: &OpWrite,
body: AsyncBody,
) -> Result<Request<AsyncBody>> {
let p = build_abs_path(&self.root, path);
Expand All @@ -317,15 +308,15 @@ impl WasabiCore {
req = req.header(CONTENT_LENGTH, size)
}

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

if let Some(pos) = content_disposition {
if let Some(pos) = args.content_disposition() {
req = req.header(CONTENT_DISPOSITION, pos)
}

if let Some(cache_control) = cache_control {
if let Some(cache_control) = args.cache_control() {
req = req.header(CACHE_CONTROL, cache_control)
}

Expand Down Expand Up @@ -471,25 +462,23 @@ impl WasabiCore {
pub async fn initiate_multipart_upload(
&self,
path: &str,
content_type: Option<&str>,
content_disposition: Option<&str>,
cache_control: Option<&str>,
args: &OpWrite,
) -> Result<Response<IncomingAsyncBody>> {
let p = build_abs_path(&self.root, path);

let url = format!("{}/{}?uploads", self.endpoint, percent_encode_path(&p));

let mut req = Request::post(&url);

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

if let Some(content_disposition) = content_disposition {
if let Some(content_disposition) = args.content_disposition() {
req = req.header(CONTENT_DISPOSITION, content_disposition)
}

if let Some(cache_control) = cache_control {
if let Some(cache_control) = args.cache_control() {
req = req.header(CACHE_CONTROL, cache_control)
}

Expand Down Expand Up @@ -638,19 +627,10 @@ impl WasabiCore {
&self,
path: &str,
size: Option<usize>,
content_type: Option<&str>,
content_disposition: Option<&str>,
cache_control: Option<&str>,
args: &OpWrite,
body: AsyncBody,
) -> Result<Response<IncomingAsyncBody>> {
let mut req = self.put_object_request(
path,
size,
content_type,
content_disposition,
cache_control,
body,
)?;
let mut req = self.put_object_request(path, size, args, body)?;

self.sign(&mut req).await?;

Expand Down
4 changes: 1 addition & 3 deletions core/src/services/wasabi/writer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,9 +49,7 @@ impl oio::OneShotWrite for WasabiWriter {
.put_object(
&self.path,
Some(bs.len()),
self.op.content_type(),
self.op.content_disposition(),
self.op.cache_control(),
&self.op,
AsyncBody::ChunkedBytes(bs),
)
.await?;
Expand Down

0 comments on commit 82a865f

Please sign in to comment.