Endpoint::lazy_connect always tries to reconnect

tonic/src/transport/channel/endpoint.rs

@@ -215,7 +215,8 @@ impl Endpoint {
         #[cfg(not(feature = "tls"))]
         let connector = service::connector(http);
 
-        Channel::connect(connector, self.clone()).await
+        let always_reconnect = false;
+        Channel::connect(connector, self.clone(), always_reconnect).await
     }
 
     /// Create a channel from this config.
@@ -234,7 +235,8 @@ impl Endpoint {
         #[cfg(not(feature = "tls"))]
         let connector = service::connector(http);
 
-        Channel::new(connector, self.clone())
+        let always_reconnect = true;
+        Channel::new(connector, self.clone(), always_reconnect)
     }
 
     /// Connect with a custom connector.
@@ -255,7 +257,8 @@ impl Endpoint {
         #[cfg(not(feature = "tls"))]
         let connector = service::connector(connector);
 
-        Channel::connect(connector, self.clone()).await
+        let always_reconnect = false;
+        Channel::connect(connector, self.clone(), always_reconnect).await
     }
 
     /// Get the endpoint uri.

tonic/src/transport/channel/mod.rs

@@ -130,7 +130,11 @@ impl Channel {
         (Self::balance(list, DEFAULT_BUFFER_SIZE), tx)
     }
 
-    pub(crate) fn new<C>(connector: C, endpoint: Endpoint) -> Result<Self, super::Error>
+    pub(crate) fn new<C>(
+        connector: C,
+        endpoint: Endpoint,
+        always_reconnect: bool,
+    ) -> Result<Self, super::Error>
     where
         C: Service<Uri> + Send + 'static,
         C::Error: Into<crate::Error> + Send,
@@ -139,13 +143,18 @@ impl Channel {
     {
         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 = Connection::new(connector, endpoint, always_reconnect)
+            .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>
+    pub(crate) async fn connect<C>(
+        connector: C,
+        endpoint: Endpoint,
+        always_reconnect: bool,
+    ) -> Result<Self, super::Error>
     where
         C: Service<Uri> + Send + 'static,
         C::Error: Into<crate::Error> + Send,
@@ -154,7 +163,7 @@ impl Channel {
     {
         let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE);
 
-        let svc = Connection::connect(connector, endpoint)
+        let svc = Connection::connect(connector, endpoint, always_reconnect)
             .await
             .map_err(super::Error::from_source)?;
         let svc = Buffer::new(Either::A(svc), buffer_size);

tonic/src/transport/service/connection.rs

@@ -29,7 +29,11 @@ pub(crate) struct Connection {
 }
 
 impl Connection {
-    pub(crate) fn new<C>(connector: C, endpoint: Endpoint) -> Result<Self, crate::Error>
+    pub(crate) fn new<C>(
+        connector: C,
+        endpoint: Endpoint,
+        always_reconnect: bool,
+    ) -> Result<Self, crate::Error>
     where
         C: Service<Uri> + Send + 'static,
         C::Error: Into<crate::Error> + Send,
@@ -61,7 +65,7 @@ impl Connection {
             .into_inner();
 
         let connector = HyperConnect::new(connector, settings);
-        let conn = Reconnect::new(connector, endpoint.uri.clone());
+        let conn = Reconnect::new(connector, endpoint.uri.clone(), always_reconnect);
 
         let inner = stack.layer(conn);
 
@@ -70,14 +74,20 @@ impl Connection {
         })
     }
 
-    pub(crate) async fn connect<C>(connector: C, endpoint: Endpoint) -> Result<Self, crate::Error>
+    pub(crate) async fn connect<C>(
+        connector: C,
+        endpoint: Endpoint,
+        always_reconnect: bool,
+    ) -> 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
+        Self::new(connector, endpoint, always_reconnect)?
+            .ready_oneshot()
+            .await
     }
 }
 

tonic/src/transport/service/discover.rs

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

tonic/src/transport/service/reconnect.rs

@@ -32,13 +32,15 @@ impl<M, Target> Reconnect<M, Target>
 where
     M: Service<Target>,
 {
-    pub(crate) fn new(mk_service: M, target: Target) -> Self {
+    pub(crate) fn new(mk_service: M, target: Target, always_reconnect: bool) -> Self {
         Reconnect {
             mk_service,
             state: State::Idle,
             target,
             error: None,
-            has_been_connected: false,
+            // If true, it makes sure that a failure (if any) on the first connection attempt
+            // will not result in an error, and we will keep trying reconnecting.
+            has_been_connected: always_reconnect,
         }
     }
 }