Consider making Streaming<T> Sync

[NeoLegends]

Feature Request

Crates

Main Crate

Motivation

I'm trying to output a unidirectional tonic gRPC stream via SSE using warp. I'm using warp from seanmonstar/warp#265 to be able to use it with async await. Changing warp to be compatible with async / await requires that all responses handed over to warp (and in turn handed over to hyper) to be Sync. However, the stream returned from a streaming gRPC response is not Sync.

Proposal

Rustc complains that the decoder specifically is the part that is not sync:

tonic/tonic/src/codec/decode.rs

Line 22 in 5d0a795

decoder: Box<dyn Decoder<Item = T, Error = Status> + Send + 'static>,

As far as I can see, decoder is not accessible from the outside, and thus it should be safe to mark the whole Streaming struct as Sync. I would be happy to file a PR for that.

Alternatives

I don't see any, because AFAIK the Sync bound stems from a layer deep down inside hyper, and I guess it's unfeasible to change that somehow to remove it.

[NeoLegends]

In case you accept this I have already filed PR #82 that applies the necessary change. Feel free to close it otherwise.

[LucioFranco]

@NeoLegends sorry for the delay, this should be done! Let me know if you run into any issues. I plan to get a new release out soon.

[parasyte]

@LucioFranco This new trait bound broke my code because hyper::client:ResponseFuture is not Sync.

error[E0277]: `(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)` cannot be shared between threads safely
   --> src/lib.rs:153:26
    |
153 |         Ok(Response::new(Box::pin(outbound)
    |                          ^^^^^^^^^^^^^^^^^^ `(dyn core::future::future::Future<Output = std::result::Result<http::response::Response<hyper::body::body::Body>, hyper::error::Error>> + std::marker::Send + 'static)` cannot be shared between threads safely

Still looking into a way to resolve this... Does hyper need to be patched to add the trait bound to ResponseFuture? Or can I wrap this future in something to make it Sync, like futures::lock::Mutex?

cc @seanmonstar Maybe you can help out here?

[LucioFranco]

@parasyte maybe I'm missing something but what is the reason you are returning a future as the response type?

[parasyte]

@LucioFranco Sorry for the confusion, I'm not returning Future as a response type at all. It will help to have a full example to the issue. I'll get to that for you soon.

The short description is that I'm calling the get method on a hyper::client::Client in an async_stream::try_stream! {} loop; the return type is a gRPC streaming response. E.g. Pin<Box<dyn Stream<Item = Result<Message, Status> + Send + Sync + 'static>>

Awaiting this non-Sync ResponseFuture from the hyper client (and adding the necessary Sync bound to my streaming response type) is what causes the compile error.

[LucioFranco]

@parasyte you're right...I'm not sure how to move forward on this one @seanmonstar any ideas?

[seanmonstar]

Hrm, is the problem that hyper::client::ResponseFuture has a Box<dyn Future + Send> inside it, and not Box<dyn Future + Send + Sync>?

Oh wait, is this because it technically needs a Sync because of await being a jerk?

[parasyte]

@seanmonstar More info in #117 Yes, I think it has to do with await seeing the Sync trait bound and expecting everything to be Sync