Remove Pin API related unsafe code by using pin-project-lite cra…

Cargo.toml

@@ -43,6 +43,7 @@ pin-utils = "0.1.0-alpha.4"
 slab = "0.4.2"
 kv-log-macro = "1.0.4"
 broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] }
+pin-project-lite = "0.1"
 
 [dev-dependencies]
 femme = "1.2.0"

src/future/timeout.rs

@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::time::Duration;
 
 use futures_timer::Delay;
+use pin_project_lite::pin_project;
 
 use crate::future::Future;
 use crate::task::{Context, Poll};
@@ -39,24 +40,24 @@ where
     f.await
 }
 
-/// A future that times out after a duration of time.
-struct TimeoutFuture<F> {
-    future: F,
-    delay: Delay,
-}
-
-impl<F> TimeoutFuture<F> {
-    pin_utils::unsafe_pinned!(future: F);
-    pin_utils::unsafe_pinned!(delay: Delay);
+pin_project! {
+    /// A future that times out after a duration of time.
+    struct TimeoutFuture<F> {
+        #[pin]
+        future: F,
+        #[pin]
+        delay: Delay,
+    }
 }
 
 impl<F: Future> Future for TimeoutFuture<F> {
     type Output = Result<F::Output, TimeoutError>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match self.as_mut().future().poll(cx) {
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        match this.future.poll(cx) {
             Poll::Ready(v) => Poll::Ready(Ok(v)),
-            Poll::Pending => match self.delay().poll(cx) {
+            Poll::Pending => match this.delay.poll(cx) {
                 Poll::Ready(_) => Poll::Ready(Err(TimeoutError { _private: () })),
                 Poll::Pending => Poll::Pending,
             },

src/io/buf_read/lines.rs

@@ -2,50 +2,55 @@ use std::mem;
 use std::pin::Pin;
 use std::str;
 
+use pin_project_lite::pin_project;
+
 use super::read_until_internal;
 use crate::io::{self, BufRead};
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-/// A stream of lines in a byte stream.
-///
-/// This stream is created by the [`lines`] method on types that implement [`BufRead`].
-///
-/// This type is an async version of [`std::io::Lines`].
-///
-/// [`lines`]: trait.BufRead.html#method.lines
-/// [`BufRead`]: trait.BufRead.html
-/// [`std::io::Lines`]: https://doc.rust-lang.org/std/io/struct.Lines.html
-#[derive(Debug)]
-pub struct Lines<R> {
-    pub(crate) reader: R,
-    pub(crate) buf: String,
-    pub(crate) bytes: Vec<u8>,
-    pub(crate) read: usize,
+pin_project! {
+    /// A stream of lines in a byte stream.
+    ///
+    /// This stream is created by the [`lines`] method on types that implement [`BufRead`].
+    ///
+    /// This type is an async version of [`std::io::Lines`].
+    ///
+    /// [`lines`]: trait.BufRead.html#method.lines
+    /// [`BufRead`]: trait.BufRead.html
+    /// [`std::io::Lines`]: https://doc.rust-lang.org/std/io/struct.Lines.html
+    #[derive(Debug)]
+    pub struct Lines<R> {
+        #[pin]
+        pub(crate) reader: R,
+        pub(crate) buf: String,
+        pub(crate) bytes: Vec<u8>,
+        pub(crate) read: usize,
+    }
 }
 
 impl<R: BufRead> Stream for Lines<R> {
     type Item = io::Result<String>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let Self {
-            reader,
-            buf,
-            bytes,
-            read,
-        } = unsafe { self.get_unchecked_mut() };
-        let reader = unsafe { Pin::new_unchecked(reader) };
-        let n = futures_core::ready!(read_line_internal(reader, cx, buf, bytes, read))?;
-        if n == 0 && buf.is_empty() {
+        let this = self.project();
+        let n = futures_core::ready!(read_line_internal(
+            this.reader,
+            cx,
+            this.buf,
+            this.bytes,
+            this.read
+        ))?;
+        if n == 0 && this.buf.is_empty() {
             return Poll::Ready(None);
         }
-        if buf.ends_with('\n') {
-            buf.pop();
-            if buf.ends_with('\r') {
-                buf.pop();
+        if this.buf.ends_with('\n') {
+            this.buf.pop();
+            if this.buf.ends_with('\r') {
+                this.buf.pop();
             }
         }
-        Poll::Ready(Some(Ok(mem::replace(buf, String::new()))))
+        Poll::Ready(Some(Ok(mem::replace(this.buf, String::new()))))
     }
 }
 

src/io/buf_read/split.rs

@@ -1,46 +1,51 @@
 use std::mem;
 use std::pin::Pin;
 
+use pin_project_lite::pin_project;
+
 use super::read_until_internal;
 use crate::io::{self, BufRead};
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
-///
-/// This stream is created by the [`split`] method on types that implement [`BufRead`].
-///
-/// This type is an async version of [`std::io::Split`].
-///
-/// [`split`]: trait.BufRead.html#method.lines
-/// [`BufRead`]: trait.BufRead.html
-/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
-#[derive(Debug)]
-pub struct Split<R> {
-    pub(crate) reader: R,
-    pub(crate) buf: Vec<u8>,
-    pub(crate) read: usize,
-    pub(crate) delim: u8,
+pin_project! {
+    /// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
+    ///
+    /// This stream is created by the [`split`] method on types that implement [`BufRead`].
+    ///
+    /// This type is an async version of [`std::io::Split`].
+    ///
+    /// [`split`]: trait.BufRead.html#method.lines
+    /// [`BufRead`]: trait.BufRead.html
+    /// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
+    #[derive(Debug)]
+    pub struct Split<R> {
+        #[pin]
+        pub(crate) reader: R,
+        pub(crate) buf: Vec<u8>,
+        pub(crate) read: usize,
+        pub(crate) delim: u8,
+    }
 }
 
 impl<R: BufRead> Stream for Split<R> {
     type Item = io::Result<Vec<u8>>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let Self {
-            reader,
-            buf,
-            read,
-            delim,
-        } = unsafe { self.get_unchecked_mut() };
-        let reader = unsafe { Pin::new_unchecked(reader) };
-        let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?;
-        if n == 0 && buf.is_empty() {
+        let this = self.project();
+        let n = futures_core::ready!(read_until_internal(
+            this.reader,
+            cx,
+            *this.delim,
+            this.buf,
+            this.read
+        ))?;
+        if n == 0 && this.buf.is_empty() {
             return Poll::Ready(None);
         }
-        if buf[buf.len() - 1] == *delim {
-            buf.pop();
+        if this.buf[this.buf.len() - 1] == *this.delim {
+            this.buf.pop();
         }
-        Poll::Ready(Some(Ok(mem::replace(buf, vec![]))))
+        Poll::Ready(Some(Ok(mem::replace(this.buf, vec![]))))
     }
 }