util: Fix call_all hang when stream is pending (#656)

Currently `call_all` will hang in a busy loop if called when the input
stream is pending.

tower/src/util/call_all/common.rs

@@ -99,20 +99,18 @@ where
                 .expect("Using CallAll after extracing inner Service");
             ready!(svc.poll_ready(cx)).map_err(Into::into)?;
 
-            // If it is, gather the next request (if there is one)
+            // If it is, gather the next request (if there is one), or return `Pending` if the
-            match this.stream.as_mut().poll_next(cx) {
+            // stream is not ready.
-                Poll::Ready(r) => match r {
+            // TODO: We probably want to "release" the slot we reserved in Svc if the
-                    Some(req) => {
+            // stream returns `Pending`. It may be a while until we get around to actually
-                        this.queue.push(svc.call(req));
+            // using it.
-                    }
+            match ready!(this.stream.as_mut().poll_next(cx)) {
-                    None => {
+                Some(req) => {
-                        // We're all done once any outstanding requests have completed
+                    this.queue.push(svc.call(req));
-                        *this.eof = true;
+                }
-                    }
+                None => {
-                },
+                    // We're all done once any outstanding requests have completed
-                Poll::Pending => {
+                    *this.eof = true;
-                    // TODO: We probably want to "release" the slot we reserved in Svc here.
-                    // It may be a while until we get around to actually using it.
                 }
             }
         }

tower/tests/util/call_all.rs

@@ -143,3 +143,24 @@ async fn unordered() {
         .unwrap();
     assert!(v.is_none());
 }
+
+#[tokio::test]
+async fn pending() {
+    let _t = support::trace_init();
+
+    let (mock, mut handle) = mock::pair::<_, &'static str>();
+
+    let mut task = task::spawn(());
+
+    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+    let ca = mock.call_all(support::IntoStream::new(rx));
+    pin_mut!(ca);
+
+    assert_pending!(task.enter(|cx, _| ca.as_mut().poll_next(cx)));
+    tx.send("req").unwrap();
+    assert_pending!(task.enter(|cx, _| ca.as_mut().poll_next(cx)));
+    assert_request_eq!(handle, "req").send_response("res");
+    let res = assert_ready!(task.enter(|cx, _| ca.as_mut().poll_next(cx)));
+    assert_eq!(res.transpose().unwrap(), Some("res"));
+    assert_pending!(task.enter(|cx, _| ca.as_mut().poll_next(cx)));
+}