feat(!): Make read_dir failable (#31)

* feat: Make read_dir failable Signed-off-by: Xuanwo <[email protected]> * Handle error correctly Signed-off-by: Xuanwo <[email protected]> --------- Signed-off-by: Xuanwo <[email protected]>
messense · Jun 18, 2024 · f668f29 · f668f29
1 parent 0e39d10
commit f668f29
Show file tree

Hide file tree

Showing 7 changed files with 45 additions and 5 deletions.
diff --git a/src/fs.rs b/src/fs.rs
@@ -115,7 +115,7 @@ pub struct DavProp {
 /// Future returned by almost all of the DavFileSystem methods.
 pub type FsFuture<'a, T> = Pin<Box<dyn Future<Output = FsResult<T>> + Send + 'a>>;
 /// Convenience alias for a boxed Stream.
-pub type FsStream<T> = Pin<Box<dyn Stream<Item = T> + Send>>;
+pub type FsStream<T> = Pin<Box<dyn Stream<Item = FsResult<T>> + Send>>;
 
 /// Used as argument to the read_dir() method.
 /// It is:

diff --git a/src/handle_copymove.rs b/src/handle_copymove.rs
@@ -85,6 +85,11 @@ impl crate::DavInner {
             // Last seen error is returned from function.
             let mut retval = Ok::<_, DavError>(());
             while let Some(dirent) = entries.next().await {
+                let dirent = match dirent {
+                    Ok(dirent) => dirent,
+                    Err(e) => return add_status(multierror, source, e).await,
+                };
+
                 // NOTE: dirent.metadata() behaves like symlink_metadata()
                 let meta = match dirent.metadata().await {
                     Ok(meta) => meta,

diff --git a/src/handle_delete.rs b/src/handle_delete.rs
@@ -66,6 +66,14 @@ impl crate::DavInner {
 
             let mut result = Ok(());
             while let Some(dirent) = entries.next().await {
+                let dirent = match dirent {
+                    Ok(dirent) => dirent,
+                    Err(e) => {
+                        result = Err(add_status(res, path, e).await);
+                        continue;
+                    }
+                };
+
                 // if metadata() fails, skip to next entry.
                 // NOTE: dirent.metadata == symlink_metadata (!)
                 let meta = match dirent.metadata().await {

diff --git a/src/handle_gethead.rs b/src/handle_gethead.rs
@@ -325,6 +325,14 @@ impl crate::DavInner {
 
                 let mut dirents: Vec<Dirent> = Vec::new();
                 while let Some(dirent) = entries.next().await {
+                    let dirent = match dirent {
+                        Ok(dirent) => dirent,
+                        Err(e) => {
+                            trace!("next dir entry error happened. Skipping {:?}", e);
+                            continue;
+                        }
+                    };
+
                     let mut name = dirent.name();
                     if name.starts_with(b".") {
                         continue;

diff --git a/src/handle_props.rs b/src/handle_props.rs
@@ -239,6 +239,14 @@ impl DavInner {
             };
 
             while let Some(dirent) = entries.next().await {
+                let dirent = match dirent {
+                    Ok(dirent) => dirent,
+                    Err(e) => {
+                        trace!("next dir entry error happened. Skipping {:?}", e);
+                        continue;
+                    }
+                };
+
                 let mut npath = path.clone();
                 npath.push_segment(&dirent.name());
                 let meta = match dirent.metadata().await {

diff --git a/src/localfs.rs b/src/localfs.rs
@@ -532,7 +532,7 @@ impl LocalFsReadDir {
 
 // The stream implementation tries to be smart and batch I/O operations
 impl<'a> Stream for LocalFsReadDir {
-    type Item = Box<dyn DavDirEntry>;
+    type Item = FsResult<Box<dyn DavDirEntry>>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = Pin::into_inner(self);
@@ -569,8 +569,18 @@ impl<'a> Stream for LocalFsReadDir {
 
         // we filled the buffer, now pop from the buffer.
         match this.buffer.pop_front() {
-            Some(Ok(item)) => Poll::Ready(Some(Box::new(item))),
-            Some(Err(_)) | None => {
+            Some(Ok(item)) => Poll::Ready(Some(Ok(Box::new(item)))),
+            Some(Err(err)) => {
+                // fuse the iterator.
+                this.iterator.take();
+                // finish the cache.
+                if let Some(ref mut nb) = this.dir_cache {
+                    nb.finish();
+                }
+                // return error of stream.
+                Poll::Ready(Some(Err(err.into())))
+            }
+            None => {
                 // fuse the iterator.
                 this.iterator.take();
                 // finish the cache.

diff --git a/src/memfs.rs b/src/memfs.rs
@@ -15,6 +15,7 @@ use bytes::{Buf, Bytes};
 use futures_util::{
     future,
     future::{BoxFuture, FutureExt},
+    StreamExt,
 };
 use http::StatusCode;
 
@@ -152,7 +153,7 @@ impl DavFileSystem for MemFs {
                     v.push(Box::new(node.as_dirent(&name)));
                 }
             }
-            let strm = futures_util::stream::iter(v.into_iter());
+            let strm = futures_util::stream::iter(v.into_iter()).map(Ok);
             Ok(Box::pin(strm) as FsStream<Box<dyn DavDirEntry>>)
         }
         .boxed()