s3 fs (#28)

* feat: support s3 write * feat: support native s3 fs * pub tokio http client * feat: support s3 list and remove
tonbo-io · Sep 29, 2024 · a7b9b00 · a7b9b00
1 parent 317b1b0
commit a7b9b00
Show file tree

Hide file tree

Showing 14 changed files with 738 additions and 181 deletions.
diff --git a/fusio/Cargo.toml b/fusio/Cargo.toml
@@ -7,22 +7,26 @@ version = "0.1.0"
 
 [features]
 aws = [
+    "base64",
     "bytes",
     "chrono",
     "chrono?/serde",
     "fs",
     "http",
+    "quick-xml",
     "ring",
     "serde",
     "serde_json",
+    "serde_urlencoded",
 ]
 bytes = ["dep:bytes"]
 completion-based = []
 default = ["dyn", "fs"]
 dyn = []
 fs = ["tokio?/rt"]
-h2 = ["dep:h2"]
 http = [
+    "async-stream",
+    "bytes",
     "dep:http",
     "http-body",
     "http-body-util",
@@ -40,7 +44,9 @@ tokio-uring = ["async-stream", "completion-based", "dep:tokio-uring", "no-send"]
 
 [dependencies]
 async-stream = { version = "0.3", optional = true }
+base64 = { version = "0.22", optional = true }
 bytes = { workspace = true, optional = true }
+cfg-if = "1.0.0"
 chrono = { version = "0.4", optional = true, default-features = false, features = [
     "now",
     "std",
@@ -62,12 +68,19 @@ object_store = { version = "0.11", optional = true, default-features = false, fe
     "aws",
 ] }
 percent-encoding = { version = "2" }
-reqwest = { version = "0.12.7", optional = true, features = ["stream"] }
+quick-xml = { version = "0.36", features = [
+    "overlapped-lists",
+    "serialize",
+], optional = true }
+reqwest = { git = "https://github.com/seanmonstar/reqwest.git", optional = true, features = [
+    "stream",
+] }
 ring = { version = "0.17", optional = true, default-features = false, features = [
     "std",
 ] }
 serde = { version = "1", optional = true, features = ["derive"] }
 serde_json = { version = "1", optional = true }
+serde_urlencoded = { version = "0.7", optional = true }
 thiserror = "1"
 tokio = { version = "1", optional = true, default-features = false, features = [
     "fs",
@@ -79,6 +92,7 @@ url = { version = "2" }
 tokio-uring = { version = "0.5", default-features = false, optional = true }
 
 [dev-dependencies]
+futures-util = { version = "0.3" }
 hyper = { version = "1", features = ["full"] }
 hyper-util = { version = "0.1", features = ["full"] }
 monoio = { version = "0.2" }

diff --git a/fusio/src/buf.rs b/fusio/src/buf.rs
@@ -47,8 +47,12 @@ pub trait IoBufMut: IoBuf {
         unsafe { std::slice::from_raw_parts_mut(self.as_mut_ptr(), self.bytes_init()) }
     }
 
+    /// # Safety
+    /// The buffer must be recovered from the same type of buffer before it drops.
     unsafe fn to_buf_mut_nocopy(self) -> BufMut;
 
+    /// # Safety
+    /// The buffer must be recovered from the same type.
     unsafe fn recover_from_buf_mut(buf: BufMut) -> Self;
 }
 

diff --git a/fusio/src/error.rs b/fusio/src/error.rs
@@ -4,6 +4,7 @@ use thiserror::Error;
 
 #[derive(Debug, Error)]
 #[error(transparent)]
+#[non_exhaustive]
 pub enum Error {
     Io(#[from] io::Error),
     #[cfg(feature = "http")]
@@ -14,8 +15,16 @@ pub enum Error {
     #[error("unsupported operation")]
     Unsupported,
     #[error(transparent)]
-    Other(BoxError),
+    Other(#[from] BoxedError),
+    #[error("invalid url: {0}")]
+    InvalidUrl(BoxedError),
+    #[cfg(feature = "http")]
+    #[error("http request failed, status: {status_code}, body: {body}")]
+    HttpNotSuccess {
+        status_code: http::StatusCode,
+        body: String,
+    },
 }
 
 #[allow(unused)]
-pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
+pub type BoxedError = Box<dyn std::error::Error + Send + Sync + 'static>;
diff --git a/fusio/src/fs/mod.rs b/fusio/src/fs/mod.rs
@@ -7,6 +7,7 @@ pub use options::*;
 
 use crate::{path::Path, Error, MaybeSend, MaybeSync, Read, Seek, Write};
 
+#[derive(Debug)]
 pub struct FileMeta {
     pub path: Path,
     pub size: u64,

diff --git a/fusio/src/local/mod.rs b/fusio/src/local/mod.rs
@@ -14,3 +14,12 @@ pub use tokio::fs::*;
 #[cfg(all(feature = "tokio-uring", target_os = "linux", feature = "fs"))]
 #[allow(unused)]
 pub use tokio_uring::fs::*;
+
+#[cfg(feature = "fs")]
+cfg_if::cfg_if! {
+    if #[cfg(feature = "tokio")] {
+        pub type LocalFs = TokioFs;
+    } else if #[cfg(feature = "monoio")] {
+        pub type LocalFs = MonoIoFs;
+    }
+}