Replace calls to Pin::new_unchecked with pin_project.

This is a breaking change, as it changes some public methods to take
`Pin<&mut Self>` rather than `&mut self`.

This brings these methods into line with `Stream::poll_next`, which also
takes a `Pin<&mut Self>`

Author: Aaron1011

actix-codec/Cargo.toml

@@ -23,4 +23,5 @@ futures-core = "0.3.1"
 futures-sink = "0.3.1"
 tokio = { version = "0.2.4", default-features=false }
 tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
-log = "0.4"
+log = "0.4"
+pin-project = "0.4.8"

actix-codec/src/framed.rs

@@ -5,6 +5,7 @@ use std::{fmt, io};
 use bytes::{Buf, BytesMut};
 use futures_core::{ready, Stream};
 use futures_sink::Sink;
+use pin_project::pin_project;
 
 use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
 
@@ -20,16 +21,16 @@ bitflags::bitflags! {
 
 /// A unified `Stream` and `Sink` interface to an underlying I/O object, using
 /// the `Encoder` and `Decoder` traits to encode and decode frames.
+#[pin_project]
 pub struct Framed<T, U> {
+    #[pin]
     io: T,
     codec: U,
     flags: Flags,
     read_buf: BytesMut,
     write_buf: BytesMut,
 }
 
-impl<T, U> Unpin for Framed<T, U> {}
-
 impl<T, U> Framed<T, U>
 where
     T: AsyncRead + AsyncWrite,
@@ -185,17 +186,18 @@ impl<T, U> Framed<T, U> {
 
 impl<T, U> Framed<T, U> {
     /// Serialize item and Write to the inner buffer
-    pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
+    pub fn write(mut self: Pin<&mut Self>, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        let remaining = self.write_buf.capacity() - self.write_buf.len();
+        let this = self.as_mut().project();
+        let remaining = this.write_buf.capacity() - this.write_buf.len();
         if remaining < LW {
-            self.write_buf.reserve(HW - remaining);
+            this.write_buf.reserve(HW - remaining);
         }
 
-        self.codec.encode(item, &mut self.write_buf)?;
+        this.codec.encode(item, this.write_buf)?;
         Ok(())
     }
 
@@ -207,21 +209,22 @@ impl<T, U> Framed<T, U> {
     }
 
     /// Try to read underlying I/O stream and decode item.
-    pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
+    pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
     where
         T: AsyncRead,
         U: Decoder,
     {
         loop {
+            let mut this = self.as_mut().project();
             // Repeatedly call `decode` or `decode_eof` as long as it is
             // "readable". Readable is defined as not having returned `None`. If
             // the upstream has returned EOF, and the decoder is no longer
             // readable, it can be assumed that the decoder will never become
             // readable again, at which point the stream is terminated.
 
-            if self.flags.contains(Flags::READABLE) {
-                if self.flags.contains(Flags::EOF) {
-                    match self.codec.decode_eof(&mut self.read_buf) {
+            if this.flags.contains(Flags::READABLE) {
+                if this.flags.contains(Flags::EOF) {
+                    match this.codec.decode_eof(&mut this.read_buf) {
                         Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
                         Ok(None) => return Poll::Ready(None),
                         Err(e) => return Poll::Ready(Some(Err(e))),
@@ -230,7 +233,7 @@ impl<T, U> Framed<T, U> {
 
                 log::trace!("attempting to decode a frame");
 
-                match self.codec.decode(&mut self.read_buf) {
+                match this.codec.decode(&mut this.read_buf) {
                     Ok(Some(frame)) => {
                         log::trace!("frame decoded from buffer");
                         return Poll::Ready(Some(Ok(frame)));
@@ -239,45 +242,44 @@ impl<T, U> Framed<T, U> {
                     _ => (), // Need more data
                 }
 
-                self.flags.remove(Flags::READABLE);
+                this.flags.remove(Flags::READABLE);
             }
 
-            debug_assert!(!self.flags.contains(Flags::EOF));
+            debug_assert!(!this.flags.contains(Flags::EOF));
 
             // Otherwise, try to read more data and try again. Make sure we've got room
-            let remaining = self.read_buf.capacity() - self.read_buf.len();
+            let remaining = this.read_buf.capacity() - this.read_buf.len();
             if remaining < LW {
-                self.read_buf.reserve(HW - remaining)
+                this.read_buf.reserve(HW - remaining)
             }
-            let cnt = match unsafe {
-                Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf)
-            } {
+            let cnt = match this.io.poll_read_buf(cx, &mut this.read_buf) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
                 Poll::Ready(Ok(cnt)) => cnt,
             };
 
             if cnt == 0 {
-                self.flags.insert(Flags::EOF);
+                this.flags.insert(Flags::EOF);
             }
-            self.flags.insert(Flags::READABLE);
+            this.flags.insert(Flags::READABLE);
         }
     }
 
     /// Flush write buffer to underlying I/O stream.
-    pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
+        let mut this = self.as_mut().project();
         log::trace!("flushing framed transport");
 
-        while !self.write_buf.is_empty() {
-            log::trace!("writing; remaining={}", self.write_buf.len());
+        while !this.write_buf.is_empty() {
+            log::trace!("writing; remaining={}", this.write_buf.len());
 
-            let n = ready!(unsafe {
-                Pin::new_unchecked(&mut self.io).poll_write(cx, &self.write_buf)
-            })?;
+            let n = ready!(
+                this.io.as_mut().poll_write(cx, this.write_buf)
+            )?;
 
             if n == 0 {
                 return Poll::Ready(Err(io::Error::new(
@@ -288,26 +290,25 @@ impl<T, U> Framed<T, U> {
             }
 
             // remove written data
-            self.write_buf.advance(n);
+            this.write_buf.advance(n);
         }
 
         // Try flushing the underlying IO
-        ready!(unsafe { Pin::new_unchecked(&mut self.io).poll_flush(cx) })?;
+        ready!(this.io.poll_flush(cx))?;
 
         log::trace!("framed transport flushed");
         Poll::Ready(Ok(()))
     }
 
     /// Flush write buffer and shutdown underlying I/O stream.
-    pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        unsafe {
-            ready!(Pin::new_unchecked(&mut self.io).poll_flush(cx))?;
-            ready!(Pin::new_unchecked(&mut self.io).poll_shutdown(cx))?;
-        }
+        let mut this = self.as_mut().project();
+        ready!(this.io.as_mut().poll_flush(cx))?;
+        ready!(this.io.as_mut().poll_shutdown(cx))?;
         Poll::Ready(Ok(()))
     }
 }
@@ -319,7 +320,7 @@ where
 {
     type Item = Result<U::Item, U::Error>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.next_item(cx)
     }
 }
@@ -341,21 +342,21 @@ where
     }
 
     fn start_send(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         item: <U as Encoder>::Item,
     ) -> Result<(), Self::Error> {
         self.write(item)
     }
 
     fn poll_flush(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.flush(cx)
     }
 
     fn poll_close(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.close(cx)

actix-ioframe/src/connect.rs

@@ -42,6 +42,7 @@ where
 pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
     pub(crate) state: St,
     pub(crate) out: Option<Out>,
+    #[pin]
     pub(crate) framed: Framed<Io, Codec>,
 }
 
@@ -97,8 +98,8 @@ where
 {
     type Error = <Codec as Encoder>::Error;
 
-    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.framed.is_write_ready() {
+    fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.as_mut().project().framed.is_write_ready() {
             Poll::Ready(Ok(()))
         } else {
             Poll::Pending
@@ -112,11 +113,11 @@ where
         self.project().framed.write(item)
     }
 
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.flush(cx)
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.close(cx)
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.close(cx)
     }
 }