feat(blocking operator): add remove_all api (#2449)

* add remove_all for blocking operator

Signed-off-by: clundro <[email protected]>

* update comment

Signed-off-by: clundro <[email protected]>

* add remove_one and remove_all in behavior test

Signed-off-by: clundro <[email protected]>

* fix clippy

Signed-off-by: clundro <[email protected]>

* update remove_all and remove

Signed-off-by: clundro <[email protected]>

* use intoIter in remove_via

Signed-off-by: clundro <[email protected]>

---------

Signed-off-by: clundro <[email protected]>

core/src/types/operator/blocking_operator.rs

@@ -680,6 +680,102 @@ impl BlockingOperator {
         Ok(())
     }
 
+    /// remove will remove files via the given paths.
+    ///
+    /// remove_via will remove files via the given vector iterators.
+    ///
+    /// # Notes
+    ///
+    /// We don't support batch delete now.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use anyhow::Result;
+    /// # use futures::io;
+    /// # use opendal::BlockingOperator;
+    /// # fn test(op: BlockingOperator) -> Result<()> {
+    /// let stream = vec!["abc".to_string(), "def".to_string()].into_iter();
+    /// op.remove_via(stream)?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn remove_via(&self, input: impl Iterator<Item = String>) -> Result<()> {
+        for path in input {
+            self.delete(&path)?;
+        }
+        Ok(())
+    }
+
+    /// # Notes
+    ///
+    /// We don't support batch delete now.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use anyhow::Result;
+    /// # use futures::io;
+    /// # use opendal::BlockingOperator;
+    /// # fn test(op: BlockingOperator) -> Result<()> {
+    /// op.remove(vec!["abc".to_string(), "def".to_string()])?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn remove(&self, paths: Vec<String>) -> Result<()> {
+        self.remove_via(paths.into_iter())?;
+
+        Ok(())
+    }
+
+    /// Remove the path and all nested dirs and files recursively.
+    ///
+    /// # Notes
+    ///
+    /// We don't support batch delete now.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use anyhow::Result;
+    /// # use futures::io;
+    /// # use opendal::BlockingOperator;
+    /// # fn test(op: BlockingOperator) -> Result<()> {
+    /// op.remove_all("path/to/dir")?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn remove_all(&self, path: &str) -> Result<()> {
+        let meta = match self.stat(path) {
+            Ok(metadata) => metadata,
+
+            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(()),
+
+            Err(e) => return Err(e),
+        };
+
+        if meta.mode() != EntryMode::DIR {
+            return self.delete(path);
+        }
+
+        let obs = self.scan(path)?;
+
+        for v in obs {
+            match v {
+                Ok(entry) => {
+                    self.inner()
+                        .blocking_delete(entry.path(), OpDelete::new())?;
+                }
+                Err(e) => return Err(e),
+            }
+        }
+
+        // Remove the directory itself.
+        self.delete(path)?;
+
+        Ok(())
+    }
+
     /// List current dir path.
     ///
     /// This function will create a new handle to list entries.

core/tests/behavior/blocking_list.rs

@@ -68,6 +68,7 @@ macro_rules! behavior_blocking_list_tests {
                 test_list_dir,
                 test_list_non_exist_dir,
                 test_scan,
+                test_remove_all,
             );
         )*
     };
@@ -152,3 +153,34 @@ pub fn test_scan(op: BlockingOperator) -> Result<()> {
     assert!(actual.contains("x/x/x/y"));
     Ok(())
 }
+
+// Remove all should remove all in this path.
+pub fn test_remove_all(op: BlockingOperator) -> Result<()> {
+    let parent = uuid::Uuid::new_v4().to_string();
+
+    let expected = vec![
+        "x/", "x/y", "x/x/", "x/x/y", "x/x/x/", "x/x/x/y", "x/x/x/x/",
+    ];
+
+    for path in expected.iter() {
+        if path.ends_with('/') {
+            op.create_dir(&format!("{parent}/{path}"))?;
+        } else {
+            op.write(&format!("{parent}/{path}"), "test_scan")?;
+        }
+    }
+
+    op.remove_all(&format!("{parent}/x/"))?;
+
+    for path in expected.iter() {
+        if path.ends_with('/') {
+            continue;
+        }
+        assert!(
+            !op.is_exist(&format!("{parent}/{path}"))?,
+            "{parent}/{path} should be removed"
+        )
+    }
+
+    Ok(())
+}

core/tests/behavior/blocking_write.rs

@@ -84,6 +84,7 @@ macro_rules! behavior_blocking_write_tests {
                 test_fuzz_offset_reader,
                 test_fuzz_part_reader,
                 test_delete_file,
+                test_remove_one_file,
             );
         )*
     };
@@ -413,3 +414,18 @@ pub fn test_delete_file(op: BlockingOperator) -> Result<()> {
 
     Ok(())
 }
+
+/// Remove one file
+pub fn test_remove_one_file(op: BlockingOperator) -> Result<()> {
+    let path = uuid::Uuid::new_v4().to_string();
+    let (content, _) = gen_bytes();
+
+    op.write(&path, content).expect("write must succeed");
+
+    op.remove(vec![path.clone()])?;
+
+    // Stat it again to check.
+    assert!(!op.is_exist(&path)?);
+
+    Ok(())
+}