Add layers

tonic-interop/src/bin/client.rs

@@ -1,5 +1,6 @@
+use std::time::Duration;
 use structopt::{clap::arg_enum, StructOpt};
-use tonic::transport::{Channel, Endpoint};
+use tonic::transport::Endpoint;
 use tonic_interop::client;
 
 #[derive(StructOpt)]
@@ -26,17 +27,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let test_cases = matches.test_case;
 
-    let addr = "localhost:10000";
+    let mut endpoint = Endpoint::from_static("http://localhost:10000")
-    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
+        .timeout(Duration::from_secs(5))
+        .concurrency_limit(30)
+        .clone();
 
-    let endpoint = if matches.use_tls {
+    if matches.use_tls {
         let ca = tokio::fs::read("tonic-interop/data/ca.pem").await?;
-        Endpoint::with_pem(origin, ca, Some("foo.test.google.fr".into()))
+        endpoint.tls_cert(ca, Some("foo.test.google.fr".into()));
-    } else {
+    }
-        Endpoint::from(origin)
-    };
 
-    let channel = Channel::builder().connect(endpoint)?;
+    let channel = endpoint.channel()?;
 
     let mut client = client::TestClient::new(channel.clone());
     let mut unimplemented_client = client::UnimplementedClient::new(channel);

tonic-interop/test.sh

@@ -27,6 +27,8 @@ echo ":; started grpc-go test server."
 # regardless of why (errors, SIGTERM, etc).
 trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT
 
+sleep 1
+
 ./target/debug/client \
  --test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\
 empty_stream,status_code_and_message,special_status_message,unimplemented_method,\
@@ -43,6 +45,8 @@ echo ":; started tonic test server."
 # regardless of why (errors, SIGTERM, etc).
 trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT
 
+sleep 1
+
 ./target/debug/client \
 --test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\
 empty_stream,status_code_and_message,special_status_message,unimplemented_method $ARG

tonic/Cargo.toml

@@ -24,12 +24,11 @@ async-trait = "0.1"
 # hyper = { version = "=0.13.0-alpha.1", optional = true }
 hyper = { git = "https://github.com/hyperium/hyper", features = ["unstable-stream"], optional = true }
 tokio = { version = "=0.2.0-alpha.4", default-features = false, features = ["tcp"], optional = true }
+tower = { version = "=0.3.0-alpha.1a", optional = true}
 tower-make = "=0.1.0-alpha.2"
-tower-reconnect = { version = "0.3.0-alpha.1", optional = true }
+tower-reconnect = { version = "=0.3.0-alpha.1", optional = true }
-tower-buffer = { version = "0.3.0-alpha.1", optional = true }
+tower-balance =  { version = "=0.3.0-alpha.1", optional = true }
-tower-balance =  { version = "0.3.0-alpha.1", optional = true }
+tower-load = { version = "=0.3.0-alpha.1", optional = true }
-tower-load = { version = "0.3.0-alpha.1", optional = true }
-tower-discover = { version = "0.3.0-alpha.1", optional = true }
 
 # openssl
 tokio-openssl = { version = "=0.4.0-alpha.4", optional = true }
@@ -42,16 +41,15 @@ tokio-rustls = { version = "0.12.0-alpha.2", optional = true }
 default = ["transport"]
 transport = [
     "hyper",
-    "tower",
+    "tower-1",
     "tokio",
     "native-tls",
 ]
-tower = [
+tower-1 = [
+    "tower",
     "tower-reconnect",
-    "tower-buffer",
     "tower-balance",
     "tower-load",
-    "tower-discover"
 ]
 native-tls = ["openssl", "tokio-openssl"]
 # rustls = ["tokio-rustls"]

tonic/src/transport/channel.rs

@@ -12,9 +12,9 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
+use tower::buffer::{future::ResponseFuture, Buffer};
+use tower::discover::Discover;
 use tower_balance::p2c::Balance;
-use tower_buffer::{future::ResponseFuture, Buffer};
-use tower_discover::Discover;
 use tower_service::Service;
 
 type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

tonic/src/transport/endpoint.rs

@@ -1,37 +1,58 @@
-use super::tls::Cert;
+use super::{channel::Channel, tls::Cert};
-use http::uri::Uri;
+use bytes::Bytes;
+use http::uri::{InvalidUriBytes, Uri};
+use std::time::Duration;
 
 #[derive(Debug, Clone)]
 pub struct Endpoint {
-    uri: Uri,
+    pub(super) uri: Uri,
-    cert: Option<Cert>,
+    pub(super) timeout: Option<Duration>,
+    pub(super) concurrency_limit: Option<usize>,
+    pub(super) cert: Option<Cert>,
 }
 
 impl Endpoint {
-    pub fn with_pem(uri: Uri, ca: Vec<u8>, domain: Option<String>) -> Self {
+    pub fn from_static(s: &'static str) -> Self {
-        let domain = domain.unwrap_or_else(|| uri.clone().to_string());
+        let uri = Uri::from_static(s);
-
+        Self::from(uri)
-        Self {
-            uri,
-            cert: Some(Cert {
-                ca,
-                domain,
-                key: None,
-            }),
-        }
     }
 
-    pub(crate) fn uri(&self) -> &Uri {
+    pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, InvalidUriBytes> {
-        &self.uri
+        let uri = Uri::from_shared(s.into())?;
+        Ok(Self::from(uri))
     }
 
-    pub(crate) fn take_cert(&mut self) -> Option<Cert> {
+    pub fn timeout(&mut self, dur: Duration) -> &mut Self {
-        self.cert.take()
+        self.timeout = Some(dur);
+        self
+    }
+
+    pub fn concurrency_limit(&mut self, limit: usize) -> &mut Self {
+        self.concurrency_limit = Some(limit);
+        self
+    }
+
+    pub fn tls_cert(&mut self, ca: Vec<u8>, domain: Option<String>) -> &mut Self {
+        self.cert = Some(Cert {
+            ca,
+            domain: domain.unwrap_or_else(|| self.uri.clone().to_string()),
+            key: None,
+        });
+        self
+    }
+
+    pub fn channel(&self) -> Result<Channel, super::Error> {
+        Channel::builder().connect(self.clone())
     }
 }
 
 impl From<Uri> for Endpoint {
     fn from(uri: Uri) -> Self {
-        Self { uri, cert: None }
+        Self {
+            uri,
+            concurrency_limit: None,
+            timeout: None,
+            cert: None,
+        }
     }
 }

tonic/src/transport/service/connect.rs

@@ -1,6 +1,5 @@
-use super::{AddOrigin, Connector};
+use super::{layer::ServiceBuilderExt, AddOrigin, Connector};
 use crate::{transport::Endpoint, BoxBody};
-use http::{Request, Response, Uri};
 use hyper::client::conn::Builder;
 use hyper::client::service::Connect as HyperConnect;
 use std::{
@@ -9,29 +8,50 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
+use tower::{
+    layer::Layer, limit::concurrency::ConcurrencyLimitLayer, timeout::TimeoutLayer,
+    util::BoxService, ServiceBuilder,
+};
 use tower_load::Load;
 use tower_reconnect::Reconnect;
 use tower_service::Service;
 
+type Request = http::Request<BoxBody>;
+type Response = http::Response<hyper::Body>;
+
 pub struct Connection {
-    inner: AddOrigin<Reconnect<HyperConnect<Connector, BoxBody, Uri>, Uri>>,
+    // inner: AddOrigin<Reconnect<HyperConnect<Connector, BoxBody, Uri>, Uri>>,
+    inner: BoxService<Request, Response, crate::Error>,
 }
 
 impl Connection {
     pub fn new(mut endpoint: Endpoint) -> Result<Self, crate::Error> {
-        let connector = Connector::new(endpoint.take_cert())?;
+        let connector = Connector::new(endpoint.cert.take())?;
 
         let settings = Builder::new().http2_only(true).clone();
-        let connect = HyperConnect::new(connector, settings);
-        let reconnect = Reconnect::new(connect, endpoint.uri().clone());
-        let inner = AddOrigin::new(reconnect, endpoint.uri().clone());
 
-        Ok(Self { inner })
+        let stack = ServiceBuilder::new()
+            .layer_fn(|s| AddOrigin::new(s, endpoint.uri.clone()))
+            .optional_layer(endpoint.timeout.map(|t| TimeoutLayer::new(t)))
+            .optional_layer(
+                endpoint
+                    .concurrency_limit
+                    .map(|l| ConcurrencyLimitLayer::new(l)),
+            )
+            .into_inner();
+
+        let conn = Reconnect::new(HyperConnect::new(connector, settings), endpoint.uri.clone());
+
+        let inner = stack.layer(conn);
+
+        Ok(Self {
+            inner: BoxService::new(inner),
+        })
     }
 }
 
-impl Service<Request<BoxBody>> for Connection {
+impl Service<Request> for Connection {
-    type Response = Response<hyper::Body>;
+    type Response = Response;
     type Error = crate::Error;
 
     type Future =
@@ -41,10 +61,8 @@ impl Service<Request<BoxBody>> for Connection {
         Service::poll_ready(&mut self.inner, cx).map_err(Into::into)
     }
 
-    fn call(&mut self, req: Request<BoxBody>) -> Self::Future {
+    fn call(&mut self, req: Request) -> Self::Future {
-        let fut = self.inner.call(req);
+        self.inner.call(req)
-        // TODO: we dont need to box here if we have too
-        Box::pin(fut)
     }
 }
 

tonic/src/transport/service/discover.rs

@@ -3,7 +3,7 @@ use crate::transport::Endpoint;
 use std::collections::VecDeque;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tower_discover::{Change, Discover};
+use tower::discover::{Change, Discover};
 
 #[derive(Debug)]
 pub struct ServiceList {

tonic/src/transport/service/layer.rs

@@ -0,0 +1,76 @@
+use tower::{
+    layer::{util::Stack, Layer},
+    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>>;
+
+    fn optional_layer_fn<F: Fn(S) -> Out, S, Out>(
+        self,
+        f: Option<F>,
+    ) -> ServiceBuilder<Stack<OptionalLayer<LayerFn<F>>, L>>;
+
+    fn optional_layer<T>(self, l: Option<T>) -> ServiceBuilder<Stack<OptionalLayer<T>, L>>;
+}
+
+impl<L> ServiceBuilderExt<L> for ServiceBuilder<L> {
+    fn layer_fn<F, S, Out>(self, f: F) -> ServiceBuilder<Stack<LayerFn<F>, L>>
+    where
+        F: Fn(S) -> Out,
+    {
+        self.layer(LayerFn(f))
+    }
+
+    fn optional_layer_fn<F, S, Out>(
+        self,
+        f: Option<F>,
+    ) -> ServiceBuilder<Stack<OptionalLayer<LayerFn<F>>, L>>
+    where
+        F: Fn(S) -> Out,
+    {
+        let layer = OptionalLayer {
+            inner: f.map(|f| LayerFn(f)),
+        };
+
+        self.layer(layer)
+    }
+
+    fn optional_layer<T>(self, inner: Option<T>) -> ServiceBuilder<Stack<OptionalLayer<T>, L>> {
+        self.layer(OptionalLayer { inner })
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct LayerFn<F>(F);
+
+impl<F, S, Out> Layer<S> for LayerFn<F>
+where
+    F: Fn(S) -> Out,
+{
+    type Service = Out;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        (self.0)(inner)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct OptionalLayer<L> {
+    inner: Option<L>,
+}
+
+impl<S, L> Layer<S> for OptionalLayer<L>
+where
+    L: Layer<S>,
+{
+    type Service = Either<L::Service, S>;
+
+    fn layer(&self, s: S) -> Self::Service {
+        if let Some(inner) = &self.inner {
+            Either::A(inner.layer(s))
+        } else {
+            Either::B(s)
+        }
+    }
+}

tonic/src/transport/service/mod.rs

@@ -4,6 +4,7 @@ mod connect;
 mod connector;
 mod discover;
 mod io;
+mod layer;
 
 pub(crate) use self::add_origin::AddOrigin;
 pub(crate) use self::boxed::BoxService;