Merge pull request #747 from rust-lang-nursery/rename-poll-complete

Rename `Sink::poll_complete` to `flush`

Author: alexcrichton

futures-channel/benches/sync_mpsc.rs

@@ -103,7 +103,7 @@ impl Stream for TestSender {
             Err(_) => panic!(),
             Ok(Ok(())) => {
                 self.last += 1;
-                assert_eq!(Ok(Async::Ready(())), self.tx.poll_complete());
+                assert_eq!(Ok(Async::Ready(())), self.tx.flush());
                 Ok(Async::Ready(Some(self.last)))
             }
             Ok(Err(_)) => {

futures-channel/tests/mpsc.rs

@@ -34,7 +34,7 @@ fn send_recv_no_buffer() {
 
     // Run on a task context
     let f = lazy(move || {
-        assert!(tx.poll_complete().unwrap().is_ready());
+        assert!(tx.flush().unwrap().is_ready());
         assert!(tx.poll_ready().unwrap().is_ready());
 
         // Send first message

futures-sink/src/channel_impls.rs

@@ -16,7 +16,7 @@ impl<T> Sink for Sender<T> {
         self.start_send(msg).map(res_to_async_sink)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+    fn flush(&mut self) -> Poll<(), SendError<T>> {
         Ok(Async::Ready(()))
     }
 
@@ -33,7 +33,7 @@ impl<T> Sink for UnboundedSender<T> {
         self.start_send(msg).map(res_to_async_sink)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+    fn flush(&mut self) -> Poll<(), SendError<T>> {
         Ok(Async::Ready(()))
     }
 
@@ -51,7 +51,7 @@ impl<'a, T> Sink for &'a UnboundedSender<T> {
         Ok(AsyncSink::Ready)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+    fn flush(&mut self) -> Poll<(), SendError<T>> {
         Ok(Async::Ready(()))
     }
 

futures-sink/src/lib.rs

@@ -27,7 +27,7 @@ use futures_core::Poll;
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum AsyncSink<T> {
     /// The `start_send` attempt succeeded, so the sending process has
-    /// *started*; you must use `Sink::poll_complete` to drive the send
+    /// *started*; you must use `Sink::flush` to drive the send
     /// to completion.
     Ready,
 
@@ -83,7 +83,7 @@ if_std! {
             Ok(::AsyncSink::Ready)
         }
 
-        fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        fn flush(&mut self) -> Poll<(), Self::SinkError> {
             Ok(::Async::Ready(()))
         }
 
@@ -105,8 +105,8 @@ if_std! {
             (**self).start_send(item)
         }
 
-        fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-            (**self).poll_complete()
+        fn flush(&mut self) -> Poll<(), Self::SinkError> {
+            (**self).flush()
         }
 
         fn close(&mut self) -> Poll<(), Self::SinkError> {
@@ -160,21 +160,21 @@ pub trait Sink {
     /// until the buffer is fully flushed. Since sinks are designed to work with
     /// asynchronous I/O, the process of actually writing out the data to an
     /// underlying object takes place asynchronously. **You *must* use
-    /// `poll_complete` in order to drive completion of a send**. In particular,
+    /// `flush` in order to drive completion of a send**. In particular,
     /// `start_send` does not begin the flushing process
     ///
     /// # Return value
     ///
     /// This method returns `AsyncSink::Ready` if the sink was able to start
     /// sending `item`. In that case, you *must* ensure that you call
-    /// `poll_complete` to process the sent item to completion. Note, however,
+    /// `flush` to process the sent item to completion. Note, however,
     /// that several calls to `start_send` can be made prior to calling
-    /// `poll_complete`, which will work on completing all pending items.
+    /// `flush`, which will work on completing all pending items.
     ///
     /// The method returns `AsyncSink::Pending` if the sink was unable to begin
     /// sending, usually due to being full. The sink must have attempted to
     /// complete processing any outstanding requests (equivalent to
-    /// `poll_complete`) before yielding this result. The current task will be
+    /// `flush`) before yielding this result. The current task will be
     /// automatically scheduled for notification when the sink may be ready to
     /// receive new values.
     ///
@@ -190,7 +190,7 @@ pub trait Sink {
     /// sink:
     ///
     /// - It is called outside of the context of a task.
-    /// - A previous call to `start_send` or `poll_complete` yielded an error.
+    /// - A previous call to `start_send` or `flush` yielded an error.
     fn start_send(&mut self, item: Self::SinkItem)
                   -> StartSend<Self::SinkItem, Self::SinkError>;
 
@@ -230,7 +230,7 @@ pub trait Sink {
     /// This method may panic in a few situations, depending on the specific sink:
     ///
     /// - It is called outside of the context of a task.
-    /// - A previous call to `start_send` or `poll_complete` yielded an error.
+    /// - A previous call to `start_send` or `flush` yielded an error.
     ///
     /// # Compatibility nodes
     ///
@@ -240,34 +240,34 @@ pub trait Sink {
     /// this method to what it's always done, just flushing.
     ///
     /// In the 0.2 release series of futures this method will be renamed to
-    /// `poll_flush`. For 0.1, however, the breaking change is not happening
+    /// `flush`. For 0.1, however, the breaking change is not happening
     /// yet.
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError>;
+    fn flush(&mut self) -> Poll<(), Self::SinkError>;
 
     /// A method to indicate that no more values will ever be pushed into this
     /// sink.
     ///
     /// This method is used to indicate that a sink will no longer even be given
     /// another value by the caller. That is, the `start_send` method above will
-    /// be called no longer (nor `poll_complete`). This method is intended to
+    /// be called no longer (nor `flush`). This method is intended to
     /// model "graceful shutdown" in various protocols where the intent to shut
     /// down is followed by a little more blocking work.
     ///
     /// Callers of this function should work it it in a similar fashion to
-    /// `poll_complete`. Once called it may return `Pending` which indicates
+    /// `flush`. Once called it may return `Pending` which indicates
     /// that more external work needs to happen to make progress. The current
     /// task will be scheduled to receive a notification in such an event,
     /// however.
     ///
-    /// Note that this function will imply `poll_complete` above. That is, if a
+    /// Note that this function will imply `flush` above. That is, if a
     /// sink has buffered data, then it'll be flushed out during a `close`
-    /// operation. It is not necessary to have `poll_complete` return `Ready`
+    /// operation. It is not necessary to have `flush` return `Ready`
     /// before a `close` is called. Once a `close` is called, though,
-    /// `poll_complete` cannot be called.
+    /// `flush` cannot be called.
     ///
     /// # Return value
     ///
-    /// This function, like `poll_complete`, returns a `Poll`. The value is
+    /// This function, like `flush`, returns a `Poll`. The value is
     /// `Ready` once the close operation has completed. At that point it should
     /// be safe to drop the sink and deallocate associated resources.
     ///
@@ -286,7 +286,7 @@ pub trait Sink {
     /// always be true.
     ///
     /// Note that it's also typically an error to call `start_send` or
-    /// `poll_complete` after the `close` function is called. This method will
+    /// `flush` after the `close` function is called. This method will
     /// *initiate* a close, and continuing to send values after that (or attempt
     /// to flush) may result in strange behavior, panics, errors, etc. Once this
     /// method is called, it must be the only method called on this `Sink`.
@@ -296,12 +296,12 @@ pub trait Sink {
     /// This method may panic or cause panics if:
     ///
     /// * It is called outside the context of a future's task
-    /// * It is called and then `start_send` or `poll_complete` is called
+    /// * It is called and then `start_send` or `flush` is called
     ///
     /// # Compatibility notes
     ///
     /// Note that this function is currently by default a provided function,
-    /// defaulted to calling `poll_complete` above. This function was added
+    /// defaulted to calling `flush` above. This function was added
     /// in the 0.1 series of the crate as a backwards-compatible addition. It
     /// is intended that in the 0.2 series the method will no longer be a
     /// default method.
@@ -321,8 +321,8 @@ impl<'a, S: ?Sized + Sink> Sink for &'a mut S {
         (**self).start_send(item)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        (**self).poll_complete()
+    fn flush(&mut self) -> Poll<(), Self::SinkError> {
+        (**self).flush()
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {

futures-util/src/sink/buffer.rs

@@ -48,7 +48,7 @@ impl<S: Sink> Buffer<S> {
                 self.buf.push_front(item);
 
                 // ensure that we attempt to complete any pushes we've started
-                self.sink.poll_complete()?;
+                self.sink.flush()?;
 
                 return Ok(Async::Pending);
             }
@@ -85,14 +85,14 @@ impl<S: Sink> Sink for Buffer<S> {
         Ok(AsyncSink::Ready)
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+    fn flush(&mut self) -> Poll<(), Self::SinkError> {
         if self.cap == 0 {
-            return self.sink.poll_complete();
+            return self.sink.flush();
         }
 
         try_ready!(self.try_empty_buffer());
         debug_assert!(self.buf.is_empty());
-        self.sink.poll_complete()
+        self.sink.flush()
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {

futures-util/src/sink/close.rs

@@ -0,0 +1,52 @@
+use futures_core::{Poll, Async, Future};
+use futures_sink::Sink;
+
+/// Future for the `close` combinator, which polls the sink until all data has
+/// been closeed.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct Close<S> {
+    sink: Option<S>,
+}
+
+/// A future that completes when the sink has finished closing.
+///
+/// The sink itself is returned after closeing is complete.
+pub fn close<S: Sink>(sink: S) -> Close<S> {
+    Close { sink: Some(sink) }
+}
+
+impl<S: Sink> Close<S> {
+    /// Get a shared reference to the inner sink.
+    /// Returns `None` if the sink has already been closeed.
+    pub fn get_ref(&self) -> Option<&S> {
+        self.sink.as_ref()
+    }
+
+    /// Get a mutable reference to the inner sink.
+    /// Returns `None` if the sink has already been closeed.
+    pub fn get_mut(&mut self) -> Option<&mut S> {
+        self.sink.as_mut()
+    }
+
+    /// Consume the `Close` and return the inner sink.
+    /// Returns `None` if the sink has already been closeed.
+    pub fn into_inner(self) -> Option<S> {
+        self.sink
+    }
+}
+
+impl<S: Sink> Future for Close<S> {
+    type Item = S;
+    type Error = S::SinkError;
+
+    fn poll(&mut self) -> Poll<S, S::SinkError> {
+        let mut sink = self.sink.take().expect("Attempted to poll Close after it completed");
+        if sink.close()?.is_ready() {
+            Ok(Async::Ready(sink))
+        } else {
+            self.sink = Some(sink);
+            Ok(Async::Pending)
+        }
+    }
+}

futures-util/src/sink/fanout.rs

@@ -67,9 +67,9 @@ impl<A, B> Sink for Fanout<A, B>
         }
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        let left_async = self.left.poll_complete()?;
-        let right_async = self.right.poll_complete()?;
+    fn flush(&mut self) -> Poll<(), Self::SinkError> {
+        let left_async = self.left.flush()?;
+        let right_async = self.right.flush()?;
         // Only if both downstream sinks are ready, signal readiness.
         if left_async.is_ready() && right_async.is_ready() {
             Ok(Async::Ready(()))
@@ -112,9 +112,9 @@ impl<S: Sink> Downstream<S> {
         Ok(())
     }
 
-    fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
+    fn flush(&mut self) -> Poll<(), S::SinkError> {
         self.keep_flushing()?;
-        let async = self.sink.poll_complete()?;
+        let async = self.sink.flush()?;
         // Only if all values have been sent _and_ the underlying
         // sink is completely flushed, signal readiness.
         if self.state.is_ready() && async.is_ready() {

futures-util/src/sink/flush.rs

@@ -1,15 +1,21 @@
 use futures_core::{Poll, Async, Future};
 use futures_sink::Sink;
 
-/// Future for the `Sink::flush` combinator, which polls the sink until all data
+/// Future for the `flush` combinator, which polls the sink until all data
 /// has been flushed.
 #[derive(Debug)]
 #[must_use = "futures do nothing unless polled"]
 pub struct Flush<S> {
     sink: Option<S>,
 }
 
-pub fn new<S: Sink>(sink: S) -> Flush<S> {
+/// A future that completes when the sink has finished processing all
+/// pending requests.
+///
+/// The sink itself is returned after flushing is complete; this adapter is
+/// intended to be used when you want to stop sending to the sink until
+/// all current requests are processed.
+pub fn flush<S: Sink>(sink: S) -> Flush<S> {
     Flush { sink: Some(sink) }
 }
 
@@ -39,7 +45,7 @@ impl<S: Sink> Future for Flush<S> {
 
     fn poll(&mut self) -> Poll<S, S::SinkError> {
         let mut sink = self.sink.take().expect("Attempted to poll Flush after it completed");
-        if sink.poll_complete()?.is_ready() {
+        if sink.flush()?.is_ready() {
             Ok(Async::Ready(sink))
         } else {
             self.sink = Some(sink);

futures-util/src/sink/from_err.rs

@@ -53,8 +53,8 @@ impl<S, E> Sink for SinkFromErr<S, E>
         self.sink.start_send(item).map_err(|e| e.into())
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        self.sink.poll_complete().map_err(|e| e.into())
+    fn flush(&mut self) -> Poll<(), Self::SinkError> {
+        self.sink.flush().map_err(|e| e.into())
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {

futures-util/src/sink/map_err.rs

@@ -44,8 +44,8 @@ impl<S, F, E> Sink for SinkMapErr<S, F>
         self.sink.start_send(item).map_err(|e| self.f.take().expect("cannot use MapErr after an error")(e))
     }
 
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        self.sink.poll_complete().map_err(|e| self.f.take().expect("cannot use MapErr after an error")(e))
+    fn flush(&mut self) -> Poll<(), Self::SinkError> {
+        self.sink.flush().map_err(|e| self.f.take().expect("cannot use MapErr after an error")(e))
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {