This is an automated email from the ASF dual-hosted git repository. xuanwo pushed a commit to branch try-into-future in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 4d121e280646fe8055a0fdb6b386e9a8e4bbb61e Author: Xuanwo <[email protected]> AuthorDate: Tue Jan 30 16:38:57 2024 +0800 Save Signed-off-by: Xuanwo <[email protected]> --- core/src/types/operator/operator.rs | 313 ++++++------- core/src/types/operator/operator_futures.rs | 659 ++++++---------------------- 2 files changed, 266 insertions(+), 706 deletions(-) diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs index 4e17530873..ea572a9729 100644 --- a/core/src/types/operator/operator.rs +++ b/core/src/types/operator/operator.rs @@ -281,20 +281,18 @@ impl Operator { /// # Ok(()) /// # } /// ``` - pub fn stat_with(&self, path: &str) -> FutureStatX<impl Future<Output = Result<Metadata>>> { + pub fn stat_with(&self, path: &str) -> FutureStat<impl Future<Output = Result<Metadata>>> { let path = normalize_path(path); - let fut = FutureStatX(OperatorFutureX { - acc: self.inner().clone(), + OperatorFuture::new( + self.inner().clone(), path, - args: OpStat::default(), - f: |inner, path, args| async move { + OpStat::default(), + |inner, path, args| async move { let rp = inner.stat(&path, args).await?; Ok(rp.into_metadata()) }, - }); - - fut + ) } /// Check if this path exists or not. @@ -406,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. @@ -486,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. @@ -708,32 +693,26 @@ impl Operator { /// # Ok(()) /// # } /// ``` - pub fn writer_with(&self, path: &str) -> FutureWriter { + pub fn writer_with(&self, path: &str) -> FutureWrite<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. @@ -759,40 +738,38 @@ 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 mut 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) + OpWrite::default(), + |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::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. @@ -836,24 +813,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 + ) } /// @@ -1250,23 +1221,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. @@ -1471,19 +1438,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 }, + ) } } @@ -1535,23 +1498,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. @@ -1610,23 +1573,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. @@ -1683,22 +1646,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 797106fe4b..1213104e4f 100644 --- a/core/src/types/operator/operator_futures.rs +++ b/core/src/types/operator/operator_futures.rs @@ -20,146 +20,63 @@ //! By using futures, users can add more options for operation. use std::future::IntoFuture; -use std::mem; 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) - } - - 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> Future for OperatorFuture<T, F> -where - T: Unpin, - F: Unpin, -{ - type Output = Result<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"); - } - } + OperatorFuture { + acc: inner, + path, + args, + f, } } } -/// 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>); - -impl FutureStat { - /// 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 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(crate) struct OperatorFutureX<T, F> { - /// The accessor to the underlying object storage - pub acc: FusedAccessor, - /// The path of string - pub path: String, - /// The input args - pub args: T, - /// The function which will move all the args and return a static future - pub f: fn(FusedAccessor, String, T) -> F, -} - -impl<T, F> OperatorFutureX<T, F> { - fn map_args(mut self, f: impl FnOnce(T) -> T) -> Self { +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> IntoFuture for OperatorFutureX<T, F> +impl<T, F> IntoFuture for OperatorFuture<T, F> where F: Future, { @@ -171,401 +88,172 @@ where } } -pub struct FutureStatX<F>(pub(crate) OperatorFutureX<OpStat, F>); - -impl<F> IntoFuture for FutureStatX<F> -where - F: Future<Output = Result<Metadata>>, -{ - type Output = F::Output; - type IntoFuture = F; - - fn into_future(self) -> Self::IntoFuture { - self.0.into_future() - } -} +/// Future that generated by [`Operator::stat_with`]. +/// +/// Users can add more options by public functions provided by this struct. +pub type FutureStat<F> = OperatorFuture<OpStat, F>; -impl<F> FutureStatX<F> { +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_read_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 - } -} - -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 + pub fn version(self, v: &str) -> Self { + self.map(|args| args.with_version(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 - } - - /// 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) - } -} - -/// 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), ()>); - -impl FutureWrite { - /// Set the append mode of op. - /// - /// If the append mode is set, the data will be appended to the end of the file. - /// - /// # 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 - } - - /// Set the buffer size of op. - /// - /// If buffer size is set, the data will be buffered by the underlying writer. - /// - /// ## NOTE - /// - /// 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 - } - - /// 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 - } - - /// 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 - } - - /// 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 - } - - /// 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 + /// Set the buffer capability to enable buffer for reader. + pub fn buffer(self, v: usize) -> Self { + self.map(|args| args.with_buffer(v)) } } -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) - } -} - -/// Future that generated by [`Operator::writer_with`]. +/// Future that generated by [`Operator::write_with`] or [`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 FutureWrite<F> = OperatorFuture<OpWrite, F>; -impl FutureWriter { +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. @@ -573,9 +261,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. @@ -593,85 +280,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 + pub fn version(self, v: &str) -> Self { + self.map(|args| args.with_version(v)) } } -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) - } -} - -/// 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. @@ -680,9 +343,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. @@ -693,64 +355,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. @@ -758,16 +364,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)) } }
