Update Forward docs, remove Unpin sink requirement

Author: ebkalderon

futures-util/src/stream/forward.rs

@@ -9,16 +9,10 @@ use pin_utils::{unsafe_pinned, unsafe_unpinned};
 const INVALID_POLL: &str = "polled `Forward` after completion";
 
 /// Future for the `Stream::forward` combinator, which sends a stream of values
-/// to a sink and then flushes the sink.
-///
-/// Note: this is only usable with `Unpin` sinks, so `Sink`s that aren't `Unpin`
-/// will need to be pinned in order to be used with this combinator.
-//
-// This limitation is necessary in order to return the sink after the forwarding
-// has completed so that it can be used again.
+/// to a sink and then flushes and closes the sink.
 #[derive(Debug)]
 #[must_use = "steams do nothing unless polled"]
-pub struct Forward<St: Stream, Si: Sink + Unpin> {
+pub struct Forward<St: Stream, Si: Sink> {
     sink: Option<Si>,
     stream: Fuse<St>,
     buffered_item: Option<Si::SinkItem>,
@@ -28,7 +22,7 @@ impl<St: Stream + Unpin, Si: Sink + Unpin> Unpin for Forward<St, Si> {}
 
 impl<St, Si> Forward<St, Si>
 where
-    Si: Sink + Unpin,
+    Si: Sink,
     St: Stream<Item = Result<Si::SinkItem, Si::SinkError>>,
 {
     unsafe_pinned!(sink: Option<Si>);
@@ -68,7 +62,7 @@ impl<St: Stream, Si: Sink + Unpin> FusedFuture for Forward<St, Si> {
 
 impl<St, Si> Future for Forward<St, Si>
 where
-    Si: Sink + Unpin,
+    Si: Sink,
     St: Stream<Item = Result<Si::SinkItem, Si::SinkError>>,
 {
     type Output = Result<(), Si::SinkError>;
@@ -91,7 +85,7 @@ where
                 Poll::Ready(None) => {
                     try_ready!(self.as_mut().sink().as_pin_mut().expect(INVALID_POLL)
                                    .poll_close(waker));
-                    let _ = self.as_mut().sink().take().unwrap();
+                    let _ = self.as_mut().sink().as_pin_mut().take().unwrap();
                     return Poll::Ready(Ok(()))
                 }
                 Poll::Pending => {