fix(oay): support WebdavFile continuous reading and writing (#4374)

* fix(oay): support `WebdavFile` continuous reading and writing

* fix(oay): make clippy happy

integrations/dav-server/src/file.rs

@@ -15,37 +15,54 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::fmt;
 use std::io::SeekFrom;
 
 use bytes::Bytes;
-use dav_server::davpath::DavPath;
 use dav_server::fs::DavFile;
 use dav_server::fs::DavMetaData;
 use dav_server::fs::FsFuture;
 use futures::FutureExt;
 use opendal::Operator;
+use opendal::Reader;
+use opendal::Writer;
 
 use super::metadata::WebdavMetaData;
 
 #[derive(Debug)]
 pub struct WebdavFile {
     op: Operator,
-    path: DavPath,
+    path: String,
+    state: WebdavFileState,
+}
+
+struct WebdavFileState {
+    reader: Reader,
+    writer: Writer,
+}
+
+impl fmt::Debug for WebdavFileState {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("WebdavFileState").finish()
+    }
 }
 
 impl WebdavFile {
-    pub fn new(op: Operator, path: DavPath) -> Self {
-        Self { op, path }
+    pub fn new(op: Operator, reader: Reader, writer: Writer, path: String) -> Self {
+        Self {
+            op,
+            path,
+            state: WebdavFileState { reader, writer },
+        }
     }
 }
 
 impl DavFile for WebdavFile {
     fn read_bytes(&mut self, count: usize) -> FsFuture<Bytes> {
         async move {
-            let file_path = self.path.as_url_string();
-            self.op
-                .read_with(&file_path)
-                .range(0..count as u64)
+            self.state
+                .reader
+                .read(count)
                 .await
                 .map(Bytes::from)
                 .map_err(convert_error)
@@ -56,7 +73,7 @@ impl DavFile for WebdavFile {
     fn metadata(&mut self) -> FsFuture<Box<dyn DavMetaData>> {
         async move {
             self.op
-                .stat(self.path.as_url_string().as_str())
+                .stat(&self.path)
                 .await
                 .map(|opendal_metadata| {
                     Box::new(WebdavMetaData::new(opendal_metadata)) as Box<dyn DavMetaData>
@@ -71,11 +88,7 @@ impl DavFile for WebdavFile {
     }
 
     fn write_bytes(&mut self, buf: Bytes) -> FsFuture<()> {
-        async move {
-            let file_path = self.path.as_url_string();
-            self.op.write(&file_path, buf).await.map_err(convert_error)
-        }
-        .boxed()
+        async move { self.state.writer.write(buf).await.map_err(convert_error) }.boxed()
     }
 
     fn seek(&mut self, _pos: SeekFrom) -> FsFuture<u64> {

integrations/dav-server/src/opendalfs.rs

@@ -48,7 +48,10 @@ impl DavFileSystem for OpendalFs {
         _options: dav_server::fs::OpenOptions,
     ) -> dav_server::fs::FsFuture<Box<dyn dav_server::fs::DavFile>> {
         async move {
-            let file = WebdavFile::new(self.op.clone(), path.clone());
+            let path = path.as_url_string();
+            let reader = self.op.reader(&path).await.map_err(convert_error)?;
+            let writer = self.op.writer(&path).await.map_err(convert_error)?;
+            let file = WebdavFile::new(self.op.clone(), reader, writer, path.clone());
             Ok(Box::new(file) as Box<dyn DavFile>)
         }
         .boxed()

integrations/dav-server/tests/test.rs

@@ -16,14 +16,15 @@
 // under the License.
 
 use anyhow::Result;
+use bytes::Bytes;
 use dav_server::davpath::DavPath;
 use dav_server::fs::DavFileSystem;
 use dav_server_opendalfs::OpendalFs;
 use opendal::services::Fs;
 use opendal::Operator;
 
 #[tokio::test]
-async fn test() -> Result<()> {
+async fn test_metadata() -> Result<()> {
     let mut builder = Fs::default();
     builder.root("/tmp");
 
@@ -35,7 +36,47 @@ async fn test() -> Result<()> {
         .metadata(&DavPath::new("/").unwrap())
         .await
         .unwrap();
-    println!("{}", metadata.is_dir());
+    assert!(metadata.is_dir());
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_write_and_read() -> Result<()> {
+    let mut builder = Fs::default();
+    builder.root("/tmp");
+
+    let op = Operator::new(builder)?.finish();
+
+    let webdavfs = OpendalFs::new(op);
+
+    let path = &DavPath::new("/test_opendalfs_write_read.txt").expect("path must be valid");
+    let content = "Hello dav-server-opendalfs.";
+
+    let mut davfile = webdavfs
+        .open(path, dav_server::fs::OpenOptions::default())
+        .await
+        .expect("open file must succeed");
+
+    let num = 999;
+    for i in 0..num {
+        davfile
+            .write_bytes(Bytes::from(format!("{}{}", content, i)))
+            .await
+            .expect("write file must succeed");
+    }
+
+    for i in 0..num {
+        let read_content = davfile
+            .read_bytes(content.len() + i.to_string().len())
+            .await
+            .expect("read file must succeed");
+        assert_eq!(read_content, Bytes::from(format!("{}{}", content, i)));
+    }
+
+    webdavfs
+        .remove_file(path)
+        .await
+        .expect("remove file must succeed");
+    Ok(())
+}