|
| 1 | +//! `GET /lean/v0/events` — Server-Sent Events stream of chain events. |
| 2 | +//! |
| 3 | +//! The [`ethlambda_blockchain::BlockChainServer`] actor publishes |
| 4 | +//! [`ChainEvent`]s on a broadcast channel; this read-only handler subscribes a |
| 5 | +//! new receiver per connection and forwards each event as an SSE message. The |
| 6 | +//! flow is strictly one-directional (actor → broadcast → SSE), so RPC never |
| 7 | +//! writes into the actor. |
| 8 | +
|
| 9 | +use std::convert::Infallible; |
| 10 | + |
| 11 | +use axum::{ |
| 12 | + Extension, Router, |
| 13 | + response::{Sse, sse::Event}, |
| 14 | + routing::get, |
| 15 | +}; |
| 16 | +use ethlambda_blockchain::ChainEvent; |
| 17 | +use ethlambda_storage::Store; |
| 18 | +use futures_core::Stream; |
| 19 | +use tokio::sync::broadcast; |
| 20 | +use tokio_stream::{ |
| 21 | + StreamExt, |
| 22 | + wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}, |
| 23 | +}; |
| 24 | + |
| 25 | +async fn get_events( |
| 26 | + Extension(tx): Extension<broadcast::Sender<ChainEvent>>, |
| 27 | +) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { |
| 28 | + let stream = BroadcastStream::new(tx.subscribe()).filter_map(|res| { |
| 29 | + // A slow client falls behind and the broadcast channel overwrites |
| 30 | + // events it never read. Surface that rather than silently dropping. |
| 31 | + let ev = match res { |
| 32 | + Ok(ev) => ev, |
| 33 | + Err(BroadcastStreamRecvError::Lagged(skipped)) => { |
| 34 | + tracing::debug!(skipped, "SSE client lagged; dropped chain events"); |
| 35 | + return None; |
| 36 | + } |
| 37 | + }; |
| 38 | + let name = match &ev { |
| 39 | + ChainEvent::Head { .. } => "head", |
| 40 | + ChainEvent::Block { .. } => "block", |
| 41 | + ChainEvent::FinalizedCheckpoint { .. } => "finalized_checkpoint", |
| 42 | + }; |
| 43 | + Some(Ok(Event::default().event(name).json_data(ev).ok()?)) |
| 44 | + }); |
| 45 | + Sse::new(stream) |
| 46 | +} |
| 47 | + |
| 48 | +pub(crate) fn routes() -> Router<Store> { |
| 49 | + Router::new().route("/lean/v0/events", get(get_events)) |
| 50 | +} |
| 51 | + |
| 52 | +#[cfg(test)] |
| 53 | +mod tests { |
| 54 | + use super::*; |
| 55 | + use axum::{body::Body, http::Request}; |
| 56 | + use ethlambda_storage::{Store, backend::InMemoryBackend}; |
| 57 | + use std::sync::Arc; |
| 58 | + use tower::ServiceExt; |
| 59 | + |
| 60 | + use crate::test_utils::create_test_state; |
| 61 | + |
| 62 | + #[tokio::test] |
| 63 | + async fn events_streams_head() { |
| 64 | + let (tx, _) = broadcast::channel::<ChainEvent>(16); |
| 65 | + let store = Store::from_anchor_state(Arc::new(InMemoryBackend::new()), create_test_state()); |
| 66 | + let app = crate::build_api_router(store).layer(Extension(tx.clone())); |
| 67 | + |
| 68 | + // Issue the request first so the handler subscribes its receiver before |
| 69 | + // we publish — `broadcast::send` errors if there are no live receivers. |
| 70 | + let resp = app |
| 71 | + .oneshot( |
| 72 | + Request::builder() |
| 73 | + .uri("/lean/v0/events") |
| 74 | + .body(Body::empty()) |
| 75 | + .unwrap(), |
| 76 | + ) |
| 77 | + .await |
| 78 | + .unwrap(); |
| 79 | + assert_eq!(resp.status(), axum::http::StatusCode::OK); |
| 80 | + |
| 81 | + tx.send(ChainEvent::Head { |
| 82 | + slot: 3, |
| 83 | + root: Default::default(), |
| 84 | + parent_root: Default::default(), |
| 85 | + }) |
| 86 | + .unwrap(); |
| 87 | + |
| 88 | + let mut body = resp.into_body().into_data_stream(); |
| 89 | + let chunk = tokio_stream::StreamExt::next(&mut body) |
| 90 | + .await |
| 91 | + .unwrap() |
| 92 | + .unwrap(); |
| 93 | + let text = String::from_utf8_lossy(&chunk); |
| 94 | + assert!(text.contains("event:head") || text.contains("event: head")); |
| 95 | + } |
| 96 | +} |
0 commit comments