feat(transport): Add Endpoint::connect_lazy method (#392)

Fixes #167

tonic/src/transport/channel/endpoint.rs

@@ -214,6 +214,25 @@ impl Endpoint {
         Channel::connect(connector, self.clone()).await
     }
 
+    /// Create a channel from this config.
+    ///
+    /// The channel returned by this method does not attempt to connect to the endpoint until first
+    /// use.
+    pub fn connect_lazy(&self) -> Result<Channel, Error> {
+        let mut http = hyper::client::connect::HttpConnector::new();
+        http.enforce_http(false);
+        http.set_nodelay(self.tcp_nodelay);
+        http.set_keepalive(self.tcp_keepalive);
+
+        #[cfg(feature = "tls")]
+        let connector = service::connector(http, self.tls.clone());
+
+        #[cfg(not(feature = "tls"))]
+        let connector = service::connector(http);
+
+        Channel::new(connector, self.clone())
+    }
+
     /// Connect with a custom connector.
     pub async fn connect_with_connector<C>(&self, connector: C) -> Result<Channel, Error>
     where

tonic/src/transport/channel/mod.rs

@@ -130,6 +130,21 @@ impl Channel {
         (Self::balance(list, DEFAULT_BUFFER_SIZE), tx)
     }
 
+    pub(crate) fn new<C>(connector: C, endpoint: Endpoint) -> Result<Self, super::Error>
+    where
+        C: Service<Uri> + Send + 'static,
+        C::Error: Into<crate::Error> + Send,
+        C::Future: Unpin + Send,
+        C::Response: AsyncRead + AsyncWrite + HyperConnection + Unpin + Send + 'static,
+    {
+        let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE);
+
+        let svc = Connection::new(connector, endpoint).map_err(super::Error::from_source)?;
+        let svc = Buffer::new(Either::A(svc), buffer_size);
+
+        Ok(Channel { svc })
+    }
+
     pub(crate) async fn connect<C>(connector: C, endpoint: Endpoint) -> Result<Self, super::Error>
     where
         C: Service<Uri> + Send + 'static,
@@ -139,10 +154,9 @@ impl Channel {
     {
         let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE);
 
-        let svc = Connection::new(connector, endpoint)
+        let svc = Connection::connect(connector, endpoint)
             .await
-            .map_err(|e| super::Error::from_source(e))?;
+            .map_err(super::Error::from_source)?;
-
         let svc = Buffer::new(Either::A(svc), buffer_size);
 
         Ok(Channel { svc })

tonic/src/transport/service/connection.rs

@@ -16,7 +16,7 @@ use tower::{
     limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
     timeout::TimeoutLayer,
     util::BoxService,
-    ServiceBuilder,
+    ServiceBuilder, ServiceExt,
 };
 use tower_load::Load;
 use tower_service::Service;
@@ -29,7 +29,7 @@ pub(crate) struct Connection {
 }
 
 impl Connection {
-    pub(crate) async fn new<C>(connector: C, endpoint: Endpoint) -> Result<Self, crate::Error>
+    pub(crate) fn new<C>(connector: C, endpoint: Endpoint) -> Result<Self, crate::Error>
     where
         C: Service<Uri> + Send + 'static,
         C::Error: Into<crate::Error> + Send,
@@ -60,9 +60,8 @@ impl Connection {
             .optional_layer(endpoint.rate_limit.map(|(l, d)| RateLimitLayer::new(l, d)))
             .into_inner();
 
-        let mut connector = HyperConnect::new(connector, settings);
+        let connector = HyperConnect::new(connector, settings);
-        let initial_conn = connector.call(endpoint.uri.clone()).await?;
+        let conn = Reconnect::new(connector, endpoint.uri.clone());
-        let conn = Reconnect::new(initial_conn, connector, endpoint.uri.clone());
 
         let inner = stack.layer(conn);
 
@@ -70,6 +69,16 @@ impl Connection {
             inner: BoxService::new(inner),
         })
     }
+
+    pub(crate) async fn connect<C>(connector: C, endpoint: Endpoint) -> Result<Self, crate::Error>
+    where
+        C: Service<Uri> + Send + 'static,
+        C::Error: Into<crate::Error> + Send,
+        C::Future: Unpin + Send,
+        C::Response: AsyncRead + AsyncWrite + HyperConnection + Unpin + Send + 'static,
+    {
+        Self::new(connector, endpoint)?.ready_oneshot().await
+    }
 }
 
 impl Service<Request> for Connection {

tonic/src/transport/service/discover.rs

@@ -64,7 +64,7 @@ impl<K: Hash + Eq + Clone> Discover for DynamicServiceStream<K> {
 
                         #[cfg(not(feature = "tls"))]
                         let connector = service::connector(http);
-                        let fut = Connection::new(connector, endpoint);
+                        let fut = Connection::connect(connector, endpoint);
                         self.connecting = Some((k, Box::pin(fut)));
                         continue;
                     }

tonic/src/transport/service/reconnect.rs

@@ -31,16 +31,10 @@ 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
+    pub(crate) fn new(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),
+            state: State::Idle,
             target,
             error: None,
         }