Replace futures-util with atomic-waker and manual poll_fn (#721)

This removes ones of the heavier dependencies. The goal is to also remove
futures-util from the rest of the dependency tree, which is hard to justify when
futures-util is used in more fundamental dependencies.

Cargo.toml

@@ -39,9 +39,9 @@ members = [
 ]
 
 [dependencies]
+atomic-waker = "1.0.0"
 futures-core = { version = "0.3", default-features = false }
 futures-sink = { version = "0.3", default-features = false }
-futures-util = { version = "0.3", default-features = false }
 tokio-util = { version = "0.7.1", features = ["codec", "io"] }
 tokio = { version = "1", features = ["io-util"] }
 bytes = "1"

src/client.rs

@@ -1492,7 +1492,7 @@ impl ResponseFuture {
 impl PushPromises {
     /// Get the next `PushPromise`.
     pub async fn push_promise(&mut self) -> Option<Result<PushPromise, crate::Error>> {
-        futures_util::future::poll_fn(move |cx| self.poll_push_promise(cx)).await
+        crate::poll_fn(move |cx| self.poll_push_promise(cx)).await
     }
 
     #[doc(hidden)]

src/lib.rs

@@ -139,3 +139,25 @@ pub use crate::share::{FlowControl, Ping, PingPong, Pong, RecvStream, SendStream
 
 #[cfg(feature = "unstable")]
 pub use codec::{Codec, SendError, UserError};
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Creates a future from a function that returns `Poll`.
+fn poll_fn<T, F: FnMut(&mut Context<'_>) -> T>(f: F) -> PollFn<F> {
+    PollFn(f)
+}
+
+/// The future created by `poll_fn`.
+struct PollFn<F>(F);
+
+impl<F> Unpin for PollFn<F> {}
+
+impl<T, F: FnMut(&mut Context<'_>) -> Poll<T>> Future for PollFn<F> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        (self.0)(cx)
+    }
+}

src/proto/ping_pong.rs

@@ -2,8 +2,8 @@ use crate::codec::Codec;
 use crate::frame::Ping;
 use crate::proto::{self, PingPayload};
 
+use atomic_waker::AtomicWaker;
 use bytes::Buf;
-use futures_util::task::AtomicWaker;
 use std::io;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;

src/server.rs

@@ -413,7 +413,7 @@ where
     pub async fn accept(
         &mut self,
     ) -> Option<Result<(Request<RecvStream>, SendResponse<B>), crate::Error>> {
-        futures_util::future::poll_fn(move |cx| self.poll_accept(cx)).await
+        crate::poll_fn(move |cx| self.poll_accept(cx)).await
     }
 
     #[doc(hidden)]

src/share.rs

@@ -410,12 +410,12 @@ impl RecvStream {
 
     /// Get the next data frame.
     pub async fn data(&mut self) -> Option<Result<Bytes, crate::Error>> {
-        futures_util::future::poll_fn(move |cx| self.poll_data(cx)).await
+        crate::poll_fn(move |cx| self.poll_data(cx)).await
     }
 
     /// Get optional trailers for this stream.
     pub async fn trailers(&mut self) -> Result<Option<HeaderMap>, crate::Error> {
-        futures_util::future::poll_fn(move |cx| self.poll_trailers(cx)).await
+        crate::poll_fn(move |cx| self.poll_trailers(cx)).await
     }
 
     /// Poll for the next data frame.
@@ -549,7 +549,7 @@ impl PingPong {
     /// Send a PING frame and wait for the peer to send the pong.
     pub async fn ping(&mut self, ping: Ping) -> Result<Pong, crate::Error> {
         self.send_ping(ping)?;
-        futures_util::future::poll_fn(|cx| self.poll_pong(cx)).await
+        crate::poll_fn(|cx| self.poll_pong(cx)).await
     }
 
     #[doc(hidden)]