feat(oay): port WebdavFs to dav-server-fs-opendal (#3119)

Cargo.toml

@@ -38,6 +38,7 @@ members = [
   "bin/oay",
 
   "integrations/object_store",
+  "integrations/dav-server",
 ]
 resolver = "2"
 

integrations/README.md

@@ -5,3 +5,5 @@ This folder contains the integrations for OpenDAL. Integrations are used to inte
 ## Available Integrations
 
 - [`object_store_opendal`](./object_store): Use OpenDAL as a backend for the [object_store](https://docs.rs/object_store/latest/object_store/).
+
+- [`dav-server-opendalfs`](./dav-server-opendalfs/): Use OpenDAL as a backend to access data in various service with WebDAV protocol.

integrations/dav-server/Cargo.toml

@@ -0,0 +1,47 @@
+# 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.
+
+[package]
+description = "Use OpenDAL as a backend to access data in various service with WebDAV protocol"
+name = "dav-server-opendalfs"
+
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+anyhow = "1"
+chrono = "0.4.28"
+dirs = "5.0.0"
+bytes = { version = "1.4.0" }
+dav-server = { version = "0.5.5" }
+futures = "0.3"
+futures-util = { version = "0.3.16" }
+opendal.workspace = true
+quick-xml = { version = "0.29", features = ["serialize", "overlapped-lists"] }
+serde = { version = "1", features = ["derive"] }
+tokio = { version = "1.27", features = [
+  "fs",
+  "macros",
+  "rt-multi-thread",
+  "io-std",
+] }
+

integrations/dav-server/README.md

@@ -0,0 +1,3 @@
+# dav-server-opendalfs
+
+`dav-server-opendalfs` is a integration which use OpenDAL as a backend to access data in various service with WebDAV protocol.

integrations/dav-server/src/dir_entry.rs

@@ -0,0 +1,54 @@
+// 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 dav_server::fs::DavDirEntry;
+use futures::FutureExt;
+use opendal::Entry;
+use opendal::Operator;
+
+use super::file::convert_error;
+use super::metadata::WebdavMetaData;
+
+pub struct WebDAVDirEntry {
+    dir_entry: Entry,
+    op: Operator,
+}
+
+impl DavDirEntry for WebDAVDirEntry {
+    fn name(&self) -> Vec<u8> {
+        self.dir_entry.name().as_bytes().to_vec()
+    }
+
+    fn metadata(&self) -> dav_server::fs::FsFuture<Box<dyn dav_server::fs::DavMetaData>> {
+        async move {
+            self.op
+                .stat(self.dir_entry.path())
+                .await
+                .map(|metadata| {
+                    Box::new(WebdavMetaData::new(metadata)) as Box<dyn dav_server::fs::DavMetaData>
+                })
+                .map_err(convert_error)
+        }
+        .boxed()
+    }
+}
+
+impl WebDAVDirEntry {
+    pub fn new(dir_entry: Entry, op: Operator) -> Self {
+        WebDAVDirEntry { dir_entry, op }
+    }
+}

integrations/dav-server/src/file.rs

@@ -0,0 +1,98 @@
+// 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 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 super::metadata::WebdavMetaData;
+
+#[derive(Debug)]
+pub struct WebdavFile {
+    op: Operator,
+    path: DavPath,
+}
+
+impl WebdavFile {
+    pub fn new(op: Operator, path: DavPath) -> Self {
+        Self { op, path }
+    }
+}
+
+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)
+                .await
+                .map(Bytes::from)
+                .map_err(convert_error)
+        }
+        .boxed()
+    }
+
+    fn metadata(&mut self) -> FsFuture<Box<dyn DavMetaData>> {
+        async move {
+            self.op
+                .stat(self.path.as_url_string().as_str())
+                .await
+                .map(|opendal_metadata| {
+                    Box::new(WebdavMetaData::new(opendal_metadata)) as Box<dyn DavMetaData>
+                })
+                .map_err(convert_error)
+        }
+        .boxed()
+    }
+
+    fn write_buf(&mut self, buf: Box<dyn bytes::Buf + Send>) -> FsFuture<()> {
+        self.write_bytes(Bytes::copy_from_slice(buf.chunk()))
+    }
+
+    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()
+    }
+
+    fn seek(&mut self, _pos: SeekFrom) -> FsFuture<u64> {
+        futures_util::future::err(dav_server::fs::FsError::NotImplemented).boxed()
+    }
+
+    fn flush(&mut self) -> FsFuture<()> {
+        futures_util::future::ok(()).boxed()
+    }
+}
+
+pub fn convert_error(opendal_error: opendal::Error) -> dav_server::fs::FsError {
+    match opendal_error.kind() {
+        opendal::ErrorKind::AlreadyExists | opendal::ErrorKind::IsSameFile => {
+            dav_server::fs::FsError::Exists
+        }
+        opendal::ErrorKind::NotFound => dav_server::fs::FsError::NotFound,
+        _ => dav_server::fs::FsError::GeneralFailure,
+    }
+}

integrations/dav-server/src/lib.rs

@@ -0,0 +1,23 @@
+// 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.
+
+mod dir_entry;
+mod file;
+mod metadata;
+mod opendalfs;
+
+pub use opendalfs::OpendalFs;

integrations/dav-server/src/metadata.rs

@@ -0,0 +1,62 @@
+// 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 dav_server::fs::DavMetaData;
+use dav_server::fs::FsError;
+use opendal::Metadata;
+
+#[derive(Debug, Clone)]
+pub struct WebdavMetaData {
+    metadata: Metadata,
+}
+
+impl WebdavMetaData {
+    pub fn new(metadata: Metadata) -> Self {
+        WebdavMetaData { metadata }
+    }
+}
+
+impl DavMetaData for WebdavMetaData {
+    fn len(&self) -> u64 {
+        self.metadata.content_length()
+    }
+
+    fn modified(&self) -> dav_server::fs::FsResult<std::time::SystemTime> {
+        match self.metadata.last_modified() {
+            Some(t) => Ok(t.into()),
+            None => Err(FsError::GeneralFailure),
+        }
+    }
+
+    fn is_dir(&self) -> bool {
+        self.metadata.is_dir()
+    }
+
+    fn is_file(&self) -> bool {
+        self.metadata.is_file()
+    }
+
+    fn etag(&self) -> Option<String> {
+        self.metadata.etag().map(|s| s.to_string())
+    }
+
+    fn status_changed(&self) -> dav_server::fs::FsResult<std::time::SystemTime> {
+        self.metadata
+            .last_modified()
+            .map_or(Err(FsError::GeneralFailure), |t| Ok(t.into()))
+    }
+}