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/11/15 15:48:38 UTC
(incubator-opendal) 01/12: Save work
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch polish-lister
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit c6123160467564fbe84ec46de3cb8751f11553c2
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Nov 15 15:28:13 2023 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/blocking.rs | 3 +-
core/src/layers/complete.rs | 11 +--
core/src/layers/concurrent_limit.rs | 4 +-
core/src/layers/error_context.rs | 4 +-
core/src/layers/immutable_index.rs | 6 +-
core/src/layers/logging.rs | 12 +--
core/src/layers/madsim.rs | 8 +-
core/src/layers/minitrace.rs | 12 +--
core/src/layers/oteltrace.rs | 4 +-
core/src/layers/retry.rs | 76 +++++++++------
core/src/layers/timeout.rs | 33 +++++--
core/src/layers/tracing.rs | 4 +-
core/src/raw/adapters/kv/backend.rs | 54 ++++++-----
core/src/raw/adapters/typed_kv/backend.rs | 53 +++++-----
core/src/raw/oio/list/api.rs | 19 ++--
core/src/raw/oio/list/into_flat_page.rs | 106 ++++++++++----------
core/src/raw/oio/list/into_hierarchy_pager.rs | 26 ++---
core/src/raw/oio/list/mod.rs | 5 +
core/src/raw/oio/list/page_list.rs | 121 +++++++++++++++++++++++
core/src/services/alluxio/backend.rs | 8 +-
core/src/services/alluxio/lister.rs | 28 ++----
core/src/services/azblob/backend.rs | 6 +-
core/src/services/azblob/lister.rs | 27 ++----
core/src/services/azdls/backend.rs | 6 +-
core/src/services/azdls/lister.rs | 44 +++------
core/src/services/azfile/backend.rs | 6 +-
core/src/services/azfile/lister.rs | 49 +++-------
core/src/services/cos/backend.rs | 8 +-
core/src/services/cos/lister.rs | 31 ++----
core/src/services/dbfs/backend.rs | 6 +-
core/src/services/dbfs/lister.rs | 28 ++----
core/src/services/fs/lister.rs | 61 ++++++++----
core/src/services/ftp/lister.rs | 49 +++++-----
core/src/services/gcs/backend.rs | 21 ++--
core/src/services/gcs/lister.rs | 34 +++----
core/src/services/gdrive/backend.rs | 8 +-
core/src/services/gdrive/core.rs | 25 ++---
core/src/services/gdrive/lister.rs | 72 +++++---------
core/src/services/hdfs/lister.rs | 49 +++++-----
core/src/services/ipfs/backend.rs | 26 ++---
core/src/services/ipmfs/backend.rs | 8 +-
core/src/services/ipmfs/lister.rs | 53 ++++------
core/src/services/obs/backend.rs | 8 +-
core/src/services/obs/lister.rs | 27 ++----
core/src/services/onedrive/backend.rs | 7 +-
core/src/services/onedrive/lister.rs | 135 +++++++++++---------------
core/src/services/oss/backend.rs | 20 ++--
core/src/services/oss/core.rs | 14 +--
core/src/services/oss/lister.rs | 34 +++----
core/src/services/s3/backend.rs | 20 ++--
core/src/services/s3/lister.rs | 35 +++----
core/src/services/sftp/lister.rs | 19 ++--
core/src/services/swift/backend.rs | 6 +-
core/src/services/swift/lister.rs | 22 ++---
core/src/services/webdav/backend.rs | 9 +-
core/src/services/webdav/lister.rs | 32 +++---
core/src/services/webhdfs/backend.rs | 75 +++-----------
core/src/services/webhdfs/lister.rs | 110 +++++++++------------
core/src/types/list.rs | 72 ++++++--------
59 files changed, 872 insertions(+), 987 deletions(-)
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index 67281ba25..3181062e7 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -315,7 +315,8 @@ 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<Vec<oio::Entry>>> {
- self.handle.block_on(self.inner.next())
+ todo!()
+ // self.handle.block_on(poll_fn(|cx| self.inner.poll_next(cx)))
}
}
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 7fd2037dc..ea4906cf6 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -27,7 +27,6 @@ use std::task::Poll;
use async_trait::async_trait;
use bytes::Bytes;
-use crate::raw::oio::Entry;
use crate::raw::oio::FlatLister;
use crate::raw::oio::HierarchyLister;
use crate::raw::oio::RangeReader;
@@ -639,13 +638,13 @@ where
A: Accessor<Lister = P>,
P: oio::List,
{
- async fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
use CompleteLister::*;
match self {
- AlreadyComplete(p) => p.next().await,
- NeedFlat(p) => p.next().await,
- NeedHierarchy(p) => p.next().await,
+ AlreadyComplete(p) => p.poll_next(cx),
+ NeedFlat(p) => p.poll_next(cx),
+ NeedHierarchy(p) => p.poll_next(cx),
}
}
}
@@ -655,7 +654,7 @@ where
A: Accessor<BlockingLister = P>,
P: oio::BlockingList,
{
- fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
use CompleteLister::*;
match self {
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index 00f9a9a8c..d839b97d9 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -310,8 +310,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for ConcurrentLimitWrapper<R> {
#[async_trait]
impl<R: oio::List> oio::List for ConcurrentLimitWrapper<R> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- self.inner.next().await
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ self.inner.poll_next(cx)
}
}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 3278d502c..c3a0feef8 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -453,8 +453,8 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
#[async_trait::async_trait]
impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- self.inner.next().await.map_err(|err| {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ self.inner.poll_next(cx).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 6137687b8..9d47d78ff 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -18,6 +18,7 @@
use std::collections::HashSet;
use std::fmt::Debug;
use std::mem;
+use std::task::{Context, Poll};
use async_trait::async_trait;
@@ -240,8 +241,9 @@ impl ImmutableDir {
#[async_trait]
impl oio::List for ImmutableDir {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ todo!()
+ // Ok(self.inner_next_page())
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index b556cc7aa..16b91cef5 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -1476,18 +1476,18 @@ impl<P> Drop for LoggingLister<P> {
#[async_trait]
impl<P: oio::List> oio::List for LoggingLister<P> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let res = self.inner.next().await;
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ let res = ready!(self.inner.poll_next(cx));
match &res {
- Ok(Some(des)) => {
+ Ok(Some(de)) => {
debug!(
target: LOGGING_TARGET,
- "service={} operation={} path={} -> listed {} entries",
+ "service={} operation={} path={} -> listed entry: {}",
self.ctx.scheme,
self.op,
self.path,
- des.len(),
+ de.path(),
);
}
Ok(None) => {
@@ -1515,7 +1515,7 @@ impl<P: oio::List> oio::List for LoggingLister<P> {
}
};
- res
+ Poll::Ready(res)
}
}
diff --git a/core/src/layers/madsim.rs b/core/src/layers/madsim.rs
index 9eb75da2d..1b5b602aa 100644
--- a/core/src/layers/madsim.rs
+++ b/core/src/layers/madsim.rs
@@ -39,8 +39,6 @@ use madsim::net::Endpoint;
#[cfg(madsim)]
use madsim::net::Payload;
-use crate::raw::oio;
-use crate::raw::oio::Entry;
use crate::raw::*;
use crate::*;
@@ -335,11 +333,11 @@ pub struct MadsimLister {}
#[async_trait]
impl oio::List for MadsimLister {
- async fn next(&mut self) -> crate::Result<Option<Vec<Entry>>> {
- Err(Error::new(
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<Option<oio::Entry>>> {
+ Poll::Ready(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 26df33004..b0f192ed3 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -372,14 +372,10 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for MinitraceWrapper<R> {
#[async_trait]
impl<R: oio::List> oio::List for MinitraceWrapper<R> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- self.inner
- .next()
- .in_span(Span::enter_with_parent(
- ListOperation::Next.into_static(),
- &self.span,
- ))
- .await
+ 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)
}
}
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 0244e3190..c4d8b6f0e 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -331,8 +331,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for OtelTraceWrapper<R> {
#[async_trait]
impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- self.inner.next().await
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ self.inner.poll_next(cx)
}
}
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index d0162ccdf..67528347d 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -1051,28 +1051,48 @@ impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWra
#[async_trait]
impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let mut backoff = self.builder.build();
-
- loop {
- match self.inner.next().await {
- Ok(v) => return Ok(v),
- Err(e) if !e.is_temporary() => return Err(e),
- Err(e) => match backoff.next() {
- None => return Err(e),
+ 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;
+ }
+
+ match ready!(self.inner.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()
+ }
+ };
+
+ match backoff.next() {
+ None => {
+ self.current_backoff = None;
+ Poll::Ready(Err(err))
+ }
Some(dur) => {
self.notify.intercept(
- &e,
+ &err,
dur,
&[
("operation", ListOperation::Next.into_static()),
("path", &self.path),
],
);
- tokio::time::sleep(dur).await;
- continue;
+ self.sleep = Some(Box::pin(tokio::time::sleep(dur)));
+ self.poll_next(cx)
}
- },
+ }
}
}
}
@@ -1296,37 +1316,33 @@ mod tests {
}
#[async_trait]
impl oio::List for MockLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
self.attempt += 1;
- match self.attempt {
+ let result = match self.attempt {
1 => Err(Error::new(
ErrorKind::RateLimited,
"retryable rate limited error from lister",
)
.set_temporary()),
- 2 => {
- let entries = vec![
- oio::Entry::new("hello", Metadata::new(EntryMode::FILE)),
- oio::Entry::new("world", Metadata::new(EntryMode::FILE)),
- ];
- Ok(Some(entries))
- }
+ 2 => Ok(Some(oio::Entry::new(
+ "hello",
+ Metadata::new(EntryMode::FILE),
+ ))),
3 => Err(
Error::new(ErrorKind::Unexpected, "retryable internal server error")
.set_temporary(),
),
- 4 => {
- let entries = vec![
- oio::Entry::new("2023/", Metadata::new(EntryMode::DIR)),
- oio::Entry::new("0208/", Metadata::new(EntryMode::DIR)),
- ];
- Ok(Some(entries))
- }
+ 4 => Ok(Some(oio::Entry::new(
+ "2023/",
+ Metadata::new(EntryMode::DIR),
+ ))),
5 => Ok(None),
_ => {
unreachable!()
}
- }
+ };
+
+ Poll::Ready(result)
}
}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 64b55ff18..8f04c3363 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -417,14 +417,33 @@ impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
#[async_trait]
impl<R: oio::List> oio::List for TimeoutWrapper<R> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- tokio::time::timeout(self.timeout, self.inner.next())
- .await
- .map_err(|_| {
- Error::new(ErrorKind::Unexpected, "operation timeout")
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ match self.start {
+ Some(start) => {
+ if start.elapsed() > self.timeout {
+ // Clean up the start time before return ready.
+ self.start = None;
+
+ return Poll::Ready(Err(Error::new(
+ ErrorKind::Unexpected,
+ "operation timeout",
+ )
.with_operation(ListOperation::Next)
.with_context("timeout", self.timeout.as_secs_f64().to_string())
- .set_temporary()
- })?
+ .set_temporary()));
+ }
+ }
+ None => {
+ self.start = Some(Instant::now());
+ }
+ }
+
+ match self.inner.poll_next(cx) {
+ Poll::Pending => Poll::Pending,
+ Poll::Ready(v) => {
+ self.start = None;
+ Poll::Ready(v)
+ }
+ }
}
}
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 418be8235..38ae5f2ad 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -366,8 +366,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for TracingWrapper<R> {
#[async_trait]
impl<R: oio::List> oio::List for TracingWrapper<R> {
#[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- self.inner.next().await
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ self.inner.poll_next(cx)
}
}
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index 4360e6f39..b5ddf9440 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -19,6 +19,7 @@ use std::sync::Arc;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
+use std::vec::IntoIter;
use async_trait::async_trait;
use bytes::Bytes;
@@ -303,47 +304,52 @@ where
pub struct KvLister {
root: String,
- inner: Option<Vec<String>>,
+ inner: IntoIter<String>,
}
impl KvLister {
fn new(root: &str, inner: Vec<String>) -> Self {
Self {
root: root.to_string(),
- inner: Some(inner),
+ inner: inner.into_iter(),
}
}
-
- fn inner_next_page(&mut self) -> Option<Vec<oio::Entry>> {
- let res = self
- .inner
- .take()?
- .into_iter()
- .map(|v| {
- let mode = if v.ends_with('/') {
- EntryMode::DIR
- } else {
- EntryMode::FILE
- };
-
- oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
- })
- .collect();
-
- Some(res)
- }
}
#[async_trait]
impl oio::List for KvLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ let entry = self.inner.next().map(|v| {
+ let mode = if v.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
+ });
+
+ Poll::Ready(Ok(entry))
}
}
impl oio::BlockingList for KvLister {
fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ Ok(Some(
+ self.inner
+ .as_slice()
+ .iter()
+ .map(|v| {
+ let mode = if v.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
+ })
+ .collect(),
+ ))
}
}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index 2a1da5966..e60e21649 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -19,6 +19,7 @@ use std::sync::Arc;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
+use std::vec::IntoIter;
use async_trait::async_trait;
use bytes::Bytes;
@@ -304,47 +305,51 @@ where
pub struct KvLister {
root: String,
- inner: Option<Vec<String>>,
+ inner: IntoIter<String>,
}
impl KvLister {
fn new(root: &str, inner: Vec<String>) -> Self {
Self {
root: root.to_string(),
- inner: Some(inner),
+ inner: inner.into_iter(),
}
}
-
- fn inner_next_page(&mut self) -> Option<Vec<oio::Entry>> {
- let res = self
- .inner
- .take()?
- .into_iter()
- .map(|v| {
- let mode = if v.ends_with('/') {
- EntryMode::DIR
- } else {
- EntryMode::FILE
- };
-
- oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
- })
- .collect();
-
- Some(res)
- }
}
#[async_trait]
impl oio::List for KvLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ let entry = self.inner.next().map(|v| {
+ let mode = if v.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
+ });
+
+ Poll::Ready(Ok(entry))
}
}
impl oio::BlockingList for KvLister {
fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ Ok(Some(
+ self.inner
+ .clone()
+ .map(|v| {
+ let mode = if v.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
+ })
+ .collect(),
+ ))
}
}
diff --git a/core/src/raw/oio/list/api.rs b/core/src/raw/oio/list/api.rs
index c08f5d294..b7899e60d 100644
--- a/core/src/raw/oio/list/api.rs
+++ b/core/src/raw/oio/list/api.rs
@@ -17,6 +17,7 @@
use std::fmt::Display;
use std::fmt::Formatter;
+use std::task::{Context, Poll};
use async_trait::async_trait;
@@ -27,7 +28,7 @@ use crate::*;
#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq)]
#[non_exhaustive]
pub enum ListOperation {
- /// Operation for [`List::next`]
+ /// Operation for [`List::poll_next`]
Next,
/// Operation for [`BlockingList::next`]
BlockingNext,
@@ -64,7 +65,7 @@ pub trait List: Send + Sync + 'static {
///
/// `Ok(None)` means all pages have been returned. Any following call
/// to `next` will always get the same result.
- async fn next(&mut self) -> Result<Option<Vec<Entry>>>;
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>>;
}
/// The boxed version of [`List`]
@@ -72,24 +73,24 @@ pub type Lister = Box<dyn List>;
#[async_trait]
impl<P: List + ?Sized> List for Box<P> {
- async fn next(&mut self) -> Result<Option<Vec<Entry>>> {
- (**self).next().await
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
+ (**self).poll_next(cx)
}
}
#[async_trait]
impl List for () {
- async fn next(&mut self) -> Result<Option<Vec<Entry>>> {
- Ok(None)
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
+ Poll::Ready(Ok(None))
}
}
#[async_trait]
impl<P: List> List for Option<P> {
- async fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
match self {
- Some(p) => p.next().await,
- None => Ok(None),
+ Some(p) => p.poll_next(cx),
+ None => Poll::Ready(Ok(None)),
}
}
}
diff --git a/core/src/raw/oio/list/into_flat_page.rs b/core/src/raw/oio/list/into_flat_page.rs
index ee9966b4c..6c2ef6207 100644
--- a/core/src/raw/oio/list/into_flat_page.rs
+++ b/core/src/raw/oio/list/into_flat_page.rs
@@ -17,6 +17,7 @@
use std::collections::VecDeque;
use std::mem;
+use std::task::{Context, Poll};
use async_trait::async_trait;
@@ -95,58 +96,59 @@ where
A: Accessor<Lister = P>,
P: oio::List,
{
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- loop {
- if let Some(de) = self.dirs.pop_back() {
- let (_, op) = self.acc.list(de.path(), OpList::new()).await?;
- self.listers.push((op, de, vec![]))
- }
-
- let (mut lister, de, mut buf) = match self.listers.pop() {
- Some((lister, de, buf)) => (lister, de, buf),
- None => {
- if !self.res.is_empty() {
- return Ok(Some(mem::take(&mut self.res)));
- }
- return Ok(None);
- }
- };
-
- if buf.is_empty() {
- match lister.next().await? {
- Some(v) => {
- buf = v;
- }
- None => {
- // Only push entry if it's not root dir
- if de.path() != self.root {
- self.res.push(de);
- }
- continue;
- }
- }
- }
-
- let mut buf = VecDeque::from(buf);
- loop {
- if let Some(oe) = buf.pop_front() {
- if oe.mode().is_dir() {
- self.dirs.push_back(oe);
- self.listers.push((lister, de, buf.into()));
- break;
- } else {
- self.res.push(oe)
- }
- } else {
- self.listers.push((lister, de, vec![]));
- break;
- }
- }
-
- if self.res.len() >= self.size {
- return Ok(Some(mem::take(&mut self.res)));
- }
- }
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ todo!()
+ // loop {
+ // if let Some(de) = self.dirs.pop_back() {
+ // let (_, op) = self.acc.list(de.path(), OpList::new()).await?;
+ // self.listers.push((op, de, vec![]))
+ // }
+ //
+ // let (mut lister, de, mut buf) = match self.listers.pop() {
+ // Some((lister, de, buf)) => (lister, de, buf),
+ // None => {
+ // if !self.res.is_empty() {
+ // return Ok(Some(mem::take(&mut self.res)));
+ // }
+ // return Ok(None);
+ // }
+ // };
+ //
+ // if buf.is_empty() {
+ // match lister.poll_next(cx)? {
+ // Some(v) => {
+ // buf = v;
+ // }
+ // None => {
+ // // Only push entry if it's not root dir
+ // if de.path() != self.root {
+ // self.res.push(de);
+ // }
+ // continue;
+ // }
+ // }
+ // }
+ //
+ // let mut buf = VecDeque::from(buf);
+ // loop {
+ // if let Some(oe) = buf.pop_front() {
+ // if oe.mode().is_dir() {
+ // self.dirs.push_back(oe);
+ // self.listers.push((lister, de, buf.into()));
+ // break;
+ // } else {
+ // self.res.push(oe)
+ // }
+ // } else {
+ // self.listers.push((lister, de, vec![]));
+ // break;
+ // }
+ // }
+ //
+ // if self.res.len() >= self.size {
+ // return Ok(Some(mem::take(&mut self.res)));
+ // }
+ // }
}
}
diff --git a/core/src/raw/oio/list/into_hierarchy_pager.rs b/core/src/raw/oio/list/into_hierarchy_pager.rs
index 2971ad2a1..217371b45 100644
--- a/core/src/raw/oio/list/into_hierarchy_pager.rs
+++ b/core/src/raw/oio/list/into_hierarchy_pager.rs
@@ -16,6 +16,7 @@
// under the License.
use std::collections::HashSet;
+use std::task::{Context, Poll};
use async_trait::async_trait;
@@ -118,18 +119,19 @@ impl<P> HierarchyLister<P> {
#[async_trait]
impl<P: oio::List> oio::List for HierarchyLister<P> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let page = self.lister.next().await?;
-
- let entries = if let Some(entries) = page {
- entries
- } else {
- return Ok(None);
- };
-
- let entries = self.filter_entries(entries);
-
- Ok(Some(entries))
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ todo!()
+ // let page = self.lister.poll_next(cx)?;
+ //
+ // let entries = if let Some(entries) = page {
+ // entries
+ // } else {
+ // return Ok(None);
+ // };
+ //
+ // let entries = self.filter_entries(entries);
+ //
+ // Ok(Some(entries))
}
}
diff --git a/core/src/raw/oio/list/mod.rs b/core/src/raw/oio/list/mod.rs
index 0a4b0e39e..63fe41292 100644
--- a/core/src/raw/oio/list/mod.rs
+++ b/core/src/raw/oio/list/mod.rs
@@ -22,6 +22,11 @@ pub use api::List;
pub use api::ListOperation;
pub use api::Lister;
+mod page_list;
+pub use page_list::PageContext;
+pub use page_list::PageList;
+pub use page_list::PageLister;
+
mod into_flat_page;
pub use into_flat_page::into_flat_page;
pub use into_flat_page::FlatLister;
diff --git a/core/src/raw/oio/list/page_list.rs b/core/src/raw/oio/list/page_list.rs
new file mode 100644
index 000000000..ed14a57d8
--- /dev/null
+++ b/core/src/raw/oio/list/page_list.rs
@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use async_trait::async_trait;
+use futures::future::BoxFuture;
+use std::collections::VecDeque;
+use std::task::{ready, Context, Poll};
+
+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`
+#[async_trait]
+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<()>;
+}
+
+pub struct PageContext {
+ /// done is used to indicate whether the list operation is done.
+ pub done: bool,
+ /// token is used by underlying storage services to fetch next page.
+ pub token: String,
+ /// entries is used to store entries fetched from underlying storage.
+ ///
+ /// Please always reuse the same `VecDeque` to avoid unnecessary memory allocation.
+ /// PageLister makes sure that entries is reset before calling `next_page`. Implementor
+ /// can calling `push_back` on `entries` directly.
+ pub entries: VecDeque<oio::Entry>,
+}
+
+pub struct PageLister<L: PageList> {
+ state: State<L>,
+}
+
+enum State<L> {
+ Idle(Option<(L, PageContext)>),
+ Fetch(BoxFuture<'static, ((L, PageContext), Result<()>)>),
+}
+
+/// # 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,
+{
+ /// 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(),
+ },
+ ))),
+ }
+ }
+}
+
+impl<L> oio::List for PageLister<L>
+where
+ L: PageList,
+{
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<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?;
+ }
+ }
+ }
+ }
+}
diff --git a/core/src/services/alluxio/backend.rs b/core/src/services/alluxio/backend.rs
index d5c9b3a12..64db4461f 100644
--- a/core/src/services/alluxio/backend.rs
+++ b/core/src/services/alluxio/backend.rs
@@ -185,7 +185,7 @@ impl Accessor for AlluxioBackend {
type BlockingReader = ();
type Writer = AlluxioWriters;
type BlockingWriter = ();
- type Lister = AlluxioLister;
+ type Lister = oio::PageLister<AlluxioLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -253,10 +253,8 @@ impl Accessor for AlluxioBackend {
}
async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- AlluxioLister::new(self.core.clone(), path),
- ))
+ let l = AlluxioLister::new(self.core.clone(), path);
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/alluxio/lister.rs b/core/src/services/alluxio/lister.rs
index c1efa6b3a..5e56dbff8 100644
--- a/core/src/services/alluxio/lister.rs
+++ b/core/src/services/alluxio/lister.rs
@@ -29,8 +29,6 @@ pub struct AlluxioLister {
core: Arc<AlluxioCore>,
path: String,
-
- done: bool,
}
impl AlluxioLister {
@@ -38,23 +36,19 @@ impl AlluxioLister {
AlluxioLister {
core,
path: path.to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for AlluxioLister {
- async fn next(&mut self) -> Result<Option<Vec<Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for AlluxioLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let result = self.core.list_status(&self.path).await;
match result {
Ok(file_infos) => {
- let mut entries = vec![];
+ ctx.done = true;
+
for file_info in file_infos {
let path: String = file_info.path.clone();
let path = if file_info.folder {
@@ -62,24 +56,18 @@ impl oio::List for AlluxioLister {
} else {
path
};
- entries.push(Entry::new(
+ ctx.entries.push_back(Entry::new(
&build_rel_path(&self.core.root, &path),
file_info.try_into()?,
));
}
- if entries.is_empty() {
- return Ok(None);
- }
-
- self.done = true;
-
- Ok(Some(entries))
+ Ok(())
}
Err(e) => {
if e.kind() == ErrorKind::NotFound {
- self.done = true;
- return Ok(None);
+ ctx.done = true;
+ return Ok(());
}
Err(e)
}
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index a87907477..b3e8caad9 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -546,7 +546,7 @@ impl Accessor for AzblobBackend {
type BlockingReader = ();
type Writer = AzblobWriters;
type BlockingWriter = ();
- type Lister = AzblobLister;
+ type Lister = oio::PageLister<AzblobLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -689,14 +689,14 @@ impl Accessor for AzblobBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = AzblobLister::new(
+ let l = AzblobLister::new(
self.core.clone(),
path.to_string(),
args.recursive(),
args.limit(),
);
- Ok((RpList::default(), op))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
diff --git a/core/src/services/azblob/lister.rs b/core/src/services/azblob/lister.rs
index 25ccc1248..96e15821e 100644
--- a/core/src/services/azblob/lister.rs
+++ b/core/src/services/azblob/lister.rs
@@ -33,9 +33,6 @@ pub struct AzblobLister {
path: String,
delimiter: &'static str,
limit: Option<usize>,
-
- next_marker: String,
- done: bool,
}
impl AzblobLister {
@@ -47,23 +44,16 @@ impl AzblobLister {
path,
delimiter,
limit,
-
- next_marker: "".to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for AzblobLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for AzblobLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
- .azblob_list_blobs(&self.path, &self.next_marker, self.delimiter, self.limit)
+ .azblob_list_blobs(&self.path, &ctx.token, self.delimiter, self.limit)
.await?;
if resp.status() != http::StatusCode::OK {
@@ -80,12 +70,11 @@ impl oio::List for AzblobLister {
//
// - Check `next_marker`
if let Some(next_marker) = output.next_marker.as_ref() {
- self.done = next_marker.is_empty();
+ ctx.done = next_marker.is_empty();
};
- self.next_marker = output.next_marker.clone().unwrap_or_default();
+ ctx.token = output.next_marker.clone().unwrap_or_default();
let prefixes = output.blobs.blob_prefix;
- let mut entries = Vec::with_capacity(prefixes.len() + output.blobs.blob.len());
for prefix in prefixes {
let de = oio::Entry::new(
@@ -93,7 +82,7 @@ impl oio::List for AzblobLister {
Metadata::new(EntryMode::DIR),
);
- entries.push(de)
+ ctx.entries.push_back(de)
}
for object in output.blobs.blob {
@@ -116,10 +105,10 @@ impl oio::List for AzblobLister {
let de = oio::Entry::new(&build_rel_path(&self.core.root, &object.name), meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index f5e08abe4..b1a7366e0 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -233,7 +233,7 @@ impl Accessor for AzdlsBackend {
type BlockingReader = ();
type Writer = AzdlsWriters;
type BlockingWriter = ();
- type Lister = AzdlsLister;
+ type Lister = oio::PageLister<AzdlsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -366,9 +366,9 @@ impl Accessor for AzdlsBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = AzdlsLister::new(self.core.clone(), path.to_string(), args.limit());
+ let l = AzdlsLister::new(self.core.clone(), path.to_string(), args.limit());
- Ok((RpList::default(), op))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/azdls/lister.rs b/core/src/services/azdls/lister.rs
index a624aa31e..db7904f94 100644
--- a/core/src/services/azdls/lister.rs
+++ b/core/src/services/azdls/lister.rs
@@ -31,41 +31,29 @@ pub struct AzdlsLister {
path: String,
limit: Option<usize>,
-
- continuation: String,
- done: bool,
}
impl AzdlsLister {
pub fn new(core: Arc<AzdlsCore>, path: String, limit: Option<usize>) -> Self {
- Self {
- core,
- path,
- limit,
-
- continuation: "".to_string(),
- done: false,
- }
+ Self { core, path, limit }
}
}
#[async_trait]
-impl oio::List for AzdlsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for AzdlsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
- .azdls_list(&self.path, &self.continuation, self.limit)
+ .azdls_list(&self.path, &ctx.token, self.limit)
.await?;
- // Azdls will return not found for not-exist path.
+ // azdls will return not found for not-exist path.
if resp.status() == http::StatusCode::NOT_FOUND {
resp.into_body().consume().await?;
- return Ok(None);
+ ctx.done = true;
+ return Ok(());
}
+
if resp.status() != http::StatusCode::OK {
return Err(parse_error(resp).await?);
}
@@ -76,19 +64,15 @@ impl oio::List for AzdlsLister {
Error::new(ErrorKind::Unexpected, "header value is not valid string")
.set_source(err)
})?;
- self.continuation = value.to_string();
+ ctx.token = value.to_string();
} else {
- self.continuation = "".to_string();
- self.done = true;
+ ctx.token = "".to_string();
+ ctx.done = true;
}
let bs = resp.into_body().bytes().await?;
- let output: Output = de::from_slice(&bs).map_err(|e| {
- Error::new(ErrorKind::Unexpected, "deserialize json from response").set_source(e)
- })?;
-
- let mut entries = Vec::with_capacity(output.paths.len());
+ let output: Output = de::from_slice(&bs).map_err(new_json_deserialize_error)?;
for object in output.paths {
// Azdls will return `"true"` and `"false"` for is_directory.
@@ -114,10 +98,10 @@ impl oio::List for AzdlsLister {
let de = oio::Entry::new(&path, meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/azfile/backend.rs b/core/src/services/azfile/backend.rs
index d11b34efd..3b87cb1f0 100644
--- a/core/src/services/azfile/backend.rs
+++ b/core/src/services/azfile/backend.rs
@@ -251,7 +251,7 @@ impl Accessor for AzfileBackend {
type BlockingReader = ();
type Writer = AzfileWriters;
type BlockingWriter = ();
- type Lister = AzfileLister;
+ type Lister = oio::PageLister<AzfileLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -396,9 +396,9 @@ impl Accessor for AzfileBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = AzfileLister::new(self.core.clone(), path.to_string(), args.limit());
+ let l = AzfileLister::new(self.core.clone(), path.to_string(), args.limit());
- Ok((RpList::default(), op))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/azfile/lister.rs b/core/src/services/azfile/lister.rs
index cd17aef33..cdd8fc980 100644
--- a/core/src/services/azfile/lister.rs
+++ b/core/src/services/azfile/lister.rs
@@ -32,39 +32,28 @@ pub struct AzfileLister {
core: Arc<AzfileCore>,
path: String,
limit: Option<usize>,
- done: bool,
- continuation: String,
}
impl AzfileLister {
pub fn new(core: Arc<AzfileCore>, path: String, limit: Option<usize>) -> Self {
- Self {
- core,
- path,
- limit,
- done: false,
- continuation: "".to_string(),
- }
+ Self { core, path, limit }
}
}
#[async_trait]
-impl oio::List for AzfileLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for AzfileLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
- .azfile_list(&self.path, &self.limit, &self.continuation)
+ .azfile_list(&self.path, &self.limit, &ctx.token)
.await?;
let status = resp.status();
if status != StatusCode::OK {
if status == StatusCode::NOT_FOUND {
- return Ok(None);
+ ctx.done = true;
+ return Ok(());
}
return Err(parse_error(resp).await?);
}
@@ -73,11 +62,13 @@ impl oio::List for AzfileLister {
let text = String::from_utf8(bs.to_vec()).expect("response convert to string must success");
- let results: EnumerationResults = from_str(&text).map_err(|e| {
- Error::new(ErrorKind::Unexpected, "deserialize xml from response").set_source(e)
- })?;
+ let results: EnumerationResults = from_str(&text).map_err(new_xml_deserialize_error)?;
- let mut entries = Vec::new();
+ if results.next_marker.is_empty() {
+ ctx.done = true;
+ } else {
+ ctx.token = results.next_marker;
+ }
for file in results.entries.file {
let meta = Metadata::new(EntryMode::FILE)
@@ -85,7 +76,7 @@ impl oio::List for AzfileLister {
.with_content_length(file.properties.content_length.unwrap_or(0))
.with_last_modified(parse_datetime_from_rfc2822(&file.properties.last_modified)?);
let path = self.path.clone().trim_start_matches('/').to_string() + &file.name;
- entries.push(oio::Entry::new(&path, meta));
+ ctx.entries.push_back(oio::Entry::new(&path, meta));
}
for dir in results.entries.directory {
@@ -93,20 +84,10 @@ impl oio::List for AzfileLister {
.with_etag(dir.properties.etag)
.with_last_modified(parse_datetime_from_rfc2822(&dir.properties.last_modified)?);
let path = self.path.clone().trim_start_matches('/').to_string() + &dir.name + "/";
- entries.push(oio::Entry::new(&path, meta));
+ ctx.entries.push_back(oio::Entry::new(&path, meta));
}
- if results.next_marker.is_empty() {
- self.done = true;
- } else {
- self.continuation = results.next_marker;
- }
-
- if entries.is_empty() {
- Ok(None)
- } else {
- Ok(Some(entries))
- }
+ Ok(())
}
}
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
index dbf2f9b1f..cf6616c34 100644
--- a/core/src/services/cos/backend.rs
+++ b/core/src/services/cos/backend.rs
@@ -245,7 +245,7 @@ impl Accessor for CosBackend {
type BlockingReader = ();
type Writer = CosWriters;
type BlockingWriter = ();
- type Lister = CosLister;
+ type Lister = oio::PageLister<CosLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -422,9 +422,7 @@ impl Accessor for CosBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- CosLister::new(self.core.clone(), path, args.recursive(), args.limit()),
- ))
+ let l = CosLister::new(self.core.clone(), path, args.recursive(), args.limit());
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/cos/lister.rs b/core/src/services/cos/lister.rs
index 55d4bd812..a6a402a0d 100644
--- a/core/src/services/cos/lister.rs
+++ b/core/src/services/cos/lister.rs
@@ -36,9 +36,6 @@ pub struct CosLister {
path: String,
delimiter: &'static str,
limit: Option<usize>,
-
- next_marker: String,
- done: bool,
}
impl CosLister {
@@ -49,23 +46,16 @@ impl CosLister {
path: path.to_string(),
delimiter,
limit,
-
- next_marker: "".to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for CosLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for CosLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
- .cos_list_objects(&self.path, &self.next_marker, self.delimiter, self.limit)
+ .cos_list_objects(&self.path, &ctx.token, self.delimiter, self.limit)
.await?;
if resp.status() != http::StatusCode::OK {
@@ -80,22 +70,19 @@ impl oio::List for CosLister {
// Try our best to check whether this list is done.
//
// - Check `next_marker`
- self.done = match output.next_marker.as_ref() {
+ ctx.done = match output.next_marker.as_ref() {
None => true,
Some(next_marker) => next_marker.is_empty(),
};
- self.next_marker = output.next_marker.clone().unwrap_or_default();
-
- let common_prefixes = output.common_prefixes;
- let mut entries = Vec::with_capacity(common_prefixes.len() + output.contents.len());
+ ctx.token = output.next_marker.clone().unwrap_or_default();
- for prefix in common_prefixes {
+ for prefix in output.common_prefixes {
let de = oio::Entry::new(
&build_rel_path(&self.core.root, &prefix.prefix),
Metadata::new(EntryMode::DIR),
);
- entries.push(de);
+ ctx.entries.push_back(de);
}
for object in output.contents {
@@ -107,10 +94,10 @@ impl oio::List for CosLister {
let de = oio::Entry::new(&build_rel_path(&self.core.root, &object.key), meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/dbfs/backend.rs b/core/src/services/dbfs/backend.rs
index 4b13e10ba..8da90e2b7 100644
--- a/core/src/services/dbfs/backend.rs
+++ b/core/src/services/dbfs/backend.rs
@@ -158,7 +158,7 @@ impl Accessor for DbfsBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<DbfsWriter>;
type BlockingWriter = ();
- type Lister = DbfsLister;
+ type Lister = oio::PageLister<DbfsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -276,9 +276,9 @@ impl Accessor for DbfsBackend {
}
async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = DbfsLister::new(self.core.clone(), path.to_string());
+ let l = DbfsLister::new(self.core.clone(), path.to_string());
- Ok((RpList::default(), op))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/dbfs/lister.rs b/core/src/services/dbfs/lister.rs
index fd27b912b..093b4c478 100644
--- a/core/src/services/dbfs/lister.rs
+++ b/core/src/services/dbfs/lister.rs
@@ -30,32 +30,24 @@ use super::error::parse_error;
pub struct DbfsLister {
core: Arc<DbfsCore>,
path: String,
- done: bool,
}
impl DbfsLister {
pub fn new(core: Arc<DbfsCore>, path: String) -> Self {
- Self {
- core,
- path,
- done: false,
- }
+ Self { core, path }
}
}
#[async_trait]
-impl oio::List for DbfsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for DbfsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let response = self.core.dbfs_list(&self.path).await?;
let status_code = response.status();
if !status_code.is_success() {
if status_code == StatusCode::NOT_FOUND {
- return Ok(None);
+ ctx.done = true;
+ return Ok(());
}
let error = parse_error(response).await?;
return Err(error);
@@ -65,11 +57,9 @@ impl oio::List for DbfsLister {
let mut decoded_response =
serde_json::from_slice::<DbfsOutputList>(&bytes).map_err(new_json_deserialize_error)?;
- self.done = true;
-
- let mut entries = Vec::with_capacity(decoded_response.files.len());
+ ctx.done = true;
- while let Some(status) = decoded_response.files.pop() {
+ for status in decoded_response.files {
let entry: oio::Entry = match status.is_dir {
true => {
let normalized_path = format!("{}/", &status.path);
@@ -88,9 +78,9 @@ impl oio::List for DbfsLister {
oio::Entry::new(&status.path, meta)
}
};
- entries.push(entry);
+ ctx.entries.push_back(entry);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/fs/lister.rs b/core/src/services/fs/lister.rs
index 403f674e8..13108b691 100644
--- a/core/src/services/fs/lister.rs
+++ b/core/src/services/fs/lister.rs
@@ -15,10 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+use std::fs::FileType;
use std::path::Path;
use std::path::PathBuf;
+use std::task::{ready, Context, Poll};
use async_trait::async_trait;
+use futures::future::BoxFuture;
+use futures::FutureExt;
use crate::raw::*;
use crate::EntryMode;
@@ -30,6 +34,8 @@ pub struct FsLister<P> {
size: usize,
rd: P,
+
+ fut: Option<BoxFuture<'static, (tokio::fs::DirEntry, Result<FileType>)>>,
}
impl<P> FsLister<P> {
@@ -38,19 +44,35 @@ impl<P> FsLister<P> {
root: root.to_owned(),
size: limit.unwrap_or(1000),
rd,
+
+ fut: None,
}
}
}
+/// # Safety
+///
+/// We will only take `&mut Self` reference for FsLister.
+unsafe impl<P> Sync for FsLister<P> {}
+
#[async_trait]
impl oio::List for FsLister<tokio::fs::ReadDir> {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let mut oes: Vec<oio::Entry> = Vec::with_capacity(self.size);
-
- for _ in 0..self.size {
- let de = match self.rd.next_entry().await.map_err(new_std_io_error)? {
- Some(de) => de,
- None => break,
+ 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();
@@ -62,25 +84,30 @@ impl oio::List for FsLister<tokio::fs::ReadDir> {
.replace('\\', "/"),
);
- // On Windows and most Unix platforms this function is free
- // (no extra system calls needed), but some Unix platforms may
- // require the equivalent call to symlink_metadata to learn about
- // the target file type.
- let file_type = de.file_type().await.map_err(new_std_io_error)?;
-
- let d = if file_type.is_file() {
+ let d = if ft.is_file() {
oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
- } else if file_type.is_dir() {
+ } 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))
};
- oes.push(d)
+ return Poll::Ready(Ok(Some(d)));
}
- Ok(if oes.is_empty() { None } else { Some(oes) })
+ 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)),
+ }
}
}
diff --git a/core/src/services/ftp/lister.rs b/core/src/services/ftp/lister.rs
index ba0892e8c..354bd4379 100644
--- a/core/src/services/ftp/lister.rs
+++ b/core/src/services/ftp/lister.rs
@@ -17,6 +17,7 @@
use std::str;
use std::str::FromStr;
+use std::task::{Context, Poll};
use std::vec::IntoIter;
use async_trait::async_trait;
@@ -43,35 +44,29 @@ impl FtpLister {
#[async_trait]
impl oio::List for FtpLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let mut oes: Vec<oio::Entry> = Vec::with_capacity(self.size);
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<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)),
+ };
- for _ in 0..self.size {
- 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 => break,
- };
+ let path = self.path.to_string() + de.name();
- let path = self.path.to_string() + de.name();
+ let entry = if de.is_file() {
+ oio::Entry::new(
+ &path,
+ Metadata::new(EntryMode::FILE)
+ .with_content_length(de.size() as u64)
+ .with_last_modified(de.modified().into()),
+ )
+ } else if de.is_directory() {
+ oio::Entry::new(&format!("{}/", &path), Metadata::new(EntryMode::DIR))
+ } else {
+ oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
+ };
- let d = if de.is_file() {
- oio::Entry::new(
- &path,
- Metadata::new(EntryMode::FILE)
- .with_content_length(de.size() as u64)
- .with_last_modified(de.modified().into()),
- )
- } else if de.is_directory() {
- oio::Entry::new(&format!("{}/", &path), Metadata::new(EntryMode::DIR))
- } else {
- oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
- };
-
- oes.push(d)
- }
-
- Ok(if oes.is_empty() { None } else { Some(oes) })
+ Poll::Ready(Ok(Some(entry)))
}
}
diff --git a/core/src/services/gcs/backend.rs b/core/src/services/gcs/backend.rs
index c2340b7c5..d528479dc 100644
--- a/core/src/services/gcs/backend.rs
+++ b/core/src/services/gcs/backend.rs
@@ -312,7 +312,7 @@ impl Accessor for GcsBackend {
type BlockingReader = ();
type Writer = GcsWriters;
type BlockingWriter = ();
- type Lister = GcsLister;
+ type Lister = oio::PageLister<GcsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -472,16 +472,15 @@ impl Accessor for GcsBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- GcsLister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- ),
- ))
+ let l = GcsLister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ );
+
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
diff --git a/core/src/services/gcs/lister.rs b/core/src/services/gcs/lister.rs
index 97496f353..244ca2a93 100644
--- a/core/src/services/gcs/lister.rs
+++ b/core/src/services/gcs/lister.rs
@@ -38,9 +38,6 @@ pub struct GcsLister {
/// Filter results to objects whose names are lexicographically
/// **equal to or after** startOffset
start_after: Option<String>,
-
- page_token: String,
- done: bool,
}
impl GcsLister {
@@ -60,28 +57,25 @@ impl GcsLister {
delimiter,
limit,
start_after: start_after.map(String::from),
-
- page_token: "".to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for GcsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for GcsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
.gcs_list_objects(
&self.path,
- &self.page_token,
+ &ctx.token,
self.delimiter,
self.limit,
- self.start_after.clone(),
+ if ctx.token.is_empty() {
+ self.start_after.clone()
+ } else {
+ None
+ },
)
.await?;
@@ -94,20 +88,18 @@ impl oio::List for GcsLister {
serde_json::from_slice(&bytes).map_err(new_json_deserialize_error)?;
if let Some(token) = &output.next_page_token {
- self.page_token = token.clone();
+ ctx.token = token.clone();
} else {
- self.done = true;
+ ctx.done = true;
}
- let mut entries = Vec::with_capacity(output.prefixes.len() + output.items.len());
-
for prefix in output.prefixes {
let de = oio::Entry::new(
&build_rel_path(&self.core.root, &prefix),
Metadata::new(EntryMode::DIR),
);
- entries.push(de);
+ ctx.entries.push_back(de);
}
for object in output.items {
@@ -139,10 +131,10 @@ impl oio::List for GcsLister {
let de = oio::Entry::new(path, meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/gdrive/backend.rs b/core/src/services/gdrive/backend.rs
index f0d9da1b1..8f82ede86 100644
--- a/core/src/services/gdrive/backend.rs
+++ b/core/src/services/gdrive/backend.rs
@@ -45,7 +45,7 @@ impl Accessor for GdriveBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<GdriveWriter>;
type BlockingWriter = ();
- type Lister = GdriveLister;
+ type Lister = oio::PageLister<GdriveLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -213,10 +213,8 @@ impl Accessor for GdriveBackend {
}
async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- GdriveLister::new(path.into(), self.core.clone()),
- ))
+ let l = GdriveLister::new(path.into(), self.core.clone());
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
diff --git a/core/src/services/gdrive/core.rs b/core/src/services/gdrive/core.rs
index 3239b281b..f6517419d 100644
--- a/core/src/services/gdrive/core.rs
+++ b/core/src/services/gdrive/core.rs
@@ -310,7 +310,7 @@ impl GdriveCore {
&self,
path: &str,
page_size: i32,
- next_page_token: Option<String>,
+ next_page_token: &str,
) -> Result<Response<IncomingAsyncBody>> {
let file_id = self.get_file_id_by_path(path).await;
@@ -341,22 +341,13 @@ impl GdriveCore {
},
};
- let url = match next_page_token {
- Some(page_token) => {
- format!(
- "https://www.googleapis.com/drive/v3/files?pageSize={}&pageToken={}&q={}",
- page_size,
- page_token,
- percent_encode_path(q.as_str())
- )
- }
- None => {
- format!(
- "https://www.googleapis.com/drive/v3/files?pageSize={}&q={}",
- page_size,
- percent_encode_path(q.as_str())
- )
- }
+ let mut url = format!(
+ "https://www.googleapis.com/drive/v3/files?pageSize={}&q={}",
+ page_size,
+ percent_encode_path(&q)
+ );
+ if !next_page_token.is_empty() {
+ url += &format!("&pageToken={next_page_token}");
};
let mut req = Request::get(&url)
diff --git a/core/src/services/gdrive/lister.rs b/core/src/services/gdrive/lister.rs
index 53d8f86c4..053a0fabb 100644
--- a/core/src/services/gdrive/lister.rs
+++ b/core/src/services/gdrive/lister.rs
@@ -23,80 +23,56 @@ use http::StatusCode;
use super::core::GdriveCore;
use super::core::GdriveFileList;
use super::error::parse_error;
-use crate::raw::build_rel_path;
-use crate::raw::build_rooted_abs_path;
-use crate::raw::new_json_deserialize_error;
-use crate::raw::oio::{self};
-use crate::EntryMode;
-use crate::Metadata;
-use crate::Result;
+use crate::raw::*;
+use crate::*;
+
pub struct GdriveLister {
path: String,
core: Arc<GdriveCore>,
- next_page_token: Option<String>,
- done: bool,
}
impl GdriveLister {
pub fn new(path: String, core: Arc<GdriveCore>) -> Self {
- Self {
- path,
- core,
- next_page_token: None,
- done: false,
- }
+ Self { path, core }
}
}
#[async_trait]
-impl oio::List for GdriveLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
- let resp = self
- .core
- .gdrive_list(&self.path, 100, self.next_page_token.clone())
- .await?;
+impl oio::PageList for GdriveLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ let resp = self.core.gdrive_list(&self.path, 100, &ctx.token).await?;
let bytes = match resp.status() {
StatusCode::OK => resp.into_body().bytes().await?,
_ => return Err(parse_error(resp).await?),
};
- if bytes.is_empty() {
- return Ok(None);
- }
-
let decoded_response =
serde_json::from_slice::<GdriveFileList>(&bytes).map_err(new_json_deserialize_error)?;
if let Some(next_page_token) = decoded_response.next_page_token {
- self.next_page_token = Some(next_page_token);
+ ctx.token = next_page_token;
} else {
- self.done = true;
+ ctx.done = true;
}
- let entries: Vec<oio::Entry> = decoded_response
- .files
- .into_iter()
- .map(|mut file| {
- let file_type = if file.mime_type.as_str() == "application/vnd.google-apps.folder" {
- file.name = format!("{}/", file.name);
- EntryMode::DIR
- } else {
- EntryMode::FILE
- };
+ for mut file in decoded_response.files {
+ let file_type = if file.mime_type.as_str() == "application/vnd.google-apps.folder" {
+ file.name = format!("{}/", file.name);
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
- let root = &self.core.root;
- let path = format!("{}{}", build_rooted_abs_path(root, &self.path), file.name);
- let normalized_path = build_rel_path(root, &path);
+ let root = &self.core.root;
+ let path = format!("{}{}", build_rooted_abs_path(root, &self.path), file.name);
+ let normalized_path = build_rel_path(root, &path);
- oio::Entry::new(&normalized_path, Metadata::new(file_type))
- })
- .collect();
+ let entry = oio::Entry::new(&normalized_path, Metadata::new(file_type));
+
+ ctx.entries.push_back(entry);
+ }
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/hdfs/lister.rs b/core/src/services/hdfs/lister.rs
index 5e31c1e14..ea8a57572 100644
--- a/core/src/services/hdfs/lister.rs
+++ b/core/src/services/hdfs/lister.rs
@@ -16,6 +16,7 @@
// under the License.
use async_trait::async_trait;
+use std::task::{Context, Poll};
use crate::raw::*;
use crate::EntryMode;
@@ -42,33 +43,27 @@ impl HdfsLister {
#[async_trait]
impl oio::List for HdfsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let mut oes: Vec<oio::Entry> = Vec::with_capacity(self.size);
-
- for _ in 0..self.size {
- let de = match self.rd.next() {
- Some(de) => de,
- None => break,
- };
-
- let path = build_rel_path(&self.root, de.path());
-
- let d = if de.is_file() {
- let meta = Metadata::new(EntryMode::FILE)
- .with_content_length(de.len())
- .with_last_modified(de.modified().into());
- oio::Entry::new(&path, meta)
- } else if de.is_dir() {
- // Make sure we are returning the correct path.
- oio::Entry::new(&format!("{path}/"), Metadata::new(EntryMode::DIR))
- } else {
- oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
- };
-
- oes.push(d)
- }
-
- Ok(if oes.is_empty() { None } else { Some(oes) })
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ let de = match self.rd.next() {
+ Some(de) => de,
+ None => return Poll::Ready(Ok(None)),
+ };
+
+ let path = build_rel_path(&self.root, de.path());
+
+ let entry = if de.is_file() {
+ let meta = Metadata::new(EntryMode::FILE)
+ .with_content_length(de.len())
+ .with_last_modified(de.modified().into());
+ oio::Entry::new(&path, meta)
+ } else if de.is_dir() {
+ // Make sure we are returning the correct path.
+ oio::Entry::new(&format!("{path}/"), Metadata::new(EntryMode::DIR))
+ } else {
+ oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
+ };
+
+ Poll::Ready(Ok(Some(entry)))
}
}
diff --git a/core/src/services/ipfs/backend.rs b/core/src/services/ipfs/backend.rs
index 2c1681dbe..915e2b798 100644
--- a/core/src/services/ipfs/backend.rs
+++ b/core/src/services/ipfs/backend.rs
@@ -165,7 +165,7 @@ impl Accessor for IpfsBackend {
type BlockingReader = ();
type Writer = ();
type BlockingWriter = ();
- type Lister = DirStream;
+ type Lister = oio::PageLister<DirStream>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -351,10 +351,8 @@ impl Accessor for IpfsBackend {
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- DirStream::new(Arc::new(self.clone()), path),
- ))
+ let l = DirStream::new(Arc::new(self.clone()), path);
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
@@ -415,7 +413,6 @@ impl IpfsBackend {
pub struct DirStream {
backend: Arc<IpfsBackend>,
path: String,
- consumed: bool,
}
impl DirStream {
@@ -423,18 +420,13 @@ impl DirStream {
Self {
backend,
path: path.to_string(),
- consumed: false,
}
}
}
#[async_trait]
-impl oio::List for DirStream {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.consumed {
- return Ok(None);
- }
-
+impl oio::PageList for DirStream {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.backend.ipfs_list(&self.path).await?;
if resp.status() != StatusCode::OK {
@@ -452,8 +444,6 @@ impl oio::List for DirStream {
.map(|v| v.name.unwrap())
.collect::<Vec<String>>();
- let mut oes = Vec::with_capacity(names.len());
-
for mut name in names {
let meta = self
.backend
@@ -465,10 +455,10 @@ impl oio::List for DirStream {
name += "/";
}
- oes.push(oio::Entry::new(&name, meta))
+ ctx.entries.push_back(oio::Entry::new(&name, meta))
}
- self.consumed = true;
- Ok(Some(oes))
+ ctx.done = true;
+ Ok(())
}
}
diff --git a/core/src/services/ipmfs/backend.rs b/core/src/services/ipmfs/backend.rs
index 26c06d569..7a9da9b4e 100644
--- a/core/src/services/ipmfs/backend.rs
+++ b/core/src/services/ipmfs/backend.rs
@@ -66,7 +66,7 @@ impl Accessor for IpmfsBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<IpmfsWriter>;
type BlockingWriter = ();
- type Lister = IpmfsLister;
+ type Lister = oio::PageLister<IpmfsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -171,10 +171,8 @@ impl Accessor for IpmfsBackend {
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- IpmfsLister::new(Arc::new(self.clone()), &self.root, path),
- ))
+ let l = IpmfsLister::new(Arc::new(self.clone()), &self.root, path);
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/ipmfs/lister.rs b/core/src/services/ipmfs/lister.rs
index 535c982b1..902d32592 100644
--- a/core/src/services/ipmfs/lister.rs
+++ b/core/src/services/ipmfs/lister.rs
@@ -47,12 +47,8 @@ impl IpmfsLister {
}
#[async_trait]
-impl oio::List for IpmfsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.consumed {
- return Ok(None);
- }
-
+impl oio::PageList for IpmfsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self.backend.ipmfs_ls(&self.path).await?;
if resp.status() != StatusCode::OK {
@@ -64,33 +60,24 @@ impl oio::List for IpmfsLister {
serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
// Mark dir stream has been consumed.
- self.consumed = true;
-
- Ok(Some(
- entries_body
- .entries
- .unwrap_or_default()
- .into_iter()
- .map(|object| {
- let path = match object.mode() {
- EntryMode::FILE => {
- format!("{}{}", &self.path, object.name)
- }
- EntryMode::DIR => {
- format!("{}{}/", &self.path, object.name)
- }
- EntryMode::Unknown => unreachable!(),
- };
-
- let path = build_rel_path(&self.root, &path);
-
- oio::Entry::new(
- &path,
- Metadata::new(object.mode()).with_content_length(object.size),
- )
- })
- .collect(),
- ))
+ ctx.done = true;
+
+ for object in entries_body.entries.unwrap_or_default() {
+ let path = match object.mode() {
+ EntryMode::FILE => format!("{}{}", &self.path, object.name),
+ EntryMode::DIR => format!("{}{}/", &self.path, object.name),
+ EntryMode::Unknown => unreachable!(),
+ };
+
+ let path = build_rel_path(&self.root, &path);
+
+ ctx.entries.push_back(oio::Entry::new(
+ &path,
+ Metadata::new(object.mode()).with_content_length(object.size),
+ ));
+ }
+
+ Ok(())
}
}
diff --git a/core/src/services/obs/backend.rs b/core/src/services/obs/backend.rs
index 6cfe43e4f..4f84852b5 100644
--- a/core/src/services/obs/backend.rs
+++ b/core/src/services/obs/backend.rs
@@ -252,7 +252,7 @@ impl Accessor for ObsBackend {
type BlockingReader = ();
type Writer = ObsWriters;
type BlockingWriter = ();
- type Lister = ObsLister;
+ type Lister = oio::PageLister<ObsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -428,9 +428,7 @@ impl Accessor for ObsBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- ObsLister::new(self.core.clone(), path, args.recursive(), args.limit()),
- ))
+ let l = ObsLister::new(self.core.clone(), path, args.recursive(), args.limit());
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/obs/lister.rs b/core/src/services/obs/lister.rs
index c96f51e3e..831427ab1 100644
--- a/core/src/services/obs/lister.rs
+++ b/core/src/services/obs/lister.rs
@@ -36,9 +36,6 @@ pub struct ObsLister {
path: String,
delimiter: &'static str,
limit: Option<usize>,
-
- next_marker: String,
- done: bool,
}
impl ObsLister {
@@ -50,23 +47,16 @@ impl ObsLister {
path: path.to_string(),
delimiter,
limit,
-
- next_marker: "".to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for ObsLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for ObsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
- .obs_list_objects(&self.path, &self.next_marker, self.delimiter, self.limit)
+ .obs_list_objects(&self.path, &ctx.token, self.delimiter, self.limit)
.await?;
if resp.status() != http::StatusCode::OK {
@@ -81,14 +71,13 @@ impl oio::List for ObsLister {
// Try our best to check whether this list is done.
//
// - Check `next_marker`
- self.done = match output.next_marker.as_ref() {
+ ctx.done = match output.next_marker.as_ref() {
None => true,
Some(next_marker) => next_marker.is_empty(),
};
- self.next_marker = output.next_marker.clone().unwrap_or_default();
+ ctx.token = output.next_marker.clone().unwrap_or_default();
let common_prefixes = output.common_prefixes;
- let mut entries = Vec::with_capacity(common_prefixes.len() + output.contents.len());
for prefix in common_prefixes {
let de = oio::Entry::new(
@@ -96,7 +85,7 @@ impl oio::List for ObsLister {
Metadata::new(EntryMode::DIR),
);
- entries.push(de);
+ ctx.entries.push_back(de);
}
for object in output.contents {
@@ -108,10 +97,10 @@ impl oio::List for ObsLister {
let de = oio::Entry::new(&build_rel_path(&self.core.root, &object.key), meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/onedrive/backend.rs b/core/src/services/onedrive/backend.rs
index bef175735..ba4e5b8d3 100644
--- a/core/src/services/onedrive/backend.rs
+++ b/core/src/services/onedrive/backend.rs
@@ -66,7 +66,7 @@ impl Accessor for OnedriveBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<OneDriveWriter>;
type BlockingWriter = ();
- type Lister = OnedriveLister;
+ type Lister = oio::PageLister<OnedriveLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -163,10 +163,9 @@ impl Accessor for OnedriveBackend {
}
async fn list(&self, path: &str, _op_list: OpList) -> Result<(RpList, Self::Lister)> {
- let lister: OnedriveLister =
- OnedriveLister::new(self.root.clone(), path.into(), self.clone());
+ let l = OnedriveLister::new(self.root.clone(), path.into(), self.clone());
- Ok((RpList::default(), lister))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
diff --git a/core/src/services/onedrive/lister.rs b/core/src/services/onedrive/lister.rs
index 90daeb866..12c557867 100644
--- a/core/src/services/onedrive/lister.rs
+++ b/core/src/services/onedrive/lister.rs
@@ -16,28 +16,19 @@
// under the License.
use async_trait::async_trait;
-use http::Response;
use super::backend::OnedriveBackend;
use super::error::parse_error;
use super::graph_model::GraphApiOnedriveListResponse;
use super::graph_model::ItemType;
-use crate::raw::build_rel_path;
-use crate::raw::build_rooted_abs_path;
-use crate::raw::new_json_deserialize_error;
-use crate::raw::oio::{self};
-use crate::raw::percent_encode_path;
-use crate::raw::IncomingAsyncBody;
-use crate::EntryMode;
-use crate::Metadata;
-use crate::Result;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
pub struct OnedriveLister {
root: String,
path: String,
backend: OnedriveBackend,
- next_link: Option<String>,
- done: bool,
}
impl OnedriveLister {
@@ -48,92 +39,80 @@ impl OnedriveLister {
root,
path,
backend,
- next_link: None,
- done: false,
}
}
}
#[async_trait]
-impl oio::List for OnedriveLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
- let response = self.onedrive_get().await?;
+impl oio::PageList for OnedriveLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ let request_url = if ctx.token.is_empty() {
+ let path = build_rooted_abs_path(&self.root, &self.path);
+ let url: String = if path == "." || path == "/" {
+ "https://graph.microsoft.com/v1.0/me/drive/root/children".to_string()
+ } else {
+ // According to OneDrive API examples, the path should not end with a slash.
+ // Reference: <https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_list_children?view=odsp-graph-online>
+ let path = path.strip_suffix('/').unwrap_or("");
+ format!(
+ "https://graph.microsoft.com/v1.0/me/drive/root:{}:/children",
+ percent_encode_path(path),
+ )
+ };
+ url
+ } else {
+ ctx.token.clone()
+ };
+
+ let resp = self
+ .backend
+ .onedrive_get_next_list_page(&request_url)
+ .await?;
- let status_code = response.status();
+ let status_code = resp.status();
if !status_code.is_success() {
if status_code == http::StatusCode::NOT_FOUND {
- return Ok(None);
+ ctx.done = true;
+ return Ok(());
}
- let error = parse_error(response).await?;
+ let error = parse_error(resp).await?;
return Err(error);
}
- let bytes = response.into_body().bytes().await?;
+ let bytes = resp.into_body().bytes().await?;
let decoded_response = serde_json::from_slice::<GraphApiOnedriveListResponse>(&bytes)
.map_err(new_json_deserialize_error)?;
if let Some(next_link) = decoded_response.next_link {
- self.next_link = Some(next_link);
+ ctx.token = next_link;
} else {
- self.done = true;
+ ctx.done = true;
}
- let entries: Vec<oio::Entry> = decoded_response
- .value
- .into_iter()
- .map(|drive_item| {
- let name = drive_item.name;
- let parent_path = drive_item.parent_reference.path;
- let parent_path = parent_path
- .strip_prefix(Self::DRIVE_ROOT_PREFIX)
- .unwrap_or("");
-
- let path = format!("{}/{}", parent_path, name);
-
- let normalized_path = build_rel_path(&self.root, &path);
-
- let entry: oio::Entry = match drive_item.item_type {
- ItemType::Folder { .. } => {
- let normalized_path = format!("{}/", normalized_path);
- oio::Entry::new(&normalized_path, Metadata::new(EntryMode::DIR))
- }
- ItemType::File { .. } => {
- oio::Entry::new(&normalized_path, Metadata::new(EntryMode::FILE))
- }
- };
- entry
- })
- .collect();
-
- Ok(Some(entries))
- }
-}
-
-impl OnedriveLister {
- async fn onedrive_get(&mut self) -> Result<Response<IncomingAsyncBody>> {
- let request_url = if let Some(next_link) = &self.next_link {
- let next_link_clone = next_link.clone();
- self.next_link = None;
- next_link_clone
- } else {
- let path = build_rooted_abs_path(&self.root, &self.path);
- let url: String = if path == "." || path == "/" {
- "https://graph.microsoft.com/v1.0/me/drive/root/children".to_string()
- } else {
- // According to OneDrive API examples, the path should not end with a slash.
- // Reference: <https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_list_children?view=odsp-graph-online>
- let path = path.strip_suffix('/').unwrap_or("");
- format!(
- "https://graph.microsoft.com/v1.0/me/drive/root:{}:/children",
- percent_encode_path(path),
- )
+ for drive_item in decoded_response.value {
+ let name = drive_item.name;
+ let parent_path = drive_item.parent_reference.path;
+ let parent_path = parent_path
+ .strip_prefix(Self::DRIVE_ROOT_PREFIX)
+ .unwrap_or("");
+
+ let path = format!("{}/{}", parent_path, name);
+
+ let normalized_path = build_rel_path(&self.root, &path);
+
+ let entry: oio::Entry = match drive_item.item_type {
+ ItemType::Folder { .. } => {
+ let normalized_path = format!("{}/", normalized_path);
+ oio::Entry::new(&normalized_path, Metadata::new(EntryMode::DIR))
+ }
+ ItemType::File { .. } => {
+ oio::Entry::new(&normalized_path, Metadata::new(EntryMode::FILE))
+ }
};
- url
- };
- self.backend.onedrive_get_next_list_page(&request_url).await
+ ctx.entries.push_back(entry)
+ }
+
+ Ok(())
}
}
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index c5fc3441c..a9e512429 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -380,7 +380,7 @@ impl Accessor for OssBackend {
type BlockingReader = ();
type Writer = OssWriters;
type BlockingWriter = ();
- type Lister = OssLister;
+ type Lister = oio::PageLister<OssLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -545,16 +545,14 @@ impl Accessor for OssBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- OssLister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- ),
- ))
+ let l = OssLister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ );
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
diff --git a/core/src/services/oss/core.rs b/core/src/services/oss/core.rs
index f0feb59d2..c90439de1 100644
--- a/core/src/services/oss/core.rs
+++ b/core/src/services/oss/core.rs
@@ -324,7 +324,7 @@ impl OssCore {
pub fn oss_list_object_request(
&self,
path: &str,
- token: Option<&str>,
+ token: &str,
delimiter: &str,
limit: Option<usize>,
start_after: Option<String>,
@@ -347,13 +347,9 @@ impl OssCore {
}
// continuation_token
- if let Some(continuation_token) = token {
- write!(
- url,
- "&continuation-token={}",
- percent_encode_path(continuation_token)
- )
- .expect("write into string must succeed");
+ if !token.is_empty() {
+ write!(url, "&continuation-token={}", percent_encode_path(token))
+ .expect("write into string must succeed");
}
// start-after
@@ -446,7 +442,7 @@ impl OssCore {
pub async fn oss_list_object(
&self,
path: &str,
- token: Option<&str>,
+ token: &str,
delimiter: &str,
limit: Option<usize>,
start_after: Option<String>,
diff --git a/core/src/services/oss/lister.rs b/core/src/services/oss/lister.rs
index 7b4b0bce3..6412cf8bf 100644
--- a/core/src/services/oss/lister.rs
+++ b/core/src/services/oss/lister.rs
@@ -41,9 +41,6 @@ pub struct OssLister {
/// Filter results to objects whose names are lexicographically
/// **equal to or after** startOffset
start_after: Option<String>,
-
- token: Option<String>,
- done: bool,
}
impl OssLister {
@@ -61,28 +58,25 @@ impl OssLister {
delimiter,
limit,
start_after: start_after.map(String::from),
- token: None,
-
- done: false,
}
}
}
#[async_trait]
-impl oio::List for OssLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for OssLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
.oss_list_object(
&self.path,
- self.token.as_deref(),
+ &ctx.token,
self.delimiter,
self.limit,
- self.start_after.clone(),
+ if ctx.token.is_empty() {
+ self.start_after.clone()
+ } else {
+ None
+ },
)
.await?;
@@ -95,17 +89,15 @@ impl oio::List for OssLister {
let output: ListBucketOutput = de::from_reader(bs.reader())
.map_err(|e| Error::new(ErrorKind::Unexpected, "deserialize xml").set_source(e))?;
- self.done = !output.is_truncated;
- self.token = output.next_continuation_token.clone();
-
- let mut entries = Vec::with_capacity(output.common_prefixes.len() + output.contents.len());
+ ctx.done = !output.is_truncated;
+ ctx.token = output.next_continuation_token.unwrap_or_default();
for prefix in output.common_prefixes {
let de = oio::Entry::new(
&build_rel_path(&self.core.root, &prefix.prefix),
Metadata::new(EntryMode::DIR),
);
- entries.push(de);
+ ctx.entries.push_back(de);
}
for object in output.contents {
@@ -129,10 +121,10 @@ impl oio::List for OssLister {
let path = unescape(&rel)
.map_err(|e| Error::new(ErrorKind::Unexpected, "excapse xml").set_source(e))?;
let de = oio::Entry::new(&path, meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 7e1ccfc85..56663f522 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -969,7 +969,7 @@ impl Accessor for S3Backend {
type BlockingReader = ();
type Writer = S3Writers;
type BlockingWriter = ();
- type Lister = S3Lister;
+ type Lister = oio::PageLister<S3Lister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -1134,16 +1134,14 @@ impl Accessor for S3Backend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- S3Lister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- ),
- ))
+ let l = S3Lister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ );
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
diff --git a/core/src/services/s3/lister.rs b/core/src/services/s3/lister.rs
index 431abd8f1..78d5e1ff3 100644
--- a/core/src/services/s3/lister.rs
+++ b/core/src/services/s3/lister.rs
@@ -38,9 +38,6 @@ pub struct S3Lister {
/// Amazon S3 starts listing **after** this specified key
start_after: Option<String>,
-
- token: String,
- done: bool,
}
impl S3Lister {
@@ -59,28 +56,26 @@ impl S3Lister {
delimiter,
limit,
start_after: start_after.map(String::from),
-
- token: "".to_string(),
- done: false,
}
}
}
#[async_trait]
-impl oio::List for S3Lister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for S3Lister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
.core
.s3_list_objects(
&self.path,
- &self.token,
+ &ctx.token,
self.delimiter,
self.limit,
- self.start_after.clone(),
+ // State after should only be set for the first page.
+ if ctx.token.is_empty() {
+ self.start_after.clone()
+ } else {
+ None
+ },
)
.await?;
@@ -97,16 +92,14 @@ impl oio::List for S3Lister {
// - Check `is_truncated`
// - Check `next_continuation_token`
// - Check the length of `common_prefixes` and `contents` (very rarely case)
- self.done = if let Some(is_truncated) = output.is_truncated {
+ ctx.done = if let Some(is_truncated) = output.is_truncated {
!is_truncated
} else if let Some(next_continuation_token) = output.next_continuation_token.as_ref() {
next_continuation_token.is_empty()
} else {
output.common_prefixes.is_empty() && output.contents.is_empty()
};
- self.token = output.next_continuation_token.clone().unwrap_or_default();
-
- let mut entries = Vec::with_capacity(output.common_prefixes.len() + output.contents.len());
+ ctx.token = output.next_continuation_token.clone().unwrap_or_default();
for prefix in output.common_prefixes {
let de = oio::Entry::new(
@@ -114,7 +107,7 @@ impl oio::List for S3Lister {
Metadata::new(EntryMode::DIR),
);
- entries.push(de);
+ ctx.entries.push_back(de);
}
for object in output.contents {
@@ -139,10 +132,10 @@ impl oio::List for S3Lister {
let de = oio::Entry::new(&build_rel_path(&self.core.root, &object.key), meta);
- entries.push(de);
+ ctx.entries.push_back(de);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/sftp/lister.rs b/core/src/services/sftp/lister.rs
index b130629ca..e69f92213 100644
--- a/core/src/services/sftp/lister.rs
+++ b/core/src/services/sftp/lister.rs
@@ -16,6 +16,7 @@
// under the License.
use std::pin::Pin;
+use std::task::{ready, Context, Poll};
use async_trait::async_trait;
use futures::StreamExt;
@@ -47,24 +48,18 @@ impl SftpLister {
#[async_trait]
impl oio::List for SftpLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.limit == 0 {
- return Ok(None);
- }
-
- let item = self.dir.next().await;
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ let item = ready!(self.dir.poll_next_unpin(cx)).transpose()?;
match item {
- Some(Ok(e)) => {
+ Some(e) => {
if e.filename().to_str() == Some(".") || e.filename().to_str() == Some("..") {
- self.next().await
+ self.poll_next(cx)
} else {
- self.limit -= 1;
- Ok(Some(vec![map_entry(self.prefix.as_str(), e.clone())]))
+ Poll::Ready(Ok(Some(map_entry(self.prefix.as_str(), e))))
}
}
- Some(Err(e)) => Err(e.into()),
- None => Ok(None),
+ None => Poll::Ready(Ok(None)),
}
}
}
diff --git a/core/src/services/swift/backend.rs b/core/src/services/swift/backend.rs
index 6f6b22dba..9fa58b7dd 100644
--- a/core/src/services/swift/backend.rs
+++ b/core/src/services/swift/backend.rs
@@ -218,7 +218,7 @@ impl Accessor for SwiftBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<SwiftWriter>;
type BlockingWriter = ();
- type Lister = SwiftLister;
+ type Lister = oio::PageLister<SwiftLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -329,8 +329,8 @@ impl Accessor for SwiftBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = SwiftLister::new(self.core.clone(), path.to_string(), args.recursive());
+ let l = SwiftLister::new(self.core.clone(), path.to_string(), args.recursive());
- Ok((RpList::default(), op))
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/swift/lister.rs b/core/src/services/swift/lister.rs
index d69bb42b5..7fdfa0d5b 100644
--- a/core/src/services/swift/lister.rs
+++ b/core/src/services/swift/lister.rs
@@ -29,7 +29,6 @@ pub struct SwiftLister {
core: Arc<SwiftCore>,
path: String,
delimiter: &'static str,
- done: bool,
}
impl SwiftLister {
@@ -39,18 +38,13 @@ impl SwiftLister {
core,
path,
delimiter,
- done: false,
}
}
}
#[async_trait]
-impl oio::List for SwiftLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
-
+impl oio::PageList for SwiftLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let response = self.core.swift_list(&self.path, self.delimiter).await?;
let status_code = response.status();
@@ -60,15 +54,13 @@ impl oio::List for SwiftLister {
return Err(error);
}
+ ctx.done = true;
+
let bytes = response.into_body().bytes().await?;
let mut decoded_response = serde_json::from_slice::<Vec<ListOpResponse>>(&bytes)
.map_err(new_json_deserialize_error)?;
- self.done = true;
-
- let mut entries = Vec::with_capacity(decoded_response.len());
-
- while let Some(status) = decoded_response.pop() {
+ for status in decoded_response {
let entry: oio::Entry = match status {
ListOpResponse::Subdir { subdir } => {
let meta = Metadata::new(EntryMode::DIR);
@@ -99,9 +91,9 @@ impl oio::List for SwiftLister {
oio::Entry::new(&name, meta)
}
};
- entries.push(entry);
+ ctx.entries.push_back(entry);
}
- Ok(Some(entries))
+ Ok(())
}
}
diff --git a/core/src/services/webdav/backend.rs b/core/src/services/webdav/backend.rs
index 09e0f839a..aa676c44c 100644
--- a/core/src/services/webdav/backend.rs
+++ b/core/src/services/webdav/backend.rs
@@ -223,7 +223,7 @@ impl Accessor for WebdavBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<WebdavWriter>;
type BlockingWriter = ();
- type Lister = Option<WebdavLister>;
+ type Lister = Option<oio::PageLister<WebdavLister>>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -398,10 +398,9 @@ impl Accessor for WebdavBackend {
let result: Multistatus =
quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
- Ok((
- RpList::default(),
- Some(WebdavLister::new(&self.base_dir, &self.root, path, result)),
- ))
+ let l = WebdavLister::new(&self.base_dir, &self.root, path, result);
+
+ Ok((RpList::default(), Some(oio::PageLister::new(l))))
}
StatusCode::NOT_FOUND if path.ends_with('/') => Ok((RpList::default(), None)),
_ => Err(parse_error(resp).await?),
diff --git a/core/src/services/webdav/lister.rs b/core/src/services/webdav/lister.rs
index 84e97aca6..e98c77db3 100644
--- a/core/src/services/webdav/lister.rs
+++ b/core/src/services/webdav/lister.rs
@@ -15,8 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::mem;
-
use async_trait::async_trait;
use serde::Deserialize;
@@ -30,6 +28,7 @@ pub struct WebdavLister {
}
impl WebdavLister {
+ /// TODO: sending request in `next_page` instead of in `new`.
pub fn new(base_dir: &str, root: &str, path: &str, multistates: Multistatus) -> Self {
Self {
base_dir: base_dir.to_string(),
@@ -41,14 +40,10 @@ impl WebdavLister {
}
#[async_trait]
-impl oio::List for WebdavLister {
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.multistates.response.is_empty() {
- return Ok(None);
- };
- let oes = mem::take(&mut self.multistates.response);
-
- let mut entries = Vec::with_capacity(oes.len());
+impl oio::PageList for WebdavLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ // Build request instead of clone here.
+ let oes = self.multistates.response.clone();
for res in oes {
let path = res
@@ -70,19 +65,20 @@ impl oio::List for WebdavLister {
}
let meta = res.parse_into_metadata()?;
- entries.push(oio::Entry::new(&decoded_path, meta))
+ ctx.entries.push_back(oio::Entry::new(&decoded_path, meta))
}
+ ctx.done = true;
- Ok(Some(entries))
+ Ok(())
}
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
pub struct Multistatus {
pub response: Vec<ListOpResponse>,
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
pub struct ListOpResponse {
pub href: String,
pub propstat: Propstat,
@@ -141,13 +137,13 @@ impl ListOpResponse {
}
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
pub struct Propstat {
pub prop: Prop,
pub status: String,
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
pub struct Prop {
#[serde(default)]
pub displayname: String,
@@ -158,13 +154,13 @@ pub struct Prop {
pub resourcetype: ResourceTypeContainer,
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
pub struct ResourceTypeContainer {
#[serde(rename = "$value")]
pub value: Option<ResourceType>,
}
-#[derive(Deserialize, Debug, PartialEq, Eq)]
+#[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
#[serde(rename_all = "lowercase")]
pub enum ResourceType {
Collection,
diff --git a/core/src/services/webhdfs/backend.rs b/core/src/services/webhdfs/backend.rs
index 3352e47d0..9fa9bd150 100644
--- a/core/src/services/webhdfs/backend.rs
+++ b/core/src/services/webhdfs/backend.rs
@@ -31,10 +31,8 @@ use super::error::parse_error;
use super::error::parse_error_msg;
use super::lister::WebhdfsLister;
use super::message::BooleanResp;
-use super::message::DirectoryListingWrapper;
use super::message::FileStatusType;
use super::message::FileStatusWrapper;
-use super::message::FileStatusesWrapper;
use super::writer::WebhdfsWriter;
use crate::raw::*;
use crate::*;
@@ -277,7 +275,10 @@ impl WebhdfsBackend {
Ok(req)
}
- fn webhdfs_list_status_request(&self, path: &str) -> Result<Request<AsyncBody>> {
+ pub async fn webhdfs_list_status_request(
+ &self,
+ path: &str,
+ ) -> Result<Response<IncomingAsyncBody>> {
let p = build_abs_path(&self.root, path);
let mut url = format!(
"{}/webhdfs/v1/{}?op=LISTSTATUS",
@@ -291,29 +292,24 @@ impl WebhdfsBackend {
let req = Request::get(&url)
.body(AsyncBody::Empty)
.map_err(new_request_build_error)?;
- Ok(req)
+ self.client.send(req).await
}
- pub(super) fn webhdfs_list_status_batch_request(
+ pub async fn webhdfs_list_status_batch_request(
&self,
path: &str,
- args: &OpList,
- ) -> Result<Request<AsyncBody>> {
+ start_after: &str,
+ ) -> Result<Response<IncomingAsyncBody>> {
let p = build_abs_path(&self.root, path);
- // if it's not the first time to call LISTSTATUS_BATCH, we will add &startAfter=<CHILD>
- let start_after_param = match args.start_after() {
- Some(sa) if sa.is_empty() => String::new(),
- Some(sa) => format!("&startAfter={}", sa),
- None => String::new(),
- };
-
let mut url = format!(
- "{}/webhdfs/v1/{}?op=LISTSTATUS_BATCH{}",
+ "{}/webhdfs/v1/{}?op=LISTSTATUS_BATCH",
self.endpoint,
percent_encode_path(&p),
- start_after_param
);
+ if !start_after.is_empty() {
+ url += format!("&startAfter={}", start_after).as_str();
+ }
if let Some(auth) = &self.auth {
url += format!("&{auth}").as_str();
}
@@ -321,7 +317,7 @@ impl WebhdfsBackend {
let req = Request::get(&url)
.body(AsyncBody::Empty)
.map_err(new_request_build_error)?;
- Ok(req)
+ self.client.send(req).await
}
async fn webhdfs_read_file(
@@ -402,7 +398,7 @@ impl Accessor for WebhdfsBackend {
type BlockingReader = ();
type Writer = oio::OneShotWriter<WebhdfsWriter>;
type BlockingWriter = ();
- type Lister = WebhdfsLister;
+ type Lister = oio::PageLister<WebhdfsLister>;
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -549,46 +545,7 @@ impl Accessor for WebhdfsBackend {
}
let path = path.trim_end_matches('/');
-
- if !self.disable_list_batch {
- let req = self.webhdfs_list_status_batch_request(path, &OpList::default())?;
- let resp = self.client.send(req).await?;
- match resp.status() {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
- let directory_listing = serde_json::from_slice::<DirectoryListingWrapper>(&bs)
- .map_err(new_json_deserialize_error)?
- .directory_listing;
- let file_statuses = directory_listing.partial_listing.file_statuses.file_status;
- let mut objects = WebhdfsLister::new(self.clone(), path, file_statuses);
- objects.set_remaining_entries(directory_listing.remaining_entries);
- Ok((RpList::default(), objects))
- }
- StatusCode::NOT_FOUND => {
- let objects = WebhdfsLister::new(self.clone(), path, vec![]);
- Ok((RpList::default(), objects))
- }
- _ => Err(parse_error(resp).await?),
- }
- } else {
- let req = self.webhdfs_list_status_request(path)?;
- let resp = self.client.send(req).await?;
- match resp.status() {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
- let file_statuses = serde_json::from_slice::<FileStatusesWrapper>(&bs)
- .map_err(new_json_deserialize_error)?
- .file_statuses
- .file_status;
- let objects = WebhdfsLister::new(self.clone(), path, file_statuses);
- Ok((RpList::default(), objects))
- }
- StatusCode::NOT_FOUND => {
- let objects = WebhdfsLister::new(self.clone(), path, vec![]);
- Ok((RpList::default(), objects))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
+ let l = WebhdfsLister::new(self.clone(), path);
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
}
diff --git a/core/src/services/webhdfs/lister.rs b/core/src/services/webhdfs/lister.rs
index 8c6579a5d..0857414fc 100644
--- a/core/src/services/webhdfs/lister.rs
+++ b/core/src/services/webhdfs/lister.rs
@@ -20,90 +20,75 @@ use http::StatusCode;
use super::backend::WebhdfsBackend;
use super::error::parse_error;
-use super::message::DirectoryListingWrapper;
-use super::message::FileStatus;
-use super::message::FileStatusType;
+use super::message::*;
use crate::raw::*;
use crate::*;
pub struct WebhdfsLister {
backend: WebhdfsBackend,
path: String,
- statuses: Vec<FileStatus>,
- batch_start_after: Option<String>,
- remaining_entries: u32,
}
impl WebhdfsLister {
- pub fn new(backend: WebhdfsBackend, path: &str, statuses: Vec<FileStatus>) -> Self {
+ pub fn new(backend: WebhdfsBackend, path: &str) -> Self {
Self {
backend,
path: path.to_string(),
- batch_start_after: statuses.last().map(|f| f.path_suffix.clone()),
- statuses,
- remaining_entries: 0,
}
}
-
- pub(super) fn set_remaining_entries(&mut self, remaining_entries: u32) {
- self.remaining_entries = remaining_entries;
- }
}
#[async_trait]
-impl oio::List for WebhdfsLister {
- /// Returns the next page of entries.
- ///
- /// Note: default list status with batch, calling next will query for next batch if `remaining_entries` > 0.
- async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.statuses.is_empty() && self.remaining_entries == 0 {
- return Ok(None);
- }
+impl oio::PageList for WebhdfsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ let file_status = if self.backend.disable_list_batch {
+ let resp = self.backend.webhdfs_list_status_request(&self.path).await?;
+ match resp.status() {
+ StatusCode::OK => {
+ ctx.done = true;
- return match self.backend.disable_list_batch {
- true => self.webhdfs_get_next_list_statuses(),
- false => {
- let args = OpList::with_start_after(
- OpList::default(),
- &self.batch_start_after.clone().unwrap(),
- );
- let req = self
- .backend
- .webhdfs_list_status_batch_request(&self.path, &args)?;
- let resp = self.backend.client.send(req).await?;
+ let bs = resp.into_body().bytes().await?;
+ serde_json::from_slice::<FileStatusesWrapper>(&bs)
+ .map_err(new_json_deserialize_error)?
+ .file_statuses
+ .file_status
+ }
+ StatusCode::NOT_FOUND => {
+ ctx.done = true;
+ return Ok(());
+ }
+ _ => return Err(parse_error(resp).await?),
+ }
+ } else {
+ let resp = self
+ .backend
+ .webhdfs_list_status_batch_request(&self.path, &ctx.token)
+ .await?;
+ match resp.status() {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+ let directory_listing = serde_json::from_slice::<DirectoryListingWrapper>(&bs)
+ .map_err(new_json_deserialize_error)?
+ .directory_listing;
+ let file_statuses = directory_listing.partial_listing.file_statuses.file_status;
- match resp.status() {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
- let directory_listing =
- serde_json::from_slice::<DirectoryListingWrapper>(&bs)
- .map_err(new_json_deserialize_error)?;
- let file_statuses = directory_listing
- .directory_listing
- .partial_listing
- .file_statuses
- .file_status;
- self.remaining_entries =
- directory_listing.directory_listing.remaining_entries;
- self.batch_start_after =
- file_statuses.last().map(|f| f.path_suffix.clone());
- self.statuses.extend(file_statuses);
- self.webhdfs_get_next_list_statuses()
+ if directory_listing.remaining_entries == 0 {
+ ctx.done = true;
+ } else if file_statuses.len() > 0 {
+ ctx.token = file_statuses.last().unwrap().path_suffix.clone();
}
- StatusCode::NOT_FOUND => self.webhdfs_get_next_list_statuses(),
- _ => Err(parse_error(resp).await?),
+
+ file_statuses
}
+ StatusCode::NOT_FOUND => {
+ ctx.done = true;
+ return Ok(());
+ }
+ _ => return Err(parse_error(resp).await?),
}
};
- }
-}
-impl WebhdfsLister {
- /// Returns the next page of entries.
- fn webhdfs_get_next_list_statuses(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let mut entries = Vec::with_capacity(self.statuses.len());
-
- while let Some(status) = self.statuses.pop() {
+ for status in file_status {
let mut path = if self.path.is_empty() {
status.path_suffix.to_string()
} else {
@@ -126,8 +111,9 @@ impl WebhdfsLister {
path += "/"
}
let entry = oio::Entry::new(&path, meta);
- entries.push(entry);
+ ctx.entries.push_back(entry);
}
- Ok(Some(entries))
+
+ Ok(())
}
}
diff --git a/core/src/types/list.rs b/core/src/types/list.rs
index ac01496b3..5493d58db 100644
--- a/core/src/types/list.rs
+++ b/core/src/types/list.rs
@@ -26,11 +26,10 @@ use futures::future::BoxFuture;
use futures::FutureExt;
use futures::Stream;
+use crate::raw::oio::List;
use crate::raw::*;
use crate::*;
-/// Future constructed by listing.
-type ListFuture = BoxFuture<'static, (oio::Lister, Result<Option<Vec<oio::Entry>>>)>;
/// Future constructed by stating.
type StatFuture = BoxFuture<'static, (String, Result<RpStat>)>;
@@ -42,12 +41,10 @@ type StatFuture = BoxFuture<'static, (String, Result<RpStat>)>;
/// User can use lister as `Stream<Item = Result<Entry>>`.
pub struct Lister {
acc: FusedAccessor,
+ lister: oio::Lister,
/// required_metakey is the metakey required by users.
required_metakey: FlagSet<Metakey>,
- buf: VecDeque<oio::Entry>,
- lister: Option<oio::Lister>,
- listing: Option<ListFuture>,
stating: Option<StatFuture>,
}
@@ -64,11 +61,9 @@ impl Lister {
Ok(Self {
acc,
+ lister,
required_metakey,
- buf: VecDeque::new(),
- lister: Some(lister),
- listing: None,
stating: None,
})
}
@@ -83,52 +78,39 @@ impl Stream for Lister {
// Make sure we will not poll this future again.
self.stating = None;
- let metadata = rp?.into_metadata();
+ // TODO: we should rebuild the future if stat failed.
+ let metadata = match rp {
+ Ok(rp) => rp.into_metadata(),
+ Err(err) => {
+ let acc = self.acc.clone();
+ let fut = async move {
+ let res = acc.stat(&path, OpStat::default()).await;
+
+ (path, res)
+ };
+ self.stating = Some(Box::pin(fut));
+ return Poll::Ready(Some(Err(err)));
+ }
+ };
return Poll::Ready(Some(Ok(Entry::new(path, metadata))));
}
- if let Some(oe) = self.buf.pop_front() {
- let (path, metadata) = oe.into_entry().into_parts();
- // TODO: we can optimize this by checking the provided metakey provided by services.
- if metadata.contains_metakey(self.required_metakey) {
- return Poll::Ready(Some(Ok(Entry::new(path, metadata))));
- }
-
- let acc = self.acc.clone();
- let fut = async move {
- let res = acc.stat(&path, OpStat::default()).await;
-
- (path, res)
- };
- self.stating = Some(Box::pin(fut));
- return self.poll_next(cx);
- }
-
- if let Some(fut) = self.listing.as_mut() {
- let (op, res) = ready!(fut.poll_unpin(cx));
-
- // Make sure we will not poll this future again.
- self.listing = None;
-
- return match res? {
- Some(oes) => {
- self.lister = Some(op);
- self.buf = oes.into();
- self.poll_next(cx)
+ match ready!(self.lister.poll_next(cx))? {
+ Some(oe) => {
+ let (path, metadata) = oe.into_entry().into_parts();
+ // TODO: we can optimize this by checking the provided metakey provided by services.
+ if metadata.contains_metakey(self.required_metakey) {
+ return Poll::Ready(Some(Ok(Entry::new(path, metadata))));
}
- None => Poll::Ready(None),
- };
- }
- match self.lister.take() {
- Some(mut lister) => {
+ let acc = self.acc.clone();
let fut = async move {
- let res = lister.next().await;
+ let res = acc.stat(&path, OpStat::default()).await;
- (lister, res)
+ (path, res)
};
- self.listing = Some(Box::pin(fut));
+ self.stating = Some(Box::pin(fut));
self.poll_next(cx)
}
None => Poll::Ready(None),