Use tower-reconnect

tonic-examples/src/helloworld/client.rs

@@ -9,7 +9,7 @@ pub mod hello_world {
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     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);
 

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<dyn std::error::Error>> {
     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();

tonic-interop/src/client.rs

@@ -25,7 +25,7 @@ const SPECIAL_TEST_STATUS_MESSAGE: &'static str =
 pub async fn create(addr: SocketAddr) -> Result<TestClient, Box<dyn std::error::Error>> {
     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<UnimplementedClient, Box<dyn std::error::Error>> {
     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))
 }

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"]

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<S> {
+    inner: S,
+}
+
+impl<S> BoxService<S> {
+    pub fn new(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S, Request> Service<Request> for BoxService<S>
+where
+    S: Service<Request>,
+    S::Future: Send + 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+
+    // type Future = BoxFuture<'static, Result<Response<Self::ResponseBody>, Self::Error>>;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, request: Request) -> Self::Future {
+        let fut = self.inner.call(request);
+        Box::pin(fut)
+    }
+}

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};

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<Box<dyn Future<Output = T> + Send + 'a>>;
-type BoxService = Box<
-    dyn GrpcService<
-            BoxBody,
-            ResponseBody = hyper::Body,
-            Error = hyper::Error,
-            Future = conn::ResponseFuture, //BoxFuture<'static, Result<Response<hyper::Body>, 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<Box<dyn Future<Output = T> + Send + 'a>>;
 // #[derive/(Clone)]
 pub struct Client {
-    svc: BoxService,
+    svc: Box<
+        dyn GrpcService<
+                BoxBody,
+                ResponseBody = hyper::Body,
+                Error = crate::Error,
+                Future = BoxFuture<'static, Result<Response<hyper::Body>, crate::Error>>,
+            > + Send
+            + 'static,
+    >,
 }
 
 impl Client {
-    pub async fn connect(addr: Uri) -> Result<Self, hyper::Error> {
+    pub fn connect(addr: Uri) -> Result<Self, super::Error> {
         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<S>(addr: Uri, service: S) -> Self
-    where
-        S: Service<
-                Request<BoxBody>,
-                Response = Response<hyper::Body>,
-                Error = hyper::Error,
-                Future = conn::ResponseFuture, //BoxFuture<'static, Result<Response<hyper::Body>, hyper::Error>>,
-            > + Send
-            + 'static,
-    {
-        let svc = AddOrigin::new(service, addr);
-
-        Self { svc: Box::new(svc) }
+        Ok(Self { svc: Box::new(svc) })
     }
 }
 
 impl GrpcService<BoxBody> for Client {
     type ResponseBody = hyper::Body;
-    type Error = hyper::Error;
+    type Error = super::Error;
 
-    // type Future = BoxFuture<'static, Result<Response<Self::ResponseBody>, Self::Error>>;
-    type Future = conn::ResponseFuture;
+    type Future = MapErr<
+        BoxFuture<'static, Result<Response<Self::ResponseBody>, crate::Error>>,
+        fn(crate::Error) -> super::Error,
+    >;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        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<BoxBody>) -> Self::Future {
-        self.svc.call(request)
+        self.svc
+            .call(request)
+            .map_err(|e| super::Error::from((super::ErrorKind::Client, e)))
     }
 }

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<crate::Error>,
+}
+
+#[derive(Debug)]
+pub(crate) enum ErrorKind {
+    Client,
+    // Server,
+}
+
+impl From<ErrorKind> 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)
+    }
+}