feat(transport): Change channel connect to be async (#107)

This makes it so you can check if the initial connection
is established. Before this we used reconnect which would
lazily attempt to connect. So if you were trying to connect
to a non existant Server you wouldn't find out until after
you attempted your first RPC. This simplifies everything
by allowing you connect before creating the RPC client.

BREAKING CHANGE: `Endpoint::channel` was removed in favor of
an async `Endpoint::connect`.

tonic-build/src/client.rs

@@ -54,12 +54,13 @@ fn generate_connect(service_ident: &syn::Ident) -> TokenStream {
     quote! {
         impl #service_ident<tonic::transport::Channel> {
             /// Attempt to create a new client by connecting to a given endpoint.
-            pub fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+            pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
             where
                 D: std::convert::TryInto<tonic::transport::Endpoint>,
                 D::Error: Into<StdError>,
             {
-                tonic::transport::Endpoint::new(dst).map(|c| Self::new(c.channel()))
+                let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+                Ok(Self::new(conn))
             }
         }
     }

tonic-examples/src/authentication/client.rs

@@ -15,7 +15,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 HeaderValue::from_static("Bearer some-secret-token"),
             );
         })
-        .channel();
+        .connect()
+        .await?;
 
     let mut client = EchoClient::new(channel);
 

tonic-examples/src/gcp/client.rs

@@ -37,7 +37,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             headers.insert("authorization", header_value.clone());
         })
         .tls_config(&tls_config)
-        .channel();
+        .connect()
+        .await?;
 
     let mut service = PublisherClient::new(channel);
 

tonic-examples/src/helloworld/client.rs

@@ -6,7 +6,7 @@ use hello_world::{client::GreeterClient, HelloRequest};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = GreeterClient::connect("http://[::1]:50051")?;
+    let mut client = GreeterClient::connect("http://[::1]:50051").await?;
 
     let request = tonic::Request::new(HelloRequest {
         name: "Tonic".into(),

tonic-examples/src/multiplex/client.rs

@@ -12,7 +12,9 @@ use tonic::transport::Endpoint;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let channel = Endpoint::from_static("http://[::1]:50051").channel();
+    let channel = Endpoint::from_static("http://[::1]:50051")
+        .connect()
+        .await?;
 
     let mut greeter_client = GreeterClient::new(channel.clone());
     let mut echo_client = EchoClient::new(channel);

tonic-examples/src/routeguide/client.rs

@@ -91,7 +91,7 @@ async fn run_route_chat(client: &mut RouteGuideClient<Channel>) -> Result<(), Bo
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = RouteGuideClient::connect("http://[::1]:10000")?;
+    let mut client = RouteGuideClient::connect("http://[::1]:10000").await?;
 
     println!("*** SIMPLE RPC ***");
     let response = client

tonic-examples/src/tls/client.rs

@@ -17,7 +17,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let channel = Channel::from_static("http://[::1]:50051")
         .tls_config(&tls)
-        .channel();
+        .connect()
+        .await?;
 
     let mut client = EchoClient::new(channel);
     let request = tonic::Request::new(EchoRequest {

tonic-examples/src/tls_client_auth/client.rs

@@ -21,8 +21,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let channel = Channel::from_static("http://[::1]:50051")
         .tls_config(&tls)
-        .clone()
+        .connect()
-        .channel();
+        .await?;
 
     let mut client = EchoClient::new(channel);
 

tonic-interop/src/bin/client.rs

@@ -41,7 +41,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         );
     }
 
-    let channel = endpoint.channel();
+    let channel = endpoint.connect().await?;
 
     let mut client = client::TestClient::new(channel.clone());
     let mut unimplemented_client = client::UnimplementedClient::new(channel);

tonic/Cargo.toml

@@ -29,7 +29,6 @@ transport = [
     "hyper",
     "tokio",
     "tower",
-    "tower-reconnect",
     "tower-balance",
     "tower-load",
 ]
@@ -68,7 +67,6 @@ hyper = { version = "=0.13.0-alpha.4", features = ["unstable-stream"], optional
 tokio = { version = "=0.2.0-alpha.6", default-features = false, features = ["tcp"], optional = true }
 tower = { version = "=0.3.0-alpha.2", optional = true}
 tower-make = "=0.3.0-alpha.2a"
-tower-reconnect = { version = "=0.3.0-alpha.2", optional = true }
 tower-balance =  { version = "=0.3.0-alpha.2", optional = true }
 tower-load = { version = "=0.3.0-alpha.2", optional = true }
 

tonic/benches/benchmarks/compiled_protos/helloworld.rs

@@ -20,12 +20,13 @@ pub mod client {
     }
     impl GreeterClient<tonic::transport::Channel> {
         #[doc = r" Attempt to create a new client by connecting to a given endpoint."]
-        pub fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
         where
             D: std::convert::TryInto<tonic::transport::Endpoint>,
             D::Error: Into<StdError>,
         {
-            tonic::transport::Endpoint::new(dst).map(|c| Self::new(c.channel()))
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
         }
     }
     impl<T> GreeterClient<T>

tonic/src/transport/channel.rs

@@ -97,18 +97,20 @@ impl Channel {
         Self::balance(discover, buffer_size, interceptor_headers)
     }
 
-    pub(crate) fn connect(endpoint: Endpoint) -> Self {
+    pub(crate) async fn connect(endpoint: Endpoint) -> Result<Self, super::Error> {
         let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE);
         let interceptor_headers = endpoint.interceptor_headers.clone();
 
-        let svc = Connection::new(endpoint);
+        let svc = Connection::new(endpoint)
+            .await
+            .map_err(|e| super::Error::from_source(super::ErrorKind::Client, e))?;
 
         let svc = Buffer::new(Either::A(svc), buffer_size);
 
-        Channel {
+        Ok(Channel {
             svc,
             interceptor_headers,
-        }
+        })
     }
 
     pub(crate) fn balance<D>(

tonic/src/transport/endpoint.rs

@@ -142,8 +142,8 @@ impl Endpoint {
     }
 
     /// Create a channel from this config.
-    pub fn channel(&self) -> Channel {
+    pub async fn connect(&self) -> Result<Channel, super::Error> {
-        Channel::connect(self.clone())
+        Channel::connect(self.clone()).await
     }
 }
 

tonic/src/transport/mod.rs

@@ -35,7 +35,8 @@
 //!     .timeout(Duration::from_secs(5))
 //!     .rate_limit(5, Duration::from_secs(1))
 //!     .concurrency_limit(256)
-//!     .channel();
+//!     .connect()
+//!     .await?;
 //!
 //! channel.call(Request::new(BoxBody::empty())).await?;
 //! # Ok(())

tonic/src/transport/service/connection.rs

@@ -1,4 +1,4 @@
-use super::{connector, layer::ServiceBuilderExt, AddOrigin};
+use super::{connector, layer::ServiceBuilderExt, reconnect::Reconnect, AddOrigin};
 use crate::{body::BoxBody, transport::Endpoint};
 use hyper::client::conn::Builder;
 use hyper::client::service::Connect as HyperConnect;
@@ -16,7 +16,6 @@ use tower::{
     ServiceBuilder,
 };
 use tower_load::Load;
-use tower_reconnect::Reconnect;
 use tower_service::Service;
 
 pub(crate) type Request = http::Request<BoxBody>;
@@ -27,7 +26,7 @@ pub(crate) struct Connection {
 }
 
 impl Connection {
-    pub(crate) fn new(endpoint: Endpoint) -> Self {
+    pub(crate) async fn new(endpoint: Endpoint) -> Result<Self, crate::Error> {
         #[cfg(feature = "tls")]
         let connector = connector(endpoint.tls.clone());
 
@@ -47,13 +46,15 @@ impl Connection {
             .optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d)))
             .into_inner();
 
-        let conn = Reconnect::new(HyperConnect::new(connector, settings), endpoint.uri.clone());
+        let mut connector = HyperConnect::new(connector, settings);
+        let initial_conn = connector.call(endpoint.uri.clone()).await?;
+        let conn = Reconnect::new(initial_conn, connector, endpoint.uri.clone());
 
         let inner = stack.layer(conn);
 
-        Self {
+        Ok(Self {
             inner: BoxService::new(inner),
-        }
+        })
     }
 }
 

tonic/src/transport/service/discover.rs

@@ -1,13 +1,18 @@
 use super::connection::Connection;
 use crate::transport::Endpoint;
-use std::collections::VecDeque;
+use std::{
-use std::pin::Pin;
+    collections::VecDeque,
-use std::task::{Context, Poll};
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
 use tower::discover::{Change, Discover};
 
-#[derive(Debug)]
 pub(crate) struct ServiceList {
     list: VecDeque<Endpoint>,
+    connecting:
+        Option<Pin<Box<dyn Future<Output = Result<Connection, crate::Error>> + Send + 'static>>>,
     i: usize,
 }
 
@@ -15,6 +20,7 @@ impl ServiceList {
     pub(crate) fn new(list: Vec<Endpoint>) -> Self {
         Self {
             list: list.into(),
+            connecting: None,
             i: 0,
         }
     }
@@ -27,19 +33,34 @@ impl Discover for ServiceList {
 
     fn poll_discover(
         mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
+        cx: &mut Context<'_>,
     ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
-        match self.list.pop_front() {
+        loop {
-            Some(endpoint) => {
+            if let Some(connecting) = &mut self.connecting {
+                let svc = futures_core::ready!(Pin::new(connecting).poll(cx))?;
+
                 let i = self.i;
                 self.i += 1;
 
-                let svc = Connection::new(endpoint);
                 let change = Ok(Change::Insert(i, svc));
 
-                Poll::Ready(change)
+                return Poll::Ready(change);
+            }
+
+            if let Some(endpoint) = self.list.pop_front() {
+                let fut = Connection::new(endpoint);
+                self.connecting = Some(Box::pin(fut));
+            } else {
+                return Poll::Pending;
             }
-            None => Poll::Pending,
         }
     }
 }
+
+impl fmt::Debug for ServiceList {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ServiceList")
+            .field("list", &self.list)
+            .finish()
+    }
+}

tonic/src/transport/service/mod.rs

@@ -5,6 +5,7 @@ mod discover;
 mod either;
 mod io;
 mod layer;
+mod reconnect;
 mod router;
 #[cfg(feature = "tls")]
 mod tls;

tonic/src/transport/service/reconnect.rs

@@ -0,0 +1,174 @@
+use crate::Error;
+use pin_project::pin_project;
+use std::fmt;
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tower_make::MakeService;
+use tower_service::Service;
+use tracing::trace;
+
+pub(crate) struct Reconnect<M, Target>
+where
+    M: Service<Target>,
+{
+    mk_service: M,
+    state: State<M::Future, M::Response>,
+    target: Target,
+}
+
+#[derive(Debug)]
+enum State<F, S> {
+    Idle,
+    Connecting(F),
+    Connected(S),
+}
+
+impl<M, Target> Reconnect<M, Target>
+where
+    M: Service<Target>,
+{
+    pub(crate) fn new<S, Request>(initial_connection: S, mk_service: M, target: Target) -> Self
+    where
+        M: Service<Target, Response = S>,
+        S: Service<Request>,
+        Error: From<M::Error> + From<S::Error>,
+        Target: Clone,
+    {
+        Reconnect {
+            mk_service,
+            state: State::Connected(initial_connection),
+            target,
+        }
+    }
+}
+
+impl<M, Target, S, Request> Service<Request> for Reconnect<M, Target>
+where
+    M: Service<Target, Response = S>,
+    S: Service<Request>,
+    M::Future: Unpin,
+    Error: From<M::Error> + From<S::Error>,
+    Target: Clone,
+{
+    type Response = S::Response;
+    type Error = Error;
+    type Future = ResponseFuture<S::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        let ret;
+        let mut state;
+
+        loop {
+            match self.state {
+                State::Idle => {
+                    trace!("poll_ready; idle");
+                    match self.mk_service.poll_ready(cx) {
+                        Poll::Ready(r) => r?,
+                        Poll::Pending => {
+                            trace!("poll_ready; MakeService not ready");
+                            return Poll::Pending;
+                        }
+                    }
+
+                    let fut = self.mk_service.make_service(self.target.clone());
+                    self.state = State::Connecting(fut);
+                    continue;
+                }
+                State::Connecting(ref mut f) => {
+                    trace!("poll_ready; connecting");
+                    match Pin::new(f).poll(cx) {
+                        Poll::Ready(Ok(service)) => {
+                            state = State::Connected(service);
+                        }
+                        Poll::Pending => {
+                            trace!("poll_ready; not ready");
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(Err(e)) => {
+                            trace!("poll_ready; error");
+                            state = State::Idle;
+                            ret = Err(e.into());
+                            break;
+                        }
+                    }
+                }
+                State::Connected(ref mut inner) => {
+                    trace!("poll_ready; connected");
+                    match inner.poll_ready(cx) {
+                        Poll::Ready(Ok(())) => {
+                            trace!("poll_ready; ready");
+                            return Poll::Ready(Ok(()));
+                        }
+                        Poll::Pending => {
+                            trace!("poll_ready; not ready");
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(Err(_)) => {
+                            trace!("poll_ready; error");
+                            state = State::Idle;
+                        }
+                    }
+                }
+            }
+
+            self.state = state;
+        }
+
+        self.state = state;
+        Poll::Ready(ret)
+    }
+
+    fn call(&mut self, request: Request) -> Self::Future {
+        let service = match self.state {
+            State::Connected(ref mut service) => service,
+            _ => panic!("service not ready; poll_ready must be called first"),
+        };
+
+        let fut = service.call(request);
+        ResponseFuture::new(fut)
+    }
+}
+
+impl<M, Target> fmt::Debug for Reconnect<M, Target>
+where
+    M: Service<Target> + fmt::Debug,
+    M::Future: fmt::Debug,
+    M::Response: fmt::Debug,
+    Target: fmt::Debug,
+{
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("Reconnect")
+            .field("mk_service", &self.mk_service)
+            .field("state", &self.state)
+            .field("target", &self.target)
+            .finish()
+    }
+}
+
+#[pin_project]
+#[derive(Debug)]
+pub(crate) struct ResponseFuture<F> {
+    #[pin]
+    inner: F,
+}
+
+impl<F> ResponseFuture<F> {
+    pub(crate) fn new(inner: F) -> Self {
+        ResponseFuture { inner }
+    }
+}
+
+impl<F, T, E> Future for ResponseFuture<F>
+where
+    F: Future<Output = Result<T, E>>,
+    E: Into<Error>,
+{
+    type Output = Result<T, Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx).map_err(Into::into)
+    }
+}