diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index 83ff8ff..6aae21d 100644 --- a/tonic-interop/src/bin/client.rs +++ b/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> { let test_cases = matches.test_case; - let addr = "localhost:10000"; - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); + let mut endpoint = Endpoint::from_static("http://localhost:10000") + .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())) - } else { - Endpoint::from(origin) - }; + endpoint.tls_cert(ca, Some("foo.test.google.fr".into())); + } - 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); diff --git a/tonic-interop/test.sh b/tonic-interop/test.sh index b1cee0f..8c58cdc 100755 --- a/tonic-interop/test.sh +++ b/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 diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 393d56a..03e838c 100644 --- a/tonic/Cargo.toml +++ b/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-buffer = { 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-discover = { version = "0.3.0-alpha.1", optional = true } +tower-reconnect = { 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 } # 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"] diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel.rs index d45ef0b..95c3b32 100644 --- a/tonic/src/transport/channel.rs +++ b/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 + Send + 'a>>; diff --git a/tonic/src/transport/endpoint.rs b/tonic/src/transport/endpoint.rs index 9a0020e..d6c20c6 100644 --- a/tonic/src/transport/endpoint.rs +++ b/tonic/src/transport/endpoint.rs @@ -1,37 +1,58 @@ -use super::tls::Cert; -use http::uri::Uri; +use super::{channel::Channel, tls::Cert}; +use bytes::Bytes; +use http::uri::{InvalidUriBytes, Uri}; +use std::time::Duration; #[derive(Debug, Clone)] pub struct Endpoint { - uri: Uri, - cert: Option, + pub(super) uri: Uri, + pub(super) timeout: Option, + pub(super) concurrency_limit: Option, + pub(super) cert: Option, } impl Endpoint { - pub fn with_pem(uri: Uri, ca: Vec, domain: Option) -> Self { - let domain = domain.unwrap_or_else(|| uri.clone().to_string()); - - Self { - uri, - cert: Some(Cert { - ca, - domain, - key: None, - }), - } + pub fn from_static(s: &'static str) -> Self { + let uri = Uri::from_static(s); + Self::from(uri) } - pub(crate) fn uri(&self) -> &Uri { - &self.uri + pub fn from_shared(s: impl Into) -> Result { + let uri = Uri::from_shared(s.into())?; + Ok(Self::from(uri)) } - pub(crate) fn take_cert(&mut self) -> Option { - self.cert.take() + pub fn timeout(&mut self, dur: Duration) -> &mut Self { + 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, domain: Option) -> &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::builder().connect(self.clone()) } } impl From for Endpoint { fn from(uri: Uri) -> Self { - Self { uri, cert: None } + Self { + uri, + concurrency_limit: None, + timeout: None, + cert: None, + } } } diff --git a/tonic/src/transport/service/connect.rs b/tonic/src/transport/service/connect.rs index e39b8e0..abc450b 100644 --- a/tonic/src/transport/service/connect.rs +++ b/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; +type Response = http::Response; + pub struct Connection { - inner: AddOrigin, Uri>>, + // inner: AddOrigin, Uri>>, + inner: BoxService, } impl Connection { pub fn new(mut endpoint: Endpoint) -> Result { - 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> for Connection { - type Response = Response; +impl Service for Connection { + type Response = Response; type Error = crate::Error; type Future = @@ -41,10 +61,8 @@ impl Service> for Connection { Service::poll_ready(&mut self.inner, cx).map_err(Into::into) } - fn call(&mut self, req: Request) -> Self::Future { - let fut = self.inner.call(req); - // TODO: we dont need to box here if we have too - Box::pin(fut) + fn call(&mut self, req: Request) -> Self::Future { + self.inner.call(req) } } diff --git a/tonic/src/transport/service/discover.rs b/tonic/src/transport/service/discover.rs index a59c8ae..d3f77e1 100644 --- a/tonic/src/transport/service/discover.rs +++ b/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 { diff --git a/tonic/src/transport/service/layer.rs b/tonic/src/transport/service/layer.rs new file mode 100644 index 0000000..5fe0402 --- /dev/null +++ b/tonic/src/transport/service/layer.rs @@ -0,0 +1,76 @@ +use tower::{ + layer::{util::Stack, Layer}, + util::Either, + ServiceBuilder, +}; +pub(crate) trait ServiceBuilderExt { + fn layer_fn Out, S, Out>(self, f: F) -> ServiceBuilder, L>>; + + fn optional_layer_fn Out, S, Out>( + self, + f: Option, + ) -> ServiceBuilder>, L>>; + + fn optional_layer(self, l: Option) -> ServiceBuilder, L>>; +} + +impl ServiceBuilderExt for ServiceBuilder { + fn layer_fn(self, f: F) -> ServiceBuilder, L>> + where + F: Fn(S) -> Out, + { + self.layer(LayerFn(f)) + } + + fn optional_layer_fn( + self, + f: Option, + ) -> ServiceBuilder>, L>> + where + F: Fn(S) -> Out, + { + let layer = OptionalLayer { + inner: f.map(|f| LayerFn(f)), + }; + + self.layer(layer) + } + + fn optional_layer(self, inner: Option) -> ServiceBuilder, L>> { + self.layer(OptionalLayer { inner }) + } +} + +#[derive(Clone, Copy, Debug)] +pub(crate) struct LayerFn(F); + +impl Layer for LayerFn +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 { + inner: Option, +} + +impl Layer for OptionalLayer +where + L: Layer, +{ + type Service = Either; + + fn layer(&self, s: S) -> Self::Service { + if let Some(inner) = &self.inner { + Either::A(inner.layer(s)) + } else { + Either::B(s) + } + } +} diff --git a/tonic/src/transport/service/mod.rs b/tonic/src/transport/service/mod.rs index b730bc3..fa89055 100644 --- a/tonic/src/transport/service/mod.rs +++ b/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;