Polish Sink

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

core/src/types/delete/deleter.rs

@@ -67,12 +67,12 @@ use std::pin::pin;
 /// ```rust
 /// use opendal::Operator;
 /// use opendal::Result;
-/// use futures::stream;
+/// use futures::{stream, Sink};
 /// use futures::SinkExt;
 ///
 /// async fn example(op: Operator) -> Result<()> {
 ///     let mut sink = op.deleter().await?.into_sink();
-///     sink.send_all(&mut stream::iter(vec!["path/to/file"])).await?;
+///     sink.send("path/to/file").await?;
 ///     sink.close().await?;
 ///     Ok(())
 /// }
@@ -210,7 +210,7 @@ impl Deleter {
     }
 
     /// Convert the deleter into a sink.
-    pub fn into_sink(self) -> FuturesDeleteSink {
+    pub fn into_sink<T: IntoDeleteInput>(self) -> FuturesDeleteSink<T> {
         FuturesDeleteSink::new(self)
     }
 }

core/src/types/delete/futures_delete_sink.rs

@@ -18,12 +18,14 @@
 use crate::raw::*;
 use crate::*;
 use futures::Sink;
+use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};
 
 /// FuturesDeleteSink is a sink that generated by [`Deleter`]
-pub struct FuturesDeleteSink {
+pub struct FuturesDeleteSink<T: IntoDeleteInput> {
     state: State,
+    _phantom: PhantomData<T>,
 }
 
 enum State {
@@ -33,16 +35,17 @@ enum State {
     Close(BoxedStaticFuture<(Deleter, Result<()>)>),
 }
 
-impl FuturesDeleteSink {
+impl<T: IntoDeleteInput> FuturesDeleteSink<T> {
     #[inline]
     pub(super) fn new(deleter: Deleter) -> Self {
         Self {
             state: State::Idle(Some(deleter)),
+            _phantom: PhantomData,
         }
     }
 }
 
-impl<T: IntoDeleteInput> Sink<T> for FuturesDeleteSink {
+impl<T: IntoDeleteInput> Sink<T> for FuturesDeleteSink<T> {
     type Error = Error;
 
     fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {

core/src/types/delete/input.rs

@@ -29,7 +29,7 @@ pub struct DeleteInput {
 }
 
 /// IntoDeleteInput is a helper trait that makes it easier for users to play with `Deleter`.
-pub trait IntoDeleteInput: Send + Sync {
+pub trait IntoDeleteInput: Send + Sync + Unpin {
     /// Convert `self` into a `DeleteInput`.
     fn into_delete_input(self) -> DeleteInput;
 }