codec: Remove Unpin requirement from Framed[Read,Write,] (#1758)

cc #1252

tokio-util/src/codec/framed_write.rs

@@ -1,6 +1,6 @@
 use crate::codec::decoder::Decoder;
 use crate::codec::encoder::Encoder;
-use crate::codec::framed::Fuse;
+use crate::codec::framed::{Fuse, ProjectFuse};
 
 use tokio::io::{AsyncBufRead, AsyncRead, AsyncWrite};
 
@@ -8,17 +8,22 @@ use bytes::BytesMut;
 use futures_core::{ready, Stream};
 use futures_sink::Sink;
 use log::trace;
+use pin_project::pin_project;
 use std::fmt;
 use std::io::{self, BufRead, Read};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 /// A `Sink` of frames encoded to an `AsyncWrite`.
+#[pin_project]
 pub struct FramedWrite<T, E> {
+    #[pin]
     inner: FramedWrite2<Fuse<T, E>>,
 }
 
+#[pin_project]
 pub(crate) struct FramedWrite2<T> {
+    #[pin]
     inner: T,
     buffer: BytesMut,
 }
@@ -34,7 +39,10 @@ where
     /// Creates a new `FramedWrite` with the given `encoder`.
     pub fn new(inner: T, encoder: E) -> FramedWrite<T, E> {
         FramedWrite {
-            inner: framed_write2(Fuse(inner, encoder)),
+            inner: framed_write2(Fuse {
+                io: inner,
+                codec: encoder,
+            }),
         }
     }
 }
@@ -47,7 +55,7 @@ impl<T, E> FramedWrite<T, E> {
     /// of data coming in as it may corrupt the stream of frames otherwise
     /// being worked with.
     pub fn get_ref(&self) -> &T {
-        &self.inner.inner.0
+        &self.inner.inner.io
     }
 
     /// Returns a mutable reference to the underlying I/O stream wrapped by
@@ -57,7 +65,7 @@ impl<T, E> FramedWrite<T, E> {
     /// of data coming in as it may corrupt the stream of frames otherwise
     /// being worked with.
     pub fn get_mut(&mut self) -> &mut T {
-        &mut self.inner.inner.0
+        &mut self.inner.inner.io
     }
 
     /// Consumes the `FramedWrite`, returning its underlying I/O stream.
@@ -66,55 +74,60 @@ impl<T, E> FramedWrite<T, E> {
     /// of data coming in as it may corrupt the stream of frames otherwise
     /// being worked with.
     pub fn into_inner(self) -> T {
-        self.inner.inner.0
+        self.inner.inner.io
     }
 
     /// Returns a reference to the underlying decoder.
     pub fn encoder(&self) -> &E {
-        &self.inner.inner.1
+        &self.inner.inner.codec
     }
 
     /// Returns a mutable reference to the underlying decoder.
     pub fn encoder_mut(&mut self) -> &mut E {
-        &mut self.inner.inner.1
+        &mut self.inner.inner.codec
     }
 }
 
 // This impl just defers to the underlying FramedWrite2
 impl<T, I, E> Sink<I> for FramedWrite<T, E>
 where
-    T: AsyncWrite + Unpin,
-    E: Encoder<Item = I> + Unpin,
+    T: AsyncWrite,
+    E: Encoder<Item = I>,
     E::Error: From<io::Error>,
 {
     type Error = E::Error;
 
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        pin!(Pin::get_mut(self).inner).poll_ready(cx)
+        self.project().inner.poll_ready(cx)
     }
 
     fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
-        pin!(Pin::get_mut(self).inner).start_send(item)
+        self.project().inner.start_send(item)
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        pin!(Pin::get_mut(self).inner).poll_flush(cx)
+        self.project().inner.poll_flush(cx)
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        pin!(Pin::get_mut(self).inner).poll_close(cx)
+        self.project().inner.poll_close(cx)
     }
 }
 
 impl<T, D> Stream for FramedWrite<T, D>
 where
-    T: Stream + Unpin,
-    D: Unpin,
+    T: Stream,
 {
     type Item = T::Item;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Pin::new(Pin::get_mut(self).get_mut()).poll_next(cx)
+        self.project()
+            .inner
+            .project()
+            .inner
+            .project()
+            .io
+            .poll_next(cx)
     }
 }
 
@@ -125,8 +138,8 @@ where
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("FramedWrite")
-            .field("inner", &self.inner.get_ref().0)
-            .field("encoder", &self.inner.get_ref().1)
+            .field("inner", &self.inner.get_ref().io)
+            .field("encoder", &self.inner.get_ref().codec)
             .field("buffer", &self.inner.buffer)
             .finish()
     }
@@ -169,9 +182,10 @@ impl<T> FramedWrite2<T> {
 
 impl<I, T> Sink<I> for FramedWrite2<T>
 where
-    T: AsyncWrite + Encoder<Item = I> + Unpin,
+    T: ProjectFuse + AsyncWrite,
+    T::Codec: Encoder<Item = I>,
 {
-    type Error = T::Error;
+    type Error = <T::Codec as Encoder>::Error;
 
     fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         // If the buffer is already over 8KiB, then attempt to flush it. If after flushing it's
@@ -191,20 +205,24 @@ where
     }
 
     fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
-        let pinned = Pin::get_mut(self);
-        pinned.inner.encode(item, &mut pinned.buffer)?;
+        let mut pinned = self.project();
+        pinned
+            .inner
+            .project()
+            .codec
+            .encode(item, &mut pinned.buffer)?;
         Ok(())
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         trace!("flushing framed transport");
-        let pinned = Pin::get_mut(self);
+        let mut pinned = self.project();
 
         while !pinned.buffer.is_empty() {
             trace!("writing; remaining={}", pinned.buffer.len());
 
             let buf = &pinned.buffer;
-            let n = ready!(pin!(pinned.inner).poll_write(cx, &buf))?;
+            let n = ready!(pinned.inner.as_mut().poll_write(cx, &buf))?;
 
             if n == 0 {
                 return Poll::Ready(Err(io::Error::new(
@@ -220,15 +238,15 @@ where
         }
 
         // Try flushing the underlying IO
-        ready!(pin!(pinned.inner).poll_flush(cx))?;
+        ready!(pinned.inner.poll_flush(cx))?;
 
         trace!("framed transport flushed");
         Poll::Ready(Ok(()))
     }
 
     fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        ready!(pin!(self).poll_flush(cx))?;
-        ready!(pin!(self.inner).poll_shutdown(cx))?;
+        ready!(self.as_mut().poll_flush(cx))?;
+        ready!(self.project().inner.poll_shutdown(cx))?;
 
         Poll::Ready(Ok(()))
     }
@@ -263,7 +281,7 @@ impl<T: BufRead> BufRead for FramedWrite2<T> {
     }
 }
 
-impl<T: AsyncRead + Unpin> AsyncRead for FramedWrite2<T> {
+impl<T: AsyncRead> AsyncRead for FramedWrite2<T> {
     unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
         self.inner.prepare_uninitialized_buffer(buf)
     }
@@ -273,16 +291,28 @@ impl<T: AsyncRead + Unpin> AsyncRead for FramedWrite2<T> {
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<Result<usize, io::Error>> {
-        pin!(self.get_mut().inner).poll_read(cx, buf)
+        self.project().inner.poll_read(cx, buf)
     }
 }
 
-impl<T: AsyncBufRead + Unpin> AsyncBufRead for FramedWrite2<T> {
+impl<T: AsyncBufRead> AsyncBufRead for FramedWrite2<T> {
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
-        pin!(self.get_mut().inner).poll_fill_buf(cx)
+        self.project().inner.poll_fill_buf(cx)
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        pin!(self.get_mut().inner).consume(amt)
+        self.project().inner.consume(amt)
+    }
+}
+
+impl<T> ProjectFuse for FramedWrite2<T>
+where
+    T: ProjectFuse,
+{
+    type Io = T::Io;
+    type Codec = T::Codec;
+
+    fn project(self: Pin<&mut Self>) -> Fuse<Pin<&mut Self::Io>, &mut Self::Codec> {
+        self.project().inner.project()
     }
 }