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 2024/03/13 09:49:27 UTC
(opendal) 01/04: Build passed
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-list
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 26906a9bb0c3eba30b1f49cbb0f2dd8235292da0
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Mar 13 17:43:53 2024 +0800
Build passed
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/blocking.rs | 2 +-
core/src/layers/complete.rs | 2 +-
core/src/layers/concurrent_limit.rs | 5 +-
core/src/layers/error_context.rs | 6 +--
core/src/layers/immutable_index.rs | 5 +-
core/src/layers/logging.rs | 9 ++--
core/src/layers/madsim.rs | 7 +--
core/src/layers/minitrace.rs | 8 +--
core/src/layers/oteltrace.rs | 7 ++-
core/src/layers/retry.rs | 77 ++++++++++++-----------------
core/src/layers/timeout.rs | 16 +++---
core/src/layers/tracing.rs | 5 +-
core/src/raw/adapters/kv/backend.rs | 5 +-
core/src/raw/adapters/typed_kv/backend.rs | 5 +-
core/src/raw/enum_utils.rs | 11 +++--
core/src/raw/oio/list/api.rs | 64 ++++++++++--------------
core/src/raw/oio/list/flat_list.rs | 42 +++++-----------
core/src/raw/oio/list/hierarchy_list.rs | 11 +++--
core/src/raw/oio/list/mod.rs | 1 -
core/src/raw/oio/list/page_list.rs | 79 +++++++-----------------------
core/src/raw/oio/list/prefix_list.rs | 7 +--
core/src/services/alluxio/lister.rs | 1 -
core/src/services/azblob/lister.rs | 2 -
core/src/services/azdls/lister.rs | 1 -
core/src/services/azfile/lister.rs | 1 -
core/src/services/b2/lister.rs | 1 -
core/src/services/chainsafe/lister.rs | 1 -
core/src/services/cos/lister.rs | 1 -
core/src/services/dbfs/lister.rs | 1 -
core/src/services/fs/lister.rs | 81 +++++++++++--------------------
core/src/services/ftp/lister.rs | 7 +--
core/src/services/gcs/lister.rs | 2 -
core/src/services/gdrive/lister.rs | 2 -
core/src/services/github/lister.rs | 1 -
core/src/services/hdfs/lister.rs | 9 ++--
core/src/services/hdfs_native/lister.rs | 3 +-
core/src/services/huggingface/lister.rs | 1 -
core/src/services/ipfs/backend.rs | 1 -
core/src/services/ipmfs/lister.rs | 1 -
core/src/services/koofr/lister.rs | 1 -
core/src/services/obs/lister.rs | 1 -
core/src/services/onedrive/lister.rs | 1 -
core/src/services/oss/lister.rs | 1 -
core/src/services/pcloud/lister.rs | 1 -
core/src/services/s3/lister.rs | 2 -
core/src/services/seafile/lister.rs | 1 -
core/src/services/sftp/lister.rs | 30 +++++++-----
core/src/services/swift/lister.rs | 1 -
core/src/services/upyun/lister.rs | 1 -
core/src/services/vercel_blob/lister.rs | 1 -
core/src/services/webdav/lister.rs | 1 -
core/src/services/webhdfs/lister.rs | 1 -
core/src/services/yandex_disk/lister.rs | 1 -
core/src/types/list.rs | 26 +++++++---
54 files changed, 219 insertions(+), 342 deletions(-)
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index 84a0c12292..529ce6bf66 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -311,7 +311,7 @@ impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
- self.handle.block_on(poll_fn(|cx| self.inner.poll_next(cx)))
+ self.handle.block_on(self.inner.next())
}
}
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 60dd2d2673..405210d320 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -212,7 +212,7 @@ impl<A: Accessor> CompleteAccessor<A> {
)
.await?;
- return if oio::ListExt::next(&mut l).await?.is_some() {
+ return if oio::List::next(&mut l).await?.is_some() {
Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
} else {
Err(Error::new(
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index 8a663cfc47..aeb6f42644 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fmt::Debug;
+use std::future::Future;
use std::io::SeekFrom;
use std::sync::Arc;
use std::task::Context;
@@ -301,8 +302,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for ConcurrentLimitWrapper<R> {
}
impl<R: oio::List> oio::List for ConcurrentLimitWrapper<R> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- self.inner.poll_next(cx)
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ self.inner.next().await
}
}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index f8112951a1..581c9e9b27 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::future::Future;
use std::io::SeekFrom;
use std::task::Context;
use std::task::Poll;
@@ -433,10 +434,9 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
}
}
-#[async_trait::async_trait]
impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- self.inner.poll_next(cx).map_err(|err| {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ self.inner.next().await.map_err(|err| {
err.with_operation(ListOperation::Next)
.with_context("service", self.scheme)
.with_context("path", &self.path)
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index eea8a343c1..37afa3cb4d 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -17,6 +17,7 @@
use std::collections::HashSet;
use std::fmt::Debug;
+use std::future::Future;
use std::task::Context;
use std::task::Poll;
use std::vec::IntoIter;
@@ -233,8 +234,8 @@ impl ImmutableDir {
}
impl oio::List for ImmutableDir {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- Poll::Ready(Ok(self.inner_next()))
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 02abb3f3df..779ea3f074 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fmt::Debug;
+use std::future::Future;
use std::io;
use std::task::ready;
use std::task::Context;
@@ -1353,11 +1354,9 @@ impl<P> Drop for LoggingLister<P> {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<P: oio::List> oio::List for LoggingLister<P> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- let res = ready!(self.inner.poll_next(cx));
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let res = self.inner.next().await;
match &res {
Ok(Some(de)) => {
@@ -1395,7 +1394,7 @@ impl<P: oio::List> oio::List for LoggingLister<P> {
}
};
- Poll::Ready(res)
+ res
}
}
diff --git a/core/src/layers/madsim.rs b/core/src/layers/madsim.rs
index d4135251a3..c2efde340b 100644
--- a/core/src/layers/madsim.rs
+++ b/core/src/layers/madsim.rs
@@ -25,6 +25,7 @@ use std::any::Any;
use std::cmp::min;
use std::collections::HashMap;
use std::fmt::Debug;
+use std::future::Future;
use std::io::Result;
use std::io::SeekFrom;
use std::net::SocketAddr;
@@ -325,11 +326,11 @@ impl oio::Write for MadsimWriter {
pub struct MadsimLister {}
impl oio::List for MadsimLister {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<Option<oio::Entry>>> {
- Poll::Ready(Err(Error::new(
+ async fn next(&mut self) -> crate::Result<Option<oio::Entry>> {
+ Err(Error::new(
ErrorKind::Unsupported,
"will be supported in the future",
- )))
+ ))
}
}
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index fdc6a1dc92..9e525f7bd9 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fmt::Debug;
+use std::future::Future;
use std::io;
use std::task::Context;
use std::task::Poll;
@@ -357,10 +358,9 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for MinitraceWrapper<R> {
}
impl<R: oio::List> oio::List for MinitraceWrapper<R> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(ListOperation::Next.into_static());
- self.inner.poll_next(cx)
+ #[trace(enter_on_poll = true)]
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ self.inner.next().await
}
}
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 90e8d65392..2aa70810f8 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::io;
use std::task::Context;
use std::task::Poll;
@@ -321,11 +322,9 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for OtelTraceWrapper<R> {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- self.inner.poll_next(cx)
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ self.inner.next().await
}
}
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index e6582114e8..b63b55e396 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::Arc;
@@ -952,53 +953,37 @@ impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWra
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- if let Some(sleep) = self.sleep.as_mut() {
- ready!(sleep.poll_unpin(cx));
- self.sleep = None;
- }
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ use backon::RetryableWithContext;
- match ready!(self.inner.as_mut().unwrap().poll_next(cx)) {
- Ok(v) => {
- self.current_backoff = None;
- Poll::Ready(Ok(v))
- }
- Err(err) if !err.is_temporary() => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Err(err) => {
- let backoff = match self.current_backoff.as_mut() {
- Some(backoff) => backoff,
- None => {
- self.current_backoff = Some(self.builder.build());
- self.current_backoff.as_mut().unwrap()
- }
- };
+ let inner = self.inner.take().expect("inner must be valid");
- match backoff.next() {
- None => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Some(dur) => {
- self.notify.intercept(
- &err,
- dur,
- &[
- ("operation", ListOperation::Next.into_static()),
- ("path", &self.path),
- ],
- );
- self.sleep = Some(Box::pin(tokio::time::sleep(dur)));
- self.poll_next(cx)
- }
- }
+ let (inner, res) = {
+ |mut p: P| async move {
+ let res = p.next().await;
+
+ (p, res)
}
}
+ .retry(&self.builder)
+ .when(|e| e.is_temporary())
+ .context(inner)
+ .notify(|err, dur| {
+ self.notify.intercept(
+ err,
+ dur,
+ &[
+ ("operation", ListOperation::Next.into_static()),
+ ("path", &self.path),
+ ],
+ )
+ })
+ .map(|(r, res)| (r, res.map_err(|err| err.set_persistent())))
+ .await;
+
+ self.inner = Some(inner);
+ res
}
}
@@ -1208,9 +1193,9 @@ mod tests {
}
impl oio::List for MockLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
self.attempt += 1;
- let result = match self.attempt {
+ match self.attempt {
1 => Err(Error::new(
ErrorKind::RateLimited,
"retryable rate limited error from lister",
@@ -1240,9 +1225,7 @@ mod tests {
_ => {
unreachable!()
}
- };
-
- Poll::Ready(result)
+ }
}
}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index f80d29cf1b..ef4a4936bf 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -360,12 +360,9 @@ impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
}
impl<R: oio::List> oio::List for TimeoutWrapper<R> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- self.poll_timeout(cx, ListOperation::Next.into_static())?;
-
- let v = ready!(self.inner.poll_next(cx));
- self.sleep = None;
- Poll::Ready(v)
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let fut = self.inner.next();
+ Self::io_timeout(self.timeout, ListOperation::Next.into_static(), fut).await
}
}
@@ -447,8 +444,8 @@ mod tests {
struct MockLister;
impl oio::List for MockLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- Poll::Pending
+ fn next(&mut self) -> impl Future<Output = Result<Option<oio::Entry>>> {
+ pending()
}
}
@@ -506,6 +503,8 @@ mod tests {
#[tokio::test]
async fn test_list_timeout_raw() {
+ use oio::List;
+
let acc = MockService;
let timeout_layer = TimeoutLayer::new()
.with_timeout(Duration::from_secs(1))
@@ -516,7 +515,6 @@ mod tests {
.await
.unwrap();
- use oio::ListExt;
let res = lister.next().await;
assert!(res.is_err());
let err = res.unwrap_err();
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 017264cd32..e0a3b8a541 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fmt::Debug;
+use std::future::Future;
use std::io;
use std::task::Context;
use std::task::Poll;
@@ -349,8 +350,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for TracingWrapper<R> {
impl<R: oio::List> oio::List for TracingWrapper<R> {
#[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- self.inner.poll_next(cx)
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ self.inner.next().await
}
}
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index 2f1c82da8a..7f5a66e94a 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::sync::Arc;
use std::task::ready;
use std::task::Context;
@@ -249,8 +250,8 @@ impl KvLister {
}
impl oio::List for KvLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- Poll::Ready(Ok(self.inner_next()))
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index 8f56a06116..38b097214e 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::sync::Arc;
use std::task::ready;
use std::task::Context;
@@ -253,8 +254,8 @@ impl KvLister {
}
impl oio::List for KvLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- Poll::Ready(Ok(self.inner_next()))
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs
index 430e513774..f43d14fee2 100644
--- a/core/src/raw/enum_utils.rs
+++ b/core/src/raw/enum_utils.rs
@@ -38,6 +38,7 @@
//! This module is used to provide some enums for the above code. We should remove this module once
//! type_alias_impl_trait has been stabilized.
+use std::future::Future;
use std::io::SeekFrom;
use std::task::Context;
use std::task::Poll;
@@ -263,12 +264,12 @@ where
THREE: oio::List,
FOUR: oio::List,
{
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
match self {
- Self::One(v) => v.poll_next(cx),
- Self::Two(v) => v.poll_next(cx),
- Self::Three(v) => v.poll_next(cx),
- Self::Four(v) => v.poll_next(cx),
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
+ Self::Three(v) => v.next().await,
+ Self::Four(v) => v.next().await,
}
}
}
diff --git a/core/src/raw/oio/list/api.rs b/core/src/raw/oio/list/api.rs
index 07dbdf774b..4596cf66b4 100644
--- a/core/src/raw/oio/list/api.rs
+++ b/core/src/raw/oio/list/api.rs
@@ -15,21 +15,25 @@
// specific language governing permissions and limitations
// under the License.
+use bytes::Bytes;
use std::fmt::Display;
use std::fmt::Formatter;
-use std::future::Future;
+use std::future::{ready, Future};
+use std::io;
+use std::ops::DerefMut;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
use crate::raw::oio::Entry;
+use crate::raw::BoxedFuture;
use crate::*;
/// PageOperation is the name for APIs of lister.
#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq)]
#[non_exhaustive]
pub enum ListOperation {
- /// Operation for [`List::poll_next`]
+ /// Operation for [`List::next`]
Next,
/// Operation for [`BlockingList::next`]
BlockingNext,
@@ -59,67 +63,51 @@ impl From<ListOperation> for &'static str {
}
}
+/// The boxed version of [`List`]
+pub type Lister = Box<dyn ListDyn>;
+
/// Page trait is used by [`raw::Accessor`] to implement `list` operation.
-pub trait List: Unpin + Send + Sync + 'static {
+pub trait List: Unpin + Send + Sync {
/// Fetch a new page of [`Entry`]
///
/// `Ok(None)` means all pages have been returned. Any following call
/// to `next` will always get the same result.
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>>;
-}
-
-/// The boxed version of [`List`]
-pub type Lister = Box<dyn List>;
-
-impl<P: List + ?Sized> List for Box<P> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
- (**self).poll_next(cx)
- }
+ fn next(&mut self) -> impl Future<Output = Result<Option<Entry>>> + Send;
}
impl List for () {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
- Poll::Ready(Ok(None))
+ async fn next(&mut self) -> Result<Option<Entry>> {
+ Ok(None)
}
}
impl<P: List> List for Option<P> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
+ async fn next(&mut self) -> Result<Option<Entry>> {
match self {
- Some(p) => p.poll_next(cx),
- None => Poll::Ready(Ok(None)),
+ Some(p) => p.next().await,
+ None => Ok(None),
}
}
}
-/// Impl ListExt for all T: List
-impl<T: List> ListExt for T {}
-
-/// Extension of [`List`] to make it easier for use.
-pub trait ListExt: List {
- /// Build a future for `poll_next`.
- fn next(&mut self) -> NextFuture<Self> {
- NextFuture { lister: self }
- }
+pub trait ListDyn: Unpin + Send + Sync {
+ fn next_dyn(&mut self) -> BoxedFuture<Result<Option<Entry>>>;
}
-pub struct NextFuture<'a, L: List + Unpin + ?Sized> {
- lister: &'a mut L,
+impl<T: List + ?Sized> ListDyn for T {
+ fn next_dyn(&mut self) -> BoxedFuture<Result<Option<Entry>>> {
+ Box::pin(self.next())
+ }
}
-impl<L> Future for NextFuture<'_, L>
-where
- L: List + Unpin + ?Sized,
-{
- type Output = Result<Option<Entry>>;
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
- self.lister.poll_next(cx)
+impl<T: ListDyn + ?Sized> List for Box<T> {
+ async fn next(&mut self) -> Result<Option<Entry>> {
+ self.deref_mut().next_dyn().await
}
}
/// BlockingList is the blocking version of [`List`].
-pub trait BlockingList: Send + 'static {
+pub trait BlockingList: Send {
/// Fetch a new page of [`Entry`]
///
/// `Ok(None)` means all pages have been returned. Any following call
diff --git a/core/src/raw/oio/list/flat_list.rs b/core/src/raw/oio/list/flat_list.rs
index 2b948408c1..66899bfe1b 100644
--- a/core/src/raw/oio/list/flat_list.rs
+++ b/core/src/raw/oio/list/flat_list.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -24,9 +25,6 @@ use futures::FutureExt;
use crate::raw::*;
use crate::*;
-/// ListFuture is the future returned while calling async list.
-type ListFuture<A, L> = BoxedStaticFuture<(A, oio::Entry, Result<(RpList, L)>)>;
-
/// FlatLister will walk dir in bottom up way:
///
/// - List nested dir first
@@ -64,12 +62,11 @@ type ListFuture<A, L> = BoxedStaticFuture<(A, oio::Entry, Result<(RpList, L)>)>;
/// may output parent dirs' files before nested dirs, this is expected because files
/// always output directly while listing.
pub struct FlatLister<A: Accessor, L> {
- acc: Option<A>,
+ acc: A,
root: String,
next_dir: Option<oio::Entry>,
active_lister: Vec<(Option<oio::Entry>, L)>,
- list_future: Option<ListFuture<A, L>>,
}
/// # Safety
@@ -88,11 +85,10 @@ where
/// Create a new flat lister
pub fn new(acc: A, path: &str) -> FlatLister<A, L> {
FlatLister {
- acc: Some(acc),
+ acc,
root: path.to_string(),
next_dir: Some(oio::Entry::new(path, Metadata::new(EntryMode::DIR))),
active_lister: vec![],
- list_future: None,
}
}
}
@@ -102,44 +98,30 @@ where
A: Accessor<Lister = L>,
L: oio::List,
{
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
loop {
- if let Some(fut) = self.list_future.as_mut() {
- let (acc, de, res) = ready!(fut.poll_unpin(cx));
- self.acc = Some(acc);
- self.list_future = None;
-
- let (_, l) = res?;
- self.active_lister.push((Some(de), l))
- }
-
if let Some(de) = self.next_dir.take() {
- let acc = self.acc.take().expect("Accessor must be valid");
- let fut = async move {
- let res = acc.list(de.path(), OpList::new()).await;
- (acc, de, res)
- };
- self.list_future = Some(Box::pin(fut));
- continue;
+ let (_, l) = self.acc.list(de.path(), OpList::new()).await?;
+ self.active_lister.push((Some(de), l));
}
let (de, lister) = match self.active_lister.last_mut() {
Some((de, lister)) => (de, lister),
- None => return Poll::Ready(Ok(None)),
+ None => return Ok(None),
};
- match ready!(lister.poll_next(cx))? {
+ match lister.next().await? {
Some(v) if v.mode().is_dir() => {
self.next_dir = Some(v);
continue;
}
- Some(v) => return Poll::Ready(Ok(Some(v))),
+ Some(v) => return Ok(Some(v)),
None => {
match de.take() {
Some(de) => {
// Only push entry if it's not root dir
if de.path() != self.root {
- return Poll::Ready(Ok(Some(de)));
+ return Ok(Some(de));
}
continue;
}
@@ -162,10 +144,8 @@ where
fn next(&mut self) -> Result<Option<oio::Entry>> {
loop {
if let Some(de) = self.next_dir.take() {
- let acc = self.acc.take().expect("Accessor must be valid");
- let (_, l) = acc.blocking_list(de.path(), OpList::new())?;
+ let (_, l) = self.acc.blocking_list(de.path(), OpList::new())?;
- self.acc = Some(acc);
self.active_lister.push((Some(de), l))
}
diff --git a/core/src/raw/oio/list/hierarchy_list.rs b/core/src/raw/oio/list/hierarchy_list.rs
index 259e455322..64fae43556 100644
--- a/core/src/raw/oio/list/hierarchy_list.rs
+++ b/core/src/raw/oio/list/hierarchy_list.rs
@@ -16,6 +16,7 @@
// under the License.
use std::collections::HashSet;
+use std::future::Future;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -125,18 +126,18 @@ impl<P> HierarchyLister<P> {
}
impl<P: oio::List> oio::List for HierarchyLister<P> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
loop {
- let mut entry = match ready!(self.lister.poll_next(cx))? {
+ let mut entry = match self.lister.next().await? {
Some(entry) => entry,
- None => return Poll::Ready(Ok(None)),
+ None => return Ok(None),
};
if self.recursive {
- return Poll::Ready(Ok(Some(entry)));
+ return Ok(Some(entry));
}
if self.keep_entry(&mut entry) {
- return Poll::Ready(Ok(Some(entry)));
+ return Ok(Some(entry));
}
}
}
diff --git a/core/src/raw/oio/list/mod.rs b/core/src/raw/oio/list/mod.rs
index 31f70d6f23..8460afb1a2 100644
--- a/core/src/raw/oio/list/mod.rs
+++ b/core/src/raw/oio/list/mod.rs
@@ -19,7 +19,6 @@ mod api;
pub use api::BlockingList;
pub use api::BlockingLister;
pub use api::List;
-pub use api::ListExt;
pub use api::ListOperation;
pub use api::Lister;
diff --git a/core/src/raw/oio/list/page_list.rs b/core/src/raw/oio/list/page_list.rs
index b793a6b63f..cec3999ca2 100644
--- a/core/src/raw/oio/list/page_list.rs
+++ b/core/src/raw/oio/list/page_list.rs
@@ -16,6 +16,7 @@
// under the License.
use std::collections::VecDeque;
+use std::future::Future;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -25,21 +26,9 @@ use async_trait::async_trait;
use crate::raw::*;
use crate::*;
-/// PageList is used to implement [`List`] based on API supporting pagination. By implementing
-/// PageList, services don't need to care about the details of page list.
-///
-/// # Architecture
-///
-/// The architecture after adopting [`PageList`]:
-///
-/// - Services impl `PageList`
-/// - `PageLister` impl `List`
-/// - Expose `PageLister` as `Accessor::Lister`
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait PageList: Send + Sync + Unpin + 'static {
/// next_page is used to fetch next page of entries from underlying storage.
- async fn next_page(&self, ctx: &mut PageContext) -> Result<()>;
+ fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output = Result<()>> + Send;
}
/// PageContext is the context passing between `PageList`.
@@ -67,23 +56,10 @@ pub struct PageContext {
/// PageLister implements [`List`] based on [`PageList`].
pub struct PageLister<L: PageList> {
- state: State<L>,
+ inner: L,
+ ctx: PageContext,
}
-enum State<L> {
- Idle(Option<(L, PageContext)>),
- Fetch(BoxedStaticFuture<((L, PageContext), Result<()>)>),
-}
-
-/// # Safety
-///
-/// wasm32 is a special target that we only have one event-loop for this state.
-unsafe impl<L: PageList> Send for State<L> {}
-/// # Safety
-///
-/// We will only take `&mut Self` reference for State.
-unsafe impl<L: PageList> Sync for State<L> {}
-
impl<L> PageLister<L>
where
L: PageList,
@@ -91,14 +67,12 @@ where
/// Create a new PageLister.
pub fn new(l: L) -> Self {
Self {
- state: State::Idle(Some((
- l,
- PageContext {
- done: false,
- token: "".to_string(),
- entries: VecDeque::new(),
- },
- ))),
+ inner: l,
+ ctx: PageContext {
+ done: false,
+ token: "".to_string(),
+ entries: VecDeque::new(),
+ },
}
}
}
@@ -107,33 +81,16 @@ impl<L> oio::List for PageLister<L>
where
L: PageList,
{
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
loop {
- match &mut self.state {
- State::Idle(st) => {
- if let Some((_, ctx)) = st.as_mut() {
- if let Some(entry) = ctx.entries.pop_front() {
- return Poll::Ready(Ok(Some(entry)));
- }
- if ctx.done {
- return Poll::Ready(Ok(None));
- }
- }
-
- let (l, mut ctx) = st.take().expect("lister must be valid");
- let fut = async move {
- let res = l.next_page(&mut ctx).await;
- ((l, ctx), res)
- };
- self.state = State::Fetch(Box::pin(fut));
- }
- State::Fetch(fut) => {
- let ((l, ctx), res) = ready!(fut.as_mut().poll(cx));
- self.state = State::Idle(Some((l, ctx)));
-
- res?;
- }
+ if let Some(entry) = self.ctx.entries.pop_front() {
+ return Ok(Some(entry));
+ }
+ if self.ctx.done {
+ return Ok(None);
}
+
+ self.inner.next_page(&mut self.ctx).await?;
}
}
}
diff --git a/core/src/raw/oio/list/prefix_list.rs b/core/src/raw/oio/list/prefix_list.rs
index 166dbcaece..fcf899bc23 100644
--- a/core/src/raw/oio/list/prefix_list.rs
+++ b/core/src/raw/oio/list/prefix_list.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -66,11 +67,11 @@ impl<L> oio::List for PrefixLister<L>
where
L: oio::List,
{
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
loop {
- match ready!(self.lister.poll_next(cx)) {
+ match self.lister.next().await {
Ok(Some(e)) if !starts_with_not_eq(&e, &self.prefix) => continue,
- v => return Poll::Ready(v),
+ v => return v,
}
}
}
diff --git a/core/src/services/alluxio/lister.rs b/core/src/services/alluxio/lister.rs
index 5e56dbff89..845c0053ac 100644
--- a/core/src/services/alluxio/lister.rs
+++ b/core/src/services/alluxio/lister.rs
@@ -40,7 +40,6 @@ impl AlluxioLister {
}
}
-#[async_trait]
impl oio::PageList for AlluxioLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let result = self.core.list_status(&self.path).await;
diff --git a/core/src/services/azblob/lister.rs b/core/src/services/azblob/lister.rs
index eeded04999..4ea2cabda2 100644
--- a/core/src/services/azblob/lister.rs
+++ b/core/src/services/azblob/lister.rs
@@ -48,8 +48,6 @@ impl AzblobLister {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::PageList for AzblobLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/azdls/lister.rs b/core/src/services/azdls/lister.rs
index db7904f946..65171d57aa 100644
--- a/core/src/services/azdls/lister.rs
+++ b/core/src/services/azdls/lister.rs
@@ -39,7 +39,6 @@ impl AzdlsLister {
}
}
-#[async_trait]
impl oio::PageList for AzdlsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/azfile/lister.rs b/core/src/services/azfile/lister.rs
index bb9f8f73b4..ae222e3339 100644
--- a/core/src/services/azfile/lister.rs
+++ b/core/src/services/azfile/lister.rs
@@ -39,7 +39,6 @@ impl AzfileLister {
}
}
-#[async_trait]
impl oio::PageList for AzfileLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/b2/lister.rs b/core/src/services/b2/lister.rs
index 0dee20a9d1..f404e38120 100644
--- a/core/src/services/b2/lister.rs
+++ b/core/src/services/b2/lister.rs
@@ -58,7 +58,6 @@ impl B2Lister {
}
}
-#[async_trait]
impl oio::PageList for B2Lister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/chainsafe/lister.rs b/core/src/services/chainsafe/lister.rs
index 69cdc55200..e0ac35c8d8 100644
--- a/core/src/services/chainsafe/lister.rs
+++ b/core/src/services/chainsafe/lister.rs
@@ -43,7 +43,6 @@ impl ChainsafeLister {
}
}
-#[async_trait]
impl oio::PageList for ChainsafeLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.core.list_objects(&self.path).await?;
diff --git a/core/src/services/cos/lister.rs b/core/src/services/cos/lister.rs
index f289b3383d..d46db1a100 100644
--- a/core/src/services/cos/lister.rs
+++ b/core/src/services/cos/lister.rs
@@ -47,7 +47,6 @@ impl CosLister {
}
}
-#[async_trait]
impl oio::PageList for CosLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/dbfs/lister.rs b/core/src/services/dbfs/lister.rs
index e86dd33810..11d36c1ae2 100644
--- a/core/src/services/dbfs/lister.rs
+++ b/core/src/services/dbfs/lister.rs
@@ -37,7 +37,6 @@ impl DbfsLister {
}
}
-#[async_trait]
impl oio::PageList for DbfsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let response = self.core.dbfs_list(&self.path).await?;
diff --git a/core/src/services/fs/lister.rs b/core/src/services/fs/lister.rs
index 29ec2c9f36..d99414ea69 100644
--- a/core/src/services/fs/lister.rs
+++ b/core/src/services/fs/lister.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fs::FileType;
+use std::future::Future;
use std::path::Path;
use std::path::PathBuf;
use std::task::ready;
@@ -25,6 +26,7 @@ use std::task::Poll;
use futures::future::BoxFuture;
use futures::FutureExt;
+use crate::raw::oio::Entry;
use crate::raw::*;
use crate::EntryMode;
use crate::Metadata;
@@ -34,8 +36,6 @@ pub struct FsLister<P> {
root: PathBuf,
rd: P,
-
- fut: Option<BoxFuture<'static, (tokio::fs::DirEntry, Result<FileType>)>>,
}
impl<P> FsLister<P> {
@@ -43,8 +43,6 @@ impl<P> FsLister<P> {
Self {
root: root.to_owned(),
rd,
-
- fut: None,
}
}
}
@@ -55,57 +53,32 @@ impl<P> FsLister<P> {
unsafe impl<P> Sync for FsLister<P> {}
impl oio::List for FsLister<tokio::fs::ReadDir> {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- if let Some(fut) = self.fut.as_mut() {
- let (de, ft) = futures::ready!(fut.poll_unpin(cx));
- let ft = match ft {
- Ok(ft) => {
- self.fut = None;
- ft
- }
- Err(e) => {
- let fut = async move {
- let ft = de.file_type().await.map_err(new_std_io_error);
- (de, ft)
- };
- self.fut = Some(Box::pin(fut));
- return Poll::Ready(Err(e));
- }
- };
-
- let entry_path = de.path();
- let rel_path = normalize_path(
- &entry_path
- .strip_prefix(&self.root)
- .expect("cannot fail because the prefix is iterated")
- .to_string_lossy()
- .replace('\\', "/"),
- );
-
- let d = if ft.is_file() {
- oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
- } else if ft.is_dir() {
- // Make sure we are returning the correct path.
- oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
- } else {
- oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
- };
-
- return Poll::Ready(Ok(Some(d)));
- }
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let Some(de) = self.rd.next_entry().await.map_err(new_std_io_error)? else {
+ return Ok(None);
+ };
- let de = ready!(self.rd.poll_next_entry(cx)).map_err(new_std_io_error)?;
- match de {
- Some(de) => {
- let fut = async move {
- let ft = de.file_type().await.map_err(new_std_io_error);
- (de, ft)
- };
- self.fut = Some(Box::pin(fut));
- self.poll_next(cx)
- }
- None => Poll::Ready(Ok(None)),
- }
+ let ft = de.file_type().await.map_err(new_std_io_error)?;
+
+ let entry_path = de.path();
+ let rel_path = normalize_path(
+ &entry_path
+ .strip_prefix(&self.root)
+ .expect("cannot fail because the prefix is iterated")
+ .to_string_lossy()
+ .replace('\\', "/"),
+ );
+
+ let d = if ft.is_file() {
+ oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
+ } else if ft.is_dir() {
+ // Make sure we are returning the correct path.
+ oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
+ } else {
+ oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
+ };
+
+ Ok(Some(d))
}
}
diff --git a/core/src/services/ftp/lister.rs b/core/src/services/ftp/lister.rs
index 4d77bf1b87..023dbd8a06 100644
--- a/core/src/services/ftp/lister.rs
+++ b/core/src/services/ftp/lister.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::str;
use std::str::FromStr;
use std::task::Context;
@@ -41,12 +42,12 @@ impl FtpLister {
}
impl oio::List for FtpLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
let de = match self.file_iter.next() {
Some(file_str) => File::from_str(file_str.as_str()).map_err(|e| {
Error::new(ErrorKind::Unexpected, "parse file from response").set_source(e)
})?,
- None => return Poll::Ready(Ok(None)),
+ None => return Ok(None),
};
let path = self.path.to_string() + de.name();
@@ -64,6 +65,6 @@ impl oio::List for FtpLister {
oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
};
- Poll::Ready(Ok(Some(entry)))
+ Ok(Some(entry))
}
}
diff --git a/core/src/services/gcs/lister.rs b/core/src/services/gcs/lister.rs
index 0a01801988..d5d6d0a891 100644
--- a/core/src/services/gcs/lister.rs
+++ b/core/src/services/gcs/lister.rs
@@ -60,8 +60,6 @@ impl GcsLister {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::PageList for GcsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/gdrive/lister.rs b/core/src/services/gdrive/lister.rs
index db0ad063a6..b61bf16b77 100644
--- a/core/src/services/gdrive/lister.rs
+++ b/core/src/services/gdrive/lister.rs
@@ -37,8 +37,6 @@ impl GdriveLister {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::PageList for GdriveLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let file_id = self.core.path_cache.get(&self.path).await?;
diff --git a/core/src/services/github/lister.rs b/core/src/services/github/lister.rs
index 320f664a02..c26a7eb383 100644
--- a/core/src/services/github/lister.rs
+++ b/core/src/services/github/lister.rs
@@ -39,7 +39,6 @@ impl GithubLister {
}
}
-#[async_trait]
impl oio::PageList for GithubLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let entries = self.core.list(&self.path).await?;
diff --git a/core/src/services/hdfs/lister.rs b/core/src/services/hdfs/lister.rs
index 404d159f5c..a14e305300 100644
--- a/core/src/services/hdfs/lister.rs
+++ b/core/src/services/hdfs/lister.rs
@@ -15,11 +15,13 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::task::Context;
use std::task::Poll;
use async_trait::async_trait;
+use crate::raw::oio::Entry;
use crate::raw::*;
use crate::EntryMode;
use crate::Metadata;
@@ -41,12 +43,11 @@ impl HdfsLister {
}
}
-#[async_trait]
impl oio::List for HdfsLister {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
let de = match self.rd.next() {
Some(de) => de,
- None => return Poll::Ready(Ok(None)),
+ None => return Ok(None),
};
let path = build_rel_path(&self.root, de.path());
@@ -63,7 +64,7 @@ impl oio::List for HdfsLister {
oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
};
- Poll::Ready(Ok(Some(entry)))
+ Ok(Some(entry))
}
}
diff --git a/core/src/services/hdfs_native/lister.rs b/core/src/services/hdfs_native/lister.rs
index 65c916c280..64f9ffa8d1 100644
--- a/core/src/services/hdfs_native/lister.rs
+++ b/core/src/services/hdfs_native/lister.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
@@ -38,7 +39,7 @@ impl HdfsNativeLister {
}
impl oio::List for HdfsNativeLister {
- fn poll_next(&mut self, _cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
+ async fn next(&mut self) -> Result<Option<Entry>> {
todo!()
}
}
diff --git a/core/src/services/huggingface/lister.rs b/core/src/services/huggingface/lister.rs
index e291dd3cdd..3ecceb8ae9 100644
--- a/core/src/services/huggingface/lister.rs
+++ b/core/src/services/huggingface/lister.rs
@@ -41,7 +41,6 @@ impl HuggingfaceLister {
}
}
-#[async_trait]
impl oio::PageList for HuggingfaceLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let response = self.core.hf_list(&self.path, self.recursive).await?;
diff --git a/core/src/services/ipfs/backend.rs b/core/src/services/ipfs/backend.rs
index 8e990e3138..964a1cef19 100644
--- a/core/src/services/ipfs/backend.rs
+++ b/core/src/services/ipfs/backend.rs
@@ -423,7 +423,6 @@ impl DirStream {
}
}
-#[async_trait]
impl oio::PageList for DirStream {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.backend.ipfs_list(&self.path).await?;
diff --git a/core/src/services/ipmfs/lister.rs b/core/src/services/ipmfs/lister.rs
index 9198b8e31a..0d97b0c438 100644
--- a/core/src/services/ipmfs/lister.rs
+++ b/core/src/services/ipmfs/lister.rs
@@ -44,7 +44,6 @@ impl IpmfsLister {
}
}
-#[async_trait]
impl oio::PageList for IpmfsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.backend.ipmfs_ls(&self.path).await?;
diff --git a/core/src/services/koofr/lister.rs b/core/src/services/koofr/lister.rs
index 8ad58a58d2..a206447682 100644
--- a/core/src/services/koofr/lister.rs
+++ b/core/src/services/koofr/lister.rs
@@ -43,7 +43,6 @@ impl KoofrLister {
}
}
-#[async_trait]
impl oio::PageList for KoofrLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.core.list(&self.path).await?;
diff --git a/core/src/services/obs/lister.rs b/core/src/services/obs/lister.rs
index ead6bbbeab..e26c00d720 100644
--- a/core/src/services/obs/lister.rs
+++ b/core/src/services/obs/lister.rs
@@ -48,7 +48,6 @@ impl ObsLister {
}
}
-#[async_trait]
impl oio::PageList for ObsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/onedrive/lister.rs b/core/src/services/onedrive/lister.rs
index 12c557867a..a84881faa6 100644
--- a/core/src/services/onedrive/lister.rs
+++ b/core/src/services/onedrive/lister.rs
@@ -43,7 +43,6 @@ impl OnedriveLister {
}
}
-#[async_trait]
impl oio::PageList for OnedriveLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let request_url = if ctx.token.is_empty() {
diff --git a/core/src/services/oss/lister.rs b/core/src/services/oss/lister.rs
index 926946a616..97606e0f15 100644
--- a/core/src/services/oss/lister.rs
+++ b/core/src/services/oss/lister.rs
@@ -56,7 +56,6 @@ impl OssLister {
}
}
-#[async_trait]
impl oio::PageList for OssLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/pcloud/lister.rs b/core/src/services/pcloud/lister.rs
index e4dd65c381..857376a1af 100644
--- a/core/src/services/pcloud/lister.rs
+++ b/core/src/services/pcloud/lister.rs
@@ -41,7 +41,6 @@ impl PcloudLister {
}
}
-#[async_trait]
impl oio::PageList for PcloudLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.core.list_folder(&self.path).await?;
diff --git a/core/src/services/s3/lister.rs b/core/src/services/s3/lister.rs
index 569d83ebee..dac9ceae4d 100644
--- a/core/src/services/s3/lister.rs
+++ b/core/src/services/s3/lister.rs
@@ -60,8 +60,6 @@ impl S3Lister {
}
}
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::PageList for S3Lister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/seafile/lister.rs b/core/src/services/seafile/lister.rs
index 792d628032..7762597e2d 100644
--- a/core/src/services/seafile/lister.rs
+++ b/core/src/services/seafile/lister.rs
@@ -44,7 +44,6 @@ impl SeafileLister {
}
}
-#[async_trait]
impl oio::PageList for SeafileLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let path = build_rooted_abs_path(&self.core.root, &self.path);
diff --git a/core/src/services/sftp/lister.rs b/core/src/services/sftp/lister.rs
index f7fe5aac9f..a99c467655 100644
--- a/core/src/services/sftp/lister.rs
+++ b/core/src/services/sftp/lister.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::pin::Pin;
use std::task::ready;
use std::task::Context;
@@ -27,6 +28,7 @@ use openssh_sftp_client::fs::ReadDir;
use super::error::parse_sftp_error;
use crate::raw::oio;
+use crate::raw::oio::Entry;
use crate::Result;
pub struct SftpLister {
@@ -45,22 +47,26 @@ impl SftpLister {
}
}
-#[async_trait]
impl oio::List for SftpLister {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- let item = ready!(self.dir.poll_next_unpin(cx))
- .transpose()
- .map_err(parse_sftp_error)?;
+ async fn next(&mut self) -> Result<Option<Entry>> {
+ loop {
+ let item = self
+ .dir
+ .next()
+ .await
+ .transpose()
+ .map_err(parse_sftp_error)?;
- match item {
- Some(e) => {
- if e.filename().to_str() == Some(".") || e.filename().to_str() == Some("..") {
- self.poll_next(cx)
- } else {
- Poll::Ready(Ok(Some(map_entry(self.prefix.as_str(), e))))
+ match item {
+ Some(e) => {
+ if e.filename().to_str() == Some(".") || e.filename().to_str() == Some("..") {
+ continue;
+ } else {
+ return Ok(Some(map_entry(self.prefix.as_str(), e)));
+ }
}
+ None => return Ok(None),
}
- None => Poll::Ready(Ok(None)),
}
}
}
diff --git a/core/src/services/swift/lister.rs b/core/src/services/swift/lister.rs
index ee90a17537..a7b6601157 100644
--- a/core/src/services/swift/lister.rs
+++ b/core/src/services/swift/lister.rs
@@ -43,7 +43,6 @@ impl SwiftLister {
}
}
-#[async_trait]
impl oio::PageList for SwiftLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let response = self
diff --git a/core/src/services/upyun/lister.rs b/core/src/services/upyun/lister.rs
index 4fdd068cef..c093ebca00 100644
--- a/core/src/services/upyun/lister.rs
+++ b/core/src/services/upyun/lister.rs
@@ -45,7 +45,6 @@ impl UpyunLister {
}
}
-#[async_trait]
impl oio::PageList for UpyunLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/vercel_blob/lister.rs b/core/src/services/vercel_blob/lister.rs
index e792d8f379..6cad3bef4e 100644
--- a/core/src/services/vercel_blob/lister.rs
+++ b/core/src/services/vercel_blob/lister.rs
@@ -42,7 +42,6 @@ impl VercelBlobLister {
}
}
-#[async_trait]
impl oio::PageList for VercelBlobLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let p = build_abs_path(&self.core.root, &self.path);
diff --git a/core/src/services/webdav/lister.rs b/core/src/services/webdav/lister.rs
index ab9d7c6085..386e99c48b 100644
--- a/core/src/services/webdav/lister.rs
+++ b/core/src/services/webdav/lister.rs
@@ -42,7 +42,6 @@ impl WebdavLister {
}
}
-#[async_trait]
impl oio::PageList for WebdavLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.core.webdav_list(&self.path, &self.args).await?;
diff --git a/core/src/services/webhdfs/lister.rs b/core/src/services/webhdfs/lister.rs
index 4a984bb4c6..2215671304 100644
--- a/core/src/services/webhdfs/lister.rs
+++ b/core/src/services/webhdfs/lister.rs
@@ -38,7 +38,6 @@ impl WebhdfsLister {
}
}
-#[async_trait]
impl oio::PageList for WebhdfsLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let file_status = if self.backend.disable_list_batch {
diff --git a/core/src/services/yandex_disk/lister.rs b/core/src/services/yandex_disk/lister.rs
index a116ce53a0..7cbcb4c012 100644
--- a/core/src/services/yandex_disk/lister.rs
+++ b/core/src/services/yandex_disk/lister.rs
@@ -44,7 +44,6 @@ impl YandexDiskLister {
}
}
-#[async_trait]
impl oio::PageList for YandexDiskLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let offset = if ctx.token.is_empty() {
diff --git a/core/src/types/list.rs b/core/src/types/list.rs
index fd0f2485ed..b780c77fbc 100644
--- a/core/src/types/list.rs
+++ b/core/src/types/list.rs
@@ -45,7 +45,11 @@ pub struct Lister {
/// required_metakey is the metakey required by users.
required_metakey: FlagSet<Metakey>,
+ fut: Option<BoxedStaticFuture<(oio::Lister, Result<Option<oio::Entry>>)>>,
+
/// tasks is used to store tasks that are run in concurrent.
+ ///
+ /// TODO: maybe we should move logic inside?
tasks: ConcurrentFutures<StatTask>,
errored: bool,
}
@@ -134,6 +138,7 @@ impl Lister {
lister: Some(lister),
required_metakey,
+ fut: None,
tasks: ConcurrentFutures::new(concurrent),
errored: false,
})
@@ -151,10 +156,13 @@ impl Stream for Lister {
// Trying to pull more tasks if there are more space.
if self.tasks.has_remaining() {
- if let Some(lister) = self.lister.as_mut() {
- match lister.poll_next(cx) {
- Poll::Pending => {}
- Poll::Ready(Ok(Some(oe))) => {
+ if let Some(fut) = self.fut.as_mut() {
+ let (lister, entry) = ready!(fut.as_mut().poll(cx));
+ self.lister = Some(lister);
+ self.fut = None;
+
+ match entry {
+ Ok(Some(oe)) => {
let (path, metadata) = oe.into_entry().into_parts();
if metadata.contains_metakey(self.required_metakey) {
self.tasks
@@ -168,14 +176,20 @@ impl Stream for Lister {
self.tasks.push_back(StatTask::Stating(Box::pin(fut)));
}
}
- Poll::Ready(Ok(None)) => {
+ Ok(None) => {
self.lister = None;
}
- Poll::Ready(Err(err)) => {
+ Err(err) => {
self.errored = true;
return Poll::Ready(Some(Err(err)));
}
+ }
+ } else if let Some(mut lister) = self.lister.take() {
+ let fut = async move {
+ let res = lister.next_dyn().await;
+ (lister, res)
};
+ self.fut = Some(Box::pin(fut));
}
}