Reuse code of Feed in Send

futures-util/src/sink/feed.rs

@@ -21,6 +21,10 @@ impl<'a, Si: Sink<Item> + Unpin + ?Sized, Item> Feed<'a, Si, Item> {
             item: Some(item),
         }
     }
+
+    pub(super) fn sink_pin_mut(&mut self) -> Pin<&mut Si> {
+        Pin::new(self.sink)
+    }
 }
 
 impl<Si: Sink<Item> + Unpin + ?Sized, Item> Future for Feed<'_, Si, Item> {

futures-util/src/sink/send.rs

@@ -1,14 +1,14 @@
+use super::Feed;
 use core::pin::Pin;
-use futures_core::future::Future;
+use futures_core::future::{Future, FusedFuture};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 
 /// Future for the [`send`](super::SinkExt::send) method.
 #[derive(Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct Send<'a, Si: ?Sized, Item> {
-    sink: &'a mut Si,
-    item: Option<Item>,
+    feed: Feed<'a, Si, Item>,
 }
 
 // Pinning is never projected to children
@@ -17,8 +17,7 @@ impl<Si: Unpin + ?Sized, Item> Unpin for Send<'_, Si, Item> {}
 impl<'a, Si: Sink<Item> + Unpin + ?Sized, Item> Send<'a, Si, Item> {
     pub(super) fn new(sink: &'a mut Si, item: Item) -> Self {
         Send {
-            sink,
-            item: Some(item),
+            feed: Feed::new(sink, item),
         }
     }
 }
@@ -31,20 +30,13 @@ impl<Si: Sink<Item> + Unpin + ?Sized, Item> Future for Send<'_, Si, Item> {
         cx: &mut Context<'_>,
     ) -> Poll<Self::Output> {
         let this = &mut *self;
-        if let Some(item) = this.item.take() {
-            let mut sink = Pin::new(&mut this.sink);
-            match sink.as_mut().poll_ready(cx)? {
-                Poll::Ready(()) => sink.as_mut().start_send(item)?,
-                Poll::Pending => {
-                    this.item = Some(item);
-                    return Poll::Pending;
-                }
-            }
+        if !this.feed.is_terminated() {
+            ready!(Pin::new(&mut this.feed).poll(cx))?;
         }
 
         // we're done sending the item, but want to block on flushing the
         // sink
-        ready!(Pin::new(&mut this.sink).poll_flush(cx))?;
+        ready!(this.feed.sink_pin_mut().poll_flush(cx))?;
 
         Poll::Ready(Ok(()))
     }