You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/01/04 15:09:35 UTC
(incubator-opendal) branch format_code created (now f4417cce8e)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch format_code
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
at f4417cce8e chore: Format code to make readers happy
This branch includes the following new commits:
new f4417cce8e chore: Format code to make readers happy
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-opendal) 01/01: chore: Format code to make readers happy
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch format_code
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit f4417cce8e4e458e07d5b713e970036446be945e
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Jan 4 23:09:16 2024 +0800
chore: Format code to make readers happy
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
Cargo.toml | 10 +-
bin/ofs/src/lib.rs | 10 +-
bindings/java/Cargo.toml | 4 +-
bindings/nodejs/Cargo.toml | 4 +-
bindings/php/Cargo.toml | 4 +-
bindings/python/Cargo.toml | 4 +-
core/Cargo.toml | 18 +-
.../Cargo.toml | 2 +-
.../src/main.rs | 3 +-
core/edge/s3_read_on_wasm/Cargo.toml | 4 +-
core/edge/s3_read_on_wasm/src/lib.rs | 3 +-
core/src/layers/complete.rs | 28 +--
core/src/layers/retry.rs | 4 +-
core/src/lib.rs | 4 +-
core/src/raw/accessor.rs | 230 ++++++++++---------
core/src/raw/enum_utils.rs | 6 +-
core/src/raw/futures_util.rs | 17 +-
core/src/raw/http_util/header.rs | 6 +-
core/src/raw/oio/read/buffer_reader.rs | 18 +-
core/src/raw/oio/read/range_read.rs | 4 +-
core/src/raw/oio/stream/into_stream.rs | 14 +-
core/src/services/alluxio/backend.rs | 28 +--
core/src/services/atomicserver/backend.rs | 3 +-
core/src/services/azblob/backend.rs | 54 ++---
core/src/services/azdls/backend.rs | 116 +++++-----
core/src/services/azfile/backend.rs | 62 +++---
core/src/services/b2/backend.rs | 118 +++++-----
core/src/services/chainsafe/backend.rs | 43 ++--
core/src/services/chainsafe/core.rs | 1 -
core/src/services/chainsafe/lister.rs | 4 +-
core/src/services/chainsafe/writer.rs | 5 +-
core/src/services/cos/backend.rs | 50 ++---
core/src/services/dbfs/backend.rs | 62 +++---
core/src/services/dropbox/backend.rs | 80 +++----
core/src/services/fs/backend.rs | 200 ++++++++---------
core/src/services/ftp/backend.rs | 42 ++--
core/src/services/gcs/backend.rs | 112 +++++-----
core/src/services/gdrive/backend.rs | 70 +++---
core/src/services/ghac/backend.rs | 60 ++---
core/src/services/hdfs/backend.rs | 247 +++++++++++----------
core/src/services/hdfs/writer.rs | 9 +-
core/src/services/http/backend.rs | 46 ++--
core/src/services/huggingface/backend.rs | 48 ++--
core/src/services/ipfs/backend.rs | 26 +--
core/src/services/ipmfs/backend.rs | 40 ++--
core/src/services/memcached/backend.rs | 9 +-
core/src/services/obs/backend.rs | 88 ++++----
core/src/services/onedrive/backend.rs | 94 ++++----
core/src/services/oss/backend.rs | 58 ++---
core/src/services/pcloud/backend.rs | 154 +++++++------
core/src/services/pcloud/core.rs | 14 +-
core/src/services/pcloud/writer.rs | 6 +-
core/src/services/rocksdb/backend.rs | 7 +-
core/src/services/s3/backend.rs | 60 ++---
core/src/services/seafile/backend.rs | 52 ++---
core/src/services/seafile/core.rs | 11 +-
core/src/services/seafile/lister.rs | 5 +-
core/src/services/seafile/writer.rs | 4 +-
core/src/services/sftp/backend.rs | 92 ++++----
core/src/services/supabase/backend.rs | 40 ++--
core/src/services/swift/backend.rs | 64 +++---
core/src/services/upyun/backend.rs | 84 ++++---
core/src/services/upyun/core.rs | 14 +-
core/src/services/upyun/lister.rs | 3 +-
core/src/services/upyun/writer.rs | 5 +-
core/src/services/vercel_artifacts/backend.rs | 34 +--
core/src/services/webdav/backend.rs | 140 ++++++------
core/src/services/webhdfs/backend.rs | 68 +++---
core/tests/behavior/blocking_write.rs | 4 +-
core/tests/behavior/main.rs | 3 +-
core/tests/behavior/utils.rs | 3 +-
deny.toml | 12 +-
72 files changed, 1538 insertions(+), 1513 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index 7bc750cf5a..267a1858dc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -18,12 +18,12 @@
[workspace]
default-members = ["core"]
exclude = [
- "examples",
+ "examples",
- "bindings/c",
- "bindings/ocaml",
- "bindings/php",
- "bindings/ruby"
+ "bindings/c",
+ "bindings/ocaml",
+ "bindings/php",
+ "bindings/ruby",
]
members = [
"core",
diff --git a/bin/ofs/src/lib.rs b/bin/ofs/src/lib.rs
index bff825fb5d..3fc0cadae6 100644
--- a/bin/ofs/src/lib.rs
+++ b/bin/ofs/src/lib.rs
@@ -15,14 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use fuse3::path::prelude::*;
-use fuse3::Result;
-
-use async_trait::async_trait;
-use futures_util::stream::{Empty, Iter};
use std::ffi::OsStr;
use std::vec::IntoIter;
+use async_trait::async_trait;
+use fuse3::path::prelude::*;
+use fuse3::Result;
+use futures_util::stream::Empty;
+use futures_util::stream::Iter;
use opendal::Operator;
pub struct Ofs {
diff --git a/bindings/java/Cargo.toml b/bindings/java/Cargo.toml
index f9c3dc5e04..9787a7325f 100644
--- a/bindings/java/Cargo.toml
+++ b/bindings/java/Cargo.toml
@@ -114,7 +114,6 @@ services-webhdfs = ["opendal/services-webhdfs"]
services-alluxio = ["opendal/services-alluxio"]
services-azfile = ["opendal/services-azfile"]
services-b2 = ["opendal/services-b2"]
-services-upyun = ["opendal/services-upyun"]
services-cacache = ["opendal/services-cacache"]
services-dashmap = ["opendal/services-dashmap"]
services-dropbox = ["opendal/services-dropbox"]
@@ -138,13 +137,14 @@ services-postgresql = ["opendal/services-postgresql"]
services-redb = ["opendal/services-redb"]
services-redis = ["opendal/services-redis"]
services-rocksdb = ["opendal/services-rocksdb"]
+services-seafile = ["opendal/services-seafile"]
services-sftp = ["opendal/services-sftp"]
services-sled = ["opendal/services-sled"]
-services-seafile = ["opendal/services-seafile"]
services-sqlite = ["opendal/services-sqlite"]
services-supabase = ["opendal/services-supabase"]
services-swift = ["opendal/services-swift"]
services-tikv = ["opendal/services-tikv"]
+services-upyun = ["opendal/services-upyun"]
services-vercel-artifacts = ["opendal/services-vercel-artifacts"]
services-wasabi = ["opendal/services-wasabi"]
diff --git a/bindings/nodejs/Cargo.toml b/bindings/nodejs/Cargo.toml
index 90c04dd4fb..51da25bd66 100644
--- a/bindings/nodejs/Cargo.toml
+++ b/bindings/nodejs/Cargo.toml
@@ -108,7 +108,6 @@ services-webhdfs = ["opendal/services-webhdfs"]
# Optional services provided by opendal.
services-alluxio = ["opendal/services-alluxio"]
services-azfile = ["opendal/services-azfile"]
-services-upyun = ["opendal/services-upyun"]
services-b2 = ["opendal/services-b2"]
services-cacache = ["opendal/services-cacache"]
services-dashmap = ["opendal/services-dashmap"]
@@ -133,13 +132,14 @@ services-postgresql = ["opendal/services-postgresql"]
services-redb = ["opendal/services-redb"]
services-redis = ["opendal/services-redis"]
services-rocksdb = ["opendal/services-rocksdb"]
+services-seafile = ["opendal/services-seafile"]
services-sftp = ["opendal/services-sftp"]
services-sled = ["opendal/services-sled"]
services-sqlite = ["opendal/services-sqlite"]
-services-seafile = ["opendal/services-seafile"]
services-supabase = ["opendal/services-supabase"]
services-swift = ["opendal/services-swift"]
services-tikv = ["opendal/services-tikv"]
+services-upyun = ["opendal/services-upyun"]
services-vercel-artifacts = ["opendal/services-vercel-artifacts"]
services-wasabi = ["opendal/services-wasabi"]
diff --git a/bindings/php/Cargo.toml b/bindings/php/Cargo.toml
index cddc68cb17..2cd7ff8248 100644
--- a/bindings/php/Cargo.toml
+++ b/bindings/php/Cargo.toml
@@ -16,13 +16,13 @@
# under the License.
[package]
-name = "opendal-php"
-version = "0.1.0"
edition = "2021"
homepage = "https://opendal.apache.org/"
license = "Apache-2.0"
+name = "opendal-php"
repository = "https://github.com/apache/incubator-opendal"
rust-version = "1.67"
+version = "0.1.0"
[lib]
crate-type = ["cdylib"]
diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml
index 2dc3d394c3..711a11a98e 100644
--- a/bindings/python/Cargo.toml
+++ b/bindings/python/Cargo.toml
@@ -111,7 +111,6 @@ services-webhdfs = ["opendal/services-webhdfs"]
services-alluxio = ["opendal/services-alluxio"]
services-azfile = ["opendal/services-azfile"]
services-b2 = ["opendal/services-b2"]
-services-upyun = ["opendal/services-upyun"]
services-cacache = ["opendal/services-cacache"]
services-dashmap = ["opendal/services-dashmap"]
services-dropbox = ["opendal/services-dropbox"]
@@ -135,13 +134,14 @@ services-postgresql = ["opendal/services-postgresql"]
services-redb = ["opendal/services-redb"]
services-redis = ["opendal/services-redis"]
services-rocksdb = ["opendal/services-rocksdb"]
+services-seafile = ["opendal/services-seafile"]
services-sftp = ["opendal/services-sftp"]
services-sled = ["opendal/services-sled"]
services-sqlite = ["opendal/services-sqlite"]
-services-seafile = ["opendal/services-seafile"]
services-supabase = ["opendal/services-supabase"]
services-swift = ["opendal/services-swift"]
services-tikv = ["opendal/services-tikv"]
+services-upyun = ["opendal/services-upyun"]
services-vercel-artifacts = ["opendal/services-vercel-artifacts"]
services-wasabi = ["opendal/services-wasabi"]
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 0128db3724..11c92212a0 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -121,11 +121,8 @@ services-azdls = [
]
services-azfile = []
services-b2 = []
-services-seafile = []
-services-upyun = ["dep:hmac", "dep:sha1"]
-services-chainsafe = []
-services-pcloud = []
services-cacache = ["dep:cacache"]
+services-chainsafe = []
services-cloudflare-kv = []
services-cos = [
"dep:reqsign",
@@ -139,7 +136,7 @@ services-dropbox = []
services-etcd = ["dep:etcd-client", "dep:bb8"]
services-foundationdb = ["dep:foundationdb"]
services-fs = ["tokio/fs"]
-services-ftp = ["dep:suppaftp", "dep:bb8", "dep:async-tls" ]
+services-ftp = ["dep:suppaftp", "dep:bb8", "dep:async-tls"]
services-gcs = [
"dep:reqsign",
"reqsign?/services-google",
@@ -171,6 +168,7 @@ services-oss = [
"reqsign?/services-aliyun",
"reqsign?/reqwest_request",
]
+services-pcloud = []
services-persy = ["dep:persy"]
services-postgresql = ["dep:tokio-postgres", "dep:bb8", "dep:bb8-postgres"]
services-redb = ["dep:redb"]
@@ -182,12 +180,14 @@ services-s3 = [
"reqsign?/services-aws",
"reqsign?/reqwest_request",
]
+services-seafile = []
services-sftp = ["dep:openssh", "dep:openssh-sftp-client", "dep:dirs"]
services-sled = ["dep:sled"]
services-sqlite = ["dep:rusqlite", "dep:r2d2"]
services-supabase = []
services-swift = []
services-tikv = ["tikv-client"]
+services-upyun = ["dep:hmac", "dep:sha1"]
services-vercel-artifacts = []
# Deprecated
# wasabi services support has been removed.
@@ -219,6 +219,7 @@ required-features = ["tests"]
anyhow = { version = "1.0.30", features = ["std"] }
async-backtrace = { version = "0.2.6", optional = true }
async-compat = "0.2"
+async-tls = { version = "0.12.0", optional = true }
async-trait = "0.1.68"
atomic_lib = { version = "0.34.5", optional = true }
await-tree = { version = "0.1.1", optional = true }
@@ -246,6 +247,7 @@ foundationdb = { version = "0.8.0", features = [
futures = { version = "0.3", default-features = false, features = ["std"] }
governor = { version = "0.6.0", optional = true, features = ["std"] }
hdrs = { version = "0.3.0", optional = true, features = ["async_file"] }
+hmac = { version = "0.12.1", optional = true }
hrana-client-proto = { version = "0.2.1", optional = true }
http = "0.2.9"
log = "0.4"
@@ -290,9 +292,8 @@ rocksdb = { version = "0.21.0", default-features = false, optional = true }
rusqlite = { version = "0.29.0", optional = true, features = ["bundled"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
-sha2 = { version = "0.10", optional = true }
-hmac = { version = "0.12.1", optional = true }
sha1 = { version = "0.10.6", optional = true }
+sha2 = { version = "0.10", optional = true }
sled = { version = "0.34.7", optional = true }
suppaftp = { version = "5.2", default-features = false, features = [
"async-secure",
@@ -300,11 +301,10 @@ suppaftp = { version = "5.2", default-features = false, features = [
"async-rustls",
], optional = true }
tikv-client = { version = "0.3.0", optional = true, default-features = false }
-tokio = { version= "1.27",features = ["sync"] }
+tokio = { version = "1.27", features = ["sync"] }
tokio-postgres = { version = "0.7.8", optional = true }
tracing = { version = "0.1", optional = true }
uuid = { version = "1", features = ["serde", "v4"] }
-async-tls = { version = "0.12.0", optional = true }
[target.'cfg(target_arch = "wasm32")'.dependencies]
getrandom = { version = "0.2", features = ["js"] }
diff --git a/core/edge/s3_aws_assume_role_with_web_identity/Cargo.toml b/core/edge/s3_aws_assume_role_with_web_identity/Cargo.toml
index e7cf869a77..c4842534ea 100644
--- a/core/edge/s3_aws_assume_role_with_web_identity/Cargo.toml
+++ b/core/edge/s3_aws_assume_role_with_web_identity/Cargo.toml
@@ -25,5 +25,5 @@ license.workspace = true
[dependencies]
opendal = { workspace = true, features = ["tests"] }
-uuid = { version = "1", features = ["serde", "v4"] }
tokio = { version = "1", features = ["full"] }
+uuid = { version = "1", features = ["serde", "v4"] }
diff --git a/core/edge/s3_aws_assume_role_with_web_identity/src/main.rs b/core/edge/s3_aws_assume_role_with_web_identity/src/main.rs
index 19446d2d74..b5a5bc6488 100644
--- a/core/edge/s3_aws_assume_role_with_web_identity/src/main.rs
+++ b/core/edge/s3_aws_assume_role_with_web_identity/src/main.rs
@@ -16,7 +16,8 @@
// under the License.
use opendal::raw::tests::init_test_service;
-use opendal::{Result, Scheme};
+use opendal::Result;
+use opendal::Scheme;
#[tokio::main]
async fn main() -> Result<()> {
diff --git a/core/edge/s3_read_on_wasm/Cargo.toml b/core/edge/s3_read_on_wasm/Cargo.toml
index 5a37dc13ca..e479a0eee6 100644
--- a/core/edge/s3_read_on_wasm/Cargo.toml
+++ b/core/edge/s3_read_on_wasm/Cargo.toml
@@ -27,7 +27,9 @@ license.workspace = true
crate-type = ["cdylib"]
[dependencies]
-opendal = { path = "../../", default-features = false, features = ["services-s3"] }
+opendal = { path = "../../", default-features = false, features = [
+ "services-s3",
+] }
wasm-bindgen = "0.2.89"
wasm-bindgen-futures = "0.4.39"
diff --git a/core/edge/s3_read_on_wasm/src/lib.rs b/core/edge/s3_read_on_wasm/src/lib.rs
index 4e8ea7f36b..d0fadff2a5 100644
--- a/core/edge/s3_read_on_wasm/src/lib.rs
+++ b/core/edge/s3_read_on_wasm/src/lib.rs
@@ -43,10 +43,11 @@ pub async fn hello_world() -> String {
#[cfg(test)]
mod tests {
- use super::*;
use wasm_bindgen_test::wasm_bindgen_test;
use wasm_bindgen_test::wasm_bindgen_test_configure;
+ use super::*;
+
wasm_bindgen_test_configure!(run_in_browser);
#[wasm_bindgen_test]
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 69e1e7b83d..60dd2d2673 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -786,10 +786,10 @@ mod tests {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for MockService {
type Reader = oio::Reader;
- type BlockingReader = oio::BlockingReader;
type Writer = oio::Writer;
- type BlockingWriter = oio::BlockingWriter;
type Lister = oio::Lister;
+ type BlockingReader = oio::BlockingReader;
+ type BlockingWriter = oio::BlockingWriter;
type BlockingLister = oio::BlockingLister;
fn info(&self) -> AccessorInfo {
@@ -803,6 +803,10 @@ mod tests {
Ok(RpCreateDir {})
}
+ async fn stat(&self, _: &str, _: OpStat) -> Result<RpStat> {
+ Ok(RpStat::new(Metadata::new(EntryMode::Unknown)))
+ }
+
async fn read(&self, _: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
Ok((RpRead::new(), Box::new(oio::Cursor::new())))
}
@@ -811,18 +815,6 @@ mod tests {
Ok((RpWrite::new(), Box::new(())))
}
- async fn copy(&self, _: &str, _: &str, _: OpCopy) -> Result<RpCopy> {
- Ok(RpCopy {})
- }
-
- async fn rename(&self, _: &str, _: &str, _: OpRename) -> Result<RpRename> {
- Ok(RpRename {})
- }
-
- async fn stat(&self, _: &str, _: OpStat) -> Result<RpStat> {
- Ok(RpStat::new(Metadata::new(EntryMode::Unknown)))
- }
-
async fn delete(&self, _: &str, _: OpDelete) -> Result<RpDelete> {
Ok(RpDelete {})
}
@@ -831,6 +823,14 @@ mod tests {
Ok((RpList {}, Box::new(())))
}
+ async fn copy(&self, _: &str, _: &str, _: OpCopy) -> Result<RpCopy> {
+ Ok(RpCopy {})
+ }
+
+ async fn rename(&self, _: &str, _: &str, _: OpRename) -> Result<RpRename> {
+ Ok(RpRename {})
+ }
+
async fn presign(&self, _: &str, _: OpPresign) -> Result<RpPresign> {
Ok(RpPresign::new(PresignedRequest::new(
HttpMethod::POST,
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 4219f445a1..1093646bcd 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -1164,10 +1164,10 @@ mod tests {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for MockService {
type Reader = MockReader;
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = MockLister;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 26dfdcd039..1e658c2ad7 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -62,10 +62,10 @@
//! every operation that OpenDAL performs.
//!
//! ```no_run
+//! use opendal::layers::LoggingLayer;
//! use opendal::services;
//! use opendal::Operator;
//! use opendal::Result;
-//! use opendal::layers::LoggingLayer;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
@@ -96,10 +96,10 @@
//! - `reader_with`: Create a reader with advanced options.
//!
//! ```no_run
+//! use opendal::layers::LoggingLayer;
//! use opendal::services;
//! use opendal::Operator;
//! use opendal::Result;
-//! use opendal::layers::LoggingLayer;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
diff --git a/core/src/raw/accessor.rs b/core/src/raw/accessor.rs
index 4d78b153a5..f74f3ac2ae 100644
--- a/core/src/raw/accessor.rs
+++ b/core/src/raw/accessor.rs
@@ -56,20 +56,18 @@ use crate::*;
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
- /// Reader is the associated reader the could return in `read` operation.
+ /// Reader is the associated reader returned in `read` operation.
type Reader: oio::Read;
- /// BlockingReader is the associated reader that could return in
- /// `blocking_read` operation.
- type BlockingReader: oio::BlockingRead;
- /// Writer is the associated writer the could return in `write` operation.
+ /// Writer is the associated writer returned in `write` operation.
type Writer: oio::Write;
- /// BlockingWriter is the associated writer the could return in
- /// `blocking_write` operation.
- type BlockingWriter: oio::BlockingWrite;
- /// Lister is the associated lister that return in `list` operation.
+ /// Lister is the associated lister returned in `list` operation.
type Lister: oio::List;
- /// BlockingLister is the associated lister that could return in
- /// `blocking_list` operation.
+
+ /// BlockingReader is the associated reader returned `blocking_read` operation.
+ type BlockingReader: oio::BlockingRead;
+ /// BlockingWriter is the associated writer returned `blocking_write` operation.
+ type BlockingWriter: oio::BlockingWrite;
+ /// BlockingLister is the associated lister returned `blocking_list` operation.
type BlockingLister: oio::BlockingList;
/// Invoke the `info` operation to get metadata of accessor.
@@ -102,6 +100,24 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
+ /// Invoke the `stat` operation on the specified path.
+ ///
+ /// Require [`Capability::stat`]
+ ///
+ /// # Behavior
+ ///
+ /// - `stat` empty path means stat backend's root path.
+ /// - `stat` a path endswith "/" means stating a dir.
+ /// - `mode` and `content_length` must be set.
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let (_, _) = (path, args);
+
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
/// Invoke the `read` operation on the specified path, returns a
/// [`Reader`][crate::Reader] if operate successful.
///
@@ -137,29 +153,16 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `copy` operation on the specified `from` path and `to` path.
- ///
- /// Require [Capability::copy]
+ /// Invoke the `delete` operation on the specified path.
///
- /// # Behaviour
+ /// Require [`Capability::delete`]
///
- /// - `from` and `to` MUST be file path, DON'T NEED to check mode.
- /// - Copy on existing file SHOULD succeed.
- /// - Copy on existing file SHOULD overwrite and truncate.
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let (_, _, _) = (from, to, args);
-
- Err(Error::new(
- ErrorKind::Unsupported,
- "operation is not supported",
- ))
- }
-
- /// Invoke the `rename` operation on the specified `from` path and `to` path.
+ /// # Behavior
///
- /// Require [Capability::rename]
- async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let (_, _, _) = (from, to, args);
+ /// - `delete` is an idempotent operation, it's safe to call `Delete` on the same path multiple times.
+ /// - `delete` SHOULD return `Ok(())` if the path is deleted successfully or not exist.
+ async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ let (_, _) = (path, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -167,16 +170,15 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `stat` operation on the specified path.
+ /// Invoke the `list` operation on the specified path.
///
- /// Require [`Capability::stat`]
+ /// Require [`Capability::list`]
///
/// # Behavior
///
- /// - `stat` empty path means stat backend's root path.
- /// - `stat` a path endswith "/" means stating a dir.
- /// - `mode` and `content_length` must be set.
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ /// - Input path MUST be dir path, DON'T NEED to check mode.
+ /// - List non-exist dir should return Empty.
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let (_, _) = (path, args);
Err(Error::new(
@@ -185,16 +187,17 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `delete` operation on the specified path.
+ /// Invoke the `copy` operation on the specified `from` path and `to` path.
///
- /// Require [`Capability::delete`]
+ /// Require [Capability::copy]
///
- /// # Behavior
+ /// # Behaviour
///
- /// - `delete` is an idempotent operation, it's safe to call `Delete` on the same path multiple times.
- /// - `delete` SHOULD return `Ok(())` if the path is deleted successfully or not exist.
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let (_, _) = (path, args);
+ /// - `from` and `to` MUST be file path, DON'T NEED to check mode.
+ /// - Copy on existing file SHOULD succeed.
+ /// - Copy on existing file SHOULD overwrite and truncate.
+ async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ let (_, _, _) = (from, to, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -202,16 +205,11 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `list` operation on the specified path.
- ///
- /// Require [`Capability::list`]
- ///
- /// # Behavior
+ /// Invoke the `rename` operation on the specified `from` path and `to` path.
///
- /// - Input path MUST be dir path, DON'T NEED to check mode.
- /// - List non-exist dir should return Empty.
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let (_, _) = (path, args);
+ /// Require [Capability::rename]
+ async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ let (_, _, _) = (from, to, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -261,6 +259,20 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
+ /// Invoke the `blocking_stat` operation on the specified path.
+ ///
+ /// This operation is the blocking version of [`Accessor::stat`]
+ ///
+ /// Require [`Capability::stat`] and [`Capability::blocking`]
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let (_, _) = (path, args);
+
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
/// Invoke the `blocking_read` operation on the specified path.
///
/// This operation is the blocking version of [`Accessor::read`]
@@ -289,13 +301,13 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `blocking_copy` operation on the specified `from` path and `to` path.
+ /// Invoke the `blocking_delete` operation on the specified path.
///
- /// This operation is the blocking version of [`Accessor::copy`]
+ /// This operation is the blocking version of [`Accessor::delete`]
///
- /// Require [`Capability::copy`] and [`Capability::blocking`]
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let (_, _, _) = (from, to, args);
+ /// Require [`Capability::write`] and [`Capability::blocking`]
+ fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ let (_, _) = (path, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -303,26 +315,16 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `blocking_rename` operation on the specified `from` path and `to` path.
+ /// Invoke the `blocking_list` operation on the specified path.
///
- /// This operation is the blocking version of [`Accessor::rename`]
+ /// This operation is the blocking version of [`Accessor::list`]
///
- /// Require [`Capability::rename`] and [`Capability::blocking`]
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let (_, _, _) = (from, to, args);
-
- Err(Error::new(
- ErrorKind::Unsupported,
- "operation is not supported",
- ))
- }
-
- /// Invoke the `blocking_stat` operation on the specified path.
+ /// Require [`Capability::list`] and [`Capability::blocking`]
///
- /// This operation is the blocking version of [`Accessor::stat`]
+ /// # Behavior
///
- /// Require [`Capability::stat`] and [`Capability::blocking`]
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ /// - List non-exist dir should return Empty.
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
let (_, _) = (path, args);
Err(Error::new(
@@ -331,13 +333,13 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `blocking_delete` operation on the specified path.
+ /// Invoke the `blocking_copy` operation on the specified `from` path and `to` path.
///
- /// This operation is the blocking version of [`Accessor::delete`]
+ /// This operation is the blocking version of [`Accessor::copy`]
///
- /// Require [`Capability::write`] and [`Capability::blocking`]
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let (_, _) = (path, args);
+ /// Require [`Capability::copy`] and [`Capability::blocking`]
+ fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ let (_, _, _) = (from, to, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -345,17 +347,13 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
))
}
- /// Invoke the `blocking_list` operation on the specified path.
- ///
- /// This operation is the blocking version of [`Accessor::list`]
- ///
- /// Require [`Capability::list`] and [`Capability::blocking`]
+ /// Invoke the `blocking_rename` operation on the specified `from` path and `to` path.
///
- /// # Behavior
+ /// This operation is the blocking version of [`Accessor::rename`]
///
- /// - List non-exist dir should return Empty.
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let (_, _) = (path, args);
+ /// Require [`Capability::rename`] and [`Capability::blocking`]
+ fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ let (_, _, _) = (from, to, args);
Err(Error::new(
ErrorKind::Unsupported,
@@ -369,10 +367,10 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for () {
type Reader = ();
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -392,10 +390,10 @@ impl Accessor for () {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<T: Accessor + ?Sized> Accessor for Arc<T> {
type Reader = T::Reader;
- type BlockingReader = T::BlockingReader;
type Writer = T::Writer;
- type BlockingWriter = T::BlockingWriter;
type Lister = T::Lister;
+ type BlockingReader = T::BlockingReader;
+ type BlockingWriter = T::BlockingWriter;
type BlockingLister = T::BlockingLister;
fn info(&self) -> AccessorInfo {
@@ -406,66 +404,66 @@ impl<T: Accessor + ?Sized> Accessor for Arc<T> {
self.as_ref().create_dir(path, args).await
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.as_ref().stat(path, args).await
+ }
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.as_ref().read(path, args).await
}
+
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.as_ref().write(path, args).await
}
+ async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.as_ref().delete(path, args).await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.as_ref().list(path, args).await
+ }
async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
self.as_ref().copy(from, to, args).await
}
-
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
self.as_ref().rename(from, to, args).await
}
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.as_ref().stat(path, args).await
- }
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.as_ref().delete(path, args).await
- }
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- self.as_ref().list(path, args).await
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ self.as_ref().presign(path, args).await
}
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
self.as_ref().batch(args).await
}
- async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
- self.as_ref().presign(path, args).await
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.as_ref().blocking_create_dir(path, args)
}
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.as_ref().blocking_stat(path, args)
+ }
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
self.as_ref().blocking_read(path, args)
}
+
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
self.as_ref().blocking_write(path, args)
}
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.as_ref().blocking_copy(from, to, args)
- }
-
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- self.as_ref().blocking_rename(from, to, args)
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.as_ref().blocking_stat(path, args)
- }
fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
self.as_ref().blocking_delete(path, args)
}
+
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.as_ref().blocking_list(path, args)
}
+ fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.as_ref().blocking_copy(from, to, args)
+ }
+ fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.as_ref().blocking_rename(from, to, args)
+ }
}
/// FusedAccessor is the type erased accessor with `Arc<dyn Accessor>`.
diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs
index f2b2625b00..56b32a9de0 100644
--- a/core/src/raw/enum_utils.rs
+++ b/core/src/raw/enum_utils.rs
@@ -38,9 +38,11 @@
//! This module is used to provide some enums for the above code. We should remove this module once
//! type_alias_impl_trait has been stabilized.
-use bytes::Bytes;
use std::io::SeekFrom;
-use std::task::{Context, Poll};
+use std::task::Context;
+use std::task::Poll;
+
+use bytes::Bytes;
use crate::raw::*;
use crate::*;
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index 318abe4ec7..08583fe6d8 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -15,12 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use futures::stream::FuturesOrdered;
-use futures::{FutureExt, StreamExt};
use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::Context;
+use std::task::Poll;
+
+use futures::stream::FuturesOrdered;
+use futures::FutureExt;
+use futures::StreamExt;
/// BoxedFuture is the type alias of [`futures::future::BoxFuture`].
///
@@ -189,12 +192,14 @@ where
#[cfg(test)]
mod tests {
- use super::*;
+ use std::task::ready;
+ use std::time::Duration;
+
use futures::future::BoxFuture;
use futures::Stream;
use rand::Rng;
- use std::task::ready;
- use std::time::Duration;
+
+ use super::*;
struct Lister {
size: usize,
diff --git a/core/src/raw/http_util/header.rs b/core/src/raw/http_util/header.rs
index ec28113a70..87748c96f4 100644
--- a/core/src/raw/http_util/header.rs
+++ b/core/src/raw/http_util/header.rs
@@ -19,16 +19,18 @@ use base64::engine::general_purpose;
use base64::Engine;
use chrono::DateTime;
use chrono::Utc;
+use http::header::CACHE_CONTROL;
use http::header::CONTENT_DISPOSITION;
+use http::header::CONTENT_ENCODING;
use http::header::CONTENT_LENGTH;
use http::header::CONTENT_RANGE;
use http::header::CONTENT_TYPE;
use http::header::ETAG;
use http::header::LAST_MODIFIED;
use http::header::LOCATION;
-use http::header::{CACHE_CONTROL, CONTENT_ENCODING};
+use http::HeaderMap;
+use http::HeaderName;
use http::HeaderValue;
-use http::{HeaderMap, HeaderName};
use md5::Digest;
use crate::raw::*;
diff --git a/core/src/raw/oio/read/buffer_reader.rs b/core/src/raw/oio/read/buffer_reader.rs
index 449f93e010..3d96b93c12 100644
--- a/core/src/raw/oio/read/buffer_reader.rs
+++ b/core/src/raw/oio/read/buffer_reader.rs
@@ -15,21 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-use bytes::BufMut;
-use bytes::Bytes;
-use tokio::io::ReadBuf;
-
use std::cmp::min;
use std::io::SeekFrom;
-
use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use crate::raw::*;
-use crate::*;
+use bytes::BufMut;
+use bytes::Bytes;
+use tokio::io::ReadBuf;
use super::BlockingRead;
+use crate::raw::*;
+use crate::*;
/// [BufferReader] allows the underlying reader to fetch data at the buffer's size
/// and is used to amortize the IO's overhead.
@@ -316,7 +314,6 @@ mod tests {
use std::io::SeekFrom;
use std::sync::Arc;
- use crate::raw::oio::RangeReader;
use async_trait::async_trait;
use bytes::Bytes;
use futures::AsyncReadExt;
@@ -326,6 +323,7 @@ mod tests {
use sha2::Sha256;
use super::*;
+ use crate::raw::oio::RangeReader;
// Generate bytes between [4MiB, 16MiB)
fn gen_bytes() -> (Bytes, usize) {
@@ -352,10 +350,10 @@ mod tests {
#[async_trait]
impl Accessor for MockReadService {
type Reader = MockReader;
- type BlockingReader = MockReader;
type Writer = ();
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = MockReader;
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
diff --git a/core/src/raw/oio/read/range_read.rs b/core/src/raw/oio/read/range_read.rs
index 4b46f47ac6..5ffacf2ec8 100644
--- a/core/src/raw/oio/read/range_read.rs
+++ b/core/src/raw/oio/read/range_read.rs
@@ -672,10 +672,10 @@ mod tests {
#[async_trait]
impl Accessor for MockReadService {
type Reader = MockReader;
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
diff --git a/core/src/raw/oio/stream/into_stream.rs b/core/src/raw/oio/stream/into_stream.rs
index d5a57970c7..3e34bbf4ba 100644
--- a/core/src/raw/oio/stream/into_stream.rs
+++ b/core/src/raw/oio/stream/into_stream.rs
@@ -20,10 +20,13 @@ pub use non_wasm32_impl::*;
#[cfg(not(target_arch = "wasm32"))]
mod non_wasm32_impl {
- use crate::raw::oio;
+ use std::task::Context;
+ use std::task::Poll;
+
use bytes::Bytes;
use futures::TryStreamExt;
- use std::task::{Context, Poll};
+
+ use crate::raw::oio;
/// Convert given futures stream into [`oio::Stream`].
pub fn into_stream<S>(stream: S) -> IntoStream<S>
@@ -51,10 +54,13 @@ mod non_wasm32_impl {
pub use wasm32_impl::*;
#[cfg(target_arch = "wasm32")]
mod wasm32_impl {
- use crate::raw::oio;
+ use std::task::Context;
+ use std::task::Poll;
+
use bytes::Bytes;
use futures::TryStreamExt;
- use std::task::{Context, Poll};
+
+ use crate::raw::oio;
/// Convert given futures stream into [`oio::Stream`].
pub fn into_stream<S>(stream: S) -> IntoStream<S>
diff --git a/core/src/services/alluxio/backend.rs b/core/src/services/alluxio/backend.rs
index 078a871da5..d14d82f085 100644
--- a/core/src/services/alluxio/backend.rs
+++ b/core/src/services/alluxio/backend.rs
@@ -182,10 +182,10 @@ pub struct AlluxioBackend {
#[async_trait]
impl Accessor for AlluxioBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = AlluxioWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<AlluxioLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -216,6 +216,12 @@ impl Accessor for AlluxioBackend {
Ok(RpCreateDir::default())
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let file_info = self.core.get_status(path).await?;
+
+ Ok(RpStat::new(file_info.try_into()?))
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let stream_id = self.core.open_file(path).await?;
@@ -231,18 +237,6 @@ impl Accessor for AlluxioBackend {
Ok((RpWrite::default(), w))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let file_info = self.core.get_status(path).await?;
-
- Ok(RpStat::new(file_info.try_into()?))
- }
-
- async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
- self.core.rename(from, to).await?;
-
- Ok(RpRename::default())
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
self.core.delete(path).await?;
@@ -253,6 +247,12 @@ impl Accessor for AlluxioBackend {
let l = AlluxioLister::new(self.core.clone(), path);
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
+ self.core.rename(from, to).await?;
+
+ Ok(RpRename::default())
+ }
}
#[cfg(test)]
diff --git a/core/src/services/atomicserver/backend.rs b/core/src/services/atomicserver/backend.rs
index 2ad7aa42a2..0742cd9301 100644
--- a/core/src/services/atomicserver/backend.rs
+++ b/core/src/services/atomicserver/backend.rs
@@ -31,16 +31,17 @@ use serde::Deserialize;
use serde::Serialize;
use crate::raw::adapters::kv;
+use crate::raw::new_json_deserialize_error;
use crate::raw::new_json_serialize_error;
use crate::raw::new_request_build_error;
use crate::raw::normalize_path;
use crate::raw::normalize_root;
use crate::raw::percent_encode_path;
use crate::raw::AsyncBody;
+use crate::raw::ConfigDeserializer;
use crate::raw::FormDataPart;
use crate::raw::HttpClient;
use crate::raw::Multipart;
-use crate::raw::{new_json_deserialize_error, ConfigDeserializer};
use crate::Builder;
use crate::Scheme;
use crate::*;
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index d7ae49344c..3a38c9977c 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -543,10 +543,10 @@ pub struct AzblobBackend {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for AzblobBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = AzblobWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<AzblobLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -593,6 +593,17 @@ impl Accessor for AzblobBackend {
am
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let resp = self.core.azblob_get_blob_properties(path, &args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.azblob_get_blob(path, &args).await?;
@@ -626,31 +637,6 @@ impl Accessor for AzblobBackend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.azblob_copy_blob(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::ACCEPTED => {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let resp = self.core.azblob_get_blob_properties(path, &args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.azblob_delete_blob(path).await?;
@@ -673,6 +659,20 @@ impl Accessor for AzblobBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.azblob_copy_blob(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::ACCEPTED => {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let mut req = match args.operation() {
PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v)?,
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index 77240e72e7..53fa53d496 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -230,10 +230,10 @@ pub struct AzdlsBackend {
#[async_trait]
impl Accessor for AzdlsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = AzdlsWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<AzdlsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -285,62 +285,6 @@ impl Accessor for AzdlsBackend {
}
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.core.azdls_read(path, args.range()).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
- let size = parse_content_length(resp.headers())?;
- let range = parse_content_range(resp.headers())?;
- Ok((
- RpRead::new().with_size(size).with_range(range),
- resp.into_body(),
- ))
- }
- StatusCode::RANGE_NOT_SATISFIABLE => {
- resp.into_body().consume().await?;
- Ok((RpRead::new().with_size(Some(0)), IncomingAsyncBody::empty()))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let w = AzdlsWriter::new(self.core.clone(), args.clone(), path.to_string());
- let w = if args.append() {
- AzdlsWriters::Two(oio::AppendObjectWriter::new(w))
- } else {
- AzdlsWriters::One(oio::OneShotWriter::new(w))
- };
- Ok((RpWrite::default(), w))
- }
-
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- if let Some(resp) = self.core.azdls_ensure_parent_path(to).await? {
- let status = resp.status();
- match status {
- StatusCode::CREATED | StatusCode::CONFLICT => {
- resp.into_body().consume().await?;
- }
- _ => return Err(parse_error(resp).await?),
- }
- }
-
- let resp = self.core.azdls_rename(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::CREATED => {
- resp.into_body().consume().await?;
- Ok(RpRename::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
// Stat root always returns a DIR.
if path == "/" {
@@ -387,6 +331,38 @@ impl Accessor for AzdlsBackend {
Ok(RpStat::new(meta))
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.azdls_read(path, args.range()).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ let range = parse_content_range(resp.headers())?;
+ Ok((
+ RpRead::new().with_size(size).with_range(range),
+ resp.into_body(),
+ ))
+ }
+ StatusCode::RANGE_NOT_SATISFIABLE => {
+ resp.into_body().consume().await?;
+ Ok((RpRead::new().with_size(Some(0)), IncomingAsyncBody::empty()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let w = AzdlsWriter::new(self.core.clone(), args.clone(), path.to_string());
+ let w = if args.append() {
+ AzdlsWriters::Two(oio::AppendObjectWriter::new(w))
+ } else {
+ AzdlsWriters::One(oio::OneShotWriter::new(w))
+ };
+ Ok((RpWrite::default(), w))
+ }
+
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.azdls_delete(path).await?;
@@ -403,6 +379,30 @@ impl Accessor for AzdlsBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ if let Some(resp) = self.core.azdls_ensure_parent_path(to).await? {
+ let status = resp.status();
+ match status {
+ StatusCode::CREATED | StatusCode::CONFLICT => {
+ resp.into_body().consume().await?;
+ }
+ _ => return Err(parse_error(resp).await?),
+ }
+ }
+
+ let resp = self.core.azdls_rename(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::CREATED => {
+ resp.into_body().consume().await?;
+ Ok(RpRename::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
fn infer_storage_name_from_endpoint(endpoint: &str) -> Option<String> {
diff --git a/core/src/services/azfile/backend.rs b/core/src/services/azfile/backend.rs
index 375b52af6c..dea84c73aa 100644
--- a/core/src/services/azfile/backend.rs
+++ b/core/src/services/azfile/backend.rs
@@ -247,10 +247,10 @@ pub struct AzfileBackend {
#[async_trait]
impl Accessor for AzfileBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = AzfileWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<AzfileLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -308,6 +308,23 @@ impl Accessor for AzfileBackend {
}
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let resp = if path.ends_with('/') {
+ self.core.azfile_get_directory_properties(path).await?
+ } else {
+ self.core.azfile_get_file_properties(path).await?
+ };
+
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let meta = parse_into_metadata(path, resp.headers())?;
+ Ok(RpStat::new(meta))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.azfile_read(path, args.range()).await?;
@@ -341,23 +358,29 @@ impl Accessor for AzfileBackend {
return Ok((RpWrite::default(), w));
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = if path.ends_with('/') {
- self.core.azfile_get_directory_properties(path).await?
+ self.core.azfile_delete_dir(path).await?
} else {
- self.core.azfile_get_file_properties(path).await?
+ self.core.azfile_delete_file(path).await?
};
let status = resp.status();
match status {
- StatusCode::OK => {
- let meta = parse_into_metadata(path, resp.headers())?;
- Ok(RpStat::new(meta))
+ StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
+ resp.into_body().consume().await?;
+ Ok(RpDelete::default())
}
_ => Err(parse_error(resp).await?),
}
}
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ let l = AzfileLister::new(self.core.clone(), path.to_string(), args.limit());
+
+ Ok((RpList::default(), oio::PageLister::new(l)))
+ }
+
async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
self.core.ensure_parent_dir_exists(to).await?;
let resp = self.core.azfile_rename(from, to).await?;
@@ -370,29 +393,6 @@ impl Accessor for AzfileBackend {
_ => Err(parse_error(resp).await?),
}
}
-
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = if path.ends_with('/') {
- self.core.azfile_delete_dir(path).await?
- } else {
- self.core.azfile_delete_file(path).await?
- };
-
- let status = resp.status();
- match status {
- StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
- resp.into_body().consume().await?;
- Ok(RpDelete::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let l = AzfileLister::new(self.core.clone(), path.to_string(), args.limit());
-
- Ok((RpList::default(), oio::PageLister::new(l)))
- }
}
#[cfg(test)]
diff --git a/core/src/services/b2/backend.rs b/core/src/services/b2/backend.rs
index 1502afe0a6..a488d60beb 100644
--- a/core/src/services/b2/backend.rs
+++ b/core/src/services/b2/backend.rs
@@ -269,14 +269,14 @@ pub struct B2Backend {
impl Accessor for B2Backend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
-
type Writer = B2Writers;
- type BlockingWriter = ();
-
type Lister = oio::PageLister<B2Lister>;
+ type BlockingReader = ();
+
+ type BlockingWriter = ();
+
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -326,6 +326,38 @@ impl Accessor for B2Backend {
am
}
+ /// B2 have a get_file_info api required a file_id field, but field_id need call list api, list api also return file info
+ /// So we call list api to get file info
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ // Stat root always returns a DIR.
+ if path == "/" {
+ return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
+ }
+
+ let delimiter = if path.ends_with('/') { Some("/") } else { None };
+ let resp = self
+ .core
+ .list_file_names(Some(path), delimiter, None, None)
+ .await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+
+ let resp: ListFileNamesResponse =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+ if resp.files.is_empty() {
+ return Err(Error::new(ErrorKind::NotFound, "no such file or directory"));
+ }
+ let meta = parse_file_info(&resp.files[0]);
+ Ok(RpStat::new(meta))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.download_file_by_name(path, &args).await?;
@@ -356,38 +388,38 @@ impl Accessor for B2Backend {
Ok((RpWrite::default(), w))
}
- /// B2 have a get_file_info api required a file_id field, but field_id need call list api, list api also return file info
- /// So we call list api to get file info
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- // Stat root always returns a DIR.
- if path == "/" {
- return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
- }
-
- let delimiter = if path.ends_with('/') { Some("/") } else { None };
- let resp = self
- .core
- .list_file_names(Some(path), delimiter, None, None)
- .await?;
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let resp = self.core.hide_file(path).await?;
let status = resp.status();
match status {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
-
- let resp: ListFileNamesResponse =
- serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
- if resp.files.is_empty() {
- return Err(Error::new(ErrorKind::NotFound, "no such file or directory"));
+ StatusCode::OK => Ok(RpDelete::default()),
+ _ => {
+ let err = parse_error(resp).await?;
+ match err.kind() {
+ ErrorKind::NotFound => Ok(RpDelete::default()),
+ // Representative deleted
+ ErrorKind::AlreadyExists => Ok(RpDelete::default()),
+ _ => Err(err),
}
- let meta = parse_file_info(&resp.files[0]);
- Ok(RpStat::new(meta))
}
- _ => Err(parse_error(resp).await?),
}
}
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ Ok((
+ RpList::default(),
+ oio::PageLister::new(B2Lister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ )),
+ ))
+ }
+
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
let resp = self
.core
@@ -426,38 +458,6 @@ impl Accessor for B2Backend {
}
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = self.core.hide_file(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => Ok(RpDelete::default()),
- _ => {
- let err = parse_error(resp).await?;
- match err.kind() {
- ErrorKind::NotFound => Ok(RpDelete::default()),
- // Representative deleted
- ErrorKind::AlreadyExists => Ok(RpDelete::default()),
- _ => Err(err),
- }
- }
- }
- }
-
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- Ok((
- RpList::default(),
- oio::PageLister::new(B2Lister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- )),
- ))
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
match args.operation() {
PresignOperation::Stat(_) => {
diff --git a/core/src/services/chainsafe/backend.rs b/core/src/services/chainsafe/backend.rs
index 17a156a688..3957f4f900 100644
--- a/core/src/services/chainsafe/backend.rs
+++ b/core/src/services/chainsafe/backend.rs
@@ -25,9 +25,6 @@ use http::StatusCode;
use log::debug;
use serde::Deserialize;
-use crate::raw::*;
-use crate::*;
-
use super::core::parse_info;
use super::core::ChainsafeCore;
use super::core::ObjectInfoResponse;
@@ -35,6 +32,8 @@ use super::error::parse_error;
use super::lister::ChainsafeLister;
use super::writer::ChainsafeWriter;
use super::writer::ChainsafeWriters;
+use crate::raw::*;
+use crate::*;
/// Config for backblaze Chainsafe services support.
#[derive(Default, Deserialize)]
@@ -206,14 +205,14 @@ pub struct ChainsafeBackend {
impl Accessor for ChainsafeBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
-
type Writer = ChainsafeWriters;
- type BlockingWriter = ();
-
type Lister = oio::PageLister<ChainsafeLister>;
+ type BlockingReader = ();
+
+ type BlockingWriter = ();
+
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -252,36 +251,36 @@ impl Accessor for ChainsafeBackend {
}
}
- async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.core.download_object(path).await?;
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.object_info(path).await?;
let status = resp.status();
match status {
StatusCode::OK => {
- let size = parse_content_length(resp.headers())?;
- let range = parse_content_range(resp.headers())?;
- Ok((
- RpRead::new().with_size(size).with_range(range),
- resp.into_body(),
- ))
+ let bs = resp.into_body().bytes().await?;
+
+ let output: ObjectInfoResponse =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+ Ok(RpStat::new(parse_info(output.content)))
}
_ => Err(parse_error(resp).await?),
}
}
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let resp = self.core.object_info(path).await?;
+ async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.download_object(path).await?;
let status = resp.status();
match status {
StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
-
- let output: ObjectInfoResponse =
- serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
- Ok(RpStat::new(parse_info(output.content)))
+ let size = parse_content_length(resp.headers())?;
+ let range = parse_content_range(resp.headers())?;
+ Ok((
+ RpRead::new().with_size(size).with_range(range),
+ resp.into_body(),
+ ))
}
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/chainsafe/core.rs b/core/src/services/chainsafe/core.rs
index 792cc22d0d..50e3e37069 100644
--- a/core/src/services/chainsafe/core.rs
+++ b/core/src/services/chainsafe/core.rs
@@ -22,7 +22,6 @@ use bytes::Bytes;
use http::header;
use http::Request;
use http::Response;
-
use serde::Deserialize;
use serde_json::json;
diff --git a/core/src/services/chainsafe/lister.rs b/core/src/services/chainsafe/lister.rs
index 997fce21f8..69cdc55200 100644
--- a/core/src/services/chainsafe/lister.rs
+++ b/core/src/services/chainsafe/lister.rs
@@ -20,7 +20,9 @@ use std::sync::Arc;
use async_trait::async_trait;
use http::StatusCode;
-use super::core::{parse_info, ChainsafeCore, Info};
+use super::core::parse_info;
+use super::core::ChainsafeCore;
+use super::core::Info;
use super::error::parse_error;
use crate::raw::oio::Entry;
use crate::raw::*;
diff --git a/core/src/services/chainsafe/writer.rs b/core/src/services/chainsafe/writer.rs
index 27158378e0..456e6ba71c 100644
--- a/core/src/services/chainsafe/writer.rs
+++ b/core/src/services/chainsafe/writer.rs
@@ -20,11 +20,10 @@ use std::sync::Arc;
use async_trait::async_trait;
use http::StatusCode;
-use crate::raw::*;
-use crate::*;
-
use super::core::ChainsafeCore;
use super::error::parse_error;
+use crate::raw::*;
+use crate::*;
pub type ChainsafeWriters = oio::OneShotWriter<ChainsafeWriter>;
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
index a31e9818d2..5c6d458858 100644
--- a/core/src/services/cos/backend.rs
+++ b/core/src/services/cos/backend.rs
@@ -242,10 +242,10 @@ pub struct CosBackend {
#[async_trait]
impl Accessor for CosBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = CosWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<CosLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -301,6 +301,17 @@ impl Accessor for CosBackend {
am
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let resp = self.core.cos_head_object(path, &args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.cos_get_object(path, &args).await?;
@@ -335,39 +346,33 @@ impl Accessor for CosBackend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.cos_copy_object(from, to).await?;
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let resp = self.core.cos_delete_object(path).await?;
let status = resp.status();
match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
+ StatusCode::NO_CONTENT | StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
+ Ok(RpDelete::default())
}
_ => Err(parse_error(resp).await?),
}
}
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let resp = self.core.cos_head_object(path, &args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- _ => Err(parse_error(resp).await?),
- }
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ let l = CosLister::new(self.core.clone(), path, args.recursive(), args.limit());
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = self.core.cos_delete_object(path).await?;
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.cos_copy_object(from, to).await?;
let status = resp.status();
match status {
- StatusCode::NO_CONTENT | StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
- Ok(RpDelete::default())
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
}
_ => Err(parse_error(resp).await?),
}
@@ -393,9 +398,4 @@ impl Accessor for CosBackend {
parts.headers,
)))
}
-
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let l = CosLister::new(self.core.clone(), path, args.recursive(), args.limit());
- Ok((RpList::default(), oio::PageLister::new(l)))
- }
}
diff --git a/core/src/services/dbfs/backend.rs b/core/src/services/dbfs/backend.rs
index 6a5fcb602d..260a5abec3 100644
--- a/core/src/services/dbfs/backend.rs
+++ b/core/src/services/dbfs/backend.rs
@@ -154,10 +154,10 @@ pub struct DbfsBackend {
#[async_trait]
impl Accessor for DbfsBackend {
type Reader = DbfsReader;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<DbfsWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<DbfsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -197,35 +197,6 @@ impl Accessor for DbfsBackend {
}
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let op = DbfsReader::new(self.core.clone(), args, path.to_string());
-
- Ok((RpRead::new(), op))
- }
-
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- Ok((
- RpWrite::default(),
- oio::OneShotWriter::new(DbfsWriter::new(self.core.clone(), args, path.to_string())),
- ))
- }
-
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- self.core.dbfs_ensure_parent_path(to).await?;
-
- let resp = self.core.dbfs_rename(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
- Ok(RpRename::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
// Stat root always returns a DIR.
if path == "/" {
@@ -261,6 +232,19 @@ impl Accessor for DbfsBackend {
}
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let op = DbfsReader::new(self.core.clone(), args, path.to_string());
+
+ Ok((RpRead::new(), op))
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ Ok((
+ RpWrite::default(),
+ oio::OneShotWriter::new(DbfsWriter::new(self.core.clone(), args, path.to_string())),
+ ))
+ }
+
/// NOTE: Server will return 200 even if the path doesn't exist.
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.dbfs_delete(path).await?;
@@ -278,6 +262,22 @@ impl Accessor for DbfsBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ self.core.dbfs_ensure_parent_path(to).await?;
+
+ let resp = self.core.dbfs_rename(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(RpRename::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
#[derive(Deserialize)]
diff --git a/core/src/services/dropbox/backend.rs b/core/src/services/dropbox/backend.rs
index 9774deed77..5d05aaf5f2 100644
--- a/core/src/services/dropbox/backend.rs
+++ b/core/src/services/dropbox/backend.rs
@@ -36,10 +36,10 @@ pub struct DropboxBackend {
#[async_trait]
impl Accessor for DropboxBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<DropboxWriter>;
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -97,6 +97,44 @@ impl Accessor for DropboxBackend {
Ok(res)
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let resp = self.core.dropbox_get_metadata(path).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let bytes = resp.into_body().bytes().await?;
+ let decoded_response = serde_json::from_slice::<DropboxMetadataResponse>(&bytes)
+ .map_err(new_json_deserialize_error)?;
+ let entry_mode: EntryMode = match decoded_response.tag.as_str() {
+ "file" => EntryMode::FILE,
+ "folder" => EntryMode::DIR,
+ _ => EntryMode::Unknown,
+ };
+
+ let mut metadata = Metadata::new(entry_mode);
+ // Only set last_modified and size if entry_mode is FILE, because Dropbox API
+ // returns last_modified and size only for files.
+ // FYI: https://www.dropbox.com/developers/documentation/http/documentation#files-get_metadata
+ if entry_mode == EntryMode::FILE {
+ let date_utc_last_modified =
+ parse_datetime_from_rfc3339(&decoded_response.client_modified)?;
+ metadata.set_last_modified(date_utc_last_modified);
+
+ if let Some(size) = decoded_response.size {
+ metadata.set_content_length(size);
+ } else {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ &format!("no size found for file {}", path),
+ ));
+ }
+ }
+ Ok(RpStat::new(metadata))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.dropbox_get(path, args).await?;
let status = resp.status();
@@ -138,44 +176,6 @@ impl Accessor for DropboxBackend {
}
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let resp = self.core.dropbox_get_metadata(path).await?;
- let status = resp.status();
- match status {
- StatusCode::OK => {
- let bytes = resp.into_body().bytes().await?;
- let decoded_response = serde_json::from_slice::<DropboxMetadataResponse>(&bytes)
- .map_err(new_json_deserialize_error)?;
- let entry_mode: EntryMode = match decoded_response.tag.as_str() {
- "file" => EntryMode::FILE,
- "folder" => EntryMode::DIR,
- _ => EntryMode::Unknown,
- };
-
- let mut metadata = Metadata::new(entry_mode);
- // Only set last_modified and size if entry_mode is FILE, because Dropbox API
- // returns last_modified and size only for files.
- // FYI: https://www.dropbox.com/developers/documentation/http/documentation#files-get_metadata
- if entry_mode == EntryMode::FILE {
- let date_utc_last_modified =
- parse_datetime_from_rfc3339(&decoded_response.client_modified)?;
- metadata.set_last_modified(date_utc_last_modified);
-
- if let Some(size) = decoded_response.size {
- metadata.set_content_length(size);
- } else {
- return Err(Error::new(
- ErrorKind::Unexpected,
- &format!("no size found for file {}", path),
- ));
- }
- }
- Ok(RpStat::new(metadata))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
if ops.len() > 1000 {
diff --git a/core/src/services/fs/backend.rs b/core/src/services/fs/backend.rs
index e2205e03c0..68ebf1b398 100644
--- a/core/src/services/fs/backend.rs
+++ b/core/src/services/fs/backend.rs
@@ -241,10 +241,10 @@ impl FsBackend {
#[async_trait]
impl Accessor for FsBackend {
type Reader = oio::TokioReader<tokio::fs::File>;
- type BlockingReader = oio::StdReader<std::fs::File>;
type Writer = FsWriter<tokio::fs::File>;
- type BlockingWriter = FsWriter<std::fs::File>;
type Lister = Option<FsLister<tokio::fs::ReadDir>>;
+ type BlockingReader = oio::StdReader<std::fs::File>;
+ type BlockingWriter = FsWriter<std::fs::File>;
type BlockingLister = Option<FsLister<std::fs::ReadDir>>;
fn info(&self) -> AccessorInfo {
@@ -286,6 +286,29 @@ impl Accessor for FsBackend {
Ok(RpCreateDir::default())
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?;
+
+ let mode = if meta.is_dir() {
+ EntryMode::DIR
+ } else if meta.is_file() {
+ EntryMode::FILE
+ } else {
+ EntryMode::Unknown
+ };
+ let m = Metadata::new(mode)
+ .with_content_length(meta.len())
+ .with_last_modified(
+ meta.modified()
+ .map(DateTime::from)
+ .map_err(new_std_io_error)?,
+ );
+
+ Ok(RpStat::new(m))
+ }
+
/// # Notes
///
/// There are three ways to get the total file length:
@@ -346,6 +369,45 @@ impl Accessor for FsBackend {
Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
}
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ let meta = tokio::fs::metadata(&p).await;
+
+ match meta {
+ Ok(meta) => {
+ if meta.is_dir() {
+ tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?;
+ } else {
+ tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?;
+ }
+
+ Ok(RpDelete::default())
+ }
+ Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(RpDelete::default()),
+ Err(err) => Err(new_std_io_error(err)),
+ }
+ }
+
+ async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ let f = match tokio::fs::read_dir(&p).await {
+ Ok(rd) => rd,
+ Err(e) => {
+ return if e.kind() == std::io::ErrorKind::NotFound {
+ Ok((RpList::default(), None))
+ } else {
+ Err(new_std_io_error(e))
+ };
+ }
+ };
+
+ let rd = FsLister::new(&self.root, f);
+
+ Ok((RpList::default(), Some(rd)))
+ }
+
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
let from = self.root.join(from.trim_end_matches('/'));
@@ -374,10 +436,18 @@ impl Accessor for FsBackend {
Ok(RpRename::default())
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
let p = self.root.join(path.trim_end_matches('/'));
- let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?;
+ std::fs::create_dir_all(p).map_err(new_std_io_error)?;
+
+ Ok(RpCreateDir::default())
+ }
+
+ fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ let meta = std::fs::metadata(p).map_err(new_std_io_error)?;
let mode = if meta.is_dir() {
EntryMode::DIR
@@ -397,53 +467,6 @@ impl Accessor for FsBackend {
Ok(RpStat::new(m))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let p = self.root.join(path.trim_end_matches('/'));
-
- let meta = tokio::fs::metadata(&p).await;
-
- match meta {
- Ok(meta) => {
- if meta.is_dir() {
- tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?;
- } else {
- tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?;
- }
-
- Ok(RpDelete::default())
- }
- Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(RpDelete::default()),
- Err(err) => Err(new_std_io_error(err)),
- }
- }
-
- async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let p = self.root.join(path.trim_end_matches('/'));
-
- let f = match tokio::fs::read_dir(&p).await {
- Ok(rd) => rd,
- Err(e) => {
- return if e.kind() == std::io::ErrorKind::NotFound {
- Ok((RpList::default(), None))
- } else {
- Err(new_std_io_error(e))
- };
- }
- };
-
- let rd = FsLister::new(&self.root, f);
-
- Ok((RpList::default(), Some(rd)))
- }
-
- fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let p = self.root.join(path.trim_end_matches('/'));
-
- std::fs::create_dir_all(p).map_err(new_std_io_error)?;
-
- Ok(RpCreateDir::default())
- }
-
fn blocking_read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
let p = self.root.join(path.trim_end_matches('/'));
@@ -495,55 +518,6 @@ impl Accessor for FsBackend {
Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
}
- fn blocking_copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let from = self.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- std::fs::metadata(&from).map_err(new_std_io_error)?;
-
- let to = Self::blocking_ensure_write_abs_path(&self.root, to.trim_end_matches('/'))?;
-
- std::fs::copy(from, to).map_err(new_std_io_error)?;
-
- Ok(RpCopy::default())
- }
-
- fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let from = self.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- std::fs::metadata(&from).map_err(new_std_io_error)?;
-
- let to = Self::blocking_ensure_write_abs_path(&self.root, to.trim_end_matches('/'))?;
-
- std::fs::rename(from, to).map_err(new_std_io_error)?;
-
- Ok(RpRename::default())
- }
-
- fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let p = self.root.join(path.trim_end_matches('/'));
-
- let meta = std::fs::metadata(p).map_err(new_std_io_error)?;
-
- let mode = if meta.is_dir() {
- EntryMode::DIR
- } else if meta.is_file() {
- EntryMode::FILE
- } else {
- EntryMode::Unknown
- };
- let m = Metadata::new(mode)
- .with_content_length(meta.len())
- .with_last_modified(
- meta.modified()
- .map(DateTime::from)
- .map_err(new_std_io_error)?,
- );
-
- Ok(RpStat::new(m))
- }
-
fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let p = self.root.join(path.trim_end_matches('/'));
@@ -582,6 +556,32 @@ impl Accessor for FsBackend {
Ok((RpList::default(), Some(rd)))
}
+
+ fn blocking_copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let from = self.root.join(from.trim_end_matches('/'));
+
+ // try to get the metadata of the source file to ensure it exists
+ std::fs::metadata(&from).map_err(new_std_io_error)?;
+
+ let to = Self::blocking_ensure_write_abs_path(&self.root, to.trim_end_matches('/'))?;
+
+ std::fs::copy(from, to).map_err(new_std_io_error)?;
+
+ Ok(RpCopy::default())
+ }
+
+ fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ let from = self.root.join(from.trim_end_matches('/'));
+
+ // try to get the metadata of the source file to ensure it exists
+ std::fs::metadata(&from).map_err(new_std_io_error)?;
+
+ let to = Self::blocking_ensure_write_abs_path(&self.root, to.trim_end_matches('/'))?;
+
+ std::fs::rename(from, to).map_err(new_std_io_error)?;
+
+ Ok(RpRename::default())
+ }
}
#[cfg(test)]
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 9888d35727..e51b64384b 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -31,14 +31,12 @@ use http::Uri;
use log::debug;
use serde::Deserialize;
use suppaftp::list::File;
-
use suppaftp::types::FileType;
use suppaftp::types::Response;
use suppaftp::AsyncRustlsConnector;
use suppaftp::AsyncRustlsFtpStream;
use suppaftp::FtpError;
use suppaftp::ImplAsyncFtpStream;
-
use suppaftp::Status;
use tokio::sync::OnceCell;
@@ -289,10 +287,10 @@ impl Debug for FtpBackend {
#[async_trait]
impl Accessor for FtpBackend {
type Reader = FtpReader;
- type BlockingReader = ();
type Writer = FtpWriters;
- type BlockingWriter = ();
type Lister = FtpLister;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -343,6 +341,24 @@ impl Accessor for FtpBackend {
return Ok(RpCreateDir::default());
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let file = self.ftp_stat(path).await?;
+
+ let mode = if file.is_file() {
+ EntryMode::FILE
+ } else if file.is_directory() {
+ EntryMode::DIR
+ } else {
+ EntryMode::Unknown
+ };
+
+ let mut meta = Metadata::new(mode);
+ meta.set_content_length(file.size() as u64);
+ meta.set_last_modified(file.modified().into());
+
+ Ok(RpStat::new(meta))
+ }
+
/// TODO: migrate to FileReader maybe?
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let mut ftp_stream = self.ftp_connect(Operation::Read).await?;
@@ -418,24 +434,6 @@ impl Accessor for FtpBackend {
Ok((RpWrite::new(), w))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let file = self.ftp_stat(path).await?;
-
- let mode = if file.is_file() {
- EntryMode::FILE
- } else if file.is_directory() {
- EntryMode::DIR
- } else {
- EntryMode::Unknown
- };
-
- let mut meta = Metadata::new(mode);
- meta.set_content_length(file.size() as u64);
- meta.set_last_modified(file.modified().into());
-
- Ok(RpStat::new(meta))
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let mut ftp_stream = self.ftp_connect(Operation::Delete).await?;
diff --git a/core/src/services/gcs/backend.rs b/core/src/services/gcs/backend.rs
index 6a01601235..3142f8a9b8 100644
--- a/core/src/services/gcs/backend.rs
+++ b/core/src/services/gcs/backend.rs
@@ -322,10 +322,10 @@ pub struct GcsBackend {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for GcsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = GcsWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<GcsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -375,37 +375,6 @@ impl Accessor for GcsBackend {
am
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.core.gcs_get_object(path, &args).await?;
-
- if resp.status().is_success() {
- let size = parse_content_length(resp.headers())?;
- Ok((RpRead::new().with_size(size), resp.into_body()))
- } else if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE {
- Ok((RpRead::new(), IncomingAsyncBody::empty()))
- } else {
- Err(parse_error(resp).await?)
- }
- }
-
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let w = GcsWriter::new(self.core.clone(), path, args);
- let w = oio::RangeWriter::new(w);
-
- Ok((RpWrite::default(), w))
- }
-
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
- let resp = self.core.gcs_copy_object(from, to).await?;
-
- if resp.status().is_success() {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
- } else {
- Err(parse_error(resp).await?)
- }
- }
-
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
let resp = self.core.gcs_get_object_metadata(path, &args).await?;
@@ -437,6 +406,26 @@ impl Accessor for GcsBackend {
Ok(RpStat::new(m))
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.gcs_get_object(path, &args).await?;
+
+ if resp.status().is_success() {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ } else if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE {
+ Ok((RpRead::new(), IncomingAsyncBody::empty()))
+ } else {
+ Err(parse_error(resp).await?)
+ }
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let w = GcsWriter::new(self.core.clone(), path, args);
+ let w = oio::RangeWriter::new(w);
+
+ Ok((RpWrite::default(), w))
+ }
+
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.gcs_delete_object(path).await?;
@@ -460,6 +449,40 @@ impl Accessor for GcsBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+ async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.gcs_copy_object(from, to).await?;
+
+ if resp.status().is_success() {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
+ } else {
+ Err(parse_error(resp).await?)
+ }
+ }
+
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ // We will not send this request out, just for signing.
+ let mut req = match args.operation() {
+ PresignOperation::Stat(v) => self.core.gcs_head_object_xml_request(path, v)?,
+ PresignOperation::Read(v) => self.core.gcs_get_object_xml_request(path, v)?,
+ PresignOperation::Write(v) => {
+ self.core
+ .gcs_insert_object_xml_request(path, v, AsyncBody::Empty)?
+ }
+ };
+
+ self.core.sign_query(&mut req, args.expire()).await?;
+
+ // We don't need this request anymore, consume it directly.
+ let (parts, _) = req.into_parts();
+
+ Ok(RpPresign::new(PresignedRequest::new(
+ parts.method,
+ parts.uri,
+ parts.headers,
+ )))
+ }
+
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
if ops.len() > 100 {
@@ -518,29 +541,6 @@ impl Accessor for GcsBackend {
Err(parse_error(resp).await?)
}
}
-
- async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
- // We will not send this request out, just for signing.
- let mut req = match args.operation() {
- PresignOperation::Stat(v) => self.core.gcs_head_object_xml_request(path, v)?,
- PresignOperation::Read(v) => self.core.gcs_get_object_xml_request(path, v)?,
- PresignOperation::Write(v) => {
- self.core
- .gcs_insert_object_xml_request(path, v, AsyncBody::Empty)?
- }
- };
-
- self.core.sign_query(&mut req, args.expire()).await?;
-
- // We don't need this request anymore, consume it directly.
- let (parts, _) = req.into_parts();
-
- Ok(RpPresign::new(PresignedRequest::new(
- parts.method,
- parts.uri,
- parts.headers,
- )))
- }
}
/// The raw json response returned by [`get`](https://cloud.google.com/storage/docs/json_api/v1/objects/get)
diff --git a/core/src/services/gdrive/backend.rs b/core/src/services/gdrive/backend.rs
index 3e88e912ca..50a2531a29 100644
--- a/core/src/services/gdrive/backend.rs
+++ b/core/src/services/gdrive/backend.rs
@@ -43,10 +43,10 @@ pub struct GdriveBackend {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for GdriveBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<GdriveWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<GdriveLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -76,17 +76,6 @@ impl Accessor for GdriveBackend {
ma
}
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let resp = self.core.gdrive_stat(path).await?;
-
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp).await?);
- }
-
- let meta = self.parse_metadata(resp.into_body().bytes().await?)?;
- Ok(RpStat::new(meta))
- }
-
async fn create_dir(&self, path: &str, _args: OpCreateDir) -> Result<RpCreateDir> {
let parent = self.core.ensure_parent_path(path).await?;
@@ -108,6 +97,17 @@ impl Accessor for GdriveBackend {
Ok(RpCreateDir::default())
}
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.gdrive_stat(path).await?;
+
+ if resp.status() != StatusCode::OK {
+ return Err(parse_error(resp).await?);
+ }
+
+ let meta = self.parse_metadata(resp.into_body().bytes().await?)?;
+ Ok(RpStat::new(meta))
+ }
+
async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.gdrive_get(path).await?;
@@ -160,28 +160,6 @@ impl Accessor for GdriveBackend {
))
}
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let resp = self.core.gdrive_patch_metadata_request(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- let body = resp.into_body().bytes().await?;
- let meta = serde_json::from_slice::<GdriveFile>(&body)
- .map_err(new_json_deserialize_error)?;
-
- let mut cache = self.core.path_cache.lock().await;
-
- cache.remove(&build_abs_path(&self.core.root, from));
- cache.insert(build_abs_path(&self.core.root, to), meta.id.clone());
-
- Ok(RpRename::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.gdrive_delete(path).await;
if let Ok(resp) = resp {
@@ -276,6 +254,28 @@ impl Accessor for GdriveBackend {
_ => Err(parse_error(resp).await?),
}
}
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ let resp = self.core.gdrive_patch_metadata_request(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let meta = serde_json::from_slice::<GdriveFile>(&body)
+ .map_err(new_json_deserialize_error)?;
+
+ let mut cache = self.core.path_cache.lock().await;
+
+ cache.remove(&build_abs_path(&self.core.root, from));
+ cache.insert(build_abs_path(&self.core.root, to), meta.id.clone());
+
+ Ok(RpRename::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
impl GdriveBackend {
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index 546ca7ec74..596189eb26 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -227,10 +227,10 @@ pub struct GhacBackend {
#[async_trait]
impl Accessor for GhacBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = GhacWriter;
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -254,6 +254,34 @@ impl Accessor for GhacBackend {
am
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let req = self.ghac_query(path).await?;
+
+ let resp = self.client.send(req).await?;
+
+ let location = if resp.status() == StatusCode::OK {
+ let slc = resp.into_body().bytes().await?;
+ let query_resp: GhacQueryResponse =
+ serde_json::from_slice(&slc).map_err(new_json_deserialize_error)?;
+ query_resp.archive_location
+ } else {
+ return Err(parse_error(resp).await?);
+ };
+
+ let req = self.ghac_head_location(&location).await?;
+ let resp = self.client.send(req).await?;
+
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let meta = parse_into_metadata(path, resp.headers())?;
+
+ Ok(RpStat::new(meta))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let req = self.ghac_query(path).await?;
@@ -308,34 +336,6 @@ impl Accessor for GhacBackend {
Ok((RpWrite::default(), GhacWriter::new(self.clone(), cache_id)))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let req = self.ghac_query(path).await?;
-
- let resp = self.client.send(req).await?;
-
- let location = if resp.status() == StatusCode::OK {
- let slc = resp.into_body().bytes().await?;
- let query_resp: GhacQueryResponse =
- serde_json::from_slice(&slc).map_err(new_json_deserialize_error)?;
- query_resp.archive_location
- } else {
- return Err(parse_error(resp).await?);
- };
-
- let req = self.ghac_head_location(&location).await?;
- let resp = self.client.send(req).await?;
-
- let status = resp.status();
- match status {
- StatusCode::OK => {
- let meta = parse_into_metadata(path, resp.headers())?;
-
- Ok(RpStat::new(meta))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
if self.api_token.is_empty() {
return Err(Error::new(
diff --git a/core/src/services/hdfs/backend.rs b/core/src/services/hdfs/backend.rs
index 1359477d3c..a03aa6a3de 100644
--- a/core/src/services/hdfs/backend.rs
+++ b/core/src/services/hdfs/backend.rs
@@ -16,7 +16,8 @@
// under the License.
use std::collections::HashMap;
-use std::fmt::{Debug, Formatter};
+use std::fmt::Debug;
+use std::fmt::Formatter;
use std::io;
use std::path::PathBuf;
use std::sync::Arc;
@@ -245,10 +246,10 @@ unsafe impl Sync for HdfsBackend {}
#[async_trait]
impl Accessor for HdfsBackend {
type Reader = oio::FuturesReader<hdrs::AsyncFile>;
- type BlockingReader = oio::StdReader<hdrs::File>;
type Writer = HdfsWriter<hdrs::AsyncFile>;
- type BlockingWriter = HdfsWriter<hdrs::File>;
type Lister = Option<HdfsLister>;
+ type BlockingReader = oio::StdReader<hdrs::File>;
+ type BlockingWriter = HdfsWriter<hdrs::File>;
type BlockingLister = Option<HdfsLister>;
fn info(&self) -> AccessorInfo {
@@ -286,6 +287,25 @@ impl Accessor for HdfsBackend {
Ok(RpCreateDir::default())
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ let meta = self.client.metadata(&p).map_err(new_std_io_error)?;
+
+ let mode = if meta.is_dir() {
+ EntryMode::DIR
+ } else if meta.is_file() {
+ EntryMode::FILE
+ } else {
+ EntryMode::Unknown
+ };
+ let mut m = Metadata::new(mode);
+ m.set_content_length(meta.len());
+ m.set_last_modified(meta.modified().into());
+
+ Ok(RpStat::new(m))
+ }
+
async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
let p = build_rooted_abs_path(&self.root, path);
@@ -358,6 +378,52 @@ impl Accessor for HdfsBackend {
))
}
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ let meta = self.client.metadata(&p);
+
+ if let Err(err) = meta {
+ return if err.kind() == io::ErrorKind::NotFound {
+ Ok(RpDelete::default())
+ } else {
+ Err(new_std_io_error(err))
+ };
+ }
+
+ // Safety: Err branch has been checked, it's OK to unwrap.
+ let meta = meta.ok().unwrap();
+
+ let result = if meta.is_dir() {
+ self.client.remove_dir(&p)
+ } else {
+ self.client.remove_file(&p)
+ };
+
+ result.map_err(new_std_io_error)?;
+
+ Ok(RpDelete::default())
+ }
+
+ async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ let f = match self.client.read_dir(&p) {
+ Ok(f) => f,
+ Err(e) => {
+ return if e.kind() == io::ErrorKind::NotFound {
+ Ok((RpList::default(), None))
+ } else {
+ Err(new_std_io_error(e))
+ }
+ }
+ };
+
+ let rd = HdfsLister::new(&self.root, f);
+
+ Ok((RpList::default(), Some(rd)))
+ }
+
async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
let from_path = build_rooted_abs_path(&self.root, from);
self.client.metadata(&from_path).map_err(new_std_io_error)?;
@@ -405,7 +471,15 @@ impl Accessor for HdfsBackend {
Ok(RpRename::new())
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ self.client.create_dir(&p).map_err(new_std_io_error)?;
+
+ Ok(RpCreateDir::default())
+ }
+
+ fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
let p = build_rooted_abs_path(&self.root, path);
let meta = self.client.metadata(&p).map_err(new_std_io_error)?;
@@ -424,60 +498,6 @@ impl Accessor for HdfsBackend {
Ok(RpStat::new(m))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let meta = self.client.metadata(&p);
-
- if let Err(err) = meta {
- return if err.kind() == io::ErrorKind::NotFound {
- Ok(RpDelete::default())
- } else {
- Err(new_std_io_error(err))
- };
- }
-
- // Safety: Err branch has been checked, it's OK to unwrap.
- let meta = meta.ok().unwrap();
-
- let result = if meta.is_dir() {
- self.client.remove_dir(&p)
- } else {
- self.client.remove_file(&p)
- };
-
- result.map_err(new_std_io_error)?;
-
- Ok(RpDelete::default())
- }
-
- async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let f = match self.client.read_dir(&p) {
- Ok(f) => f,
- Err(e) => {
- return if e.kind() == io::ErrorKind::NotFound {
- Ok((RpList::default(), None))
- } else {
- Err(new_std_io_error(e))
- }
- }
- };
-
- let rd = HdfsLister::new(&self.root, f);
-
- Ok((RpList::default(), Some(rd)))
- }
-
- fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let p = build_rooted_abs_path(&self.root, path);
-
- self.client.create_dir(&p).map_err(new_std_io_error)?;
-
- Ok(RpCreateDir::default())
- }
-
fn blocking_read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
let p = build_rooted_abs_path(&self.root, path);
@@ -546,6 +566,52 @@ impl Accessor for HdfsBackend {
))
}
+ fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ let meta = self.client.metadata(&p);
+
+ if let Err(err) = meta {
+ return if err.kind() == io::ErrorKind::NotFound {
+ Ok(RpDelete::default())
+ } else {
+ Err(new_std_io_error(err))
+ };
+ }
+
+ // Safety: Err branch has been checked, it's OK to unwrap.
+ let meta = meta.ok().unwrap();
+
+ let result = if meta.is_dir() {
+ self.client.remove_dir(&p)
+ } else {
+ self.client.remove_file(&p)
+ };
+
+ result.map_err(new_std_io_error)?;
+
+ Ok(RpDelete::default())
+ }
+
+ fn blocking_list(&self, path: &str, _: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ let p = build_rooted_abs_path(&self.root, path);
+
+ let f = match self.client.read_dir(&p) {
+ Ok(f) => f,
+ Err(e) => {
+ return if e.kind() == io::ErrorKind::NotFound {
+ Ok((RpList::default(), None))
+ } else {
+ Err(new_std_io_error(e))
+ }
+ }
+ };
+
+ let rd = HdfsLister::new(&self.root, f);
+
+ Ok((RpList::default(), Some(rd)))
+ }
+
fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
let from_path = build_rooted_abs_path(&self.root, from);
self.client.metadata(&from_path).map_err(new_std_io_error)?;
@@ -592,69 +658,4 @@ impl Accessor for HdfsBackend {
Ok(RpRename::new())
}
-
- fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let meta = self.client.metadata(&p).map_err(new_std_io_error)?;
-
- let mode = if meta.is_dir() {
- EntryMode::DIR
- } else if meta.is_file() {
- EntryMode::FILE
- } else {
- EntryMode::Unknown
- };
- let mut m = Metadata::new(mode);
- m.set_content_length(meta.len());
- m.set_last_modified(meta.modified().into());
-
- Ok(RpStat::new(m))
- }
-
- fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let meta = self.client.metadata(&p);
-
- if let Err(err) = meta {
- return if err.kind() == io::ErrorKind::NotFound {
- Ok(RpDelete::default())
- } else {
- Err(new_std_io_error(err))
- };
- }
-
- // Safety: Err branch has been checked, it's OK to unwrap.
- let meta = meta.ok().unwrap();
-
- let result = if meta.is_dir() {
- self.client.remove_dir(&p)
- } else {
- self.client.remove_file(&p)
- };
-
- result.map_err(new_std_io_error)?;
-
- Ok(RpDelete::default())
- }
-
- fn blocking_list(&self, path: &str, _: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let f = match self.client.read_dir(&p) {
- Ok(f) => f,
- Err(e) => {
- return if e.kind() == io::ErrorKind::NotFound {
- Ok((RpList::default(), None))
- } else {
- Err(new_std_io_error(e))
- }
- }
- };
-
- let rd = HdfsLister::new(&self.root, f);
-
- Ok((RpList::default(), Some(rd)))
- }
}
diff --git a/core/src/services/hdfs/writer.rs b/core/src/services/hdfs/writer.rs
index a2ddd5c002..6c77097d84 100644
--- a/core/src/services/hdfs/writer.rs
+++ b/core/src/services/hdfs/writer.rs
@@ -15,15 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-use futures::future::BoxFuture;
use std::io::Write;
use std::pin::Pin;
use std::sync::Arc;
+use std::task::ready;
+use std::task::Context;
use std::task::Poll;
-use std::task::{ready, Context};
use async_trait::async_trait;
-use futures::{AsyncWrite, AsyncWriteExt, FutureExt};
+use futures::future::BoxFuture;
+use futures::AsyncWrite;
+use futures::AsyncWriteExt;
+use futures::FutureExt;
use crate::raw::*;
use crate::*;
diff --git a/core/src/services/http/backend.rs b/core/src/services/http/backend.rs
index f295333ba2..0a907e0ae8 100644
--- a/core/src/services/http/backend.rs
+++ b/core/src/services/http/backend.rs
@@ -224,10 +224,10 @@ impl Debug for HttpBackend {
#[async_trait]
impl Accessor for HttpBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -251,6 +251,27 @@ impl Accessor for HttpBackend {
ma
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ // Stat root always returns a DIR.
+ if path == "/" {
+ return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
+ }
+
+ let resp = self.http_head(path, &args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ // HTTP Server like nginx could return FORBIDDEN if auto-index
+ // is not enabled, we should ignore them.
+ StatusCode::NOT_FOUND | StatusCode::FORBIDDEN if path.ends_with('/') => {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.http_get(path, &args).await?;
@@ -272,27 +293,6 @@ impl Accessor for HttpBackend {
_ => Err(parse_error(resp).await?),
}
}
-
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- // Stat root always returns a DIR.
- if path == "/" {
- return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
- }
-
- let resp = self.http_head(path, &args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- // HTTP Server like nginx could return FORBIDDEN if auto-index
- // is not enabled, we should ignore them.
- StatusCode::NOT_FOUND | StatusCode::FORBIDDEN if path.ends_with('/') => {
- Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
}
impl HttpBackend {
diff --git a/core/src/services/huggingface/backend.rs b/core/src/services/huggingface/backend.rs
index 70e6d39b51..defbc7b608 100644
--- a/core/src/services/huggingface/backend.rs
+++ b/core/src/services/huggingface/backend.rs
@@ -244,10 +244,10 @@ pub struct HuggingfaceBackend {
#[async_trait]
impl Accessor for HuggingfaceBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = oio::PageLister<HuggingfaceLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -268,28 +268,6 @@ impl Accessor for HuggingfaceBackend {
am
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.core.hf_resolve(path, args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
- let size = parse_content_length(resp.headers())?;
- let range = parse_content_range(resp.headers())?;
- Ok((
- RpRead::new().with_size(size).with_range(range),
- resp.into_body(),
- ))
- }
- StatusCode::RANGE_NOT_SATISFIABLE => {
- resp.into_body().consume().await?;
- Ok((RpRead::new().with_size(Some(0)), IncomingAsyncBody::empty()))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
// Stat root always returns a DIR.
if path == "/" {
@@ -333,6 +311,28 @@ impl Accessor for HuggingfaceBackend {
}
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.hf_resolve(path, args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ let range = parse_content_range(resp.headers())?;
+ Ok((
+ RpRead::new().with_size(size).with_range(range),
+ resp.into_body(),
+ ))
+ }
+ StatusCode::RANGE_NOT_SATISFIABLE => {
+ resp.into_body().consume().await?;
+ Ok((RpRead::new().with_size(Some(0)), IncomingAsyncBody::empty()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let l = HuggingfaceLister::new(self.core.clone(), path.to_string(), args.recursive());
diff --git a/core/src/services/ipfs/backend.rs b/core/src/services/ipfs/backend.rs
index 081a56e73d..8e990e3138 100644
--- a/core/src/services/ipfs/backend.rs
+++ b/core/src/services/ipfs/backend.rs
@@ -162,10 +162,10 @@ impl Debug for IpfsBackend {
#[async_trait]
impl Accessor for IpfsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = ();
- type BlockingWriter = ();
type Lister = oio::PageLister<DirStream>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -187,17 +187,6 @@ impl Accessor for IpfsBackend {
ma
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.ipfs_get(path, args.range()).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
- _ => Err(parse_error(resp).await?),
- }
- }
-
/// IPFS's stat behavior highly depends on its implementation.
///
/// Based on IPFS [Path Gateway Specification](https://github.com/ipfs/specs/blob/main/http-gateways/PATH_GATEWAY.md),
@@ -349,6 +338,17 @@ impl Accessor for IpfsBackend {
}
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.ipfs_get(path, args.range()).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
let l = DirStream::new(Arc::new(self.clone()), path);
Ok((RpList::default(), oio::PageLister::new(l)))
diff --git a/core/src/services/ipmfs/backend.rs b/core/src/services/ipmfs/backend.rs
index 0433028f0c..a6565a6322 100644
--- a/core/src/services/ipmfs/backend.rs
+++ b/core/src/services/ipmfs/backend.rs
@@ -63,10 +63,10 @@ impl IpmfsBackend {
#[async_trait]
impl Accessor for IpmfsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<IpmfsWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<IpmfsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -105,24 +105,6 @@ impl Accessor for IpmfsBackend {
}
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.ipmfs_read(path, args.range()).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => Ok((RpRead::new(), resp.into_body())),
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- Ok((
- RpWrite::default(),
- oio::OneShotWriter::new(IpmfsWriter::new(self.clone(), path.to_string())),
- ))
- }
-
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
// Stat root always returns a DIR.
if path == "/" {
@@ -155,6 +137,24 @@ impl Accessor for IpmfsBackend {
}
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.ipmfs_read(path, args.range()).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => Ok((RpRead::new(), resp.into_body())),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ Ok((
+ RpWrite::default(),
+ oio::OneShotWriter::new(IpmfsWriter::new(self.clone(), path.to_string())),
+ ))
+ }
+
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.ipmfs_rm(path).await?;
diff --git a/core/src/services/memcached/backend.rs b/core/src/services/memcached/backend.rs
index 60250f1a98..f0d48e4a91 100644
--- a/core/src/services/memcached/backend.rs
+++ b/core/src/services/memcached/backend.rs
@@ -18,16 +18,17 @@
use std::collections::HashMap;
use std::time::Duration;
-use super::ascii;
-use crate::raw::adapters::kv;
-use crate::raw::*;
-use crate::*;
use async_trait::async_trait;
use bb8::RunError;
use serde::Deserialize;
use tokio::net::TcpStream;
use tokio::sync::OnceCell;
+use super::ascii;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::*;
+
/// Config for MemCached services support
#[derive(Default, Deserialize, Clone)]
#[serde(default)]
diff --git a/core/src/services/obs/backend.rs b/core/src/services/obs/backend.rs
index ce562dd51b..d00fbe423a 100644
--- a/core/src/services/obs/backend.rs
+++ b/core/src/services/obs/backend.rs
@@ -249,10 +249,10 @@ pub struct ObsBackend {
#[async_trait]
impl Accessor for ObsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = ObsWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<ObsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -307,25 +307,19 @@ impl Accessor for ObsBackend {
am
}
- async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
- let mut req = match args.operation() {
- PresignOperation::Stat(v) => self.core.obs_head_object_request(path, v)?,
- PresignOperation::Read(v) => self.core.obs_get_object_request(path, v)?,
- PresignOperation::Write(v) => {
- self.core
- .obs_put_object_request(path, None, v, AsyncBody::Empty)?
- }
- };
- self.core.sign_query(&mut req, args.expire()).await?;
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let resp = self.core.obs_head_object(path, &args).await?;
- // We don't need this request anymore, consume it directly.
- let (parts, _) = req.into_parts();
+ let status = resp.status();
- Ok(RpPresign::new(PresignedRequest::new(
- parts.method,
- parts.uri,
- parts.headers,
- )))
+ // The response is very similar to azblob.
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ StatusCode::NOT_FOUND if path.ends_with('/') => {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
@@ -362,50 +356,56 @@ impl Accessor for ObsBackend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.obs_copy_object(from, to).await?;
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let resp = self.core.obs_delete_object(path).await?;
let status = resp.status();
match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
+ StatusCode::NO_CONTENT | StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
+ Ok(RpDelete::default())
}
_ => Err(parse_error(resp).await?),
}
}
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let resp = self.core.obs_head_object(path, &args).await?;
-
- let status = resp.status();
-
- // The response is very similar to azblob.
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- StatusCode::NOT_FOUND if path.ends_with('/') => {
- Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
- }
- _ => Err(parse_error(resp).await?),
- }
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ let l = ObsLister::new(self.core.clone(), path, args.recursive(), args.limit());
+ Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = self.core.obs_delete_object(path).await?;
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.obs_copy_object(from, to).await?;
let status = resp.status();
match status {
- StatusCode::NO_CONTENT | StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {
- Ok(RpDelete::default())
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
}
_ => Err(parse_error(resp).await?),
}
}
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let l = ObsLister::new(self.core.clone(), path, args.recursive(), args.limit());
- Ok((RpList::default(), oio::PageLister::new(l)))
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ let mut req = match args.operation() {
+ PresignOperation::Stat(v) => self.core.obs_head_object_request(path, v)?,
+ PresignOperation::Read(v) => self.core.obs_get_object_request(path, v)?,
+ PresignOperation::Write(v) => {
+ self.core
+ .obs_put_object_request(path, None, v, AsyncBody::Empty)?
+ }
+ };
+ self.core.sign_query(&mut req, args.expire()).await?;
+
+ // We don't need this request anymore, consume it directly.
+ let (parts, _) = req.into_parts();
+
+ Ok(RpPresign::new(PresignedRequest::new(
+ parts.method,
+ parts.uri,
+ parts.headers,
+ )))
}
}
diff --git a/core/src/services/onedrive/backend.rs b/core/src/services/onedrive/backend.rs
index f491c98106..b63288c38c 100644
--- a/core/src/services/onedrive/backend.rs
+++ b/core/src/services/onedrive/backend.rs
@@ -63,10 +63,10 @@ impl Debug for OnedriveBackend {
#[async_trait]
impl Accessor for OnedriveBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<OneDriveWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<OnedriveLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -86,31 +86,28 @@ impl Accessor for OnedriveBackend {
ma
}
- async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.onedrive_get_content(path).await?;
+ async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
+ let path = build_rooted_abs_path(&self.root, path);
+ let path_before_last_slash = get_parent(&path);
+ let encoded_path = percent_encode_path(path_before_last_slash);
- let status = resp.status();
+ let uri = format!(
+ "https://graph.microsoft.com/v1.0/me/drive/root:{}:/children",
+ encoded_path
+ );
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
- let size = parse_content_length(resp.headers())?;
- let range = parse_content_range(resp.headers())?;
- Ok((
- RpRead::new().with_size(size).with_range(range),
- resp.into_body(),
- ))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
+ let folder_name = get_basename(&path);
+ let folder_name = folder_name.strip_suffix('/').unwrap_or(folder_name);
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let path = build_rooted_abs_path(&self.root, path);
+ let body = CreateDirPayload::new(folder_name.to_string());
- Ok((
- RpWrite::default(),
- oio::OneShotWriter::new(OneDriveWriter::new(self.clone(), args, path)),
- ))
+ let response = self.onedrive_create_dir(&uri, body).await?;
+
+ let status = response.status();
+ match status {
+ StatusCode::CREATED | StatusCode::OK => Ok(RpCreateDir::default()),
+ _ => Err(parse_error(response).await?),
+ }
}
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
@@ -152,6 +149,33 @@ impl Accessor for OnedriveBackend {
}
}
+ async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.onedrive_get_content(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ let range = parse_content_range(resp.headers())?;
+ Ok((
+ RpRead::new().with_size(size).with_range(range),
+ resp.into_body(),
+ ))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let path = build_rooted_abs_path(&self.root, path);
+
+ Ok((
+ RpWrite::default(),
+ oio::OneShotWriter::new(OneDriveWriter::new(self.clone(), args, path)),
+ ))
+ }
+
/// Delete operation
/// Documentation: https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_delete?view=odsp-graph-online
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
@@ -170,30 +194,6 @@ impl Accessor for OnedriveBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
-
- async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let path = build_rooted_abs_path(&self.root, path);
- let path_before_last_slash = get_parent(&path);
- let encoded_path = percent_encode_path(path_before_last_slash);
-
- let uri = format!(
- "https://graph.microsoft.com/v1.0/me/drive/root:{}:/children",
- encoded_path
- );
-
- let folder_name = get_basename(&path);
- let folder_name = folder_name.strip_suffix('/').unwrap_or(folder_name);
-
- let body = CreateDirPayload::new(folder_name.to_string());
-
- let response = self.onedrive_create_dir(&uri, body).await?;
-
- let status = response.status();
- match status {
- StatusCode::CREATED | StatusCode::OK => Ok(RpCreateDir::default()),
- _ => Err(parse_error(response).await?),
- }
- }
}
impl OnedriveBackend {
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index 6530453bd3..6e109e1321 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -377,10 +377,10 @@ pub struct OssBackend {
#[async_trait]
impl Accessor for OssBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = OssWriters;
- type BlockingWriter = ();
type Lister = oio::PageLister<OssLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -441,6 +441,20 @@ impl Accessor for OssBackend {
am
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let resp = self
+ .core
+ .oss_head_object(path, args.if_match(), args.if_none_match())
+ .await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self
.core
@@ -484,33 +498,6 @@ impl Accessor for OssBackend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.oss_copy_object(from, to).await?;
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let resp = self
- .core
- .oss_head_object(path, args.if_match(), args.if_none_match())
- .await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.oss_delete_object(path).await?;
let status = resp.status();
@@ -534,6 +521,19 @@ impl Accessor for OssBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.oss_copy_object(from, to).await?;
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
// We will not send this request out, just for signing.
let mut req = match args.operation() {
diff --git a/core/src/services/pcloud/backend.rs b/core/src/services/pcloud/backend.rs
index 9338fb4893..8283ae8846 100644
--- a/core/src/services/pcloud/backend.rs
+++ b/core/src/services/pcloud/backend.rs
@@ -15,15 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use async_trait::async_trait;
-use http::StatusCode;
-use log::debug;
-use serde::Deserialize;
use std::collections::HashMap;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
+use async_trait::async_trait;
+use http::StatusCode;
+use log::debug;
+use serde::Deserialize;
+
use super::core::*;
use super::error::parse_error;
use super::error::PcloudError;
@@ -228,15 +229,10 @@ pub struct PcloudBackend {
#[async_trait]
impl Accessor for PcloudBackend {
type Reader = IncomingAsyncBody;
-
- type BlockingReader = ();
-
type Writer = PcloudWriters;
-
- type BlockingWriter = ();
-
type Lister = oio::PageLister<PcloudLister>;
-
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -270,61 +266,30 @@ impl Accessor for PcloudBackend {
Ok(RpCreateDir::default())
}
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- self.core.ensure_dir_exists(to).await?;
-
- let resp = if from.ends_with('/') {
- self.core.rename_folder(from, to).await?
- } else {
- self.core.rename_file(from, to).await?
- };
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.stat(path).await?;
let status = resp.status();
match status {
StatusCode::OK => {
let bs = resp.into_body().bytes().await?;
- let resp: PcloudError =
+ let resp: StatResponse =
serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
let result = resp.result;
- if result == 2009 || result == 2010 || result == 2055 || result == 2002 {
+ if result == 2010 || result == 2055 || result == 2002 {
return Err(Error::new(ErrorKind::NotFound, &format!("{resp:?}")));
}
if result != 0 {
return Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")));
}
- Ok(RpRename::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- self.core.ensure_dir_exists(to).await?;
-
- let resp = if from.ends_with('/') {
- self.core.copy_folder(from, to).await?
- } else {
- self.core.copy_file(from, to).await?
- };
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
- let resp: PcloudError =
- serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
- let result = resp.result;
- if result == 2009 || result == 2010 || result == 2055 || result == 2002 {
- return Err(Error::new(ErrorKind::NotFound, &format!("{resp:?}")));
- }
- if result != 0 {
- return Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")));
+ if let Some(md) = resp.metadata {
+ let md = parse_stat_metadata(md);
+ return md.map(RpStat::new);
}
- Ok(RpCopy::default())
+ Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")))
}
_ => Err(parse_error(resp).await?),
}
@@ -350,35 +315,6 @@ impl Accessor for PcloudBackend {
}
}
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let resp = self.core.stat(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
- let resp: StatResponse =
- serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
- let result = resp.result;
- if result == 2010 || result == 2055 || result == 2002 {
- return Err(Error::new(ErrorKind::NotFound, &format!("{resp:?}")));
- }
- if result != 0 {
- return Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")));
- }
-
- if let Some(md) = resp.metadata {
- let md = parse_stat_metadata(md);
- return md.map(RpStat::new);
- }
-
- Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn write(&self, path: &str, _args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
let writer = PcloudWriter::new(self.core.clone(), path.to_string());
@@ -418,4 +354,64 @@ impl Accessor for PcloudBackend {
let l = PcloudLister::new(self.core.clone(), path);
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ self.core.ensure_dir_exists(to).await?;
+
+ let resp = if from.ends_with('/') {
+ self.core.copy_folder(from, to).await?
+ } else {
+ self.core.copy_file(from, to).await?
+ };
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+ let resp: PcloudError =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+ let result = resp.result;
+ if result == 2009 || result == 2010 || result == 2055 || result == 2002 {
+ return Err(Error::new(ErrorKind::NotFound, &format!("{resp:?}")));
+ }
+ if result != 0 {
+ return Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")));
+ }
+
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ self.core.ensure_dir_exists(to).await?;
+
+ let resp = if from.ends_with('/') {
+ self.core.rename_folder(from, to).await?
+ } else {
+ self.core.rename_file(from, to).await?
+ };
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+ let resp: PcloudError =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+ let result = resp.result;
+ if result == 2009 || result == 2010 || result == 2055 || result == 2002 {
+ return Err(Error::new(ErrorKind::NotFound, &format!("{resp:?}")));
+ }
+ if result != 0 {
+ return Err(Error::new(ErrorKind::Unexpected, &format!("{resp:?}")));
+ }
+
+ Ok(RpRename::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
diff --git a/core/src/services/pcloud/core.rs b/core/src/services/pcloud/core.rs
index cdbf46c36c..d71d752aee 100644
--- a/core/src/services/pcloud/core.rs
+++ b/core/src/services/pcloud/core.rs
@@ -15,15 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::{Debug, Formatter};
+use std::fmt::Debug;
+use std::fmt::Formatter;
-use crate::raw::*;
-use crate::*;
use bytes::Bytes;
-use http::{Request, Response, StatusCode};
+use http::Request;
+use http::Response;
+use http::StatusCode;
use serde::Deserialize;
-use super::error::{parse_error, PcloudError};
+use super::error::parse_error;
+use super::error::PcloudError;
+use crate::raw::*;
+use crate::*;
#[derive(Clone)]
pub struct PcloudCore {
diff --git a/core/src/services/pcloud/writer.rs b/core/src/services/pcloud/writer.rs
index 9bd47bb023..108080ff17 100644
--- a/core/src/services/pcloud/writer.rs
+++ b/core/src/services/pcloud/writer.rs
@@ -20,12 +20,12 @@ use std::sync::Arc;
use async_trait::async_trait;
use http::StatusCode;
+use super::core::PcloudCore;
+use super::error::parse_error;
+use super::error::PcloudError;
use crate::raw::*;
use crate::*;
-use super::core::PcloudCore;
-use super::error::{parse_error, PcloudError};
-
pub type PcloudWriters = oio::OneShotWriter<PcloudWriter>;
pub struct PcloudWriter {
diff --git a/core/src/services/rocksdb/backend.rs b/core/src/services/rocksdb/backend.rs
index d4fec8b24e..2787ca79e5 100644
--- a/core/src/services/rocksdb/backend.rs
+++ b/core/src/services/rocksdb/backend.rs
@@ -15,13 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use async_trait::async_trait;
-use rocksdb::DB;
-use serde::Deserialize;
use std::collections::HashMap;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
+
+use async_trait::async_trait;
+use rocksdb::DB;
+use serde::Deserialize;
use tokio::task;
use crate::raw::adapters::kv;
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 9d996ca1c2..94acf49c37 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -983,10 +983,10 @@ pub struct S3Backend {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for S3Backend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = S3Writers;
- type BlockingWriter = ();
type Lister = oio::PageLister<S3Lister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -1051,6 +1051,17 @@ impl Accessor for S3Backend {
am
}
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let resp = self.core.s3_head_object(path, args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.s3_get_object(path, args).await?;
@@ -1081,34 +1092,6 @@ impl Accessor for S3Backend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.s3_copy_object(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- // According to the documentation, when using copy_object, a 200 error may occur and we need to detect it.
- // https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html#API_CopyObject_RequestSyntax
- resp.into_body().consume().await?;
-
- Ok(RpCopy::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let resp = self.core.s3_head_object(path, args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.s3_delete_object(path).await?;
@@ -1135,6 +1118,23 @@ impl Accessor for S3Backend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.s3_copy_object(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ // According to the documentation, when using copy_object, a 200 error may occur and we need to detect it.
+ // https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html#API_CopyObject_RequestSyntax
+ resp.into_body().consume().await?;
+
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let (expire, op) = args.into_parts();
diff --git a/core/src/services/seafile/backend.rs b/core/src/services/seafile/backend.rs
index 34641c567d..5bf5879c14 100644
--- a/core/src/services/seafile/backend.rs
+++ b/core/src/services/seafile/backend.rs
@@ -15,14 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use async_trait::async_trait;
-use http::StatusCode;
-use log::debug;
-use serde::Deserialize;
use std::collections::HashMap;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
+
+use async_trait::async_trait;
+use http::StatusCode;
+use log::debug;
+use serde::Deserialize;
use tokio::sync::RwLock;
use super::core::parse_dir_detail;
@@ -256,15 +257,10 @@ pub struct SeafileBackend {
#[async_trait]
impl Accessor for SeafileBackend {
type Reader = IncomingAsyncBody;
-
- type BlockingReader = ();
-
type Writer = SeafileWriters;
-
- type BlockingWriter = ();
-
type Lister = oio::PageLister<SeafileLister>;
-
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -290,6 +286,23 @@ impl Accessor for SeafileBackend {
am
}
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ if path == "/" {
+ return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
+ }
+
+ let metadata = if path.ends_with('/') {
+ let dir_detail = self.core.dir_detail(path).await?;
+ parse_dir_detail(dir_detail)
+ } else {
+ let file_detail = self.core.file_detail(path).await?;
+
+ parse_file_detail(file_detail)
+ };
+
+ metadata.map(RpStat::new)
+ }
+
async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.download_file(path).await?;
@@ -308,23 +321,6 @@ impl Accessor for SeafileBackend {
}
}
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- if path == "/" {
- return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
- }
-
- let metadata = if path.ends_with('/') {
- let dir_detail = self.core.dir_detail(path).await?;
- parse_dir_detail(dir_detail)
- } else {
- let file_detail = self.core.file_detail(path).await?;
-
- parse_file_detail(file_detail)
- };
-
- metadata.map(RpStat::new)
- }
-
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
let w = SeafileWriter::new(self.core.clone(), args, path.to_string());
let w = oio::OneShotWriter::new(w);
diff --git a/core/src/services/seafile/core.rs b/core/src/services/seafile/core.rs
index a7cb8af7f5..c8af7e1701 100644
--- a/core/src/services/seafile/core.rs
+++ b/core/src/services/seafile/core.rs
@@ -15,23 +15,22 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
use bytes::Bytes;
use http::header;
use http::Request;
use http::Response;
use http::StatusCode;
use serde::Deserialize;
-use std::sync::Arc;
use tokio::sync::RwLock;
-use std::fmt::Debug;
-use std::fmt::Formatter;
-
+use super::error::parse_error;
use crate::raw::*;
use crate::*;
-use super::error::parse_error;
-
/// Core of [seafile](https://www.seafile.com) services support.
#[derive(Clone)]
pub struct SeafileCore {
diff --git a/core/src/services/seafile/lister.rs b/core/src/services/seafile/lister.rs
index c1ee6b9470..792d628032 100644
--- a/core/src/services/seafile/lister.rs
+++ b/core/src/services/seafile/lister.rs
@@ -18,12 +18,13 @@
use std::sync::Arc;
use async_trait::async_trait;
-use http::{header, Request, StatusCode};
+use http::header;
+use http::Request;
+use http::StatusCode;
use serde::Deserialize;
use super::core::SeafileCore;
use super::error::parse_error;
-
use crate::raw::oio::Entry;
use crate::raw::*;
use crate::*;
diff --git a/core/src/services/seafile/writer.rs b/core/src/services/seafile/writer.rs
index 8c7c0ce9db..0ff6ae9190 100644
--- a/core/src/services/seafile/writer.rs
+++ b/core/src/services/seafile/writer.rs
@@ -18,7 +18,9 @@
use std::sync::Arc;
use async_trait::async_trait;
-use http::{header, Request, StatusCode};
+use http::header;
+use http::Request;
+use http::StatusCode;
use super::core::SeafileCore;
use super::error::parse_error;
diff --git a/core/src/services/sftp/backend.rs b/core/src/services/sftp/backend.rs
index 572bb4dd1e..839c1ffd86 100644
--- a/core/src/services/sftp/backend.rs
+++ b/core/src/services/sftp/backend.rs
@@ -243,10 +243,10 @@ impl Debug for SftpBackend {
#[async_trait]
impl Accessor for SftpBackend {
type Reader = oio::TokioReader<Pin<Box<TokioCompatFile>>>;
- type BlockingReader = ();
type Writer = SftpWriter;
- type BlockingWriter = ();
type Lister = Option<SftpLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -300,6 +300,16 @@ impl Accessor for SftpBackend {
return Ok(RpCreateDir::default());
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let client = self.connect().await?;
+ let mut fs = client.fs();
+ fs.set_cwd(&self.root);
+
+ let meta: Metadata = fs.metadata(path).await?.into();
+
+ Ok(RpStat::new(meta))
+ }
+
async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
let client = self.connect().await?;
@@ -344,50 +354,6 @@ impl Accessor for SftpBackend {
Ok((RpWrite::new(), SftpWriter::new(file)))
}
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
- let client = self.connect().await?;
-
- let mut fs = client.fs();
- fs.set_cwd(&self.root);
-
- if let Some((dir, _)) = to.rsplit_once('/') {
- self.create_dir(dir, OpCreateDir::default()).await?;
- }
-
- let src = fs.canonicalize(from).await?;
- let dst = fs.canonicalize(to).await?;
- let mut src_file = client.open(&src).await?;
- let mut dst_file = client.create(dst).await?;
-
- src_file.copy_all_to(&mut dst_file).await?;
-
- Ok(RpCopy::default())
- }
-
- async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
- let client = self.connect().await?;
-
- let mut fs = client.fs();
- fs.set_cwd(&self.root);
-
- if let Some((dir, _)) = to.rsplit_once('/') {
- self.create_dir(dir, OpCreateDir::default()).await?;
- }
- fs.rename(from, to).await?;
-
- Ok(RpRename::default())
- }
-
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let client = self.connect().await?;
- let mut fs = client.fs();
- fs.set_cwd(&self.root);
-
- let meta: Metadata = fs.metadata(path).await?.into();
-
- Ok(RpStat::new(meta))
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let client = self.connect().await?;
@@ -468,6 +434,40 @@ impl Accessor for SftpBackend {
Some(SftpLister::new(dir, path.to_owned())),
))
}
+
+ async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ let client = self.connect().await?;
+
+ let mut fs = client.fs();
+ fs.set_cwd(&self.root);
+
+ if let Some((dir, _)) = to.rsplit_once('/') {
+ self.create_dir(dir, OpCreateDir::default()).await?;
+ }
+
+ let src = fs.canonicalize(from).await?;
+ let dst = fs.canonicalize(to).await?;
+ let mut src_file = client.open(&src).await?;
+ let mut dst_file = client.create(dst).await?;
+
+ src_file.copy_all_to(&mut dst_file).await?;
+
+ Ok(RpCopy::default())
+ }
+
+ async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
+ let client = self.connect().await?;
+
+ let mut fs = client.fs();
+ fs.set_cwd(&self.root);
+
+ if let Some((dir, _)) = to.rsplit_once('/') {
+ self.create_dir(dir, OpCreateDir::default()).await?;
+ }
+ fs.rename(from, to).await?;
+
+ Ok(RpRename::default())
+ }
}
impl SftpBackend {
diff --git a/core/src/services/supabase/backend.rs b/core/src/services/supabase/backend.rs
index cb7bffc7d1..99f04723ad 100644
--- a/core/src/services/supabase/backend.rs
+++ b/core/src/services/supabase/backend.rs
@@ -157,11 +157,11 @@ pub struct SupabaseBackend {
#[async_trait]
impl Accessor for SupabaseBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<SupabaseWriter>;
- type BlockingWriter = ();
// todo: implement Lister to support list and scan
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -183,24 +183,6 @@ impl Accessor for SupabaseBackend {
am
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let resp = self.core.supabase_get_object(path, args.range()).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- Ok((
- RpWrite::default(),
- oio::OneShotWriter::new(SupabaseWriter::new(self.core.clone(), path, args)),
- ))
- }
-
async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
// The get_object_info does not contain the file size. Therefore
// we first try the get the metadata through head, if we fail,
@@ -221,6 +203,24 @@ impl Accessor for SupabaseBackend {
}
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.supabase_get_object(path, args.range()).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ Ok((
+ RpWrite::default(),
+ oio::OneShotWriter::new(SupabaseWriter::new(self.core.clone(), path, args)),
+ ))
+ }
+
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.core.supabase_delete_object(path).await?;
diff --git a/core/src/services/swift/backend.rs b/core/src/services/swift/backend.rs
index 01e123a4d2..8c36a8d3f9 100644
--- a/core/src/services/swift/backend.rs
+++ b/core/src/services/swift/backend.rs
@@ -215,10 +215,10 @@ pub struct SwiftBackend {
#[async_trait]
impl Accessor for SwiftBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<SwiftWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<SwiftLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -244,6 +244,20 @@ impl Accessor for SwiftBackend {
am
}
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.swift_get_metadata(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::NO_CONTENT => {
+ let meta = parse_into_metadata(path, resp.headers())?;
+ Ok(RpStat::new(meta))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.core.swift_read(path, args).await?;
@@ -272,36 +286,6 @@ impl Accessor for SwiftBackend {
return Ok((RpWrite::default(), w));
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- // cannot copy objects larger than 5 GB.
- // Reference: https://docs.openstack.org/api-ref/object-store/#copy-object
- let resp = self.core.swift_copy(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::CREATED | StatusCode::OK => {
- resp.into_body().consume().await?;
- Ok(RpCopy::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let resp = self.core.swift_get_metadata(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::NO_CONTENT => {
- let meta = parse_into_metadata(path, resp.headers())?;
- Ok(RpStat::new(meta))
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _args: OpDelete) -> Result<RpDelete> {
let resp = self.core.swift_delete(path).await?;
@@ -324,4 +308,20 @@ impl Accessor for SwiftBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ // cannot copy objects larger than 5 GB.
+ // Reference: https://docs.openstack.org/api-ref/object-store/#copy-object
+ let resp = self.core.swift_copy(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::CREATED | StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
diff --git a/core/src/services/upyun/backend.rs b/core/src/services/upyun/backend.rs
index f2cd2a8c0c..091977f6f5 100644
--- a/core/src/services/upyun/backend.rs
+++ b/core/src/services/upyun/backend.rs
@@ -15,15 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use async_trait::async_trait;
-use http::StatusCode;
-use log::debug;
-use serde::Deserialize;
use std::collections::HashMap;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
+use async_trait::async_trait;
+use http::StatusCode;
+use log::debug;
+use serde::Deserialize;
+
use super::core::parse_info;
use super::core::UpyunCore;
use super::error::parse_error;
@@ -233,15 +234,10 @@ pub struct UpyunBackend {
#[async_trait]
impl Accessor for UpyunBackend {
type Reader = IncomingAsyncBody;
-
- type BlockingReader = ();
-
type Writer = UpyunWriters;
-
- type BlockingWriter = ();
-
type Lister = oio::PageLister<UpyunLister>;
-
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -290,32 +286,13 @@ impl Accessor for UpyunBackend {
}
}
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let resp = self.core.move_object(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
-
- Ok(RpRename::default())
- }
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let resp = self.core.copy(from, to).await?;
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.info(path).await?;
let status = resp.status();
match status {
- StatusCode::OK => {
- resp.into_body().consume().await?;
-
- Ok(RpCopy::default())
- }
+ StatusCode::OK => parse_info(resp.headers()).map(RpStat::new),
_ => Err(parse_error(resp).await?),
}
}
@@ -338,17 +315,6 @@ impl Accessor for UpyunBackend {
}
}
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let resp = self.core.info(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => parse_info(resp.headers()).map(RpStat::new),
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
let writer = UpyunWriter::new(self.core.clone(), args, path.to_string());
@@ -374,4 +340,34 @@ impl Accessor for UpyunBackend {
let l = UpyunLister::new(self.core.clone(), path, args.limit());
Ok((RpList::default(), oio::PageLister::new(l)))
}
+
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ let resp = self.core.copy(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+
+ Ok(RpCopy::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ let resp = self.core.move_object(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+
+ Ok(RpRename::default())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
diff --git a/core/src/services/upyun/core.rs b/core/src/services/upyun/core.rs
index 4c02d1b338..49b430768f 100644
--- a/core/src/services/upyun/core.rs
+++ b/core/src/services/upyun/core.rs
@@ -15,20 +15,24 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::{Debug, Formatter};
+use std::fmt::Debug;
+use std::fmt::Formatter;
use base64::Engine;
-use hmac::{Hmac, Mac};
-use http::{header, HeaderMap, Request, Response};
+use hmac::Hmac;
+use hmac::Mac;
+use http::header;
+use http::HeaderMap;
+use http::Request;
+use http::Response;
use md5::Digest;
use serde::Deserialize;
use sha1::Sha1;
+use self::constants::*;
use crate::raw::*;
use crate::*;
-use self::constants::*;
-
pub(super) mod constants {
pub const X_UPYUN_FILE_TYPE: &str = "x-upyun-file-type";
pub const X_UPYUN_FILE_SIZE: &str = "x-upyun-file-size";
diff --git a/core/src/services/upyun/lister.rs b/core/src/services/upyun/lister.rs
index e9f17f0871..4fdd068cef 100644
--- a/core/src/services/upyun/lister.rs
+++ b/core/src/services/upyun/lister.rs
@@ -19,7 +19,8 @@ use std::sync::Arc;
use async_trait::async_trait;
-use super::core::{ListObjectsResponse, UpyunCore};
+use super::core::ListObjectsResponse;
+use super::core::UpyunCore;
use super::error::parse_error;
use crate::raw::oio::Entry;
use crate::raw::*;
diff --git a/core/src/services/upyun/writer.rs b/core/src/services/upyun/writer.rs
index a245ae23c4..b4f5082122 100644
--- a/core/src/services/upyun/writer.rs
+++ b/core/src/services/upyun/writer.rs
@@ -20,12 +20,11 @@ use std::sync::Arc;
use async_trait::async_trait;
use http::StatusCode;
-use crate::raw::*;
-use crate::*;
-
use super::core::constants::X_UPYUN_MULTI_UUID;
use super::core::UpyunCore;
use super::error::parse_error;
+use crate::raw::*;
+use crate::*;
pub type UpyunWriters = oio::MultipartUploadWriter<UpyunWriter>;
diff --git a/core/src/services/vercel_artifacts/backend.rs b/core/src/services/vercel_artifacts/backend.rs
index 28af379781..27ef788e19 100644
--- a/core/src/services/vercel_artifacts/backend.rs
+++ b/core/src/services/vercel_artifacts/backend.rs
@@ -46,10 +46,10 @@ impl Debug for VercelArtifactsBackend {
#[async_trait]
impl Accessor for VercelArtifactsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<VercelArtifactsWriter>;
- type BlockingWriter = ();
type Lister = ();
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -69,6 +69,21 @@ impl Accessor for VercelArtifactsBackend {
ma
}
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let res = self.vercel_artifacts_stat(path).await?;
+
+ let status = res.status();
+
+ match status {
+ StatusCode::OK => {
+ let meta = parse_into_metadata(path, res.headers())?;
+ Ok(RpStat::new(meta))
+ }
+
+ _ => Err(parse_error(res).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.vercel_artifacts_get(path, args).await?;
@@ -91,21 +106,6 @@ impl Accessor for VercelArtifactsBackend {
)),
))
}
-
- async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let res = self.vercel_artifacts_stat(path).await?;
-
- let status = res.status();
-
- match status {
- StatusCode::OK => {
- let meta = parse_into_metadata(path, res.headers())?;
- Ok(RpStat::new(meta))
- }
-
- _ => Err(parse_error(res).await?),
- }
- }
}
impl VercelArtifactsBackend {
diff --git a/core/src/services/webdav/backend.rs b/core/src/services/webdav/backend.rs
index e1a3e895cf..67e6a9b4ac 100644
--- a/core/src/services/webdav/backend.rs
+++ b/core/src/services/webdav/backend.rs
@@ -246,10 +246,10 @@ impl Debug for WebdavBackend {
#[async_trait]
impl Accessor for WebdavBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<WebdavWriter>;
- type BlockingWriter = ();
type Lister = Option<oio::PageLister<WebdavLister>>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -288,6 +288,36 @@ impl Accessor for WebdavBackend {
Ok(RpCreateDir::default())
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let mut header_map = HeaderMap::new();
+ // not include children
+ header_map.insert("Depth", "0".parse().unwrap());
+ header_map.insert(header::ACCEPT, "application/xml".parse().unwrap());
+
+ let resp = self.webdav_propfind(path, Some(header_map)).await?;
+
+ let status = resp.status();
+
+ if !status.is_success() {
+ return Err(parse_error(resp).await?);
+ }
+
+ let bs = resp.into_body().bytes().await?;
+ let result: Multistatus =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+ let item = result
+ .response
+ .first()
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Failed getting item stat: bad response",
+ )
+ })?
+ .parse_into_metadata()?;
+ Ok(RpStat::new(item))
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let resp = self.webdav_get(path, args).await?;
let status = resp.status();
@@ -319,74 +349,6 @@ impl Accessor for WebdavBackend {
))
}
- /// # Notes
- ///
- /// There is a strange dead lock issues when copying a non-exist file, so we will check
- /// if the source exists first.
- ///
- /// For example: <https://github.com/apache/incubator-opendal/pull/2809>
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- if let Err(err) = self.stat(from, OpStat::default()).await {
- if err.kind() == ErrorKind::NotFound {
- return Err(err);
- }
- }
-
- self.ensure_parent_path(to).await?;
-
- let resp = self.webdav_copy(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::CREATED | StatusCode::NO_CONTENT => Ok(RpCopy::default()),
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- self.ensure_parent_path(to).await?;
-
- let resp = self.webdav_move(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::CREATED | StatusCode::NO_CONTENT => Ok(RpRename::default()),
- _ => Err(parse_error(resp).await?),
- }
- }
-
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let mut header_map = HeaderMap::new();
- // not include children
- header_map.insert("Depth", "0".parse().unwrap());
- header_map.insert(header::ACCEPT, "application/xml".parse().unwrap());
-
- let resp = self.webdav_propfind(path, Some(header_map)).await?;
-
- let status = resp.status();
-
- if !status.is_success() {
- return Err(parse_error(resp).await?);
- }
-
- let bs = resp.into_body().bytes().await?;
- let result: Multistatus =
- quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
- let item = result
- .response
- .first()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "Failed getting item stat: bad response",
- )
- })?
- .parse_into_metadata()?;
- Ok(RpStat::new(item))
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.webdav_delete(path).await?;
@@ -426,6 +388,44 @@ impl Accessor for WebdavBackend {
_ => Err(parse_error(resp).await?),
}
}
+
+ /// # Notes
+ ///
+ /// There is a strange dead lock issues when copying a non-exist file, so we will check
+ /// if the source exists first.
+ ///
+ /// For example: <https://github.com/apache/incubator-opendal/pull/2809>
+ async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ if let Err(err) = self.stat(from, OpStat::default()).await {
+ if err.kind() == ErrorKind::NotFound {
+ return Err(err);
+ }
+ }
+
+ self.ensure_parent_path(to).await?;
+
+ let resp = self.webdav_copy(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::CREATED | StatusCode::NO_CONTENT => Ok(RpCopy::default()),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+ self.ensure_parent_path(to).await?;
+
+ let resp = self.webdav_move(from, to).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::CREATED | StatusCode::NO_CONTENT => Ok(RpRename::default()),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
}
impl WebdavBackend {
diff --git a/core/src/services/webhdfs/backend.rs b/core/src/services/webhdfs/backend.rs
index 2759c1e6a1..18f4b55ca1 100644
--- a/core/src/services/webhdfs/backend.rs
+++ b/core/src/services/webhdfs/backend.rs
@@ -395,10 +395,10 @@ impl WebhdfsBackend {
#[async_trait]
impl Accessor for WebhdfsBackend {
type Reader = IncomingAsyncBody;
- type BlockingReader = ();
type Writer = oio::OneShotWriter<WebhdfsWriter>;
- type BlockingWriter = ();
type Lister = oio::PageLister<WebhdfsLister>;
+ type BlockingReader = ();
+ type BlockingWriter = ();
type BlockingLister = ();
fn info(&self) -> AccessorInfo {
@@ -460,6 +460,38 @@ impl Accessor for WebhdfsBackend {
}
}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ // if root exists and is a directory, stat will be ok
+ self.root_checker
+ .get_or_try_init(|| async { self.check_root().await })
+ .await?;
+
+ let resp = self.webhdfs_get_file_status(path).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+
+ let file_status = serde_json::from_slice::<FileStatusWrapper>(&bs)
+ .map_err(new_json_deserialize_error)?
+ .file_status;
+
+ let meta = match file_status.ty {
+ FileStatusType::Directory => Metadata::new(EntryMode::DIR),
+ FileStatusType::File => Metadata::new(EntryMode::FILE)
+ .with_content_length(file_status.length)
+ .with_last_modified(parse_datetime_from_from_timestamp_millis(
+ file_status.modification_time,
+ )?),
+ };
+
+ Ok(RpStat::new(meta))
+ }
+
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let range = args.range();
let resp = self.webhdfs_read_file(path, range).await?;
@@ -498,38 +530,6 @@ impl Accessor for WebhdfsBackend {
))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- // if root exists and is a directory, stat will be ok
- self.root_checker
- .get_or_try_init(|| async { self.check_root().await })
- .await?;
-
- let resp = self.webhdfs_get_file_status(path).await?;
- let status = resp.status();
- match status {
- StatusCode::OK => {
- let bs = resp.into_body().bytes().await?;
-
- let file_status = serde_json::from_slice::<FileStatusWrapper>(&bs)
- .map_err(new_json_deserialize_error)?
- .file_status;
-
- let meta = match file_status.ty {
- FileStatusType::Directory => Metadata::new(EntryMode::DIR),
- FileStatusType::File => Metadata::new(EntryMode::FILE)
- .with_content_length(file_status.length)
- .with_last_modified(parse_datetime_from_from_timestamp_millis(
- file_status.modification_time,
- )?),
- };
-
- Ok(RpStat::new(meta))
- }
-
- _ => Err(parse_error(resp).await?),
- }
- }
-
async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
let resp = self.webhdfs_delete(path).await?;
diff --git a/core/tests/behavior/blocking_write.rs b/core/tests/behavior/blocking_write.rs
index 9e29a8274c..7f4e6dc6c8 100644
--- a/core/tests/behavior/blocking_write.rs
+++ b/core/tests/behavior/blocking_write.rs
@@ -15,12 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+use std::io::BufReader;
+use std::io::Cursor;
+
use anyhow::Result;
use log::debug;
use log::warn;
use sha2::Digest;
use sha2::Sha256;
-use std::io::{BufReader, Cursor};
use crate::*;
diff --git a/core/tests/behavior/main.rs b/core/tests/behavior/main.rs
index c8ae40fe92..20eb832c82 100644
--- a/core/tests/behavior/main.rs
+++ b/core/tests/behavior/main.rs
@@ -45,7 +45,8 @@ mod blocking_write;
// External dependencies
use libtest_mimic::Arguments;
use libtest_mimic::Trial;
-use opendal::raw::tests::{init_test_service, TEST_RUNTIME};
+use opendal::raw::tests::init_test_service;
+use opendal::raw::tests::TEST_RUNTIME;
use opendal::*;
pub static TEST_FIXTURE: Fixture = Fixture::new();
diff --git a/core/tests/behavior/utils.rs b/core/tests/behavior/utils.rs
index f8ebae75a2..63f8bce841 100644
--- a/core/tests/behavior/utils.rs
+++ b/core/tests/behavior/utils.rs
@@ -15,8 +15,9 @@
// specific language governing permissions and limitations
// under the License.
+use std::mem;
use std::sync::Mutex;
-use std::{mem, usize};
+use std::usize;
use futures::Future;
use libtest_mimic::Failed;
diff --git a/deny.toml b/deny.toml
index 4d4ee1c53a..cf3f93b25e 100644
--- a/deny.toml
+++ b/deny.toml
@@ -16,8 +16,6 @@
# under the License.
[licenses]
-unlicensed = "deny"
-copyleft = "deny"
allow = [
"Apache-2.0",
"Apache-2.0 WITH LLVM-exception",
@@ -27,10 +25,16 @@ allow = [
"ISC",
"CC0-1.0",
]
+copyleft = "deny"
exceptions = [
- { allow = ["OpenSSL"], name = "ring" },
- { allow = ["Unicode-DFS-2016"], name = "unicode-ident" },
+ { allow = [
+ "OpenSSL",
+ ], name = "ring" },
+ { allow = [
+ "Unicode-DFS-2016",
+ ], name = "unicode-ident" },
]
+unlicensed = "deny"
[[licenses.clarify]]
name = "ring"