chore(tonic): Reduce futures-util (#1457)

* chore(tonic): Reduce futures-util

* chore(transport): Vendor futures_util::future::BoxFuture

tonic/Cargo.toml

@@ -27,11 +27,12 @@ codegen = ["dep:async-trait"]
 gzip = ["dep:flate2"]
 default = ["transport", "codegen", "prost"]
 prost = ["dep:prost"]
-tls = ["dep:rustls-pemfile", "transport", "dep:tokio-rustls", "dep:async-stream", "tokio/rt"]
+tls = ["dep:rustls-pemfile", "transport", "dep:tokio-rustls", "tokio/rt"]
 tls-roots = ["tls-roots-common", "dep:rustls-native-certs"]
 tls-roots-common = ["tls"]
 tls-webpki-roots = ["tls-roots-common", "dep:webpki-roots"]
 transport = [
+  "dep:async-stream",
   "dep:axum",
   "channel",
   "dep:h2",

tonic/src/codec/encode.rs

@@ -2,15 +2,15 @@ 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, StreamExt, TryStream, TryStreamExt};
+use futures_util::ready;
 use http::HeaderMap;
 use http_body::Body;
 use pin_project::pin_project;
 use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio_stream::Stream;
+use tokio_stream::{Stream, StreamExt};
 
 pub(super) const BUFFER_SIZE: usize = 8 * 1024;
 
@@ -31,8 +31,7 @@ where
         compression_encoding,
         compression_override,
         max_message_size,
-    )
-    .into_stream();
+    );
 
     EncodeBody::new_server(stream)
 }
@@ -53,8 +52,7 @@ where
         compression_encoding,
         SingleMessageCompressionOverride::default(),
         max_message_size,
-    )
-    .into_stream();
+    );
     EncodeBody::new_client(stream)
 }
 
@@ -64,7 +62,7 @@ fn encode<T, U>(
     compression_encoding: Option<CompressionEncoding>,
     compression_override: SingleMessageCompressionOverride,
     max_message_size: Option<usize>,
-) -> impl TryStream<Ok = Bytes, Error = Status>
+) -> impl Stream<Item = Result<Bytes, Status>>
 where
     T: Encoder<Error = Status>,
     U: Stream<Item = Result<T::Item, Status>>,
@@ -251,8 +249,8 @@ where
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        let mut self_proj = self.project();
-        match ready!(self_proj.inner.try_poll_next_unpin(cx)) {
+        let self_proj = self.project();
+        match ready!(self_proj.inner.poll_next(cx)) {
             Some(Ok(d)) => Some(Ok(d)).into(),
             Some(Err(status)) => match self_proj.state.role {
                 Role::Client => Some(Err(status)).into(),

tonic/src/codegen.rs

@@ -1,7 +1,6 @@
 //! Codegen exports used by `tonic-build`.
 
 pub use async_trait::async_trait;
-pub use futures_util::future::{ok, poll_fn, Ready};
 pub use tokio_stream;
 
 pub use std::future::Future;

tonic/src/server/grpc.rs

@@ -7,9 +7,9 @@ use crate::{
     server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService},
     Code, Request, Status,
 };
-use futures_util::{TryStream, TryStreamExt};
 use http_body::Body;
 use std::fmt;
+use tokio_stream::{Stream, StreamExt};
 
 macro_rules! t {
     ($result:expr) => {
@@ -428,7 +428,7 @@ where
         max_message_size: Option<usize>,
     ) -> http::Response<BoxBody>
     where
-        B: TryStream<Ok = T::Encode, Error = Status> + Send + 'static,
+        B: Stream<Item = Result<T::Encode, Status>> + Send + 'static,
     {
         let response = match response {
             Ok(r) => r,
@@ -453,7 +453,7 @@ where
 
         let body = encode_server(
             self.codec.encoder(),
-            body.into_stream(),
+            body,
             accept_encoding,
             compression_override,
             max_message_size,

tonic/src/transport/channel/mod.rs

@@ -188,7 +188,7 @@ impl Channel {
         D: Discover<Service = Connection> + Unpin + Send + 'static,
         D::Error: Into<crate::Error>,
         D::Key: Hash + Send + Clone,
-        E: Executor<futures_util::future::BoxFuture<'static, ()>> + Send + Sync + 'static,
+        E: Executor<crate::transport::BoxFuture<'static, ()>> + Send + Sync + 'static,
     {
         let svc = Balance::new(discover);
 

tonic/src/transport/mod.rs

@@ -118,5 +118,4 @@ pub use self::server::ServerTlsConfig;
 #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
 pub use self::tls::Identity;
 
-type BoxFuture<T, E> =
-    std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, E>> + Send + 'static>>;
+type BoxFuture<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;

tonic/src/transport/server/incoming.rs

@@ -14,7 +14,7 @@ use tokio::{
     io::{AsyncRead, AsyncWrite},
     net::TcpListener,
 };
-use tokio_stream::Stream;
+use tokio_stream::{Stream, StreamExt};
 
 #[cfg(not(feature = "tls"))]
 pub(crate) fn tcp_incoming<IO, IE, L>(
@@ -25,8 +25,13 @@ where
     IO: AsyncRead + AsyncWrite + Connected + Unpin + Send + 'static,
     IE: Into<crate::Error>,
 {
-    use futures_util::stream::TryStreamExt;
-    incoming.err_into().map_ok(ServerIo::new_io)
+    async_stream::try_stream! {
+        futures_util::pin_mut!(incoming);
+
+        while let Some(item) = incoming.next().await {
+            yield item.map(ServerIo::new_io)?
+        }
+    }
 }
 
 #[cfg(feature = "tls")]
@@ -81,8 +86,6 @@ async fn select<IO: 'static, IE>(
 where
     IE: Into<crate::Error>,
 {
-    use tokio_stream::StreamExt;
-
     if tasks.is_empty() {
         return match incoming.try_next().await {
             Ok(Some(stream)) => SelectOutput::Incoming(stream),

tonic/src/transport/server/mod.rs

@@ -36,15 +36,15 @@ use self::recover_error::RecoverError;
 use super::service::{GrpcTimeout, ServerIo};
 use crate::body::BoxBody;
 use bytes::Bytes;
-use futures_util::{future, ready};
+use futures_util::ready;
 use http::{Request, Response};
 use http_body::Body as _;
 use hyper::{server::accept, Body};
 use pin_project::pin_project;
 use std::{
     convert::Infallible,
     fmt,
-    future::Future,
+    future::{self, Future},
     marker::PhantomData,
     net::SocketAddr,
     pin::Pin,

tonic/src/transport/service/add_origin.rs

@@ -1,4 +1,4 @@
-use futures_util::future::BoxFuture;
+use crate::transport::BoxFuture;
 use http::uri::Authority;
 use http::uri::Scheme;
 use http::{Request, Uri};

tonic/src/transport/service/connection.rs

@@ -1,6 +1,8 @@
-use super::super::BoxFuture;
 use super::{grpc_timeout::GrpcTimeout, reconnect::Reconnect, AddOrigin, UserAgent};
-use crate::{body::BoxBody, transport::Endpoint};
+use crate::{
+    body::BoxBody,
+    transport::{BoxFuture, Endpoint},
+};
 use http::Uri;
 use hyper::client::conn::Builder;
 use hyper::client::connect::Connection as HyperConnection;
@@ -100,7 +102,7 @@ impl Connection {
 impl Service<Request> for Connection {
     type Response = Response;
     type Error = crate::Error;
-    type Future = BoxFuture<Self::Response, Self::Error>;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Service::poll_ready(&mut self.inner, cx).map_err(Into::into)

tonic/src/transport/service/connector.rs

@@ -64,7 +64,7 @@ where
 {
     type Response = BoxedIo;
     type Error = crate::Error;
-    type Future = BoxFuture<Self::Response, Self::Error>;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         MakeConnection::poll_ready(&mut self.inner, cx).map_err(Into::into)

tonic/src/transport/service/executor.rs

@@ -1,4 +1,4 @@
-use futures_util::future::BoxFuture;
+use crate::transport::BoxFuture;
 use std::{future::Future, sync::Arc};
 
 pub(crate) use hyper::rt::Executor;