Use (try_)poll_(next_)unpin more

Author: taiki-e

futures-util/src/async_await/poll.rs

@@ -1,3 +1,4 @@
+use crate::future::FutureExt;
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::task::{Context, Poll};
@@ -29,6 +30,6 @@ pub struct PollOnce<F: Future + Unpin> {
 impl<F: Future + Unpin> Future for PollOnce<F> {
     type Output = Poll<F::Output>;
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        Poll::Ready(Pin::new(&mut self.future).poll(cx))
+        Poll::Ready(self.future.poll_unpin(cx))
     }
 }

futures-util/src/future/select.rs

@@ -1,7 +1,7 @@
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::task::{Context, Poll};
-use crate::future::Either;
+use crate::future::{Either, FutureExt};
 
 /// Future for the [`select()`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
@@ -55,9 +55,9 @@ impl<A: Unpin, B: Unpin> Future for Select<A, B> where A: Future, B: Future {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let (mut a, mut b) = self.inner.take().expect("cannot poll Select twice");
-        match Pin::new(&mut a).poll(cx) {
+        match a.poll_unpin(cx) {
             Poll::Ready(x) => Poll::Ready(Either::Left((x, b))),
-            Poll::Pending => match Pin::new(&mut b).poll(cx) {
+            Poll::Pending => match b.poll_unpin(cx) {
                 Poll::Ready(x) => Poll::Ready(Either::Right((x, a))),
                 Poll::Pending => {
                     self.inner = Some((a, b));

futures-util/src/future/select_all.rs

@@ -1,3 +1,4 @@
+use crate::future::FutureExt;
 use core::iter::FromIterator;
 use core::mem;
 use core::pin::Pin;
@@ -42,7 +43,7 @@ impl<Fut: Future + Unpin> Future for SelectAll<Fut> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let item = self.inner.iter_mut().enumerate().find_map(|(i, f)| {
-            match Pin::new(f).poll(cx) {
+            match f.poll_unpin(cx) {
                 Poll::Pending => None,
                 Poll::Ready(e) => Some((i, e)),
             }

futures-util/src/stream/buffer_unordered.rs

@@ -1,4 +1,4 @@
-use crate::stream::{Fuse, FuturesUnordered};
+use crate::stream::{Fuse, FuturesUnordered, StreamExt};
 use futures_core::future::Future;
 use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
@@ -114,7 +114,7 @@ where
         }
 
         // Attempt to pull the next value from the in_progress_queue
-        match Pin::new(self.as_mut().in_progress_queue()).poll_next(cx) {
+        match self.as_mut().in_progress_queue().poll_next_unpin(cx) {
             x @ Poll::Pending | x @ Poll::Ready(Some(_)) => return x,
             Poll::Ready(None) => {}
         }

futures-util/src/stream/buffered.rs

@@ -1,4 +1,4 @@
-use crate::stream::{Fuse, FuturesOrdered};
+use crate::stream::{Fuse, FuturesOrdered, StreamExt};
 use futures_core::future::Future;
 use futures_core::stream::Stream;
 use futures_core::task::{Context, Poll};
@@ -109,7 +109,7 @@ where
         }
 
         // Attempt to pull the next value from the in_progress_queue
-        let res = Pin::new(self.as_mut().in_progress_queue()).poll_next(cx);
+        let res = self.as_mut().in_progress_queue().poll_next_unpin(cx);
         if let Some(val) = ready!(res) {
             return Poll::Ready(Some(val))
         }

futures-util/src/stream/into_future.rs

@@ -1,3 +1,4 @@
+use crate::stream::StreamExt;
 use core::pin::Pin;
 use futures_core::future::{FusedFuture, Future};
 use futures_core::stream::Stream;
@@ -85,7 +86,7 @@ impl<St: Stream + Unpin> Future for StreamFuture<St> {
     ) -> Poll<Self::Output> {
         let item = {
             let s = self.stream.as_mut().expect("polling StreamFuture twice");
-            ready!(Pin::new(s).poll_next(cx))
+            ready!(s.poll_next_unpin(cx))
         };
         let stream = self.stream.take().unwrap();
         Poll::Ready((item, stream))

futures-util/src/try_future/select_ok.rs

@@ -1,3 +1,4 @@
+use crate::try_future::TryFutureExt;
 use core::iter::FromIterator;
 use core::mem;
 use core::pin::Pin;
@@ -45,7 +46,7 @@ impl<Fut: TryFuture + Unpin> Future for SelectOk<Fut> {
         // loop until we've either exhausted all errors, a success was hit, or nothing is ready
         loop {
             let item = self.inner.iter_mut().enumerate().find_map(|(i, f)| {
-                match Pin::new(f).try_poll(cx) {
+                match f.try_poll_unpin(cx) {
                     Poll::Pending => None,
                     Poll::Ready(e) => Some((i, e)),
                 }

futures-util/src/try_future/try_select.rs

@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::{Future, TryFuture};
 use futures_core::task::{Context, Poll};
 use crate::future::Either;
+use crate::try_future::TryFutureExt;
 
 /// Future for the [`try_select()`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
@@ -64,10 +65,10 @@ impl<A: Unpin, B: Unpin> Future for TrySelect<A, B>
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let (mut a, mut b) = self.inner.take().expect("cannot poll Select twice");
-        match Pin::new(&mut a).try_poll(cx) {
+        match a.try_poll_unpin(cx) {
             Poll::Ready(Err(x)) => Poll::Ready(Err(Either::Left((x, b)))),
             Poll::Ready(Ok(x)) => Poll::Ready(Ok(Either::Left((x, b)))),
-            Poll::Pending => match Pin::new(&mut b).try_poll(cx) {
+            Poll::Pending => match b.try_poll_unpin(cx) {
                 Poll::Ready(Err(x)) => Poll::Ready(Err(Either::Right((x, a)))),
                 Poll::Ready(Ok(x)) => Poll::Ready(Ok(Either::Right((x, a)))),
                 Poll::Pending => {

futures-util/src/try_stream/into_async_read.rs

@@ -1,3 +1,4 @@
+use crate::try_stream::TryStreamExt;
 use core::pin::Pin;
 use futures_core::stream::TryStream;
 use futures_core::task::{Context, Poll};
@@ -72,7 +73,7 @@ where
                     return Poll::Ready(Ok(len));
                 }
                 ReadState::PendingChunk => {
-                    match ready!(Pin::new(&mut self.stream).try_poll_next(cx)) {
+                    match ready!(self.stream.try_poll_next_unpin(cx)) {
                         Some(Ok(chunk)) => {
                             if !chunk.as_ref().is_empty() {
                                 self.state = ReadState::Ready {
@@ -109,7 +110,7 @@ where
         cx: &mut Context<'_>,
     ) -> Poll<Result<&'a [u8]>> {
         while let ReadState::PendingChunk = self.state {
-            match ready!(Pin::new(&mut self.stream).try_poll_next(cx)) {
+            match ready!(self.stream.try_poll_next_unpin(cx)) {
                 Some(Ok(chunk)) => {
                     if !chunk.as_ref().is_empty() {
                         self.state = ReadState::Ready {

futures-util/src/try_stream/try_buffer_unordered.rs

@@ -92,7 +92,7 @@ impl<St> Stream for TryBufferUnordered<St>
         }
 
         // Attempt to pull the next value from the in_progress_queue
-        match Pin::new(self.as_mut().in_progress_queue()).poll_next(cx) {
+        match self.as_mut().in_progress_queue().poll_next_unpin(cx) {
             x @ Poll::Pending | x @ Poll::Ready(Some(_)) => return x,
             Poll::Ready(None) => {}
         }

futures/tests/io_buf_reader.rs

@@ -1,5 +1,5 @@
 use futures::executor::block_on;
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures::io::{
     AsyncSeek, AsyncSeekExt, AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt,
     AllowStdIo, BufReader, SeekFrom,
@@ -259,7 +259,7 @@ impl AsyncBufRead for MaybePending<'_> {
 fn run<F: Future + Unpin>(mut f: F) -> F::Output {
     let mut cx = noop_context();
     loop {
-        if let Poll::Ready(x) = Pin::new(&mut f).poll(&mut cx) {
+        if let Poll::Ready(x) = f.poll_unpin(&mut cx) {
             return x;
         }
     }

futures/tests/io_buf_writer.rs

@@ -1,5 +1,5 @@
 use futures::executor::block_on;
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures::io::{AsyncSeek, AsyncSeekExt, AsyncWrite, AsyncWriteExt, BufWriter, SeekFrom};
 use futures::task::{Context, Poll};
 use futures_test::task::noop_context;
@@ -111,7 +111,7 @@ impl AsyncWrite for MaybePending {
 fn run<F: Future + Unpin>(mut f: F) -> F::Output {
     let mut cx = noop_context();
     loop {
-        if let Poll::Ready(x) = Pin::new(&mut f).poll(&mut cx) {
+        if let Poll::Ready(x) = f.poll_unpin(&mut cx) {
             return x;
         }
     }

futures/tests/io_lines.rs

@@ -1,12 +1,11 @@
 use futures::executor::block_on;
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures::stream::{self, StreamExt, TryStreamExt};
 use futures::io::AsyncBufReadExt;
 use futures::task::Poll;
 use futures_test::io::AsyncReadTestExt;
 use futures_test::task::noop_context;
 use std::io::Cursor;
-use std::pin::Pin;
 
 macro_rules! block_on_next {
     ($expr:expr) => {
@@ -31,7 +30,7 @@ fn lines() {
 fn run<F: Future + Unpin>(mut f: F) -> F::Output {
     let mut cx = noop_context();
     loop {
-        if let Poll::Ready(x) = Pin::new(&mut f).poll(&mut cx) {
+        if let Poll::Ready(x) = f.poll_unpin(&mut cx) {
             return x;
         }
     }

futures/tests/io_read_line.rs

@@ -1,12 +1,11 @@
 use futures::executor::block_on;
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures::stream::{self, StreamExt, TryStreamExt};
 use futures::io::AsyncBufReadExt;
 use futures::task::Poll;
 use futures_test::io::AsyncReadTestExt;
 use futures_test::task::noop_context;
 use std::io::Cursor;
-use std::pin::Pin;
 
 #[test]
 fn read_line() {
@@ -30,7 +29,7 @@ fn read_line() {
 fn run<F: Future + Unpin>(mut f: F) -> F::Output {
     let mut cx = noop_context();
     loop {
-        if let Poll::Ready(x) = Pin::new(&mut f).poll(&mut cx) {
+        if let Poll::Ready(x) = f.poll_unpin(&mut cx) {
             return x;
         }
     }

futures/tests/io_read_until.rs

@@ -1,12 +1,11 @@
 use futures::executor::block_on;
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures::stream::{self, StreamExt, TryStreamExt};
 use futures::io::AsyncBufReadExt;
 use futures::task::Poll;
 use futures_test::io::AsyncReadTestExt;
 use futures_test::task::noop_context;
 use std::io::Cursor;
-use std::pin::Pin;
 
 #[test]
 fn read_until() {
@@ -30,7 +29,7 @@ fn read_until() {
 fn run<F: Future + Unpin>(mut f: F) -> F::Output {
     let mut cx = noop_context();
     loop {
-        if let Poll::Ready(x) = Pin::new(&mut f).poll(&mut cx) {
+        if let Poll::Ready(x) = f.poll_unpin(&mut cx) {
             return x;
         }
     }