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.

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

refactor(core): Implement RFC-2299 for reader_with #2316

Merged
merged 2 commits into from
May 24, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 27 additions & 14 deletions core/src/types/operator/operator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -552,7 +552,7 @@ impl Operator {
/// # }
/// ```
pub async fn reader(&self, path: &str) -> Result<Reader> {
self.reader_with(path, OpRead::default()).await
self.reader_with(path).await
}

/// Create a new reader which can read the specified range.
Expand All @@ -574,8 +574,7 @@ impl Operator {
/// # }
/// ```
pub async fn range_reader(&self, path: &str, range: impl RangeBounds<u64>) -> Result<Reader> {
self.reader_with(path, OpRead::new().with_range(range.into()))
.await
self.reader_with(path).range(range).await
}

/// Create a new reader with extra options
Expand All @@ -591,24 +590,38 @@ impl Operator {
/// # #[tokio::main]
/// # async fn test(op: Operator) -> Result<()> {
/// let r = op
/// .reader_with("path/to/file", OpRead::default().with_range((0..10).into()))
/// .reader_with("path/to/file")
/// .range((0..10))
/// .await?;
/// # Ok(())
/// # }
/// ```
pub async fn reader_with(&self, path: &str, args: OpRead) -> Result<Reader> {
pub fn reader_with(&self, path: &str) -> FutureReader {
let path = normalize_path(path);

if !validate_path(&path, EntryMode::FILE) {
return Err(
Error::new(ErrorKind::IsADirectory, "read path is a directory")
.with_operation("Operator::range_reader")
.with_context("service", self.info().scheme())
.with_context("path", path),
);
}
let fut = FutureReader(OperatorFuture::new(
self.inner().clone(),
path,
OpRead::default(),
|inner, path, args| {
let fut = async move {
if !validate_path(&path, EntryMode::FILE) {
return Err(Error::new(
ErrorKind::IsADirectory,
"read path is a directory",
)
.with_operation("Operator::range_reader")
.with_context("service", inner.info().scheme())
.with_context("path", path));
}

Reader::create_dir(self.inner().clone(), &path, args).await
Reader::create_dir(inner.clone(), &path, args).await
};

Box::pin(fut)
},
));
fut
}

/// Write bytes into path.
Expand Down
49 changes: 49 additions & 0 deletions core/src/types/operator/operator_futures.rs
Original file line number Diff line number Diff line change
Expand Up @@ -263,3 +263,52 @@ impl Future for FutureRead {
self.0.poll_unpin(cx)
}
}

/// Future that generated by [`Operator::reader_with`].
///
/// Users can add more options by public functions provided by this struct.
pub struct FutureReader(pub(crate) OperatorFuture<OpRead, Reader>);

impl FutureReader {
/// Set the range header for this operation.
pub fn range(mut self, range: impl RangeBounds<u64>) -> Self {
self.0 = self.0.map_args(|args| args.with_range(range.into()));
self
}

/// Sets the content-disposition header that should be send back by the remote read operation.
pub fn override_content_disposition(mut self, content_disposition: &str) -> Self {
self.0 = self
.0
.map_args(|args| args.with_override_content_disposition(content_disposition));
self
}

/// Sets the cache-control header that should be send back by the remote read operation.
pub fn override_cache_control(mut self, cache_control: &str) -> Self {
self.0 = self
.0
.map_args(|args| args.with_override_cache_control(cache_control));
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));
self
}

/// Set the If-None-Match for this operation.
pub fn if_none_match(mut self, v: &str) -> Self {
self.0 = self.0.map_args(|args| args.with_if_none_match(v));
self
}
}

impl Future for FutureReader {
type Output = Result<Reader>;

fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.0.poll_unpin(cx)
}
}