You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/03/17 12:41:16 UTC
(opendal) branch main updated: refactor(core/raw)!: Merge blocking and async operations (#5789)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new db38acf97 refactor(core/raw)!: Merge blocking and async operations (#5789)
db38acf97 is described below
commit db38acf9772ee64ed55acd82acadf661c3b6d0a3
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Mar 17 20:41:07 2025 +0800
refactor(core/raw)!: Merge blocking and async operations (#5789)
* refactor(core/raw)!: Merge blocking and async operations
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Fix tests
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/await_tree.rs | 8 +--
core/src/layers/capability_check.rs | 16 ++---
core/src/layers/correctness_check.rs | 35 ++++-----
core/src/layers/error_context.rs | 32 ++++-----
core/src/layers/fastrace.rs | 32 ++++-----
core/src/layers/logging.rs | 117 +++++++++++++++---------------
core/src/layers/observe/metrics.rs | 36 +++++-----
core/src/layers/timeout.rs | 8 +--
core/src/raw/operation.rs | 133 ++++++++++++++---------------------
core/src/services/ftp/backend.rs | 6 +-
core/src/services/ftp/delete.rs | 2 +-
11 files changed, 201 insertions(+), 224 deletions(-)
diff --git a/core/src/layers/await_tree.rs b/core/src/layers/await_tree.rs
index 238616aed..d04866c3d 100644
--- a/core/src/layers/await_tree.rs
+++ b/core/src/layers/await_tree.rs
@@ -86,7 +86,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.inner
.read(path, args)
- .instrument_await(format!("opendal::{}", Operation::Read))
+ .instrument_await(format!("opendal::{}", Operation::ReaderStart))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -94,7 +94,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.inner
.write(path, args)
- .instrument_await(format!("opendal::{}", Operation::Write))
+ .instrument_await(format!("opendal::{}", Operation::WriterStart))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -123,7 +123,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner
.delete()
- .instrument_await(format!("opendal::{}", Operation::Delete))
+ .instrument_await(format!("opendal::{}", Operation::DeleterStart))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -131,7 +131,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner
.list(path, args)
- .instrument_await(format!("opendal::{}", Operation::List))
+ .instrument_await(format!("opendal::{}", Operation::ListerStart))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
diff --git a/core/src/layers/capability_check.rs b/core/src/layers/capability_check.rs
index 30858766c..6d412c61d 100644
--- a/core/src/layers/capability_check.rs
+++ b/core/src/layers/capability_check.rs
@@ -102,21 +102,21 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.write_with_content_type && args.content_type().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Write,
+ Operation::WriterStart,
"content_type",
));
}
if !capability.write_with_cache_control && args.cache_control().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Write,
+ Operation::WriterStart,
"cache_control",
));
}
if !capability.write_with_content_disposition && args.content_disposition().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Write,
+ Operation::WriterStart,
"content_disposition",
));
}
@@ -133,7 +133,7 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.list_with_versions && args.versions() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::List,
+ Operation::ListerStart,
"version",
));
}
@@ -158,21 +158,21 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.write_with_content_type && args.content_type().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingWrite,
+ Operation::WriterStart,
"content_type",
));
}
if !capability.write_with_cache_control && args.cache_control().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingWrite,
+ Operation::WriterStart,
"cache_control",
));
}
if !capability.write_with_content_disposition && args.content_disposition().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingWrite,
+ Operation::WriterStart,
"content_disposition",
));
}
@@ -193,7 +193,7 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.list_with_versions && args.versions() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingList,
+ Operation::ListerStart,
"version",
));
}
diff --git a/core/src/layers/correctness_check.rs b/core/src/layers/correctness_check.rs
index dd73d3d7a..e15976358 100644
--- a/core/src/layers/correctness_check.rs
+++ b/core/src/layers/correctness_check.rs
@@ -98,35 +98,35 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if !capability.read_with_version && args.version().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Read,
+ Operation::ReaderStart,
"version",
));
}
if !capability.read_with_if_match && args.if_match().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Read,
+ Operation::ReaderStart,
"if_match",
));
}
if !capability.read_with_if_none_match && args.if_none_match().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Read,
+ Operation::ReaderStart,
"if_none_match",
));
}
if !capability.read_with_if_modified_since && args.if_modified_since().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Read,
+ Operation::ReaderStart,
"if_modified_since",
));
}
if !capability.read_with_if_unmodified_since && args.if_unmodified_since().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::Read,
+ Operation::ReaderStart,
"if_unmodified_since",
));
}
@@ -139,21 +139,24 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if args.append() && !capability.write_can_append {
return Err(new_unsupported_error(
&self.info,
- Operation::Write,
+ Operation::WriterStart,
"append",
));
}
if args.if_not_exists() && !capability.write_with_if_not_exists {
return Err(new_unsupported_error(
&self.info,
- Operation::Write,
+ Operation::WriterStart,
"if_not_exists",
));
}
if let Some(if_none_match) = args.if_none_match() {
if !capability.write_with_if_none_match {
- let mut err =
- new_unsupported_error(self.info.as_ref(), Operation::Write, "if_none_match");
+ let mut err = new_unsupported_error(
+ self.info.as_ref(),
+ Operation::WriterStart,
+ "if_none_match",
+ );
if if_none_match == "*" && capability.write_with_if_not_exists {
err = err.with_context("hint", "use if_not_exists instead");
}
@@ -222,7 +225,7 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if !capability.read_with_version && args.version().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingRead,
+ Operation::ReaderStart,
"version",
));
}
@@ -235,21 +238,21 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if args.append() && !capability.write_can_append {
return Err(new_unsupported_error(
&self.info,
- Operation::BlockingWrite,
+ Operation::WriterStart,
"append",
));
}
if args.if_not_exists() && !capability.write_with_if_not_exists {
return Err(new_unsupported_error(
&self.info,
- Operation::BlockingWrite,
+ Operation::WriterStart,
"if_not_exists",
));
}
if args.if_none_match().is_some() && !capability.write_with_if_none_match {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingWrite,
+ Operation::WriterStart,
"if_none_match",
));
}
@@ -262,7 +265,7 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if !capability.stat_with_version && args.version().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::BlockingStat,
+ Operation::Stat,
"version",
));
}
@@ -466,7 +469,7 @@ mod tests {
assert!(res.is_err());
assert_eq!(
res.unwrap_err().to_string(),
- "Unsupported (permanent) at write => The service memory does not support the operation write with the arguments if_none_match. Please verify if the relevant flags have been enabled, or submit an issue if you believe this is incorrect."
+ "Unsupported (permanent) at writer::start => The service memory does not support the operation writer::start with the arguments if_none_match. Please verify if the relevant flags have been enabled, or submit an issue if you believe this is incorrect."
);
// Now try a wildcard if-none-match
@@ -477,7 +480,7 @@ mod tests {
assert!(res.is_err());
assert_eq!(
res.unwrap_err().to_string(),
- "Unsupported (permanent) at write, context: { hint: use if_not_exists instead } => The service memory does not support the operation write with the arguments if_none_match. Please verify if the relevant flags have been enabled, or submit an issue if you believe this is incorrect."
+ "Unsupported (permanent) at writer::start, context: { hint: use if_not_exists instead } => The service memory does not support the operation writer::start with the arguments if_none_match. Please verify if the relevant flags have been enabled, or submit an issue if you believe this is incorrect."
);
let res = op
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 34ccb0c6e..a6a1908a4 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -99,7 +99,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::Read)
+ err.with_operation(Operation::ReaderStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
.with_context("range", range.to_string())
@@ -117,7 +117,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::Write)
+ err.with_operation(Operation::WriterStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -160,7 +160,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::Delete)
+ err.with_operation(Operation::DeleterStart)
.with_context("service", self.info.scheme())
})
}
@@ -176,7 +176,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::List)
+ err.with_operation(Operation::ListerStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -192,7 +192,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.inner.blocking_create_dir(path, args).map_err(|err| {
- err.with_operation(Operation::BlockingCreateDir)
+ err.with_operation(Operation::CreateDir)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -210,7 +210,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::BlockingRead)
+ err.with_operation(Operation::ReaderStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
.with_context("range", range.to_string())
@@ -227,7 +227,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::BlockingWrite)
+ err.with_operation(Operation::WriterStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -235,7 +235,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
self.inner.blocking_copy(from, to, args).map_err(|err| {
- err.with_operation(Operation::BlockingCopy)
+ err.with_operation(Operation::Copy)
.with_context("service", self.info.scheme())
.with_context("from", from)
.with_context("to", to)
@@ -244,7 +244,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
self.inner.blocking_rename(from, to, args).map_err(|err| {
- err.with_operation(Operation::BlockingRename)
+ err.with_operation(Operation::Rename)
.with_context("service", self.info.scheme())
.with_context("from", from)
.with_context("to", to)
@@ -253,7 +253,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.inner.blocking_stat(path, args).map_err(|err| {
- err.with_operation(Operation::BlockingStat)
+ err.with_operation(Operation::Stat)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -269,7 +269,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::BlockingDelete)
+ err.with_operation(Operation::DeleterStart)
.with_context("service", self.info.scheme())
})
}
@@ -284,7 +284,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::BlockingList)
+ err.with_operation(Operation::ListerStart)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -344,7 +344,7 @@ impl<T: oio::BlockingRead> oio::BlockingRead for ErrorContextWrapper<T> {
bs
})
.map_err(|err| {
- err.with_operation(Operation::BlockingReaderRead)
+ err.with_operation(Operation::ReaderRead)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("range", self.range.to_string())
@@ -399,7 +399,7 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
self.processed += size as u64;
})
.map_err(|err| {
- err.with_operation(Operation::BlockingWriterWrite)
+ err.with_operation(Operation::WriterWrite)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("size", size.to_string())
@@ -409,7 +409,7 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
fn close(&mut self) -> Result<Metadata> {
self.inner.close().map_err(|err| {
- err.with_operation(Operation::BlockingWriterClose)
+ err.with_operation(Operation::WriterClose)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("written", self.processed.to_string())
@@ -444,7 +444,7 @@ impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
bs
})
.map_err(|err| {
- err.with_operation(Operation::BlockingListerNext)
+ err.with_operation(Operation::ListerNext)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("listed", self.processed.to_string())
diff --git a/core/src/layers/fastrace.rs b/core/src/layers/fastrace.rs
index 8b0371b23..fabef47db 100644
--- a/core/src/layers/fastrace.rs
+++ b/core/src/layers/fastrace.rs
@@ -149,7 +149,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Read.into_static()),
+ Span::enter_with_local_parent(Operation::ReaderStart.into_static()),
r,
),
)
@@ -162,7 +162,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Write.into_static()),
+ Span::enter_with_local_parent(Operation::WriterStart.into_static()),
r,
),
)
@@ -190,7 +190,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Delete.into_static()),
+ Span::enter_with_local_parent(Operation::DeleterStart.into_static()),
r,
),
)
@@ -203,7 +203,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::List.into_static()),
+ Span::enter_with_local_parent(Operation::ListerStart.into_static()),
s,
),
)
@@ -226,7 +226,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::BlockingRead.into_static()),
+ Span::enter_with_local_parent(Operation::ReaderStart.into_static()),
r,
),
)
@@ -239,7 +239,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::BlockingWrite.into_static()),
+ Span::enter_with_local_parent(Operation::WriterStart.into_static()),
r,
),
)
@@ -267,7 +267,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::BlockingDelete.into_static()),
+ Span::enter_with_local_parent(Operation::DeleterStart.into_static()),
r,
),
)
@@ -280,7 +280,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::BlockingList.into_static()),
+ Span::enter_with_local_parent(Operation::ListerStart.into_static()),
it,
),
)
@@ -309,7 +309,7 @@ impl<R: oio::Read> oio::Read for FastraceWrapper<R> {
impl<R: oio::BlockingRead> oio::BlockingRead for FastraceWrapper<R> {
fn read(&mut self) -> Result<Buffer> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::BlockingReaderRead.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::ReaderRead.into_static());
self.inner.read()
}
}
@@ -337,15 +337,13 @@ impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
impl<R: oio::BlockingWrite> oio::BlockingWrite for FastraceWrapper<R> {
fn write(&mut self, bs: Buffer) -> Result<()> {
let _g = self.span.set_local_parent();
- let _span =
- LocalSpan::enter_with_local_parent(Operation::BlockingWriterWrite.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::WriterWrite.into_static());
self.inner.write(bs)
}
fn close(&mut self) -> Result<Metadata> {
let _g = self.span.set_local_parent();
- let _span =
- LocalSpan::enter_with_local_parent(Operation::BlockingWriterClose.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::WriterClose.into_static());
self.inner.close()
}
}
@@ -360,7 +358,7 @@ impl<R: oio::List> oio::List for FastraceWrapper<R> {
impl<R: oio::BlockingList> oio::BlockingList for FastraceWrapper<R> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::BlockingListerNext.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::ListerNext.into_static());
self.inner.next()
}
}
@@ -381,15 +379,13 @@ impl<R: oio::Delete> oio::Delete for FastraceWrapper<R> {
impl<R: oio::BlockingDelete> oio::BlockingDelete for FastraceWrapper<R> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
let _g = self.span.set_local_parent();
- let _span =
- LocalSpan::enter_with_local_parent(Operation::BlockingDeleterDelete.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::DeleterDelete.into_static());
self.inner.delete(path, args)
}
fn flush(&mut self) -> Result<usize> {
let _g = self.span.set_local_parent();
- let _span =
- LocalSpan::enter_with_local_parent(Operation::BlockingDeleterFlush.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::DeleterFlush.into_static());
self.inner.flush()
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 054131f84..953624166 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -313,7 +313,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.logger.log(
&self.info,
- Operation::Read,
+ Operation::ReaderStart,
&[("path", path)],
"started",
None,
@@ -325,7 +325,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, r)| {
self.logger.log(
&self.info,
- Operation::Read,
+ Operation::ReaderStart,
&[("path", path)],
"created reader",
None,
@@ -338,7 +338,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::Read,
+ Operation::ReaderStart,
&[("path", path)],
"failed",
Some(&err),
@@ -350,7 +350,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.logger.log(
&self.info,
- Operation::Write,
+ Operation::WriterStart,
&[("path", path)],
"started",
None,
@@ -362,7 +362,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, w)| {
self.logger.log(
&self.info,
- Operation::Write,
+ Operation::WriterStart,
&[("path", path)],
"created writer",
None,
@@ -373,7 +373,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::Write,
+ Operation::WriterStart,
&[("path", path)],
"failed",
Some(&err),
@@ -486,20 +486,25 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.logger
- .log(&self.info, Operation::Delete, &[], "started", None);
+ .log(&self.info, Operation::DeleterStart, &[], "started", None);
self.inner
.delete()
.await
.map(|(rp, d)| {
self.logger
- .log(&self.info, Operation::Delete, &[], "finished", None);
+ .log(&self.info, Operation::DeleterStart, &[], "finished", None);
let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
(rp, d)
})
.map_err(|err| {
- self.logger
- .log(&self.info, Operation::Delete, &[], "failed", Some(&err));
+ self.logger.log(
+ &self.info,
+ Operation::DeleterStart,
+ &[],
+ "failed",
+ Some(&err),
+ );
err
})
}
@@ -507,7 +512,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.logger.log(
&self.info,
- Operation::List,
+ Operation::ListerStart,
&[("path", path)],
"started",
None,
@@ -519,7 +524,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, v)| {
self.logger.log(
&self.info,
- Operation::List,
+ Operation::ListerStart,
&[("path", path)],
"created lister",
None,
@@ -530,7 +535,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::List,
+ Operation::ListerStart,
&[("path", path)],
"failed",
Some(&err),
@@ -576,7 +581,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.logger.log(
&self.info,
- Operation::BlockingCreateDir,
+ Operation::CreateDir,
&[("path", path)],
"started",
None,
@@ -587,7 +592,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|v| {
self.logger.log(
&self.info,
- Operation::BlockingCreateDir,
+ Operation::CreateDir,
&[("path", path)],
"finished",
None,
@@ -597,7 +602,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingCreateDir,
+ Operation::CreateDir,
&[("path", path)],
"failed",
Some(&err),
@@ -609,7 +614,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
self.logger.log(
&self.info,
- Operation::BlockingRead,
+ Operation::ReaderStart,
&[("path", path)],
"started",
None,
@@ -620,7 +625,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, r)| {
self.logger.log(
&self.info,
- Operation::BlockingRead,
+ Operation::ReaderStart,
&[("path", path)],
"created reader",
None,
@@ -631,7 +636,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingRead,
+ Operation::ReaderStart,
&[("path", path)],
"failed",
Some(&err),
@@ -643,7 +648,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
self.logger.log(
&self.info,
- Operation::BlockingWrite,
+ Operation::WriterStart,
&[("path", path)],
"started",
None,
@@ -654,7 +659,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, w)| {
self.logger.log(
&self.info,
- Operation::BlockingWrite,
+ Operation::WriterStart,
&[("path", path)],
"created writer",
None,
@@ -665,7 +670,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingWrite,
+ Operation::WriterStart,
&[("path", path)],
"failed",
Some(&err),
@@ -677,7 +682,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
self.logger.log(
&self.info,
- Operation::BlockingCopy,
+ Operation::Copy,
&[("from", from), ("to", to)],
"started",
None,
@@ -688,7 +693,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|v| {
self.logger.log(
&self.info,
- Operation::BlockingCopy,
+ Operation::Copy,
&[("from", from), ("to", to)],
"finished",
None,
@@ -698,7 +703,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingCopy,
+ Operation::Copy,
&[("from", from), ("to", to)],
"",
Some(&err),
@@ -710,7 +715,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
self.logger.log(
&self.info,
- Operation::BlockingRename,
+ Operation::Rename,
&[("from", from), ("to", to)],
"started",
None,
@@ -721,7 +726,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|v| {
self.logger.log(
&self.info,
- Operation::BlockingRename,
+ Operation::Rename,
&[("from", from), ("to", to)],
"finished",
None,
@@ -731,7 +736,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingRename,
+ Operation::Rename,
&[("from", from), ("to", to)],
"failed",
Some(&err),
@@ -743,7 +748,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.logger.log(
&self.info,
- Operation::BlockingStat,
+ Operation::Stat,
&[("path", path)],
"started",
None,
@@ -754,7 +759,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|v| {
self.logger.log(
&self.info,
- Operation::BlockingStat,
+ Operation::Stat,
&[("path", path)],
"finished",
None,
@@ -764,7 +769,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingStat,
+ Operation::Stat,
&[("path", path)],
"failed",
Some(&err),
@@ -775,20 +780,20 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
self.logger
- .log(&self.info, Operation::BlockingDelete, &[], "started", None);
+ .log(&self.info, Operation::DeleterStart, &[], "started", None);
self.inner
.blocking_delete()
.map(|(rp, d)| {
self.logger
- .log(&self.info, Operation::BlockingDelete, &[], "finished", None);
+ .log(&self.info, Operation::DeleterStart, &[], "finished", None);
let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
(rp, d)
})
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingDelete,
+ Operation::DeleterStart,
&[],
"failed",
Some(&err),
@@ -800,7 +805,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.logger.log(
&self.info,
- Operation::BlockingList,
+ Operation::ListerStart,
&[("path", path)],
"started",
None,
@@ -811,7 +816,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, v)| {
self.logger.log(
&self.info,
- Operation::BlockingList,
+ Operation::ListerStart,
&[("path", path)],
"created lister",
None,
@@ -822,7 +827,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map_err(|err| {
self.logger.log(
&self.info,
- Operation::BlockingList,
+ Operation::ListerStart,
&[("path", path)],
"",
Some(&err),
@@ -903,7 +908,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
fn read(&mut self) -> Result<Buffer> {
self.logger.log(
&self.info,
- Operation::BlockingReaderRead,
+ Operation::ReaderRead,
&[("path", &self.path), ("read", &self.read.to_string())],
"started",
None,
@@ -914,7 +919,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
self.read += bs.len() as u64;
self.logger.log(
&self.info,
- Operation::BlockingReaderRead,
+ Operation::ReaderRead,
&[
("path", &self.path),
("read", &self.read.to_string()),
@@ -932,7 +937,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingReaderRead,
+ Operation::ReaderRead,
&[("path", &self.path), ("read", &self.read.to_string())],
"failed",
Some(&err),
@@ -1087,7 +1092,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
self.logger.log(
&self.info,
- Operation::BlockingWriterWrite,
+ Operation::WriterWrite,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1101,7 +1106,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Ok(_) => {
self.logger.log(
&self.info,
- Operation::BlockingWriterWrite,
+ Operation::WriterWrite,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1115,7 +1120,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingWriterWrite,
+ Operation::WriterWrite,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1132,7 +1137,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
fn close(&mut self) -> Result<Metadata> {
self.logger.log(
&self.info,
- Operation::BlockingWriterClose,
+ Operation::WriterClose,
&[("path", &self.path), ("written", &self.written.to_string())],
"started",
None,
@@ -1142,7 +1147,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Ok(meta) => {
self.logger.log(
&self.info,
- Operation::BlockingWriterWrite,
+ Operation::WriterWrite,
&[("path", &self.path), ("written", &self.written.to_string())],
"succeeded",
None,
@@ -1152,7 +1157,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingWriterClose,
+ Operation::WriterClose,
&[("path", &self.path), ("written", &self.written.to_string())],
"failed",
Some(&err),
@@ -1240,7 +1245,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
fn next(&mut self) -> Result<Option<oio::Entry>> {
self.logger.log(
&self.info,
- Operation::BlockingListerNext,
+ Operation::ListerNext,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"started",
None,
@@ -1252,7 +1257,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
self.listed += 1;
self.logger.log(
&self.info,
- Operation::BlockingListerNext,
+ Operation::ListerNext,
&[
("path", &self.path),
("listed", &self.listed.to_string()),
@@ -1265,7 +1270,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
Ok(None) => {
self.logger.log(
&self.info,
- Operation::BlockingListerNext,
+ Operation::ListerNext,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"finished",
None,
@@ -1274,7 +1279,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingListerNext,
+ Operation::ListerNext,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"failed",
Some(err),
@@ -1416,7 +1421,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.logger.log(
&self.info,
- Operation::BlockingDeleterDelete,
+ Operation::DeleterDelete,
&[("path", path), ("version", &version)],
"started",
None,
@@ -1429,7 +1434,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.queued += 1;
self.logger.log(
&self.info,
- Operation::BlockingDeleterDelete,
+ Operation::DeleterDelete,
&[
("path", path),
("version", &version),
@@ -1443,7 +1448,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingDeleterDelete,
+ Operation::DeleterDelete,
&[
("path", path),
("version", &version),
@@ -1462,7 +1467,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
fn flush(&mut self) -> Result<usize> {
self.logger.log(
&self.info,
- Operation::BlockingDeleterFlush,
+ Operation::DeleterFlush,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1479,7 +1484,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.deleted += flushed;
self.logger.log(
&self.info,
- Operation::BlockingDeleterFlush,
+ Operation::DeleterFlush,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1491,7 +1496,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
Err(err) => {
self.logger.log(
&self.info,
- Operation::BlockingDeleterFlush,
+ Operation::DeleterFlush,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
diff --git a/core/src/layers/observe/metrics.rs b/core/src/layers/observe/metrics.rs
index 7855427e3..004b8d474 100644
--- a/core/src/layers/observe/metrics.rs
+++ b/core/src/layers/observe/metrics.rs
@@ -218,7 +218,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let op = Operation::Read;
+ let op = Operation::ReaderStart;
let start = Instant::now();
let (rp, reader) = self
@@ -262,7 +262,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let op = Operation::Write;
+ let op = Operation::WriterStart;
let start = Instant::now();
let (rp, writer) = self
@@ -399,7 +399,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- let op = Operation::Delete;
+ let op = Operation::DeleterStart;
let start = Instant::now();
let (rp, writer) = self
@@ -443,7 +443,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = Operation::List;
+ let op = Operation::ListerStart;
let start = Instant::now();
let (rp, lister) = self
@@ -518,7 +518,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let op = Operation::BlockingCreateDir;
+ let op = Operation::CreateDir;
let start = Instant::now();
self.inner()
@@ -548,7 +548,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let op = Operation::BlockingRead;
+ let op = Operation::ReaderStart;
let start = Instant::now();
let (rp, reader) = self
@@ -591,7 +591,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let op = Operation::BlockingWrite;
+ let op = Operation::WriterStart;
let start = Instant::now();
let (rp, writer) = self
@@ -634,7 +634,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let op = Operation::BlockingCopy;
+ let op = Operation::Copy;
let start = Instant::now();
self.inner()
@@ -664,7 +664,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let op = Operation::BlockingRename;
+ let op = Operation::Rename;
let start = Instant::now();
self.inner()
@@ -694,7 +694,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let op = Operation::BlockingStat;
+ let op = Operation::Stat;
let start = Instant::now();
self.inner()
@@ -724,7 +724,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- let op = Operation::BlockingDelete;
+ let op = Operation::DeleterStart;
let start = Instant::now();
let (rp, writer) = self
@@ -767,7 +767,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let op = Operation::BlockingList;
+ let op = Operation::ListerStart;
let start = Instant::now();
let (rp, lister) = self
@@ -884,7 +884,7 @@ impl<R: oio::Read, I: MetricsIntercept> oio::Read for MetricsWrapper<R, I> {
impl<R: oio::BlockingRead, I: MetricsIntercept> oio::BlockingRead for MetricsWrapper<R, I> {
fn read(&mut self) -> Result<Buffer> {
- let op = Operation::BlockingReaderRead;
+ let op = Operation::ReaderRead;
let start = Instant::now();
@@ -1029,7 +1029,7 @@ impl<R: oio::Write, I: MetricsIntercept> oio::Write for MetricsWrapper<R, I> {
impl<R: oio::BlockingWrite, I: MetricsIntercept> oio::BlockingWrite for MetricsWrapper<R, I> {
fn write(&mut self, bs: Buffer) -> Result<()> {
- let op = Operation::BlockingWriterWrite;
+ let op = Operation::WriterWrite;
let start = Instant::now();
let size = bs.len();
@@ -1070,7 +1070,7 @@ impl<R: oio::BlockingWrite, I: MetricsIntercept> oio::BlockingWrite for MetricsW
}
fn close(&mut self) -> Result<Metadata> {
- let op = Operation::BlockingWriterClose;
+ let op = Operation::WriterClose;
let start = Instant::now();
@@ -1134,7 +1134,7 @@ impl<R: oio::List, I: MetricsIntercept> oio::List for MetricsWrapper<R, I> {
impl<R: oio::BlockingList, I: MetricsIntercept> oio::BlockingList for MetricsWrapper<R, I> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
- let op = Operation::BlockingListerNext;
+ let op = Operation::ListerNext;
let start = Instant::now();
@@ -1228,7 +1228,7 @@ impl<R: oio::Delete, I: MetricsIntercept> oio::Delete for MetricsWrapper<R, I> {
impl<R: oio::BlockingDelete, I: MetricsIntercept> oio::BlockingDelete for MetricsWrapper<R, I> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let op = Operation::BlockingDeleterDelete;
+ let op = Operation::DeleterDelete;
let start = Instant::now();
@@ -1258,7 +1258,7 @@ impl<R: oio::BlockingDelete, I: MetricsIntercept> oio::BlockingDelete for Metric
}
fn flush(&mut self) -> Result<usize> {
- let op = Operation::BlockingDeleterFlush;
+ let op = Operation::DeleterFlush;
let start = Instant::now();
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index b1177d72e..0a1dfaac8 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -240,7 +240,7 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
)));
}
- self.io_timeout(Operation::Read, self.inner.read(path, args))
+ self.io_timeout(Operation::ReaderStart, self.inner.read(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
@@ -253,7 +253,7 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
)));
}
- self.io_timeout(Operation::Write, self.inner.write(path, args))
+ self.io_timeout(Operation::WriterStart, self.inner.write(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
@@ -274,13 +274,13 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- self.timeout(Operation::Delete, self.inner.delete())
+ self.timeout(Operation::DeleterStart, self.inner.delete())
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- self.io_timeout(Operation::List, self.inner.list(path, args))
+ self.io_timeout(Operation::ListerStart, self.inner.list(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
diff --git a/core/src/raw/operation.rs b/core/src/raw/operation.rs
index dc03db26d..633b823e4 100644
--- a/core/src/raw/operation.rs
+++ b/core/src/raw/operation.rs
@@ -18,73 +18,64 @@
use std::fmt::Display;
use std::fmt::Formatter;
-/// Operation is the name for APIs in `Accessor`.
+/// Operation is the name of the operation that is being performed.
+///
+/// Most operations can be mapped to the methods of the `Access` trait,
+/// but we modify the names to make them more readable and clear for users.
+///
+/// The same operation might have different meanings and costs in different
+/// storage services.
#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, Default)]
#[non_exhaustive]
pub enum Operation {
- /// Operation for [`crate::raw::Access::info`]
+ /// Operation to retrieve information about the specified storage services.
#[default]
Info,
- /// Operation for [`crate::raw::Access::create_dir`]
+ /// Operation to create a directory.
CreateDir,
- /// Operation for [`crate::raw::Access::read`]
- Read,
- /// Operation for [`crate::raw::oio::Read::read`]
+ /// Operation to start reading from a file.
+ ///
+ /// This operation only start to read from the file, it does not read any data.
+ ///
+ /// For example:
+ ///
+ /// - On fs alike storage, this operation is used to open a file.
+ /// - On s3 alike storage, this operation is used to send a request but not consume it's body.
+ ReaderStart,
+ /// Operation to perform a single read I/O operation on a file.
+ ///
+ /// Reader might have to call this operation multiple times to read the whole file.
ReaderRead,
- /// Operation for [`crate::raw::Access::write`]
- Write,
- /// Operation for [`crate::raw::oio::Write::write`]
+ /// Operation to start writing to a file.
+ ///
+ /// This operation only start to write into the file, it does not write any data.
+ WriterStart,
+ /// Operation to perform a single write I/O operation on a file.
+ ///
+ /// Writer might have to call this operation multiple times to write the whole file.
WriterWrite,
- /// Operation for [`crate::raw::oio::Write::close`]
+ /// Operation to close the writer and finish writing.
WriterClose,
- /// Operation for [`crate::raw::oio::Write::abort`]
+ /// Operation to abort the writer and discard the data.
WriterAbort,
- /// Operation for [`crate::raw::Access::copy`]
+ /// Operation to copy a file.
Copy,
- /// Operation for [`crate::raw::Access::rename`]
+ /// Operation to rename a file.
Rename,
- /// Operation for [`crate::raw::Access::stat`]
+ /// Operation to stat a file or a directory.
Stat,
- /// Operation for [`crate::raw::Access::delete`]
- Delete,
- /// Operation for [`crate::raw::oio::Delete::delete`]
+ /// Operation to start deleting files.
+ DeleterStart,
+ /// Operation to push a delete operation into the queue.
DeleterDelete,
- /// Operation for [`crate::raw::oio::Delete::flush`]
+ /// Operation to flush the deleter queue and perform real delete operations.
DeleterFlush,
- /// Operation for [`crate::raw::Access::list`]
- List,
- /// Operation for [`crate::raw::oio::List::next`]
+ /// Operation to start listing files.
+ ListerStart,
+ /// Operation to get the next file from the list.
ListerNext,
- /// Operation for [`crate::raw::Access::presign`]
+ /// Operation to generate a presigned URL.
Presign,
- /// Operation for [`crate::raw::Access::blocking_create_dir`]
- BlockingCreateDir,
- /// Operation for [`crate::raw::Access::blocking_read`]
- BlockingRead,
- /// Operation for [`crate::raw::oio::BlockingRead::read`]
- BlockingReaderRead,
- /// Operation for [`crate::raw::Access::blocking_write`]
- BlockingWrite,
- /// Operation for [`crate::raw::oio::BlockingWrite::write`]
- BlockingWriterWrite,
- /// Operation for [`crate::raw::oio::BlockingWrite::close`]
- BlockingWriterClose,
- /// Operation for [`crate::raw::Access::blocking_copy`]
- BlockingCopy,
- /// Operation for [`crate::raw::Access::blocking_rename`]
- BlockingRename,
- /// Operation for [`crate::raw::Access::blocking_stat`]
- BlockingStat,
- /// Operation for [`crate::raw::Access::blocking_delete`]
- BlockingDelete,
- /// Operation for [`crate::raw::oio::BlockingDelete::delete`]
- BlockingDeleterDelete,
- /// Operation for [`crate::raw::oio::BlockingDelete::flush`]
- BlockingDeleterFlush,
- /// Operation for [`crate::raw::Access::blocking_list`]
- BlockingList,
- /// Operation for [`crate::raw::oio::BlockingList::next`]
- BlockingListerNext,
}
impl Operation {
@@ -105,10 +96,6 @@ impl Operation {
| Operation::WriterWrite
| Operation::ListerNext
| Operation::DeleterDelete
- | Operation::BlockingReaderRead
- | Operation::BlockingWriterWrite
- | Operation::BlockingListerNext
- | Operation::BlockingDeleterDelete
)
}
}
@@ -122,37 +109,23 @@ impl Display for Operation {
impl From<Operation> for &'static str {
fn from(v: Operation) -> &'static str {
match v {
- Operation::Info => "metadata",
+ Operation::Info => "info",
Operation::CreateDir => "create_dir",
- Operation::Read => "read",
- Operation::ReaderRead => "Reader::read",
- Operation::Write => "write",
- Operation::WriterWrite => "Writer::write",
- Operation::WriterClose => "Writer::close",
- Operation::WriterAbort => "Writer::abort",
+ Operation::ReaderStart => "reader::start",
+ Operation::ReaderRead => "reader::read",
+ Operation::WriterStart => "writer::start",
+ Operation::WriterWrite => "writer::write",
+ Operation::WriterClose => "writer::close",
+ Operation::WriterAbort => "writer::abort",
Operation::Copy => "copy",
Operation::Rename => "rename",
Operation::Stat => "stat",
- Operation::Delete => "delete",
- Operation::List => "list",
- Operation::ListerNext => "List::next",
+ Operation::DeleterStart => "deleter::start",
+ Operation::DeleterDelete => "deleter::delete",
+ Operation::DeleterFlush => "deleter::flush",
+ Operation::ListerStart => "lister::start",
+ Operation::ListerNext => "lister::next",
Operation::Presign => "presign",
- Operation::BlockingCreateDir => "blocking_create_dir",
- Operation::BlockingRead => "blocking_read",
- Operation::BlockingReaderRead => "BlockingReader::read",
- Operation::BlockingWrite => "blocking_write",
- Operation::BlockingWriterWrite => "BlockingWriter::write",
- Operation::BlockingWriterClose => "BlockingWriter::close",
- Operation::BlockingCopy => "blocking_copy",
- Operation::BlockingRename => "blocking_rename",
- Operation::BlockingStat => "blocking_stat",
- Operation::BlockingDelete => "blocking_delete",
- Operation::BlockingList => "blocking_list",
- Operation::BlockingListerNext => "BlockingLister::next",
- Operation::DeleterDelete => "Deleter::delete",
- Operation::DeleterFlush => "Deleter::flush",
- Operation::BlockingDeleterDelete => "BlockingDeleter::delete",
- Operation::BlockingDeleterFlush => "BlockingDeleter::flush",
}
}
}
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 1a4f83a1c..5294824d4 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -344,7 +344,7 @@ impl Access for FtpBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let ftp_stream = self.ftp_connect(Operation::Read).await?;
+ let ftp_stream = self.ftp_connect(Operation::ReaderStart).await?;
let reader = FtpReader::new(ftp_stream, path.to_string(), args).await?;
Ok((RpRead::new(), reader))
@@ -356,7 +356,7 @@ impl Access for FtpBackend {
let paths: Vec<&str> = parent.split('/').collect();
// TODO: we can optimize this by checking dir existence first.
- let mut ftp_stream = self.ftp_connect(Operation::Write).await?;
+ let mut ftp_stream = self.ftp_connect(Operation::WriterStart).await?;
let mut curr_path = String::new();
for path in paths {
@@ -398,7 +398,7 @@ impl Access for FtpBackend {
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let mut ftp_stream = self.ftp_connect(Operation::List).await?;
+ let mut ftp_stream = self.ftp_connect(Operation::ListerStart).await?;
let pathname = if path == "/" { None } else { Some(path) };
let files = ftp_stream.list(pathname).await.map_err(parse_error)?;
diff --git a/core/src/services/ftp/delete.rs b/core/src/services/ftp/delete.rs
index 12ebed770..bc9bd2942 100644
--- a/core/src/services/ftp/delete.rs
+++ b/core/src/services/ftp/delete.rs
@@ -36,7 +36,7 @@ impl FtpDeleter {
impl oio::OneShotDelete for FtpDeleter {
async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let mut ftp_stream = self.core.ftp_connect(Operation::Delete).await?;
+ let mut ftp_stream = self.core.ftp_connect(Operation::DeleterStart).await?;
let result = if path.ends_with('/') {
ftp_stream.rmdir(&path).await