feat(transport): Make tower internal dependency (#1947)

* feat(channel): Add Change type to make tower internal dependency

* feat(server): Remove NamedService implementation for Either

* feat(web): Add CorsGrpcWebResponseFuture to make tower-http internal dependency

Author: tottoto

examples/Cargo.toml

@@ -51,12 +51,10 @@ path = "src/load_balance/server.rs"
 [[bin]]
 name = "dynamic-load-balance-client"
 path = "src/dynamic_load_balance/client.rs"
-required-features = ["dynamic-load-balance"]
 
 [[bin]]
 name = "dynamic-load-balance-server"
 path = "src/dynamic_load_balance/server.rs"
-required-features = ["dynamic-load-balance"]
 
 [[bin]]
 name = "tls-client"
@@ -267,13 +265,12 @@ json-codec = ["dep:serde", "dep:serde_json", "dep:bytes"]
 compression = ["tonic/gzip"]
 tls = ["tonic/tls"]
 tls-rustls = ["dep:http", "dep:hyper", "dep:hyper-util", "dep:hyper-rustls", "dep:tower", "tower-http/util", "tower-http/add-extension", "dep:rustls-pki-types", "dep:tokio-rustls", "dep:pin-project", "dep:http-body-util"]
-dynamic-load-balance = ["dep:tower"]
 tls-client-auth = ["tonic/tls"]
 types = ["dep:tonic-types"]
 h2c = ["dep:hyper", "dep:tower", "dep:http", "dep:hyper-util"]
 cancellation = ["dep:tokio-util"]
 
-full = ["gcp", "routeguide", "reflection", "autoreload", "health", "grpc-web", "tracing", "uds", "streaming", "mock", "tower", "json-codec", "compression", "tls", "tls-rustls", "dynamic-load-balance", "tls-client-auth", "types", "cancellation", "h2c"]
+full = ["gcp", "routeguide", "reflection", "autoreload", "health", "grpc-web", "tracing", "uds", "streaming", "mock", "tower", "json-codec", "compression", "tls", "tls-rustls", "tls-client-auth", "types", "cancellation", "h2c"]
 default = ["full"]
 
 [dependencies]

examples/src/dynamic_load_balance/client.rs

@@ -3,15 +3,14 @@ pub mod pb {
 }
 
 use pb::{echo_client::EchoClient, EchoRequest};
+use tonic::transport::channel::Change;
 use tonic::transport::Channel;
-
 use tonic::transport::Endpoint;
 
 use std::sync::Arc;
 
 use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
 use tokio::time::timeout;
-use tower::discover::Change;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {

tonic-web/Cargo.toml

@@ -43,7 +43,6 @@ allowed_external_types = [
   # not major released
   "futures_core::stream::Stream",
   "http_body_util::combinators::box_body::UnsyncBoxBody",
-  "tower_http::cors::Cors",
   "tower_layer::Layer",
   "tower_service::Service",
 ]

tonic-web/src/lib.rs

@@ -108,7 +108,14 @@ mod layer;
 mod service;
 
 use http::header::HeaderName;
-use std::time::Duration;
+use pin_project::pin_project;
+use std::{
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
 use tonic::{body::BoxBody, server::NamedService, Status};
 use tower_http::cors::{AllowOrigin, CorsLayer};
 use tower_layer::Layer;
@@ -159,18 +166,37 @@ where
 {
     type Response = S::Response;
     type Error = S::Error;
-    type Future =
-        <tower_http::cors::Cors<GrpcWebService<S>> as Service<http::Request<BoxBody>>>::Future;
+    type Future = CorsGrpcWebResponseFuture<S::Future>;
 
-    fn poll_ready(
-        &mut self,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), Self::Error>> {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.0.poll_ready(cx)
     }
 
     fn call(&mut self, req: http::Request<BoxBody>) -> Self::Future {
-        self.0.call(req)
+        CorsGrpcWebResponseFuture(self.0.call(req))
+    }
+}
+
+/// Response Future for the [`CorsGrpcWeb`].
+#[pin_project]
+pub struct CorsGrpcWebResponseFuture<F>(
+    #[pin] tower_http::cors::ResponseFuture<service::ResponseFuture<F>>,
+);
+
+impl<F, E> Future for CorsGrpcWebResponseFuture<F>
+where
+    F: Future<Output = Result<http::Response<BoxBody>, E>>,
+{
+    type Output = Result<http::Response<BoxBody>, E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().0.poll(cx)
+    }
+}
+
+impl<F> fmt::Debug for CorsGrpcWebResponseFuture<F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("CorsGrpcWebResponseFuture").finish()
     }
 }
 

tonic/Cargo.toml

@@ -137,7 +137,6 @@ allowed_external_types = [
   "futures_core::stream::Stream",
   "h2::error::Error",
   "http_body_util::combinators::box_body::UnsyncBoxBody",
-  "tower::discover::Change",
   "tower_service::Service",
   "tower_layer::Layer",
   "tower_layer::stack::Stack",

tonic/src/transport/channel/mod.rs

@@ -5,6 +5,7 @@ pub(crate) mod service;
 #[cfg(feature = "_tls-any")]
 mod tls;
 
+pub use self::service::Change;
 pub use endpoint::Endpoint;
 #[cfg(feature = "_tls-any")]
 pub use tls::ClientTlsConfig;
@@ -29,7 +30,7 @@ use hyper::rt;
 use tower::balance::p2c::Balance;
 use tower::{
     buffer::{future::ResponseFuture as BufferResponseFuture, Buffer},
-    discover::{Change, Discover},
+    discover::Discover,
     util::BoxService,
     Service,
 };

tonic/src/transport/channel/service/discover.rs

@@ -7,11 +7,17 @@ use std::{
     task::{Context, Poll},
 };
 use tokio::sync::mpsc::Receiver;
-
 use tokio_stream::Stream;
-use tower::discover::Change;
-
-type DiscoverResult<K, S, E> = Result<Change<K, S>, E>;
+use tower::discover::Change as TowerChange;
+
+/// A change in the service set.
+#[derive(Debug, Clone)]
+pub enum Change<K, V> {
+    /// A new service identified by key `K` was identified.
+    Insert(K, V),
+    /// The service identified by key `K` disappeared.
+    Remove(K),
+}
 
 pub(crate) struct DynamicServiceStream<K: Hash + Eq + Clone> {
     changes: Receiver<Change<K, Endpoint>>,
@@ -24,7 +30,7 @@ impl<K: Hash + Eq + Clone> DynamicServiceStream<K> {
 }
 
 impl<K: Hash + Eq + Clone> Stream for DynamicServiceStream<K> {
-    type Item = DiscoverResult<K, Connection, crate::BoxError>;
+    type Item = Result<TowerChange<K, Connection>, crate::BoxError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let c = &mut self.changes;
@@ -39,10 +45,10 @@ impl<K: Hash + Eq + Clone> Stream for DynamicServiceStream<K> {
                     http.enforce_http(false);
 
                     let connection = Connection::lazy(endpoint.connector(http), endpoint);
-                    let change = Ok(Change::Insert(k, connection));
+                    let change = Ok(TowerChange::Insert(k, connection));
                     Poll::Ready(Some(change))
                 }
-                Change::Remove(k) => Poll::Ready(Some(Ok(Change::Remove(k)))),
+                Change::Remove(k) => Poll::Ready(Some(Ok(TowerChange::Remove(k)))),
             },
         }
     }

tonic/src/transport/channel/service/mod.rs

@@ -11,6 +11,7 @@ mod connection;
 pub(super) use self::connection::Connection;
 
 mod discover;
+pub use self::discover::Change;
 pub(super) use self::discover::DynamicServiceStream;
 
 mod io;

tonic/src/transport/server/mod.rs

@@ -64,7 +64,7 @@ use tower::{
     layer::util::{Identity, Stack},
     layer::Layer,
     limit::concurrency::ConcurrencyLimitLayer,
-    util::{BoxCloneService, Either},
+    util::BoxCloneService,
     Service, ServiceBuilder, ServiceExt,
 };
 
@@ -138,10 +138,6 @@ pub struct Router<L = Identity> {
     routes: Routes,
 }
 
-impl<S: NamedService, T> NamedService for Either<S, T> {
-    const NAME: &'static str = S::NAME;
-}
-
 impl Server {
     /// Create a new server builder that can configure a [`Server`].
     pub fn builder() -> Self {