Forwarding a reqwest bytestream in a Rocket response without copying

[tamas]

I would like to create an endpoint in Rocket that forwards a bytestream from a reqwest response to the client. I have some trouble figuring out how to manage this. Particularly, the ByteStream! macro is a bit confusing to me in the documentation, I don't really know what its parameter is supposed to be and what it expands to. My code currently:

#[get("/github/<org>/<repo>/<sha>")]
async fn grab(org: String, repo: String, sha: String) -> ByteStream![Vec<u8>] {
  if let Ok(res) = reqwest::get(format!("https://github.com/{}/{}/archive/{}.tar.gz", org, repo, sha)).await {
    res.bytes_stream()
  }
}

This results in

error[E0308]: mismatched types
   --> src/main.rs:13:5
    |
13  |     res.bytes_stream()
    |     ^^^^^^^^^^^^^^^^^^ expected struct `ByteStream`, found opaque type
    |
   ::: /home/tamas/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.11.4/src/async_impl/response.rs:314:34
    |
314 |     pub fn bytes_stream(self) -> impl futures_core::Stream<Item = crate::Result<Bytes>> {
    |                                  ------------------------------------------------------ the found opaque type
    |
    = note:   expected struct `ByteStream<_>`
            found opaque type `impl rocket::futures::Stream`

Which is not surprising, but I'm stuck on how to convert between the two types (ideally while also avoiding copying. Any pointers? (no pun intended)

[bjorn3]

I think res.bytes_stream().into() will work. Rocket's ByteStream has an impl From<S> for ByteStream<S> where S: Stream. As return type you will have to use impl Responder I think. You need ByteStream<whatever the return type of `res.byte_stream()` is>. This type is not writable due to reqwest returning an impl Stream<_>, so you need to return the opaque type impl Responder which I think is the exact trait #[get] expects the function return type to implement.

[tamas]

Thanks, that seems to work. However, I don't know what to return when the query fails. I tried wrapping the return type in Option (which seems to be idiomatic in Rocket), but that doesn't work.

[bjorn3]

Try to keep the return type the same, but still return an option. I think this would work:

#[get("/github/<org>/<repo>/<sha>")]
async fn grab(org: String, repo: String, sha: String) -> impl Responder {
  if let Ok(res) = reqwest::get(format!("https://github.com/{}/{}/archive/{}.tar.gz", org, repo, sha)).await {
    Some(ByteStream(res.bytes_stream()))
  } else {
    None
  }
}

Note: I switched from res.bytes_stream().into() to ByteStream(res.bytes_stream()) as I am not sure if type inference would break otherwise ByteStream is the type rocket::response::stream::ByteStream here.

[tamas]

That results in

   Compiling zustad-server v0.1.0 (/home/tamas/projects/rust/zustad/zustad-server)
error[E0726]: implicit elided lifetime not allowed here
  --> zustad-server/src/main.rs:11:63
   |
11 | async fn grab(org: String, repo: String, sha: String) -> impl Responder {
   |                                                               ^^^^^^^^^- help: indicate the anonymous lifetimes: `<'_, '_>`
   |
   = note: assuming a `'static` lifetime...

error[E0277]: the trait bound `Result<rocket::http::hyper::Bytes, reqwest::Error>: AsRef<[u8]>` is not satisfied
  --> zustad-server/src/main.rs:11:58
   |
11 | async fn grab(org: String, repo: String, sha: String) -> impl Responder {
   |                                                          ^^^^^^^^^^^^^^ the trait `AsRef<[u8]>` is not implemented for `Result<rocket::http::hyper::Bytes, reqwest::Error>`
   |
   = note: required because of the requirements on the impl of `Responder<'_, '_>` for `ByteStream<impl rocket::futures::Stream>`
   = note: 1 redundant requirements hidden
   = note: required because of the requirements on the impl of `Responder<'static, 'static>` for `std::option::Option<ByteStream<impl rocket::futures::Stream>>`

error[E0759]: `__req` has lifetime `'__r` but it needs to satisfy a `'static` lifetime requirement
  --> zustad-server/src/main.rs:11:58
   |
10 | #[get("/github/<org>/<repo>/<sha>")]
   | ------------------------------------ this data with lifetime `'__r`...
11 | async fn grab(org: String, repo: String, sha: String) -> impl Responder {
   |                                                          ^^^^^^^^^^^^^^ ...is captured and required to live as long as `'static` here

Some errors have detailed explanations: E0277, E0759.
For more information about an error, try `rustc --explain E0277`.
warning: `zustad-server` (bin "zustad-server") generated 2 warnings
error: could not compile `zustad-server` due to 3 previous errors; 2 warnings emitted

Sorry to just paste the error here, this is a bit over my head. I tried adding the lifetimes as the error suggested but that did't work either.

[alice]

It seems like the main complaint is that the byte arrays are wrapped in a Result, but rocket wants the stream item type to just be the byte array without a result. I'm not sure what you should do with the errors though.

[tamas]

That appears to be the case, but I don't see what actually has a Result type in that expression.