This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 48383aa536 refactor: Implement `IntoFuture` for operator futures to
remove an alloc (#4098)
48383aa536 is described below
commit 48383aa5369afe97533570e40b055769b2ff3484
Author: Xuanwo <[email protected]>
AuthorDate: Tue Jan 30 16:55:07 2024 +0800
refactor: Implement `IntoFuture` for operator futures to remove an alloc
(#4098)
* Save work
Signed-off-by: Xuanwo <[email protected]>
* Save
Signed-off-by: Xuanwo <[email protected]>
* Fix build
Signed-off-by: Xuanwo <[email protected]>
---------
Signed-off-by: Xuanwo <[email protected]>
---
core/src/types/operator/operator.rs | 315 +++++++--------
core/src/types/operator/operator_futures.rs | 583 +++++++---------------------
2 files changed, 286 insertions(+), 612 deletions(-)
diff --git a/core/src/types/operator/operator.rs
b/core/src/types/operator/operator.rs
index 8ba9899b84..8c50d7db4c 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::time::Duration;
use bytes::Buf;
@@ -280,24 +281,18 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn stat_with(&self, path: &str) -> FutureStat {
+ pub fn stat_with(&self, path: &str) -> FutureStat<impl Future<Output =
Result<Metadata>>> {
let path = normalize_path(path);
- let fut = FutureStat(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
OpStat::default(),
- |inner, path, args| {
- let fut = async move {
- let rp = inner.stat(&path, args).await?;
- Ok(rp.into_metadata())
- };
-
- Box::pin(fut)
+ |inner, path, args| async move {
+ let rp = inner.stat(&path, args).await?;
+ Ok(rp.into_metadata())
},
- ));
-
- fut
+ )
}
/// Check if this path exists or not.
@@ -409,50 +404,43 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn read_with(&self, path: &str) -> FutureRead {
+ pub fn read_with(&self, path: &str) -> FutureRead<impl Future<Output =
Result<Vec<u8>>>> {
let path = normalize_path(path);
- let fut = FutureRead(OperatorFuture::new(
+ 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("read")
- .with_context("service", inner.info().scheme())
- .with_context("path", &path));
- }
-
- let range = args.range();
- let (size_hint, range) = if let Some(size) = range.size() {
- (size, range)
- } else {
- let size = inner
- .stat(&path, OpStat::default())
- .await?
- .into_metadata()
- .content_length();
- let range = range.complete(size);
- (range.size().unwrap(), range)
- };
-
- let (_, mut s) = inner.read(&path,
args.with_range(range)).await?;
- let mut buf = Vec::with_capacity(size_hint as usize);
- s.read_to_end(&mut buf).await?;
-
- Ok(buf)
+ |inner, path, args| async move {
+ if !validate_path(&path, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "read path is a
directory")
+ .with_operation("read")
+ .with_context("service", inner.info().scheme())
+ .with_context("path", &path),
+ );
+ }
+
+ let range = args.range();
+ let (size_hint, range) = if let Some(size) = range.size() {
+ (size, range)
+ } else {
+ let size = inner
+ .stat(&path, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length();
+ let range = range.complete(size);
+ (range.size().unwrap(), range)
};
- Box::pin(fut)
- },
- ));
+ let (_, mut s) = inner.read(&path,
args.with_range(range)).await?;
+ let mut buf = Vec::with_capacity(size_hint as usize);
+ s.read_to_end(&mut buf).await?;
- fut
+ Ok(buf)
+ },
+ )
}
/// Create a new reader which can read the whole path.
@@ -489,32 +477,26 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn reader_with(&self, path: &str) -> FutureReader {
+ pub fn reader_with(&self, path: &str) -> FutureRead<impl Future<Output =
Result<Reader>>> {
let path = normalize_path(path);
- let fut = FutureReader(OperatorFuture::new(
+ 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::reader")
- .with_context("service", inner.info().scheme())
- .with_context("path", path));
- }
-
- Reader::create(inner.clone(), &path, args).await
- };
+ |inner, path, args| async move {
+ if !validate_path(&path, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "read path is a
directory")
+ .with_operation("Operator::reader")
+ .with_context("service", inner.info().scheme())
+ .with_context("path", path),
+ );
+ }
- Box::pin(fut)
+ Reader::create(inner.clone(), &path, args).await
},
- ));
- fut
+ )
}
/// Write bytes into path.
@@ -711,32 +693,26 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn writer_with(&self, path: &str) -> FutureWriter {
+ pub fn writer_with(&self, path: &str) -> FutureWriter<impl Future<Output =
Result<Writer>>> {
let path = normalize_path(path);
- let fut = FutureWriter(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
OpWrite::default(),
- |inner, path, args| {
- let fut = async move {
- if !validate_path(&path, EntryMode::FILE) {
- return Err(Error::new(
- ErrorKind::IsADirectory,
- "write path is a directory",
- )
- .with_operation("Operator::writer")
- .with_context("service",
inner.info().scheme().into_static())
- .with_context("path", &path));
- }
-
- Writer::create(inner, &path, args).await
- };
- Box::pin(fut)
- },
- ));
+ |inner, path, args| async move {
+ if !validate_path(&path, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "write path is a
directory")
+ .with_operation("Operator::writer")
+ .with_context("service",
inner.info().scheme().into_static())
+ .with_context("path", &path),
+ );
+ }
- fut
+ Writer::create(inner, &path, args).await
+ },
+ )
}
/// Write data with extra options.
@@ -762,40 +738,39 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn write_with(&self, path: &str, bs: impl Into<Bytes>) -> FutureWrite {
+ pub fn write_with(
+ &self,
+ path: &str,
+ bs: impl Into<Bytes>,
+ ) -> FutureWrite<impl Future<Output = Result<()>>> {
let path = normalize_path(path);
let bs = bs.into();
- let fut = FutureWrite(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
(OpWrite::default(), bs),
- |inner, path, (args, mut bs)| {
- let fut = async move {
- if !validate_path(&path, EntryMode::FILE) {
- return Err(Error::new(
- ErrorKind::IsADirectory,
- "write path is a directory",
- )
- .with_operation("Operator::write_with")
- .with_context("service",
inner.info().scheme().into_static())
- .with_context("path", &path));
- }
-
- let (_, mut w) = inner.write(&path, args).await?;
- while bs.remaining() > 0 {
- let n = w.write(&bs).await?;
- bs.advance(n);
- }
-
- w.close().await?;
-
- Ok(())
- };
- Box::pin(fut)
+ |inner, path, (args, mut bs)| async move {
+ if !validate_path(&path, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "write path is a
directory")
+ .with_operation("Operator::write_with")
+ .with_context("service",
inner.info().scheme().into_static())
+ .with_context("path", &path),
+ );
+ }
+
+ let (_, mut w) = inner.write(&path, args).await?;
+ while bs.remaining() > 0 {
+ let n = w.write(&bs).await?;
+ bs.advance(n);
+ }
+
+ w.close().await?;
+
+ Ok(())
},
- ));
- fut
+ )
}
/// Delete the given path.
@@ -839,24 +814,18 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn delete_with(&self, path: &str) -> FutureDelete {
+ pub fn delete_with(&self, path: &str) -> FutureDelete<impl Future<Output =
Result<()>>> {
let path = normalize_path(path);
- let fut = FutureDelete(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
OpDelete::default(),
- |inner, path, args| {
- let fut = async move {
- let _ = inner.delete(&path, args).await?;
- Ok(())
- };
-
- Box::pin(fut)
+ |inner, path, args| async move {
+ let _ = inner.delete(&path, args).await?;
+ Ok(())
},
- ));
-
- fut
+ )
}
///
@@ -1253,23 +1222,19 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn list_with(&self, path: &str) -> FutureList {
+ pub fn list_with(&self, path: &str) -> FutureList<impl Future<Output =
Result<Vec<Entry>>>> {
let path = normalize_path(path);
- let fut = FutureList(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
OpList::default(),
- |inner, path, args| {
- let fut = async move {
- let lister = Lister::create(inner, &path, args).await?;
+ |inner, path, args| async move {
+ let lister = Lister::create(inner, &path, args).await?;
- lister.try_collect().await
- };
- Box::pin(fut)
+ lister.try_collect().await
},
- ));
- fut
+ )
}
/// List entries that starts with given `path` in parent dir.
@@ -1474,19 +1439,15 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn lister_with(&self, path: &str) -> FutureLister {
+ pub fn lister_with(&self, path: &str) -> FutureList<impl Future<Output =
Result<Lister>>> {
let path = normalize_path(path);
- let fut = FutureLister(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
OpList::default(),
- |inner, path, args| {
- let fut = async move { Lister::create(inner, &path,
args).await };
- Box::pin(fut)
- },
- ));
- fut
+ |inner, path, args| async move { Lister::create(inner, &path,
args).await },
+ )
}
}
@@ -1538,23 +1499,23 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn presign_stat_with(&self, path: &str, expire: Duration) ->
FuturePresignStat {
+ pub fn presign_stat_with(
+ &self,
+ path: &str,
+ expire: Duration,
+ ) -> FuturePresignStat<impl Future<Output = Result<PresignedRequest>>> {
let path = normalize_path(path);
- let fut = FuturePresignStat(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
(OpStat::default(), expire),
- |inner, path, (args, dur)| {
- let fut = async move {
- let op = OpPresign::new(args, dur);
- let rp = inner.presign(&path, op).await?;
- Ok(rp.into_presigned_request())
- };
- Box::pin(fut)
+ |inner, path, (args, dur)| async move {
+ let op = OpPresign::new(args, dur);
+ let rp = inner.presign(&path, op).await?;
+ Ok(rp.into_presigned_request())
},
- ));
- fut
+ )
}
/// Presign an operation for read.
@@ -1613,23 +1574,23 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn presign_read_with(&self, path: &str, expire: Duration) ->
FuturePresignRead {
+ pub fn presign_read_with(
+ &self,
+ path: &str,
+ expire: Duration,
+ ) -> FuturePresignRead<impl Future<Output = Result<PresignedRequest>>> {
let path = normalize_path(path);
- let fut = FuturePresignRead(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
(OpRead::default(), expire),
- |inner, path, (args, dur)| {
- let fut = async move {
- let op = OpPresign::new(args, dur);
- let rp = inner.presign(&path, op).await?;
- Ok(rp.into_presigned_request())
- };
- Box::pin(fut)
+ |inner, path, (args, dur)| async move {
+ let op = OpPresign::new(args, dur);
+ let rp = inner.presign(&path, op).await?;
+ Ok(rp.into_presigned_request())
},
- ));
- fut
+ )
}
/// Presign an operation for write.
@@ -1686,22 +1647,22 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn presign_write_with(&self, path: &str, expire: Duration) ->
FuturePresignWrite {
+ pub fn presign_write_with(
+ &self,
+ path: &str,
+ expire: Duration,
+ ) -> FuturePresignWrite<impl Future<Output = Result<PresignedRequest>>> {
let path = normalize_path(path);
- let fut = FuturePresignWrite(OperatorFuture::new(
+ OperatorFuture::new(
self.inner().clone(),
path,
(OpWrite::default(), expire),
- |inner, path, (args, dur)| {
- let fut = async move {
- let op = OpPresign::new(args, dur);
- let rp = inner.presign(&path, op).await?;
- Ok(rp.into_presigned_request())
- };
- Box::pin(fut)
+ |inner, path, (args, dur)| async move {
+ let op = OpPresign::new(args, dur);
+ let rp = inner.presign(&path, op).await?;
+ Ok(rp.into_presigned_request())
},
- ));
- fut
+ )
}
}
diff --git a/core/src/types/operator/operator_futures.rs
b/core/src/types/operator/operator_futures.rs
index d8c090a618..15ce16e0c6 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -19,418 +19,242 @@
//!
//! By using futures, users can add more options for operation.
-use std::mem;
+use std::future::IntoFuture;
use std::ops::RangeBounds;
-use std::pin::Pin;
-use std::task::Context;
-use std::task::Poll;
use std::time::Duration;
use bytes::Bytes;
use flagset::FlagSet;
use futures::Future;
-use futures::FutureExt;
use crate::raw::*;
use crate::*;
/// OperatorFuture is the future generated by [`Operator`].
///
-/// The future will consume all the input to generate a result.
-pub(crate) enum OperatorFuture<T, F> {
- /// Idle state, waiting for the future to be polled
- Idle(
- /// The accessor to the underlying object storage
- FusedAccessor,
- /// The path of string
- String,
- /// The input args
- T,
- /// The function which will move all the args and return a static
future
- fn(FusedAccessor, String, T) -> BoxedFuture<Result<F>>,
- ),
- /// Polling state, waiting for the future to be ready
- Poll(BoxedFuture<Result<F>>),
- /// Empty state, the future has been polled and completed or
- /// something is broken during state switch.
- Empty,
+/// The future will consume all the input to generate a future.
+///
+/// # NOTES
+///
+/// This struct is by design to keep in crate. We don't want
+/// users to use this struct directly.
+pub struct OperatorFuture<T, F> {
+ /// The accessor to the underlying object storage
+ acc: FusedAccessor,
+ /// The path of string
+ path: String,
+ /// The input args
+ args: T,
+ /// The function which will move all the args and return a static future
+ f: fn(FusedAccessor, String, T) -> F,
}
-impl<T, F> OperatorFuture<T, F> {
- pub fn new(
+impl<T, F: Future> OperatorFuture<T, F> {
+ /// # NOTES
+ ///
+ /// This struct is by design to keep in crate. We don't want
+ /// users to use this struct directly.
+ pub(crate) fn new(
inner: FusedAccessor,
path: String,
args: T,
- f: fn(FusedAccessor, String, T) -> BoxedFuture<Result<F>>,
+ f: fn(FusedAccessor, String, T) -> F,
) -> Self {
- OperatorFuture::Idle(inner, path, args, f)
+ OperatorFuture {
+ acc: inner,
+ path,
+ args,
+ f,
+ }
}
+}
- fn map_args(self, f: impl FnOnce(T) -> T) -> Self {
- match self {
- OperatorFuture::Idle(inner, path, args, func) => {
- OperatorFuture::Idle(inner, path, f(args), func)
- }
- _ => unreachable!("future has been polled and should not be
changed again"),
- }
+impl<T, F> OperatorFuture<T, F> {
+ /// Change the operation's args.
+ fn map(mut self, f: impl FnOnce(T) -> T) -> Self {
+ self.args = (f)(self.args);
+ self
}
}
-impl<T, F> Future for OperatorFuture<T, F>
+impl<T, F> IntoFuture for OperatorFuture<T, F>
where
- T: Unpin,
- F: Unpin,
+ F: Future,
{
- type Output = Result<F>;
+ type Output = F::Output;
+ type IntoFuture = F;
- /// We will move the self state out by replace a `Empty` into.
- ///
- /// - If the future is `Idle`, we will move all args out to build
- /// a new future, and update self state to `Poll`.
- /// - If the future is `Poll`, we will poll the inner future
- /// - If future is `Ready`, we will return it directly with
- /// self state is `Empty`
- /// - If future is `Pending`, we will set self state to `Poll`
- /// and wait for next poll
- ///
- /// In general, `Empty` state should not be polled.
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- loop {
- match mem::replace(self.as_mut().get_mut(), OperatorFuture::Empty)
{
- OperatorFuture::Idle(inner, path, args, f) => {
- *self = OperatorFuture::Poll(f(inner, path, args))
- }
- OperatorFuture::Poll(mut fut) => match fut.as_mut().poll(cx) {
- Poll::Ready(v) => return Poll::Ready(v),
- Poll::Pending => {
- *self = OperatorFuture::Poll(fut);
- return Poll::Pending;
- }
- },
- OperatorFuture::Empty => {
- panic!("future polled after completion");
- }
- }
- }
+ fn into_future(self) -> Self::IntoFuture {
+ (self.f)(self.acc, self.path, self.args)
}
}
/// Future that generated by [`Operator::stat_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureStat(pub(crate) OperatorFuture<OpStat, Metadata>);
+pub type FutureStat<F> = OperatorFuture<OpStat, F>;
-impl FutureStat {
+impl<F> FutureStat<F> {
/// 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
+ pub fn if_match(self, v: &str) -> Self {
+ self.map(|args| args.with_if_match(v))
}
/// 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
+ pub fn if_none_match(self, v: &str) -> Self {
+ self.map(|args| args.with_if_none_match(v))
}
/// Set the version for this operation.
- pub fn version(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_version(v));
- self
- }
-}
-
-impl Future for FutureStat {
- type Output = Result<Metadata>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn version(self, v: &str) -> Self {
+ self.map(|args| args.with_version(v))
}
}
/// Future that generated by [`Operator::presign_stat_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FuturePresignStat(pub(crate) OperatorFuture<(OpStat, Duration),
PresignedRequest>);
+pub type FuturePresignStat<F> = OperatorFuture<(OpStat, Duration), F>;
-impl FuturePresignStat {
+impl<F> FuturePresignStat<F> {
/// Sets the content-disposition header that should be send back by the
remote read operation.
- pub fn override_content_disposition(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)|
(args.with_override_content_disposition(v), dur));
- self
+ pub fn override_content_disposition(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_content_disposition(v),
dur))
}
/// Sets the cache-control header that should be send back by the remote
read operation.
- pub fn override_cache_control(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_override_cache_control(v),
dur));
- self
+ pub fn override_cache_control(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_cache_control(v), dur))
}
/// 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
+ pub fn override_content_type(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_content_type(v), dur))
}
/// 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));
- self
+ pub fn if_match(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_if_match(v), dur))
}
/// Set the If-None-Match of the option
- pub fn if_none_match(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_if_none_match(v), dur));
- self
- }
-}
-
-impl Future for FuturePresignStat {
- type Output = Result<PresignedRequest>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn if_none_match(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_if_none_match(v), dur))
}
}
/// Future that generated by [`Operator::presign_read_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FuturePresignRead(pub(crate) OperatorFuture<(OpRead, Duration),
PresignedRequest>);
+pub type FuturePresignRead<F> = OperatorFuture<(OpRead, Duration), F>;
-impl FuturePresignRead {
+impl<F> FuturePresignRead<F> {
/// Create a new OpRead with range.
- pub fn range(mut self, v: BytesRange) -> Self {
- self.0 = self.0.map_args(|(args, dur)| (args.with_range(v), dur));
- self
+ pub fn range(self, v: BytesRange) -> Self {
+ self.map(|(args, dur)| (args.with_range(v), dur))
}
/// Sets the content-disposition header that should be send back by the
remote read operation.
- pub fn override_content_disposition(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)|
(args.with_override_content_disposition(v), dur));
- self
+ pub fn override_content_disposition(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_content_disposition(v),
dur))
}
/// Sets the cache-control header that should be send back by the remote
read operation.
- pub fn override_cache_control(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_override_cache_control(v),
dur));
- self
+ pub fn override_cache_control(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_cache_control(v), dur))
}
/// 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
+ pub fn override_content_type(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_override_content_type(v), dur))
}
/// 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));
- self
+ pub fn if_match(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_if_match(v), dur))
}
/// Set the If-None-Match of the option
- pub fn if_none_match(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_if_none_match(v), dur));
- self
- }
-}
-
-impl Future for FuturePresignRead {
- type Output = Result<PresignedRequest>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn if_none_match(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_if_none_match(v), dur))
}
}
/// Future that generated by [`Operator::presign_write_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FuturePresignWrite(pub(crate) OperatorFuture<(OpWrite, Duration),
PresignedRequest>);
+pub type FuturePresignWrite<F> = OperatorFuture<(OpWrite, Duration), F>;
-impl FuturePresignWrite {
+impl<F> FuturePresignWrite<F> {
/// Set the content type of option
- pub fn content_type(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_content_type(v), dur));
- self
+ pub fn content_type(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_content_type(v), dur))
}
/// Set the content disposition of option
- pub fn content_disposition(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_content_disposition(v), dur));
- self
+ pub fn content_disposition(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_content_disposition(v), dur))
}
/// Set the content type of option
- pub fn cache_control(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, dur)| (args.with_cache_control(v), dur));
- self
+ pub fn cache_control(self, v: &str) -> Self {
+ self.map(|(args, dur)| (args.with_cache_control(v), dur))
}
}
-impl Future for FuturePresignWrite {
- type Output = Result<PresignedRequest>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
- }
-}
-
-/// Future that generated by [`Operator::read_with`].
+/// Future that generated by [`Operator::read_with`] or
[`Operator::reader_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureRead(pub(crate) OperatorFuture<OpRead, Vec<u8>>);
+pub type FutureRead<F> = OperatorFuture<OpRead, F>;
-impl FutureRead {
+impl<F> FutureRead<F> {
/// 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
+ pub fn range(self, range: impl RangeBounds<u64>) -> Self {
+ self.map(|args| args.with_range(range.into()))
}
/// 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
+ pub fn override_content_disposition(self, v: &str) -> Self {
+ self.map(|args| args.with_override_content_disposition(v))
}
/// 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
+ pub fn override_cache_control(self, v: &str) -> Self {
+ self.map(|args| args.with_override_cache_control(v))
}
/// 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
+ pub fn override_content_type(self, v: &str) -> Self {
+ self.map(|args| args.with_override_content_type(v))
}
/// 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
+ pub fn if_match(self, v: &str) -> Self {
+ self.map(|args| args.with_if_match(v))
}
/// 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
+ pub fn if_none_match(self, v: &str) -> Self {
+ self.map(|args| args.with_if_none_match(v))
}
/// Set the version for this operation.
- pub fn version(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_version(v));
- self
+ pub fn version(self, v: &str) -> Self {
+ self.map(|args| args.with_version(v))
}
-}
-
-impl Future for FutureRead {
- type Output = Result<Vec<u8>>;
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- 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
- }
-
- /// 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));
- 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
- }
-
- /// Set the buffer capability to enable `BufferReader`.
- pub fn buffer(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_buffer(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)
+ /// Set the buffer capability to enable buffer for reader.
+ pub fn buffer(self, v: usize) -> Self {
+ self.map(|args| args.with_buffer(v))
}
}
/// Future that generated by [`Operator::write_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureWrite(pub(crate) OperatorFuture<(OpWrite, Bytes), ()>);
+pub type FutureWrite<F> = OperatorFuture<(OpWrite, Bytes), F>;
-impl FutureWrite {
+impl<F> FutureWrite<F> {
/// Set the append mode of op.
///
/// If the append mode is set, the data will be appended to the end of the
file.
@@ -438,9 +262,8 @@ impl FutureWrite {
/// # Notes
///
/// Service could return `Unsupported` if the underlying storage does not
support append.
- pub fn append(mut self, v: bool) -> Self {
- self.0 = self.0.map_args(|(args, bs)| (args.with_append(v), bs));
- self
+ pub fn append(self, v: bool) -> Self {
+ self.map(|(args, bs)| (args.with_append(v), bs))
}
/// Set the buffer size of op.
@@ -451,56 +274,37 @@ impl FutureWrite {
///
/// Service could have their own minimum buffer size while perform write
operations like
/// multipart uploads. So the buffer size may be larger than the given
buffer size.
- pub fn buffer(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|(args, bs)| (args.with_buffer(v), bs));
- self
+ pub fn buffer(self, v: usize) -> Self {
+ self.map(|(args, bs)| (args.with_buffer(v), bs))
}
/// Set the content type of option
- pub fn content_type(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, bs)| (args.with_content_type(v), bs));
- self
+ pub fn content_type(self, v: &str) -> Self {
+ self.map(|(args, bs)| (args.with_content_type(v), bs))
}
/// Set the content disposition of option
- pub fn content_disposition(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, bs)| (args.with_content_disposition(v), bs));
- self
+ pub fn content_disposition(self, v: &str) -> Self {
+ self.map(|(args, bs)| (args.with_content_disposition(v), bs))
}
/// Set the content type of option
- pub fn cache_control(mut self, v: &str) -> Self {
- self.0 = self
- .0
- .map_args(|(args, bs)| (args.with_cache_control(v), bs));
- self
+ pub fn cache_control(self, v: &str) -> Self {
+ self.map(|(args, bs)| (args.with_cache_control(v), bs))
}
/// Set the maximum concurrent write task amount.
- pub fn concurrent(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|(args, bs)| (args.with_buffer(v), bs));
- self
- }
-}
-
-impl Future for FutureWrite {
- type Output = Result<()>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn concurrent(self, v: usize) -> Self {
+ self.map(|(args, bs)| (args.with_buffer(v), bs))
}
}
/// Future that generated by [`Operator::writer_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureWriter(pub(crate) OperatorFuture<OpWrite, Writer>);
+pub type FutureWriter<F> = OperatorFuture<OpWrite, F>;
-impl FutureWriter {
+impl<F> FutureWriter<F> {
/// Set the append mode of op.
///
/// If the append mode is set, the data will be appended to the end of the
file.
@@ -508,9 +312,8 @@ impl FutureWriter {
/// ## Notes
///
/// Service could return `Unsupported` if the underlying storage does not
support append.
- pub fn append(mut self, v: bool) -> Self {
- self.0 = self.0.map_args(|args| args.with_append(v));
- self
+ pub fn append(self, v: bool) -> Self {
+ self.map(|args| args.with_append(v))
}
/// Set the buffer size of op.
@@ -528,85 +331,61 @@ impl FutureWriter {
/// - GCS requires the part size to be aligned with 256 KiB.
///
/// The services will alter the buffer size to meet their requirements.
- pub fn buffer(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_buffer(v));
- self
+ pub fn buffer(self, v: usize) -> Self {
+ self.map(|args| args.with_buffer(v))
}
/// Set the content type of option
- pub fn content_type(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_content_type(v));
- self
+ pub fn content_type(self, v: &str) -> Self {
+ self.map(|args| args.with_content_type(v))
}
/// Set the content disposition of option
- pub fn content_disposition(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_content_disposition(v));
- self
+ pub fn content_disposition(self, v: &str) -> Self {
+ self.map(|args| args.with_content_disposition(v))
}
/// Set the content type of option
- pub fn cache_control(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_cache_control(v));
- self
+ pub fn cache_control(self, v: &str) -> Self {
+ self.map(|args| args.with_cache_control(v))
}
/// Set the maximum concurrent write task amount.
- pub fn concurrent(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_concurrent(v));
- self
- }
-}
-
-impl Future for FutureWriter {
- type Output = Result<Writer>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn concurrent(self, v: usize) -> Self {
+ self.map(|args| args.with_concurrent(v))
}
}
/// Future that generated by [`Operator::delete_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureDelete(pub(crate) OperatorFuture<OpDelete, ()>);
+pub type FutureDelete<F> = OperatorFuture<OpDelete, F>;
-impl FutureDelete {
+impl<F> FutureDelete<F> {
/// Change the version of this delete operation.
- pub fn version(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_version(v));
- self
- }
-}
-
-impl Future for FutureDelete {
- type Output = Result<()>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn version(self, v: &str) -> Self {
+ self.map(|args| args.with_version(v))
}
}
-/// Future that generated by [`Operator::list_with`].
+/// Future that generated by [`Operator::list_with`] or
[`Operator::lister_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub struct FutureList(pub(crate) OperatorFuture<OpList, Vec<Entry>>);
+pub type FutureList<F> = OperatorFuture<OpList, F>;
-impl FutureList {
+impl<F> FutureList<F> {
/// The limit passed to underlying service to specify the max results
/// that could return per-request.
///
/// Users could use this to control the memory usage of list operation.
- pub fn limit(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_limit(v));
- self
+ pub fn limit(self, v: usize) -> Self {
+ self.map(|args| args.with_limit(v))
}
/// The start_after passes to underlying service to specify the specified
key
/// to start listing from.
- pub fn start_after(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_start_after(v));
- self
+ pub fn start_after(self, v: &str) -> Self {
+ self.map(|args| args.with_start_after(v))
}
/// The recursive is used to control whether the list operation is
recursive.
@@ -615,9 +394,8 @@ impl FutureList {
/// - If `true`, list operation will list all entries that starts with
given path.
///
/// Default to `false`.
- pub fn recursive(mut self, v: bool) -> Self {
- self.0 = self.0.map_args(|args| args.with_recursive(v));
- self
+ pub fn recursive(self, v: bool) -> Self {
+ self.map(|args| args.with_recursive(v))
}
/// Metakey is used to control which meta should be returned.
@@ -628,64 +406,8 @@ impl FutureList {
/// - `None` means services doesn't have this meta.
///
/// The default metakey is `Metakey::Mode`.
- pub fn metakey(mut self, v: impl Into<FlagSet<Metakey>>) -> Self {
- self.0 = self.0.map_args(|args| args.with_metakey(v));
- self
- }
-}
-
-impl Future for FutureList {
- type Output = Result<Vec<Entry>>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
- }
-}
-
-/// Future that generated by [`Operator::lister_with`].
-///
-/// Users can add more options by public functions provided by this struct.
-pub struct FutureLister(pub(crate) OperatorFuture<OpList, Lister>);
-
-impl FutureLister {
- /// The limit passed to underlying service to specify the max results
- /// that could return per-request.
- ///
- /// Users could use this to control the memory usage of list operation.
- pub fn limit(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_limit(v));
- self
- }
-
- /// The start_after passes to underlying service to specify the specified
key
- /// to start listing from.
- pub fn start_after(mut self, v: &str) -> Self {
- self.0 = self.0.map_args(|args| args.with_start_after(v));
- self
- }
-
- /// The recursive is used to control whether the list operation is
recursive.
- ///
- /// - If `false`, list operation will only list the entries under the
given path.
- /// - If `true`, list operation will list all entries that starts with
given path.
- ///
- /// Default to `false`.
- pub fn recursive(mut self, v: bool) -> Self {
- self.0 = self.0.map_args(|args| args.with_recursive(v));
- self
- }
-
- /// Metakey is used to control which meta should be returned.
- ///
- /// Lister will make sure the result for specified meta is **known**:
- ///
- /// - `Some(v)` means exist.
- /// - `None` means services doesn't have this meta.
- ///
- /// The default metakey is `Metakey::Mode`.
- pub fn metakey(mut self, v: impl Into<FlagSet<Metakey>>) -> Self {
- self.0 = self.0.map_args(|args| args.with_metakey(v));
- self
+ pub fn metakey(self, v: impl Into<FlagSet<Metakey>>) -> Self {
+ self.map(|args| args.with_metakey(v))
}
/// Concurrent is used to control the number of concurrent stat requests.
@@ -693,16 +415,7 @@ impl FutureLister {
/// If concurrent is set to <=1, the lister will perform stat requests
sequentially.
///
/// The default concurrent is 1.
- pub fn concurrent(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_concurrent(v));
- self
- }
-}
-
-impl Future for FutureLister {
- type Output = Result<Lister>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Self::Output> {
- self.0.poll_unpin(cx)
+ pub fn concurrent(self, v: usize) -> Self {
+ self.map(|args| args.with_concurrent(v))
}
}
