chore: Use released version of tower (#199)

* chore: Use released version of tower

* Update the rest

examples/Cargo.toml

@@ -95,7 +95,7 @@ tokio = { version = "0.2", features = ["rt-threaded", "time", "stream", "fs", "m
 futures = { version = "0.3", default-features = false, features = ["alloc"]}
 async-stream = "0.2"
 http = "0.2"
-tower = { git = "https://github.com/tower-rs/tower" }
+tower = "0.3" 
 
 # Required for routeguide
 serde = { version = "1.0", features = ["derive"] }

interop/Cargo.toml

@@ -24,8 +24,7 @@ http = "0.2"
 futures-core = "0.3"
 futures-util = "0.3"
 async-stream = "0.2"
-# tower = "=0.3.0-alpha.2"
+tower = "0.3"
-tower = { git = "https://github.com/tower-rs/tower" }
 http-body = "0.3"
 
 console = "0.9"

tonic/Cargo.toml

@@ -65,10 +65,10 @@ async-trait = { version = "0.1.13", optional = true }
 # transport
 hyper = { version = "0.13", features = ["stream"], optional = true }
 tokio = { version = "0.2", features = ["tcp"], optional = true }
-tower = { git = "https://github.com/tower-rs/tower", optional = true}
+tower = { version = "0.3", optional = true}
 tower-make = { version = "0.3", features = ["connect"] }
-tower-balance =  { git = "https://github.com/tower-rs/tower", optional = true }
+tower-balance =  { version = "0.3", optional = true }
-tower-load = { git = "https://github.com/tower-rs/tower", optional = true }
+tower-load = { version = "0.3", optional = true }
 tracing-futures = { version = "0.2", optional = true } 
 
 # rustls

tonic/src/transport/channel/endpoint.rs

@@ -155,6 +155,18 @@ impl Endpoint {
     }
 
     /// Intercept outbound HTTP Request headers;
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use tonic::transport::Endpoint;
+    /// # use std::time::Duration;
+    /// # let mut builder = Endpoint::from_static("https://example.com");
+    /// builder.intercept_headers(|headers| {
+    ///    // Do something with headers
+    ///    headers.insert("hello", "world".parse().unwrap());
+    /// });
+    /// ```
     pub fn intercept_headers<F>(self, f: F) -> Self
     where
         F: Fn(&mut http::HeaderMap) + Send + Sync + 'static,

tonic/src/transport/server/mod.rs

@@ -37,9 +37,8 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use tower::{
     layer::{Layer, Stack},
     limit::concurrency::ConcurrencyLimitLayer,
-    // timeout::TimeoutLayer,
+    timeout::TimeoutLayer,
-    Service,
+    Service, ServiceBuilder,
-    ServiceBuilder,
 };
 use tracing_futures::{Instrument, Instrumented};
 
@@ -60,7 +59,7 @@ pub struct Server {
     interceptor: Option<Interceptor>,
     trace_interceptor: Option<TraceInterceptor>,
     concurrency_limit: Option<usize>,
-    // timeout: Option<Duration>,
+    timeout: Option<Duration>,
     #[cfg(feature = "tls")]
     tls: Option<TlsAcceptor>,
     init_stream_window_size: Option<u32>,
@@ -109,6 +108,8 @@ impl Server {
 
     /// Set the concurrency limit applied to on requests inbound per connection.
     ///
+    /// # Example
+    ///
     /// ```
     /// # use tonic::transport::Server;
     /// # use tower_service::Service;
@@ -122,12 +123,21 @@ impl Server {
         }
     }
 
-    // FIXME: tower-timeout currentlly uses `From` instead of `Into` for the error
+    /// Set a timeout on for all request handlers.
-    // so our services do not align.
+    ///
-    // pub fn timeout(&mut self, timeout: Duration) -> &mut Self {
+    /// # Example
-    //     self.timeout = Some(timeout);
+    ///
-    //     self
+    /// ```
-    // }
+    /// # use tonic::transport::Server;
+    /// # use tower_service::Service;
+    /// # use std::time::Duration;
+    /// # let mut builder = Server::builder();
+    /// builder.timeout(Duration::from_secs(30));
+    /// ```
+    pub fn timeout(&mut self, timeout: Duration) -> &mut Self {
+        self.timeout = Some(timeout);
+        self
+    }
 
     /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
     /// stream-level flow control.
@@ -266,7 +276,7 @@ impl Server {
         let init_connection_window_size = self.init_connection_window_size;
         let init_stream_window_size = self.init_stream_window_size;
         let max_concurrent_streams = self.max_concurrent_streams;
-        // let timeout = self.timeout.clone();
+        let timeout = self.timeout.clone();
 
         let tcp = incoming::tcp_incoming(incoming, self);
         let incoming = accept::from_stream::<_, _, crate::Error>(tcp);
@@ -275,7 +285,7 @@ impl Server {
             inner: svc,
             interceptor,
             concurrency_limit,
-            // timeout,
+            timeout,
             span,
         };
 
@@ -454,7 +464,7 @@ impl<S> fmt::Debug for Svc<S> {
 struct MakeSvc<S> {
     interceptor: Option<Interceptor>,
     concurrency_limit: Option<usize>,
-    // timeout: Option<Duration>,
+    timeout: Option<Duration>,
     inner: S,
     span: Option<TraceInterceptor>,
 }
@@ -482,13 +492,13 @@ where
         let interceptor = self.interceptor.clone();
         let svc = self.inner.clone();
         let concurrency_limit = self.concurrency_limit;
-        // let timeout = self.timeout.clone();
+        let timeout = self.timeout.clone();
         let span = self.span.clone();
 
         Box::pin(async move {
             let svc = ServiceBuilder::new()
                 .optional_layer(concurrency_limit.map(ConcurrencyLimitLayer::new))
-                // .optional_layer(timeout.map(TimeoutLayer::new))
+                .optional_layer(timeout.map(TimeoutLayer::new))
                 .service(svc);
 
             let svc = if let Some(interceptor) = interceptor {

tonic/src/transport/service/either.rs

@@ -1,72 +0,0 @@
-use futures_util::future::{MapErr, TryFutureExt};
-use std::{
-    future::Future,
-    pin::Pin,
-    task::{Context, Poll},
-};
-use tower::Service;
-
-pub(crate) enum Either<A, B> {
-    A(A),
-    B(B),
-}
-
-impl<A, B, Request, Response> Service<Request> for Either<A, B>
-where
-    A: Service<Request, Response = Response>,
-    B: Service<Request, Response = Response>,
-    A::Error: Into<crate::Error>,
-    B::Error: Into<crate::Error>,
-{
-    type Response = Response;
-    type Error = crate::Error;
-    type Future = Either<
-        MapErr<A::Future, fn(A::Error) -> crate::Error>,
-        MapErr<B::Future, fn(B::Error) -> crate::Error>,
-    >;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self {
-            Either::A(svc) => svc.poll_ready(cx).map_err(Into::into),
-            Either::B(svc) => svc.poll_ready(cx).map_err(Into::into),
-        }
-    }
-
-    fn call(&mut self, req: Request) -> Self::Future {
-        match self {
-            Either::A(svc) => {
-                let fut = svc
-                    .call(req)
-                    .map_err((|e| e.into()) as fn(A::Error) -> crate::Error);
-                Either::A(fut)
-            }
-
-            Either::B(svc) => {
-                let fut = svc
-                    .call(req)
-                    .map_err((|e| e.into()) as fn(B::Error) -> crate::Error);
-                Either::B(fut)
-            }
-        }
-    }
-}
-
-impl<A: Unpin, B: Unpin> Unpin for Either<A, B> {}
-
-impl<A, B> Future for Either<A, B>
-where
-    A: Future,
-    B: Future<Output = A::Output>,
-{
-    type Output = A::Output;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        // safe because we do not exposed an unchecked mut beyond this projection.
-        let mut me = unsafe { self.get_unchecked_mut() };
-
-        match &mut me {
-            Either::A(fut) => unsafe { Pin::new_unchecked(fut) }.poll(cx),
-            Either::B(fut) => unsafe { Pin::new_unchecked(fut) }.poll(cx),
-        }
-    }
-}

tonic/src/transport/service/layer.rs

@@ -1,8 +1,9 @@
-use super::either::Either;
 use tower::{
     layer::{Layer, Stack},
+    util::Either,
     ServiceBuilder,
 };
+
 pub(crate) trait ServiceBuilderExt<L> {
     fn layer_fn<F: Fn(S) -> Out, S, Out>(self, f: F) -> ServiceBuilder<Stack<LayerFn<F>, L>>;
 

tonic/src/transport/service/mod.rs

@@ -2,7 +2,6 @@ mod add_origin;
 mod connection;
 mod connector;
 mod discover;
-mod either;
 mod io;
 mod layer;
 mod reconnect;