chore: Remove futures crates (#1478)

* chore: Remove futures crates

* chore(tonic): Replace pin_utils::pin_mut with tokio::pin

tests/compression/Cargo.toml

@@ -8,7 +8,6 @@ version = "0.1.0"
 
 [dependencies]
 bytes = "1"
-futures-core = "0.3"
 http = "0.2"
 http-body = "0.4"
 hyper = "0.14.3"

tests/compression/src/util.rs

@@ -1,6 +1,5 @@
 use super::*;
 use bytes::Bytes;
-use futures_core::ready;
 use http_body::Body;
 use pin_project::pin_project;
 use std::{
@@ -9,7 +8,7 @@ use std::{
         atomic::{AtomicUsize, Ordering::SeqCst},
         Arc,
     },
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tonic::transport::{server::Connected, Channel};

tonic-web/Cargo.toml

@@ -17,7 +17,6 @@ version = "0.9.2"
 [dependencies]
 base64 = "0.21"
 bytes = "1.0"
-futures-core = "0.3"
 tokio-stream = "0.1"
 http = "0.2"
 http-body = "0.4"

tonic-web/src/call.rs

@@ -1,10 +1,9 @@
 use std::error::Error;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 use base64::Engine as _;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
-use futures_core::ready;
 use http::{header, HeaderMap, HeaderName, HeaderValue};
 use http_body::{Body, SizeHint};
 use pin_project::pin_project;

tonic-web/src/client.rs

@@ -1,13 +1,12 @@
 use bytes::Bytes;
-use futures_core::ready;
 use http::header::CONTENT_TYPE;
 use http::{Request, Response, Version};
 use http_body::Body;
 use pin_project::pin_project;
 use std::error::Error;
 use std::future::Future;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tower_layer::Layer;
 use tower_service::Service;
 use tracing::debug;

tonic-web/src/service.rs

@@ -1,7 +1,6 @@
-use futures_core::ready;
 use std::future::Future;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 use http::{header, HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Version};
 use hyper::Body;

tonic/Cargo.toml

@@ -37,7 +37,8 @@ transport = [
   "channel",
   "dep:h2",
   "dep:hyper",
-  "tokio",
+  "tokio/net",
+  "tokio/time",
   "dep:tower",
   "dep:hyper-timeout",
 ]
@@ -50,10 +51,10 @@ channel = []
 [dependencies]
 base64 = "0.21"
 bytes = "1.0"
-futures-util = {version = "0.3", default-features = false}
 http = "0.2"
 tracing = "0.1"
 
+tokio = "1.0.1"
 http-body = "0.4.4"
 percent-encoding = "2.1"
 pin-project = "1.0.11"
@@ -70,7 +71,6 @@ async-trait = {version = "0.1.13", optional = true}
 h2 = {version = "0.3.17", optional = true}
 hyper = {version = "0.14.14", features = ["full"], optional = true}
 hyper-timeout = {version = "0.4", optional = true}
-tokio = {version = "1.0.1", features = ["net", "time"], optional = true}
 tokio-stream = "0.1"
 tower = {version = "0.4.7", default-features = false, features = ["balance", "buffer", "discover", "limit", "load", "make", "timeout", "util"], optional = true}
 axum = {version = "0.6.9", default_features = false, optional = true}

tonic/src/client/grpc.rs

@@ -6,13 +6,12 @@ use crate::{
     request::SanitizeHeaders,
     Code, Request, Response, Status,
 };
-use futures_util::future;
 use http::{
     header::{HeaderValue, CONTENT_TYPE, TE},
     uri::{Parts, PathAndQuery, Uri},
 };
 use http_body::Body;
-use std::fmt;
+use std::{fmt, future};
 use tokio_stream::{Stream, StreamExt};
 
 /// A gRPC client dispatcher.
@@ -240,7 +239,7 @@ impl<T> Grpc<T> {
         let (mut parts, body, extensions) =
             self.streaming(request, path, codec).await?.into_parts();
 
-        futures_util::pin_mut!(body);
+        tokio::pin!(body);
 
         let message = body
             .try_next()

tonic/src/codec/decode.rs

@@ -2,12 +2,12 @@ use super::compression::{decompress, CompressionEncoding};
 use super::{DecodeBuf, Decoder, DEFAULT_MAX_RECV_MESSAGE_SIZE, HEADER_SIZE};
 use crate::{body::BoxBody, metadata::MetadataMap, Code, Status};
 use bytes::{Buf, BufMut, BytesMut};
-use futures_util::{future, ready};
 use http::StatusCode;
 use http_body::Body;
 use std::{
-    fmt,
+    fmt, future,
     pin::Pin,
+    task::ready,
     task::{Context, Poll},
 };
 use tokio_stream::Stream;

tonic/src/codec/encode.rs

@@ -2,13 +2,12 @@ use super::compression::{compress, CompressionEncoding, SingleMessageCompression
 use super::{EncodeBuf, Encoder, DEFAULT_MAX_SEND_MESSAGE_SIZE, HEADER_SIZE};
 use crate::{Code, Status};
 use bytes::{BufMut, Bytes, BytesMut};
-use futures_util::ready;
 use http::HeaderMap;
 use http_body::Body;
 use pin_project::pin_project;
 use std::{
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 use tokio_stream::{Stream, StreamExt};
 

tonic/src/codec/prost.rs

@@ -165,7 +165,7 @@ mod tests {
             None,
         );
 
-        futures_util::pin_mut!(body);
+        tokio::pin!(body);
 
         while let Some(r) = body.data().await {
             r.unwrap();
@@ -189,7 +189,7 @@ mod tests {
             Some(MAX_MESSAGE_SIZE),
         );
 
-        futures_util::pin_mut!(body);
+        tokio::pin!(body);
 
         assert!(body.data().await.is_none());
         assert_eq!(
@@ -223,7 +223,7 @@ mod tests {
             Some(usize::MAX),
         );
 
-        futures_util::pin_mut!(body);
+        tokio::pin!(body);
 
         assert!(body.data().await.is_none());
         assert_eq!(

tonic/src/server/grpc.rs

@@ -382,7 +382,7 @@ where
             self.max_decoding_message_size,
         );
 
-        futures_util::pin_mut!(stream);
+        tokio::pin!(stream);
 
         let message = stream
             .try_next()

tonic/src/transport/channel/mod.rs

@@ -23,7 +23,7 @@ use std::{
     future::Future,
     hash::Hash,
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 use tokio::{
     io::{AsyncRead, AsyncWrite},
@@ -220,8 +220,7 @@ impl Future for ResponseFuture {
     type Output = Result<Response<hyper::Body>, super::Error>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let val = futures_util::ready!(Pin::new(&mut self.inner).poll(cx))
-            .map_err(super::Error::from_source)?;
+        let val = ready!(Pin::new(&mut self.inner).poll(cx)).map_err(super::Error::from_source)?;
         Ok(val).into()
     }
 }

tonic/src/transport/server/incoming.rs

@@ -26,7 +26,7 @@ where
     IE: Into<crate::Error>,
 {
     async_stream::try_stream! {
-        futures_util::pin_mut!(incoming);
+        tokio::pin!(incoming);
 
         while let Some(item) = incoming.next().await {
             yield item.map(ServerIo::new_io)?
@@ -44,7 +44,7 @@ where
     IE: Into<crate::Error>,
 {
     async_stream::try_stream! {
-        futures_util::pin_mut!(incoming);
+        tokio::pin!(incoming);
 
         let mut tasks = tokio::task::JoinSet::new();
 

tonic/src/transport/server/mod.rs

@@ -38,7 +38,6 @@ use self::recover_error::RecoverError;
 use super::service::{GrpcTimeout, ServerIo};
 use crate::body::BoxBody;
 use bytes::Bytes;
-use futures_util::ready;
 use http::{Request, Response};
 use http_body::Body as _;
 use hyper::{server::accept, Body};
@@ -51,7 +50,7 @@ use std::{
     net::SocketAddr,
     pin::Pin,
     sync::Arc,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
     time::Duration,
 };
 use tokio::io::{AsyncRead, AsyncWrite};

tonic/src/transport/server/recover_error.rs

@@ -1,11 +1,10 @@
 use crate::Status;
-use futures_util::ready;
 use http::Response;
 use pin_project::pin_project;
 use std::{
     future::Future,
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 use tower::Service;
 

tonic/src/transport/service/grpc_timeout.rs

@@ -5,7 +5,7 @@ use std::{
     fmt,
     future::Future,
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
     time::Duration,
 };
 use tokio::time::Sleep;
@@ -89,7 +89,7 @@ where
         }
 
         if let Some(sleep) = this.sleep.as_pin_mut() {
-            futures_util::ready!(sleep.poll(cx));
+            ready!(sleep.poll(cx));
             return Poll::Ready(Err(TimeoutExpired(()).into()));
         }
 

tonic/src/transport/service/router.rs

@@ -10,7 +10,7 @@ use std::{
     fmt,
     future::Future,
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 use tower::ServiceExt;
 use tower_service::Service;
@@ -131,7 +131,7 @@ impl Future for RoutesFuture {
     type Output = Result<Response<BoxBody>, crate::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match futures_util::ready!(self.project().0.poll(cx)) {
+        match ready!(self.project().0.poll(cx)) {
             Ok(res) => Ok(res.map(boxed)).into(),
             Err(err) => match err {},
         }