Merge pull request #44 from Fishrock123/fix-hyper-body

src: correctly translate hyper body stream to ours

Author: Fishrock123

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://book.async.rs/overview
 
 ## [Unreleased]
 
+## Fixed
+- Fixed a body stream translation bug in the `hyper_client`.
+
 ## [5.0.0] - 2020-09-18
 
 This release includes an optional backend using [hyper.rs](https://hyper.rs/), and uses [async-trait](https://crates.io/crates/async-trait) for `HttpClient`.

Cargo.toml

@@ -22,7 +22,7 @@ h1_client = ["async-h1", "async-std", "async-native-tls"]
 native_client = ["curl_client", "wasm_client"]
 curl_client = ["isahc", "async-std"]
 wasm_client = ["js-sys", "web-sys", "wasm-bindgen", "wasm-bindgen-futures", "futures"]
-hyper_client = ["hyper", "hyper-tls", "http-types/hyperium_http"]
+hyper_client = ["hyper", "hyper-tls", "http-types/hyperium_http", "futures-util"]
 
 [dependencies]
 async-trait = "0.1.37"
@@ -37,6 +37,7 @@ async-native-tls = { version = "0.3.1", optional = true }
 # hyper_client
 hyper = { version = "0.13.6", features = ["tcp"], optional = true }
 hyper-tls = { version = "0.4.3", optional = true }
+futures-util = { version = "0.3.5", optional = true }
 
 # curl_client
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

src/hyper.rs

@@ -1,11 +1,13 @@
 //! http-client implementation for reqwest
 
 use super::{async_trait, Error, HttpClient, Request, Response};
+use futures_util::stream::TryStreamExt;
 use http_types::headers::{HeaderName, HeaderValue};
 use http_types::StatusCode;
 use hyper::body::HttpBody;
 use hyper_tls::HttpsConnector;
 use std::convert::TryFrom;
+use std::io;
 use std::str::FromStr;
 
 /// Hyper-based HTTP Client.
@@ -100,20 +102,11 @@ struct HttpTypesResponse {
 
 impl HttpTypesResponse {
     async fn try_from(value: hyper::Response<hyper::Body>) -> Result<Self, Error> {
-        let (parts, mut body) = value.into_parts();
-
-        let body = match body.data().await {
-            None => None,
-            Some(Ok(b)) => Some(b),
-            Some(Err(_)) => {
-                return Err(Error::from_str(
-                    StatusCode::BadGateway,
-                    "unable to read HTTP response body",
-                ))
-            }
-        }
-        .map(|b| http_types::Body::from_bytes(b.to_vec()))
-        .unwrap_or(http_types::Body::empty());
+        let (parts, body) = value.into_parts();
+
+        let size_hint = body.size_hint().upper().map(|s| s as usize);
+        let body = body.map_err(|err| io::Error::new(io::ErrorKind::Other, err.to_string()));
+        let body = http_types::Body::from_reader(body.into_async_read(), size_hint);
 
         let mut res = Response::new(parts.status);
         res.set_version(Some(parts.version.into()));