eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
38bd063109c1abd44620fc9b82b157ced3ab23a6
pmomusic/pmocontrol/src/sse.rs
Eric Coissac ea5328450b Implémentation de la détection de flux continu et gestion de la barre de progression 
Cette mise à jour implémente la détection des flux continus (radio, streaming) et améliore la gestion de la barre de progression pour ces flux. Les changements incluent : 
- Ajout d'une méthode predicat `is_playing_a_stream()` au niveau de `MusicRenderer`
- Développement d'une fonction utilitaire `is_continuous_stream_url()` pour analyser les headers HTTP
- Mise à jour des backends (UPnP, OpenHome, LinkPlay, ArylicTcp, Chromecast) pour détecter les flux continus
- Ajout d'un flag `continuous_stream` dans chaque backend
- Export du module de détection de flux
- Mise à jour de l'API REST et des événements SSE pour transmettre l'état du flux
- Correction de la gestion de la position de lecture pour les flux continus

Les flux continus sont maintenant correctement détectés via une analyse HTTP des headers, permettant une gestion appropriée de la barre de progression dans l'interface web.
2026-01-31 10:33:06 +01:00

712 lines
26 KiB
Rust
Raw Blame History

//! SSE endpoints pour les événements du Control Point
//!
//! Ce module fournit des endpoints Server-Sent Events pour permettre aux clients
//! web de recevoir en temps réel :
//! - Les événements des renderers (state, volume, position, queue, etc.)
//! - Les événements des serveurs de médias (global updates, container updates)
//!
//! Routes:
//! - GET /api/control/events/renderers - Événements renderers uniquement
//! - GET /api/control/events/servers - Événements serveurs uniquement
//! - GET /api/control/events - Tous les événements (agrégés)
//!
//! ⚠️ Les payloads SSE servent uniquement de signaux de rafraîchissement :
//! l'UI doit toujours refetch l'instantané complet auprès du ControlPoint,
//! seule source de vérité de l'état renderer.
#[cfg(feature = "pmoserver")]
use crate::control_point::ControlPoint;
#[cfg(feature = "pmoserver")]
use crate::model::{MediaServerEvent, RendererEvent};
#[cfg(feature = "pmoserver")]
use async_stream::stream;
#[cfg(feature = "pmoserver")]
use axum::{
Router,
extract::State,
response::IntoResponse,
response::sse::{Event, KeepAlive, Sse},
};
#[cfg(feature = "pmoserver")]
use serde::Serialize;
#[cfg(feature = "pmoserver")]
use std::sync::Arc;
use crate::{DeviceIdentity, DeviceOnline};
use tracing::error;
// ============================================================================
// PAYLOADS SSE
// ============================================================================
/// Payload SSE pour un événement renderer
#[cfg(feature = "pmoserver")]
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RendererEventPayload {
StateChanged {
renderer_id: String,
state: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
PositionChanged {
renderer_id: String,
track: Option<u32>,
rel_time: Option<String>,
track_duration: Option<String>,
timestamp: chrono::DateTime<chrono::Utc>,
},
VolumeChanged {
renderer_id: String,
volume: u16,
timestamp: chrono::DateTime<chrono::Utc>,
},
MuteChanged {
renderer_id: String,
mute: bool,
timestamp: chrono::DateTime<chrono::Utc>,
},
MetadataChanged {
renderer_id: String,
title: Option<String>,
artist: Option<String>,
album: Option<String>,
album_art_uri: Option<String>,
timestamp: chrono::DateTime<chrono::Utc>,
},
QueueUpdated {
renderer_id: String,
queue_length: usize,
timestamp: chrono::DateTime<chrono::Utc>,
},
BindingChanged {
renderer_id: String,
server_id: Option<String>,
container_id: Option<String>,
timestamp: chrono::DateTime<chrono::Utc>,
},
StreamStateChanged {
renderer_id: String,
is_stream: bool,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerStarted {
renderer_id: String,
duration_seconds: u32,
remaining_seconds: u32,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerUpdated {
renderer_id: String,
duration_seconds: u32,
remaining_seconds: u32,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerTick {
renderer_id: String,
remaining_seconds: u32,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerExpired {
renderer_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
TimerCancelled {
renderer_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
Online {
renderer_id: String,
friendly_name: String,
model_name: String,
manufacturer: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
Offline {
renderer_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
}
/// Payload SSE pour un événement serveur de médias
#[cfg(feature = "pmoserver")]
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum MediaServerEventPayload {
GlobalUpdated {
server_id: String,
system_update_id: Option<u32>,
timestamp: chrono::DateTime<chrono::Utc>,
},
ContainersUpdated {
server_id: String,
container_ids: Vec<String>,
timestamp: chrono::DateTime<chrono::Utc>,
},
Online {
server_id: String,
friendly_name: String,
model_name: String,
manufacturer: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
Offline {
server_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
}
/// Payload SSE unifié pour tous les événements
#[cfg(feature = "pmoserver")]
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "category", rename_all = "snake_case")]
pub enum UnifiedEventPayload {
Renderer(RendererEventPayload),
MediaServer(MediaServerEventPayload),
}
// ============================================================================
// HELPER FUNCTIONS
// ============================================================================
/// Convertit un RendererEvent en RendererEventPayload pour SSE
///
/// Cette fonction centralise la conversion pour éviter la duplication de code
/// entre les différents streams SSE (renderers-only et all-events).
#[cfg(feature = "pmoserver")]
fn renderer_event_to_payload(
event: RendererEvent,
timestamp: chrono::DateTime<chrono::Utc>,
) -> RendererEventPayload {
match event {
RendererEvent::StateChanged { id, state } => RendererEventPayload::StateChanged {
renderer_id: id.0,
state: state.as_str().to_string(),
timestamp,
},
RendererEvent::PositionChanged { id, position } => RendererEventPayload::PositionChanged {
renderer_id: id.0,
track: position.track,
rel_time: position.rel_time,
track_duration: position.track_duration,
timestamp,
},
RendererEvent::VolumeChanged { id, volume } => RendererEventPayload::VolumeChanged {
renderer_id: id.0,
volume,
timestamp,
},
RendererEvent::MuteChanged { id, mute } => RendererEventPayload::MuteChanged {
renderer_id: id.0,
mute,
timestamp,
},
RendererEvent::MetadataChanged { id, metadata } => RendererEventPayload::MetadataChanged {
renderer_id: id.0,
title: metadata.title,
artist: metadata.artist,
album: metadata.album,
album_art_uri: metadata.album_art_uri,
timestamp,
},
RendererEvent::QueueUpdated { id, queue_length } => RendererEventPayload::QueueUpdated {
renderer_id: id.0,
queue_length,
timestamp,
},
RendererEvent::BindingChanged { id, binding } => RendererEventPayload::BindingChanged {
renderer_id: id.0,
server_id: binding.as_ref().map(|b| b.server_id.0.clone()),
container_id: binding.as_ref().map(|b| b.container_id.clone()),
timestamp,
},
RendererEvent::StreamStateChanged { id, is_stream } => {
RendererEventPayload::StreamStateChanged {
renderer_id: id.0,
is_stream,
timestamp,
}
}
RendererEvent::TimerStarted {
id,
duration_seconds,
remaining_seconds,
} => RendererEventPayload::TimerStarted {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
},
RendererEvent::TimerUpdated {
id,
duration_seconds,
remaining_seconds,
} => RendererEventPayload::TimerUpdated {
renderer_id: id.0,
duration_seconds,
remaining_seconds,
timestamp,
},
RendererEvent::TimerTick {
id,
remaining_seconds,
} => RendererEventPayload::TimerTick {
renderer_id: id.0,
remaining_seconds,
timestamp,
},
RendererEvent::TimerExpired { id } => RendererEventPayload::TimerExpired {
renderer_id: id.0,
timestamp,
},
RendererEvent::TimerCancelled { id } => RendererEventPayload::TimerCancelled {
renderer_id: id.0,
timestamp,
},
RendererEvent::Online { id, info } => RendererEventPayload::Online {
renderer_id: id.0,
friendly_name: info.friendly_name,
model_name: info.model_name,
manufacturer: info.manufacturer,
timestamp,
},
RendererEvent::Offline { id } => RendererEventPayload::Offline {
renderer_id: id.0,
timestamp,
},
}
}
/// Convertit un MediaServerEvent en MediaServerEventPayload pour SSE
///
/// Cette fonction centralise la conversion pour éviter la duplication de code
/// entre les différents streams SSE (servers-only et all-events).
#[cfg(feature = "pmoserver")]
fn media_server_event_to_payload(
event: MediaServerEvent,
timestamp: chrono::DateTime<chrono::Utc>,
) -> MediaServerEventPayload {
match event {
MediaServerEvent::GlobalUpdated {
server_id,
system_update_id,
} => MediaServerEventPayload::GlobalUpdated {
server_id: server_id.0,
system_update_id,
timestamp,
},
MediaServerEvent::ContainersUpdated {
server_id,
container_ids,
} => MediaServerEventPayload::ContainersUpdated {
server_id: server_id.0,
container_ids,
timestamp,
},
MediaServerEvent::Online { server_id, info } => MediaServerEventPayload::Online {
server_id: server_id.0,
friendly_name: info.friendly_name,
model_name: info.model_name,
manufacturer: info.manufacturer,
timestamp,
},
MediaServerEvent::Offline { server_id } => MediaServerEventPayload::Offline {
server_id: server_id.0,
timestamp,
},
}
}
// ============================================================================
// HANDLERS SSE
// ============================================================================
/// Handler SSE pour les événements renderers
///
/// Route: GET /api/control/events/renderers
///
/// Diffuse tous les événements liés aux renderers (state, volume, position, queue, etc.)
/// en temps réel via Server-Sent Events.
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/events/renderers",
responses(
(status = 200, description = "Flux SSE des événements renderers", content_type = "text/event-stream")
),
tag = "control"
)]
pub async fn renderer_events_sse(
State(control_point): State<Arc<ControlPoint>>,
) -> impl IntoResponse {
// Convert crossbeam channel to tokio channel for async compatibility
let (tx, mut rx_tokio) = tokio::sync::mpsc::unbounded_channel();
let rx = control_point.subscribe_events();
// Spawn blocking task to bridge crossbeam -> tokio
tokio::task::spawn_blocking(move || {
while let Ok(event) = rx.recv() {
if tx.send(event).is_err() {
break;
}
}
});
let cp_for_heartbeat = control_point.clone();
let stream = stream! {
// INITIAL SNAPSHOT: Send Online events for all currently discovered renderers
// This ensures clients see devices that were discovered before they connected
let initial_renderers = {
let registry = control_point.registry();
let reg = registry.read().unwrap();
match reg.list_renderers() {
Ok(renderers) => renderers,
Err(e) => {
error!("Failed to list renderers: {}", e);
Vec::new()
}
}
};
for renderer in initial_renderers {
if renderer.is_online() {
let timestamp = chrono::Utc::now();
let payload = RendererEventPayload::Online {
renderer_id: renderer.id().0,
friendly_name: renderer.friendly_name().to_string(),
model_name: renderer.model_name().to_string(),
manufacturer: renderer.manufacturer().to_string(),
timestamp,
};
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("renderer").data(json));
}
}
}
// Heartbeat interval: re-send Online events every 2 minutes for all online devices
// This allows UIs that reconnect to quickly recover device state without waiting
// for an actual state change event
let mut heartbeat_interval = tokio::time::interval(tokio::time::Duration::from_secs(120));
heartbeat_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
// Then stream future events with periodic heartbeat
loop {
tokio::select! {
// Regular events from the control point
Some(event) = rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let payload = renderer_event_to_payload(event, timestamp);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("renderer").data(json));
}
}
// Periodic heartbeat: re-send Online events for all online renderers
_ = heartbeat_interval.tick() => {
let heartbeat_renderers = {
let registry = cp_for_heartbeat.registry();
let reg = registry.read().unwrap();
match reg.list_renderers() {
Ok(renderers) => renderers,
Err(e) => {
error!("Failed to list renderers for heartbeat: {}", e);
Vec::new()
}
}
};
for renderer in heartbeat_renderers {
if renderer.is_online() {
let timestamp = chrono::Utc::now();
let payload = RendererEventPayload::Online {
renderer_id: renderer.id().0,
friendly_name: renderer.friendly_name().to_string(),
model_name: renderer.model_name().to_string(),
manufacturer: renderer.manufacturer().to_string(),
timestamp,
};
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("renderer_heartbeat").data(json));
}
}
}
}
}
}
};
Sse::new(stream).keep_alive(KeepAlive::default())
}
/// Handler SSE pour les événements serveurs de médias
///
/// Route: GET /api/control/events/servers
///
/// Diffuse tous les événements liés aux serveurs de médias (global updates, container updates)
/// en temps réel via Server-Sent Events.
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/events/servers",
responses(
(status = 200, description = "Flux SSE des événements serveurs de médias", content_type = "text/event-stream")
),
tag = "control"
)]
pub async fn media_server_events_sse(
State(control_point): State<Arc<ControlPoint>>,
) -> impl IntoResponse {
// Convert crossbeam channel to tokio channel for async compatibility
let (tx, mut rx_tokio) = tokio::sync::mpsc::unbounded_channel();
let rx = control_point.subscribe_media_server_events();
// Spawn blocking task to bridge crossbeam -> tokio
tokio::task::spawn_blocking(move || {
while let Ok(event) = rx.recv() {
if tx.send(event).is_err() {
break;
}
}
});
let cp_for_heartbeat = control_point.clone();
let stream = stream! {
// INITIAL SNAPSHOT: Send Online events for all currently discovered media servers
// This ensures clients see servers that were discovered before they connected
let initial_servers = {
let registry = control_point.registry();
let reg = registry.read().unwrap();
match reg.list_servers() {
Ok(servers) => servers,
Err(e) => {
error!("Failed to list servers: {}", e);
Vec::new()
}
}
};
for server in initial_servers {
if server.is_online() {
let timestamp = chrono::Utc::now();
let payload = MediaServerEventPayload::Online {
server_id: server.id().0,
friendly_name: server.friendly_name().to_string(),
model_name: server.model_name().to_string(),
manufacturer: server.manufacturer().to_string(),
timestamp,
};
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("media_server").data(json));
}
}
}
// Heartbeat interval: re-send Online events every 2 minutes for all online servers
let mut heartbeat_interval = tokio::time::interval(tokio::time::Duration::from_secs(120));
heartbeat_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
// Then stream future events with periodic heartbeat
loop {
tokio::select! {
// Regular events from the control point
Some(event) = rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let payload = media_server_event_to_payload(event, timestamp);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("media_server").data(json));
}
}
// Periodic heartbeat: re-send Online events for all online servers
_ = heartbeat_interval.tick() => {
let heartbeat_servers = {
let registry = cp_for_heartbeat.registry();
let reg = registry.read().unwrap();
match reg.list_servers() {
Ok(servers) => servers,
Err(e) => {
error!("Failed to list servers for heartbeat: {}", e);
Vec::new()
}
}
};
for server in heartbeat_servers {
if server.is_online() {
let timestamp = chrono::Utc::now();
let payload = MediaServerEventPayload::Online {
server_id: server.id().0,
friendly_name: server.friendly_name().to_string(),
model_name: server.model_name().to_string(),
manufacturer: server.manufacturer().to_string(),
timestamp,
};
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("media_server_heartbeat").data(json));
}
}
}
}
}
}
};
Sse::new(stream).keep_alive(KeepAlive::default())
}
/// Handler SSE pour tous les événements (renderers + serveurs)
///
/// Route: GET /api/control/events
///
/// Diffuse tous les événements du control point (renderers et serveurs) en temps réel.
/// Chaque événement est catégorisé et inclut un timestamp.
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/events",
responses(
(status = 200, description = "Flux SSE de tous les événements du control point", content_type = "text/event-stream")
),
tag = "control"
)]
pub async fn all_events_sse(State(control_point): State<Arc<ControlPoint>>) -> impl IntoResponse {
// Convert crossbeam channels to tokio channels for async compatibility
let (renderer_tx, mut renderer_rx_tokio) = tokio::sync::mpsc::unbounded_channel();
let (server_tx, mut server_rx_tokio) = tokio::sync::mpsc::unbounded_channel();
let renderer_rx = control_point.subscribe_events();
let server_rx = control_point.subscribe_media_server_events();
// Spawn blocking tasks to bridge crossbeam -> tokio
tokio::task::spawn_blocking(move || {
while let Ok(event) = renderer_rx.recv() {
if renderer_tx.send(event).is_err() {
break;
}
}
});
tokio::task::spawn_blocking(move || {
while let Ok(event) = server_rx.recv() {
if server_tx.send(event).is_err() {
break;
}
}
});
let stream = stream! {
// INITIAL SNAPSHOT: Send Online events for all currently discovered devices
// This ensures clients see devices that were discovered before they connected
let (initial_renderers, initial_servers) = {
let registry = control_point.registry();
let reg = registry.read().unwrap();
let renderers = match reg.list_renderers() {
Ok(r) => r,
Err(e) => {
error!("Failed to list renderers: {}", e);
Vec::new()
}
};
let servers = match reg.list_servers() {
Ok(s) => s,
Err(e) => {
error!("Failed to list servers: {}", e);
Vec::new()
}
};
(renderers, servers)
};
// Send renderer Online events
for renderer in initial_renderers {
if renderer.is_online() {
let timestamp = chrono::Utc::now();
let renderer_payload = RendererEventPayload::Online {
renderer_id: renderer.id().0,
friendly_name: renderer.friendly_name().to_string(),
model_name: renderer.model_name().to_string(),
manufacturer: renderer.manufacturer().to_string(),
timestamp,
};
let payload = UnifiedEventPayload::Renderer(renderer_payload);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("control").data(json));
}
}
}
// Send server Online events
for server in initial_servers {
if server.is_online() {
let timestamp = chrono::Utc::now();
let server_payload = MediaServerEventPayload::Online {
server_id: server.id().0,
friendly_name: server.friendly_name().to_string(),
model_name: server.model_name().to_string(),
manufacturer: server.manufacturer().to_string(),
timestamp,
};
let payload = UnifiedEventPayload::MediaServer(server_payload);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("control").data(json));
}
}
}
// Then stream future events
loop {
tokio::select! {
Some(event) = renderer_rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let renderer_payload = renderer_event_to_payload(event, timestamp);
let payload = UnifiedEventPayload::Renderer(renderer_payload);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("control").data(json));
}
}
Some(event) = server_rx_tokio.recv() => {
let timestamp = chrono::Utc::now();
let server_payload = media_server_event_to_payload(event, timestamp);
let payload = UnifiedEventPayload::MediaServer(server_payload);
if let Ok(json) = serde_json::to_string(&payload) {
yield Ok::<_, axum::Error>(Event::default().event("control").data(json));
}
}
else => break
}
}
};
Sse::new(stream).keep_alive(KeepAlive::default())
}
// ============================================================================
// ROUTER
// ============================================================================
/// Crée le router SSE pour les événements du Control Point
#[cfg(feature = "pmoserver")]
pub fn create_sse_router(control_point: Arc<ControlPoint>) -> Router {
use axum::routing::get;
Router::new()
.route("/events", get(all_events_sse))
.route("/events/renderers", get(renderer_events_sse))
.route("/events/servers", get(media_server_events_sse))
.with_state(control_point)
}
Reference in New Issue View Git Blame Copy Permalink