diff --git a/tonic-examples/src/helloworld/client.rs b/tonic-examples/src/helloworld/client.rs index 816f51f..42cdb77 100644 --- a/tonic-examples/src/helloworld/client.rs +++ b/tonic-examples/src/helloworld/client.rs @@ -9,7 +9,7 @@ pub mod hello_world { async fn main() -> Result<(), Box> { let origin = http::Uri::from_static("http://[::1]:50051"); - let svc = Client::connect(origin).await?; + let svc = Client::connect(origin)?; let mut client = hello_world::GreeterClient::new(svc); diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index 7405de3..34d267c 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/tonic-examples/src/routeguide/client.rs @@ -1,7 +1,4 @@ use futures::TryStreamExt; -use hyper::client::conn::Builder; -use hyper::client::connect::HttpConnector; -use hyper::client::service::{Connect, MakeService}; use route_guide::{Point, RouteNote}; use std::time::{Duration, Instant}; use tokio::timer::Interval; @@ -16,7 +13,7 @@ mod route_guide { async fn main() -> Result<(), Box> { let origin = http::Uri::from_static("http://[::1]:10000"); - let svc = Client::connect(origin).await?; + let svc = Client::connect(origin)?; let mut client = route_guide::RouteGuideClient::new(svc); let start = Instant::now(); diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs index 3aad2e8..0cece99 100644 --- a/tonic-interop/src/client.rs +++ b/tonic-interop/src/client.rs @@ -25,7 +25,7 @@ const SPECIAL_TEST_STATUS_MESSAGE: &'static str = pub async fn create(addr: SocketAddr) -> Result> { let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - let svc = Client::connect(origin).await?; + let svc = Client::connect(origin)?; Ok(TestServiceClient::new(svc)) } @@ -35,7 +35,7 @@ pub async fn create_unimplemented( ) -> Result> { let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - let svc = Client::connect(origin).await?; + let svc = Client::connect(origin)?; Ok(UnimplementedServiceClient::new(svc)) } diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 356c610..4baf9a3 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -24,6 +24,9 @@ http-body = "0.2.0-alpha.1" pin-project = "0.4.0-alpha.2" hyper = { git = "https://github.com/hyperium/hyper", optional = true} +# tower +tower-reconnect = { path = "../../tower/tower-reconnect", optional = true } + [features] default = ["transport"] -transport = ["hyper"] +transport = ["hyper", "tower-reconnect"] diff --git a/tonic/src/service/boxed.rs b/tonic/src/service/boxed.rs new file mode 100644 index 0000000..73494ad --- /dev/null +++ b/tonic/src/service/boxed.rs @@ -0,0 +1,39 @@ +use std::{ + future::Future, + pin::Pin, + task::{Context, Poll}, +}; +use tower_service::Service; + +#[derive(Debug, Clone)] +pub struct BoxService { + inner: S, +} + +impl BoxService { + pub fn new(inner: S) -> Self { + Self { inner } + } +} + +impl Service for BoxService +where + S: Service, + S::Future: Send + 'static, +{ + type Response = S::Response; + type Error = S::Error; + + // type Future = BoxFuture<'static, Result, Self::Error>>; + type Future = + Pin> + Send + 'static>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, request: Request) -> Self::Future { + let fut = self.inner.call(request); + Box::pin(fut) + } +} diff --git a/tonic/src/service/mod.rs b/tonic/src/service/mod.rs index b4d3380..a819efa 100644 --- a/tonic/src/service/mod.rs +++ b/tonic/src/service/mod.rs @@ -1,7 +1,9 @@ // TODO: make this private again pub mod add_origin; +mod boxed; pub use self::add_origin::AddOrigin; +pub use self::boxed::BoxService; use crate::body::Body; use http::{Request, Response}; diff --git a/tonic/src/transport/client.rs b/tonic/src/transport/client.rs index e46ec17..02f2b87 100644 --- a/tonic/src/transport/client.rs +++ b/tonic/src/transport/client.rs @@ -1,68 +1,62 @@ use crate::{ body::BoxBody, - service::{AddOrigin, GrpcService}, + service::{AddOrigin, BoxService, GrpcService}, }; +use futures_util::try_future::{MapErr, TryFutureExt}; use http::Uri; -use hyper::client::conn; -use hyper::{Request, Response}; -use std::task::{Context, Poll}; -use tower_service::Service; use hyper::client::conn::Builder; use hyper::client::connect::HttpConnector; -use hyper::client::service::{Connect, MakeService}; - -//type BoxFuture<'a, T> = Pin + Send + 'a>>; -type BoxService = Box< - dyn GrpcService< - BoxBody, - ResponseBody = hyper::Body, - Error = hyper::Error, - Future = conn::ResponseFuture, //BoxFuture<'static, Result, hyper::Error>>, - > + Send - + 'static, ->; +use hyper::client::service::Connect; +use hyper::{Request, Response}; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +type BoxFuture<'a, T> = Pin + Send + 'a>>; // #[derive/(Clone)] pub struct Client { - svc: BoxService, + svc: Box< + dyn GrpcService< + BoxBody, + ResponseBody = hyper::Body, + Error = crate::Error, + Future = BoxFuture<'static, Result, crate::Error>>, + > + Send + + 'static, + >, } impl Client { - pub async fn connect(addr: Uri) -> Result { + pub fn connect(addr: Uri) -> Result { let settings = Builder::new().http2_only(true).clone(); - let mut maker = Connect::new(HttpConnector::new(), settings); + let maker = Connect::new(HttpConnector::new(), settings); + let svc = tower_reconnect::Reconnect::new(maker, addr.clone()); - maker.make_service(addr.clone()).await.map(|svc| Self::new(addr, svc)) - } + let svc = AddOrigin::new(svc, addr); + let svc = BoxService::new(svc); - fn new(addr: Uri, service: S) -> Self - where - S: Service< - Request, - Response = Response, - Error = hyper::Error, - Future = conn::ResponseFuture, //BoxFuture<'static, Result, hyper::Error>>, - > + Send - + 'static, - { - let svc = AddOrigin::new(service, addr); - - Self { svc: Box::new(svc) } + Ok(Self { svc: Box::new(svc) }) } } impl GrpcService for Client { type ResponseBody = hyper::Body; - type Error = hyper::Error; + type Error = super::Error; - // type Future = BoxFuture<'static, Result, Self::Error>>; - type Future = conn::ResponseFuture; + type Future = MapErr< + BoxFuture<'static, Result, crate::Error>>, + fn(crate::Error) -> super::Error, + >; fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - self.svc.poll_ready(cx) + self.svc + .poll_ready(cx) + .map_err(|e| super::Error::from((super::ErrorKind::Client, e))) } fn call(&mut self, request: Request) -> Self::Future { - self.svc.call(request) + self.svc + .call(request) + .map_err(|e| super::Error::from((super::ErrorKind::Client, e))) } } diff --git a/tonic/src/transport/mod.rs b/tonic/src/transport/mod.rs index f535227..a7790cb 100644 --- a/tonic/src/transport/mod.rs +++ b/tonic/src/transport/mod.rs @@ -1,3 +1,69 @@ mod client; pub use self::client::Client; + +use std::{error, fmt}; + +pub struct Error { + kind: ErrorKind, + source: Option, +} + +#[derive(Debug)] +pub(crate) enum ErrorKind { + Client, + // Server, +} + +impl From for Error { + fn from(t: ErrorKind) -> Self { + Self { + kind: t, + source: None, + } + } +} + +impl From<(ErrorKind, crate::Error)> for Error { + fn from(t: (ErrorKind, crate::Error)) -> Self { + Self { + kind: t.0, + source: Some(t.1), + } + } +} + +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + let mut f = f.debug_tuple("Error"); + f.field(&self.kind); + if let Some(source) = &self.source { + f.field(source); + } + f.finish() + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + if let Some(source) = &self.source { + write!(f, "{}: {}", self.kind, source) + } else { + write!(f, "{}", self.kind) + } + } +} + +impl error::Error for Error { + fn source(&self) -> Option<&(dyn error::Error + 'static)> { + self.source + .as_ref() + .map(|e| &**e as &(dyn error::Error + 'static)) + } +} + +impl fmt::Display for ErrorKind { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "{:?}", self) + } +}