fix(http2): fix race, drain client dispatch reciever

src/client/dispatch.rs

@@ -197,13 +197,11 @@ impl<T, U> Receiver<T, U> {
         }
     }
 
-    #[cfg(feature = "http1")]
     pub(crate) fn close(&mut self) {
         self.taker.cancel();
         self.inner.close();
     }
 
-    #[cfg(feature = "http1")]
     pub(crate) fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll<()> {
         self.close();
         // A send that started before close may still be publishing its envelope.

src/proto/h2/client.rs

@@ -201,6 +201,7 @@ where
         h2_tx,
         req_rx,
         fut_ctx: None,
+        closing: None,
         marker: PhantomData,
     })
 }
@@ -434,6 +435,7 @@ where
     h2_tx: SendRequest<SendBuf<B::Data>>,
     req_rx: ClientRx<B>,
     fut_ctx: Option<FutCtx<B>>,
+    closing: Option<crate::Result<Dispatched>>,
     marker: PhantomData<T>,
 }
 
@@ -676,6 +678,24 @@ where
     type Output = crate::Result<Dispatched>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.closing.is_none() {
+            let res = ready!(self.poll_dispatch(cx));
+            self.closing = Some(res);
+        }
+        ready!(self.req_rx.poll_close(cx));
+        Poll::Ready(self.closing.take().expect("Future polled twice"))
+    }
+}
+
+impl<B, E, T> ClientTask<B, E, T>
+where
+    B: Body + 'static + Unpin,
+    B::Data: Send,
+    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+    E: Http2ClientConnExec<B, T> + Clone + Unpin,
+    T: Read + Write + Unpin,
+{
+    fn poll_dispatch(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<Dispatched>> {
         loop {
             match ready!(self.h2_tx.poll_ready(cx)) {
                 Ok(()) => (),