feat(transport): Connect lazily in the load balanced channel (#493)

tonic/src/transport/service/discover.rs

@@ -1,9 +1,8 @@
-use super::super::{service, BoxFuture};
+use super::super::service;
 use super::connection::Connection;
 use crate::transport::Endpoint;
 
 use std::{
-    future::Future,
     hash::Hash,
     pin::Pin,
     task::{Context, Poll},
@@ -16,15 +15,11 @@ type DiscoverResult<K, S, E> = Result<Change<K, S>, E>;
 
 pub(crate) struct DynamicServiceStream<K: Hash + Eq + Clone> {
     changes: Receiver<Change<K, Endpoint>>,
-    connecting: Option<(K, BoxFuture<Connection, crate::Error>)>,
 }
 
 impl<K: Hash + Eq + Clone> DynamicServiceStream<K> {
     pub(crate) fn new(changes: Receiver<Change<K, Endpoint>>) -> Self {
-        Self {
+        Self { changes }
-            changes,
-            connecting: None,
-        }
     }
 }
 
@@ -37,21 +32,9 @@ impl<K: Hash + Eq + Clone> Discover for DynamicServiceStream<K> {
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<DiscoverResult<Self::Key, Self::Service, Self::Error>> {
-        loop {
-            if let Some((key, connecting)) = &mut self.connecting {
-                let svc = futures_core::ready!(Pin::new(connecting).poll(cx))?;
-                let key = key.to_owned();
-                self.connecting = None;
-                let change = Ok(Change::Insert(key, svc));
-                return Poll::Ready(change);
-            };
-
         let c = &mut self.changes;
         match Pin::new(&mut *c).poll_next(cx) {
-                Poll::Pending => return Poll::Pending,
+            Poll::Pending | Poll::Ready(None) => Poll::Pending,
-                Poll::Ready(None) => {
-                    return Poll::Pending;
-                }
             Poll::Ready(Some(change)) => match change {
                 Change::Insert(k, endpoint) => {
                     let mut http = hyper::client::connect::HttpConnector::new();
@@ -63,15 +46,14 @@ 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 connection = Connection::lazy(connector, endpoint);
-                        self.connecting = Some((k, Box::pin(fut)));
+                    let change = Ok(Change::Insert(k, connection));
-                        continue;
+                    Poll::Ready(change)
                 }
-                    Change::Remove(k) => return Poll::Ready(Ok(Change::Remove(k))),
+                Change::Remove(k) => Poll::Ready(Ok(Change::Remove(k))),
             },
         }
     }
-    }
 }
 
 impl<K: Hash + Eq + Clone> Unpin for DynamicServiceStream<K> {}