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:39 UTC
(incubator-opendal) 02/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 0849e07d249381ba2125c5daeb6647ae6a2f2c41
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Nov 15 15:41:37 2023 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/blocking.rs | 2 +-
core/src/layers/complete.rs | 2 +-
core/src/layers/concurrent_limit.rs | 2 +-
core/src/layers/error_context.rs | 2 +-
core/src/layers/immutable_index.rs | 47 ++++----
core/src/layers/logging.rs | 6 +-
core/src/layers/minitrace.rs | 2 +-
core/src/layers/oteltrace.rs | 2 +-
core/src/layers/retry.rs | 2 +-
core/src/layers/tracing.rs | 2 +-
core/src/raw/adapters/kv/backend.rs | 34 ++----
core/src/raw/adapters/typed_kv/backend.rs | 33 ++----
core/src/raw/oio/list/api.rs | 8 +-
core/src/raw/oio/list/into_flat_page.rs | 147 +++++++++++++-------------
core/src/raw/oio/list/into_hierarchy_pager.rs | 81 +++++++-------
core/src/services/fs/lister.rs | 68 ++++++------
core/src/services/hdfs/lister.rs | 42 ++++----
core/src/types/list.rs | 44 +++-----
18 files changed, 234 insertions(+), 292 deletions(-)
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index 3181062e7..3b167a491 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -314,7 +314,7 @@ impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
}
impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
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 ea4906cf6..cc7fdb61a 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -654,7 +654,7 @@ where
A: Accessor<BlockingLister = P>,
P: oio::BlockingList,
{
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
use CompleteLister::*;
match self {
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index d839b97d9..cd82aa650 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -316,7 +316,7 @@ impl<R: oio::List> oio::List for ConcurrentLimitWrapper<R> {
}
impl<R: oio::BlockingList> oio::BlockingList for ConcurrentLimitWrapper<R> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner.next()
}
}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index c3a0feef8..3baed1be5 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -463,7 +463,7 @@ impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
}
impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner.next().map_err(|err| {
err.with_operation(ListOperation::BlockingNext)
.with_context("service", self.scheme)
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index 9d47d78ff..764f5ecc8 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -17,8 +17,8 @@
use std::collections::HashSet;
use std::fmt::Debug;
-use std::mem;
use std::task::{Context, Poll};
+use std::vec::IntoIter;
use async_trait::async_trait;
@@ -208,48 +208,39 @@ impl<A: Accessor> LayeredAccessor for ImmutableIndexAccessor<A> {
}
pub struct ImmutableDir {
- idx: Vec<String>,
+ idx: IntoIter<String>,
}
impl ImmutableDir {
fn new(idx: Vec<String>) -> Self {
- Self { idx }
- }
-
- fn inner_next_page(&mut self) -> Option<Vec<oio::Entry>> {
- if self.idx.is_empty() {
- return None;
+ Self {
+ idx: idx.into_iter(),
}
+ }
- let vs = mem::take(&mut self.idx);
-
- Some(
- vs.into_iter()
- .map(|v| {
- let mode = if v.ends_with('/') {
- EntryMode::DIR
- } else {
- EntryMode::FILE
- };
- let meta = Metadata::new(mode);
- oio::Entry::with(v, meta)
- })
- .collect(),
- )
+ fn inner_next(&mut self) -> Option<oio::Entry> {
+ self.idx.next().map(|v| {
+ let mode = if v.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+ let meta = Metadata::new(mode);
+ oio::Entry::with(v, meta)
+ })
}
}
#[async_trait]
impl oio::List for ImmutableDir {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- todo!()
- // Ok(self.inner_next_page())
+ fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ Poll::Ready(Ok(self.inner_next()))
}
}
impl oio::BlockingList for ImmutableDir {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- Ok(self.inner_next_page())
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 16b91cef5..6d48dcbd0 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -1520,18 +1520,18 @@ impl<P: oio::List> oio::List for LoggingLister<P> {
}
impl<P: oio::BlockingList> oio::BlockingList for LoggingLister<P> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
let res = self.inner.next();
match &res {
Ok(Some(des)) => {
debug!(
target: LOGGING_TARGET,
- "service={} operation={} path={} -> got {} entries",
+ "service={} operation={} path={} -> listed entry: {}",
self.ctx.scheme,
self.op,
self.path,
- des.len(),
+ des.path(),
);
}
Ok(None) => {
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index b0f192ed3..0af413940 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -380,7 +380,7 @@ impl<R: oio::List> oio::List for MinitraceWrapper<R> {
}
impl<R: oio::BlockingList> oio::BlockingList for MinitraceWrapper<R> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
let _g = self.span.set_local_parent();
let _span = LocalSpan::enter_with_local_parent(ListOperation::BlockingNext.into_static());
self.inner.next()
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index c4d8b6f0e..bddb0b30f 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -337,7 +337,7 @@ impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
}
impl<R: oio::BlockingList> oio::BlockingList for OtelTraceWrapper<R> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner.next()
}
}
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 67528347d..bd47598a1 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -1099,7 +1099,7 @@ impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
}
impl<P: oio::BlockingList, I: RetryInterceptor> oio::BlockingList for RetryWrapper<P, I> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
{ || self.inner.next() }
.retry(&self.builder)
.when(|e| e.is_temporary())
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 38ae5f2ad..74cb89a55 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -373,7 +373,7 @@ impl<R: oio::List> oio::List for TracingWrapper<R> {
impl<R: oio::BlockingList> oio::BlockingList for TracingWrapper<R> {
#[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner.next()
}
}
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index b5ddf9440..856f65f1f 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -314,12 +314,9 @@ impl KvLister {
inner: inner.into_iter(),
}
}
-}
-#[async_trait]
-impl oio::List for KvLister {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- let entry = self.inner.next().map(|v| {
+ fn inner_next(&mut self) -> Option<oio::Entry> {
+ self.inner.next().map(|v| {
let mode = if v.ends_with('/') {
EntryMode::DIR
} else {
@@ -327,29 +324,20 @@ impl oio::List for KvLister {
};
oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
- });
+ })
+ }
+}
- Poll::Ready(Ok(entry))
+#[async_trait]
+impl oio::List for KvLister {
+ fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ Poll::Ready(Ok(self.inner_next()))
}
}
impl oio::BlockingList for KvLister {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- 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(),
- ))
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index e60e21649..89c5b2ea6 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -315,12 +315,9 @@ impl KvLister {
inner: inner.into_iter(),
}
}
-}
-#[async_trait]
-impl oio::List for KvLister {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
- let entry = self.inner.next().map(|v| {
+ fn inner_next(&mut self) -> Option<oio::Entry> {
+ self.inner.next().map(|v| {
let mode = if v.ends_with('/') {
EntryMode::DIR
} else {
@@ -328,28 +325,20 @@ impl oio::List for KvLister {
};
oio::Entry::new(&build_rel_path(&self.root, &v), Metadata::new(mode))
- });
+ })
+ }
+}
- Poll::Ready(Ok(entry))
+#[async_trait]
+impl oio::List for KvLister {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
+ Poll::Ready(Ok(self.inner_next()))
}
}
impl oio::BlockingList for KvLister {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- 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(),
- ))
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ Ok(self.inner_next())
}
}
diff --git a/core/src/raw/oio/list/api.rs b/core/src/raw/oio/list/api.rs
index b7899e60d..b90f995cf 100644
--- a/core/src/raw/oio/list/api.rs
+++ b/core/src/raw/oio/list/api.rs
@@ -101,26 +101,26 @@ pub trait BlockingList: Send + 'static {
///
/// `Ok(None)` means all pages have been returned. Any following call
/// to `next` will always get the same result.
- fn next(&mut self) -> Result<Option<Vec<Entry>>>;
+ fn next(&mut self) -> Result<Option<Entry>>;
}
/// BlockingLister is a boxed [`BlockingList`]
pub type BlockingLister = Box<dyn BlockingList>;
impl<P: BlockingList + ?Sized> BlockingList for Box<P> {
- fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn next(&mut self) -> Result<Option<Entry>> {
(**self).next()
}
}
impl BlockingList for () {
- fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn next(&mut self) -> Result<Option<Entry>> {
Ok(None)
}
}
impl<P: BlockingList> BlockingList for Option<P> {
- fn next(&mut self) -> Result<Option<Vec<Entry>>> {
+ fn next(&mut self) -> Result<Option<Entry>> {
match self {
Some(p) => p.next(),
None => 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 6c2ef6207..e7215fa45 100644
--- a/core/src/raw/oio/list/into_flat_page.rs
+++ b/core/src/raw/oio/list/into_flat_page.rs
@@ -16,7 +16,6 @@
// under the License.
use std::collections::VecDeque;
-use std::mem;
use std::task::{Context, Poll};
use async_trait::async_trait;
@@ -157,58 +156,59 @@ where
A: Accessor<BlockingLister = P>,
P: oio::BlockingList,
{
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- loop {
- if let Some(de) = self.dirs.pop_back() {
- let (_, op) = self.acc.blocking_list(de.path(), OpList::new())?;
- 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()? {
- 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 next(&mut self) -> Result<Option<oio::Entry>> {
+ todo!()
+ // loop {
+ // if let Some(de) = self.dirs.pop_back() {
+ // let (_, op) = self.acc.blocking_list(de.path(), OpList::new())?;
+ // 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()? {
+ // 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)));
+ // }
+ // }
}
}
@@ -273,25 +273,26 @@ mod tests {
}
impl BlockingList for MockLister {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
- self.done = true;
-
- let entries = self
- .inner
- .iter()
- .map(|path| {
- if path.ends_with('/') {
- oio::Entry::new(path, Metadata::new(EntryMode::DIR))
- } else {
- oio::Entry::new(path, Metadata::new(EntryMode::FILE))
- }
- })
- .collect();
-
- Ok(Some(entries))
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ todo!()
+ // if self.done {
+ // return Ok(None);
+ // }
+ // self.done = true;
+ //
+ // let entries = self
+ // .inner
+ // .iter()
+ // .map(|path| {
+ // if path.ends_with('/') {
+ // oio::Entry::new(path, Metadata::new(EntryMode::DIR))
+ // } else {
+ // oio::Entry::new(path, Metadata::new(EntryMode::FILE))
+ // }
+ // })
+ // .collect();
+ //
+ // Ok(Some(entries))
}
}
@@ -305,7 +306,7 @@ mod tests {
let mut entries = Vec::default();
while let Some(e) = lister.next()? {
- entries.extend_from_slice(&e)
+ entries.push(e)
}
assert_eq!(
diff --git a/core/src/raw/oio/list/into_hierarchy_pager.rs b/core/src/raw/oio/list/into_hierarchy_pager.rs
index 217371b45..852a33548 100644
--- a/core/src/raw/oio/list/into_hierarchy_pager.rs
+++ b/core/src/raw/oio/list/into_hierarchy_pager.rs
@@ -136,18 +136,19 @@ impl<P: oio::List> oio::List for HierarchyLister<P> {
}
impl<P: oio::BlockingList> oio::BlockingList for HierarchyLister<P> {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- let page = self.lister.next()?;
-
- let entries = if let Some(entries) = page {
- entries
- } else {
- return Ok(None);
- };
-
- let entries = self.filter_entries(entries);
-
- Ok(Some(entries))
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ todo!()
+ // let page = self.lister.next()?;
+ //
+ // let entries = if let Some(entries) = page {
+ // entries
+ // } else {
+ // return Ok(None);
+ // };
+ //
+ // let entries = self.filter_entries(entries);
+ //
+ // Ok(Some(entries))
}
}
@@ -176,25 +177,26 @@ mod tests {
}
impl BlockingList for MockLister {
- fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
- if self.done {
- return Ok(None);
- }
- self.done = true;
-
- let entries = self
- .inner
- .iter()
- .map(|path| {
- if path.ends_with('/') {
- oio::Entry::new(path, Metadata::new(EntryMode::DIR))
- } else {
- oio::Entry::new(path, Metadata::new(EntryMode::FILE))
- }
- })
- .collect();
-
- Ok(Some(entries))
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ todo!()
+ // if self.done {
+ // return Ok(None);
+ // }
+ // self.done = true;
+ //
+ // let entries = self
+ // .inner
+ // .iter()
+ // .map(|path| {
+ // if path.ends_with('/') {
+ // oio::Entry::new(path, Metadata::new(EntryMode::DIR))
+ // } else {
+ // oio::Entry::new(path, Metadata::new(EntryMode::FILE))
+ // }
+ // })
+ // .collect();
+ //
+ // Ok(Some(entries))
}
}
@@ -209,15 +211,14 @@ mod tests {
let mut set = HashSet::new();
while let Some(e) = lister.next()? {
- for i in &e {
- debug!("got path {}", i.path());
- assert!(
- set.insert(i.path().to_string()),
- "duplicated value: {}",
- i.path()
- );
- }
- entries.extend_from_slice(&e)
+ debug!("got path {}", e.path());
+ assert!(
+ set.insert(e.path().to_string()),
+ "duplicated value: {}",
+ e.path()
+ );
+
+ entries.push(e)
}
assert_eq!(
diff --git a/core/src/services/fs/lister.rs b/core/src/services/fs/lister.rs
index 13108b691..cce70f98c 100644
--- a/core/src/services/fs/lister.rs
+++ b/core/src/services/fs/lister.rs
@@ -112,42 +112,36 @@ impl oio::List for FsLister<tokio::fs::ReadDir> {
}
impl oio::BlockingList for FsLister<std::fs::ReadDir> {
- 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.map_err(new_std_io_error)?,
- None => break,
- };
-
- let entry_path = de.path();
- let rel_path = normalize_path(
- &entry_path
- .strip_prefix(&self.root)
- .expect("cannot fail because the prefix is iterated")
- .to_string_lossy()
- .replace('\\', "/"),
- );
-
- // 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().map_err(new_std_io_error)?;
-
- let d = if file_type.is_file() {
- oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
- } else if file_type.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)
- }
-
- Ok(if oes.is_empty() { None } else { Some(oes) })
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let de = match self.rd.next() {
+ Some(de) => de.map_err(new_std_io_error)?,
+ None => return Ok(None),
+ };
+
+ let entry_path = de.path();
+ let rel_path = normalize_path(
+ &entry_path
+ .strip_prefix(&self.root)
+ .expect("cannot fail because the prefix is iterated")
+ .to_string_lossy()
+ .replace('\\', "/"),
+ );
+
+ // 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().map_err(new_std_io_error)?;
+
+ let entry = if file_type.is_file() {
+ oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
+ } else if file_type.is_dir() {
+ // Make sure we are returning the correct path.
+ oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
+ } else {
+ oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
+ };
+
+ Ok(Some(entry))
}
}
diff --git a/core/src/services/hdfs/lister.rs b/core/src/services/hdfs/lister.rs
index ea8a57572..7a6f7788a 100644
--- a/core/src/services/hdfs/lister.rs
+++ b/core/src/services/hdfs/lister.rs
@@ -68,32 +68,26 @@ impl oio::List for HdfsLister {
}
impl oio::BlockingList for HdfsLister {
- 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());
+ fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let de = match self.rd.next() {
+ Some(de) => de,
+ None => return Ok(None),
+ };
- 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))
- };
+ let path = build_rel_path(&self.root, de.path());
- oes.push(d)
- }
+ 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))
+ };
- Ok(if oes.is_empty() { None } else { Some(oes) })
+ Ok(Some(entry))
}
}
diff --git a/core/src/types/list.rs b/core/src/types/list.rs
index 5493d58db..4b7813139 100644
--- a/core/src/types/list.rs
+++ b/core/src/types/list.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::VecDeque;
use std::pin::Pin;
use std::task::ready;
use std::task::Context;
@@ -127,8 +126,7 @@ pub struct BlockingLister {
/// required_metakey is the metakey required by users.
required_metakey: FlagSet<Metakey>,
- lister: Option<oio::BlockingLister>,
- buf: VecDeque<oio::Entry>,
+ lister: oio::BlockingLister,
}
/// # Safety
@@ -146,8 +144,7 @@ impl BlockingLister {
acc,
required_metakey,
- buf: VecDeque::new(),
- lister: Some(lister),
+ lister,
})
}
}
@@ -157,38 +154,25 @@ impl Iterator for BlockingLister {
type Item = Result<Entry>;
fn next(&mut self) -> Option<Self::Item> {
- 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 Some(Ok(Entry::new(path, metadata)));
+ let entry = match self.lister.next() {
+ Ok(Some(entry)) => entry,
+ Ok(None) => {
+ return None;
}
+ Err(err) => return Some(Err(err)),
+ };
- let metadata = match self.acc.blocking_stat(&path, OpStat::default()) {
- Ok(rp) => rp.into_metadata(),
- Err(err) => return Some(Err(err)),
- };
+ let (path, metadata) = entry.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 Some(Ok(Entry::new(path, metadata)));
}
- let lister = match self.lister.as_mut() {
- Some(lister) => lister,
- None => return None,
- };
-
- self.buf = match lister.next() {
- // Ideally, the convert from `Vec` to `VecDeque` will not do reallocation.
- //
- // However, this could be changed as described in [impl<T, A> From<Vec<T, A>> for VecDeque<T, A>](https://doc.rust-lang.org/std/collections/struct.VecDeque.html#impl-From%3CVec%3CT%2C%20A%3E%3E-for-VecDeque%3CT%2C%20A%3E)
- Ok(Some(entries)) => entries.into(),
- Ok(None) => {
- self.lister = None;
- return None;
- }
+ let metadata = match self.acc.blocking_stat(&path, OpStat::default()) {
+ Ok(rp) => rp.into_metadata(),
Err(err) => return Some(Err(err)),
};
-
- self.next()
+ Some(Ok(Entry::new(path, metadata)))
}
}