SSE - build stream that streams when message is received on channel

[myz-dev]

Hi,
I want to serve connected clients server side events when a condition on my server is met.
I know you can use web sockets for this (like in the chat app) but my front end is already written and I am currently trying to port my back-end.
So I have set up my route as shown in the example. I am trying to figure out how I can trigger the stream on my own condition but could not make it work yet. Can somebody help me?
My route has access to a tokio::sync::broadcast::Receiver<T>. I would like to await the receive method of this receiver and send out a new event every time I get a new message.

/// Sends a new event every time a message is recevied on `rx_broadcast`.
pub async fn stream_events(
    TypedHeader(user_agent): TypedHeader<headers::UserAgent>,
    State(rx_broadcast): State<SharedStreamer>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    info!("`{}` connected", user_agent.as_str());
    // A `Stream` that repeats an event every second
    let stream =
        stream::repeat_with(|| Event::default().event(MY_EVENT_NAME).data(I_WOULD_LIKE_TO_UPDATE_THIS_DYNAMICALLY))
            .map(Ok)
            .throttle(Duration::from_secs(1)); // ideally I would not need this, the stream would just yield when there is new data

    Sse::new(stream).keep_alive(
        axum::response::sse::KeepAlive::new()
            .interval(Duration::from_secs(1))
            .text("keep-stream-alive"),
    )
}

I tried to write the logic in the .data() call, but it will only let me write synchronous code there.

let stream = stream::repeat_with(|| {
        Event::default().event(DEFECT_STREAM_EVENT_NAME).data(
            async {
                let k = defect_broadcast.0.recv().await.unwrap();
                serde_json::to_string(&k).unwrap()
            }
            .await,
        )
    })
    .map(Ok)
    .throttle(Duration::from_secs(1));

(says that the .repeat_with() call needs to be async but it is not.

[ryanmcgrath]

If you're still hitting issues with this, you have two options that could work:

You could use BroadcastStream to wrap the Receiver into a Stream implementing type. Dunno what type your State is holding though, so...

You could also do something like the following, using async-stream:

use std::convert::Infallible;

use async_stream::try_stream;
use axum::extract::State;
use axum::routing::get;
use axum::response::sse::{Event, KeepAlive, Sse};
use futures::Stream;

use my_crate::ServerState;

async fn event_stream(
    State(state): State<ServerState>
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let mut receiver = state.event_stream.subscribe();

    Sse::new(try_stream! {
        loop {
            match receiver.recv().await {
                Ok(i) => {
                    let event = Event::default()
                        .data(i);

                    yield event;
                },

                Err(e) => {
                    tracing::error!(error = ?e, "Failed to get");
                }
            }
        }
    }).keep_alive(KeepAlive::default())
}

In this case my Sender and Receiver types are just passing strings across, but you can of course make this pass just about whatever over the channel.

[will-break-it]

How would break out of this loop for a graceful shutdown?

[ryanmcgrath]

with_graceful_shutdown coupled with stashing a sender reference somewhere that could pass in a "exit" message.