Using actix-web Payload as ByteStream

[stefanhoelzl]

Hello,
how is it possible to use an actix_web::web::Playload as input parameter to aws_sdk_s3::client::fluent_builders::PutObject::body accepting the type aws_sdk_s3::ByteStream?

my actix request handler currently looks like this:

use actix_web::{post, web::Payload, HttpResponse, Responder};
use aws_config;
use aws_sdk_s3::{ByteStream, Client};

#[post("/")]
async fn create(body: Payload) -> impl Responder {
    let config = aws_config::load_from_env().await;
    let client = Client::new(&config);
    client
        .put_object()
        .bucket("my-bucket")
        .key("my-key")
        .body(ByteStream::from(body.to_vec())) // <- here I want to stream the body directly
        .send()
        .await
        .unwrap();
    HttpResponse::Ok()
}

Here is an example from the Payload documentation on how to use it as a stream, but I don't know how to use this for aws_sdk_s3::ByteStream

Thanks for your help!

[rcoh]

Hello! This is pattern we should probably add a better API for. You can do this via hyper::Body (note: Hyper Body does not use the Hyper client at all, just the body implementation):

let body: Payload = ....;
let body = hyper::Body::wrap_stream(body);
// convert hyper::Body into SdkBody
let byte_stream = ByteStream::new(body.into());

[stefanhoelzl]

Thanks a lot for your help!
I tried your suggestions but get the following error message:

error[E0277]: `Rc<RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely
   --> src/main.rs:21:43
    |
21  |     let stream = hyper::Body::wrap_stream(body);
    |                  ------------------------ ^^^^ `Rc<RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely
    |                  |
    |                  required by a bound introduced by this call
    |
    = help: within `actix_web::web::Payload`, the trait `Send` is not implemented for `Rc<RefCell<actix_http::h1::payload::Inner>>`
    = note: required because it appears within the type `actix_http::h1::payload::Payload`
    = note: required because it appears within the type `actix_web::dev::Payload`
    = note: required because it appears within the type `actix_web::web::Payload`
note: required by a bound in `hyper::Body::wrap_stream`
   --> /Users/stefan/.cargo/registry/src/github.com-1ecc6299db9ec823/hyper-0.14.17/src/body/body.rs:192:42
    |
192 |         S: Stream<Item = Result<O, E>> + Send + 'static,
    |                                          ^^^^ required by this bound in `hyper::Body::wrap_stream`

error[E0277]: `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` cannot be sent between threads safely
   --> src/main.rs:21:43
    |
21  |     let stream = hyper::Body::wrap_stream(body);
    |                  ------------------------ ^^^^ `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` cannot be sent between threads safely
    |                  |
    |                  required by a bound introduced by this call
    |
    = help: the trait `Send` is not implemented for `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)`
    = note: required because of the requirements on the impl of `Send` for `Unique<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>`
    = note: required because it appears within the type `Box<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>`
    = note: required because it appears within the type `Pin<Box<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>>`
    = note: required because it appears within the type `actix_web::dev::Payload`
    = note: required because it appears within the type `actix_web::web::Payload`
note: required by a bound in `hyper::Body::wrap_stream`
   --> /Users/stefan/.cargo/registry/src/github.com-1ecc6299db9ec823/hyper-0.14.17/src/body/body.rs:192:42
    |
192 |         S: Stream<Item = Result<O, E>> + Send + 'static,
    |                                          ^^^^ required by this bound in `hyper::Body::wrap_stream`

For more information about this error, try `rustc --explain E0277`.
error: could not compile `backend` due to 2 previous errors

It fails when trying to wrap the Payload in a hyper::Body

let stream = hyper::Body::wrap_stream(body);

[rcoh]

Hmmm....I think this might be problematic. In general, the body definitely _can_ be passed around to multiple threads (unless you're using a single threaded runtime). I'll need to investigate some more.

…

On Fri, Feb 18, 2022 at 9:33 AM Stefan Hoelzl ***@***.***> wrote: Thanks a lot for your help! I tried your suggestions but get the following error message: error[E0277]: `Rc<RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely --> src/main.rs:21:43 | 21 | let stream = hyper::Body::wrap_stream(body); | ------------------------ ^^^^ `Rc<RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely | | | required by a bound introduced by this call | = help: within `actix_web::web::Payload`, the trait `Send` is not implemented for `Rc<RefCell<actix_http::h1::payload::Inner>>` = note: required because it appears within the type `actix_http::h1::payload::Payload` = note: required because it appears within the type `actix_web::dev::Payload` = note: required because it appears within the type `actix_web::web::Payload` note: required by a bound in `hyper::Body::wrap_stream` --> /Users/stefan/.cargo/registry/src/github.com-1ecc6299db9ec823/hyper-0.14.17/src/body/body.rs:192:42 | 192 | S: Stream<Item = Result<O, E>> + Send + 'static, | ^^^^ required by this bound in `hyper::Body::wrap_stream` error[E0277]: `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` cannot be sent between threads safely --> src/main.rs:21:43 | 21 | let stream = hyper::Body::wrap_stream(body); | ------------------------ ^^^^ `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` cannot be sent between threads safely | | | required by a bound introduced by this call | = help: the trait `Send` is not implemented for `(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` = note: required because of the requirements on the impl of `Send` for `Unique<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>` = note: required because it appears within the type `Box<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>` = note: required because it appears within the type `Pin<Box<(dyn futures_core::stream::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>>` = note: required because it appears within the type `actix_web::dev::Payload` = note: required because it appears within the type `actix_web::web::Payload` note: required by a bound in `hyper::Body::wrap_stream` --> /Users/stefan/.cargo/registry/src/github.com-1ecc6299db9ec823/hyper-0.14.17/src/body/body.rs:192:42 | 192 | S: Stream<Item = Result<O, E>> + Send + 'static, | ^^^^ required by this bound in `hyper::Body::wrap_stream` For more information about this error, try `rustc --explain E0277`. error: could not compile `backend` due to 2 previous errors It fails when trying to wrap the Payload in a hyper::Body let stream = hyper::Body::wrap_stream(body); — Reply to this email directly, view it on GitHub <#458 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AADYKZ23AWHAA4OGGLDD4ZLU3ZKCFANCNFSM5OX2H3WQ> . You are receiving this because you commented.Message ID: ***@***.***>

[rcoh]

I think you'll need something like SyncWrapper described in https://internals.rust-lang.org/t/what-shall-sync-mean-across-an-await/12020/2

The discussion is actually around this specific issue of a streaming HTTP body

[seabo]

Hello! I ran into the exact same issue as @stefanhoelzl (wanting to use an actix_web::web::Playload as an aws_sdk_s3::types::ByteStream).

Since the discussion above, it seems the implementation of hyper::Body::wrap_stream has incorporated the SyncWrapper idea @rcoh mentioned above. This means the Sync bound no longer exists on the Stream passed into wrap_stream. However, this approach still fails to compile because the Send bound remains, and actix_web::web::Playload is !Send.

I would appreciate any pointers on how to get this working.

[martijnarts]

I don't know if this problem is the same with actix-web, but I ran into a (similar) issue with Axum. I managed to solve it through an ownership-taking wrapper, as described in this blog post: https://blog.adamchalmers.com/streaming-proxy/