How to respond streaming data from Receiver?

[kenmasumitsu]

I want to respond streaming data continuously gotten from Receiver.

I found axum::response::Sse that might be usable.
Are there any example codes? Or other better approach?

[davidpdrsn]

You can use ReceiverStream to convert the receiver into a stream and then use SSE similar to what's shown here.

However these days it's more common to use websockets than sse. More details on that here. There is a complete chat app example here.

[kenmasumitsu]

Thanks for it. Yes, I did various trial, but none of them worked fine. So that I am considering to use websockets.

There are some of my trial.
Firstly I tried BroadcastStream, like,

    let (tx, rx) = tokio::sync::broadcast::channel::<String>(4);
    let stream = tokio_stream::wrappers::BroadcastStream::new(rx);
    Sse::new(stream)

But I got the compile error.

  --> examples/sse/src/main.rs:54:10
   |
54 | ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
   |          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected struct `axum::response::sse::Event`, found struct `String`
   |
   = note: expected enum `Result<axum::response::sse::Event, Infallible>`
              found enum `Result<String, BroadcastStreamRecvError>`

So that, I changed String to Event.

    let (tx, rx) = tokio::sync::broadcast::channel::<Event>(4);
    let stream = tokio_stream::wrappers::BroadcastStream::new(rx);
    Sse::new(stream);

However, I got the errors.

error[E0277]: the trait bound `axum::response::sse::Event: Clone` is not satisfied
  --> examples/sse/src/main.rs:55:12
   |
55 |   ) -> Sse<impl Stream<Item = Result<Event, BroadcastStreamRecvError>>> {
   |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `Clone` is not implemented for `axum::response::sse::Event`
   |
   = note: required because of the requirements on the impl of `futures::Stream` for `BroadcastStream<axum::response::sse::Event>`

[davidpdrsn]

You didn't specify that you were using a broadcast receiver before. By Receiver I assumed you meant mpsc.

Broadcast is useful when you have multiple receivers which is why the messages must be cloneable. You could have a task that forwards from the broadcast receiver into an mpsc receiver that is hooked up to the sse stream.