Spaces:
Sleeping
Sleeping
File size: 1,276 Bytes
03e3b1b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | use axum::{
body::Body,
extract::{Path, State},
http::{header, StatusCode},
response::Response,
};
use std::sync::Arc;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::StreamExt;
use super::routes::AppState;
pub async fn events(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Result<Response<Body>, StatusCode> {
let rx = state.solver.subscribe(&id).ok_or(StatusCode::NOT_FOUND)?;
let bootstrap_json = state
.solver
.bootstrap_event(&id)
.map_err(|_| StatusCode::NOT_FOUND)?;
let bootstrap = tokio_stream::iter(std::iter::once(Ok::<_, std::convert::Infallible>(
format!("data: {}\n\n", bootstrap_json).into_bytes(),
)));
let live = BroadcastStream::new(rx).filter_map(|msg| match msg {
Ok(json) => Some(Ok::<_, std::convert::Infallible>(
format!("data: {}\n\n", json).into_bytes(),
)),
Err(_) => None, // Lagged — skip missed messages
});
let stream = bootstrap.chain(live);
Ok(Response::builder()
.header(header::CONTENT_TYPE, "text/event-stream")
.header(header::CACHE_CONTROL, "no-cache")
.header("X-Accel-Buffering", "no")
.body(Body::from_stream(stream))
.unwrap())
}
|