How to correcly use body::stream()

[Bassetts]

I am trying to do almost exactly what was asked in #139; take the request as is, forward it to another host using hyper::Client and then return the response straight back to the client. I thought I had hit the jackpot when I found that issue but it seems that since then BodyStream has been made private in #345.

I am struggling to figure out how to take the stream returned by body::stream() and create my new response from it. When I try to use wrap_stream it complains that Into<Bytes>, which is required by wrap_stream is not satisfied for impl Buf

error[E0277]: the trait bound `bytes::bytes::Bytes: std::convert::From<impl Buf>` is not satisfied
   --> src/handlers.rs:67:30
    |
67  |   let request = request.body(hyper::Body::wrap_stream(body)).unwrap();
    |                              ^^^^^^^^^^^^^^^^^^^^^^^^ the trait `std::convert::From<impl Buf>` is not implemented for `bytes::bytes::Bytes`
    | 
   ::: /Users/jason/.cargo/registry/src/github.com-1ecc6299db9ec823/hyper-0.13.2/src/body/body.rs:158:12
    |
158 |         O: Into<Bytes> + 'static,
    |            ----------- required by this bound in `hyper::body::body::Body::wrap_stream`
    |
    = note: required because of the requirements on the impl of `std::convert::Into<bytes::bytes::Bytes>` for `impl Buf

If I make BodyStream public again I can get this working fine. I am fairly new to rust, am I doing something wrong here or should BodyStream be exported for this to work?

[seanmonstar]

What does your code look like?

Without looking, you could probably get this to work using TryStreamExt::map_ok to map the impl Buf items into Bytes.

[Bassetts]

The nearest I seem to have got to it working is with the following code

fn proxy(
  client: HttpClient,
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
  path_and_query()
    .and(warp::method())
    .and(header::headers_cloned())
    .and(warp::body::stream())
    .and(with_client(client))
    .and_then(handlers::proxy_request)
}

async fn proxy_request(
  url: url::Url,
  method: Method,
  headers: HeaderMap,
  body: impl Stream<Item = Result<impl Buf, warp::Error>> + Send + Sync,
  client: hyper::Client<hyper_rustls::HttpsConnector<hyper::client::HttpConnector>>,
) -> Result<impl warp::Reply, warp::Rejection> {
  let mut request = hyper::Request::builder()
    .uri(url.as_str())
    .method(method)
    .body(hyper::Body::wrap_stream(body))
    .unwrap();

  *request.headers_mut() = headers;
  let response = client.request(request).await;

  Ok(response.unwrap())

This is what gives me the above error. The aim is ultimately to pass the body around as efficiently as possible to avoid allocating a lot of memory if the body is large.

[seanmonstar]

If you add use futures::TryStreamExt to your module, then this should conceivably work:

let body = body.map_ok(|mut buf| {
    buf.to_bytes()
});

[Bassetts]

Thanks, doing that and adding + 'static to the body parameter on proxy_request seems to do the trick