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 2026/05/13 15:05:26 UTC
(opendal) 01/01: feat(core): add copier api
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/copier-api-phase1
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 68802fa5b996b6757d531d2e818388c68430350b
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed May 13 23:05:05 2026 +0800
feat(core): add copier api
---
core/core/src/docs/internals/accessor.rs | 1 +
core/core/src/layers/complete.rs | 16 ++-
core/core/src/layers/correctness_check.rs | 12 +-
core/core/src/layers/error_context.rs | 56 +++++++--
core/core/src/layers/http_client.rs | 11 +-
core/core/src/layers/simulate.rs | 11 ++
core/core/src/layers/type_eraser.rs | 14 +++
core/core/src/raw/accessor.rs | 34 ++++--
core/core/src/raw/layer.rs | 25 +++-
core/core/src/raw/oio/copy/api.rs | 114 ++++++++++++++++++
core/core/src/raw/oio/{ => copy}/mod.rs | 25 +---
core/core/src/raw/oio/mod.rs | 3 +
core/core/src/raw/ops.rs | 37 ++++++
core/core/src/services/memory/backend.rs | 1 +
core/core/src/types/copy.rs | 143 +++++++++++++++++++++++
core/core/src/types/mod.rs | 3 +
core/core/src/types/operator/operator.rs | 123 +++++++++++++------
core/core/src/types/operator/operator_futures.rs | 23 ++++
core/core/src/types/options.rs | 6 +
core/layers/async-backtrace/src/lib.rs | 11 +-
core/layers/await-tree/src/lib.rs | 11 +-
core/layers/capability-check/src/lib.rs | 21 ++++
core/layers/chaos/src/lib.rs | 11 ++
core/layers/concurrent-limit/src/lib.rs | 23 +++-
core/layers/dtrace/src/lib.rs | 15 +++
core/layers/fastrace/src/lib.rs | 11 +-
core/layers/foyer/src/lib.rs | 11 ++
core/layers/hotpath/src/lib.rs | 11 +-
core/layers/immutable-index/src/lib.rs | 12 ++
core/layers/logging/src/lib.rs | 11 +-
core/layers/mime-guess/src/lib.rs | 11 ++
core/layers/observe-metrics-common/src/lib.rs | 11 +-
core/layers/oteltrace/src/lib.rs | 14 ++-
core/layers/retry/src/lib.rs | 12 +-
core/layers/route/src/lib.rs | 11 +-
core/layers/tail-cut/src/lib.rs | 17 ++-
core/layers/throttle/src/lib.rs | 11 ++
core/layers/timeout/src/lib.rs | 17 ++-
core/layers/tracing/src/lib.rs | 11 +-
core/services/aliyun-drive/src/backend.rs | 13 ++-
core/services/alluxio/src/backend.rs | 1 +
core/services/azblob/src/backend.rs | 11 +-
core/services/azdls/src/backend.rs | 1 +
core/services/azfile/src/backend.rs | 1 +
core/services/b2/src/backend.rs | 11 +-
core/services/cacache/src/backend.rs | 1 +
core/services/cloudflare-kv/src/backend.rs | 1 +
core/services/compfs/src/backend.rs | 11 +-
core/services/cos/src/backend.rs | 11 +-
core/services/d1/src/backend.rs | 1 +
core/services/dashmap/src/backend.rs | 1 +
core/services/dbfs/src/backend.rs | 1 +
core/services/dropbox/src/backend.rs | 13 ++-
core/services/etcd/src/backend.rs | 1 +
core/services/foundationdb/src/backend.rs | 1 +
core/services/foyer/src/backend.rs | 1 +
core/services/fs/src/backend.rs | 11 +-
core/services/ftp/src/backend.rs | 1 +
core/services/gcs/src/backend.rs | 11 +-
core/services/gdrive/src/backend.rs | 13 ++-
core/services/ghac/src/backend.rs | 1 +
core/services/github/src/backend.rs | 1 +
core/services/goosefs/src/backend.rs | 1 +
core/services/gridfs/src/backend.rs | 1 +
core/services/hdfs-native/src/backend.rs | 1 +
core/services/hdfs/src/backend.rs | 1 +
core/services/hf/src/backend.rs | 1 +
core/services/http/src/backend.rs | 1 +
core/services/ipfs/src/backend.rs | 1 +
core/services/ipmfs/src/backend.rs | 1 +
core/services/koofr/src/backend.rs | 13 ++-
core/services/lakefs/src/backend.rs | 11 +-
core/services/memcached/src/backend.rs | 1 +
core/services/mini_moka/src/backend.rs | 1 +
core/services/moka/src/backend.rs | 1 +
core/services/mongodb/src/backend.rs | 1 +
core/services/monoiofs/src/backend.rs | 11 +-
core/services/mysql/src/backend.rs | 1 +
core/services/obs/src/backend.rs | 11 +-
core/services/onedrive/src/backend.rs | 11 +-
core/services/opfs/src/backend.rs | 1 +
core/services/oss/src/backend.rs | 11 +-
core/services/pcloud/src/backend.rs | 11 +-
core/services/persy/src/backend.rs | 1 +
core/services/postgresql/src/backend.rs | 1 +
core/services/redb/src/backend.rs | 1 +
core/services/redis/src/backend.rs | 1 +
core/services/rocksdb/src/backend.rs | 1 +
core/services/s3/src/backend.rs | 11 +-
core/services/seafile/src/backend.rs | 1 +
core/services/sftp/src/backend.rs | 11 +-
core/services/sled/src/backend.rs | 1 +
core/services/sqlite/src/backend.rs | 1 +
core/services/surrealdb/src/backend.rs | 1 +
core/services/swift/src/backend.rs | 11 +-
core/services/tikv/src/backend.rs | 1 +
core/services/tos/src/backend.rs | 1 +
core/services/upyun/src/backend.rs | 11 +-
core/services/vercel-artifacts/src/backend.rs | 1 +
core/services/vercel-blob/src/backend.rs | 11 +-
core/services/webdav/src/backend.rs | 11 +-
core/services/webhdfs/src/backend.rs | 1 +
core/services/yandex-disk/src/backend.rs | 11 +-
core/tests/behavior/async_copy.rs | 133 ++++++++++++++++++++-
104 files changed, 1187 insertions(+), 169 deletions(-)
diff --git a/core/core/src/docs/internals/accessor.rs b/core/core/src/docs/internals/accessor.rs
index 96d2c019a..c1c9bf351 100644
--- a/core/core/src/docs/internals/accessor.rs
+++ b/core/core/src/docs/internals/accessor.rs
@@ -268,6 +268,7 @@
//! type Writer = ();
//! type Lister = ();
//! type Deleter = ();
+//! type Copier = ();
//!
//! fn info(&self) -> Arc<AccessorInfo> {
//! let am = AccessorInfo::default();
diff --git a/core/core/src/layers/complete.rs b/core/core/src/layers/complete.rs
index cc2808393..c53669f38 100644
--- a/core/core/src/layers/complete.rs
+++ b/core/core/src/layers/complete.rs
@@ -22,8 +22,9 @@ use std::sync::Arc;
use crate::raw::oio;
use crate::raw::{
- Access, AccessorInfo, Layer, LayeredAccess, OpCreateDir, OpList, OpPresign, OpRead, OpStat,
- OpWrite, RpCreateDir, RpDelete, RpList, RpPresign, RpRead, RpStat, RpWrite,
+ Access, AccessorInfo, Layer, LayeredAccess, OpCopier, OpCopy, OpCreateDir, OpList, OpPresign,
+ OpRead, OpStat, OpWrite, RpCopy, RpCreateDir, RpDelete, RpList, RpPresign, RpRead, RpStat,
+ RpWrite,
};
use crate::*;
@@ -59,6 +60,7 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
type Writer = CompleteWriter<A::Writer>;
type Lister = CompleteLister<A>;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -86,6 +88,16 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
Ok((rp, w))
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.inner.stat(path, args).await
}
diff --git a/core/core/src/layers/correctness_check.rs b/core/core/src/layers/correctness_check.rs
index 2620111c6..af73f7608 100644
--- a/core/core/src/layers/correctness_check.rs
+++ b/core/core/src/layers/correctness_check.rs
@@ -81,6 +81,7 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = CheckWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -217,7 +218,13 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
})
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let capability = self.info.full_capability();
if args.if_not_exists() && !capability.copy_with_if_not_exists {
return Err(new_unsupported_error(
@@ -227,7 +234,7 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
));
}
- self.inner.copy(from, to, args).await
+ self.inner.copy(from, to, args, opts).await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -296,6 +303,7 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/core/src/layers/error_context.rs b/core/core/src/layers/error_context.rs
index 1bd7a4dcd..c3593f452 100644
--- a/core/core/src/layers/error_context.rs
+++ b/core/core/src/layers/error_context.rs
@@ -69,6 +69,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
type Writer = ErrorContextWrapper<A::Writer>;
type Lister = ErrorContextWrapper<A::Lister>;
type Deleter = ErrorContextWrapper<A::Deleter>;
+ type Copier = ErrorContextWrapper<A::Copier>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -119,13 +120,28 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
})
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner.copy(from, to, args).await.map_err(|err| {
- err.with_operation(Operation::Copy)
- .with_context("service", self.info.scheme())
- .with_context("from", from)
- .with_context("to", to)
- })
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner
+ .copy(from, to, args, opts)
+ .await
+ .map(|(rp, p)| {
+ (
+ rp,
+ ErrorContextWrapper::new(self.info.scheme(), to.to_string(), p),
+ )
+ })
+ .map_err(|err| {
+ err.with_operation(Operation::Copy)
+ .with_context("service", self.info.scheme())
+ .with_context("from", from)
+ .with_context("to", to)
+ })
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -302,3 +318,29 @@ impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
})
}
}
+
+impl<T: oio::Copy> oio::Copy for ErrorContextWrapper<T> {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ self.inner
+ .next()
+ .await
+ .inspect(|size| {
+ self.processed += size.unwrap_or_default() as u64;
+ })
+ .map_err(|err| {
+ err.with_operation(Operation::Copy)
+ .with_context("service", self.scheme)
+ .with_context("path", &self.path)
+ .with_context("copied", self.processed.to_string())
+ })
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.inner.abort().await.map_err(|err| {
+ err.with_operation(Operation::Copy)
+ .with_context("service", self.scheme)
+ .with_context("path", &self.path)
+ .with_context("copied", self.processed.to_string())
+ })
+ }
+}
diff --git a/core/core/src/layers/http_client.rs b/core/core/src/layers/http_client.rs
index 1b1eed4b0..8e4b02d59 100644
--- a/core/core/src/layers/http_client.rs
+++ b/core/core/src/layers/http_client.rs
@@ -122,6 +122,7 @@ impl<A: Access> LayeredAccess for HttpClientAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -139,8 +140,14 @@ impl<A: Access> LayeredAccess for HttpClientAccessor<A> {
self.inner.write(path, args).await
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner.copy(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
diff --git a/core/core/src/layers/simulate.rs b/core/core/src/layers/simulate.rs
index 1d64d6408..5d546b279 100644
--- a/core/core/src/layers/simulate.rs
+++ b/core/core/src/layers/simulate.rs
@@ -257,6 +257,7 @@ impl<A: Access> LayeredAccess for SimulateAccessor<A> {
type Writer = A::Writer;
type Lister = SimulateLister<A, A::Lister>;
type Deleter = SimulateDeleter<A, A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -278,6 +279,16 @@ impl<A: Access> LayeredAccess for SimulateAccessor<A> {
self.inner.write(path, args).await
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.simulate_stat(path, args).await
}
diff --git a/core/core/src/layers/type_eraser.rs b/core/core/src/layers/type_eraser.rs
index bb019d14f..c041b8ed2 100644
--- a/core/core/src/layers/type_eraser.rs
+++ b/core/core/src/layers/type_eraser.rs
@@ -56,6 +56,7 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -88,4 +89,17 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
.await
.map(|(rp, p)| (rp, Box::new(p) as oio::Lister))
}
+
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner
+ .copy(from, to, args, opts)
+ .await
+ .map(|(rp, p)| (rp, Box::new(p) as oio::Copier))
+ }
}
diff --git a/core/core/src/raw/accessor.rs b/core/core/src/raw/accessor.rs
index 5a9283a13..c033981dd 100644
--- a/core/core/src/raw/accessor.rs
+++ b/core/core/src/raw/accessor.rs
@@ -66,6 +66,8 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
type Lister: oio::List;
/// Deleter is the associated deleter returned in `delete` operation.
type Deleter: oio::Delete;
+ /// Copier is the associated copier returned in `copy` operation.
+ type Copier: oio::Copy;
/// Invoke the `info` operation to get metadata of accessor.
///
@@ -212,8 +214,9 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
from: &str,
to: &str,
args: OpCopy,
- ) -> impl Future<Output = Result<RpCopy>> + MaybeSend {
- let (_, _, _) = (from, to, args);
+ opts: OpCopier,
+ ) -> impl Future<Output = Result<(RpCopy, Self::Copier)>> + MaybeSend {
+ let (_, _, _, _) = (from, to, args, opts);
ready(Err(Error::new(
ErrorKind::Unsupported,
@@ -298,7 +301,8 @@ pub trait AccessDyn: Send + Sync + Debug + Unpin {
from: &'a str,
to: &'a str,
args: OpCopy,
- ) -> BoxedFuture<'a, Result<RpCopy>>;
+ opts: OpCopier,
+ ) -> BoxedFuture<'a, Result<(RpCopy, oio::Copier)>>;
/// Dyn version of [`Accessor::rename`]
fn rename_dyn<'a>(
&'a self,
@@ -321,6 +325,7 @@ where
Writer = oio::Writer,
Lister = oio::Lister,
Deleter = oio::Deleter,
+ Copier = oio::Copier,
>,
{
fn info_dyn(&self) -> Arc<AccessorInfo> {
@@ -372,8 +377,9 @@ where
from: &'a str,
to: &'a str,
args: OpCopy,
- ) -> BoxedFuture<'a, Result<RpCopy>> {
- Box::pin(self.copy(from, to, args))
+ opts: OpCopier,
+ ) -> BoxedFuture<'a, Result<(RpCopy, oio::Copier)>> {
+ Box::pin(self.copy(from, to, args, opts))
}
fn rename_dyn<'a>(
@@ -399,6 +405,7 @@ impl Access for dyn AccessDyn {
type Writer = oio::Writer;
type Deleter = oio::Deleter;
type Lister = oio::Lister;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
self.info_dyn()
@@ -428,8 +435,14 @@ impl Access for dyn AccessDyn {
self.list_dyn(path, args).await
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.copy_dyn(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.copy_dyn(from, to, args, opts).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -447,6 +460,7 @@ impl Access for () {
type Writer = ();
type Lister = ();
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
let ai = AccessorInfo::default();
@@ -468,6 +482,7 @@ impl<T: Access + ?Sized> Access for Arc<T> {
type Writer = T::Writer;
type Lister = T::Lister;
type Deleter = T::Deleter;
+ type Copier = T::Copier;
fn info(&self) -> Arc<AccessorInfo> {
self.as_ref().info()
@@ -518,8 +533,9 @@ impl<T: Access + ?Sized> Access for Arc<T> {
from: &str,
to: &str,
args: OpCopy,
- ) -> impl Future<Output = Result<RpCopy>> + MaybeSend {
- async move { self.as_ref().copy(from, to, args).await }
+ opts: OpCopier,
+ ) -> impl Future<Output = Result<(RpCopy, Self::Copier)>> + MaybeSend {
+ async move { self.as_ref().copy(from, to, args, opts).await }
}
fn rename(
diff --git a/core/core/src/raw/layer.rs b/core/core/src/raw/layer.rs
index d35e9223f..09c9e3203 100644
--- a/core/core/src/raw/layer.rs
+++ b/core/core/src/raw/layer.rs
@@ -19,6 +19,7 @@ use std::fmt::Debug;
use std::sync::Arc;
use futures::Future;
+use futures::future::ready;
use crate::raw::*;
use crate::*;
@@ -60,6 +61,7 @@ use crate::*;
/// type Writer = A::Writer;
/// type Lister = A::Lister;
/// type Deleter = A::Deleter;
+/// type Copier = A::Copier;
///
/// fn inner(&self) -> &Self::Inner {
/// &self.inner
@@ -113,6 +115,7 @@ pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
type Writer: oio::Write;
type Lister: oio::List;
type Deleter: oio::Delete;
+ type Copier: oio::Copy;
fn inner(&self) -> &Self::Inner;
@@ -145,8 +148,14 @@ pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
from: &str,
to: &str,
args: OpCopy,
- ) -> impl Future<Output = Result<RpCopy>> + MaybeSend {
- self.inner().copy(from, to, args)
+ opts: OpCopier,
+ ) -> impl Future<Output = Result<(RpCopy, Self::Copier)>> + MaybeSend {
+ let (_, _, _, _) = (from, to, args, opts);
+
+ ready(Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ )))
}
fn rename(
@@ -184,6 +193,7 @@ impl<L: LayeredAccess> Access for L {
type Writer = L::Writer;
type Lister = L::Lister;
type Deleter = L::Deleter;
+ type Copier = L::Copier;
fn info(&self) -> Arc<AccessorInfo> {
LayeredAccess::info(self)
@@ -201,8 +211,14 @@ impl<L: LayeredAccess> Access for L {
LayeredAccess::write(self, path, args).await
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- LayeredAccess::copy(self, from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ LayeredAccess::copy(self, from, to, args, opts).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -258,6 +274,7 @@ mod tests {
type Writer = ();
type Lister = ();
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/core/src/raw/oio/copy/api.rs b/core/core/src/raw/oio/copy/api.rs
new file mode 100644
index 000000000..08f938577
--- /dev/null
+++ b/core/core/src/raw/oio/copy/api.rs
@@ -0,0 +1,114 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::future::Future;
+use std::ops::DerefMut;
+
+use crate::raw::*;
+use crate::*;
+
+/// Copier is the type-erased [`Copy`].
+pub type Copier = Box<dyn CopyDyn>;
+
+/// Copy is the trait that OpenDAL returns for stateful copy operations.
+pub trait Copy: Unpin + Send + Sync {
+ /// Drive the copy operation forward.
+ ///
+ /// `Ok(Some(n))` means the copy operation made progress by `n` bytes.
+ /// `Ok(None)` means the copy operation has completed.
+ fn next(&mut self) -> impl Future<Output = Result<Option<usize>>> + MaybeSend;
+
+ /// Abort the pending copy operation.
+ fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend;
+}
+
+impl Copy for () {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ Ok(None)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// OneShotCopier drives a single asynchronous copy step.
+pub struct OneShotCopier {
+ fut: Option<BoxedStaticFuture<Result<Option<usize>>>>,
+}
+
+/// # Safety
+///
+/// OneShotCopier is only accessed by `&mut self`.
+unsafe impl Sync for OneShotCopier {}
+
+impl OneShotCopier {
+ /// Create a new one-shot copier.
+ pub fn new(fut: impl Future<Output = Result<Option<usize>>> + MaybeSend + 'static) -> Self {
+ Self {
+ fut: Some(Box::pin(fut)),
+ }
+ }
+
+ /// Create a one-shot copier that has already completed.
+ pub fn completed() -> Self {
+ Self { fut: None }
+ }
+}
+
+impl Copy for OneShotCopier {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ match self.fut.take() {
+ Some(fut) => fut.await,
+ None => Ok(None),
+ }
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.fut = None;
+ Ok(())
+ }
+}
+
+/// CopyDyn is the dyn version of [`Copy`].
+pub trait CopyDyn: Unpin + Send + Sync {
+ /// The dyn version of [`Copy::next`].
+ fn next_dyn(&mut self) -> BoxedFuture<'_, Result<Option<usize>>>;
+
+ /// The dyn version of [`Copy::abort`].
+ fn abort_dyn(&mut self) -> BoxedFuture<'_, Result<()>>;
+}
+
+impl<T: Copy + ?Sized> CopyDyn for T {
+ fn next_dyn(&mut self) -> BoxedFuture<'_, Result<Option<usize>>> {
+ Box::pin(self.next())
+ }
+
+ fn abort_dyn(&mut self) -> BoxedFuture<'_, Result<()>> {
+ Box::pin(self.abort())
+ }
+}
+
+impl<T: CopyDyn + ?Sized> Copy for Box<T> {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ self.deref_mut().next_dyn().await
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.deref_mut().abort_dyn().await
+ }
+}
diff --git a/core/core/src/raw/oio/mod.rs b/core/core/src/raw/oio/copy/mod.rs
similarity index 68%
copy from core/core/src/raw/oio/mod.rs
copy to core/core/src/raw/oio/copy/mod.rs
index c44340646..defb32e6b 100644
--- a/core/core/src/raw/oio/mod.rs
+++ b/core/core/src/raw/oio/copy/mod.rs
@@ -15,26 +15,5 @@
// specific language governing permissions and limitations
// under the License.
-//! `oio` provides OpenDAL's raw traits and types that opendal returns as
-//! output.
-//!
-//! Those types should only be used internally and we don't want users to
-//! depend on them.
-
-mod delete;
-pub use delete::*;
-
-mod read;
-pub use read::*;
-
-mod write;
-pub use write::*;
-
-mod list;
-pub use list::*;
-
-mod entry;
-pub use entry::Entry;
-
-mod buf;
-pub use buf::*;
+mod api;
+pub use api::*;
diff --git a/core/core/src/raw/oio/mod.rs b/core/core/src/raw/oio/mod.rs
index c44340646..ec5abf324 100644
--- a/core/core/src/raw/oio/mod.rs
+++ b/core/core/src/raw/oio/mod.rs
@@ -24,6 +24,9 @@
mod delete;
pub use delete::*;
+mod copy;
+pub use copy::*;
+
mod read;
pub use read::*;
diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs
index 41182df6f..51b814c66 100644
--- a/core/core/src/raw/ops.rs
+++ b/core/core/src/raw/ops.rs
@@ -901,6 +901,43 @@ impl OpCopy {
}
}
+/// Args for `copier` operation.
+#[derive(Debug, Clone, Default)]
+pub struct OpCopier {
+ concurrent: usize,
+}
+
+impl OpCopier {
+ /// Create a new `OpCopier`.
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Set the concurrent tasks for the copier.
+ pub fn with_concurrent(mut self, concurrent: usize) -> Self {
+ self.concurrent = concurrent.max(1);
+ self
+ }
+
+ /// Get the concurrent tasks for the copier.
+ pub fn concurrent(&self) -> usize {
+ self.concurrent.max(1)
+ }
+}
+
+impl From<options::CopyOptions> for (OpCopy, OpCopier) {
+ fn from(value: options::CopyOptions) -> Self {
+ (
+ OpCopy {
+ if_not_exists: value.if_not_exists,
+ },
+ OpCopier {
+ concurrent: value.concurrent.max(1),
+ },
+ )
+ }
+}
+
/// Args for `rename` operation.
#[derive(Debug, Clone, Default)]
pub struct OpRename {}
diff --git a/core/core/src/services/memory/backend.rs b/core/core/src/services/memory/backend.rs
index f8384cb7b..9eda17fc9 100644
--- a/core/core/src/services/memory/backend.rs
+++ b/core/core/src/services/memory/backend.rs
@@ -104,6 +104,7 @@ impl Access for MemoryBackend {
type Writer = MemoryWriter;
type Lister = oio::HierarchyLister<MemoryLister>;
type Deleter = oio::OneShotDeleter<MemoryDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/core/src/types/copy.rs b/core/core/src/types/copy.rs
new file mode 100644
index 000000000..759db4a85
--- /dev/null
+++ b/core/core/src/types/copy.rs
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+use std::task::ready;
+use std::{fmt, fmt::Formatter};
+
+use futures::Stream;
+
+use crate::raw::*;
+use crate::*;
+
+/// Copier is designed to drive long-running copy operations.
+///
+/// - Copier implements `Stream<Item = Result<usize>>`.
+/// - `Some(Ok(n))` means the copy operation made progress by `n` bytes.
+/// - `None` means the copy operation has completed.
+pub struct Copier {
+ copier: Option<oio::Copier>,
+
+ fut: Option<BoxedStaticFuture<(oio::Copier, Result<Option<usize>>)>>,
+ errored: bool,
+}
+
+impl fmt::Debug for Copier {
+ fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Copier")
+ .field("active", &self.copier.is_some())
+ .field("errored", &self.errored)
+ .finish()
+ }
+}
+
+/// # Safety
+///
+/// Copier will only be accessed by `&mut Self`.
+unsafe impl Sync for Copier {}
+
+impl Copier {
+ /// Create a new copier.
+ pub(crate) async fn create(
+ acc: Accessor,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<Self> {
+ let (_, copier) = acc.copy(from, to, args, opts).await?;
+
+ Ok(Self {
+ copier: Some(copier),
+
+ fut: None,
+ errored: false,
+ })
+ }
+
+ /// Drive the copy operation forward.
+ pub async fn next(&mut self) -> Result<Option<usize>> {
+ if self.errored {
+ return Ok(None);
+ }
+
+ let Some(copier) = self.copier.as_mut() else {
+ return Ok(None);
+ };
+
+ match copier.next_dyn().await {
+ Ok(Some(n)) => Ok(Some(n)),
+ Ok(None) => {
+ self.copier = None;
+ Ok(None)
+ }
+ Err(err) => {
+ self.errored = true;
+ Err(err)
+ }
+ }
+ }
+
+ /// Abort the pending copy operation.
+ pub async fn abort(&mut self) -> Result<()> {
+ let Some(copier) = self.copier.as_mut() else {
+ return Ok(());
+ };
+
+ copier.abort_dyn().await
+ }
+}
+
+impl Stream for Copier {
+ type Item = Result<usize>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ if self.errored {
+ return Poll::Ready(None);
+ }
+
+ if let Some(mut copier) = self.copier.take() {
+ let fut = async move {
+ let res = copier.next_dyn().await;
+ (copier, res)
+ };
+ self.fut = Some(Box::pin(fut));
+ }
+
+ if let Some(fut) = self.fut.as_mut() {
+ let (copier, res) = ready!(fut.as_mut().poll(cx));
+ self.copier = Some(copier);
+ self.fut = None;
+
+ return match res {
+ Ok(Some(n)) => Poll::Ready(Some(Ok(n))),
+ Ok(None) => {
+ self.copier = None;
+ Poll::Ready(None)
+ }
+ Err(err) => {
+ self.errored = true;
+ Poll::Ready(Some(Err(err)))
+ }
+ };
+ }
+
+ Poll::Ready(None)
+ }
+}
diff --git a/core/core/src/types/mod.rs b/core/core/src/types/mod.rs
index 8042804ae..41203ce65 100644
--- a/core/core/src/types/mod.rs
+++ b/core/core/src/types/mod.rs
@@ -36,6 +36,9 @@ pub use write::*;
mod list;
pub use list::Lister;
+mod copy;
+pub use copy::Copier;
+
mod delete;
pub use delete::*;
diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs
index b0153bda9..3bbe21d6b 100644
--- a/core/core/src/types/operator/operator.rs
+++ b/core/core/src/types/operator/operator.rs
@@ -990,41 +990,14 @@ impl Operator {
/// # }
/// ```
pub async fn copy(&self, from: &str, to: &str) -> Result<()> {
- let from = normalize_path(from);
-
- if !validate_path(&from, EntryMode::FILE) {
- return Err(
- Error::new(ErrorKind::IsADirectory, "from path is a directory")
- .with_operation("Operator::copy")
- .with_context("service", self.info().scheme())
- .with_context("from", from),
- );
- }
-
- let to = normalize_path(to);
-
- if !validate_path(&to, EntryMode::FILE) {
- return Err(
- Error::new(ErrorKind::IsADirectory, "to path is a directory")
- .with_operation("Operator::copy")
- .with_context("service", self.info().scheme())
- .with_context("to", to),
- );
- }
-
- if from == to {
- return Err(
- Error::new(ErrorKind::IsSameFile, "from and to paths are same")
- .with_operation("Operator::copy")
- .with_context("service", self.info().scheme())
- .with_context("from", from)
- .with_context("to", to),
- );
- }
-
- self.inner().copy(&from, &to, OpCopy::new()).await?;
+ self.copy_options(from, to, options::CopyOptions::default())
+ .await
+ }
- Ok(())
+ /// Create a copier from `from` to `to`.
+ pub async fn copier(&self, from: &str, to: &str) -> Result<Copier> {
+ self.copier_options(from, to, options::CopyOptions::default())
+ .await
}
/// Copy a file from `from` to `to` with additional options.
@@ -1066,6 +1039,23 @@ impl Operator {
)
}
+ /// Create a copier from `from` to `to` with additional options.
+ pub fn copier_with(
+ &self,
+ from: &str,
+ to: &str,
+ ) -> FutureCopier<impl Future<Output = Result<Copier>>> {
+ let from = normalize_path(from);
+ let to = normalize_path(to);
+
+ OperatorFuture::new(
+ self.inner().clone(),
+ from,
+ (options::CopyOptions::default(), to),
+ Self::copier_inner,
+ )
+ }
+
/// Copy a file from `from` to `to` with additional options.
///
/// # Notes
@@ -1107,6 +1097,20 @@ impl Operator {
Self::copy_inner(self.inner().clone(), from, (opts, to)).await
}
+ /// Create a copier from `from` to `to` with additional options.
+ pub async fn copier_options(
+ &self,
+ from: &str,
+ to: &str,
+ opts: impl Into<options::CopyOptions>,
+ ) -> Result<Copier> {
+ let from = normalize_path(from);
+ let to = normalize_path(to);
+ let opts = opts.into();
+
+ Self::copier_inner(self.inner().clone(), from, (opts, to)).await
+ }
+
async fn copy_inner(
acc: Accessor,
from: String,
@@ -1140,12 +1144,55 @@ impl Operator {
);
}
- let mut op = OpCopy::new();
- if opts.if_not_exists {
- op = op.with_if_not_exists(true);
+ let mut copier = Self::copier_inner(acc, from, (opts, to)).await?;
+
+ loop {
+ match copier.next().await {
+ Ok(Some(_)) => continue,
+ Ok(None) => return Ok(()),
+ Err(err) => {
+ let _ = copier.abort().await;
+ return Err(err);
+ }
+ }
+ }
+ }
+
+ async fn copier_inner(
+ acc: Accessor,
+ from: String,
+ (opts, to): (options::CopyOptions, String),
+ ) -> Result<Copier> {
+ if !validate_path(&from, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "from path is a directory")
+ .with_operation("Operator::copier")
+ .with_context("service", acc.info().scheme())
+ .with_context("from", from),
+ );
+ }
+
+ if !validate_path(&to, EntryMode::FILE) {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "to path is a directory")
+ .with_operation("Operator::copier")
+ .with_context("service", acc.info().scheme())
+ .with_context("to", to),
+ );
+ }
+
+ if from == to {
+ return Err(
+ Error::new(ErrorKind::IsSameFile, "from and to paths are same")
+ .with_operation("Operator::copier")
+ .with_context("service", acc.info().scheme())
+ .with_context("from", &from)
+ .with_context("to", &to),
+ );
}
- acc.copy(&from, &to, op).await.map(|_| ())
+ let (args, opts) = opts.into();
+ Copier::create(acc, &from, &to, args, opts).await
}
/// Rename a file from `from` to `to`.
diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs
index 7482bf75b..c635981f8 100644
--- a/core/core/src/types/operator/operator_futures.rs
+++ b/core/core/src/types/operator/operator_futures.rs
@@ -1421,3 +1421,26 @@ impl<F: Future<Output = Result<()>>> FutureCopy<F> {
self
}
}
+
+/// Future that generated by [`Operator::copier_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub type FutureCopier<F> = OperatorFuture<(options::CopyOptions, String), Copier, F>;
+
+impl<F: Future<Output = Result<Copier>>> FutureCopier<F> {
+ /// Sets the condition that copy operation will succeed only if target does not exist.
+ ///
+ /// Refer to [`options::CopyOptions::if_not_exists`] for more details.
+ pub fn if_not_exists(mut self, v: bool) -> Self {
+ self.args.0.if_not_exists = v;
+ self
+ }
+
+ /// Sets concurrent copy operations for this copier.
+ ///
+ /// Refer to [`options::CopyOptions::concurrent`] for more details.
+ pub fn concurrent(mut self, v: usize) -> Self {
+ self.args.0.concurrent = v.max(1);
+ self
+ }
+}
diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs
index 4ef109c61..3ed7302f3 100644
--- a/core/core/src/types/options.rs
+++ b/core/core/src/types/options.rs
@@ -532,4 +532,10 @@ pub struct CopyOptions {
/// This operation provides a way to ensure copy operations only create new resources
/// without overwriting existing ones, useful for implementing "copy if not exists" logic.
pub if_not_exists: bool,
+
+ /// Sets concurrent copy operations for this copier.
+ ///
+ /// This is a best-effort execution option. Services that cannot split copy
+ /// into concurrent server-side tasks can ignore it.
+ pub concurrent: usize,
}
diff --git a/core/layers/async-backtrace/src/lib.rs b/core/layers/async-backtrace/src/lib.rs
index f8222097a..afa572b4d 100644
--- a/core/layers/async-backtrace/src/lib.rs
+++ b/core/layers/async-backtrace/src/lib.rs
@@ -76,6 +76,7 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
type Writer = AsyncBacktraceWrapper<A::Writer>;
type Lister = AsyncBacktraceWrapper<A::Lister>;
type Deleter = AsyncBacktraceWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -98,8 +99,14 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
}
#[async_backtrace::framed]
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner.copy(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts.clone()).await
}
#[async_backtrace::framed]
diff --git a/core/layers/await-tree/src/lib.rs b/core/layers/await-tree/src/lib.rs
index 1b98de408..b2a89ecda 100644
--- a/core/layers/await-tree/src/lib.rs
+++ b/core/layers/await-tree/src/lib.rs
@@ -79,6 +79,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
type Writer = AwaitTreeWrapper<A::Writer>;
type Lister = AwaitTreeWrapper<A::Lister>;
type Deleter = AwaitTreeWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -100,9 +101,15 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.inner()
- .copy(from, to, args)
+ .copy(from, to, args, opts.clone())
.instrument_await(format!("opendal::{}", Operation::Copy))
.await
}
diff --git a/core/layers/capability-check/src/lib.rs b/core/layers/capability-check/src/lib.rs
index b0ba046f0..96f1c8cde 100644
--- a/core/layers/capability-check/src/lib.rs
+++ b/core/layers/capability-check/src/lib.rs
@@ -110,6 +110,7 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -146,6 +147,25 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
self.inner.write(path, args).await
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ let capability = self.info.full_capability();
+ if args.if_not_exists() && !capability.copy_with_if_not_exists {
+ return Err(new_unsupported_error(
+ self.info.as_ref(),
+ Operation::Copy,
+ "if_not_exists",
+ ));
+ }
+
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner.delete().await
}
@@ -178,6 +198,7 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/layers/chaos/src/lib.rs b/core/layers/chaos/src/lib.rs
index c382961c4..06ac4b964 100644
--- a/core/layers/chaos/src/lib.rs
+++ b/core/layers/chaos/src/lib.rs
@@ -107,6 +107,7 @@ impl<A: Access> LayeredAccess for ChaosAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -125,6 +126,16 @@ impl<A: Access> LayeredAccess for ChaosAccessor<A> {
self.inner.write(path, args).await
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.list(path, args).await
}
diff --git a/core/layers/concurrent-limit/src/lib.rs b/core/layers/concurrent-limit/src/lib.rs
index 305f68fb2..6ea9bdc66 100644
--- a/core/layers/concurrent-limit/src/lib.rs
+++ b/core/layers/concurrent-limit/src/lib.rs
@@ -249,6 +249,7 @@ where
type Writer = ConcurrentLimitWrapper<A::Writer, S::Permit>;
type Lister = ConcurrentLimitWrapper<A::Lister, S::Permit>;
type Deleter = ConcurrentLimitWrapper<A::Deleter, S::Permit>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -278,10 +279,16 @@ where
.map(|(rp, w)| (rp, ConcurrentLimitWrapper::new(w, permit)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let _permit = self.semaphore.acquire().await;
- self.inner.copy(from, to, args).await
+ self.inner.copy(from, to, args, opts.clone()).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -422,13 +429,20 @@ mod tests {
type Writer = ();
type Lister = ();
type Deleter = ();
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
}
- async fn copy(&self, _: &str, _: &str, _: OpCopy) -> Result<RpCopy> {
- Ok(RpCopy::default())
+ async fn copy(
+ &self,
+ _: &str,
+ _: &str,
+ _: OpCopy,
+ _: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ Ok((RpCopy::default(), Box::new(())))
}
async fn rename(&self, _: &str, _: &str, _: OpRename) -> Result<RpRename> {
@@ -501,6 +515,7 @@ mod tests {
type Writer = ();
type Lister = ();
type Deleter = ();
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/layers/dtrace/src/lib.rs b/core/layers/dtrace/src/lib.rs
index 0d288dc21..e0cb5bf7e 100644
--- a/core/layers/dtrace/src/lib.rs
+++ b/core/layers/dtrace/src/lib.rs
@@ -163,6 +163,7 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
type Writer = DtraceLayerWrapper<A::Writer>;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -201,6 +202,20 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
result
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ let c_from = CString::new(from).unwrap();
+ probe_lazy!(opendal, copy_start, c_from.as_ptr());
+ let result = self.inner.copy(from, to, args, opts).await;
+ probe_lazy!(opendal, copy_end, c_from.as_ptr());
+ result
+ }
+
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
let c_path = CString::new(path).unwrap();
probe_lazy!(opendal, stat_start, c_path.as_ptr());
diff --git a/core/layers/fastrace/src/lib.rs b/core/layers/fastrace/src/lib.rs
index 8a3ebd805..eaad75fb7 100644
--- a/core/layers/fastrace/src/lib.rs
+++ b/core/layers/fastrace/src/lib.rs
@@ -134,6 +134,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
type Writer = FastraceWrapper<A::Writer>;
type Lister = FastraceWrapper<A::Lister>;
type Deleter = FastraceWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -176,8 +177,14 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
}
#[trace(enter_on_poll = true)]
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner().copy(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner().copy(from, to, args, opts.clone()).await
}
#[trace(enter_on_poll = true)]
diff --git a/core/layers/foyer/src/lib.rs b/core/layers/foyer/src/lib.rs
index e0797949d..519d88eee 100644
--- a/core/layers/foyer/src/lib.rs
+++ b/core/layers/foyer/src/lib.rs
@@ -188,6 +188,7 @@ impl<A: Access> LayeredAccess for FoyerAccessor<A> {
type Writer = Writer<A>;
type Lister = A::Lister;
type Deleter = Deleter<A>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner.accessor
@@ -225,6 +226,16 @@ impl<A: Access> LayeredAccess for FoyerAccessor<A> {
}
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.accessor.copy(from, to, args, opts).await
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.accessor.list(path, args).await
}
diff --git a/core/layers/hotpath/src/lib.rs b/core/layers/hotpath/src/lib.rs
index 6e327eff8..23b8350ca 100644
--- a/core/layers/hotpath/src/lib.rs
+++ b/core/layers/hotpath/src/lib.rs
@@ -113,6 +113,7 @@ impl<A: Access> LayeredAccess for HotpathAccessor<A> {
type Writer = HotpathWrapper<A::Writer>;
type Lister = HotpathWrapper<A::Lister>;
type Deleter = HotpathWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -133,8 +134,14 @@ impl<A: Access> LayeredAccess for HotpathAccessor<A> {
Ok((rp, HotpathWrapper::new(writer)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- hotpath::measure_async(LABEL_COPY, self.inner.copy(from, to, args)).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ hotpath::measure_async(LABEL_COPY, self.inner.copy(from, to, args, opts.clone())).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
diff --git a/core/layers/immutable-index/src/lib.rs b/core/layers/immutable-index/src/lib.rs
index 6b633b40f..96fc27e9d 100644
--- a/core/layers/immutable-index/src/lib.rs
+++ b/core/layers/immutable-index/src/lib.rs
@@ -159,6 +159,7 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
type Writer = A::Writer;
type Lister = ImmutableDir;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -172,6 +173,16 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
self.inner.write(path, args).await
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let mut path = path;
if path == "/" {
@@ -243,6 +254,7 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/layers/logging/src/lib.rs b/core/layers/logging/src/lib.rs
index 47c5c2b74..2a32abff8 100644
--- a/core/layers/logging/src/lib.rs
+++ b/core/layers/logging/src/lib.rs
@@ -251,6 +251,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
type Writer = LoggingWriter<A::Writer, I>;
type Lister = LoggingLister<A::Lister, I>;
type Deleter = LoggingDeleter<A::Deleter, I>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -362,7 +363,13 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
})
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.logger.log(
&self.info,
Operation::Copy,
@@ -372,7 +379,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
);
self.inner
- .copy(from, to, args)
+ .copy(from, to, args, opts.clone())
.await
.inspect(|_| {
self.logger.log(
diff --git a/core/layers/mime-guess/src/lib.rs b/core/layers/mime-guess/src/lib.rs
index a7a068b4b..5e6c460fd 100644
--- a/core/layers/mime-guess/src/lib.rs
+++ b/core/layers/mime-guess/src/lib.rs
@@ -121,6 +121,7 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.0
@@ -136,6 +137,16 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
.await
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner().copy(from, to, args, opts).await
+ }
+
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.inner()
.stat(path, args)
diff --git a/core/layers/observe-metrics-common/src/lib.rs b/core/layers/observe-metrics-common/src/lib.rs
index 72b7dd59f..898138803 100644
--- a/core/layers/observe-metrics-common/src/lib.rs
+++ b/core/layers/observe-metrics-common/src/lib.rs
@@ -720,6 +720,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
type Writer = MetricsWrapper<A::Writer, I>;
type Lister = MetricsWrapper<A::Lister, I>;
type Deleter = MetricsWrapper<A::Deleter, I>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -818,7 +819,13 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let labels = MetricLabels::new(self.info.clone(), Operation::Copy.into_static());
let start = Instant::now();
@@ -830,7 +837,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
let res = self
.inner()
- .copy(from, to, args)
+ .copy(from, to, args, opts.clone())
.await
.inspect(|_| {
self.interceptor.observe(
diff --git a/core/layers/oteltrace/src/lib.rs b/core/layers/oteltrace/src/lib.rs
index 170e1f270..588f5bb43 100644
--- a/core/layers/oteltrace/src/lib.rs
+++ b/core/layers/oteltrace/src/lib.rs
@@ -84,6 +84,7 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
type Writer = OtelTraceWrapper<A::Writer>;
type Lister = OtelTraceWrapper<A::Lister>;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -125,14 +126,23 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
.map(|(rp, r)| (rp, OtelTraceWrapper::new(span, r)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let tracer = global::tracer("opendal");
let mut span = tracer.start("copy");
span.set_attribute(KeyValue::new("from", from.to_string()));
span.set_attribute(KeyValue::new("to", to.to_string()));
span.set_attribute(KeyValue::new("args", format!("{args:?}")));
let cx = TraceContext::current_with_span(span);
- self.inner().copy(from, to, args).with_context(cx).await
+ self.inner()
+ .copy(from, to, args, opts.clone())
+ .with_context(cx)
+ .await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
diff --git a/core/layers/retry/src/lib.rs b/core/layers/retry/src/lib.rs
index 65d1eea74..df87ed02b 100644
--- a/core/layers/retry/src/lib.rs
+++ b/core/layers/retry/src/lib.rs
@@ -293,6 +293,7 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
type Writer = RetryWrapper<A::Writer, I>;
type Lister = RetryWrapper<A::Lister, I>;
type Deleter = RetryWrapper<A::Deleter, I>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -395,9 +396,15 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
.map_err(|e| e.set_persistent())
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let mut attempt: u32 = 0;
- { || self.inner.copy(from, to, args.clone()) }
+ { || self.inner.copy(from, to, args.clone(), opts.clone()) }
.retry(self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -788,6 +795,7 @@ mod tests {
type Writer = MockWriter;
type Lister = MockLister;
type Deleter = MockDeleter;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/layers/route/src/lib.rs b/core/layers/route/src/lib.rs
index 6e555cb1f..83d32a1e2 100644
--- a/core/layers/route/src/lib.rs
+++ b/core/layers/route/src/lib.rs
@@ -152,6 +152,7 @@ impl LayeredAccess for RouteAccessor {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = RouteDeleter;
+ type Copier = oio::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -169,8 +170,14 @@ impl LayeredAccess for RouteAccessor {
self.select(path).write(path, args).await
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.select(from).copy(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.select(from).copy(from, to, args, opts.clone()).await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
diff --git a/core/layers/tail-cut/src/lib.rs b/core/layers/tail-cut/src/lib.rs
index fd473ac3b..17e75ea3d 100644
--- a/core/layers/tail-cut/src/lib.rs
+++ b/core/layers/tail-cut/src/lib.rs
@@ -364,6 +364,7 @@ impl<A: Access> LayeredAccess for TailCutAccessor<A> {
type Writer = TailCutWrapper<A::Writer>;
type Lister = TailCutWrapper<A::Lister>;
type Deleter = TailCutWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -401,9 +402,19 @@ impl<A: Access> LayeredAccess for TailCutAccessor<A> {
})
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.with_deadline(Operation::Copy, None, self.inner.copy(from, to, args))
- .await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.with_deadline(
+ Operation::Copy,
+ None,
+ self.inner.copy(from, to, args, opts.clone()),
+ )
+ .await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
diff --git a/core/layers/throttle/src/lib.rs b/core/layers/throttle/src/lib.rs
index 05be4409a..df0ffcd0a 100644
--- a/core/layers/throttle/src/lib.rs
+++ b/core/layers/throttle/src/lib.rs
@@ -119,6 +119,7 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
type Writer = ThrottleWrapper<A::Writer>;
type Lister = A::Lister;
type Deleter = A::Deleter;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -142,6 +143,16 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
.map(|(rp, w)| (rp, ThrottleWrapper::new(w, limiter)))
}
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner.copy(from, to, args, opts).await
+ }
+
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner.delete().await
}
diff --git a/core/layers/timeout/src/lib.rs b/core/layers/timeout/src/lib.rs
index e9e316503..9a24ae718 100644
--- a/core/layers/timeout/src/lib.rs
+++ b/core/layers/timeout/src/lib.rs
@@ -213,6 +213,7 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
type Writer = TimeoutWrapper<A::Writer>;
type Lister = TimeoutWrapper<A::Lister>;
type Deleter = TimeoutWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -235,9 +236,18 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.timeout(Operation::Copy, self.inner.copy(from, to, args))
- .await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.timeout(
+ Operation::Copy,
+ self.inner.copy(from, to, args, opts.clone()),
+ )
+ .await
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
@@ -377,6 +387,7 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
+ type Copier = oio::Copier;
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/layers/tracing/src/lib.rs b/core/layers/tracing/src/lib.rs
index 31fb79c3b..d61de599b 100644
--- a/core/layers/tracing/src/lib.rs
+++ b/core/layers/tracing/src/lib.rs
@@ -202,6 +202,7 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
type Writer = TracingWrapper<A::Writer>;
type Lister = TracingWrapper<A::Lister>;
type Deleter = TracingWrapper<A::Deleter>;
+ type Copier = A::Copier;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -233,8 +234,14 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
}
#[tracing::instrument(level = "debug", skip(self))]
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner().copy(from, to, args).await
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
+ self.inner().copy(from, to, args, opts.clone()).await
}
#[tracing::instrument(level = "debug", skip(self))]
diff --git a/core/services/aliyun-drive/src/backend.rs b/core/services/aliyun-drive/src/backend.rs
index de752ac99..854a01afa 100644
--- a/core/services/aliyun-drive/src/backend.rs
+++ b/core/services/aliyun-drive/src/backend.rs
@@ -196,6 +196,7 @@ impl Access for AliyunDriveBackend {
type Writer = AliyunDriveWriter;
type Lister = oio::PageLister<AliyunDriveLister>;
type Deleter = oio::OneShotDeleter<AliyunDriveDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -238,9 +239,15 @@ impl Access for AliyunDriveBackend {
Ok(RpRename::default())
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
if from == to {
- return Ok(RpCopy::default());
+ return Ok((RpCopy::default(), ()));
}
let res = self.core.get_by_path(from).await?;
let file: AliyunDriveFile =
@@ -279,7 +286,7 @@ impl Access for AliyunDriveBackend {
self.core.update_path(&file_id, to_name).await?;
}
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
diff --git a/core/services/alluxio/src/backend.rs b/core/services/alluxio/src/backend.rs
index e865dfb15..e5e16a799 100644
--- a/core/services/alluxio/src/backend.rs
+++ b/core/services/alluxio/src/backend.rs
@@ -139,6 +139,7 @@ impl Access for AlluxioBackend {
type Writer = AlluxioWriters;
type Lister = oio::PageLister<AlluxioLister>;
type Deleter = oio::OneShotDeleter<AlluxioDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs
index 5b9d98936..765465c76 100644
--- a/core/services/azblob/src/backend.rs
+++ b/core/services/azblob/src/backend.rs
@@ -473,6 +473,7 @@ impl Access for AzblobBackend {
type Writer = AzblobWriters;
type Lister = oio::PageLister<AzblobLister>;
type Deleter = oio::BatchDeleter<AzblobDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -552,13 +553,19 @@ impl Access for AzblobBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.azblob_copy_blob(from, to, args).await?;
let status = resp.status();
match status {
- StatusCode::ACCEPTED => Ok(RpCopy::default()),
+ StatusCode::ACCEPTED => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/azdls/src/backend.rs b/core/services/azdls/src/backend.rs
index 8409dcb04..ececcfbce 100644
--- a/core/services/azdls/src/backend.rs
+++ b/core/services/azdls/src/backend.rs
@@ -371,6 +371,7 @@ impl Access for AzdlsBackend {
type Writer = AzdlsWriters;
type Lister = oio::PageLister<AzdlsLister>;
type Deleter = oio::OneShotDeleter<AzdlsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/azfile/src/backend.rs b/core/services/azfile/src/backend.rs
index ede0296c1..06412c1d3 100644
--- a/core/services/azfile/src/backend.rs
+++ b/core/services/azfile/src/backend.rs
@@ -276,6 +276,7 @@ impl Access for AzfileBackend {
type Writer = AzfileWriters;
type Lister = oio::PageLister<AzfileLister>;
type Deleter = oio::OneShotDeleter<AzfileDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/b2/src/backend.rs b/core/services/b2/src/backend.rs
index 5f1d0fea8..76fb9eac6 100644
--- a/core/services/b2/src/backend.rs
+++ b/core/services/b2/src/backend.rs
@@ -228,6 +228,7 @@ impl Access for B2Backend {
type Writer = B2Writers;
type Lister = oio::PageLister<B2Lister>;
type Deleter = oio::OneShotDeleter<B2Deleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -296,7 +297,13 @@ impl Access for B2Backend {
))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let file_info = self.core.get_file_info(from, None).await?;
let source_file_id = file_info.file_id;
@@ -310,7 +317,7 @@ impl Access for B2Backend {
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/cacache/src/backend.rs b/core/services/cacache/src/backend.rs
index 277686174..2b3b5d880 100644
--- a/core/services/cacache/src/backend.rs
+++ b/core/services/cacache/src/backend.rs
@@ -87,6 +87,7 @@ impl Access for CacacheBackend {
type Writer = CacacheWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<CacacheDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/cloudflare-kv/src/backend.rs b/core/services/cloudflare-kv/src/backend.rs
index ee3610b44..c7af780e3 100644
--- a/core/services/cloudflare-kv/src/backend.rs
+++ b/core/services/cloudflare-kv/src/backend.rs
@@ -195,6 +195,7 @@ impl Access for CloudflareKvBackend {
type Writer = oio::OneShotWriter<CloudflareWriter>;
type Lister = oio::PageLister<CloudflareKvLister>;
type Deleter = oio::BatchDeleter<CloudflareKvDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/compfs/src/backend.rs b/core/services/compfs/src/backend.rs
index d4c4611b5..8b23445a5 100644
--- a/core/services/compfs/src/backend.rs
+++ b/core/services/compfs/src/backend.rs
@@ -128,6 +128,7 @@ impl Access for CompfsBackend {
type Writer = CompfsWriter;
type Lister = Option<CompfsLister>;
type Deleter = oio::OneShotDeleter<CompfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -171,7 +172,13 @@ impl Access for CompfsBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let from = self.core.prepare_path(from);
let to = self.core.prepare_path(to);
@@ -195,7 +202,7 @@ impl Access for CompfsBackend {
})
.await?;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
diff --git a/core/services/cos/src/backend.rs b/core/services/cos/src/backend.rs
index fc3eed8f8..956c3c0e0 100644
--- a/core/services/cos/src/backend.rs
+++ b/core/services/cos/src/backend.rs
@@ -303,6 +303,7 @@ impl Access for CosBackend {
type Writer = CosWriters;
type Lister = CosListers;
type Deleter = oio::OneShotDeleter<CosDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -394,13 +395,19 @@ impl Access for CosBackend {
Ok((RpList::default(), l))
}
- async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.cos_copy_object(from, to, &args).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/d1/src/backend.rs b/core/services/d1/src/backend.rs
index 4c2cfb11b..1df7101e7 100644
--- a/core/services/d1/src/backend.rs
+++ b/core/services/d1/src/backend.rs
@@ -227,6 +227,7 @@ impl Access for D1Backend {
type Writer = D1Writer;
type Lister = ();
type Deleter = oio::OneShotDeleter<D1Deleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/dashmap/src/backend.rs b/core/services/dashmap/src/backend.rs
index 365e7e828..b7c33b5da 100644
--- a/core/services/dashmap/src/backend.rs
+++ b/core/services/dashmap/src/backend.rs
@@ -116,6 +116,7 @@ impl Access for DashmapBackend {
type Writer = DashmapWriter;
type Lister = oio::HierarchyLister<DashmapLister>;
type Deleter = oio::OneShotDeleter<DashmapDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/dbfs/src/backend.rs b/core/services/dbfs/src/backend.rs
index fead859dc..29e04fb71 100644
--- a/core/services/dbfs/src/backend.rs
+++ b/core/services/dbfs/src/backend.rs
@@ -128,6 +128,7 @@ impl Access for DbfsBackend {
type Writer = oio::OneShotWriter<DbfsWriter>;
type Lister = oio::PageLister<DbfsLister>;
type Deleter = oio::OneShotDeleter<DbfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/services/dropbox/src/backend.rs b/core/services/dropbox/src/backend.rs
index fd25f4de5..1a05deed7 100644
--- a/core/services/dropbox/src/backend.rs
+++ b/core/services/dropbox/src/backend.rs
@@ -39,6 +39,7 @@ impl Access for DropboxBackend {
type Writer = oio::OneShotWriter<DropboxWriter>;
type Lister = oio::PageLister<DropboxLister>;
type Deleter = oio::OneShotDeleter<DropboxDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -150,17 +151,23 @@ impl Access for DropboxBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.dropbox_copy(from, to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => {
let err = parse_error(resp);
match err.kind() {
- ErrorKind::NotFound => Ok(RpCopy::default()),
+ ErrorKind::NotFound => Ok((RpCopy::default(), ())),
_ => Err(err),
}
}
diff --git a/core/services/etcd/src/backend.rs b/core/services/etcd/src/backend.rs
index 3012021ef..cd1fea9b1 100644
--- a/core/services/etcd/src/backend.rs
+++ b/core/services/etcd/src/backend.rs
@@ -208,6 +208,7 @@ impl Access for EtcdBackend {
type Writer = EtcdWriter;
type Lister = oio::HierarchyLister<EtcdLister>;
type Deleter = oio::OneShotDeleter<EtcdDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/foundationdb/src/backend.rs b/core/services/foundationdb/src/backend.rs
index 0e7a46049..865f92985 100644
--- a/core/services/foundationdb/src/backend.rs
+++ b/core/services/foundationdb/src/backend.rs
@@ -125,6 +125,7 @@ impl Access for FoundationdbBackend {
type Writer = FoundationdbWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<FoundationdbDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/foyer/src/backend.rs b/core/services/foyer/src/backend.rs
index feefbac7b..303a2054f 100644
--- a/core/services/foyer/src/backend.rs
+++ b/core/services/foyer/src/backend.rs
@@ -244,6 +244,7 @@ impl Access for FoyerBackend {
type Writer = FoyerWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<FoyerDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/fs/src/backend.rs b/core/services/fs/src/backend.rs
index 7f2040ef7..fd01b1d57 100644
--- a/core/services/fs/src/backend.rs
+++ b/core/services/fs/src/backend.rs
@@ -188,6 +188,7 @@ impl Access for FsBackend {
type Writer = FsWriters;
type Lister = Option<FsLister<tokio::fs::ReadDir>>;
type Deleter = oio::OneShotDeleter<FsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -258,9 +259,15 @@ impl Access for FsBackend {
}
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.core.fs_copy(from, to).await?;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
diff --git a/core/services/ftp/src/backend.rs b/core/services/ftp/src/backend.rs
index 048c6b220..56458fffd 100644
--- a/core/services/ftp/src/backend.rs
+++ b/core/services/ftp/src/backend.rs
@@ -195,6 +195,7 @@ impl Access for FtpBackend {
type Writer = FtpWriter;
type Lister = FtpLister;
type Deleter = oio::OneShotDeleter<FtpDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info()
diff --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs
index 7890e1f31..d066a5e9a 100644
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@ -410,6 +410,7 @@ impl Access for GcsBackend {
type Writer = GcsWriters;
type Lister = oio::PageLister<GcsLister>;
type Deleter = oio::BatchDeleter<GcsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -475,11 +476,17 @@ impl Access for GcsBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.gcs_copy_object(from, to).await?;
if resp.status().is_success() {
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
} else {
Err(parse_error(resp))
}
diff --git a/core/services/gdrive/src/backend.rs b/core/services/gdrive/src/backend.rs
index 082171fd6..b928a8c01 100644
--- a/core/services/gdrive/src/backend.rs
+++ b/core/services/gdrive/src/backend.rs
@@ -47,6 +47,7 @@ impl Access for GdriveBackend {
type Writer = oio::OneShotWriter<GdriveWriter>;
type Lister = GdriveListers;
type Deleter = oio::OneShotDeleter<GdriveDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -208,7 +209,13 @@ impl Access for GdriveBackend {
}
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let source = build_abs_path(&self.core.root, from);
let target = build_abs_path(&self.core.root, to);
let resp = self.core.gdrive_copy(from, to).await?;
@@ -238,7 +245,7 @@ impl Access for GdriveBackend {
}
self.core.record_recent_upsert(&to_path, metadata).await;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
StatusCode::NOT_FOUND => {
self.core.refresh_path(&source).await;
@@ -272,7 +279,7 @@ impl Access for GdriveBackend {
}
self.core.record_recent_upsert(&to_path, metadata).await;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
_ => Err(parse_error(resp)),
}
diff --git a/core/services/ghac/src/backend.rs b/core/services/ghac/src/backend.rs
index 16570dfbd..fed800c4e 100644
--- a/core/services/ghac/src/backend.rs
+++ b/core/services/ghac/src/backend.rs
@@ -214,6 +214,7 @@ impl Access for GhacBackend {
type Writer = GhacWriter;
type Lister = ();
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/github/src/backend.rs b/core/services/github/src/backend.rs
index 2fca63d4f..dff01aaa3 100644
--- a/core/services/github/src/backend.rs
+++ b/core/services/github/src/backend.rs
@@ -165,6 +165,7 @@ impl Access for GithubBackend {
type Writer = GithubWriters;
type Lister = oio::PageLister<GithubLister>;
type Deleter = oio::OneShotDeleter<GithubDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/goosefs/src/backend.rs b/core/services/goosefs/src/backend.rs
index 52fdfd240..fa5323405 100644
--- a/core/services/goosefs/src/backend.rs
+++ b/core/services/goosefs/src/backend.rs
@@ -307,6 +307,7 @@ impl Access for GoosefsBackend {
type Writer = GoosefsWriters;
type Lister = oio::PageLister<GoosefsLister>;
type Deleter = oio::OneShotDeleter<GoosefsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/gridfs/src/backend.rs b/core/services/gridfs/src/backend.rs
index 9e76b64d9..cc31f0e0c 100644
--- a/core/services/gridfs/src/backend.rs
+++ b/core/services/gridfs/src/backend.rs
@@ -191,6 +191,7 @@ impl Access for GridfsBackend {
type Writer = GridfsWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<GridfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/hdfs-native/src/backend.rs b/core/services/hdfs-native/src/backend.rs
index 3ad8e07f7..535aa5780 100644
--- a/core/services/hdfs-native/src/backend.rs
+++ b/core/services/hdfs-native/src/backend.rs
@@ -173,6 +173,7 @@ impl Access for HdfsNativeBackend {
type Writer = HdfsNativeWriter;
type Lister = Option<HdfsNativeLister>;
type Deleter = oio::OneShotDeleter<HdfsNativeDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs
index 49ae41337..c652308a9 100644
--- a/core/services/hdfs/src/backend.rs
+++ b/core/services/hdfs/src/backend.rs
@@ -201,6 +201,7 @@ impl Access for HdfsBackend {
type Writer = HdfsWriter<hdrs::AsyncFile>;
type Lister = Option<HdfsLister>;
type Deleter = oio::OneShotDeleter<HdfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/hf/src/backend.rs b/core/services/hf/src/backend.rs
index 525aa4647..6297b7164 100644
--- a/core/services/hf/src/backend.rs
+++ b/core/services/hf/src/backend.rs
@@ -236,6 +236,7 @@ impl Access for HfBackend {
type Writer = HfWriter;
type Lister = oio::PageLister<HfLister>;
type Deleter = oio::BatchDeleter<HfDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/http/src/backend.rs b/core/services/http/src/backend.rs
index 9d99cf089..96cfe74bc 100644
--- a/core/services/http/src/backend.rs
+++ b/core/services/http/src/backend.rs
@@ -174,6 +174,7 @@ impl Access for HttpBackend {
type Writer = ();
type Lister = ();
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/ipfs/src/backend.rs b/core/services/ipfs/src/backend.rs
index bb7a4fc47..fb60ce3b5 100644
--- a/core/services/ipfs/src/backend.rs
+++ b/core/services/ipfs/src/backend.rs
@@ -147,6 +147,7 @@ impl Access for IpfsBackend {
type Writer = ();
type Lister = oio::PageLister<DirStream>;
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/ipmfs/src/backend.rs b/core/services/ipmfs/src/backend.rs
index 0d39d665a..4407f31c2 100644
--- a/core/services/ipmfs/src/backend.rs
+++ b/core/services/ipmfs/src/backend.rs
@@ -42,6 +42,7 @@ impl Access for IpmfsBackend {
type Writer = oio::OneShotWriter<IpmfsWriter>;
type Lister = oio::PageLister<IpmfsLister>;
type Deleter = oio::OneShotDeleter<IpmfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/koofr/src/backend.rs b/core/services/koofr/src/backend.rs
index f43fe69fc..075f2d539 100644
--- a/core/services/koofr/src/backend.rs
+++ b/core/services/koofr/src/backend.rs
@@ -194,6 +194,7 @@ impl Access for KoofrBackend {
type Writer = KoofrWriters;
type Lister = oio::PageLister<KoofrLister>;
type Deleter = oio::OneShotDeleter<KoofrDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -274,11 +275,17 @@ impl Access for KoofrBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.core.ensure_dir_exists(to).await?;
if from == to {
- return Ok(RpCopy::default());
+ return Ok((RpCopy::default(), ()));
}
let resp = self.core.remove(to).await?;
@@ -294,7 +301,7 @@ impl Access for KoofrBackend {
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/lakefs/src/backend.rs b/core/services/lakefs/src/backend.rs
index e789cc33c..1b62ea4a6 100644
--- a/core/services/lakefs/src/backend.rs
+++ b/core/services/lakefs/src/backend.rs
@@ -193,6 +193,7 @@ impl Access for LakefsBackend {
type Writer = oio::OneShotWriter<LakefsWriter>;
type Lister = oio::PageLister<LakefsLister>;
type Deleter = oio::OneShotDeleter<LakefsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -270,13 +271,19 @@ impl Access for LakefsBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.copy_object(from, to).await?;
let status = resp.status();
match status {
- StatusCode::CREATED => Ok(RpCopy::default()),
+ StatusCode::CREATED => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/memcached/src/backend.rs b/core/services/memcached/src/backend.rs
index 4ea91c0c9..9062267a8 100644
--- a/core/services/memcached/src/backend.rs
+++ b/core/services/memcached/src/backend.rs
@@ -220,6 +220,7 @@ impl Access for MemcachedBackend {
type Writer = MemcachedWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<MemcachedDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/mini_moka/src/backend.rs b/core/services/mini_moka/src/backend.rs
index dcad3c875..1458f5439 100644
--- a/core/services/mini_moka/src/backend.rs
+++ b/core/services/mini_moka/src/backend.rs
@@ -135,6 +135,7 @@ impl Access for MiniMokaBackend {
type Writer = MiniMokaWriter;
type Lister = oio::HierarchyLister<MiniMokaLister>;
type Deleter = oio::OneShotDeleter<MiniMokaDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/services/moka/src/backend.rs b/core/services/moka/src/backend.rs
index e01387a1d..d54075a65 100644
--- a/core/services/moka/src/backend.rs
+++ b/core/services/moka/src/backend.rs
@@ -227,6 +227,7 @@ impl Access for MokaBackend {
type Writer = MokaWriter;
type Lister = oio::HierarchyLister<MokaLister>;
type Deleter = oio::OneShotDeleter<MokaDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/mongodb/src/backend.rs b/core/services/mongodb/src/backend.rs
index 676ac36d1..be3a3f93f 100644
--- a/core/services/mongodb/src/backend.rs
+++ b/core/services/mongodb/src/backend.rs
@@ -209,6 +209,7 @@ impl Access for MongodbBackend {
type Writer = MongodbWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<MongodbDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/monoiofs/src/backend.rs b/core/services/monoiofs/src/backend.rs
index 59c802901..1d56d31a1 100644
--- a/core/services/monoiofs/src/backend.rs
+++ b/core/services/monoiofs/src/backend.rs
@@ -97,6 +97,7 @@ impl Access for MonoiofsBackend {
type Writer = MonoiofsWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<MonoiofsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -170,7 +171,13 @@ impl Access for MonoiofsBackend {
Ok(RpCreateDir::default())
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let from = self.core.prepare_path(from);
// ensure file exists
self.core
@@ -223,6 +230,6 @@ impl Access for MonoiofsBackend {
})
.await
.map_err(new_std_io_error)?;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
}
diff --git a/core/services/mysql/src/backend.rs b/core/services/mysql/src/backend.rs
index ff0d68f0a..5ce0b4a6a 100644
--- a/core/services/mysql/src/backend.rs
+++ b/core/services/mysql/src/backend.rs
@@ -194,6 +194,7 @@ impl Access for MysqlBackend {
type Writer = MysqlWriter;
type Lister = oio::HierarchyLister<MysqlLister>;
type Deleter = oio::OneShotDeleter<MysqlDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/obs/src/backend.rs b/core/services/obs/src/backend.rs
index 99c5fe2ca..eba4b232e 100644
--- a/core/services/obs/src/backend.rs
+++ b/core/services/obs/src/backend.rs
@@ -273,6 +273,7 @@ impl Access for ObsBackend {
type Writer = ObsWriters;
type Lister = oio::PageLister<ObsLister>;
type Deleter = oio::OneShotDeleter<ObsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -363,13 +364,19 @@ impl Access for ObsBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.obs_copy_object(from, to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/onedrive/src/backend.rs b/core/services/onedrive/src/backend.rs
index e6ecb8150..31b0a2b97 100644
--- a/core/services/onedrive/src/backend.rs
+++ b/core/services/onedrive/src/backend.rs
@@ -39,6 +39,7 @@ impl Access for OnedriveBackend {
type Writer = oio::OneShotWriter<OneDriveWriter>;
type Lister = oio::PageLister<OneDriveLister>;
type Deleter = oio::OneShotDeleter<OneDriveDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -95,10 +96,16 @@ impl Access for OnedriveBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let monitor_url = self.core.initialize_copy(from, to).await?;
self.core.wait_until_complete(monitor_url).await?;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
diff --git a/core/services/opfs/src/backend.rs b/core/services/opfs/src/backend.rs
index ab50212a5..63a80bbda 100644
--- a/core/services/opfs/src/backend.rs
+++ b/core/services/opfs/src/backend.rs
@@ -75,6 +75,7 @@ impl Access for OpfsBackend {
type Lister = OpfsLister;
type Deleter = oio::OneShotDeleter<OpfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/oss/src/backend.rs b/core/services/oss/src/backend.rs
index 887a0e0a8..f5b0a3f57 100644
--- a/core/services/oss/src/backend.rs
+++ b/core/services/oss/src/backend.rs
@@ -600,6 +600,7 @@ impl Access for OssBackend {
type Writer = OssWriters;
type Lister = OssListers;
type Deleter = oio::BatchDeleter<OssDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -688,12 +689,18 @@ impl Access for OssBackend {
Ok((RpList::default(), l))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.oss_copy_object(from, to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/pcloud/src/backend.rs b/core/services/pcloud/src/backend.rs
index 2b300e0ac..7e6d1898d 100644
--- a/core/services/pcloud/src/backend.rs
+++ b/core/services/pcloud/src/backend.rs
@@ -183,6 +183,7 @@ impl Access for PcloudBackend {
type Writer = PcloudWriters;
type Lister = oio::PageLister<PcloudLister>;
type Deleter = oio::OneShotDeleter<PcloudDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -261,7 +262,13 @@ impl Access for PcloudBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.core.ensure_dir_exists(to).await?;
let resp = if from.ends_with('/') {
@@ -285,7 +292,7 @@ impl Access for PcloudBackend {
return Err(Error::new(ErrorKind::Unexpected, format!("{resp:?}")));
}
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
_ => Err(parse_error(resp)),
}
diff --git a/core/services/persy/src/backend.rs b/core/services/persy/src/backend.rs
index a43dd4a5b..519c29dc9 100644
--- a/core/services/persy/src/backend.rs
+++ b/core/services/persy/src/backend.rs
@@ -154,6 +154,7 @@ impl Access for PersyBackend {
type Writer = PersyWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<PersyDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/postgresql/src/backend.rs b/core/services/postgresql/src/backend.rs
index f92dd90f5..ea1b0c5f5 100644
--- a/core/services/postgresql/src/backend.rs
+++ b/core/services/postgresql/src/backend.rs
@@ -186,6 +186,7 @@ impl Access for PostgresqlBackend {
type Writer = PostgresqlWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<PostgresqlDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/redb/src/backend.rs b/core/services/redb/src/backend.rs
index 48ad108df..11d9f1c2c 100644
--- a/core/services/redb/src/backend.rs
+++ b/core/services/redb/src/backend.rs
@@ -172,6 +172,7 @@ impl Access for RedbBackend {
type Writer = RedbWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<RedbDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/redis/src/backend.rs b/core/services/redis/src/backend.rs
index 6d809a504..a06012348 100644
--- a/core/services/redis/src/backend.rs
+++ b/core/services/redis/src/backend.rs
@@ -312,6 +312,7 @@ impl Access for RedisBackend {
type Writer = RedisWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<RedisDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/rocksdb/src/backend.rs b/core/services/rocksdb/src/backend.rs
index c3a27e2b6..8fb30be06 100644
--- a/core/services/rocksdb/src/backend.rs
+++ b/core/services/rocksdb/src/backend.rs
@@ -122,6 +122,7 @@ impl Access for RocksdbBackend {
type Writer = RocksdbWriter;
type Lister = oio::HierarchyLister<RocksdbLister>;
type Deleter = oio::OneShotDeleter<RocksdbDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs
index de04e42e2..c998267d2 100644
--- a/core/services/s3/src/backend.rs
+++ b/core/services/s3/src/backend.rs
@@ -999,6 +999,7 @@ impl Access for S3Backend {
type Writer = S3Writers;
type Lister = S3Listers;
type Deleter = oio::BatchDeleter<S3Deleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -1095,7 +1096,13 @@ impl Access for S3Backend {
Ok((RpList::default(), l))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.s3_copy_object(from, to).await?;
let status = resp.status();
@@ -1122,7 +1129,7 @@ impl Access for S3Backend {
);
}
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
_ => Err(parse_error(resp)),
}
diff --git a/core/services/seafile/src/backend.rs b/core/services/seafile/src/backend.rs
index 0c7653d4a..636d78595 100644
--- a/core/services/seafile/src/backend.rs
+++ b/core/services/seafile/src/backend.rs
@@ -203,6 +203,7 @@ impl Access for SeafileBackend {
type Writer = SeafileWriters;
type Lister = oio::PageLister<SeafileLister>;
type Deleter = oio::OneShotDeleter<SeafileDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/sftp/src/backend.rs b/core/services/sftp/src/backend.rs
index 272c02ba6..c685939ca 100644
--- a/core/services/sftp/src/backend.rs
+++ b/core/services/sftp/src/backend.rs
@@ -214,6 +214,7 @@ impl Access for SftpBackend {
type Writer = SftpWriter;
type Lister = Option<SftpLister>;
type Deleter = oio::OneShotDeleter<SftpDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -354,7 +355,13 @@ impl Access for SftpBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let client = self.core.connect().await?;
let mut fs = client.fs();
@@ -374,7 +381,7 @@ impl Access for SftpBackend {
.await
.map_err(parse_sftp_error)?;
- Ok(RpCopy::default())
+ Ok((RpCopy::default(), ()))
}
async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
diff --git a/core/services/sled/src/backend.rs b/core/services/sled/src/backend.rs
index 9f0cec989..3832363e5 100644
--- a/core/services/sled/src/backend.rs
+++ b/core/services/sled/src/backend.rs
@@ -146,6 +146,7 @@ impl Access for SledBackend {
type Writer = SledWriter;
type Lister = oio::HierarchyLister<SledLister>;
type Deleter = oio::OneShotDeleter<SledDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/sqlite/src/backend.rs b/core/services/sqlite/src/backend.rs
index 5e1113596..9bdbc73df 100644
--- a/core/services/sqlite/src/backend.rs
+++ b/core/services/sqlite/src/backend.rs
@@ -212,6 +212,7 @@ impl Access for SqliteBackend {
type Writer = SqliteWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<SqliteDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/surrealdb/src/backend.rs b/core/services/surrealdb/src/backend.rs
index 3255caf9a..286f737c8 100644
--- a/core/services/surrealdb/src/backend.rs
+++ b/core/services/surrealdb/src/backend.rs
@@ -240,6 +240,7 @@ impl Access for SurrealdbBackend {
type Writer = SurrealdbWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<SurrealdbDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/swift/src/backend.rs b/core/services/swift/src/backend.rs
index 594e65f46..8aad1f7e4 100644
--- a/core/services/swift/src/backend.rs
+++ b/core/services/swift/src/backend.rs
@@ -241,6 +241,7 @@ impl Access for SwiftBackend {
type Writer = oio::MultipartWriter<SwiftWriter>;
type Lister = oio::PageLister<SwiftLister>;
type Deleter = oio::BatchDeleter<SwiftDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -335,7 +336,13 @@ impl Access for SwiftBackend {
)))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
// cannot copy objects larger than 5 GB.
// Reference: https://docs.openstack.org/api-ref/object-store/#copy-object
let resp = self.core.swift_copy(from, to).await?;
@@ -343,7 +350,7 @@ impl Access for SwiftBackend {
let status = resp.status();
match status {
- StatusCode::CREATED | StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::CREATED | StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/tikv/src/backend.rs b/core/services/tikv/src/backend.rs
index 281dd0b69..4268c91cf 100644
--- a/core/services/tikv/src/backend.rs
+++ b/core/services/tikv/src/backend.rs
@@ -148,6 +148,7 @@ impl Access for TikvBackend {
type Writer = TikvWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<TikvDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
diff --git a/core/services/tos/src/backend.rs b/core/services/tos/src/backend.rs
index 21c4694c6..30149ccd3 100644
--- a/core/services/tos/src/backend.rs
+++ b/core/services/tos/src/backend.rs
@@ -252,6 +252,7 @@ impl Access for TosBackend {
type Writer = oio::MultipartWriter<TosWriter>;
type Lister = ();
type Deleter = oio::BatchDeleter<TosDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/upyun/src/backend.rs b/core/services/upyun/src/backend.rs
index 7918d6350..bf17987f6 100644
--- a/core/services/upyun/src/backend.rs
+++ b/core/services/upyun/src/backend.rs
@@ -193,6 +193,7 @@ impl Access for UpyunBackend {
type Writer = UpyunWriters;
type Lister = oio::PageLister<UpyunLister>;
type Deleter = oio::OneShotDeleter<UpyunDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -258,13 +259,19 @@ impl Access for UpyunBackend {
Ok((RpList::default(), oio::PageLister::new(l)))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.copy(from, to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/vercel-artifacts/src/backend.rs b/core/services/vercel-artifacts/src/backend.rs
index 9228926c7..a2ef7db2d 100644
--- a/core/services/vercel-artifacts/src/backend.rs
+++ b/core/services/vercel-artifacts/src/backend.rs
@@ -37,6 +37,7 @@ impl Access for VercelArtifactsBackend {
type Writer = oio::OneShotWriter<VercelArtifactsWriter>;
type Lister = ();
type Deleter = ();
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/vercel-blob/src/backend.rs b/core/services/vercel-blob/src/backend.rs
index 264b56271..d9f792fe6 100644
--- a/core/services/vercel-blob/src/backend.rs
+++ b/core/services/vercel-blob/src/backend.rs
@@ -142,6 +142,7 @@ impl Access for VercelBlobBackend {
type Writer = VercelBlobWriters;
type Lister = oio::PageLister<VercelBlobLister>;
type Deleter = oio::OneShotDeleter<VercelBlobDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -191,13 +192,19 @@ impl Access for VercelBlobBackend {
Ok((RpWrite::default(), w))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.copy(from, to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(RpCopy::default()),
+ StatusCode::OK => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/webdav/src/backend.rs b/core/services/webdav/src/backend.rs
index 807c8d525..05cae24d3 100644
--- a/core/services/webdav/src/backend.rs
+++ b/core/services/webdav/src/backend.rs
@@ -257,6 +257,7 @@ impl Access for WebdavBackend {
type Writer = oio::OneShotWriter<WebdavWriter>;
type Lister = oio::PageLister<WebdavLister>;
type Deleter = oio::OneShotDeleter<WebdavDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -315,13 +316,19 @@ impl Access for WebdavBackend {
))
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
let resp = self.core.webdav_copy(from, to).await?;
let status = resp.status();
match status {
- StatusCode::CREATED | StatusCode::NO_CONTENT => Ok(RpCopy::default()),
+ StatusCode::CREATED | StatusCode::NO_CONTENT => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/webhdfs/src/backend.rs b/core/services/webhdfs/src/backend.rs
index f463a4396..dec4f374a 100644
--- a/core/services/webhdfs/src/backend.rs
+++ b/core/services/webhdfs/src/backend.rs
@@ -241,6 +241,7 @@ impl Access for WebhdfsBackend {
type Writer = WebhdfsWriters;
type Lister = oio::PageLister<WebhdfsLister>;
type Deleter = oio::OneShotDeleter<WebhdfsDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/services/yandex-disk/src/backend.rs b/core/services/yandex-disk/src/backend.rs
index 3bcb4d268..19fd116f1 100644
--- a/core/services/yandex-disk/src/backend.rs
+++ b/core/services/yandex-disk/src/backend.rs
@@ -141,6 +141,7 @@ impl Access for YandexDiskBackend {
type Writer = YandexDiskWriters;
type Lister = oio::PageLister<YandexDiskLister>;
type Deleter = oio::OneShotDeleter<YandexDiskDeleter>;
+ type Copier = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -165,7 +166,13 @@ impl Access for YandexDiskBackend {
}
}
- async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+ async fn copy(
+ &self,
+ from: &str,
+ to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> Result<(RpCopy, Self::Copier)> {
self.core.ensure_dir_exists(to).await?;
let resp = self.core.copy(from, to).await?;
@@ -173,7 +180,7 @@ impl Access for YandexDiskBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::CREATED => Ok(RpCopy::default()),
+ StatusCode::OK | StatusCode::CREATED => Ok((RpCopy::default(), ())),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/tests/behavior/async_copy.rs b/core/tests/behavior/async_copy.rs
index 7734cafb7..ca3e5b289 100644
--- a/core/tests/behavior/async_copy.rs
+++ b/core/tests/behavior/async_copy.rs
@@ -32,7 +32,10 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
test_copy_target_dir,
test_copy_self,
test_copy_nested,
- test_copy_overwrite
+ test_copy_overwrite,
+ test_copier_file,
+ test_copier_completion,
+ test_copier_abort
))
}
@@ -40,7 +43,9 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
tests.extend(async_trials!(
op,
test_copy_with_if_not_exists_to_new_file,
- test_copy_with_if_not_exists_to_existing_file
+ test_copy_with_if_not_exists_to_existing_file,
+ test_copier_with_if_not_exists_to_new_file,
+ test_copier_with_if_not_exists_to_existing_file
))
}
}
@@ -311,3 +316,127 @@ pub async fn test_copy_with_if_not_exists_to_existing_file(op: Operator) -> Resu
op.delete(&target_path).await.expect("delete must succeed");
Ok(())
}
+
+/// Copier should copy a file and commit the target after completion.
+pub async fn test_copier_file(op: Operator) -> Result<()> {
+ let source_path = uuid::Uuid::new_v4().to_string();
+ let (source_content, _) = gen_bytes(op.info().full_capability());
+
+ op.write(&source_path, source_content.clone()).await?;
+
+ let target_path = uuid::Uuid::new_v4().to_string();
+ let mut copier = op.copier(&source_path, &target_path).await?;
+
+ while copier.next().await?.is_some() {}
+
+ let target_content = op
+ .read(&target_path)
+ .await
+ .expect("read must succeed")
+ .to_bytes();
+ assert_eq!(
+ sha256_digest(target_content),
+ sha256_digest(&source_content),
+ );
+
+ op.delete(&source_path).await.expect("delete must succeed");
+ op.delete(&target_path).await.expect("delete must succeed");
+ Ok(())
+}
+
+/// Completed copier should keep returning `None`.
+pub async fn test_copier_completion(op: Operator) -> Result<()> {
+ let source_path = uuid::Uuid::new_v4().to_string();
+ let (source_content, _) = gen_bytes(op.info().full_capability());
+
+ op.write(&source_path, source_content).await?;
+
+ let target_path = uuid::Uuid::new_v4().to_string();
+ let mut copier = op.copier(&source_path, &target_path).await?;
+
+ while copier.next().await?.is_some() {}
+ assert!(copier.next().await?.is_none());
+
+ op.delete(&source_path).await.expect("delete must succeed");
+ op.delete(&target_path).await.expect("delete must succeed");
+ Ok(())
+}
+
+/// Aborting a copier should be explicit and idempotent for completed one-shot copies.
+pub async fn test_copier_abort(op: Operator) -> Result<()> {
+ let source_path = uuid::Uuid::new_v4().to_string();
+ let (source_content, _) = gen_bytes(op.info().full_capability());
+
+ op.write(&source_path, source_content).await?;
+
+ let target_path = uuid::Uuid::new_v4().to_string();
+ let mut copier = op.copier(&source_path, &target_path).await?;
+ copier.abort().await?;
+
+ op.delete(&source_path).await.expect("delete must succeed");
+ let _ = op.delete(&target_path).await;
+ Ok(())
+}
+
+/// Copier with if_not_exists to a new file should succeed.
+pub async fn test_copier_with_if_not_exists_to_new_file(op: Operator) -> Result<()> {
+ let source_path = uuid::Uuid::new_v4().to_string();
+ let (source_content, _) = gen_bytes(op.info().full_capability());
+
+ op.write(&source_path, source_content.clone()).await?;
+
+ let target_path = uuid::Uuid::new_v4().to_string();
+ let mut copier = op
+ .copier_with(&source_path, &target_path)
+ .if_not_exists(true)
+ .await?;
+
+ while copier.next().await?.is_some() {}
+
+ let target_content = op
+ .read(&target_path)
+ .await
+ .expect("read must succeed")
+ .to_bytes();
+ assert_eq!(
+ sha256_digest(target_content),
+ sha256_digest(&source_content),
+ );
+
+ op.delete(&source_path).await.expect("delete must succeed");
+ op.delete(&target_path).await.expect("delete must succeed");
+ Ok(())
+}
+
+/// Copier with if_not_exists to an existing file should fail.
+pub async fn test_copier_with_if_not_exists_to_existing_file(op: Operator) -> Result<()> {
+ let source_path = uuid::Uuid::new_v4().to_string();
+ let (source_content, _) = gen_bytes(op.info().full_capability());
+
+ op.write(&source_path, source_content).await?;
+
+ let target_path = uuid::Uuid::new_v4().to_string();
+ let (target_content, _) = gen_bytes(op.info().full_capability());
+ op.write(&target_path, target_content.clone()).await?;
+
+ let err = op
+ .copier_with(&source_path, &target_path)
+ .if_not_exists(true)
+ .await
+ .expect_err("copier must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ let current_content = op
+ .read(&target_path)
+ .await
+ .expect("read must succeed")
+ .to_bytes();
+ assert_eq!(
+ sha256_digest(current_content),
+ sha256_digest(&target_content),
+ );
+
+ op.delete(&source_path).await.expect("delete must succeed");
+ op.delete(&target_path).await.expect("delete must succeed");
+ Ok(())
+}