Add the SinkExt::feed combinator (#2155)

Like send, except no flushing is done at the end.
This allows sequentially feeding the sink with items in async code
without intermittent flushing.

Reuse code of Feed internally in the Send future.

Author: mzabaluev

futures-util/src/sink/feed.rs

@@ -0,0 +1,49 @@
+use core::pin::Pin;
+use futures_core::future::Future;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_sink::Sink;
+
+/// Future for the [`feed`](super::SinkExt::feed) method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Feed<'a, Si: ?Sized, Item> {
+    sink: &'a mut Si,
+    item: Option<Item>,
+}
+
+// Pinning is never projected to children
+impl<Si: Unpin + ?Sized, Item> Unpin for Feed<'_, Si, Item> {}
+
+impl<'a, Si: Sink<Item> + Unpin + ?Sized, Item> Feed<'a, Si, Item> {
+    pub(super) fn new(sink: &'a mut Si, item: Item) -> Self {
+        Feed {
+            sink,
+            item: Some(item),
+        }
+    }
+
+    pub(super) fn sink_pin_mut(&mut self) -> Pin<&mut Si> {
+        Pin::new(self.sink)
+    }
+
+    pub(super) fn is_item_pending(&self) -> bool {
+        self.item.is_some()
+    }
+}
+
+impl<Si: Sink<Item> + Unpin + ?Sized, Item> Future for Feed<'_, Si, Item> {
+    type Output = Result<(), Si::Error>;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        let mut sink = Pin::new(&mut this.sink);
+        ready!(sink.as_mut().poll_ready(cx))?;
+        let item = this.item.take().expect("polled Feed after completion");
+        sink.as_mut().start_send(item)?;
+        Poll::Ready(Ok(()))
+    }
+}

futures-util/src/sink/mod.rs

@@ -26,6 +26,9 @@ pub use self::drain::{drain, Drain};
 mod fanout;
 pub use self::fanout::Fanout;
 
+mod feed;
+pub use self::feed::Feed;
+
 mod flush;
 pub use self::flush::Flush;
 
@@ -212,15 +215,27 @@ pub trait SinkExt<Item>: Sink<Item> {
     /// into the sink, including flushing.
     ///
     /// Note that, **because of the flushing requirement, it is usually better
-    /// to batch together items to send via `send_all`, rather than flushing
-    /// between each item.**
+    /// to batch together items to send via `feed` or `send_all`,
+    /// rather than flushing between each item.**
     fn send(&mut self, item: Item) -> Send<'_, Self, Item>
     where
         Self: Unpin,
     {
         Send::new(self, item)
     }
 
+    /// A future that completes after the given item has been received
+    /// by the sink.
+    ///
+    /// Unlike `send`, the returned future does not flush the sink.
+    /// It is the caller's responsibility to ensure all pending items
+    /// are processed, which can be done via `flush` or `close`.
+    fn feed(&mut self, item: Item) -> Feed<'_, Self, Item>
+        where Self: Unpin,
+    {
+        Feed::new(self, item)
+    }
+
     /// A future that completes after the given stream has been fully processed
     /// into the sink, including flushing.
     ///

futures-util/src/sink/send.rs

@@ -1,3 +1,4 @@
+use super::Feed;
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::ready;
@@ -8,8 +9,7 @@ use futures_sink::Sink;
 #[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
@@ -18,8 +18,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 {
         Self {
-            sink,
-            item: Some(item),
+            feed: Feed::new(sink, item),
         }
     }
 }
@@ -32,20 +31,15 @@ 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_item_pending() {
+            ready!(Pin::new(&mut this.feed).poll(cx))?;
+            debug_assert!(!this.feed.is_item_pending());
         }
 
         // 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(()))
     }

futures/src/lib.rs

@@ -417,7 +417,7 @@ pub mod sink {
     pub use futures_sink::Sink;
 
     pub use futures_util::sink::{
-        Close, Flush, Send, SendAll, SinkErrInto, SinkMapErr, With,
+        Close, Feed, Flush, Send, SendAll, SinkErrInto, SinkMapErr, With,
         SinkExt, Fanout, Drain, drain, Unfold, unfold,
         WithFlatMap,
     };