feat(oay): Add webdav basic read impl (#2658)

* add oay_webdav basic read impl

* update

* cargo fmt

bin/oay/Cargo.toml

@@ -29,6 +29,18 @@ repository.workspace = true
 rust-version.workspace = true
 version.workspace = true
 
+[features]
+default = [
+  "frontends-webdav",
+  "frontends-s3"
+]
+
+frontends-webdav = [
+  "dep:dav-server",
+  "dep:bytes"
+]
+frontends-s3 = []
+
 [dependencies]
 anyhow = "1"
 axum = "0.6"
@@ -52,3 +64,5 @@ tracing = "0.1"
 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 }

bin/oay/src/bin/oay.rs

@@ -21,7 +21,9 @@ use std::sync::Arc;
 use anyhow::Context;
 use anyhow::Result;
 use oay::services::S3Service;
+use oay::services::WebdavService;
 use oay::Config;
+use opendal::services::Fs;
 use opendal::Operator;
 use opendal::Scheme;
 use tracing_subscriber::fmt;
@@ -30,6 +32,11 @@ use tracing_subscriber::EnvFilter;
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    let _ = s3().await;
+    webdav().await
+}
+
+async fn s3() -> Result<()> {
     tracing_subscriber::registry()
         .with(fmt::layer().pretty())
         .with(EnvFilter::from_default_env())
@@ -46,3 +53,30 @@ async fn main() -> Result<()> {
 
     Ok(())
 }
+
+async fn webdav() -> Result<()> {
+    let cfg: Config = Config {
+        backend: oay::BackendConfig {
+            typ: "fs".to_string(),
+            ..Default::default()
+        },
+        frontends: oay::FrontendsConfig {
+            webdav: oay::WebdavConfig {
+                enable: true,
+                addr: "127.0.0.1:3000".to_string(),
+            },
+            ..Default::default()
+        },
+    };
+
+    let mut builder = Fs::default();
+    builder.root("/tmp");
+
+    let op = Operator::new(builder)?.finish();
+
+    let webdav = WebdavService::new(Arc::new(cfg), op);
+
+    webdav.serve().await?;
+
+    Ok(())
+}

bin/oay/src/bin/webdav.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 std::sync::Arc;
+
+use anyhow::Result;
+
+use oay::services::WebdavService;
+use oay::Config;
+use opendal::services::Fs;
+
+use opendal::Operator;
+use tracing_subscriber::fmt;
+use tracing_subscriber::prelude::*;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::registry()
+        .with(fmt::layer().pretty())
+        .with(EnvFilter::from_default_env())
+        .init();
+
+    let cfg: Config = Config {
+        backend: oay::BackendConfig {
+            typ: "fs".to_string(),
+            ..Default::default()
+        },
+        frontends: oay::FrontendsConfig {
+            webdav: oay::WebdavConfig {
+                enable: true,
+                addr: "127.0.0.1:3000".to_string(),
+            },
+            ..Default::default()
+        },
+    };
+
+    let mut builder = Fs::default();
+    builder.root("/tmp");
+
+    let op = Operator::new(builder)?.finish();
+
+    let webdav = WebdavService::new(Arc::new(cfg), op);
+
+    webdav.serve().await?;
+
+    Ok(())
+}

bin/oay/src/config.rs

@@ -26,21 +26,28 @@ pub struct Config {
     pub frontends: FrontendsConfig,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct BackendConfig {
     #[serde(rename = "type")]
     pub typ: String,
     #[serde(flatten)]
     pub map: HashMap<String, String>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct FrontendsConfig {
     pub s3: S3Config,
+    pub webdav: WebdavConfig,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct S3Config {
     pub enable: bool,
     pub addr: String,
 }
+
+#[derive(Serialize, Deserialize, Default)]
+pub struct WebdavConfig {
+    pub enable: bool,
+    pub addr: String,
+}

bin/oay/src/services/mod.rs

@@ -15,5 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#[cfg(feature = "frontends-s3")]
 mod s3;
+#[cfg(feature = "frontends-s3")]
 pub use s3::S3Service;
+
+#[cfg(feature = "frontends-webdav")]
+mod webdav;
+#[cfg(feature = "frontends-webdav")]
+pub use webdav::WebdavService;

bin/oay/src/services/webdav/mod.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 service;
+mod webdav_dir_entry;
+mod webdav_file;
+mod webdav_metadata;
+pub mod webdavfs;
+pub use service::*;

bin/oay/src/services/webdav/service.rs

@@ -0,0 +1,65 @@
+// 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::convert::Infallible;
+use std::sync::Arc;
+
+use axum::body::Body;
+use axum::http::Request;
+use axum::routing::any_service;
+use axum::Router;
+use dav_server::DavHandler;
+use opendal::Operator;
+
+use crate::Config;
+
+use super::webdavfs::WebdavFs;
+
+pub struct WebdavService {
+    cfg: Arc<Config>,
+    webdavfs: Box<WebdavFs>,
+}
+
+impl WebdavService {
+    pub fn new(cfg: Arc<Config>, op: Operator) -> Self {
+        Self {
+            cfg,
+            webdavfs: WebdavFs::new(op),
+        }
+    }
+
+    pub async fn serve(&self) -> anyhow::Result<()> {
+        let webdav_cfg = &self.cfg.frontends.webdav;
+
+        let webdav_handler = DavHandler::builder()
+            .filesystem(self.webdavfs.clone())
+            .build_handler();
+
+        let webdav_service = tower::service_fn(move |req: Request<Body>| {
+            let webdav_server = webdav_handler.clone();
+            async move { Ok::<_, Infallible>(webdav_server.handle(req).await) }
+        });
+
+        let app = Router::new().route("/*path", any_service(webdav_service));
+
+        axum::Server::bind(&webdav_cfg.addr.parse().unwrap())
+            .serve(app.into_make_service())
+            .await?;
+
+        Ok(())
+    }
+}

bin/oay/src/services/webdav/webdav_dir_entry.rs

@@ -0,0 +1,30 @@
+// 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;
+
+struct WebDAVDirEntry {}
+
+impl DavDirEntry for WebDAVDirEntry {
+    fn name(&self) -> Vec<u8> {
+        todo!()
+    }
+
+    fn metadata(&self) -> dav_server::fs::FsFuture<Box<dyn dav_server::fs::DavMetaData>> {
+        todo!()
+    }
+}

bin/oay/src/services/webdav/webdav_file.rs

@@ -0,0 +1,78 @@
+// 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 dav_server::{
+    davpath::DavPath,
+    fs::{DavFile, DavMetaData, FsFuture, OpenOptions},
+};
+use futures::FutureExt;
+use opendal::Operator;
+
+use super::webdav_metadata::WebdavMetaData;
+
+#[derive(Debug)]
+pub struct WebdavFile {
+    pub op: Operator,
+    pub path: DavPath,
+    pub options: OpenOptions,
+}
+
+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 content = self
+                .op
+                .range_read(file_path.to_str().unwrap(), 0..count as u64)
+                .await
+                .unwrap();
+            //error handle ?
+            Ok(bytes::Bytes::from(content))
+        }
+        .boxed()
+    }
+
+    fn metadata(&mut self) -> FsFuture<Box<dyn DavMetaData>> {
+        async move {
+            let opendal_metadata = self
+                .op
+                .stat(self.path.as_rel_ospath().to_str().unwrap())
+                .await
+                .unwrap();
+            Ok(Box::new(WebdavMetaData::new(opendal_metadata)) as Box<dyn DavMetaData>)
+        }
+        .boxed()
+    }
+
+    fn write_buf(&mut self, _buf: Box<dyn bytes::Buf + Send>) -> FsFuture<()> {
+        todo!()
+    }
+
+    fn write_bytes(&mut self, _buf: bytes::Bytes) -> FsFuture<()> {
+        todo!()
+    }
+
+    fn seek(&mut self, _pos: SeekFrom) -> FsFuture<u64> {
+        todo!()
+    }
+
+    fn flush(&mut self) -> FsFuture<()> {
+        todo!()
+    }
+}