[incubator-opendal] branch main updated: refactor(core): Implement RFC-2299 for reader_with (#2316)

[xu...@apache.org]

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/incubator-opendal.git


The following commit(s) were added to refs/heads/main by this push:
     new fd708e25 refactor(core): Implement RFC-2299 for reader_with (#2316)
fd708e25 is described below

commit fd708e251f096ce280d64d0a5c90245409007080
Author: clundro <in...@outlook.com>
AuthorDate: Thu May 25 03:52:53 2023 +0800

    refactor(core): Implement RFC-2299 for reader_with (#2316)
    
    Signed-off-by: clundro <in...@outlook.com>
---
 core/src/types/operator/operator.rs         | 41 +++++++++++++++---------
 core/src/types/operator/operator_futures.rs | 49 +++++++++++++++++++++++++++++
 2 files changed, 76 insertions(+), 14 deletions(-)

diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 37e49c9b..e5d8b366 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -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.
@@ -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
@@ -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.
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 9a5ddd30..34c5b21f 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -314,3 +314,52 @@ impl Future for FutureWrite {
         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)
+    }
+}