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;
@@ -968,14 +968,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::Sleep;

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

@@ -2,7 +2,7 @@ mod test_utils;
 
 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;
@@ -143,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();
@@ -193,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;
 }
@@ -249,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;
 }
@@ -302,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;
 }
@@ -349,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;
 }
@@ -572,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;
 }

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};