From ac177872642a1e5fd515abf2a4d0653d618cbda5 Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Thu, 9 Apr 2026 03:38:06 -0600 Subject: [PATCH] feat(qbzd): unified event bus + SSE endpoint (Phase 3) Playback event loop polls QbzCore player state with adaptive timing (250ms playing, 1s paused, 5s idle) and broadcasts PlaybackSnapshot events to the event bus. New SSE endpoint at /api/events streams all DaemonEvents to clients. Broadcast channel bounded to 256 events with lagged client handling. cc --- crates/qbzd/Cargo.toml | 2 + crates/qbzd/src/api/events.rs | 96 +++++++++++++++++++++++++++++++++++ crates/qbzd/src/api/mod.rs | 1 + crates/qbzd/src/daemon.rs | 76 +++++++++++++++++++++++++++ crates/qbzd/src/main.rs | 1 + 5 files changed, 176 insertions(+) create mode 100644 crates/qbzd/src/api/events.rs create mode 100644 crates/qbzd/src/api/mod.rs diff --git a/crates/qbzd/Cargo.toml b/crates/qbzd/Cargo.toml index e5215e081..2bbdbd943 100644 --- a/crates/qbzd/Cargo.toml +++ b/crates/qbzd/Cargo.toml @@ -26,7 +26,9 @@ tower-http = { version = "0.6", features = ["cors"] } # Async tokio = { workspace = true } +tokio-stream = { workspace = true } async-trait = { workspace = true } +futures-util = { workspace = true } # Serialization serde = { workspace = true } diff --git a/crates/qbzd/src/api/events.rs b/crates/qbzd/src/api/events.rs new file mode 100644 index 000000000..568d94a8d --- /dev/null +++ b/crates/qbzd/src/api/events.rs @@ -0,0 +1,96 @@ +//! SSE (Server-Sent Events) endpoint for real-time event streaming. +//! +//! Clients connect to `/api/events` and receive a continuous stream of +//! DaemonEvents formatted as SSE. Slow clients that fall behind the +//! broadcast buffer are disconnected (SSE auto-reconnect handles this). + +use std::convert::Infallible; +use std::sync::Arc; + +use axum::response::sse::{Event, KeepAlive, Sse}; +use futures_util::stream::Stream; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::StreamExt; + +use crate::adapter::{DaemonEvent, PlaybackSnapshot, RuntimeEvent}; +use crate::daemon::DaemonCore; + +/// SSE endpoint handler. Each connected client gets its own broadcast receiver. +pub async fn sse_handler( + daemon: Arc, +) -> Sse>> { + let rx = daemon.event_bus.subscribe(); + let stream = BroadcastStream::new(rx).filter_map(|result| match result { + Ok(event) => Some(Ok(format_sse_event(&event))), + Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => { + log::warn!("[SSE] Client lagged, dropped {} events", n); + Some(Ok(Event::default().comment(format!("lagged:{}", n)))) + } + }); + + Sse::new(stream).keep_alive(KeepAlive::default()) +} + +/// Convert a DaemonEvent to an SSE Event with typed event name. +fn format_sse_event(event: &DaemonEvent) -> Event { + match event { + DaemonEvent::Core(core_event) => { + let (event_type, data) = format_core_event(core_event); + Event::default().event(event_type).data(data) + } + DaemonEvent::Playback(snapshot) => Event::default() + .event("playback") + .data(serde_json::to_string(snapshot).unwrap_or_default()), + DaemonEvent::Runtime(runtime_event) => Event::default() + .event("runtime") + .data(serde_json::to_string(runtime_event).unwrap_or_default()), + } +} + +/// Map CoreEvent variants to SSE event type + JSON data. +fn format_core_event(event: &qbz_models::CoreEvent) -> (&'static str, String) { + use qbz_models::CoreEvent; + match event { + CoreEvent::TrackStarted { track, .. } => ( + "track-started", + serde_json::to_string(track).unwrap_or_default(), + ), + CoreEvent::TrackEnded { track_id } => ( + "track-ended", + serde_json::json!({"track_id": track_id}).to_string(), + ), + CoreEvent::QueueUpdated { state } => ( + "queue", + serde_json::to_string(state).unwrap_or_default(), + ), + CoreEvent::FavoritesUpdated { favorite_type } => ( + "favorites-updated", + serde_json::json!({"type": favorite_type}).to_string(), + ), + CoreEvent::PlaylistCreated { playlist } => ( + "playlist-created", + serde_json::to_string(playlist).unwrap_or_default(), + ), + CoreEvent::PlaylistUpdated { playlist_id } => ( + "playlist-updated", + serde_json::json!({"playlist_id": playlist_id}).to_string(), + ), + CoreEvent::Error { + code, + message, + recoverable, + } => ( + "error", + serde_json::json!({"code": code, "message": message, "recoverable": recoverable}) + .to_string(), + ), + CoreEvent::LoggedIn { session } => ( + "logged-in", + serde_json::json!({"user_id": session.user_id, "display_name": session.display_name}) + .to_string(), + ), + CoreEvent::LoggedOut => ("logged-out", "{}".to_string()), + // Catch-all for any future CoreEvent variants + _ => ("core-event", format!("{:?}", event)), + } +} diff --git a/crates/qbzd/src/api/mod.rs b/crates/qbzd/src/api/mod.rs new file mode 100644 index 000000000..a9970c28f --- /dev/null +++ b/crates/qbzd/src/api/mod.rs @@ -0,0 +1 @@ +pub mod events; diff --git a/crates/qbzd/src/daemon.rs b/crates/qbzd/src/daemon.rs index d4f910cc8..de41e5ecc 100644 --- a/crates/qbzd/src/daemon.rs +++ b/crates/qbzd/src/daemon.rs @@ -99,6 +99,9 @@ pub async fn run(mut config: DaemonConfig) -> Result<(), String> { } } + // Start playback state polling loop (broadcasts to event bus) + spawn_playback_loop(daemon.core.clone(), daemon.event_bus.clone()); + log::info!( "[qbzd] Daemon ready on {}:{}", config.server.bind, @@ -192,6 +195,72 @@ fn load_oauth_token() -> Option { } } +/// Spawn the playback state polling loop. +/// Reads player state and broadcasts PlaybackSnapshot events. +/// Adaptive polling: 250ms playing, 1s paused, 5s idle. +fn spawn_playback_loop( + core: Arc>, + event_tx: broadcast::Sender, +) { + tokio::spawn(async move { + let mut last_position: u64 = 0; + let mut last_is_playing = false; + let mut last_track_id: u64 = 0; + + loop { + let player = core.player(); + let state = &player.state; + + let is_playing = state.is_playing(); + let track_id = state.current_track_id(); + let position = state.current_position(); + let duration = state.duration(); + let volume = state.volume(); + let sample_rate = state.get_sample_rate(); + let bit_depth = state.get_bit_depth(); + + let track_cleared = track_id == 0 && last_track_id != 0; + let should_emit = (track_id != 0 + && (is_playing != last_is_playing + || track_id != last_track_id + || (is_playing && position != last_position))) + || track_cleared; + + if should_emit { + let snapshot = crate::adapter::PlaybackSnapshot { + state: if is_playing { + "Playing".to_string() + } else if track_id != 0 { + "Paused".to_string() + } else { + "Stopped".to_string() + }, + track_id, + position_secs: position, + duration_secs: duration, + volume, + sample_rate, + bit_depth, + }; + let _ = event_tx.send(DaemonEvent::Playback(snapshot)); + last_position = position; + last_is_playing = is_playing; + last_track_id = track_id; + } + + // Adaptive polling + let sleep_ms = if is_playing { + 250 + } else if track_id == 0 { + 5000 + } else { + 1000 + }; + tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await; + } + }); +} + /// Build the Axum HTTP router. fn build_router(daemon: Arc) -> axum::Router { use axum::routing::get; @@ -212,6 +281,13 @@ fn build_router(daemon: Arc) -> axum::Router { move || status_handler(d.clone()) }), ) + .route( + "/api/events", + get({ + let d = daemon.clone(); + move || crate::api::events::sse_handler(d.clone()) + }), + ) } async fn ping_handler() -> &'static str { diff --git a/crates/qbzd/src/main.rs b/crates/qbzd/src/main.rs index 7e55e2def..4506b2ef6 100644 --- a/crates/qbzd/src/main.rs +++ b/crates/qbzd/src/main.rs @@ -1,3 +1,4 @@ +mod api; mod config; mod daemon; mod adapter;