refactor: reduce dependency on futures-util

Author: paolobarbolini

src/client/legacy/connect/http.rs

@@ -4,7 +4,7 @@ use std::future::Future;
 use std::io;
 use std::marker::PhantomData;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
-use std::pin::Pin;
+use std::pin::{pin, Pin};
 use std::sync::Arc;
 use std::task::{self, Poll};
 use std::time::Duration;
@@ -970,14 +970,11 @@ impl ConnectingTcp<'_> {
         match self.fallback {
             None => self.preferred.connect(self.config).await,
             Some(mut fallback) => {
-                let preferred_fut = self.preferred.connect(self.config);
-                futures_util::pin_mut!(preferred_fut);
+                let preferred_fut = pin!(self.preferred.connect(self.config));
 
-                let fallback_fut = fallback.remote.connect(self.config);
-                futures_util::pin_mut!(fallback_fut);
+                let fallback_fut = pin!(fallback.remote.connect(self.config));
 
-                let fallback_delay = fallback.delay;
-                futures_util::pin_mut!(fallback_delay);
+                let fallback_delay = pin!(fallback.delay);
 
                 let (result, future) =
                     match futures_util::future::select(preferred_fut, fallback_delay).await {

src/client/legacy/pool.rs

@@ -9,12 +9,11 @@ use std::hash::Hash;
 use std::ops::{Deref, DerefMut};
 use std::pin::Pin;
 use std::sync::{Arc, Mutex, Weak};
-use std::task::{self, Poll};
+use std::task::{self, ready, Poll};
 
 use std::time::{Duration, Instant};
 
 use futures_channel::oneshot;
-use futures_core::ready;
 use tracing::{debug, trace};
 
 use hyper::rt::Timer as _;

src/server/conn/auto/mod.rs

@@ -7,11 +7,10 @@ use std::future::Future;
 use std::marker::PhantomPinned;
 use std::mem::MaybeUninit;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use std::{error::Error as StdError, io, time::Duration};
 
 use bytes::Bytes;
-use futures_core::ready;
 use http::{Request, Response};
 use http_body::Body;
 use hyper::{

src/service/oneshot.rs

@@ -1,8 +1,7 @@
-use futures_core::ready;
 use pin_project_lite::pin_project;
 use std::future::Future;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tower_service::Service;
 
 // Vendored from tower::util to reduce dependencies, the code is small enough.

tests/legacy_client.rs

@@ -1,18 +1,19 @@
 mod test_utils;
 
+use std::future::Future;
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener};
-use std::pin::Pin;
+use std::pin::{pin, Pin};
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::task::Poll;
 use std::thread;
 use std::time::Duration;
 
 use futures_channel::{mpsc, oneshot};
+use futures_core::Stream;
 use futures_util::future::{self, FutureExt, TryFutureExt};
 use futures_util::stream::StreamExt;
-use futures_util::{self, Stream};
 use http_body_util::BodyExt;
 use http_body_util::{Empty, Full, StreamBody};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -132,7 +133,7 @@ async fn drop_client_closes_idle_connections() {
     res.unwrap();
 
     // not closed yet, just idle
-    future::poll_fn(|ctx| {
+    std::future::poll_fn(|ctx| {
         assert!(Pin::new(&mut closes).poll_next(ctx).is_pending());
         Poll::Ready(())
     })
@@ -142,8 +143,7 @@ async fn drop_client_closes_idle_connections() {
     drop(client);
 
     // and wait a few ticks for the connections to close
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(t, close).await;
     t1.await.unwrap();
@@ -192,8 +192,7 @@ async fn drop_response_future_closes_in_progress_connection() {
     future::select(res, rx1).await;
 
     // res now dropped
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(t, close).await;
 }
@@ -248,8 +247,7 @@ async fn drop_response_body_closes_in_progress_connection() {
     res.unwrap();
 
     // and wait a few ticks to see the connection drop
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(t, close).await;
 }
@@ -301,8 +299,7 @@ async fn no_keep_alive_closes_connection() {
     let (res, _) = future::join(res, rx).await;
     res.unwrap();
 
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(close, t).await;
 }
@@ -348,8 +345,7 @@ async fn socket_disconnect_closes_idle_conn() {
     let (res, _) = future::join(res, rx).await;
     res.unwrap();
 
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(t, close).await;
 }
@@ -562,7 +558,7 @@ async fn client_keep_alive_when_response_before_request_body_ends() {
     });
 
     future::join(res, rx2).await.0.unwrap();
-    future::poll_fn(|ctx| {
+    std::future::poll_fn(|ctx| {
         assert!(Pin::new(&mut closes).poll_next(ctx).is_pending());
         Poll::Ready(())
     })
@@ -571,8 +567,7 @@ async fn client_keep_alive_when_response_before_request_body_ends() {
     assert_eq!(connects.load(Ordering::Relaxed), 1);
 
     drop(client);
-    let t = tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out"));
-    futures_util::pin_mut!(t);
+    let t = pin!(tokio::time::sleep(Duration::from_millis(100)).map(|_| panic!("time out")));
     let close = closes.into_future().map(|(opt, _)| opt.expect("closes"));
     future::select(t, close).await;
 }
@@ -1255,10 +1250,7 @@ impl tower_service::Service<hyper::Uri> for MockConnector {
     type Response = crate::MockConnection;
     type Error = std::io::Error;
     type Future = std::pin::Pin<
-        Box<
-            dyn futures_util::Future<Output = std::result::Result<Self::Response, Self::Error>>
-                + Send,
-        >,
+        Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>,
     >;
 
     // Polls the connector to check if it’s ready to handle a request.

tests/test_utils/mod.rs

@@ -1,10 +1,10 @@
+use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::task::{Context, Poll};
 
 use futures_channel::mpsc;
-use futures_util::task::{Context, Poll};
-use futures_util::Future;
 use futures_util::TryFutureExt;
 use hyper::Uri;
 use tokio::io::{self, AsyncRead, AsyncWrite, ReadBuf};