You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2023/05/25 13:32:24 UTC
[incubator-opendal] branch main updated: refactor(core): Implement RFC-2299 for list_with (#2323)
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 19241c7a refactor(core): Implement RFC-2299 for list_with (#2323)
19241c7a is described below
commit 19241c7a2d0d608f827e4c107919a617c094c166
Author: clundro <in...@outlook.com>
AuthorDate: Thu May 25 21:32:20 2023 +0800
refactor(core): Implement RFC-2299 for list_with (#2323)
* refactor(core): Implement RFC-2299 for list_with
Signed-off-by: clundro <in...@outlook.com>
* add brace
Signed-off-by: clundro <in...@outlook.com>
* change scan func
Signed-off-by: clundro <in...@outlook.com>
* add oplist bound
Signed-off-by: clundro <in...@outlook.com>
* Fix build
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: clundro <in...@outlook.com>
Signed-off-by: Xuanwo <gi...@xuanwo.io>
Co-authored-by: Xuanwo <gi...@xuanwo.io>
---
bin/oay/src/services/s3/service.rs | 7 +---
core/src/types/operator/operator.rs | 65 +++++++++++++----------------
core/src/types/operator/operator_futures.rs | 33 +++++++++++++++
core/tests/behavior/list.rs | 4 +-
4 files changed, 64 insertions(+), 45 deletions(-)
diff --git a/bin/oay/src/services/s3/service.rs b/bin/oay/src/services/s3/service.rs
index 84b9c777..f8bf8828 100644
--- a/bin/oay/src/services/s3/service.rs
+++ b/bin/oay/src/services/s3/service.rs
@@ -25,7 +25,6 @@ use axum::response::Response;
use axum::routing::get;
use axum::Router;
use chrono::SecondsFormat;
-use opendal::ops::OpList;
use opendal::Metakey;
use opendal::Operator;
use serde::Deserialize;
@@ -102,10 +101,8 @@ async fn handle_list_objects(
let mut lister = state
.op
- .list_with(
- ¶ms.prefix,
- OpList::new().with_start_after(¶ms.start_after),
- )
+ .list_with(¶ms.prefix)
+ .start_after(¶ms.start_after)
.await?;
let page = lister.next_page().await?.unwrap_or_default();
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 0cb858d9..1d1d1958 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -1264,7 +1264,7 @@ impl Operator {
/// # }
/// ```
pub async fn list(&self, path: &str) -> Result<Lister> {
- self.list_with(path, OpList::new()).await
+ self.list_with(path).await
}
/// List given path with OpList.
@@ -1287,8 +1287,7 @@ impl Operator {
/// use opendal::Operator;
/// # #[tokio::main]
/// # async fn test(op: Operator) -> Result<()> {
- /// let option = OpList::new().with_limit(10).with_start_after("start");
- /// let mut ds = op.list_with("path/to/dir/", option).await?;
+ /// let mut ds = op.list_with("path/to/dir/").limit(10).start_after("start").await?;
/// while let Some(mut de) = ds.try_next().await? {
/// let meta = op.metadata(&de, Metakey::Mode).await?;
/// match meta.mode() {
@@ -1319,8 +1318,7 @@ impl Operator {
/// use opendal::Operator;
/// # #[tokio::main]
/// # async fn test(op: Operator) -> Result<()> {
- /// let option = OpList::new().with_delimiter("");
- /// let mut ds = op.list_with("path/to/dir/", option).await?;
+ /// let mut ds = op.list_with("path/to/dir/").delimiter("").await?;
/// while let Some(mut de) = ds.try_next().await? {
/// let meta = op.metadata(&de, Metakey::Mode).await?;
/// match meta.mode() {
@@ -1336,22 +1334,33 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub async fn list_with(&self, path: &str, op: OpList) -> Result<Lister> {
+ pub fn list_with(&self, path: &str) -> FutureList {
let path = normalize_path(path);
- if !validate_path(&path, EntryMode::DIR) {
- return Err(Error::new(
- ErrorKind::NotADirectory,
- "the path trying to list should end with `/`",
- )
- .with_operation("Operator::list")
- .with_context("service", self.info().scheme().into_static())
- .with_context("path", &path));
- }
+ let fut = FutureList(OperatorFuture::new(
+ self.inner().clone(),
+ path,
+ OpList::default(),
+ |inner, path, args| {
+ let fut = async move {
+ if !validate_path(&path, EntryMode::DIR) {
+ return Err(Error::new(
+ ErrorKind::NotADirectory,
+ "the path trying to list should end with `/`",
+ )
+ .with_operation("Operator::list")
+ .with_context("service", inner.info().scheme().into_static())
+ .with_context("path", &path));
+ }
- let (_, pager) = self.inner().list(&path, op).await?;
+ let (_, pager) = inner.list(&path, args).await?;
- Ok(Lister::new(pager))
+ Ok(Lister::new(pager))
+ };
+ Box::pin(fut)
+ },
+ ));
+ fut
}
/// List dir in flat way.
@@ -1363,7 +1372,7 @@ impl Operator {
/// # Notes
///
/// - `scan` will not return the prefix itself.
- /// - `scan` is an alias of `list_with(OpList::new().with_delimiter(""))`
+ /// - `scan` is an alias of `list_with(path).delimiter("")`
///
/// # Examples
///
@@ -1394,27 +1403,9 @@ impl Operator {
/// # }
/// ```
pub async fn scan(&self, path: &str) -> Result<Lister> {
- let path = normalize_path(path);
-
- if !validate_path(&path, EntryMode::DIR) {
- return Err(Error::new(
- ErrorKind::NotADirectory,
- "the path trying to scan should end with `/`",
- )
- .with_operation("list")
- .with_context("service", self.info().scheme().into_static())
- .with_context("path", &path));
- }
-
- let (_, pager) = self
- .inner()
- .list(&path, OpList::new().with_delimiter(""))
- .await?;
-
- Ok(Lister::new(pager))
+ self.list_with(path).delimiter("").await
}
}
-
/// Operator presign API.
impl Operator {
/// Presign an operation for stat(head).
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 2016b073..478f25ce 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -509,3 +509,36 @@ impl Future for FutureWriter {
self.0.poll_unpin(cx)
}
}
+
+/// Future that generated by [`Operator::list_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub struct FutureList(pub(crate) OperatorFuture<OpList, Lister>);
+
+impl FutureList {
+ /// Change the limit of this list operation.
+ pub fn limit(mut self, v: usize) -> Self {
+ self.0 = self.0.map_args(|args| args.with_limit(v));
+ self
+ }
+
+ /// Change the start_after of this list operation.
+ pub fn start_after(mut self, v: &str) -> Self {
+ self.0 = self.0.map_args(|args| args.with_start_after(v));
+ self
+ }
+
+ /// Change the delimiter. The default delimiter is "/"
+ pub fn delimiter(mut self, v: &str) -> Self {
+ self.0 = self.0.map_args(|args| args.with_delimiter(v));
+ self
+ }
+}
+
+impl Future for FutureList {
+ type Output = Result<Lister>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ self.0.poll_unpin(cx)
+ }
+}
diff --git a/core/tests/behavior/list.rs b/core/tests/behavior/list.rs
index 0549d6ca..98bcd255 100644
--- a/core/tests/behavior/list.rs
+++ b/core/tests/behavior/list.rs
@@ -23,7 +23,6 @@ use futures::stream::FuturesUnordered;
use futures::StreamExt;
use futures::TryStreamExt;
use log::debug;
-use opendal::ops::OpList;
use opendal::EntryMode;
use opendal::ErrorKind;
use opendal::Operator;
@@ -304,8 +303,7 @@ pub async fn test_list_with_start_after(op: Operator) -> Result<()> {
.collect::<Vec<_>>()
.await;
- let option = OpList::new().with_start_after(&given[2]);
- let mut objects = op.list_with(dir, option).await?;
+ let mut objects = op.list_with(dir).start_after(&given[2]).await?;
let mut actual = vec![];
while let Some(o) = objects.try_next().await? {
let path = o.path().to_string();