You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ti...@apache.org on 2025/11/25 11:58:31 UTC
(opendal) branch bb8-to-fastpool created (now 052cb4cca)
This is an automated email from the ASF dual-hosted git repository.
tison pushed a change to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 052cb4cca run format
This branch includes the following new commits:
new 8105434d3 chore: use fastpool as object pool impl
new cabb34802 for redis
new 3debc8f62 for sftp
new e7fc638f5 for memcached
new 052cb4cca run format
The 5 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.
(opendal) 04/05: for memcached
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit e7fc638f5ab369231be75d2b79285efab9edd4ce
Author: tison <wa...@gmail.com>
AuthorDate: Tue Nov 25 19:55:50 2025 +0800
for memcached
Signed-off-by: tison <wa...@gmail.com>
---
core/src/services/memcached/backend.rs | 26 ++++------
core/src/services/memcached/binary.rs | 1 +
core/src/services/memcached/config.rs | 2 +-
core/src/services/memcached/core.rs | 92 +++++++++++++++++-----------------
4 files changed, 56 insertions(+), 65 deletions(-)
diff --git a/core/src/services/memcached/backend.rs b/core/src/services/memcached/backend.rs
index 22ce701c6..22b5f835a 100644
--- a/core/src/services/memcached/backend.rs
+++ b/core/src/services/memcached/backend.rs
@@ -85,7 +85,7 @@ impl MemcachedBuilder {
///
/// Will panic if `max_size` is 0.
#[must_use]
- pub fn connection_pool_max_size(mut self, max_size: u32) -> Self {
+ pub fn connection_pool_max_size(mut self, max_size: usize) -> Self {
assert!(max_size > 0, "max_size must be greater than zero!");
self.config.connection_pool_max_size = Some(max_size);
self
@@ -144,23 +144,15 @@ impl Builder for MemcachedBuilder {
};
let endpoint = format!("{host}:{port}",);
- let root = normalize_root(
- self.config
- .root
- .clone()
- .unwrap_or_else(|| "/".to_string())
- .as_str(),
- );
-
- let conn = OnceCell::new();
- Ok(MemcachedBackend::new(MemcachedCore {
- conn,
+ let root = normalize_root(self.config.root.unwrap_or_else(|| "/".to_string()).as_str());
+
+ Ok(MemcachedBackend::new(MemcachedCore::new(
endpoint,
- username: self.config.username.clone(),
- password: self.config.password.clone(),
- default_ttl: self.config.default_ttl,
- connection_pool_max_size: self.config.connection_pool_max_size,
- })
+ self.config.username,
+ self.config.password,
+ self.config.default_ttl,
+ self.config.connection_pool_max_size,
+ ))
.with_normalized_root(root))
}
}
diff --git a/core/src/services/memcached/binary.rs b/core/src/services/memcached/binary.rs
index e3db22abc..d433f3fce 100644
--- a/core/src/services/memcached/binary.rs
+++ b/core/src/services/memcached/binary.rs
@@ -97,6 +97,7 @@ pub struct Response {
value: Vec<u8>,
}
+#[derive(Debug)]
pub struct Connection {
io: BufReader<TcpStream>,
}
diff --git a/core/src/services/memcached/config.rs b/core/src/services/memcached/config.rs
index 8b82537d7..ac8990a1d 100644
--- a/core/src/services/memcached/config.rs
+++ b/core/src/services/memcached/config.rs
@@ -46,7 +46,7 @@ pub struct MemcachedConfig {
/// The maximum number of connections allowed.
///
/// default is 10
- pub connection_pool_max_size: Option<u32>,
+ pub connection_pool_max_size: Option<usize>,
}
impl Debug for MemcachedConfig {
diff --git a/core/src/services/memcached/core.rs b/core/src/services/memcached/core.rs
index add582f80..c88d7b0be 100644
--- a/core/src/services/memcached/core.rs
+++ b/core/src/services/memcached/core.rs
@@ -15,19 +15,18 @@
// specific language governing permissions and limitations
// under the License.
+use fastpool::{ManageObject, ObjectStatus, bounded};
+use std::sync::Arc;
use std::time::Duration;
-
-use bb8::RunError;
-use mea::once::OnceCell;
use tokio::net::TcpStream;
use super::binary;
use crate::raw::*;
use crate::*;
-/// A `bb8::ManageConnection` for `memcache_async::ascii::Protocol`.
-#[derive(Clone, Debug)]
-pub struct MemcacheConnectionManager {
+/// A connection manager for `memcache_async::ascii::Protocol`.
+#[derive(Clone)]
+struct MemcacheConnectionManager {
address: String,
username: Option<String>,
password: Option<String>,
@@ -43,12 +42,12 @@ impl MemcacheConnectionManager {
}
}
-impl bb8::ManageConnection for MemcacheConnectionManager {
- type Connection = binary::Connection;
+impl ManageObject for MemcacheConnectionManager {
+ type Object = binary::Connection;
type Error = Error;
/// TODO: Implement unix stream support.
- async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
let conn = TcpStream::connect(&self.address)
.await
.map_err(new_std_io_error)?;
@@ -60,53 +59,52 @@ impl bb8::ManageConnection for MemcacheConnectionManager {
Ok(conn)
}
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- conn.version().await.map(|_| ())
- }
-
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- false
+ async fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ match o.version().await {
+ Ok(_) => Ok(()),
+ Err(err) => Err(err),
+ }
}
}
#[derive(Clone, Debug)]
pub struct MemcachedCore {
- pub conn: OnceCell<bb8::Pool<MemcacheConnectionManager>>,
- pub endpoint: String,
- pub username: Option<String>,
- pub password: Option<String>,
- pub default_ttl: Option<Duration>,
- pub connection_pool_max_size: Option<u32>,
+ default_ttl: Option<Duration>,
+ conn: Arc<bounded::Pool<MemcacheConnectionManager>>,
}
impl MemcachedCore {
- async fn conn(&self) -> Result<bb8::PooledConnection<'_, MemcacheConnectionManager>> {
- let pool = self
- .conn
- .get_or_try_init(|| async {
- let mgr = MemcacheConnectionManager::new(
- &self.endpoint,
- self.username.clone(),
- self.password.clone(),
- );
-
- bb8::Pool::builder()
- .max_size(self.connection_pool_max_size.unwrap_or(10))
- .build(mgr)
- .await
- .map_err(|err| {
- Error::new(ErrorKind::ConfigInvalid, "connect to memcached failed")
- .set_source(err)
- })
- })
- .await?;
-
- pool.get().await.map_err(|err| match err {
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "get connection from pool failed").set_temporary()
+ pub fn new(
+ endpoint: String,
+ username: Option<String>,
+ password: Option<String>,
+ default_ttl: Option<Duration>,
+ connection_pool_max_size: Option<usize>,
+ ) -> Self {
+ let conn = bounded::Pool::new(
+ bounded::PoolConfig::new(connection_pool_max_size.unwrap_or(10)),
+ MemcacheConnectionManager::new(endpoint.as_str(), username, password),
+ );
+
+ Self { default_ttl, conn }
+ }
+
+ async fn conn(&self) -> Result<bounded::Object<MemcacheConnectionManager>> {
+ let fut = self.conn.get();
+
+ tokio::select! {
+ _ = tokio::time::sleep(Duration::from_secs(10)) => {
+ Err(Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary())
}
- RunError::User(err) => err,
- })
+ result = fut => match result {
+ Ok(conn) => Ok(conn),
+ Err(err) => Err(err),
+ }
+ }
}
pub async fn get(&self, key: &str) -> Result<Option<Buffer>> {
(opendal) 01/05: chore: use fastpool as object pool impl
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8105434d3577fceda07d071ee665270f45b25437
Author: tison <wa...@gmail.com>
AuthorDate: Tue Nov 25 19:03:55 2025 +0800
chore: use fastpool as object pool impl
Signed-off-by: tison <wa...@gmail.com>
---
core/Cargo.lock | 23 +++++----
core/Cargo.toml | 12 ++---
core/src/services/etcd/backend.rs | 10 +---
core/src/services/etcd/core.rs | 100 ++++++++++++++++++++------------------
core/src/services/ftp/backend.rs | 20 +++-----
core/src/services/ftp/core.rs | 74 +++++++++++++---------------
core/src/services/ftp/deleter.rs | 4 +-
core/src/services/ftp/err.rs | 2 +-
core/src/services/ftp/reader.rs | 12 ++---
core/src/services/ftp/writer.rs | 14 +++---
10 files changed, 130 insertions(+), 141 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 7ab3d1485..d07ed620c 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -1156,17 +1156,6 @@ version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "55248b47b0caf0546f7988906588779981c43bb1bc9d0c44087278f80cdb44ba"
-[[package]]
-name = "bb8"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "212d8b8e1a22743d9241575c6ba822cf9c8fef34771c86ab7e477a4fbfd254e5"
-dependencies = [
- "futures-util",
- "parking_lot 0.12.5",
- "tokio",
-]
-
[[package]]
name = "bcrypt"
version = "0.15.1"
@@ -2885,6 +2874,16 @@ dependencies = [
"paste",
]
+[[package]]
+name = "fastpool"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6777b4743839a42fd32141d95d7adc4c98e3a1d5200a2598cf32ffd102c81e1"
+dependencies = [
+ "mea",
+ "scopeguard",
+]
+
[[package]]
name = "fastrace"
version = "0.7.14"
@@ -5351,7 +5350,6 @@ dependencies = [
"await-tree",
"backon",
"base64 0.22.1",
- "bb8",
"bytes",
"cacache",
"compio",
@@ -5362,6 +5360,7 @@ dependencies = [
"dotenvy",
"etcd-client",
"fastmetrics",
+ "fastpool",
"fastrace",
"fastrace-jaeger",
"flume",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 16224b93b..eccca34ef 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -141,12 +141,12 @@ services-d1 = []
services-dashmap = ["dep:dashmap"]
services-dbfs = []
services-dropbox = []
-services-etcd = ["dep:etcd-client", "dep:bb8"]
+services-etcd = ["dep:etcd-client", "dep:fastpool"]
services-foundationdb = ["dep:foundationdb"]
services-fs = ["tokio/fs", "internal-tokio-rt"]
services-ftp = [
"dep:suppaftp",
- "dep:bb8",
+ "dep:fastpool",
"dep:futures-rustls",
"dep:rustls-native-certs",
]
@@ -167,7 +167,7 @@ services-ipfs = ["dep:prost"]
services-ipmfs = []
services-koofr = []
services-lakefs = []
-services-memcached = ["dep:bb8"]
+services-memcached = ["dep:fastpool"]
services-memory = []
services-mini-moka = ["dep:mini-moka"]
services-moka = ["dep:moka"]
@@ -195,7 +195,7 @@ services-pcloud = []
services-persy = ["dep:persy", "internal-tokio-rt"]
services-postgresql = ["dep:sqlx", "sqlx?/postgres"]
services-redb = ["dep:redb", "internal-tokio-rt"]
-services-redis = ["dep:redis", "dep:bb8", "redis?/tokio-rustls-comp"]
+services-redis = ["dep:redis", "dep:fastpool", "redis?/tokio-rustls-comp"]
services-redis-native-tls = ["services-redis", "redis?/tokio-native-tls-comp"]
services-rocksdb = ["dep:rocksdb", "internal-tokio-rt"]
services-s3 = [
@@ -206,7 +206,7 @@ services-s3 = [
"dep:crc32c",
]
services-seafile = []
-services-sftp = ["dep:openssh", "dep:openssh-sftp-client", "dep:bb8"]
+services-sftp = ["dep:openssh", "dep:openssh-sftp-client", "dep:fastpool"]
services-sled = ["dep:sled", "internal-tokio-rt"]
services-sqlite = ["dep:sqlx", "sqlx?/sqlite", "dep:ouroboros"]
services-surrealdb = ["dep:surrealdb"]
@@ -272,7 +272,7 @@ rand = { version = "0.8", optional = true }
# Services
# general dependencies.
-bb8 = { version = "0.9", optional = true }
+fastpool = { version = "1.0.2", optional = true }
prost = { version = "0.13", optional = true }
sha1 = { version = "0.10.6", optional = true }
sha2 = { version = "0.10", optional = true }
diff --git a/core/src/services/etcd/backend.rs b/core/src/services/etcd/backend.rs
index 8a9d916f0..317345240 100644
--- a/core/src/services/etcd/backend.rs
+++ b/core/src/services/etcd/backend.rs
@@ -21,7 +21,6 @@ use etcd_client::Certificate;
use etcd_client::ConnectOptions;
use etcd_client::Identity;
use etcd_client::TlsOptions;
-use mea::once::OnceCell;
use super::ETCD_SCHEME;
use super::config::EtcdConfig;
@@ -158,14 +157,7 @@ impl Builder for EtcdBuilder {
.as_str(),
);
- let client = OnceCell::new();
-
- let core = EtcdCore {
- endpoints,
- client,
- options,
- };
-
+ let core = EtcdCore::new(endpoints, options);
Ok(EtcdBackend::new(core, &root))
}
}
diff --git a/core/src/services/etcd/core.rs b/core/src/services/etcd/core.rs
index fcb07a73d..a2fa54b6f 100644
--- a/core/src/services/etcd/core.rs
+++ b/core/src/services/etcd/core.rs
@@ -15,56 +15,54 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::Debug;
-
-use bb8::Pool;
-use bb8::PooledConnection;
-use bb8::RunError;
-use etcd_client::Client;
-use etcd_client::ConnectOptions;
-use mea::once::OnceCell;
-
use crate::services::etcd::error::format_etcd_error;
use crate::{Buffer, Error, ErrorKind, Result};
+use etcd_client::Client;
+use etcd_client::ConnectOptions;
+use fastpool::ManageObject;
+use fastpool::ObjectStatus;
+use fastpool::bounded;
+use std::fmt::Debug;
+use std::sync::Arc;
+use std::time::Duration;
pub mod constants {
pub const DEFAULT_ETCD_ENDPOINTS: &str = "http://127.0.0.1:2379";
}
#[derive(Clone)]
-pub struct Manager {
+struct Manager {
endpoints: Vec<String>,
options: ConnectOptions,
}
-impl bb8::ManageConnection for Manager {
- type Connection = Client;
+impl ManageObject for Manager {
+ type Object = Client;
type Error = Error;
- async fn connect(&self) -> Result<Self::Connection, Self::Error> {
- let conn = Client::connect(self.endpoints.clone(), Some(self.options.clone()))
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
+ Client::connect(self.endpoints.clone(), Some(self.options.clone()))
.await
- .map_err(format_etcd_error)?;
-
- Ok(conn)
- }
-
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- let _ = conn.status().await.map_err(format_etcd_error)?;
- Ok(())
+ .map_err(format_etcd_error)
}
- /// Always allow reuse conn.
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- false
+ async fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ match o.status().await {
+ Ok(_) => Ok(()),
+ Err(err) => Err(format_etcd_error(err)),
+ }
}
}
#[derive(Clone)]
pub struct EtcdCore {
- pub endpoints: Vec<String>,
- pub options: ConnectOptions,
- pub client: OnceCell<Pool<Manager>>,
+ endpoints: Vec<String>,
+ options: ConnectOptions,
+ client: Arc<bounded::Pool<Manager>>,
}
impl Debug for EtcdCore {
@@ -77,26 +75,34 @@ impl Debug for EtcdCore {
}
impl EtcdCore {
- pub async fn conn(&self) -> Result<PooledConnection<'static, Manager>> {
- let client = self
- .client
- .get_or_try_init(|| async {
- Pool::builder()
- .max_size(64)
- .build(Manager {
- endpoints: self.endpoints.clone(),
- options: self.options.clone(),
- })
- .await
- })
- .await?;
-
- client.get_owned().await.map_err(|err| match err {
- RunError::User(err) => err,
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary()
+ pub fn new(endpoints: Vec<String>, options: ConnectOptions) -> Self {
+ let client = bounded::Pool::new(
+ bounded::PoolConfig::new(64),
+ Manager {
+ endpoints: endpoints.clone(),
+ options: options.clone(),
+ },
+ );
+
+ Self {
+ endpoints,
+ options,
+ client,
+ }
+ }
+
+ pub async fn conn(&self) -> Result<bounded::Object<Manager>> {
+ let fut = self.client.get();
+
+ tokio::select! {
+ _ = tokio::time::sleep(Duration::from_secs(10)) => {
+ Err(Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary())
+ }
+ result = fut => match result {
+ Ok(conn) => Ok(conn),
+ Err(err) => Err(err),
}
- })
+ }
}
pub async fn has_prefix(&self, prefix: &str) -> Result<bool> {
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 0b0c87b13..555487717 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -21,7 +21,6 @@ use std::sync::Arc;
use http::Uri;
use log::debug;
-use mea::once::OnceCell;
use services::ftp::core::Manager;
use suppaftp::FtpError;
use suppaftp::Status;
@@ -32,7 +31,7 @@ use super::FTP_SCHEME;
use super::config::FtpConfig;
use super::core::FtpCore;
use super::deleter::FtpDeleter;
-use super::err::parse_error;
+use super::err::format_ftp_error;
use super::lister::FtpLister;
use super::reader::FtpReader;
use super::writer::FtpWriter;
@@ -165,6 +164,7 @@ impl Builder for FtpBuilder {
..Default::default()
});
+
let manager = Manager {
endpoint: endpoint.clone(),
root: root.clone(),
@@ -172,12 +172,8 @@ impl Builder for FtpBuilder {
password: password.clone(),
enable_secure,
};
- let core = Arc::new(FtpCore {
- info: accessor_info.into(),
- manager,
- pool: OnceCell::new(),
- });
+ let core = Arc::new(FtpCore::new(accessor_info.into(), manager.clone()));
Ok(FtpBackend { core })
}
}
@@ -201,7 +197,7 @@ impl Access for FtpBackend {
type Deleter = oio::OneShotDeleter<FtpDeleter>;
fn info(&self) -> Arc<AccessorInfo> {
- self.core.info.clone()
+ self.core.info()
}
async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
@@ -221,7 +217,7 @@ impl Access for FtpBackend {
}))
| Ok(()) => (),
Err(e) => {
- return Err(parse_error(e));
+ return Err(format_ftp_error(e));
}
}
}
@@ -277,7 +273,7 @@ impl Access for FtpBackend {
}))
| Ok(()) => (),
Err(e) => {
- return Err(parse_error(e));
+ return Err(format_ftp_error(e));
}
}
}
@@ -299,7 +295,7 @@ impl Access for FtpBackend {
let mut ftp_stream = self.core.ftp_connect(Operation::List).await?;
let pathname = if path == "/" { None } else { Some(path) };
- let files = ftp_stream.list(pathname).await.map_err(parse_error)?;
+ let files = ftp_stream.list(pathname).await.map_err(format_ftp_error)?;
Ok((
RpList::default(),
@@ -316,7 +312,7 @@ impl FtpBackend {
let pathname = if parent == "/" { None } else { Some(parent) };
- let resp = ftp_stream.list(pathname).await.map_err(parse_error)?;
+ let resp = ftp_stream.list(pathname).await.map_err(format_ftp_error)?;
// Get stat of file.
let mut files = resp
diff --git a/core/src/services/ftp/core.rs b/core/src/services/ftp/core.rs
index 42d427096..afe227add 100644
--- a/core/src/services/ftp/core.rs
+++ b/core/src/services/ftp/core.rs
@@ -15,14 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use std::sync::Arc;
-
-use bb8::Pool;
-use bb8::PooledConnection;
-use bb8::RunError;
+use fastpool::{ManageObject, ObjectStatus, bounded};
use futures_rustls::TlsConnector;
-use mea::once::OnceCell;
use raw::Operation;
+use std::sync::Arc;
+use std::time::Duration;
use suppaftp::AsyncRustlsConnector;
use suppaftp::AsyncRustlsFtpStream;
use suppaftp::FtpError;
@@ -31,35 +28,37 @@ use suppaftp::Status;
use suppaftp::rustls::ClientConfig;
use suppaftp::types::FileType;
-use super::err::parse_error;
+use super::err::format_ftp_error;
use crate::raw::AccessorInfo;
use crate::*;
pub struct FtpCore {
- pub info: Arc<AccessorInfo>,
- pub manager: Manager,
- pub pool: OnceCell<Pool<Manager>>,
+ info: Arc<AccessorInfo>,
+ pool: Arc<bounded::Pool<Manager>>,
}
impl FtpCore {
- pub async fn ftp_connect(&self, _: Operation) -> Result<PooledConnection<'static, Manager>> {
- let pool = self
- .pool
- .get_or_try_init(|| async {
- Pool::builder()
- .max_size(64)
- .build(self.manager.clone())
- .await
- })
- .await
- .map_err(parse_error)?;
-
- pool.get_owned().await.map_err(|err| match err {
- RunError::User(err) => parse_error(err),
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary()
+ pub fn new(info: Arc<AccessorInfo>, manager: Manager) -> Self {
+ let pool = bounded::Pool::new(bounded::PoolConfig::new(64), manager);
+ Self { info, pool }
+ }
+
+ pub fn info(&self) -> Arc<AccessorInfo> {
+ self.info.clone()
+ }
+
+ pub async fn ftp_connect(&self, _: Operation) -> Result<bounded::Object<Manager>> {
+ let fut = self.pool.get();
+
+ tokio::select! {
+ _ = tokio::time::sleep(Duration::from_secs(10)) => {
+ Err(Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary())
+ }
+ result = fut => match result {
+ Ok(conn) => Ok(conn),
+ Err(err) => Err(format_ftp_error(err)),
}
- })
+ }
}
}
@@ -72,11 +71,11 @@ pub struct Manager {
pub enable_secure: bool,
}
-impl bb8::ManageConnection for Manager {
- type Connection = AsyncRustlsFtpStream;
+impl ManageObject for Manager {
+ type Object = AsyncRustlsFtpStream;
type Error = FtpError;
- async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
let stream = ImplAsyncFtpStream::connect(&self.endpoint).await?;
// switch to secure mode if ssl/tls is on.
let mut ftp_stream = if self.enable_secure {
@@ -123,14 +122,11 @@ impl bb8::ManageConnection for Manager {
Ok(ftp_stream)
}
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- conn.noop().await
- }
-
- /// Don't allow reuse conn.
- ///
- /// We need to investigate why reuse conn will cause error.
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- true
+ async fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ o.noop().await
}
}
diff --git a/core/src/services/ftp/deleter.rs b/core/src/services/ftp/deleter.rs
index 38a5b9fdd..10b688912 100644
--- a/core/src/services/ftp/deleter.rs
+++ b/core/src/services/ftp/deleter.rs
@@ -22,7 +22,7 @@ use suppaftp::Status;
use suppaftp::types::Response;
use super::core::FtpCore;
-use super::err::parse_error;
+use super::err::format_ftp_error;
use crate::raw::*;
use crate::*;
@@ -53,7 +53,7 @@ impl oio::OneShotDelete for FtpDeleter {
}))
| Ok(_) => (),
Err(e) => {
- return Err(parse_error(e));
+ return Err(format_ftp_error(e));
}
}
diff --git a/core/src/services/ftp/err.rs b/core/src/services/ftp/err.rs
index eac889880..a2ea33674 100644
--- a/core/src/services/ftp/err.rs
+++ b/core/src/services/ftp/err.rs
@@ -21,7 +21,7 @@ use suppaftp::Status;
use crate::Error;
use crate::ErrorKind;
-pub(super) fn parse_error(err: FtpError) -> Error {
+pub(super) fn format_ftp_error(err: FtpError) -> Error {
let (kind, retryable) = match err {
// Allow retry for error
//
diff --git a/core/src/services/ftp/reader.rs b/core/src/services/ftp/reader.rs
index bebc53d8f..144b03d2b 100644
--- a/core/src/services/ftp/reader.rs
+++ b/core/src/services/ftp/reader.rs
@@ -15,19 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-use bb8::PooledConnection;
use bytes::BytesMut;
+use fastpool::bounded;
use futures::AsyncRead;
use futures::AsyncReadExt;
use super::core::Manager;
-use super::err::parse_error;
+use super::err::format_ftp_error;
use crate::raw::*;
use crate::*;
pub struct FtpReader {
/// Keep the connection alive while data stream is alive.
- _ftp_stream: PooledConnection<'static, Manager>,
+ _ftp_stream: bounded::Object<Manager>,
data_stream: Box<dyn AsyncRead + Sync + Send + Unpin + 'static>,
chunk: usize,
@@ -41,7 +41,7 @@ unsafe impl Sync for FtpReader {}
impl FtpReader {
pub async fn new(
- mut ftp_stream: PooledConnection<'static, Manager>,
+ mut ftp_stream: bounded::Object<Manager>,
path: String,
args: OpRead,
) -> Result<Self> {
@@ -53,12 +53,12 @@ impl FtpReader {
ftp_stream
.resume_transfer(offset as usize)
.await
- .map_err(parse_error)?;
+ .map_err(format_ftp_error)?;
}
let ds = ftp_stream
.retr_as_stream(path)
.await
- .map_err(parse_error)?
+ .map_err(format_ftp_error)?
.take(size as _);
Ok(Self {
diff --git a/core/src/services/ftp/writer.rs b/core/src/services/ftp/writer.rs
index 52db47378..838e0dda7 100644
--- a/core/src/services/ftp/writer.rs
+++ b/core/src/services/ftp/writer.rs
@@ -15,20 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-use bb8::PooledConnection;
use bytes::Buf;
+use fastpool::bounded;
use futures::AsyncWrite;
use futures::AsyncWriteExt;
use super::core::Manager;
-use super::err::parse_error;
+use super::err::format_ftp_error;
use crate::raw::*;
use crate::*;
pub struct FtpWriter {
target_path: String,
tmp_path: Option<String>,
- ftp_stream: PooledConnection<'static, Manager>,
+ ftp_stream: bounded::Object<Manager>,
data_stream: Option<Box<dyn AsyncWrite + Sync + Send + Unpin + 'static>>,
}
@@ -44,7 +44,7 @@ unsafe impl Sync for FtpWriter {}
/// After we can use data stream, we should return it directly.
impl FtpWriter {
pub fn new(
- ftp_stream: PooledConnection<'static, Manager>,
+ ftp_stream: bounded::Object<Manager>,
target_path: String,
tmp_path: Option<String>,
) -> Self {
@@ -70,7 +70,7 @@ impl oio::Write for FtpWriter {
self.ftp_stream
.append_with_stream(path)
.await
- .map_err(parse_error)?,
+ .map_err(format_ftp_error)?,
));
}
@@ -100,13 +100,13 @@ impl oio::Write for FtpWriter {
self.ftp_stream
.finalize_put_stream(data_stream)
.await
- .map_err(parse_error)?;
+ .map_err(format_ftp_error)?;
if let Some(tmp_path) = &self.tmp_path {
self.ftp_stream
.rename(tmp_path, &self.target_path)
.await
- .map_err(parse_error)?;
+ .map_err(format_ftp_error)?;
}
}
(opendal) 03/05: for sftp
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 3debc8f621ae7d88fde5825289561f48a3a6b13b
Author: tison <wa...@gmail.com>
AuthorDate: Tue Nov 25 19:29:01 2025 +0800
for sftp
Signed-off-by: tison <wa...@gmail.com>
---
core/src/services/sftp/backend.rs | 20 +++----
core/src/services/sftp/core.rs | 115 ++++++++++++++++++++------------------
core/src/services/sftp/reader.rs | 6 +-
3 files changed, 73 insertions(+), 68 deletions(-)
diff --git a/core/src/services/sftp/backend.rs b/core/src/services/sftp/backend.rs
index ca2e17a81..384c0b3e4 100644
--- a/core/src/services/sftp/backend.rs
+++ b/core/src/services/sftp/backend.rs
@@ -21,7 +21,6 @@ use std::path::PathBuf;
use std::sync::Arc;
use log::debug;
-use mea::once::OnceCell;
use openssh::KnownHosts;
use tokio::io::AsyncSeekExt;
@@ -187,17 +186,14 @@ impl Builder for SftpBuilder {
..Default::default()
});
- let accessor_info = Arc::new(info);
- let core = Arc::new(SftpCore {
- info: accessor_info,
+ let core = Arc::new(SftpCore::new(
+ Arc::new(info),
endpoint,
root,
user,
- key: self.config.key.clone(),
+ self.config.key.clone(),
known_hosts_strategy,
-
- client: OnceCell::new(),
- });
+ ));
debug!("sftp backend finished: {:?}", &self);
Ok(SftpBackend { core })
@@ -319,11 +315,11 @@ impl Access for SftpBackend {
let dir = match fs.open_dir(&file_path).await {
Ok(dir) => dir,
Err(e) => {
- if is_not_found(&e) {
- return Ok((RpList::default(), None));
+ return if is_not_found(&e) {
+ Ok((RpList::default(), None))
} else {
- return Err(parse_sftp_error(e));
- }
+ Err(parse_sftp_error(e))
+ };
}
}
.read_dir();
diff --git a/core/src/services/sftp/core.rs b/core/src/services/sftp/core.rs
index 327cf57af..2520a03d9 100644
--- a/core/src/services/sftp/core.rs
+++ b/core/src/services/sftp/core.rs
@@ -15,35 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::Debug;
-use std::path::Path;
-use std::path::PathBuf;
-use std::sync::Arc;
-
-use bb8::PooledConnection;
-use bb8::RunError;
-use log::debug;
-use mea::once::OnceCell;
-use openssh::KnownHosts;
-use openssh::SessionBuilder;
-use openssh_sftp_client::Sftp;
-use openssh_sftp_client::SftpOptions;
-
use super::error::is_sftp_protocol_error;
use super::error::parse_sftp_error;
use super::error::parse_ssh_error;
use crate::raw::*;
use crate::*;
+use fastpool::{ManageObject, ObjectStatus, bounded};
+use log::debug;
+use openssh::KnownHosts;
+use openssh::SessionBuilder;
+use openssh_sftp_client::Sftp;
+use openssh_sftp_client::SftpOptions;
+use std::fmt::Debug;
+use std::path::Path;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
pub struct SftpCore {
pub info: Arc<AccessorInfo>,
pub endpoint: String,
pub root: String,
- pub user: Option<String>,
- pub key: Option<String>,
- pub known_hosts_strategy: KnownHosts,
-
- pub client: OnceCell<bb8::Pool<Manager>>,
+ client: Arc<bounded::Pool<Manager>>,
}
impl Debug for SftpCore {
@@ -56,29 +49,45 @@ impl Debug for SftpCore {
}
impl SftpCore {
- pub async fn connect(&self) -> Result<PooledConnection<'static, Manager>> {
- let client = self
- .client
- .get_or_try_init(|| async {
- bb8::Pool::builder()
- .max_size(64)
- .build(Manager {
- endpoint: self.endpoint.clone(),
- root: self.root.clone(),
- user: self.user.clone(),
- key: self.key.clone(),
- known_hosts_strategy: self.known_hosts_strategy.clone(),
- })
- .await
- })
- .await?;
-
- client.get_owned().await.map_err(|err| match err {
- RunError::User(err) => err,
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary()
+ pub fn new(
+ info: Arc<AccessorInfo>,
+ endpoint: String,
+ root: String,
+ user: Option<String>,
+ key: Option<String>,
+ known_hosts_strategy: KnownHosts,
+ ) -> Self {
+ let client = bounded::Pool::new(
+ bounded::PoolConfig::new(64),
+ Manager {
+ endpoint: endpoint.clone(),
+ root: root.clone(),
+ user,
+ key,
+ known_hosts_strategy,
+ },
+ );
+
+ SftpCore {
+ info,
+ endpoint,
+ root,
+ client,
+ }
+ }
+
+ pub async fn connect(&self) -> Result<bounded::Object<Manager>> {
+ let fut = self.client.get();
+
+ tokio::select! {
+ _ = tokio::time::sleep(Duration::from_secs(10)) => {
+ Err(Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary())
+ }
+ result = fut => match result {
+ Ok(conn) => Ok(conn),
+ Err(err) => Err(err),
}
- })
+ }
}
}
@@ -90,11 +99,11 @@ pub struct Manager {
known_hosts_strategy: KnownHosts,
}
-impl bb8::ManageConnection for Manager {
- type Connection = Sftp;
+impl ManageObject for Manager {
+ type Object = Sftp;
type Error = Error;
- async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+ async fn create(&self) -> Result<Self::Object, Self::Error> {
let mut session = SessionBuilder::default();
if let Some(user) = &self.user {
@@ -140,14 +149,14 @@ impl bb8::ManageConnection for Manager {
}
// Check if connect valid by checking the root path.
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- let _ = conn.fs().metadata("./").await.map_err(parse_sftp_error)?;
-
- Ok(())
- }
-
- /// Always allow reuse conn.
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- false
+ async fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ match o.fs().metadata("./").await {
+ Ok(_) => Ok(()),
+ Err(e) => Err(parse_sftp_error(e)),
+ }
}
}
diff --git a/core/src/services/sftp/reader.rs b/core/src/services/sftp/reader.rs
index 781f27dd7..676389123 100644
--- a/core/src/services/sftp/reader.rs
+++ b/core/src/services/sftp/reader.rs
@@ -15,8 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use bb8::PooledConnection;
use bytes::BytesMut;
+use fastpool::bounded;
use openssh_sftp_client::file::File;
use super::core::Manager;
@@ -26,7 +26,7 @@ use crate::*;
pub struct SftpReader {
/// Keep the connection alive while data stream is alive.
- _conn: PooledConnection<'static, Manager>,
+ _conn: bounded::Object<Manager>,
file: File,
chunk: usize,
@@ -36,7 +36,7 @@ pub struct SftpReader {
}
impl SftpReader {
- pub fn new(conn: PooledConnection<'static, Manager>, file: File, size: Option<u64>) -> Self {
+ pub fn new(conn: bounded::Object<Manager>, file: File, size: Option<u64>) -> Self {
Self {
_conn: conn,
file,
(opendal) 02/05: for redis
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit cabb348023372712d0315f7a96e202f7eed6a35e
Author: tison <wa...@gmail.com>
AuthorDate: Tue Nov 25 19:21:37 2025 +0800
for redis
Signed-off-by: tison <wa...@gmail.com>
---
core/src/services/redis/backend.rs | 38 ++++++-------
core/src/services/redis/config.rs | 2 +-
core/src/services/redis/core.rs | 111 +++++++++++++++++++------------------
3 files changed, 73 insertions(+), 78 deletions(-)
diff --git a/core/src/services/redis/backend.rs b/core/src/services/redis/backend.rs
index 5365422cc..e9b3d58a1 100644
--- a/core/src/services/redis/backend.rs
+++ b/core/src/services/redis/backend.rs
@@ -20,7 +20,6 @@ use std::sync::Arc;
use std::time::Duration;
use http::Uri;
-use mea::once::OnceCell;
use redis::Client;
use redis::ConnectionAddr;
use redis::ConnectionInfo;
@@ -133,7 +132,7 @@ impl RedisBuilder {
///
/// Will panic if `max_size` is 0.
#[must_use]
- pub fn connection_pool_max_size(mut self, max_size: u32) -> Self {
+ pub fn connection_pool_max_size(mut self, max_size: usize) -> Self {
assert!(max_size > 0, "max_size must be greater than zero!");
self.config.connection_pool_max_size = Some(max_size);
self
@@ -166,16 +165,13 @@ impl Builder for RedisBuilder {
}
let client = client_builder.build().map_err(format_redis_error)?;
- let conn = OnceCell::new();
-
- Ok(RedisBackend::new(RedisCore {
- addr: endpoints,
- client: None,
- cluster_client: Some(client),
- conn,
- default_ttl: self.config.default_ttl,
- connection_pool_max_size: self.config.connection_pool_max_size,
- })
+ Ok(RedisBackend::new(RedisCore::new(
+ endpoints,
+ None,
+ Some(client),
+ self.config.default_ttl,
+ self.config.connection_pool_max_size,
+ ))
.with_normalized_root(root))
} else {
let endpoint = self
@@ -193,15 +189,13 @@ impl Builder for RedisBuilder {
.set_source(e)
})?;
- let conn = OnceCell::new();
- Ok(RedisBackend::new(RedisCore {
- addr: endpoint,
- client: Some(client),
- cluster_client: None,
- conn,
- default_ttl: self.config.default_ttl,
- connection_pool_max_size: self.config.connection_pool_max_size,
- })
+ Ok(RedisBackend::new(RedisCore::new(
+ endpoint,
+ Some(client),
+ None,
+ self.config.default_ttl,
+ self.config.connection_pool_max_size,
+ ))
.with_normalized_root(root))
}
}
@@ -277,7 +271,7 @@ impl RedisBackend {
fn new(core: RedisCore) -> Self {
let info = AccessorInfo::default();
info.set_scheme(REDIS_SCHEME);
- info.set_name(&core.addr);
+ info.set_name(core.addr());
info.set_root("/");
info.set_native_capability(Capability {
read: true,
diff --git a/core/src/services/redis/config.rs b/core/src/services/redis/config.rs
index c5d134f80..1f07f4a04 100644
--- a/core/src/services/redis/config.rs
+++ b/core/src/services/redis/config.rs
@@ -40,7 +40,7 @@ pub struct RedisConfig {
/// The maximum number of connections allowed.
///
/// default is 10
- pub connection_pool_max_size: Option<u32>,
+ pub connection_pool_max_size: Option<usize>,
/// the username to connect redis service.
///
/// default is None
diff --git a/core/src/services/redis/core.rs b/core/src/services/redis/core.rs
index 21f609f51..f0f5de20a 100644
--- a/core/src/services/redis/core.rs
+++ b/core/src/services/redis/core.rs
@@ -16,11 +16,12 @@
// under the License.
use std::fmt::Debug;
+use std::sync::Arc;
use std::time::Duration;
-use bb8::RunError;
+use crate::*;
use bytes::Bytes;
-use mea::once::OnceCell;
+use fastpool::{ManageObject, ObjectStatus, bounded};
use redis::AsyncCommands;
use redis::Client;
use redis::Cmd;
@@ -33,8 +34,6 @@ use redis::aio::ConnectionManager;
use redis::cluster::ClusterClient;
use redis::cluster_async::ClusterConnection;
-use crate::*;
-
#[derive(Clone)]
pub enum RedisConnection {
Normal(ConnectionManager),
@@ -71,15 +70,15 @@ impl ConnectionLike for RedisConnection {
#[derive(Clone)]
pub struct RedisConnectionManager {
- pub client: Option<Client>,
- pub cluster_client: Option<ClusterClient>,
+ client: Option<Client>,
+ cluster_client: Option<ClusterClient>,
}
-impl bb8::ManageConnection for RedisConnectionManager {
- type Connection = RedisConnection;
+impl ManageObject for RedisConnectionManager {
+ type Object = RedisConnection;
type Error = Error;
- async fn connect(&self) -> Result<RedisConnection, Self::Error> {
+ async fn create(&self) -> Result<RedisConnection, Self::Error> {
if let Some(client) = self.client.clone() {
ConnectionManager::new(client.clone())
.await
@@ -96,30 +95,27 @@ impl bb8::ManageConnection for RedisConnectionManager {
}
}
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- let pong: String = conn.ping().await.map_err(format_redis_error)?;
-
- if pong == "PONG" {
- Ok(())
- } else {
- Err(Error::new(ErrorKind::Unexpected, "PING ERROR"))
+ async fn is_recyclable(
+ &self,
+ o: &mut Self::Object,
+ _: &ObjectStatus,
+ ) -> Result<(), Self::Error> {
+ match o.ping::<String>().await {
+ Ok(ref pong) => match pong.as_bytes() {
+ b"PONG" => Ok(()),
+ _ => Err(Error::new(ErrorKind::Unexpected, "PING ERROR")),
+ },
+ Err(err) => Err(format_redis_error(err)),
}
}
-
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- false
- }
}
/// RedisCore holds the Redis connection and configuration
#[derive(Clone)]
pub struct RedisCore {
- pub addr: String,
- pub client: Option<Client>,
- pub cluster_client: Option<ClusterClient>,
- pub conn: OnceCell<bb8::Pool<RedisConnectionManager>>,
- pub default_ttl: Option<Duration>,
- pub connection_pool_max_size: Option<u32>,
+ addr: String,
+ conn: Arc<bounded::Pool<RedisConnectionManager>>,
+ default_ttl: Option<Duration>,
}
impl Debug for RedisCore {
@@ -131,38 +127,43 @@ impl Debug for RedisCore {
}
impl RedisCore {
- pub async fn conn(&self) -> Result<bb8::PooledConnection<'_, RedisConnectionManager>> {
- let pool = self
- .conn
- .get_or_try_init(|| async {
- bb8::Pool::builder()
- .max_size(self.connection_pool_max_size.unwrap_or(10))
- .build(self.get_redis_connection_manager())
- .await
- .map_err(|err| {
- Error::new(ErrorKind::ConfigInvalid, "connect to redis failed")
- .set_source(err)
- })
- })
- .await?;
- pool.get().await.map_err(|err| match err {
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "get connection from pool failed").set_temporary()
- }
- RunError::User(err) => err,
- })
+ pub fn new(
+ endpoint: String,
+ client: Option<Client>,
+ cluster_client: Option<ClusterClient>,
+ default_ttl: Option<Duration>,
+ connection_pool_max_size: Option<usize>,
+ ) -> Self {
+ let manager = RedisConnectionManager {
+ client,
+ cluster_client,
+ };
+ let pool = bounded::Pool::new(
+ bounded::PoolConfig::new(connection_pool_max_size.unwrap_or(10)),
+ manager,
+ );
+
+ Self {
+ addr: endpoint,
+ conn: pool,
+ default_ttl,
+ }
+ }
+
+ pub fn addr(&self) -> &str {
+ &self.addr
}
- pub fn get_redis_connection_manager(&self) -> RedisConnectionManager {
- if let Some(_client) = self.client.clone() {
- RedisConnectionManager {
- client: self.client.clone(),
- cluster_client: None,
+ pub async fn conn(&self) -> Result<bounded::Object<RedisConnectionManager>> {
+ let fut = self.conn.get();
+
+ tokio::select! {
+ _ = tokio::time::sleep(Duration::from_secs(10)) => {
+ Err(Error::new(ErrorKind::Unexpected, "connection request: timeout").set_temporary())
}
- } else {
- RedisConnectionManager {
- client: None,
- cluster_client: self.cluster_client.clone(),
+ result = fut => match result {
+ Ok(conn) => Ok(conn),
+ Err(err) => Err(err),
}
}
}
(opendal) 05/05: run format
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch bb8-to-fastpool
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 052cb4cca1cf49a0577dcfbba3c641fe6637a704
Author: tison <wa...@gmail.com>
AuthorDate: Tue Nov 25 19:58:07 2025 +0800
run format
Signed-off-by: tison <wa...@gmail.com>
---
core/benches/vs_fs/src/main.rs | 11 +-
core/src/blocking/operator.rs | 8 +-
core/src/blocking/write/writer.rs | 2 +-
core/src/docs/concepts.rs | 4 +-
core/src/layers/http_client.rs | 12 ++-
core/src/layers/retry.rs | 28 ++---
core/src/layers/tail_cut.rs | 13 ++-
core/src/lib.rs | 10 +-
core/src/raw/azure.rs | 14 +--
core/src/raw/http_util/bytes_range.rs | 22 ++--
core/src/raw/ops.rs | 5 +-
core/src/raw/path.rs | 4 +-
core/src/raw/serde_util.rs | 58 +++++-----
core/src/raw/time.rs | 14 ++-
core/src/services/aliyun_drive/lister.rs | 3 +-
core/src/services/azblob/core.rs | 9 +-
core/src/services/azblob/error.rs | 11 +-
core/src/services/azdls/backend.rs | 3 +-
core/src/services/azdls/config.rs | 8 +-
core/src/services/azdls/error.rs | 11 +-
core/src/services/azdls/lister.rs | 19 ++--
core/src/services/azfile/backend.rs | 3 +-
core/src/services/azfile/config.rs | 8 +-
core/src/services/azfile/error.rs | 11 +-
core/src/services/b2/config.rs | 8 +-
core/src/services/cacache/deleter.rs | 3 +-
core/src/services/cacache/writer.rs | 3 +-
core/src/services/cloudflare_kv/core.rs | 8 +-
core/src/services/cloudflare_kv/lister.rs | 3 +-
core/src/services/cloudflare_kv/model.rs | 3 +-
core/src/services/cloudflare_kv/writer.rs | 3 +-
core/src/services/compfs/core.rs | 4 +-
core/src/services/cos/core.rs | 7 +-
core/src/services/dashmap/deleter.rs | 4 +-
core/src/services/dashmap/lister.rs | 9 +-
core/src/services/dashmap/writer.rs | 15 ++-
core/src/services/dropbox/writer.rs | 3 +-
core/src/services/etcd/core.rs | 26 ++---
core/src/services/etcd/error.rs | 3 +-
core/src/services/etcd/lister.rs | 4 +-
core/src/services/ftp/backend.rs | 11 +-
core/src/services/ftp/core.rs | 9 +-
core/src/services/ghac/config.rs | 8 +-
core/src/services/http/config.rs | 11 +-
core/src/services/huggingface/config.rs | 8 +-
core/src/services/huggingface/core.rs | 11 +-
core/src/services/ipmfs/builder.rs | 2 +-
core/src/services/lakefs/config.rs | 8 +-
core/src/services/lakefs/lister.rs | 3 +-
core/src/services/memcached/core.rs | 5 +-
core/src/services/mini_moka/config.rs | 8 +-
core/src/services/mini_moka/writer.rs | 6 +-
core/src/services/moka/backend.rs | 6 +-
core/src/services/moka/mod.rs | 3 +-
core/src/services/moka/writer.rs | 3 +-
core/src/services/obs/core.rs | 7 +-
core/src/services/opfs/core.rs | 5 +-
core/src/services/opfs/error.rs | 3 +-
core/src/services/opfs/utils.rs | 11 +-
core/src/services/oss/core.rs | 43 ++++----
core/src/services/pcloud/config.rs | 8 +-
core/src/services/redis/core.rs | 7 +-
core/src/services/s3/core.rs | 153 ++++++++++++---------------
core/src/services/sftp/config.rs | 13 +--
core/src/services/sftp/core.rs | 43 ++++----
core/src/services/sqlite/backend.rs | 3 +-
core/src/services/sqlite/core.rs | 3 +-
core/src/services/sqlite/deleter.rs | 4 +-
core/src/services/sqlite/writer.rs | 4 +-
core/src/services/swift/config.rs | 8 +-
core/src/services/swift/core.rs | 39 +++----
core/src/services/vercel_artifacts/config.rs | 8 +-
core/src/services/vercel_blob/deleter.rs | 3 +-
core/src/services/webdav/config.rs | 19 ++--
core/src/services/yandex_disk/config.rs | 8 +-
core/src/types/buffer.rs | 18 ++--
core/src/types/builder.rs | 8 +-
core/src/types/delete/deleter.rs | 2 +-
core/src/types/metadata.rs | 3 +-
core/src/types/operator/builder.rs | 18 ++--
core/src/types/operator/mod.rs | 7 +-
core/src/types/operator/operator.rs | 17 +--
core/src/types/operator/operator_futures.rs | 3 +-
core/src/types/operator/registry.rs | 14 ++-
core/src/types/operator/uri.rs | 12 ++-
core/src/types/options.rs | 4 +-
rustfmt.toml | 1 -
87 files changed, 504 insertions(+), 494 deletions(-)
diff --git a/core/benches/vs_fs/src/main.rs b/core/benches/vs_fs/src/main.rs
index 22e13d9e7..1f471663b 100644
--- a/core/benches/vs_fs/src/main.rs
+++ b/core/benches/vs_fs/src/main.rs
@@ -56,13 +56,10 @@ fn bench_vs_fs(c: &mut Criterion) {
let path = prepare();
b.iter(|| {
let _ = op
- .read_options(
- &path,
- options::ReadOptions {
- range: (0..16 * 1024 * 1024).into(),
- ..Default::default()
- },
- )
+ .read_options(&path, options::ReadOptions {
+ range: (0..16 * 1024 * 1024).into(),
+ ..Default::default()
+ })
.unwrap();
});
});
diff --git a/core/src/blocking/operator.rs b/core/src/blocking/operator.rs
index 037b65a19..a596c1248 100644
--- a/core/src/blocking/operator.rs
+++ b/core/src/blocking/operator.rs
@@ -199,9 +199,9 @@ impl Operator {
/// ```
/// # use anyhow::Result;
/// # use futures::io;
- /// use opendal::blocking;
/// # use opendal::blocking::Operator;
/// use opendal::ErrorKind;
+ /// use opendal::blocking;
/// #
/// # fn test(op: blocking::Operator) -> Result<()> {
/// if let Err(e) = op.stat("test") {
@@ -598,9 +598,9 @@ impl Operator {
///
/// ```no_run
/// # use anyhow::Result;
+ /// use opendal::EntryMode;
/// use opendal::blocking;
/// use opendal::blocking::Operator;
- /// use opendal::EntryMode;
/// # fn test(op: blocking::Operator) -> Result<()> {
/// let mut entries = op.list("path/to/dir/")?;
/// for entry in entries {
@@ -656,9 +656,9 @@ impl Operator {
/// # use anyhow::Result;
/// # use futures::io;
/// use futures::TryStreamExt;
+ /// use opendal::EntryMode;
/// use opendal::blocking;
/// use opendal::blocking::Operator;
- /// use opendal::EntryMode;
/// # fn test(op: blocking::Operator) -> Result<()> {
/// let mut ds = op.lister("path/to/dir/")?;
/// for de in ds {
@@ -701,9 +701,9 @@ impl Operator {
/// ```
/// # use std::sync::Arc;
/// # use anyhow::Result;
+ /// use opendal::ErrorKind;
/// use opendal::blocking;
/// use opendal::blocking::Operator;
- /// use opendal::ErrorKind;
///
/// # fn test(op: blocking::Operator) -> Result<()> {
/// op.check()?;
diff --git a/core/src/blocking/write/writer.rs b/core/src/blocking/write/writer.rs
index cdb617b49..e1e07dfe5 100644
--- a/core/src/blocking/write/writer.rs
+++ b/core/src/blocking/write/writer.rs
@@ -49,9 +49,9 @@ impl Writer {
///
/// ```
/// use bytes::Bytes;
+ /// use opendal::Result;
/// use opendal::blocking;
/// use opendal::blocking::Operator;
- /// use opendal::Result;
///
/// fn test(op: blocking::Operator) -> Result<()> {
/// let mut w = op.writer("hello.txt")?;
diff --git a/core/src/docs/concepts.rs b/core/src/docs/concepts.rs
index e512ba603..4fe0c67de 100644
--- a/core/src/docs/concepts.rs
+++ b/core/src/docs/concepts.rs
@@ -79,8 +79,8 @@
//!
//! ```no_run
//! # use opendal::Result;
-//! use opendal::services::S3;
//! use opendal::Operator;
+//! use opendal::services::S3;
//!
//! # fn test() -> Result<()> {
//! let mut builder = S3::default();
@@ -117,8 +117,8 @@
//!
//! ```no_run
//! # use opendal::Result;
-//! use opendal::services::S3;
//! use opendal::Operator;
+//! use opendal::services::S3;
//!
//! # async fn test() -> Result<()> {
//! let mut builder = S3::default();
diff --git a/core/src/layers/http_client.rs b/core/src/layers/http_client.rs
index 8f8d2632d..84e31ef63 100644
--- a/core/src/layers/http_client.rs
+++ b/core/src/layers/http_client.rs
@@ -28,11 +28,11 @@ use crate::*;
/// # Examples
///
/// ```no_run
-/// use opendal::layers::HttpClientLayer;
-/// use opendal::services;
/// use opendal::Operator;
/// use opendal::Result;
+/// use opendal::layers::HttpClientLayer;
/// use opendal::raw::HttpClient;
+/// use opendal::services;
///
/// # fn main() -> Result<()> {
/// // Create a custom HTTP client
@@ -48,10 +48,12 @@ use crate::*;
/// # Custom HTTP Client Implementation
///
/// ```no_run
-/// use opendal::raw::{HttpFetch, HttpBody};
+/// use http::Request;
+/// use http::Response;
/// use opendal::Buffer;
-/// use http::{Request, Response};
/// use opendal::Result;
+/// use opendal::raw::HttpBody;
+/// use opendal::raw::HttpFetch;
///
/// struct CustomHttpClient {
/// // Your custom HTTP client fields
@@ -79,9 +81,9 @@ impl HttpClientLayer {
/// # Examples
///
/// ```no_run
+ /// use opendal::Result;
/// use opendal::layers::HttpClientLayer;
/// use opendal::raw::HttpClient;
- /// use opendal::Result;
///
/// # fn main() -> Result<()> {
/// let client = HttpClient::new()?;
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 59a63830f..b42b08d97 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -143,9 +143,9 @@ impl RetryLayer {
///
/// ```no_run
/// use anyhow::Result;
+ /// use opendal::Operator;
/// use opendal::layers::RetryLayer;
/// use opendal::services;
- /// use opendal::Operator;
///
/// let _ = Operator::new(services::Memory::default())
/// .expect("must init")
@@ -160,9 +160,9 @@ impl<I: RetryInterceptor> RetryLayer<I> {
/// Set the retry interceptor as new notify.
///
/// ```no_run
+ /// use opendal::Operator;
/// use opendal::layers::RetryLayer;
/// use opendal::services;
- /// use opendal::Operator;
///
/// fn notify(_err: &opendal::Error, _dur: std::time::Duration) {}
///
@@ -660,24 +660,18 @@ mod tests {
}
async fn read(&self, _: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- Ok((
- RpRead::new(),
- MockReader {
- buf: Bytes::from("Hello, World!").into(),
- range: args.range(),
- attempt: self.attempt.clone(),
- },
- ))
+ Ok((RpRead::new(), MockReader {
+ buf: Bytes::from("Hello, World!").into(),
+ range: args.range(),
+ attempt: self.attempt.clone(),
+ }))
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- Ok((
- RpDelete::default(),
- MockDeleter {
- size: 0,
- attempt: self.attempt.clone(),
- },
- ))
+ Ok((RpDelete::default(), MockDeleter {
+ size: 0,
+ attempt: self.attempt.clone(),
+ }))
}
async fn write(&self, _: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
diff --git a/core/src/layers/tail_cut.rs b/core/src/layers/tail_cut.rs
index 42a071c60..5ebddbbc7 100644
--- a/core/src/layers/tail_cut.rs
+++ b/core/src/layers/tail_cut.rs
@@ -18,7 +18,9 @@
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
@@ -33,8 +35,9 @@ use crate::*;
/// # Examples
///
/// ```no_run
-/// use opendal::layers::TailCutLayer;
/// use std::time::Duration;
+///
+/// use opendal::layers::TailCutLayer;
/// # use opendal::services;
/// # use opendal::Operator;
/// # use opendal::Result;
@@ -176,8 +179,9 @@ impl TailCutLayerBuilder {
/// # Examples
///
/// ```no_run
- /// use opendal::layers::TailCutLayer;
/// use std::time::Duration;
+ ///
+ /// use opendal::layers::TailCutLayer;
/// # use opendal::services;
/// # use opendal::Operator;
/// # use opendal::Result;
@@ -237,8 +241,9 @@ struct TailCutConfig {
/// # Examples
///
/// ```no_run
-/// use opendal::layers::TailCutLayer;
/// use std::time::Duration;
+///
+/// use opendal::layers::TailCutLayer;
/// # use opendal::services;
/// # use opendal::Operator;
/// # use opendal::Result;
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 89177dbbb..73d751464 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -41,9 +41,9 @@
//! Let's take [`services::S3`] as an example:
//!
//! ```no_run
-//! use opendal::services;
//! use opendal::Operator;
//! use opendal::Result;
+//! use opendal::services;
//!
//! fn main() -> Result<()> {
//! // Pick a builder and configure it.
@@ -64,10 +64,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;
+//! use opendal::services;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
@@ -102,11 +102,11 @@
//! into [`futures::AsyncRead`] or [`futures::Stream`] for broader ecosystem compatibility.
//!
//! ```no_run
+//! use opendal::Operator;
+//! use opendal::Result;
//! use opendal::layers::LoggingLayer;
//! use opendal::options;
//! use opendal::services;
-//! use opendal::Operator;
-//! use opendal::Result;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
diff --git a/core/src/raw/azure.rs b/core/src/raw/azure.rs
index c9ef55bbc..3645a4dae 100644
--- a/core/src/raw/azure.rs
+++ b/core/src/raw/azure.rs
@@ -25,9 +25,12 @@ use std::collections::HashMap;
use http::Uri;
use http::response::Parts;
-use reqsign::{AzureStorageConfig, AzureStorageCredential};
+use reqsign::AzureStorageConfig;
+use reqsign::AzureStorageCredential;
-use crate::{Error, ErrorKind, Result};
+use crate::Error;
+use crate::ErrorKind;
+use crate::Result;
/// Parses an [Azure connection string][1] into a configuration object.
///
@@ -329,12 +332,11 @@ mod tests {
use http::Uri;
use reqsign::AzureStorageConfig;
+ use super::AzureStorageService;
+ use super::azure_account_name_from_endpoint;
+ use super::azure_config_from_connection_string;
use crate::raw::azure::censor_sas_uri;
- use super::{
- AzureStorageService, azure_account_name_from_endpoint, azure_config_from_connection_string,
- };
-
#[test]
fn test_azure_config_from_connection_string() {
#[allow(unused_mut)]
diff --git a/core/src/raw/http_util/bytes_range.rs b/core/src/raw/http_util/bytes_range.rs
index 23b4a3d9b..96bbb806e 100644
--- a/core/src/raw/http_util/bytes_range.rs
+++ b/core/src/raw/http_util/bytes_range.rs
@@ -95,24 +95,18 @@ impl BytesRange {
/// Convert bytes range into rust range.
pub fn to_range(&self) -> impl RangeBounds<u64> {
- (
- Bound::Included(self.0),
- match self.1 {
- Some(size) => Bound::Excluded(self.0 + size),
- None => Bound::Unbounded,
- },
- )
+ (Bound::Included(self.0), match self.1 {
+ Some(size) => Bound::Excluded(self.0 + size),
+ None => Bound::Unbounded,
+ })
}
/// Convert bytes range into rust range with usize.
pub(crate) fn to_range_as_usize(self) -> impl RangeBounds<usize> {
- (
- Bound::Included(self.0 as usize),
- match self.1 {
- Some(size) => Bound::Excluded((self.0 + size) as usize),
- None => Bound::Unbounded,
- },
- )
+ (Bound::Included(self.0 as usize), match self.1 {
+ Some(size) => Bound::Excluded((self.0 + size) as usize),
+ None => Bound::Unbounded,
+ })
}
}
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index abd89be9e..7af9d86da 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -19,11 +19,12 @@
//!
//! By using ops, users can add more context for operation.
-use crate::options;
-use crate::raw::*;
use std::collections::HashMap;
use std::time::Duration;
+use crate::options;
+use crate::raw::*;
+
/// Args for `create` operation.
///
/// The path must be normalized.
diff --git a/core/src/raw/path.rs b/core/src/raw/path.rs
index 55733f47c..5b71d03b0 100644
--- a/core/src/raw/path.rs
+++ b/core/src/raw/path.rs
@@ -15,8 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+use std::hash::BuildHasher;
+use std::hash::Hasher;
+
use crate::*;
-use std::hash::{BuildHasher, Hasher};
/// build_abs_path will build an absolute path with root.
///
diff --git a/core/src/raw/serde_util.rs b/core/src/raw/serde_util.rs
index 3c5ded9b8..3760d5488 100644
--- a/core/src/raw/serde_util.rs
+++ b/core/src/raw/serde_util.rs
@@ -375,31 +375,28 @@ mod tests {
.collect();
let output = TestConfig::deserialize(ConfigDeserializer::new(map)).unwrap();
- assert_eq!(
- output,
- TestConfig {
- bool_value: true,
- bool_option_value_none: None,
- bool_option_value_some: Some(false),
- bool_value_with_on: true,
- bool_value_with_off: false,
- string_value: "hello".to_string(),
- string_option_value_none: None,
- string_option_value_some: Some("hello".to_string()),
- u8_value: 8,
- u16_value: 16,
- u32_value: 32,
- u64_value: 64,
- i8_value: -8,
- i16_value: 16,
- i32_value: -32,
- i64_value: 64,
- vec_value: vec!["hello".to_string()],
- vec_value_two: vec!["hello".to_string(), "world".to_string()],
- vec_none: None,
- vec_empty: vec![],
- }
- );
+ assert_eq!(output, TestConfig {
+ bool_value: true,
+ bool_option_value_none: None,
+ bool_option_value_some: Some(false),
+ bool_value_with_on: true,
+ bool_value_with_off: false,
+ string_value: "hello".to_string(),
+ string_option_value_none: None,
+ string_option_value_some: Some("hello".to_string()),
+ u8_value: 8,
+ u16_value: 16,
+ u32_value: 32,
+ u64_value: 64,
+ i8_value: -8,
+ i16_value: 16,
+ i32_value: -32,
+ i64_value: 64,
+ vec_value: vec!["hello".to_string()],
+ vec_value_two: vec!["hello".to_string(), "world".to_string()],
+ vec_none: None,
+ vec_empty: vec![],
+ });
}
#[test]
@@ -412,12 +409,9 @@ mod tests {
.collect();
let output = TestConfig::deserialize(ConfigDeserializer::new(map)).unwrap();
- assert_eq!(
- output,
- TestConfig {
- bool_value: true,
- ..TestConfig::default()
- }
- );
+ assert_eq!(output, TestConfig {
+ bool_value: true,
+ ..TestConfig::default()
+ });
}
}
diff --git a/core/src/raw/time.rs b/core/src/raw/time.rs
index 9d67931b3..1fbbbc4e4 100644
--- a/core/src/raw/time.rs
+++ b/core/src/raw/time.rs
@@ -17,12 +17,18 @@
//! Time related utils.
-use crate::*;
-use jiff::SignedDuration;
use std::fmt;
-use std::ops::{Add, AddAssign, Sub, SubAssign};
+use std::ops::Add;
+use std::ops::AddAssign;
+use std::ops::Sub;
+use std::ops::SubAssign;
use std::str::FromStr;
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
+use std::time::SystemTime;
+
+use jiff::SignedDuration;
+
+use crate::*;
/// An instant in time represented as the number of nanoseconds since the Unix epoch.
#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
diff --git a/core/src/services/aliyun_drive/lister.rs b/core/src/services/aliyun_drive/lister.rs
index 8fa9e795a..1b929475f 100644
--- a/core/src/services/aliyun_drive/lister.rs
+++ b/core/src/services/aliyun_drive/lister.rs
@@ -17,6 +17,8 @@
use std::sync::Arc;
+use bytes::Buf;
+
use self::oio::Entry;
use super::core::AliyunDriveCore;
use super::core::AliyunDriveFileList;
@@ -26,7 +28,6 @@ use crate::ErrorKind;
use crate::Metadata;
use crate::Result;
use crate::raw::*;
-use bytes::Buf;
pub struct AliyunDriveLister {
core: Arc<AliyunDriveCore>,
diff --git a/core/src/services/azblob/core.rs b/core/src/services/azblob/core.rs
index fe533807b..d1293e70d 100644
--- a/core/src/services/azblob/core.rs
+++ b/core/src/services/azblob/core.rs
@@ -921,9 +921,10 @@ mod tests {
let out: PutBlockListRequest =
de::from_reader(Bytes::from(bs).reader()).expect("must success");
- assert_eq!(
- out.latest,
- vec!["1".to_string(), "2".to_string(), "3".to_string()]
- );
+ assert_eq!(out.latest, vec![
+ "1".to_string(),
+ "2".to_string(),
+ "3".to_string()
+ ]);
}
}
diff --git a/core/src/services/azblob/error.rs b/core/src/services/azblob/error.rs
index 0d26b8cd5..f8df7536e 100644
--- a/core/src/services/azblob/error.rs
+++ b/core/src/services/azblob/error.rs
@@ -86,13 +86,10 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
if message.is_empty() {
if let Some(v) = parts.headers.get("x-ms-error-code") {
if let Ok(code) = v.to_str() {
- message = format!(
- "{:?}",
- AzblobError {
- code: code.to_string(),
- ..Default::default()
- }
- )
+ message = format!("{:?}", AzblobError {
+ code: code.to_string(),
+ ..Default::default()
+ })
}
}
}
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index 989a16b36..4209eedaf 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -214,7 +214,8 @@ impl AzdlsBuilder {
/// use opendal::Builder;
/// use opendal::services::Azdls;
///
- /// let conn_str = "AccountName=example;DefaultEndpointsProtocol=https;EndpointSuffix=core.windows.net";
+ /// let conn_str =
+ /// "AccountName=example;DefaultEndpointsProtocol=https;EndpointSuffix=core.windows.net";
///
/// let mut config = Azdls::from_connection_string(&conn_str)
/// .unwrap()
diff --git a/core/src/services/azdls/config.rs b/core/src/services/azdls/config.rs
index c4e5a7643..61b82b0f5 100644
--- a/core/src/services/azdls/config.rs
+++ b/core/src/services/azdls/config.rs
@@ -155,10 +155,10 @@ mod tests {
#[test]
fn from_uri_accepts_filesystem_from_query() {
- let uri = OperatorUri::new(
- "azdls://account.dfs.core.windows.net",
- vec![("filesystem".to_string(), "logs".to_string())],
- )
+ let uri = OperatorUri::new("azdls://account.dfs.core.windows.net", vec![(
+ "filesystem".to_string(),
+ "logs".to_string(),
+ )])
.unwrap();
let cfg = AzdlsConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/azdls/error.rs b/core/src/services/azdls/error.rs
index 83770195e..f7d22b584 100644
--- a/core/src/services/azdls/error.rs
+++ b/core/src/services/azdls/error.rs
@@ -84,13 +84,10 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
if message.is_empty() {
if let Some(v) = parts.headers.get("x-ms-error-code") {
if let Ok(code) = v.to_str() {
- message = format!(
- "{:?}",
- AzdlsError {
- code: code.to_string(),
- ..Default::default()
- }
- )
+ message = format!("{:?}", AzdlsError {
+ code: code.to_string(),
+ ..Default::default()
+ })
}
}
}
diff --git a/core/src/services/azdls/lister.rs b/core/src/services/azdls/lister.rs
index 2e5744a81..97aee6b66 100644
--- a/core/src/services/azdls/lister.rs
+++ b/core/src/services/azdls/lister.rs
@@ -150,16 +150,13 @@ mod tests {
let out: Output = de::from_slice(&bs).expect("must success");
println!("{out:?}");
- assert_eq!(
- out.paths[0],
- Path {
- content_length: "1977097".to_string(),
- etag: "0x8DACF9B0061305F".to_string(),
- is_directory: "".to_string(),
- last_modified: "Sat, 26 Nov 2022 10:43:05 GMT".to_string(),
- name: "c3b3ef48-7783-4946-81bc-dc07e1728878/d4ea21d7-a533-4011-8b1f-d0e566d63725"
- .to_string()
- }
- );
+ assert_eq!(out.paths[0], Path {
+ content_length: "1977097".to_string(),
+ etag: "0x8DACF9B0061305F".to_string(),
+ is_directory: "".to_string(),
+ last_modified: "Sat, 26 Nov 2022 10:43:05 GMT".to_string(),
+ name: "c3b3ef48-7783-4946-81bc-dc07e1728878/d4ea21d7-a533-4011-8b1f-d0e566d63725"
+ .to_string()
+ });
}
}
diff --git a/core/src/services/azfile/backend.rs b/core/src/services/azfile/backend.rs
index 45c02b663..96a821160 100644
--- a/core/src/services/azfile/backend.rs
+++ b/core/src/services/azfile/backend.rs
@@ -149,7 +149,8 @@ impl AzfileBuilder {
/// use opendal::Builder;
/// use opendal::services::Azfile;
///
- /// let conn_str = "AccountName=example;DefaultEndpointsProtocol=https;EndpointSuffix=core.windows.net";
+ /// let conn_str =
+ /// "AccountName=example;DefaultEndpointsProtocol=https;EndpointSuffix=core.windows.net";
///
/// let mut config = Azfile::from_connection_string(&conn_str)
/// .unwrap()
diff --git a/core/src/services/azfile/config.rs b/core/src/services/azfile/config.rs
index ed7bd1ef7..baade64ee 100644
--- a/core/src/services/azfile/config.rs
+++ b/core/src/services/azfile/config.rs
@@ -135,10 +135,10 @@ mod tests {
#[test]
fn from_uri_accepts_share_from_query() {
- let uri = OperatorUri::new(
- "azfile://account.file.core.windows.net",
- vec![("share_name".to_string(), "data".to_string())],
- )
+ let uri = OperatorUri::new("azfile://account.file.core.windows.net", vec![(
+ "share_name".to_string(),
+ "data".to_string(),
+ )])
.unwrap();
let cfg = AzfileConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/azfile/error.rs b/core/src/services/azfile/error.rs
index ec0bd6970..9c15d146c 100644
--- a/core/src/services/azfile/error.rs
+++ b/core/src/services/azfile/error.rs
@@ -85,13 +85,10 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
if message.is_empty() {
if let Some(v) = parts.headers.get("x-ms-error-code") {
if let Ok(code) = v.to_str() {
- message = format!(
- "{:?}",
- AzfileError {
- code: code.to_string(),
- ..Default::default()
- }
- )
+ message = format!("{:?}", AzfileError {
+ code: code.to_string(),
+ ..Default::default()
+ })
}
}
}
diff --git a/core/src/services/b2/config.rs b/core/src/services/b2/config.rs
index 7910a4e99..667091030 100644
--- a/core/src/services/b2/config.rs
+++ b/core/src/services/b2/config.rs
@@ -96,10 +96,10 @@ mod tests {
#[test]
fn from_uri_extracts_bucket_and_root() {
- let uri = OperatorUri::new(
- "b2://example-bucket/path/to/root",
- vec![("bucket_id".to_string(), "bucket-id".to_string())],
- )
+ let uri = OperatorUri::new("b2://example-bucket/path/to/root", vec![(
+ "bucket_id".to_string(),
+ "bucket-id".to_string(),
+ )])
.unwrap();
let cfg = B2Config::from_uri(&uri).unwrap();
diff --git a/core/src/services/cacache/deleter.rs b/core/src/services/cacache/deleter.rs
index 225fbf207..ccf8d4be6 100644
--- a/core/src/services/cacache/deleter.rs
+++ b/core/src/services/cacache/deleter.rs
@@ -17,11 +17,10 @@
use std::sync::Arc;
+use super::core::CacacheCore;
use crate::raw::*;
use crate::*;
-use super::core::CacacheCore;
-
pub struct CacacheDeleter {
core: Arc<CacacheCore>,
}
diff --git a/core/src/services/cacache/writer.rs b/core/src/services/cacache/writer.rs
index e82e74aae..401650e5b 100644
--- a/core/src/services/cacache/writer.rs
+++ b/core/src/services/cacache/writer.rs
@@ -17,11 +17,10 @@
use std::sync::Arc;
+use super::core::CacacheCore;
use crate::raw::*;
use crate::*;
-use super::core::CacacheCore;
-
pub struct CacacheWriter {
core: Arc<CacacheCore>,
path: String,
diff --git a/core/src/services/cloudflare_kv/core.rs b/core/src/services/cloudflare_kv/core.rs
index 13797b587..699dfa967 100644
--- a/core/src/services/cloudflare_kv/core.rs
+++ b/core/src/services/cloudflare_kv/core.rs
@@ -24,13 +24,17 @@ use http::header;
use http::request;
use serde_json::json;
+use crate::Buffer;
+use crate::Result;
+use crate::raw::AccessorInfo;
+use crate::raw::FormDataPart;
+use crate::raw::Multipart;
use crate::raw::Operation;
use crate::raw::QueryPairsWriter;
use crate::raw::new_json_serialize_error;
+use crate::raw::new_request_build_error;
use crate::raw::percent_encode_path;
-use crate::raw::{AccessorInfo, FormDataPart, Multipart, new_request_build_error};
use crate::services::cloudflare_kv::model::CfKvMetadata;
-use crate::{Buffer, Result};
#[derive(Debug, Clone)]
pub struct CloudflareKvCore {
diff --git a/core/src/services/cloudflare_kv/lister.rs b/core/src/services/cloudflare_kv/lister.rs
index 1eeca9906..43b6d566e 100644
--- a/core/src/services/cloudflare_kv/lister.rs
+++ b/core/src/services/cloudflare_kv/lister.rs
@@ -22,7 +22,8 @@ use bytes::Buf;
use super::core::CloudflareKvCore;
use super::error::parse_error;
use crate::raw::*;
-use crate::services::cloudflare_kv::model::{CfKvListKey, CfKvListResponse};
+use crate::services::cloudflare_kv::model::CfKvListKey;
+use crate::services::cloudflare_kv::model::CfKvListResponse;
use crate::*;
pub struct CloudflareKvLister {
diff --git a/core/src/services/cloudflare_kv/model.rs b/core/src/services/cloudflare_kv/model.rs
index 390f11d82..5f56d6450 100644
--- a/core/src/services/cloudflare_kv/model.rs
+++ b/core/src/services/cloudflare_kv/model.rs
@@ -17,7 +17,8 @@
// use std::collections::HashMap;
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
+use serde::Serialize;
#[derive(Debug, Deserialize)]
pub struct CfKvResponse {
diff --git a/core/src/services/cloudflare_kv/writer.rs b/core/src/services/cloudflare_kv/writer.rs
index c644611bd..e9a2cd8ec 100644
--- a/core/src/services/cloudflare_kv/writer.rs
+++ b/core/src/services/cloudflare_kv/writer.rs
@@ -17,12 +17,13 @@
use std::sync::Arc;
+use http::StatusCode;
+
use super::core::CloudflareKvCore;
use super::error::parse_error;
use crate::raw::*;
use crate::services::cloudflare_kv::model::CfKvMetadata;
use crate::*;
-use http::StatusCode;
pub struct CloudflareWriter {
core: Arc<CloudflareKvCore>,
diff --git a/core/src/services/compfs/core.rs b/core/src/services/compfs/core.rs
index 6151ac8e7..5998ebc3e 100644
--- a/core/src/services/compfs/core.rs
+++ b/core/src/services/compfs/core.rs
@@ -19,7 +19,9 @@ use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
-use compio::buf::{IoBuf, IoBuffer, IoVectoredBuf};
+use compio::buf::IoBuf;
+use compio::buf::IoBuffer;
+use compio::buf::IoVectoredBuf;
use compio::dispatcher::Dispatcher;
use crate::raw::*;
diff --git a/core/src/services/cos/core.rs b/core/src/services/cos/core.rs
index cb3c75b38..346627f0c 100644
--- a/core/src/services/cos/core.rs
+++ b/core/src/services/cos/core.rs
@@ -739,10 +739,9 @@ mod tests {
.collect::<Vec<String>>(),
["obj002", "obj003"],
);
- assert_eq!(
- out.contents.iter().map(|v| v.size).collect::<Vec<u64>>(),
- [9, 10],
- );
+ assert_eq!(out.contents.iter().map(|v| v.size).collect::<Vec<u64>>(), [
+ 9, 10
+ ],);
assert_eq!(
out.common_prefixes
.iter()
diff --git a/core/src/services/dashmap/deleter.rs b/core/src/services/dashmap/deleter.rs
index 0f4e10e50..f1990bff0 100644
--- a/core/src/services/dashmap/deleter.rs
+++ b/core/src/services/dashmap/deleter.rs
@@ -18,7 +18,9 @@
use std::sync::Arc;
use super::core::DashmapCore;
-use crate::raw::{OpDelete, build_abs_path, oio};
+use crate::raw::OpDelete;
+use crate::raw::build_abs_path;
+use crate::raw::oio;
use crate::*;
pub struct DashmapDeleter {
diff --git a/core/src/services/dashmap/lister.rs b/core/src/services/dashmap/lister.rs
index e276ac3d1..df9d5b35d 100644
--- a/core/src/services/dashmap/lister.rs
+++ b/core/src/services/dashmap/lister.rs
@@ -15,12 +15,15 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::Arc;
+use std::vec::IntoIter;
+
use super::core::DashmapCore;
+use crate::raw::build_abs_path;
+use crate::raw::build_rel_path;
+use crate::raw::oio;
use crate::raw::oio::Entry;
-use crate::raw::{build_abs_path, build_rel_path, oio};
use crate::*;
-use std::sync::Arc;
-use std::vec::IntoIter;
pub struct DashmapLister {
root: String,
diff --git a/core/src/services/dashmap/writer.rs b/core/src/services/dashmap/writer.rs
index 54efe86cb..494bf9dba 100644
--- a/core/src/services/dashmap/writer.rs
+++ b/core/src/services/dashmap/writer.rs
@@ -20,7 +20,9 @@ use std::time::SystemTime;
use super::core::DashmapCore;
use super::core::DashmapValue;
-use crate::raw::{OpWrite, Timestamp, oio};
+use crate::raw::OpWrite;
+use crate::raw::Timestamp;
+use crate::raw::oio;
use crate::*;
pub struct DashmapWriter {
@@ -69,13 +71,10 @@ impl oio::Write for DashmapWriter {
meta.set_content_encoding(v);
}
- self.core.set(
- &self.path,
- DashmapValue {
- metadata: meta.clone(),
- content,
- },
- )?;
+ self.core.set(&self.path, DashmapValue {
+ metadata: meta.clone(),
+ content,
+ })?;
Ok(meta)
}
diff --git a/core/src/services/dropbox/writer.rs b/core/src/services/dropbox/writer.rs
index eff7c23cb..30e77808d 100644
--- a/core/src/services/dropbox/writer.rs
+++ b/core/src/services/dropbox/writer.rs
@@ -20,7 +20,8 @@ use std::sync::Arc;
use bytes::Buf;
use http::StatusCode;
-use super::core::{DropboxCore, DropboxMetadataResponse};
+use super::core::DropboxCore;
+use super::core::DropboxMetadataResponse;
use super::error::parse_error;
use crate::raw::*;
use crate::*;
diff --git a/core/src/services/etcd/core.rs b/core/src/services/etcd/core.rs
index a2fa54b6f..6e6d9eb6e 100644
--- a/core/src/services/etcd/core.rs
+++ b/core/src/services/etcd/core.rs
@@ -15,16 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-use crate::services::etcd::error::format_etcd_error;
-use crate::{Buffer, Error, ErrorKind, Result};
+use std::fmt::Debug;
+use std::sync::Arc;
+use std::time::Duration;
+
use etcd_client::Client;
use etcd_client::ConnectOptions;
use fastpool::ManageObject;
use fastpool::ObjectStatus;
use fastpool::bounded;
-use std::fmt::Debug;
-use std::sync::Arc;
-use std::time::Duration;
+
+use crate::Buffer;
+use crate::Error;
+use crate::ErrorKind;
+use crate::Result;
+use crate::services::etcd::error::format_etcd_error;
pub mod constants {
pub const DEFAULT_ETCD_ENDPOINTS: &str = "http://127.0.0.1:2379";
@@ -76,13 +81,10 @@ impl Debug for EtcdCore {
impl EtcdCore {
pub fn new(endpoints: Vec<String>, options: ConnectOptions) -> Self {
- let client = bounded::Pool::new(
- bounded::PoolConfig::new(64),
- Manager {
- endpoints: endpoints.clone(),
- options: options.clone(),
- },
- );
+ let client = bounded::Pool::new(bounded::PoolConfig::new(64), Manager {
+ endpoints: endpoints.clone(),
+ options: options.clone(),
+ });
Self {
endpoints,
diff --git a/core/src/services/etcd/error.rs b/core/src/services/etcd/error.rs
index 1293d9ecd..9d355df28 100644
--- a/core/src/services/etcd/error.rs
+++ b/core/src/services/etcd/error.rs
@@ -17,7 +17,8 @@
use etcd_client::Error as EtcdError;
-use crate::{Error, ErrorKind};
+use crate::Error;
+use crate::ErrorKind;
pub fn format_etcd_error(e: EtcdError) -> Error {
Error::new(ErrorKind::Unexpected, e.to_string().as_str())
diff --git a/core/src/services/etcd/lister.rs b/core/src/services/etcd/lister.rs
index 39d143c3e..a9950a1ba 100644
--- a/core/src/services/etcd/lister.rs
+++ b/core/src/services/etcd/lister.rs
@@ -19,8 +19,10 @@ use std::sync::Arc;
use std::vec::IntoIter;
use super::core::EtcdCore;
+use crate::raw::build_abs_path;
+use crate::raw::build_rel_path;
+use crate::raw::oio;
use crate::raw::oio::Entry;
-use crate::raw::{build_abs_path, build_rel_path, oio};
use crate::*;
pub struct EtcdLister {
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 555487717..8e5e7842b 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -382,13 +382,10 @@ mod build_test {
#[test]
fn from_uri_applies_credentials_from_query() {
- let uri = OperatorUri::new(
- "ftp://example.com/data",
- vec![
- ("user".to_string(), "alice".to_string()),
- ("password".to_string(), "secret".to_string()),
- ],
- )
+ let uri = OperatorUri::new("ftp://example.com/data", vec![
+ ("user".to_string(), "alice".to_string()),
+ ("password".to_string(), "secret".to_string()),
+ ])
.unwrap();
let cfg = FtpConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/ftp/core.rs b/core/src/services/ftp/core.rs
index afe227add..be7fbef44 100644
--- a/core/src/services/ftp/core.rs
+++ b/core/src/services/ftp/core.rs
@@ -15,11 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use fastpool::{ManageObject, ObjectStatus, bounded};
-use futures_rustls::TlsConnector;
-use raw::Operation;
use std::sync::Arc;
use std::time::Duration;
+
+use fastpool::ManageObject;
+use fastpool::ObjectStatus;
+use fastpool::bounded;
+use futures_rustls::TlsConnector;
+use raw::Operation;
use suppaftp::AsyncRustlsConnector;
use suppaftp::AsyncRustlsFtpStream;
use suppaftp::FtpError;
diff --git a/core/src/services/ghac/config.rs b/core/src/services/ghac/config.rs
index 07469326e..1f715c4ae 100644
--- a/core/src/services/ghac/config.rs
+++ b/core/src/services/ghac/config.rs
@@ -111,10 +111,10 @@ mod tests {
#[test]
fn from_uri_respects_version_override() {
- let uri = OperatorUri::new(
- "ghac://cache.githubactions.io/cache-prefix",
- vec![("version".to_string(), "v2".to_string())],
- )
+ let uri = OperatorUri::new("ghac://cache.githubactions.io/cache-prefix", vec![(
+ "version".to_string(),
+ "v2".to_string(),
+ )])
.unwrap();
let cfg = GhacConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/http/config.rs b/core/src/services/http/config.rs
index 196bc8e75..ba929b092 100644
--- a/core/src/services/http/config.rs
+++ b/core/src/services/http/config.rs
@@ -114,13 +114,10 @@ mod tests {
#[test]
fn from_uri_ignores_endpoint_override() {
- let uri = OperatorUri::new(
- "http://example.com/data",
- vec![(
- "endpoint".to_string(),
- "https://cdn.example.com".to_string(),
- )],
- )
+ let uri = OperatorUri::new("http://example.com/data", vec![(
+ "endpoint".to_string(),
+ "https://cdn.example.com".to_string(),
+ )])
.unwrap();
let cfg = HttpConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/huggingface/config.rs b/core/src/services/huggingface/config.rs
index 6b4034bfb..0217d7673 100644
--- a/core/src/services/huggingface/config.rs
+++ b/core/src/services/huggingface/config.rs
@@ -164,10 +164,10 @@ mod tests {
#[test]
fn from_uri_uses_existing_revision_and_sets_root() {
- let uri = OperatorUri::new(
- "huggingface://dataset/opendal/sample/data/train",
- vec![("revision".to_string(), "dev".to_string())],
- )
+ let uri = OperatorUri::new("huggingface://dataset/opendal/sample/data/train", vec![(
+ "revision".to_string(),
+ "dev".to_string(),
+ )])
.unwrap();
let cfg = HuggingfaceConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/huggingface/core.rs b/core/src/services/huggingface/core.rs
index 6aa635c59..f3b33414d 100644
--- a/core/src/services/huggingface/core.rs
+++ b/core/src/services/huggingface/core.rs
@@ -22,7 +22,8 @@ use bytes::Bytes;
use http::Request;
use http::Response;
use http::header;
-use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
+use percent_encoding::NON_ALPHANUMERIC;
+use percent_encoding::utf8_percent_encode;
use serde::Deserialize;
use super::backend::RepoType;
@@ -248,13 +249,17 @@ pub(super) struct HuggingfaceImport {
#[cfg(test)]
mod tests {
+ use std::sync::Arc;
+ use std::sync::Mutex;
+
use bytes::Bytes;
+ use http::Request;
+ use http::Response;
+ use http::StatusCode;
use super::*;
use crate::raw::new_json_deserialize_error;
use crate::types::Result;
- use http::{Request, Response, StatusCode};
- use std::sync::{Arc, Mutex};
// Mock HTTP client that captures the request URL and headers
#[derive(Clone)]
diff --git a/core/src/services/ipmfs/builder.rs b/core/src/services/ipmfs/builder.rs
index 975bcaa20..7907c1c85 100644
--- a/core/src/services/ipmfs/builder.rs
+++ b/core/src/services/ipmfs/builder.rs
@@ -52,8 +52,8 @@ use crate::*;
///
/// ```no_run
/// use anyhow::Result;
-/// use opendal::services::Ipmfs;
/// use opendal::Operator;
+/// use opendal::services::Ipmfs;
///
/// #[tokio::main]
/// async fn main() -> Result<()> {
diff --git a/core/src/services/lakefs/config.rs b/core/src/services/lakefs/config.rs
index d6c374e3d..430eb23ac 100644
--- a/core/src/services/lakefs/config.rs
+++ b/core/src/services/lakefs/config.rs
@@ -167,10 +167,10 @@ mod tests {
#[test]
fn from_uri_respects_branch_override_and_sets_root() {
- let uri = OperatorUri::new(
- "lakefs://api.example.com/sample/content",
- vec![("branch".to_string(), "develop".to_string())],
- )
+ let uri = OperatorUri::new("lakefs://api.example.com/sample/content", vec![(
+ "branch".to_string(),
+ "develop".to_string(),
+ )])
.unwrap();
let cfg = LakefsConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/lakefs/lister.rs b/core/src/services/lakefs/lister.rs
index fe49f1529..20e4f33b0 100644
--- a/core/src/services/lakefs/lister.rs
+++ b/core/src/services/lakefs/lister.rs
@@ -17,12 +17,13 @@
use std::sync::Arc;
+use bytes::Buf;
+
use super::core::LakefsCore;
use super::core::LakefsListResponse;
use super::error::parse_error;
use crate::raw::*;
use crate::*;
-use bytes::Buf;
pub struct LakefsLister {
core: Arc<LakefsCore>,
diff --git a/core/src/services/memcached/core.rs b/core/src/services/memcached/core.rs
index c88d7b0be..1946fb83f 100644
--- a/core/src/services/memcached/core.rs
+++ b/core/src/services/memcached/core.rs
@@ -15,9 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-use fastpool::{ManageObject, ObjectStatus, bounded};
use std::sync::Arc;
use std::time::Duration;
+
+use fastpool::ManageObject;
+use fastpool::ObjectStatus;
+use fastpool::bounded;
use tokio::net::TcpStream;
use super::binary;
diff --git a/core/src/services/mini_moka/config.rs b/core/src/services/mini_moka/config.rs
index 8106858f2..9c82004df 100644
--- a/core/src/services/mini_moka/config.rs
+++ b/core/src/services/mini_moka/config.rs
@@ -70,10 +70,10 @@ mod tests {
#[test]
fn from_uri_sets_root_and_preserves_ttl() {
- let uri = OperatorUri::new(
- "mini-moka:///session",
- vec![("time_to_live".to_string(), "300s".to_string())],
- )
+ let uri = OperatorUri::new("mini-moka:///session", vec![(
+ "time_to_live".to_string(),
+ "300s".to_string(),
+ )])
.unwrap();
let cfg = MiniMokaConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/mini_moka/writer.rs b/core/src/services/mini_moka/writer.rs
index aafdacfec..0c331b12e 100644
--- a/core/src/services/mini_moka/writer.rs
+++ b/core/src/services/mini_moka/writer.rs
@@ -15,11 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use super::core::{MiniMokaCore, MiniMokaValue};
+use std::sync::Arc;
+
+use super::core::MiniMokaCore;
+use super::core::MiniMokaValue;
use crate::raw::oio;
use crate::raw::*;
use crate::*;
-use std::sync::Arc;
pub struct MiniMokaWriter {
core: Arc<MiniMokaCore>,
diff --git a/core/src/services/moka/backend.rs b/core/src/services/moka/backend.rs
index ee28e15b9..fdc592b78 100644
--- a/core/src/services/moka/backend.rs
+++ b/core/src/services/moka/backend.rs
@@ -78,9 +78,11 @@ impl MokaBuilder {
/// .eviction_listener(|k: Arc<String>, v: MokaValue, cause: RemovalCause| {
/// debug!(
/// "moka cache eviction listener, key = {}, value = {:?}, cause = {:?}",
- /// k.as_str(), v.content.to_vec(), cause
+ /// k.as_str(),
+ /// v.content.to_vec(),
+ /// cause
/// );
- /// })
+ /// }),
/// );
/// ```
pub fn new(builder: MokaCacheBuilder<String, MokaValue>) -> Self {
diff --git a/core/src/services/moka/mod.rs b/core/src/services/moka/mod.rs
index ea5bfc216..588daf085 100644
--- a/core/src/services/moka/mod.rs
+++ b/core/src/services/moka/mod.rs
@@ -25,7 +25,8 @@ mod deleter;
mod lister;
mod writer;
+pub use core::MokaValue;
+
pub use backend::MokaBuilder as Moka;
pub use backend::MokaCacheBuilder;
pub use config::MokaConfig;
-pub use core::MokaValue;
diff --git a/core/src/services/moka/writer.rs b/core/src/services/moka/writer.rs
index f1bdf09ab..1f6fc0fde 100644
--- a/core/src/services/moka/writer.rs
+++ b/core/src/services/moka/writer.rs
@@ -17,7 +17,8 @@
use std::sync::Arc;
-use super::core::{MokaCore, MokaValue};
+use super::core::MokaCore;
+use super::core::MokaValue;
use crate::raw::oio;
use crate::raw::*;
use crate::*;
diff --git a/core/src/services/obs/core.rs b/core/src/services/obs/core.rs
index 33f03f549..938bf898c 100644
--- a/core/src/services/obs/core.rs
+++ b/core/src/services/obs/core.rs
@@ -592,10 +592,9 @@ mod tests {
.collect::<Vec<String>>(),
["obj002", "obj003"],
);
- assert_eq!(
- out.contents.iter().map(|v| v.size).collect::<Vec<u64>>(),
- [9, 10],
- );
+ assert_eq!(out.contents.iter().map(|v| v.size).collect::<Vec<u64>>(), [
+ 9, 10
+ ],);
assert_eq!(
out.common_prefixes
.iter()
diff --git a/core/src/services/opfs/core.rs b/core/src/services/opfs/core.rs
index 643af84bd..ecb51c5f7 100644
--- a/core/src/services/opfs/core.rs
+++ b/core/src/services/opfs/core.rs
@@ -22,11 +22,10 @@ use wasm_bindgen_futures::JsFuture;
use web_sys::File;
use web_sys::FileSystemWritableFileStream;
-use crate::Error;
-use crate::Result;
-
use super::error::*;
use super::utils::*;
+use crate::Error;
+use crate::Result;
#[derive(Default, Debug)]
pub struct OpfsCore {}
diff --git a/core/src/services/opfs/error.rs b/core/src/services/opfs/error.rs
index 1f7d5cb87..8507ca797 100644
--- a/core/src/services/opfs/error.rs
+++ b/core/src/services/opfs/error.rs
@@ -17,7 +17,8 @@
use wasm_bindgen::JsValue;
-use crate::{Error, ErrorKind};
+use crate::Error;
+use crate::ErrorKind;
pub(crate) fn parse_js_error(msg: JsValue) -> Error {
Error::new(
diff --git a/core/src/services/opfs/utils.rs b/core/src/services/opfs/utils.rs
index 9cc3bd25e..69c163e7a 100644
--- a/core/src/services/opfs/utils.rs
+++ b/core/src/services/opfs/utils.rs
@@ -15,15 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use crate::Result;
use wasm_bindgen::JsCast;
use wasm_bindgen_futures::JsFuture;
-use web_sys::{
- FileSystemDirectoryHandle, FileSystemFileHandle, FileSystemGetDirectoryOptions,
- FileSystemGetFileOptions, window,
-};
+use web_sys::FileSystemDirectoryHandle;
+use web_sys::FileSystemFileHandle;
+use web_sys::FileSystemGetDirectoryOptions;
+use web_sys::FileSystemGetFileOptions;
+use web_sys::window;
use super::error::*;
+use crate::Result;
pub(crate) async fn get_root_directory_handle() -> Result<FileSystemDirectoryHandle> {
let navigator = window().unwrap().navigator();
diff --git a/core/src/services/oss/core.rs b/core/src/services/oss/core.rs
index 78423dd76..0061e801f 100644
--- a/core/src/services/oss/core.rs
+++ b/core/src/services/oss/core.rs
@@ -1054,28 +1054,25 @@ mod tests {
assert_eq!(out.next_continuation_token, Some("CgJiYw--".to_string()));
assert!(out.common_prefixes.is_empty());
- assert_eq!(
- out.contents,
- vec![
- ListObjectsOutputContent {
- key: "b/c".to_string(),
- last_modified: "2020-05-18T05:45:54.000Z".to_string(),
- etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
- size: 25,
- },
- ListObjectsOutputContent {
- key: "ba".to_string(),
- last_modified: "2020-05-18T11:17:58.000Z".to_string(),
- etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
- size: 25,
- },
- ListObjectsOutputContent {
- key: "bc".to_string(),
- last_modified: "2020-05-18T05:45:59.000Z".to_string(),
- etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
- size: 25,
- }
- ]
- )
+ assert_eq!(out.contents, vec![
+ ListObjectsOutputContent {
+ key: "b/c".to_string(),
+ last_modified: "2020-05-18T05:45:54.000Z".to_string(),
+ etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
+ size: 25,
+ },
+ ListObjectsOutputContent {
+ key: "ba".to_string(),
+ last_modified: "2020-05-18T11:17:58.000Z".to_string(),
+ etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
+ size: 25,
+ },
+ ListObjectsOutputContent {
+ key: "bc".to_string(),
+ last_modified: "2020-05-18T05:45:59.000Z".to_string(),
+ etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
+ size: 25,
+ }
+ ])
}
}
diff --git a/core/src/services/pcloud/config.rs b/core/src/services/pcloud/config.rs
index d8a05c24a..f5072dba9 100644
--- a/core/src/services/pcloud/config.rs
+++ b/core/src/services/pcloud/config.rs
@@ -88,10 +88,10 @@ mod tests {
#[test]
fn from_uri_sets_endpoint_and_root() {
- let uri = OperatorUri::new(
- "pcloud://api.pcloud.com/drive/photos",
- vec![("username".to_string(), "alice".to_string())],
- )
+ let uri = OperatorUri::new("pcloud://api.pcloud.com/drive/photos", vec![(
+ "username".to_string(),
+ "alice".to_string(),
+ )])
.unwrap();
let cfg = PcloudConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/redis/core.rs b/core/src/services/redis/core.rs
index f0f5de20a..127e831ad 100644
--- a/core/src/services/redis/core.rs
+++ b/core/src/services/redis/core.rs
@@ -19,9 +19,10 @@ use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
-use crate::*;
use bytes::Bytes;
-use fastpool::{ManageObject, ObjectStatus, bounded};
+use fastpool::ManageObject;
+use fastpool::ObjectStatus;
+use fastpool::bounded;
use redis::AsyncCommands;
use redis::Client;
use redis::Cmd;
@@ -34,6 +35,8 @@ use redis::aio::ConnectionManager;
use redis::cluster::ClusterClient;
use redis::cluster_async::ClusterConnection;
+use crate::*;
+
#[derive(Clone)]
pub enum RedisConnection {
Normal(ConnectionManager),
diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs
index 28576e4cd..fa2b39b95 100644
--- a/core/src/services/s3/core.rs
+++ b/core/src/services/s3/core.rs
@@ -1235,14 +1235,10 @@ impl ChecksumAlgorithm {
}
impl Display for ChecksumAlgorithm {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{}",
- match self {
- Self::Crc32c => "CRC32C",
- Self::Md5 => "MD5",
- }
- )
+ write!(f, "{}", match self {
+ Self::Crc32c => "CRC32C",
+ Self::Md5 => "MD5",
+ })
}
}
@@ -1497,23 +1493,20 @@ mod tests {
assert!(!out.is_truncated.unwrap());
assert!(out.next_marker.is_none());
assert!(out.common_prefixes.is_empty());
- assert_eq!(
- out.contents,
- vec![
- ListObjectsOutputContent {
- key: "my-image.jpg".to_string(),
- size: 434234,
- etag: Some("\"fba9dede5f27731c9771645a39863328\"".to_string()),
- last_modified: "2009-10-12T17:50:30.000Z".to_string(),
- },
- ListObjectsOutputContent {
- key: "my-third-image.jpg".to_string(),
- size: 64994,
- last_modified: "2009-10-12T17:50:30.000Z".to_string(),
- etag: Some("\"1b2cf535f27731c974343645a3985328\"".to_string()),
- },
- ]
- )
+ assert_eq!(out.contents, vec![
+ ListObjectsOutputContent {
+ key: "my-image.jpg".to_string(),
+ size: 434234,
+ etag: Some("\"fba9dede5f27731c9771645a39863328\"".to_string()),
+ last_modified: "2009-10-12T17:50:30.000Z".to_string(),
+ },
+ ListObjectsOutputContent {
+ key: "my-third-image.jpg".to_string(),
+ size: 64994,
+ last_modified: "2009-10-12T17:50:30.000Z".to_string(),
+ etag: Some("\"1b2cf535f27731c974343645a3985328\"".to_string()),
+ },
+ ])
}
#[test]
@@ -1567,29 +1560,26 @@ mod tests {
.collect::<Vec<String>>(),
vec!["photos/2006/February/", "photos/2006/January/"]
);
- assert_eq!(
- out.contents,
- vec![
- ListObjectsOutputContent {
- key: "photos/2006".to_string(),
- size: 56,
- etag: Some("\"d41d8cd98f00b204e9800998ecf8427e\"".to_string()),
- last_modified: "2016-04-30T23:51:29.000Z".to_string(),
- },
- ListObjectsOutputContent {
- key: "photos/2007".to_string(),
- size: 100,
- last_modified: "2016-04-30T23:51:29.000Z".to_string(),
- etag: Some("\"d41d8cd98f00b204e9800998ecf8427e\"".to_string()),
- },
- ListObjectsOutputContent {
- key: "photos/2008".to_string(),
- size: 42,
- last_modified: "2016-05-30T23:51:29.000Z".to_string(),
- etag: None,
- },
- ]
- )
+ assert_eq!(out.contents, vec![
+ ListObjectsOutputContent {
+ key: "photos/2006".to_string(),
+ size: 56,
+ etag: Some("\"d41d8cd98f00b204e9800998ecf8427e\"".to_string()),
+ last_modified: "2016-04-30T23:51:29.000Z".to_string(),
+ },
+ ListObjectsOutputContent {
+ key: "photos/2007".to_string(),
+ size: 100,
+ last_modified: "2016-04-30T23:51:29.000Z".to_string(),
+ etag: Some("\"d41d8cd98f00b204e9800998ecf8427e\"".to_string()),
+ },
+ ListObjectsOutputContent {
+ key: "photos/2008".to_string(),
+ size: 42,
+ last_modified: "2016-05-30T23:51:29.000Z".to_string(),
+ etag: None,
+ },
+ ])
}
#[test]
@@ -1657,48 +1647,41 @@ mod tests {
output.next_version_id_marker,
Some("d-d309mfjFrUmoQ0DBsVqmcMV15OI.".to_owned())
);
- assert_eq!(
- output.common_prefixes,
- vec![
- OutputCommonPrefix {
- prefix: "photos/".to_owned()
- },
- OutputCommonPrefix {
- prefix: "videos/".to_owned()
- }
- ]
- );
+ assert_eq!(output.common_prefixes, vec![
+ OutputCommonPrefix {
+ prefix: "photos/".to_owned()
+ },
+ OutputCommonPrefix {
+ prefix: "videos/".to_owned()
+ }
+ ]);
- assert_eq!(
- output.version,
- vec![
- ListObjectVersionsOutputVersion {
- key: "key3".to_owned(),
- version_id: "8XECiENpj8pydEDJdd-_VRrvaGKAHOaGMNW7tg6UViI.".to_owned(),
- is_latest: true,
- size: 217,
- last_modified: "2009-12-09T00:18:23.000Z".to_owned(),
- etag: Some("\"396fefef536d5ce46c7537ecf978a360\"".to_owned()),
- },
- ListObjectVersionsOutputVersion {
- key: "key3".to_owned(),
- version_id: "d-d309mfjFri40QYukDozqBt3UmoQ0DBsVqmcMV15OI.".to_owned(),
- is_latest: false,
- size: 217,
- last_modified: "2009-12-09T00:18:08.000Z".to_owned(),
- etag: Some("\"396fefef536d5ce46c7537ecf978a360\"".to_owned()),
- }
- ]
- );
+ assert_eq!(output.version, vec![
+ ListObjectVersionsOutputVersion {
+ key: "key3".to_owned(),
+ version_id: "8XECiENpj8pydEDJdd-_VRrvaGKAHOaGMNW7tg6UViI.".to_owned(),
+ is_latest: true,
+ size: 217,
+ last_modified: "2009-12-09T00:18:23.000Z".to_owned(),
+ etag: Some("\"396fefef536d5ce46c7537ecf978a360\"".to_owned()),
+ },
+ ListObjectVersionsOutputVersion {
+ key: "key3".to_owned(),
+ version_id: "d-d309mfjFri40QYukDozqBt3UmoQ0DBsVqmcMV15OI.".to_owned(),
+ is_latest: false,
+ size: 217,
+ last_modified: "2009-12-09T00:18:08.000Z".to_owned(),
+ etag: Some("\"396fefef536d5ce46c7537ecf978a360\"".to_owned()),
+ }
+ ]);
- assert_eq!(
- output.delete_marker,
- vec![ListObjectVersionsOutputDeleteMarker {
+ assert_eq!(output.delete_marker, vec![
+ ListObjectVersionsOutputDeleteMarker {
key: "my-third-image.jpg".to_owned(),
version_id: "03jpff543dhffds434rfdsFDN943fdsFkdmqnh892".to_owned(),
is_latest: true,
last_modified: "2009-10-15T17:50:30.000Z".to_owned(),
- },]
- );
+ },
+ ]);
}
}
diff --git a/core/src/services/sftp/config.rs b/core/src/services/sftp/config.rs
index 999da4378..82ad894a4 100644
--- a/core/src/services/sftp/config.rs
+++ b/core/src/services/sftp/config.rs
@@ -96,14 +96,11 @@ mod tests {
#[test]
fn from_uri_applies_connection_overrides() {
- let uri = OperatorUri::new(
- "sftp://host",
- vec![
- ("user".to_string(), "alice".to_string()),
- ("key".to_string(), "/home/alice/.ssh/id_rsa".to_string()),
- ("known_hosts_strategy".to_string(), "accept".to_string()),
- ],
- )
+ let uri = OperatorUri::new("sftp://host", vec![
+ ("user".to_string(), "alice".to_string()),
+ ("key".to_string(), "/home/alice/.ssh/id_rsa".to_string()),
+ ("known_hosts_strategy".to_string(), "accept".to_string()),
+ ])
.unwrap();
let cfg = SftpConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/sftp/core.rs b/core/src/services/sftp/core.rs
index 2520a03d9..754005309 100644
--- a/core/src/services/sftp/core.rs
+++ b/core/src/services/sftp/core.rs
@@ -15,23 +15,27 @@
// specific language governing permissions and limitations
// under the License.
-use super::error::is_sftp_protocol_error;
-use super::error::parse_sftp_error;
-use super::error::parse_ssh_error;
-use crate::raw::*;
-use crate::*;
-use fastpool::{ManageObject, ObjectStatus, bounded};
-use log::debug;
-use openssh::KnownHosts;
-use openssh::SessionBuilder;
-use openssh_sftp_client::Sftp;
-use openssh_sftp_client::SftpOptions;
use std::fmt::Debug;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
+use fastpool::ManageObject;
+use fastpool::ObjectStatus;
+use fastpool::bounded;
+use log::debug;
+use openssh::KnownHosts;
+use openssh::SessionBuilder;
+use openssh_sftp_client::Sftp;
+use openssh_sftp_client::SftpOptions;
+
+use super::error::is_sftp_protocol_error;
+use super::error::parse_sftp_error;
+use super::error::parse_ssh_error;
+use crate::raw::*;
+use crate::*;
+
pub struct SftpCore {
pub info: Arc<AccessorInfo>,
pub endpoint: String,
@@ -57,16 +61,13 @@ impl SftpCore {
key: Option<String>,
known_hosts_strategy: KnownHosts,
) -> Self {
- let client = bounded::Pool::new(
- bounded::PoolConfig::new(64),
- Manager {
- endpoint: endpoint.clone(),
- root: root.clone(),
- user,
- key,
- known_hosts_strategy,
- },
- );
+ let client = bounded::Pool::new(bounded::PoolConfig::new(64), Manager {
+ endpoint: endpoint.clone(),
+ root: root.clone(),
+ user,
+ key,
+ known_hosts_strategy,
+ });
SftpCore {
info,
diff --git a/core/src/services/sqlite/backend.rs b/core/src/services/sqlite/backend.rs
index 8b05e765b..bb453187f 100644
--- a/core/src/services/sqlite/backend.rs
+++ b/core/src/services/sqlite/backend.rs
@@ -313,9 +313,10 @@ impl Access for SqliteBackend {
#[cfg(test)]
mod test {
- use super::*;
use sqlx::SqlitePool;
+ use super::*;
+
async fn build_client() -> OnceCell<SqlitePool> {
let config = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
let pool = SqlitePool::connect_with(config).await.unwrap();
diff --git a/core/src/services/sqlite/core.rs b/core/src/services/sqlite/core.rs
index fa55643e9..a533a9db1 100644
--- a/core/src/services/sqlite/core.rs
+++ b/core/src/services/sqlite/core.rs
@@ -15,10 +15,11 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::Debug;
+
use mea::once::OnceCell;
use sqlx::SqlitePool;
use sqlx::sqlite::SqliteConnectOptions;
-use std::fmt::Debug;
use crate::services::sqlite::backend::parse_sqlite_error;
use crate::*;
diff --git a/core/src/services/sqlite/deleter.rs b/core/src/services/sqlite/deleter.rs
index 7b3ee442c..d707c58a6 100644
--- a/core/src/services/sqlite/deleter.rs
+++ b/core/src/services/sqlite/deleter.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::raw::{OpDelete, build_abs_path, oio};
+use crate::raw::OpDelete;
+use crate::raw::build_abs_path;
+use crate::raw::oio;
use crate::services::sqlite::core::SqliteCore;
pub struct SqliteDeleter {
diff --git a/core/src/services/sqlite/writer.rs b/core/src/services/sqlite/writer.rs
index bda34b255..0b45a5c6e 100644
--- a/core/src/services/sqlite/writer.rs
+++ b/core/src/services/sqlite/writer.rs
@@ -15,9 +15,11 @@
// specific language governing permissions and limitations
// under the License.
+use crate::Buffer;
+use crate::EntryMode;
+use crate::Metadata;
use crate::raw::oio;
use crate::services::sqlite::core::SqliteCore;
-use crate::{Buffer, EntryMode, Metadata};
pub struct SqliteWriter {
core: std::sync::Arc<SqliteCore>,
diff --git a/core/src/services/swift/config.rs b/core/src/services/swift/config.rs
index 14fd0c58e..764fb156e 100644
--- a/core/src/services/swift/config.rs
+++ b/core/src/services/swift/config.rs
@@ -115,10 +115,10 @@ mod tests {
#[test]
fn from_uri_accepts_container_from_query() {
- let uri = OperatorUri::new(
- "swift://swift.example.com",
- vec![("container".to_string(), "logs".to_string())],
- )
+ let uri = OperatorUri::new("swift://swift.example.com", vec![(
+ "container".to_string(),
+ "logs".to_string(),
+ )])
.unwrap();
let cfg = SwiftConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/swift/core.rs b/core/src/services/swift/core.rs
index acff08ec6..9f665e195 100644
--- a/core/src/services/swift/core.rs
+++ b/core/src/services/swift/core.rs
@@ -280,30 +280,21 @@ mod tests {
.map_err(new_json_deserialize_error)?;
assert_eq!(out.len(), 3);
- assert_eq!(
- out.pop().unwrap(),
- ListOpResponse::FileInfo {
- bytes: 147,
- hash: "5e6b5b70b0426b1cc1968003e1afa5ad".to_string(),
- name: "test.txt".to_string(),
- last_modified: "2023-11-01T03:00:23.147480".to_string(),
- content_type: Some("text/plain".to_string()),
- }
- );
-
- assert_eq!(
- out.pop().unwrap(),
- ListOpResponse::Subdir {
- subdir: "fruit/".to_string()
- }
- );
-
- assert_eq!(
- out.pop().unwrap(),
- ListOpResponse::Subdir {
- subdir: "animals/".to_string()
- }
- );
+ assert_eq!(out.pop().unwrap(), ListOpResponse::FileInfo {
+ bytes: 147,
+ hash: "5e6b5b70b0426b1cc1968003e1afa5ad".to_string(),
+ name: "test.txt".to_string(),
+ last_modified: "2023-11-01T03:00:23.147480".to_string(),
+ content_type: Some("text/plain".to_string()),
+ });
+
+ assert_eq!(out.pop().unwrap(), ListOpResponse::Subdir {
+ subdir: "fruit/".to_string()
+ });
+
+ assert_eq!(out.pop().unwrap(), ListOpResponse::Subdir {
+ subdir: "animals/".to_string()
+ });
Ok(())
}
diff --git a/core/src/services/vercel_artifacts/config.rs b/core/src/services/vercel_artifacts/config.rs
index 864d9e7c1..60fc88b22 100644
--- a/core/src/services/vercel_artifacts/config.rs
+++ b/core/src/services/vercel_artifacts/config.rs
@@ -62,10 +62,10 @@ mod tests {
#[test]
fn from_uri_loads_access_token() {
- let uri = OperatorUri::new(
- "vercel-artifacts://cache",
- vec![("access_token".to_string(), "token123".to_string())],
- )
+ let uri = OperatorUri::new("vercel-artifacts://cache", vec![(
+ "access_token".to_string(),
+ "token123".to_string(),
+ )])
.unwrap();
let cfg = VercelArtifactsConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/vercel_blob/deleter.rs b/core/src/services/vercel_blob/deleter.rs
index 77f67c3b5..c71fb0a19 100644
--- a/core/src/services/vercel_blob/deleter.rs
+++ b/core/src/services/vercel_blob/deleter.rs
@@ -19,7 +19,8 @@ use std::sync::Arc;
use super::core::VercelBlobCore;
use crate::Result;
-use crate::raw::{OpDelete, oio};
+use crate::raw::OpDelete;
+use crate::raw::oio;
pub struct VercelBlobDeleter {
core: Arc<VercelBlobCore>,
diff --git a/core/src/services/webdav/config.rs b/core/src/services/webdav/config.rs
index de1c9c76a..d02f34858 100644
--- a/core/src/services/webdav/config.rs
+++ b/core/src/services/webdav/config.rs
@@ -102,13 +102,10 @@ mod tests {
#[test]
fn from_uri_ignores_endpoint_override() {
- let uri = OperatorUri::new(
- "webdav://dav.internal/data",
- vec![(
- "endpoint".to_string(),
- "http://dav.internal:8080".to_string(),
- )],
- )
+ let uri = OperatorUri::new("webdav://dav.internal/data", vec![(
+ "endpoint".to_string(),
+ "http://dav.internal:8080".to_string(),
+ )])
.unwrap();
let cfg = WebdavConfig::from_uri(&uri).unwrap();
@@ -117,10 +114,10 @@ mod tests {
#[test]
fn from_uri_propagates_disable_copy() {
- let uri = OperatorUri::new(
- "webdav://dav.example.com",
- vec![("disable_copy".to_string(), "true".to_string())],
- )
+ let uri = OperatorUri::new("webdav://dav.example.com", vec![(
+ "disable_copy".to_string(),
+ "true".to_string(),
+ )])
.unwrap();
let cfg = WebdavConfig::from_uri(&uri).unwrap();
diff --git a/core/src/services/yandex_disk/config.rs b/core/src/services/yandex_disk/config.rs
index b128d6272..3dc1eef10 100644
--- a/core/src/services/yandex_disk/config.rs
+++ b/core/src/services/yandex_disk/config.rs
@@ -75,10 +75,10 @@ mod tests {
#[test]
fn from_uri_sets_root_and_preserves_token() {
- let uri = OperatorUri::new(
- "yandex-disk://disk/root/path",
- vec![("access_token".to_string(), "secret".to_string())],
- )
+ let uri = OperatorUri::new("yandex-disk://disk/root/path", vec![(
+ "access_token".to_string(),
+ "secret".to_string(),
+ )])
.unwrap();
let cfg = YandexDiskConfig::from_uri(&uri).unwrap();
diff --git a/core/src/types/buffer.rs b/core/src/types/buffer.rs
index 26c4110fa..5456bc097 100644
--- a/core/src/types/buffer.rs
+++ b/core/src/types/buffer.rs
@@ -860,10 +860,11 @@ mod tests {
.map(|chunk| chunk.to_bytes())
.collect::<Vec<Bytes>>();
- assert_eq!(
- chunks,
- vec![Bytes::from("abc"), Bytes::from("def"), Bytes::from("g")]
- );
+ assert_eq!(chunks, vec![
+ Bytes::from("abc"),
+ Bytes::from("def"),
+ Bytes::from("g")
+ ]);
assert_eq!(Buffer::new().chunks(4).count(), 0);
}
@@ -881,10 +882,11 @@ mod tests {
.map(|chunk| chunk.to_bytes())
.collect::<Vec<Bytes>>();
- assert_eq!(
- chunks,
- vec![Bytes::from("ab"), Bytes::from("cd"), Bytes::from("ef"),]
- );
+ assert_eq!(chunks, vec![
+ Bytes::from("ab"),
+ Bytes::from("cd"),
+ Bytes::from("ef"),
+ ]);
}
#[test]
diff --git a/core/src/types/builder.rs b/core/src/types/builder.rs
index 903459775..f9fa12908 100644
--- a/core/src/types/builder.rs
+++ b/core/src/types/builder.rs
@@ -37,8 +37,8 @@ use crate::*;
///
/// ```
/// # use anyhow::Result;
-/// use opendal::services::Fs;
/// use opendal::Operator;
+/// use opendal::services::Fs;
/// async fn test() -> Result<()> {
/// // Create fs backend builder.
/// let mut builder = Fs::default().root("/tmp");
@@ -81,8 +81,8 @@ impl Builder for () {
/// # use anyhow::Result;
/// use std::collections::HashMap;
///
-/// use opendal::services::MemoryConfig;
/// use opendal::Operator;
+/// use opendal::services::MemoryConfig;
/// async fn test() -> Result<()> {
/// let mut cfg = MemoryConfig::default();
/// cfg.root = Some("/".to_string());
@@ -101,10 +101,10 @@ impl Builder for () {
/// # use anyhow::Result;
/// use std::collections::HashMap;
///
-/// use opendal::raw::HttpClient;
-/// use opendal::services::S3Config;
/// use opendal::Configurator;
/// use opendal::Operator;
+/// use opendal::raw::HttpClient;
+/// use opendal::services::S3Config;
///
/// async fn test() -> Result<()> {
/// let mut cfg = S3Config::default();
diff --git a/core/src/types/delete/deleter.rs b/core/src/types/delete/deleter.rs
index 4675b9751..c11ef6694 100644
--- a/core/src/types/delete/deleter.rs
+++ b/core/src/types/delete/deleter.rs
@@ -68,9 +68,9 @@ use crate::*;
/// Deleter can be used as a Sink for file deletion:
///
/// ```rust
-/// use futures::stream;
/// use futures::Sink;
/// use futures::SinkExt;
+/// use futures::stream;
/// use opendal::Operator;
/// use opendal::Result;
///
diff --git a/core/src/types/metadata.rs b/core/src/types/metadata.rs
index 901979dc4..53e642b8b 100644
--- a/core/src/types/metadata.rs
+++ b/core/src/types/metadata.rs
@@ -15,9 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
+
use crate::raw::*;
use crate::*;
-use std::collections::HashMap;
/// Metadata contains all the information related to a specific path.
///
diff --git a/core/src/types/operator/builder.rs b/core/src/types/operator/builder.rs
index 1954ea85d..c789fdc44 100644
--- a/core/src/types/operator/builder.rs
+++ b/core/src/types/operator/builder.rs
@@ -28,8 +28,8 @@ use crate::*;
///
/// ```
/// # use anyhow::Result;
-/// use opendal::services::Fs;
/// use opendal::Operator;
+/// use opendal::services::Fs;
/// async fn test() -> Result<()> {
/// // Create fs backend builder.
/// let builder = Fs::default().root("/tmp");
@@ -52,8 +52,8 @@ impl Operator {
///
/// ```
/// # use anyhow::Result;
- /// use opendal::services::Fs;
/// use opendal::Operator;
+ /// use opendal::services::Fs;
/// async fn test() -> Result<()> {
/// // Create fs backend builder.
/// let builder = Fs::default().root("/tmp");
@@ -78,8 +78,8 @@ impl Operator {
/// # use anyhow::Result;
/// use std::collections::HashMap;
///
- /// use opendal::services::MemoryConfig;
/// use opendal::Operator;
+ /// use opendal::services::MemoryConfig;
/// async fn test() -> Result<()> {
/// let cfg = MemoryConfig::default();
///
@@ -107,8 +107,8 @@ impl Operator {
/// # use anyhow::Result;
/// use std::collections::HashMap;
///
- /// use opendal::services::Fs;
/// use opendal::Operator;
+ /// use opendal::services::Fs;
/// async fn test() -> Result<()> {
/// let map = HashMap::from([
/// // Set the root for fs, all operations will happen under this root.
@@ -350,9 +350,9 @@ impl Operator {
/// ```no_run
/// # use std::sync::Arc;
/// # use anyhow::Result;
+ /// use opendal::Operator;
/// use opendal::layers::LoggingLayer;
/// use opendal::services::Memory;
- /// use opendal::Operator;
///
/// # async fn test() -> Result<()> {
/// let op = Operator::new(Memory::default())?.finish();
@@ -387,12 +387,12 @@ impl Operator {
/// ```
/// use std::collections::HashMap;
///
-/// use opendal::layers::LoggingLayer;
-/// use opendal::layers::RetryLayer;
-/// use opendal::services;
/// use opendal::Builder;
/// use opendal::Operator;
/// use opendal::Result;
+/// use opendal::layers::LoggingLayer;
+/// use opendal::layers::RetryLayer;
+/// use opendal::services;
///
/// fn init_service<B: Builder>(cfg: HashMap<String, String>) -> Result<Operator> {
/// let op = Operator::from_iter::<B>(cfg)?
@@ -443,9 +443,9 @@ impl<A: Access> OperatorBuilder<A> {
/// ```no_run
/// # use std::sync::Arc;
/// # use anyhow::Result;
+ /// use opendal::Operator;
/// use opendal::layers::LoggingLayer;
/// use opendal::services::Memory;
- /// use opendal::Operator;
///
/// # async fn test() -> Result<()> {
/// let op = Operator::new(Memory::default())?
diff --git a/core/src/types/operator/mod.rs b/core/src/types/operator/mod.rs
index 3de1ea7c1..6d8faf185 100644
--- a/core/src/types/operator/mod.rs
+++ b/core/src/types/operator/mod.rs
@@ -30,7 +30,10 @@ pub use info::OperatorInfo;
pub mod operator_futures;
mod registry;
-pub use registry::{DEFAULT_OPERATOR_REGISTRY, OperatorFactory, OperatorRegistry};
+pub use registry::DEFAULT_OPERATOR_REGISTRY;
+pub use registry::OperatorFactory;
+pub use registry::OperatorRegistry;
mod uri;
-pub use uri::{IntoOperatorUri, OperatorUri};
+pub use uri::IntoOperatorUri;
+pub use uri::OperatorUri;
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 5f255ab52..0ff8ab20f 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -46,8 +46,8 @@ use crate::*;
///
/// ```
/// # use anyhow::Result;
-/// use opendal::services::Memory;
/// use opendal::Operator;
+/// use opendal::services::Memory;
/// async fn test() -> Result<()> {
/// // Build an `Operator` to start operating the storage.
/// let _: Operator = Operator::new(Memory::default())?.finish();
@@ -69,9 +69,9 @@ use crate::*;
///
/// ```
/// # use anyhow::Result;
+/// use opendal::Operator;
/// use opendal::layers::RetryLayer;
/// use opendal::services::Memory;
-/// use opendal::Operator;
/// async fn test() -> Result<()> {
/// let op: Operator = Operator::new(Memory::default())?.finish();
///
@@ -102,11 +102,11 @@ use crate::*;
/// into [`futures::AsyncRead`] or [`futures::Stream`] for broader ecosystem compatibility.
///
/// ```no_run
+/// use opendal::Operator;
+/// use opendal::Result;
/// use opendal::layers::LoggingLayer;
/// use opendal::options;
/// use opendal::services;
-/// use opendal::Operator;
-/// use opendal::Result;
///
/// #[tokio::main]
/// async fn main() -> Result<()> {
@@ -372,8 +372,8 @@ impl Operator {
/// # use anyhow::Result;
/// # use futures::io;
/// # use opendal::Operator;
- /// use opendal::options;
/// use opendal::ErrorKind;
+ /// use opendal::options;
/// #
/// # async fn test(op: Operator) -> Result<()> {
/// let res = op
@@ -1134,7 +1134,8 @@ impl Operator {
/// # async fn test(op: Operator) -> Result<()> {
/// let mut opts = CopyOptions::default();
/// opts.if_not_exists = true;
- /// op.copy_options("path/to/file", "path/to/file2", opts).await?;
+ /// op.copy_options("path/to/file", "path/to/file2", opts)
+ /// .await?;
/// # Ok(())
/// # }
/// ```
@@ -1577,9 +1578,9 @@ impl Operator {
///
/// ```
/// # use anyhow::Result;
- /// use opendal::options;
/// use opendal::EntryMode;
/// use opendal::Operator;
+ /// use opendal::options;
/// # async fn test(op: Operator) -> Result<()> {
/// let mut entries = op
/// .list_options("path/to/prefix", options::ListOptions {
@@ -1705,9 +1706,9 @@ impl Operator {
/// ```
/// # use anyhow::Result;
/// use futures::TryStreamExt;
- /// use opendal::options;
/// use opendal::EntryMode;
/// use opendal::Operator;
+ /// use opendal::options;
/// # async fn test(op: Operator) -> Result<()> {
/// let mut lister = op
/// .lister_options("path/to/dir/", options::ListOptions {
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 1a1276bb1..66a0d0f33 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -24,9 +24,10 @@ use std::future::IntoFuture;
use std::ops::RangeBounds;
use std::time::Duration;
+use futures::Future;
+
use crate::raw::*;
use crate::*;
-use futures::Future;
/// OperatorFuture is the future generated by [`Operator`].
///
diff --git a/core/src/types/operator/registry.rs b/core/src/types/operator/registry.rs
index 4aedc77d7..fb788c47f 100644
--- a/core/src/types/operator/registry.rs
+++ b/core/src/types/operator/registry.rs
@@ -16,11 +16,17 @@
// under the License.
use std::collections::HashMap;
-use std::sync::{LazyLock, Mutex};
+use std::sync::LazyLock;
+use std::sync::Mutex;
-use crate::types::builder::{Builder, Configurator};
-use crate::types::{IntoOperatorUri, OperatorUri};
-use crate::{Error, ErrorKind, Operator, Result};
+use crate::Error;
+use crate::ErrorKind;
+use crate::Operator;
+use crate::Result;
+use crate::types::IntoOperatorUri;
+use crate::types::OperatorUri;
+use crate::types::builder::Builder;
+use crate::types::builder::Configurator;
/// Factory signature used to construct [`Operator`] from a URI and extra options.
pub type OperatorFactory = fn(&OperatorUri) -> Result<Operator>;
diff --git a/core/src/types/operator/uri.rs b/core/src/types/operator/uri.rs
index 4a4e3bed3..0eca5b06c 100644
--- a/core/src/types/operator/uri.rs
+++ b/core/src/types/operator/uri.rs
@@ -21,7 +21,9 @@ use http::Uri;
use percent_encoding::percent_decode_str;
use url::Url;
-use crate::{Error, ErrorKind, Result};
+use crate::Error;
+use crate::ErrorKind;
+use crate::Result;
/// Parsed representation of an operator URI with normalized components.
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -268,10 +270,10 @@ mod tests {
#[test]
fn into_operator_uri_merges_extra_options() {
- let uri = (
- "s3://bucket/path?region=us-east-1",
- vec![("region", "override"), ("endpoint", "https://custom")],
- )
+ let uri = ("s3://bucket/path?region=us-east-1", vec![
+ ("region", "override"),
+ ("endpoint", "https://custom"),
+ ])
.into_operator_uri()
.unwrap();
diff --git a/core/src/types/options.rs b/core/src/types/options.rs
index d6d81a5a8..ee5064eec 100644
--- a/core/src/types/options.rs
+++ b/core/src/types/options.rs
@@ -17,9 +17,11 @@
//! Options module provides options definitions for operations.
-use crate::raw::{BytesRange, Timestamp};
use std::collections::HashMap;
+use crate::raw::BytesRange;
+use crate::raw::Timestamp;
+
/// Options for delete operations.
#[derive(Debug, Clone, Default, Eq, PartialEq)]
pub struct DeleteOptions {
diff --git a/rustfmt.toml b/rustfmt.toml
index 383062d6b..d39acfe67 100644
--- a/rustfmt.toml
+++ b/rustfmt.toml
@@ -23,4 +23,3 @@ reorder_imports = true
# imports_granularity = "Item"
# overflow_delimited_expr = true
# trailing_comma = "Vertical"
-# where_single_line = true