Remove infine loop around poll

Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>

Author: zaharidichev

linkerd/proxy/api-resolve/src/resolve.rs

@@ -108,50 +108,48 @@ where
     S::Future: Send,
 {
     try_stream! {
-            let rsp = client.get(grpc::Request::new(req)).await?;
-            trace!(metadata = ?rsp.metadata());
-            let mut stream = rsp.into_inner();
-    loop {
-            while let Some(update) = stream.next().await {
-                match update?.update {
-                    Some(api::update::Update::Add(api::WeightedAddrSet {
-                        addrs,
-                        metric_labels,
-                    })) => {
-                        let addr_metas = addrs
-                            .into_iter()
-                            .filter_map(|addr| pb::to_addr_meta(addr, &metric_labels))
-                            .collect::<Vec<_>>();
-                        if !addr_metas.is_empty() {
-                            debug!(endpoints = %addr_metas.len(), "Add");
-                            yield Update::Add(addr_metas);
-                        }
-                    }
-
-                    Some(api::update::Update::Remove(api::AddrSet { addrs })) => {
-                        let sock_addrs = addrs
-                            .into_iter()
-                            .filter_map(pb::to_sock_addr)
-                            .collect::<Vec<_>>();
-                        if !sock_addrs.is_empty() {
-                            debug!(endpoints = %sock_addrs.len(), "Remove");
-                            yield Update::Remove(sock_addrs);
-                        }
+        let rsp = client.get(grpc::Request::new(req)).await?;
+        trace!(metadata = ?rsp.metadata());
+        let mut stream = rsp.into_inner();
+        while let Some(update) = stream.next().await {
+            match update?.update {
+                Some(api::update::Update::Add(api::WeightedAddrSet {
+                    addrs,
+                    metric_labels,
+                })) => {
+                    let addr_metas = addrs
+                        .into_iter()
+                        .filter_map(|addr| pb::to_addr_meta(addr, &metric_labels))
+                        .collect::<Vec<_>>();
+                    if !addr_metas.is_empty() {
+                        debug!(endpoints = %addr_metas.len(), "Add");
+                        yield Update::Add(addr_metas);
                     }
+                }
 
-                    Some(api::update::Update::NoEndpoints(api::NoEndpoints { exists })) => {
-                        info!("No endpoints");
-                        let update = if exists {
-                            Update::Empty
-                        } else {
-                            Update::DoesNotExist
-                        };
-                        yield update.into();
+                Some(api::update::Update::Remove(api::AddrSet { addrs })) => {
+                    let sock_addrs = addrs
+                        .into_iter()
+                        .filter_map(pb::to_sock_addr)
+                        .collect::<Vec<_>>();
+                    if !sock_addrs.is_empty() {
+                        debug!(endpoints = %sock_addrs.len(), "Remove");
+                        yield Update::Remove(sock_addrs);
                     }
+                }
 
-                    None => {} // continue
+                Some(api::update::Update::NoEndpoints(api::NoEndpoints { exists })) => {
+                    info!("No endpoints");
+                    let update = if exists {
+                        Update::Empty
+                    } else {
+                        Update::DoesNotExist
+                    };
+                    yield update.into();
                 }
+
+                None => {} // continue
             }
         }
-        }
+    }
 }

linkerd/proxy/discover/src/make_endpoint.rs

@@ -158,8 +158,10 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Change<D::Key, E::Response>, Error>>> {
-        if let Poll::Ready(key) = self.poll_removals(cx) {
-            return Poll::Ready(Some(Ok(Change::Remove(key?))));
+        if let Poll::Ready(result) = self.poll_removals(cx) {
+            if let Some(key) = result? {
+                return Poll::Ready(Some(Ok(Change::Remove(key))));
+            }
         }
 
         if let Poll::Ready(Some(res)) = self.project().make_futures.poll_next(cx) {
@@ -182,21 +184,21 @@ where
     fn poll_removals(
         self: &mut Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Result<D::Key, Error>> {
+    ) -> Poll<Result<Option<D::Key>, Error>> {
         loop {
             let mut this = self.as_mut().project();
             if let Some(key) = this.pending_removals.pop() {
                 this.make_futures.remove(&key);
-                return Poll::Ready(Ok(key));
+                return Poll::Ready(Ok(Some(key)));
             }
 
             // Before polling the resolution, where we could potentially receive
             // an `Add`, poll_ready to ensure that `make` is ready to build new
             // services. Don't process any updates until we can do so.
             ready!(this.make_endpoint.poll_ready(cx)).map_err(Into::into)?;
 
-            if let Some(change) = ready!(this.discover.poll_discover(cx)) {
-                match change.map_err(Into::into)? {
+            match ready!(this.discover.poll_discover(cx)) {
+                Some(change) => match change.map_err(Into::into)? {
                     Change::Insert(key, target) => {
                         // Start building the service and continue. If a pending
                         // service exists for this addr, it will be canceled.
@@ -206,7 +208,9 @@ where
                     Change::Remove(key) => {
                         this.pending_removals.push(key);
                     }
-                }
+                },
+
+                None => return Poll::Ready(Ok(None)),
             }
         }
     }

linkerd/proxy/resolve/src/recover.rs

@@ -214,7 +214,12 @@ where
                                 backoff: None,
                             }
                         }
-                        None => return Poll::Ready(None),
+                        None => {
+                            this.inner.state = State::Recover {
+                                error: Some(Eos(()).into()),
+                                backoff: None,
+                            }
+                        }
                     }
                 }
                 // XXX(eliza): note that this match was originally an `if let`,