feat(core): impl Drop for BlockingWrapper (#6036)

* feat(blocking): implement Drop trait for BlockingWrapper to ensure proper resource cleanup

* remove a drop

* rm white

* add blocking test

* fix(blocking): restrict Drop implementation for BlockingWrapper to specific trait bounds

* try drop in async

* Option<I>

* fix(blocking): change Drop implementation to use block_on for inner cleanup

* fix(blocking): relax trait bounds for Drop implementation in BlockingWrapper

* test(blocking): add test_check to the blocking trials

Author: asukaminato0721

core/src/layers/blocking.rs

@@ -267,44 +267,57 @@ impl<A: Access> LayeredAccess for BlockingAccessor<A> {
 
 pub struct BlockingWrapper<I> {
     handle: Handle,
-    inner: I,
+    inner: Option<I>,
 }
 
 impl<I> BlockingWrapper<I> {
     fn new(handle: Handle, inner: I) -> Self {
-        Self { handle, inner }
+        Self {
+            handle,
+            inner: Some(inner),
+        }
     }
 }
 
 impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
     fn read(&mut self) -> Result<Buffer> {
-        self.handle.block_on(self.inner.read())
+        self.handle.block_on(self.inner.as_mut().unwrap().read())
     }
 }
 
 impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
     fn write(&mut self, bs: Buffer) -> Result<()> {
-        self.handle.block_on(self.inner.write(bs))
+        self.handle.block_on(self.inner.as_mut().unwrap().write(bs))
     }
 
     fn close(&mut self) -> Result<Metadata> {
-        self.handle.block_on(self.inner.close())
+        self.handle.block_on(self.inner.as_mut().unwrap().close())
+    }
+}
+
+impl<I> Drop for BlockingWrapper<I> {
+    fn drop(&mut self) {
+        if let Some(inner) = self.inner.take() {
+            self.handle.block_on(async move {
+                drop(inner);
+            });
+        }
     }
 }
 
 impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
     fn next(&mut self) -> Result<Option<oio::Entry>> {
-        self.handle.block_on(self.inner.next())
+        self.handle.block_on(self.inner.as_mut().unwrap().next())
     }
 }
 
 impl<I: oio::Delete + 'static> oio::BlockingDelete for BlockingWrapper<I> {
     fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
-        self.inner.delete(path, args)
+        self.inner.as_mut().unwrap().delete(path, args)
     }
 
     fn flush(&mut self) -> Result<usize> {
-        self.handle.block_on(self.inner.flush())
+        self.handle.block_on(self.inner.as_mut().unwrap().flush())
     }
 }
 

core/tests/behavior/blocking_list.rs

@@ -28,6 +28,7 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
     if cap.read && cap.write && cap.copy && cap.blocking && cap.list {
         tests.extend(blocking_trials!(
             op,
+            test_check,
             test_blocking_list_dir,
             test_blocking_list_non_exist_dir,
             test_blocking_list_not_exist_dir_with_recursive,
@@ -39,6 +40,13 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
     }
 }
 
+/// Check should be OK.
+pub fn test_check(op: BlockingOperator) -> Result<()> {
+    op.check().expect("operator check is ok");
+
+    Ok(())
+}
+
 /// List dir should return newly created file.
 pub fn test_blocking_list_dir(op: BlockingOperator) -> Result<()> {
     let parent = uuid::Uuid::new_v4().to_string();