feat(oay): Add read_dir (#2736)

* add for oay-webdav * port to DavStream
apache · Aug 2, 2023 · ef1aeeb · ef1aeeb
1 parent 4958c84
commit ef1aeeb
Show file tree

Hide file tree

Showing 6 changed files with 84 additions and 18 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bin/oay/Cargo.toml b/bin/oay/Cargo.toml
@@ -37,7 +37,8 @@ default = [
 
 frontends-webdav = [
   "dep:dav-server",
-  "dep:bytes"
+  "dep:bytes",
+  "dep:futures-util"
 ]
 frontends-s3 = []
 
@@ -65,4 +66,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 url = "2.3.1"
 uuid = { version = "1", features = ["v4", "fast-rng"] }
 dav-server = { version = "0.5.5", optional = true }
-bytes = { version = "1.4.0", optional = true }
+bytes = { version = "1.4.0", optional = true }
+futures-util = { version = "0.3.16", optional = true }
diff --git a/bin/oay/src/services/webdav/webdav_dir_entry.rs b/bin/oay/src/services/webdav/webdav_dir_entry.rs
@@ -15,16 +15,33 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use dav_server::fs::DavDirEntry;
+use dav_server::fs::{DavDirEntry, DavMetaData};
+use futures::FutureExt;
+use opendal::{Entry, Operator};
 
-struct WebDAVDirEntry {}
+use super::webdav_metadata::WebdavMetaData;
+
+pub struct WebDAVDirEntry {
+    dir_entry: Entry,
+    op: Operator,
+}
 
 impl DavDirEntry for WebDAVDirEntry {
     fn name(&self) -> Vec<u8> {
-        todo!()
+        self.dir_entry.name().as_bytes().to_vec()
     }
 
     fn metadata(&self) -> dav_server::fs::FsFuture<Box<dyn dav_server::fs::DavMetaData>> {
-        todo!()
+        async move {
+            let metedata = self.op.stat(self.dir_entry.path()).await.unwrap();
+            Ok(Box::new(WebdavMetaData::new(metedata)) as Box<dyn DavMetaData>)
+        }
+        .boxed()
+    }
+}
+
+impl WebDAVDirEntry {
+    pub fn new(dir_entry: Entry, op: Operator) -> Self {
+        WebDAVDirEntry { dir_entry, op }
     }
 }
diff --git a/bin/oay/src/services/webdav/webdav_file.rs b/bin/oay/src/services/webdav/webdav_file.rs
@@ -37,10 +37,10 @@ pub struct WebdavFile {
 impl DavFile for WebdavFile {
     fn read_bytes(&mut self, count: usize) -> FsFuture<bytes::Bytes> {
         async move {
-            let file_path = self.path.as_rel_ospath();
+            let file_path = self.path.as_url_string();
             let content = self
                 .op
-                .range_read(file_path.to_str().unwrap(), 0..count as u64)
+                .range_read(&file_path, 0..count as u64)
                 .await
                 .unwrap();
             //error handle ?
@@ -53,7 +53,7 @@ impl DavFile for WebdavFile {
         async move {
             let opendal_metadata = self
                 .op
-                .stat(self.path.as_rel_ospath().to_str().unwrap())
+                .stat(self.path.as_url_string().as_str())
                 .await
                 .unwrap();
             Ok(Box::new(WebdavMetaData::new(opendal_metadata)) as Box<dyn DavMetaData>)

diff --git a/bin/oay/src/services/webdav/webdav_metadata.rs b/bin/oay/src/services/webdav/webdav_metadata.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use dav_server::fs::DavMetaData;
+use dav_server::fs::{DavMetaData, FsError};
 use opendal::Metadata;
 
 #[derive(Debug, Clone)]
@@ -35,7 +35,10 @@ impl DavMetaData for WebdavMetaData {
     }
 
     fn modified(&self) -> dav_server::fs::FsResult<std::time::SystemTime> {
-        Ok(self.metadata.last_modified().unwrap().into())
+        match self.metadata.last_modified() {
+            Some(t) => Ok(t.into()),
+            None => Err(FsError::GeneralFailure),
+        }
     }
 
     fn is_dir(&self) -> bool {

diff --git a/bin/oay/src/services/webdav/webdavfs.rs b/bin/oay/src/services/webdav/webdavfs.rs
@@ -15,12 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::pin::Pin;
+use std::task::Poll::{Pending, Ready};
+
+use dav_server::fs::DavDirEntry;
 use dav_server::fs::DavFile;
 use dav_server::fs::DavFileSystem;
 use dav_server::fs::DavMetaData;
 use futures::FutureExt;
+use futures_util::Stream;
+use futures_util::StreamExt;
+use opendal::Lister;
 use opendal::Operator;
 
+use crate::services::webdav::webdav_dir_entry::WebDAVDirEntry;
+
 use super::webdav_file::WebdavFile;
 use super::webdav_metadata::WebdavMetaData;
 
@@ -48,23 +57,23 @@ impl DavFileSystem for WebdavFs {
 
     fn read_dir<'a>(
         &'a self,
-        _path: &'a dav_server::davpath::DavPath,
+        path: &'a dav_server::davpath::DavPath,
         _meta: dav_server::fs::ReadDirMeta,
     ) -> dav_server::fs::FsFuture<dav_server::fs::FsStream<Box<dyn dav_server::fs::DavDirEntry>>>
     {
-        todo!()
+        async move {
+            let lister = self.op.list(path.as_url_string().as_str()).await.unwrap();
+            Ok(DavStream::new(self.op.clone(), lister).boxed())
+        }
+        .boxed()
     }
 
     fn metadata<'a>(
         &'a self,
         path: &'a dav_server::davpath::DavPath,
     ) -> dav_server::fs::FsFuture<Box<dyn dav_server::fs::DavMetaData>> {
         async move {
-            let opendal_metadata = self
-                .op
-                .stat(path.as_rel_ospath().to_str().unwrap())
-                .await
-                .unwrap();
+            let opendal_metadata = self.op.stat(path.as_url_string().as_str()).await.unwrap();
             Ok(Box::new(WebdavMetaData::new(opendal_metadata)) as Box<dyn DavMetaData>)
         }
         .boxed()
@@ -76,3 +85,37 @@ impl WebdavFs {
         Box::new(WebdavFs { op })
     }
 }
+
+struct DavStream {
+    op: Operator,
+    lister: Lister,
+}
+
+impl Stream for DavStream {
+    type Item = Box<dyn DavDirEntry>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let dav_stream = self.get_mut();
+        let lister = Pin::new(&mut dav_stream.lister).get_mut();
+
+        match Pin::new(lister).poll_next(cx) {
+            Ready(entry) => match entry {
+                Some(entry) => {
+                    let webdav_entry = WebDAVDirEntry::new(entry.unwrap(), dav_stream.op.clone());
+                    Ready(Some(Box::new(webdav_entry) as Box<dyn DavDirEntry>))
+                }
+                None => Ready(None),
+            },
+            Pending => Pending,
+        }
+    }
+}
+
+impl DavStream {
+    fn new(op: Operator, lister: Lister) -> Self {
+        DavStream { op, lister }
+    }
+}