fix: propagate Body::size_hint when wrapping bodies (#2503)

src/async_impl/body.rs

@@ -148,10 +148,7 @@ impl Body {
     {
         use http_body_util::BodyExt;
 
-        let boxed = inner
-            .map_frame(|f| f.map_data(Into::into))
-            .map_err(Into::into)
-            .boxed();
+        let boxed = IntoBytesBody { inner }.map_err(Into::into).boxed();
 
         Body {
             inner: Inner::Streaming(boxed),
@@ -461,6 +458,47 @@ where
     }
 }
 
+// ===== impl IntoBytesBody =====
+
+pin_project! {
+    struct IntoBytesBody<B> {
+        #[pin]
+        inner: B,
+    }
+}
+
+// We can't use `map_frame()` because that loses the hint data (for good reason).
+// But we aren't transforming the data.
+impl<B> hyper::body::Body for IntoBytesBody<B>
+where
+    B: hyper::body::Body,
+    B::Data: Into<Bytes>,
+{
+    type Data = Bytes;
+    type Error = B::Error;
+
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+    ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
+        match futures_core::ready!(self.project().inner.poll_frame(cx)) {
+            Some(Ok(f)) => Poll::Ready(Some(Ok(f.map_data(Into::into)))),
+            Some(Err(e)) => Poll::Ready(Some(Err(e))),
+            None => Poll::Ready(None),
+        }
+    }
+
+    #[inline]
+    fn size_hint(&self) -> http_body::SizeHint {
+        self.inner.size_hint()
+    }
+
+    #[inline]
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use http_body::Body as _;
@@ -484,8 +522,9 @@ mod tests {
         assert!(!bytes_body.is_end_stream());
         assert_eq!(bytes_body.size_hint().exact(), Some(3));
 
-        let stream_body = Body::wrap(bytes_body);
-        assert!(!stream_body.is_end_stream());
-        assert_eq!(stream_body.size_hint().exact(), None);
+        // can delegate even when wrapped
+        let stream_body = Body::wrap(empty_body);
+        assert!(stream_body.is_end_stream());
+        assert_eq!(stream_body.size_hint().exact(), Some(0));
     }
 }

tests/client.rs

@@ -197,7 +197,7 @@ async fn body_pipe_response() {
             http::Response::new("pipe me".into())
         } else {
             assert_eq!(req.uri(), "/pipe");
-            assert_eq!(req.headers()["transfer-encoding"], "chunked");
+            assert_eq!(req.headers()["content-length"], "7");
 
             let full: Vec<u8> = req
                 .into_body()