You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/12/05 11:43:36 UTC
(opendal) 08/09: Check pass
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch deleter
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit accaf94c5cdb0e755a283c061b7ee90627bcce9b
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Dec 5 18:31:03 2024 +0800
Check pass
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/async_backtrace.rs | 1 -
core/src/layers/chaos.rs | 1 -
core/src/layers/concurrent_limit.rs | 1 -
core/src/layers/error_context.rs | 1 -
core/src/layers/immutable_index.rs | 1 -
core/src/layers/logging.rs | 1 -
core/src/layers/mime_guess.rs | 1 -
core/src/layers/observe/metrics.rs | 1 -
core/src/layers/retry.rs | 3 +-
core/src/layers/throttle.rs | 1 -
core/src/layers/type_eraser.rs | 1 -
core/src/raw/adapters/kv/backend.rs | 1 -
core/src/raw/oio/delete/batch_delete.rs | 2 +-
core/src/services/azblob/backend.rs | 8 ---
core/src/services/azblob/core.rs | 1 -
core/src/services/dbfs/delete.rs | 2 +-
core/src/services/dropbox/backend.rs | 1 -
core/src/services/dropbox/core.rs | 5 --
core/src/services/github/delete.rs | 2 -
core/src/services/oss/backend.rs | 2 -
core/src/services/oss/delete.rs | 2 -
core/src/services/s3/backend.rs | 2 -
core/src/types/delete/blocking_deleter.rs | 40 ++++++++++---
core/src/types/delete/deleter.rs | 84 +++++++++++++++++++++++-----
core/src/types/delete/futures_delete_sink.rs | 1 +
core/src/types/delete/input.rs | 2 +-
core/src/types/error.rs | 8 ---
core/src/types/operator/blocking_operator.rs | 33 ++++++++---
core/src/types/operator/operator.rs | 73 ++++++++++++++++++------
core/tests/behavior/blocking_delete.rs | 2 +-
30 files changed, 191 insertions(+), 93 deletions(-)
diff --git a/core/src/layers/async_backtrace.rs b/core/src/layers/async_backtrace.rs
index 96cf64deb..f9dc06b47 100644
--- a/core/src/layers/async_backtrace.rs
+++ b/core/src/layers/async_backtrace.rs
@@ -17,7 +17,6 @@
use crate::raw::*;
use crate::*;
-use std::future::Future;
/// Add Efficient, logical 'stack' traces of async functions for the underlying services.
///
diff --git a/core/src/layers/chaos.rs b/core/src/layers/chaos.rs
index 0c39a7c68..342e56857 100644
--- a/core/src/layers/chaos.rs
+++ b/core/src/layers/chaos.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::future::Future;
use std::sync::Arc;
use std::sync::Mutex;
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index 2aeae54e6..2fca982aa 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -16,7 +16,6 @@
// under the License.
use std::fmt::Debug;
-use std::future::Future;
use std::sync::Arc;
use tokio::sync::OwnedSemaphorePermit;
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 17ce634cb..305fc82d3 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -17,7 +17,6 @@
use std::fmt::Debug;
use std::fmt::Formatter;
-use std::future::Future;
use std::sync::Arc;
use crate::raw::*;
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index 1d8c1c9d2..5a41019d4 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -17,7 +17,6 @@
use std::collections::HashSet;
use std::fmt::Debug;
-use std::future::Future;
use std::sync::Arc;
use std::vec::IntoIter;
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 7a9a455a0..2540dc39d 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -17,7 +17,6 @@
use std::fmt::Debug;
use std::fmt::Display;
-use std::future::Future;
use std::sync::Arc;
use log::log;
diff --git a/core/src/layers/mime_guess.rs b/core/src/layers/mime_guess.rs
index 62bb6280e..245aecf55 100644
--- a/core/src/layers/mime_guess.rs
+++ b/core/src/layers/mime_guess.rs
@@ -17,7 +17,6 @@
use crate::raw::*;
use crate::Result;
-use std::future::Future;
/// A layer that can automatically set `Content-Type` based on the file extension in the path.
///
diff --git a/core/src/layers/observe/metrics.rs b/core/src/layers/observe/metrics.rs
index 4da2cfeb8..858314113 100644
--- a/core/src/layers/observe/metrics.rs
+++ b/core/src/layers/observe/metrics.rs
@@ -18,7 +18,6 @@
use std::fmt::Debug;
use std::fmt::Formatter;
use std::fmt::Write;
-use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index d9b7a69c1..1a93ed3a9 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -17,7 +17,6 @@
use std::fmt::Debug;
use std::fmt::Formatter;
-use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
@@ -799,7 +798,7 @@ mod tests {
use std::sync::Mutex;
use bytes::Bytes;
- use futures::{stream, StreamExt, TryStreamExt};
+ use futures::{stream, TryStreamExt};
use tracing_subscriber::filter::LevelFilter;
use super::*;
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index 8ffeb0edb..5b7154c92 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::future::Future;
use std::num::NonZeroU32;
use std::sync::Arc;
use std::thread;
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index b9a0567b9..cfe161aae 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -19,7 +19,6 @@ use crate::raw::*;
use crate::*;
use std::fmt::Debug;
use std::fmt::Formatter;
-use std::future::Future;
/// TypeEraseLayer will erase the types on internal accessor.
///
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index 3ed7a6cfd..56146aef0 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::future::Future;
use std::sync::Arc;
use std::vec::IntoIter;
diff --git a/core/src/raw/oio/delete/batch_delete.rs b/core/src/raw/oio/delete/batch_delete.rs
index 6c17927a2..639499998 100644
--- a/core/src/raw/oio/delete/batch_delete.rs
+++ b/core/src/raw/oio/delete/batch_delete.rs
@@ -94,7 +94,7 @@ impl<D: BatchDelete> oio::Delete for BatchDeleter<D> {
}
let batch = self.buffer.iter().cloned().collect();
- let mut result = self.inner.delete_batch(batch).await?;
+ let result = self.inner.delete_batch(batch).await?;
debug_assert!(
!result.succeeded.is_empty(),
"the number of succeeded operations must be greater than 0"
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index 946faa1de..a8baf13ba 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -22,8 +22,6 @@ use std::sync::Arc;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
-use bytes::Buf;
-use http::header::CONTENT_TYPE;
use http::Response;
use http::StatusCode;
use log::debug;
@@ -434,11 +432,6 @@ impl Builder for AzblobBuilder {
let signer = AzureStorageSigner::new();
- let batch_max_operations = self
- .config
- .batch_max_operations
- .unwrap_or(AZBLOB_BATCH_LIMIT);
-
Ok(AzblobBackend {
core: Arc::new(AzblobCore {
root,
@@ -451,7 +444,6 @@ impl Builder for AzblobBuilder {
client,
loader: cred_loader,
signer,
- batch_max_operations,
}),
has_sas_token: self.config.sas_token.is_some(),
})
diff --git a/core/src/services/azblob/core.rs b/core/src/services/azblob/core.rs
index 867f07651..6a208497f 100644
--- a/core/src/services/azblob/core.rs
+++ b/core/src/services/azblob/core.rs
@@ -68,7 +68,6 @@ pub struct AzblobCore {
pub client: HttpClient,
pub loader: AzureStorageLoader,
pub signer: AzureStorageSigner,
- pub batch_max_operations: usize,
}
impl Debug for AzblobCore {
diff --git a/core/src/services/dbfs/delete.rs b/core/src/services/dbfs/delete.rs
index 9603a9a1e..089ec756d 100644
--- a/core/src/services/dbfs/delete.rs
+++ b/core/src/services/dbfs/delete.rs
@@ -39,7 +39,7 @@ impl oio::OneShotDelete for DbfsDeleter {
let status = resp.status();
match status {
- /// NOTE: Server will return 200 even if the path doesn't exist.
+ // NOTE: Server will return 200 even if the path doesn't exist.
StatusCode::OK => Ok(()),
_ => Err(parse_error(resp)),
}
diff --git a/core/src/services/dropbox/backend.rs b/core/src/services/dropbox/backend.rs
index ee188e5f2..83f400755 100644
--- a/core/src/services/dropbox/backend.rs
+++ b/core/src/services/dropbox/backend.rs
@@ -18,7 +18,6 @@
use std::fmt::Debug;
use std::sync::Arc;
-use backon::Retryable;
use bytes::Buf;
use http::Response;
use http::StatusCode;
diff --git a/core/src/services/dropbox/core.rs b/core/src/services/dropbox/core.rs
index 23a8c6223..3f784fc18 100644
--- a/core/src/services/dropbox/core.rs
+++ b/core/src/services/dropbox/core.rs
@@ -15,14 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
use std::default::Default;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
-use std::time::Duration;
-use backon::ExponentialBuilder;
use bytes::Buf;
use bytes::Bytes;
use chrono::DateTime;
@@ -33,7 +30,6 @@ use http::header::CONTENT_TYPE;
use http::Request;
use http::Response;
use http::StatusCode;
-use once_cell::sync::Lazy;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::Mutex;
@@ -520,7 +516,6 @@ pub struct DropboxDeleteBatchResponseEntry {
#[serde(rename(deserialize = ".tag"))]
pub tag: String,
pub metadata: Option<DropboxMetadataResponse>,
- pub failure: Option<DropboxDeleteBatchFailureResponse>,
}
#[derive(Default, Debug, Deserialize)]
diff --git a/core/src/services/github/delete.rs b/core/src/services/github/delete.rs
index 2636e5f30..8f246d7e6 100644
--- a/core/src/services/github/delete.rs
+++ b/core/src/services/github/delete.rs
@@ -16,10 +16,8 @@
// under the License.
use super::core::*;
-use super::error::parse_error;
use crate::raw::*;
use crate::*;
-use http::StatusCode;
use std::sync::Arc;
pub struct GithubDeleter {
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index daba1e345..9ede2b0d3 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -15,12 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashSet;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
-use bytes::Buf;
use http::Response;
use http::StatusCode;
use http::Uri;
diff --git a/core/src/services/oss/delete.rs b/core/src/services/oss/delete.rs
index c6af25d39..544cf242c 100644
--- a/core/src/services/oss/delete.rs
+++ b/core/src/services/oss/delete.rs
@@ -22,9 +22,7 @@ use crate::raw::*;
use crate::*;
use bytes::Buf;
use http::StatusCode;
-use prost::encoding::key_len;
use std::collections::HashSet;
-use std::future::Future;
use std::sync::Arc;
pub struct OssDeleter {
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 925be76eb..7b2bbd6b2 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -25,7 +25,6 @@ use std::sync::Arc;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
-use bytes::Buf;
use constants::X_AMZ_META_PREFIX;
use http::Response;
use http::StatusCode;
@@ -44,7 +43,6 @@ use reqwest::Url;
use super::core::*;
use super::delete::S3Deleter;
use super::error::parse_error;
-use super::error::parse_s3_error_code;
use super::lister::{S3Lister, S3Listers, S3ObjectVersionsLister};
use super::writer::S3Writer;
use super::writer::S3Writers;
diff --git a/core/src/types/delete/blocking_deleter.rs b/core/src/types/delete/blocking_deleter.rs
index 22305a61a..fda26be78 100644
--- a/core/src/types/delete/blocking_deleter.rs
+++ b/core/src/types/delete/blocking_deleter.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use crate::raw::oio::Delete;
use crate::raw::*;
use crate::*;
@@ -59,20 +58,45 @@ impl BlockingDeleter {
Ok(())
}
- /// Delete a stream of paths.
- pub fn delete_iter<I, D, E>(&mut self, mut iter: I) -> Result<()>
+ /// Delete an infallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`BlockingDeleter::delete_try_iter`]: delete an fallible iterator of paths.
+ pub fn delete_iter<I, D>(&mut self, iter: I) -> Result<()>
where
- I: IntoIterator,
+ I: IntoIterator<Item = D>,
D: IntoDeleteInput,
- I::Item: Into<Result<D, Error>>,
{
let mut iter = iter.into_iter();
loop {
- match iter.next().into() {
- Some(Ok(entry)) => {
+ match iter.next() {
+ Some(entry) => {
self.delete(entry)?;
}
- Some(Err(err)) => return Err(err),
+ None => break,
+ }
+ }
+
+ Ok(())
+ }
+
+ /// Delete an fallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`BlockingDeleter::delete_iter`]: delete an infallible iterator of paths.
+ pub fn delete_try_iter<I, D>(&mut self, try_iter: I) -> Result<()>
+ where
+ I: IntoIterator<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut iter = try_iter.into_iter();
+ loop {
+ match iter.next() {
+ Some(entry) => {
+ self.delete(entry?)?;
+ }
None => break,
}
}
diff --git a/core/src/types/delete/deleter.rs b/core/src/types/delete/deleter.rs
index 939aa3063..3b0078685 100644
--- a/core/src/types/delete/deleter.rs
+++ b/core/src/types/delete/deleter.rs
@@ -19,7 +19,6 @@ use crate::raw::oio::Delete;
use crate::raw::*;
use crate::*;
use futures::{Stream, StreamExt};
-use std::future::IntoFuture;
use std::pin::pin;
/// Deleter is designed to continuously remove content from storage.
@@ -115,34 +114,93 @@ impl Deleter {
Ok(())
}
- /// Delete an iterator of paths.
- pub async fn delete_iter<I, D, E>(&mut self, mut iter: I) -> Result<()>
+ /// Delete an infallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Deleter::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Deleter::delete_stream`]: delete an infallible stream of paths.
+ /// - [`Deleter::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_iter<I, D>(&mut self, iter: I) -> Result<()>
where
- I: IntoIterator,
+ I: IntoIterator<Item = D>,
D: IntoDeleteInput,
- I::Item: Into<Result<D, Error>>,
{
let iter = iter.into_iter();
for entry in iter {
- self.delete(entry?).await?;
+ self.delete(entry).await?;
}
Ok(())
}
- /// Delete a stream of paths.
- pub async fn delete_stream<S, D, E>(&mut self, mut stream: S) -> Result<()>
+ /// Delete an fallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Deleter::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Deleter::delete_stream`]: delete an infallible stream of paths.
+ /// - [`Deleter::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_try_iter<I, D>(&mut self, try_iter: I) -> Result<()>
where
- S: Stream,
+ I: IntoIterator<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut iter = try_iter.into_iter();
+ loop {
+ match iter.next() {
+ Some(entry) => {
+ self.delete(entry?).await?;
+ }
+ None => break,
+ }
+ }
+
+ Ok(())
+ }
+
+ /// Delete an infallible stream of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Deleter::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Deleter::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Deleter::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_stream<S, D>(&mut self, mut stream: S) -> Result<()>
+ where
+ S: Stream<Item = D>,
D: IntoDeleteInput,
- S::Item: Into<Result<D, Error>>,
{
let mut stream = pin!(stream);
loop {
- match stream.next().await.into() {
- Some(Ok(entry)) => {
+ match stream.next().await {
+ Some(entry) => {
+ self.delete(entry).await?;
+ }
+ None => break,
+ }
+ }
+
+ Ok(())
+ }
+
+ /// Delete an fallible stream of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Deleter::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Deleter::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Deleter::delete_stream`]: delete an infallible stream of paths.
+ pub async fn delete_try_stream<S, D>(&mut self, mut try_stream: S) -> Result<()>
+ where
+ S: Stream<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut stream = pin!(try_stream);
+ loop {
+ match stream.next().await.transpose()? {
+ Some(entry) => {
self.delete(entry).await?;
}
- Some(Err(err)) => return Err(err.into()),
None => break,
}
}
diff --git a/core/src/types/delete/futures_delete_sink.rs b/core/src/types/delete/futures_delete_sink.rs
index 863fd2deb..78a25dbed 100644
--- a/core/src/types/delete/futures_delete_sink.rs
+++ b/core/src/types/delete/futures_delete_sink.rs
@@ -112,6 +112,7 @@ impl<T: IntoDeleteInput> Sink<T> for FuturesDeleteSink {
State::Delete(fut) => {
let (deleter, res) = ready!(fut.as_mut().poll(cx));
self.state = State::Idle(Some(deleter));
+ res?;
continue;
}
State::Flush(fut) => {
diff --git a/core/src/types/delete/input.rs b/core/src/types/delete/input.rs
index 285b714d5..9c4b056fd 100644
--- a/core/src/types/delete/input.rs
+++ b/core/src/types/delete/input.rs
@@ -82,7 +82,7 @@ impl IntoDeleteInput for (String, OpDelete) {
/// Implement `IntoDeleteInput` for `Entry` so we can use `Lister` as a DeleteInput stream.
impl IntoDeleteInput for Entry {
fn into_delete_input(self) -> DeleteInput {
- let (path, mut meta) = self.into_parts();
+ let (path, meta) = self.into_parts();
let mut input = DeleteInput {
path,
diff --git a/core/src/types/error.rs b/core/src/types/error.rs
index 23cf75925..4d5514273 100644
--- a/core/src/types/error.rs
+++ b/core/src/types/error.rs
@@ -36,7 +36,6 @@
use std::backtrace::Backtrace;
use std::backtrace::BacktraceStatus;
-use std::convert::Infallible;
use std::fmt;
use std::fmt::Debug;
use std::fmt::Display;
@@ -403,13 +402,6 @@ impl Error {
}
}
-/// Impl From<Infallible> for Error so that users can use `Into<Error>` even when there is no error.
-impl From<Infallible> for Error {
- fn from(value: Infallible) -> Self {
- unreachable!("this error should never be created: {}", value)
- }
-}
-
impl From<Error> for io::Error {
fn from(err: Error) -> Self {
let kind = match err.kind() {
diff --git a/core/src/types/operator/blocking_operator.rs b/core/src/types/operator/blocking_operator.rs
index 167d91b64..7a1b6b699 100644
--- a/core/src/types/operator/blocking_operator.rs
+++ b/core/src/types/operator/blocking_operator.rs
@@ -775,12 +775,15 @@ impl BlockingOperator {
))
}
- /// Delete an iterator of paths.
- pub fn delete_iter<I, D, E>(&self, iter: I) -> Result<()>
+ /// Delete an infallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`BlockingOperator::delete_try_iter`]: delete an fallible iterator of paths.
+ pub fn delete_iter<I, D>(&self, iter: I) -> Result<()>
where
- I: IntoIterator<Item = Result<D, E>>,
+ I: IntoIterator<Item = D>,
D: IntoDeleteInput,
- E: Into<Error>,
{
let mut deleter = self.deleter()?;
deleter.delete_iter(iter)?;
@@ -788,6 +791,22 @@ impl BlockingOperator {
Ok(())
}
+ /// Delete an fallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`BlockingOperator::delete_iter`]: delete an infallible iterator of paths.
+ pub fn delete_try_iter<I, D>(&self, try_iter: I) -> Result<()>
+ where
+ I: IntoIterator<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut deleter = self.deleter()?;
+ deleter.delete_try_iter(try_iter)?;
+ deleter.close()?;
+ Ok(())
+ }
+
/// Create a [`BlockingDeleter`] to continuously remove content from storage.
///
/// It leverages batch deletion capabilities provided by storage services for efficient removal.
@@ -842,9 +861,7 @@ impl BlockingOperator {
/// ```
#[deprecated(note = "use `BlockingOperator::delete_iter` instead", since = "0.52")]
pub fn remove(&self, paths: Vec<String>) -> Result<()> {
- self.remove_via(paths.into_iter())?;
-
- Ok(())
+ self.delete_iter(paths.into_iter())
}
/// Remove the path and all nested dirs and files recursively.
@@ -881,7 +898,7 @@ impl BlockingOperator {
};
let lister = self.lister_with(path).recursive(true).call()?;
- self.delete_iter(lister)?;
+ self.delete_try_iter(lister)?;
Ok(())
}
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 06e1b757e..6724f9bf3 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -16,10 +16,8 @@
// under the License.
use std::future::Future;
-use std::pin::pin;
use std::time::Duration;
-use futures::stream;
use futures::Stream;
use futures::StreamExt;
use futures::TryStreamExt;
@@ -1512,12 +1510,17 @@ impl Operator {
)
}
- /// Delete an iterator of paths.
- pub async fn delete_iter<I, D, E>(&mut self, iter: I) -> Result<()>
+ /// Delete an infallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Operator::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Operator::delete_stream`]: delete an infallible stream of paths.
+ /// - [`Operator::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_iter<I, D>(&self, iter: I) -> Result<()>
where
- I: IntoIterator<Item = Result<D, E>>,
+ I: IntoIterator<Item = D>,
D: IntoDeleteInput,
- E: Into<Error>,
{
let mut deleter = self.deleter().await?;
deleter.delete_iter(iter).await?;
@@ -1525,12 +1528,35 @@ impl Operator {
Ok(())
}
- /// Delete a stream of paths.
- pub async fn delete_stream<S, D, E>(&self, stream: S) -> Result<()>
+ /// Delete an infallible iterator of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Operator::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Operator::delete_stream`]: delete an infallible stream of paths.
+ /// - [`Operator::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_try_iter<I, D>(&self, try_iter: I) -> Result<()>
+ where
+ I: IntoIterator<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut deleter = self.deleter().await?;
+ deleter.delete_try_iter(try_iter).await?;
+ deleter.close().await?;
+ Ok(())
+ }
+
+ /// Delete an infallible stream of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Operator::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Operator::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Operator::delete_try_stream`]: delete an fallible stream of paths.
+ pub async fn delete_stream<S, D>(&self, stream: S) -> Result<()>
where
- S: Stream<Item = Result<D, E>>,
+ S: Stream<Item = D>,
D: IntoDeleteInput,
- E: Into<Error>,
{
let mut deleter = self.deleter().await?;
deleter.delete_stream(stream).await?;
@@ -1538,6 +1564,24 @@ impl Operator {
Ok(())
}
+ /// Delete an infallible stream of paths.
+ ///
+ /// Also see:
+ ///
+ /// - [`Operator::delete_iter`]: delete an infallible iterator of paths.
+ /// - [`Operator::delete_try_iter`]: delete an fallible iterator of paths.
+ /// - [`Operator::delete_stream`]: delete an infallible stream of paths.
+ pub async fn delete_try_stream<S, D>(&self, try_stream: S) -> Result<()>
+ where
+ S: Stream<Item = Result<D>>,
+ D: IntoDeleteInput,
+ {
+ let mut deleter = self.deleter().await?;
+ deleter.delete_try_stream(try_stream).await?;
+ deleter.close().await?;
+ Ok(())
+ }
+
/// Create a [`Deleter`] to continuously remove content from storage.
///
/// It leverages batch deletion capabilities provided by storage services for efficient removal.
@@ -1568,9 +1612,7 @@ impl Operator {
#[deprecated(note = "use `Operator::delete_iter` instead", since = "0.52")]
pub async fn remove(&self, paths: Vec<String>) -> Result<()> {
let mut deleter = self.deleter().await?;
- deleter
- .delete_stream(&mut stream::iter(paths).map(|v| Ok(normalize_path(&v))))
- .await?;
+ deleter.delete_iter(paths).await?;
deleter.close().await?;
Ok(())
}
@@ -1649,9 +1691,8 @@ impl Operator {
Err(e) => return Err(e),
};
- let mut lister = self.lister_with(path).recursive(true).await?;
- self.delete_stream(lister).await?;
-
+ let lister = self.lister_with(path).recursive(true).await?;
+ self.delete_try_stream(lister).await?;
Ok(())
}
diff --git a/core/tests/behavior/blocking_delete.rs b/core/tests/behavior/blocking_delete.rs
index 79e0ef24d..7b66e520d 100644
--- a/core/tests/behavior/blocking_delete.rs
+++ b/core/tests/behavior/blocking_delete.rs
@@ -64,7 +64,7 @@ pub fn test_blocking_remove_one_file(op: BlockingOperator) -> Result<()> {
op.write(&path, content).expect("write must succeed");
- op.remove(vec![path.clone()])?;
+ op.delete_iter(vec![path.clone()])?;
// Stat it again to check.
assert!(!op.exists(&path)?);