diff --git a/pmoaudio-ext/src/lib.rs b/pmoaudio-ext/src/lib.rs index 6dfc44b8..ce2b9a25 100755 --- a/pmoaudio-ext/src/lib.rs +++ b/pmoaudio-ext/src/lib.rs @@ -43,3 +43,6 @@ pub use sources::PlaylistSource; #[cfg(feature = "http-stream")] pub use sources::UriSource; + +#[cfg(feature = "http-stream")] +pub use sources::{PlayerCommand, PlayerEvent, PlayerHandle, PlayerSource}; diff --git a/pmoaudio-ext/src/sources/mod.rs b/pmoaudio-ext/src/sources/mod.rs index ce87b685..86d5174c 100644 --- a/pmoaudio-ext/src/sources/mod.rs +++ b/pmoaudio-ext/src/sources/mod.rs @@ -19,3 +19,9 @@ mod uri_source; #[cfg(feature = "http-stream")] pub use uri_source::UriSource; + +#[cfg(feature = "http-stream")] +mod player_source; + +#[cfg(feature = "http-stream")] +pub use player_source::{PlayerCommand, PlayerEvent, PlayerHandle, PlayerSource}; diff --git a/pmoaudio-ext/src/sources/player_source.rs b/pmoaudio-ext/src/sources/player_source.rs new file mode 100644 index 00000000..7fdfecfa --- /dev/null +++ b/pmoaudio-ext/src/sources/player_source.rs @@ -0,0 +1,540 @@ +//! PlayerSource — Source audio avec contrôle de transport AVTransport UPnP. +//! +//! Implémente un nœud `AudioPipelineNode` qui encapsule le cycle de vie complet +//! d'un renderer UPnP : LoadUri, LoadNextUri, Play, Pause, Stop, Seek. +//! +//! # Architecture +//! +//! ```text +//! PlayerHandle (clonable, envoyé aux handlers UPnP) +//! │ PlayerCommand (mpsc) +//! ▼ +//! PlayerSource (AudioPipelineNode — nœud source) +//! │ AudioSegment (mpsc) +//! ▼ +//! ResamplingNode → ToI24Node → StreamingOggFlacSink → HTTP clients +//! ``` +//! +//! # Gestion de la Pause +//! +//! La Pause ne coupe pas la source brutalement. Elle bloque l'émission de chunks. +//! À la reprise, un `TrackBoundary` est injecté avant les données — cela déclenche +//! EOS + nouveau BOS OGG dans `StreamingOggFlacSink`, garantissant un bitstream +//! propre aligné sur un frame boundary. +//! +//! # Transitions gapless +//! +//! Si `LoadNextUri` a été appelé avant la fin de la piste courante, la transition +//! se fait via un `TrackBoundary` sans interruption du flux OGG. + +use std::sync::Arc; + +use async_trait::async_trait; +use pmometadata::{MemoryTrackMetadata, TrackMetadata}; +use pmoaudio::{ + AudioSegment, SyncMarker, + nodes::AudioError, + pipeline::{AudioPipelineNode, Node, NodeLogic, send_to_children}, +}; +use tokio::sync::{broadcast, mpsc}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, info, warn}; + +use super::uri_source::UriSource; + +// ─── Commandes de transport ─────────────────────────────────────────────────── + +/// Commandes de transport AVTransport UPnP +#[derive(Debug)] +pub enum PlayerCommand { + /// SetAVTransportURI — charge une URI (démarre pas encore) + LoadUri(String), + /// SetNextAVTransportURI — pré-charge pour transition gapless + LoadNextUri(String), + /// Play — démarre ou reprend la lecture + Play, + /// Pause — suspend la lecture, position préservée + Pause, + /// Stop — arrête la lecture, position remise à 0 + Stop, + /// Seek — reprend depuis la position donnée (en secondes) + Seek(f64), +} + +// ─── Événements remontés ────────────────────────────────────────────────────── + +/// Événements remontés par la PlayerSource vers les handlers UPnP +#[derive(Debug, Clone)] +pub enum PlayerEvent { + /// Lecture démarrée ou reprise + Playing { + uri: String, + duration_sec: Option, + }, + /// Lecture suspendue + Paused { + position_sec: f64, + }, + /// Lecture arrêtée + Stopped, + /// Fin de piste (pour que le ControlPoint avance la queue) + TrackEnded, + /// Erreur lors de l'ouverture ou de la lecture + Error(String), +} + +// ─── Handle de contrôle ─────────────────────────────────────────────────────── + +/// Handle de contrôle clonable, envoyé aux handlers UPnP +#[derive(Clone)] +pub struct PlayerHandle { + command_tx: mpsc::Sender, + event_tx: broadcast::Sender, +} + +impl PlayerHandle { + /// Charge une URI (sans démarrer la lecture) + pub async fn load_uri(&self, uri: impl Into) { + let _ = self.command_tx.send(PlayerCommand::LoadUri(uri.into())).await; + } + + /// Pré-charge l'URI suivante pour transition gapless + pub async fn load_next_uri(&self, uri: impl Into) { + let _ = self.command_tx.send(PlayerCommand::LoadNextUri(uri.into())).await; + } + + /// Démarre ou reprend la lecture + pub async fn play(&self) { + let _ = self.command_tx.send(PlayerCommand::Play).await; + } + + /// Suspend la lecture (position préservée) + pub async fn pause(&self) { + let _ = self.command_tx.send(PlayerCommand::Pause).await; + } + + /// Arrête la lecture (position remise à 0) + pub async fn stop(&self) { + let _ = self.command_tx.send(PlayerCommand::Stop).await; + } + + /// Reprend depuis la position donnée (en secondes) + pub async fn seek(&self, pos_sec: f64) { + let _ = self.command_tx.send(PlayerCommand::Seek(pos_sec)).await; + } + + /// Souscrit aux événements de transport + pub fn subscribe_events(&self) -> broadcast::Receiver { + self.event_tx.subscribe() + } +} + +// ─── État de transport ──────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq)] +enum TransportState { + /// Aucune URI chargée + Idle, + /// URI chargée, en attente de Play + Loaded, + /// Lecture en cours + Playing, + /// Lecture suspendue + Paused, +} + +// ─── Logique interne ────────────────────────────────────────────────────────── + +struct PlayerSourceLogic { + command_rx: mpsc::Receiver, + event_tx: broadcast::Sender, +} + +#[async_trait] +impl NodeLogic for PlayerSourceLogic { + async fn process( + &mut self, + _input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut state = TransportState::Idle; + let mut current_uri: Option = None; + let mut next_uri: Option = None; + let mut paused_at_sec: f64 = 0.0; + + info!("PlayerSource: started"); + + loop { + match state { + TransportState::Idle | TransportState::Loaded | TransportState::Paused => { + // Attendre une commande + tokio::select! { + _ = stop_token.cancelled() => { + debug!("PlayerSource: stop requested"); + break; + } + cmd = self.command_rx.recv() => { + match cmd { + None => break, + Some(cmd) => { + self.handle_command( + cmd, &mut state, &mut current_uri, + &mut next_uri, &mut paused_at_sec, + &output, &stop_token, + ).await?; + } + } + } + } + } + + TransportState::Playing => { + let uri = match current_uri.clone() { + Some(u) => u, + None => { + state = TransportState::Idle; + continue; + } + }; + + // Ouvrir la source depuis la position courante + let source = match UriSource::open(&uri, paused_at_sec, stop_token.clone()).await { + Ok(s) => s, + Err(e) => { + warn!("PlayerSource: failed to open {:?}: {}", uri, e); + let _ = self.event_tx.send(PlayerEvent::Error(e.to_string())); + state = TransportState::Idle; + continue; + } + }; + + let duration_sec = source.duration_sec(); + let _ = self.event_tx.send(PlayerEvent::Playing { + uri: uri.clone(), + duration_sec, + }); + info!("PlayerSource: playing {:?} from {:.1}s", uri, paused_at_sec); + + // Pompe audio — s'arrête sur EOF, Pause, Stop, ou commande + let result = self.pump( + source, + &mut state, + &mut current_uri, + &mut next_uri, + &mut paused_at_sec, + &output, + &stop_token, + ).await; + + match result { + Ok(()) => {} + Err(AudioError::ChildDied) => { + debug!("PlayerSource: child died, stopping"); + break; + } + Err(e) => { + warn!("PlayerSource: pump error: {}", e); + let _ = self.event_tx.send(PlayerEvent::Error(e.to_string())); + state = TransportState::Idle; + } + } + } + } + } + + info!("PlayerSource: stopped"); + Ok(()) + } +} + +impl PlayerSourceLogic { + /// Traite une commande de transport dans les états non-Playing. + async fn handle_command( + &mut self, + cmd: PlayerCommand, + state: &mut TransportState, + current_uri: &mut Option, + next_uri: &mut Option, + paused_at_sec: &mut f64, + output: &[mpsc::Sender>], + stop_token: &CancellationToken, + ) -> Result<(), AudioError> { + match cmd { + PlayerCommand::LoadUri(uri) => { + info!("PlayerSource: LoadUri {:?}", uri); + *current_uri = Some(uri); + *next_uri = None; + *paused_at_sec = 0.0; + *state = TransportState::Loaded; + } + + PlayerCommand::LoadNextUri(uri) => { + debug!("PlayerSource: LoadNextUri {:?}", uri); + *next_uri = Some(uri); + } + + PlayerCommand::Play => { + match state { + TransportState::Paused => { + info!("PlayerSource: Play (resume from {:.1}s)", paused_at_sec); + // Injecter un TrackBoundary pour EOS + nouveau BOS OGG propre + send_track_boundary(current_uri.as_deref(), output, *paused_at_sec, stop_token).await?; + *state = TransportState::Playing; + } + TransportState::Loaded => { + info!("PlayerSource: Play (start)"); + *paused_at_sec = 0.0; + // TrackBoundary initial pour le premier BOS OGG + send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await?; + *state = TransportState::Playing; + } + TransportState::Idle => { + debug!("PlayerSource: Play ignored (no URI loaded)"); + } + TransportState::Playing => { + debug!("PlayerSource: Play ignored (already playing)"); + } + } + } + + PlayerCommand::Pause => { + if *state == TransportState::Playing { + // Géré dans pump() — ne devrait pas arriver ici + debug!("PlayerSource: Pause ignored in non-Playing state"); + } else { + debug!("PlayerSource: Pause ignored (not playing)"); + } + } + + PlayerCommand::Stop => { + info!("PlayerSource: Stop"); + *paused_at_sec = 0.0; + *state = TransportState::Idle; + let _ = self.event_tx.send(PlayerEvent::Stopped); + } + + PlayerCommand::Seek(pos) => { + if current_uri.is_some() { + info!("PlayerSource: Seek to {:.1}s", pos); + *paused_at_sec = pos; + send_track_boundary(current_uri.as_deref(), output, pos, stop_token).await?; + *state = TransportState::Playing; + } + } + } + Ok(()) + } + + /// Pompe audio : émet les chunks depuis `source` vers `output`. + /// + /// Surveille simultanément les commandes de contrôle. + /// Se termine quand : EOF, Pause, Stop, cancel, ou erreur. + async fn pump( + &mut self, + source: UriSource, + state: &mut TransportState, + current_uri: &mut Option, + next_uri: &mut Option, + paused_at_sec: &mut f64, + output: &[mpsc::Sender>], + stop_token: &CancellationToken, + ) -> Result<(), AudioError> { + // Canal interne pour recevoir les chunks de UriSource + let (chunk_tx, mut chunk_rx) = mpsc::channel::>(16); + let source_stop = stop_token.child_token(); + let source_stop_clone = source_stop.clone(); + + // Spawner l'émission de la source dans une tâche séparée + let emit_task = tokio::spawn(async move { + source.emit_to_channel(&chunk_tx, &source_stop_clone).await + }); + + let mut result = Ok(()); + + loop { + tokio::select! { + _ = stop_token.cancelled() => { + source_stop.cancel(); + break; + } + + cmd = self.command_rx.recv() => { + match cmd { + None => { + source_stop.cancel(); + break; + } + Some(PlayerCommand::Pause) => { + info!("PlayerSource: Pause at {:.1}s", *paused_at_sec); + source_stop.cancel(); + *state = TransportState::Paused; + let _ = self.event_tx.send(PlayerEvent::Paused { + position_sec: *paused_at_sec, + }); + break; + } + Some(PlayerCommand::Stop) => { + info!("PlayerSource: Stop"); + source_stop.cancel(); + *paused_at_sec = 0.0; + *state = TransportState::Idle; + let _ = self.event_tx.send(PlayerEvent::Stopped); + break; + } + Some(PlayerCommand::LoadUri(uri)) => { + info!("PlayerSource: LoadUri (replacing current) {:?}", uri); + source_stop.cancel(); + *current_uri = Some(uri); + *next_uri = None; + *paused_at_sec = 0.0; + // TrackBoundary pour clore le bitstream OGG proprement + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await { + result = Err(e); + } + *state = TransportState::Playing; + break; + } + Some(PlayerCommand::LoadNextUri(uri)) => { + debug!("PlayerSource: LoadNextUri {:?}", uri); + *next_uri = Some(uri); + } + Some(PlayerCommand::Seek(pos)) => { + info!("PlayerSource: Seek to {:.1}s", pos); + source_stop.cancel(); + *paused_at_sec = pos; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, pos, stop_token).await { + result = Err(e); + break; + } + *state = TransportState::Playing; + break; + } + Some(PlayerCommand::Play) => { + debug!("PlayerSource: Play ignored (already playing)"); + } + } + } + + segment = chunk_rx.recv() => { + match segment { + None => { + // EOF de la source + debug!("PlayerSource: EOF"); + // Signaler la fin de piste dans tous les cas (gapless ou non) + // pour que le ControlPoint avance sa queue et mette à jour son état. + let _ = self.event_tx.send(PlayerEvent::TrackEnded); + if let Some(next) = next_uri.take() { + // Transition gapless vers la piste suivante + info!("PlayerSource: gapless transition to {:?}", next); + *current_uri = Some(next); + *paused_at_sec = 0.0; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stop_token).await { + result = Err(e); + } + *state = TransportState::Playing; + } else { + *state = TransportState::Loaded; + *paused_at_sec = 0.0; + } + break; + } + Some(seg) => { + // Mettre à jour la position courante + if seg.is_audio_chunk() { + *paused_at_sec = seg.timestamp_sec; + } + // Envoyer au pipeline en aval + if let Err(e) = send_to_children("PlayerSource", output, seg).await { + source_stop.cancel(); + result = Err(e); + break; + } + } + } + } + } + } + + // Attendre que la tâche source se termine proprement + let _ = emit_task.await; + result + } +} + +// ─── Helpers ────────────────────────────────────────────────────────────────── + +/// Envoie un TrackBoundary à tous les enfants. +/// +/// Utilisé pour déclencher EOS + nouveau BOS OGG dans StreamingOggFlacSink, +/// garantissant un bitstream propre à chaque démarrage, reprise ou seek. +async fn send_track_boundary( + uri: Option<&str>, + output: &[mpsc::Sender>], + timestamp_sec: f64, + stop_token: &CancellationToken, +) -> Result<(), AudioError> { + if output.is_empty() || stop_token.is_cancelled() { + return Ok(()); + } + + let mut meta = MemoryTrackMetadata::new(); + if let Some(u) = uri { + let _ = meta.set_title(Some(u.to_string())).await; + } + let meta_arc = Arc::new(tokio::sync::RwLock::new(meta)); + let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc); + + send_to_children("PlayerSource", output, boundary).await +} + +// ─── Nœud public ───────────────────────────────────────────────────────────── + +/// Source audio avec contrôle de transport AVTransport UPnP. +/// +/// Utilisée comme nœud racine d'un pipeline audio. Le `PlayerHandle` retourné +/// permet d'envoyer des commandes (Play, Pause, Stop, Seek, LoadUri) depuis +/// les handlers UPnP. +pub struct PlayerSource { + inner: Node, +} + +impl PlayerSource { + /// Crée une nouvelle PlayerSource et son handle de contrôle. + pub fn new() -> (Self, PlayerHandle) { + let (command_tx, command_rx) = mpsc::channel::(32); + let (event_tx, _) = broadcast::channel::(16); + + let logic = PlayerSourceLogic { + command_rx, + event_tx: event_tx.clone(), + }; + + let handle = PlayerHandle { + command_tx, + event_tx, + }; + + ( + Self { + inner: Node::new_source(logic), + }, + handle, + ) + } +} + +#[async_trait] +impl AudioPipelineNode for PlayerSource { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } +} diff --git a/pmowebrenderer/src/handlers.rs b/pmowebrenderer/src/handlers.rs index c29d8152..97503cc9 100644 --- a/pmowebrenderer/src/handlers.rs +++ b/pmowebrenderer/src/handlers.rs @@ -226,13 +226,11 @@ pub fn get_media_info_handler(state: SharedState) -> ActionHandler { // ─── RenderingControl Handlers ────────────────────────────────────────────── -pub fn set_volume_handler(pipeline: PipelineHandle, state: SharedState) -> ActionHandler { +pub fn set_volume_handler(_pipeline: PipelineHandle, state: SharedState) -> ActionHandler { Arc::new(move |data: ActionData| -> ActionFuture { - let pipeline = pipeline.clone(); let state = state.clone(); Box::pin(async move { let volume: u16 = get!(&data, "DesiredVolume", u16); - pipeline.send(PipelineControl::SetVolume(volume)).await; state.write().volume = volume; Ok(data) }) @@ -251,13 +249,11 @@ pub fn get_volume_handler(state: SharedState) -> ActionHandler { }) } -pub fn set_mute_handler(pipeline: PipelineHandle, state: SharedState) -> ActionHandler { +pub fn set_mute_handler(_pipeline: PipelineHandle, state: SharedState) -> ActionHandler { Arc::new(move |data: ActionData| -> ActionFuture { - let pipeline = pipeline.clone(); let state = state.clone(); Box::pin(async move { let mute: bool = get!(&data, "DesiredMute", bool); - pipeline.send(PipelineControl::SetMute(mute)).await; state.write().mute = mute; Ok(data) }) diff --git a/pmowebrenderer/src/pipeline.rs b/pmowebrenderer/src/pipeline.rs index 71c4d2d7..90edd380 100644 --- a/pmowebrenderer/src/pipeline.rs +++ b/pmowebrenderer/src/pipeline.rs @@ -1,55 +1,51 @@ //! Pipeline audio serveur par instance WebRenderer //! //! Chaque instance WebRenderer possède un pipeline indépendant : -//! - Un `StreamingFlacSink` qui encode et diffuse le flux FLAC aux clients HTTP -//! - Un canal de contrôle `PipelineControl` alimenté par les handlers UPnP -//! - Une task background qui orchestre sources et sink +//! - Une `PlayerSource` qui gère le cycle de vie AVTransport (Play/Pause/Stop/Seek/LoadUri) +//! - Un `StreamingOggFlacSink` qui encode et diffuse le flux OGG-FLAC aux clients HTTP +//! - Des nœuds de normalisation (resampling → 96 kHz, conversion → I24) use std::sync::Arc; -use pmoaudio::{AudioSegment, ResamplingNode, ToI24Node}; -use pmometadata::{MemoryTrackMetadata, TrackMetadata}; -use pmoaudio_ext::sinks::{ - OggFlacStreamHandle, StreamingOggFlacSink, -}; -use pmoaudio_ext::UriSource; +use pmoaudio::{AudioPipelineNode, ResamplingNode, ToI24Node}; +use pmoaudio_ext::{PlayerCommand, PlayerHandle, PlayerSource}; +use pmoaudio_ext::sinks::{OggFlacStreamHandle, StreamingOggFlacSink}; use pmoflac::EncoderOptions; -use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, warn}; +use tracing::{debug, warn}; -use crate::messages::PlaybackState; use crate::state::SharedState; -// ─── Commandes de contrôle ─────────────────────────────────────────────────── +// ─── Ré-export des commandes pour les handlers ──────────────────────────────── -/// Commandes envoyées au pipeline audio de l'instance -#[derive(Debug)] -pub enum PipelineControl { - LoadUri(String), - LoadNextUri(String), - Play, - Pause, - Stop, - Seek(f64), - SetVolume(u16), - SetMute(bool), - /// Notification interne : la source courante s'est terminée (EOF ou erreur) - SourceEnded, -} +/// Commandes de transport — alias vers PlayerCommand pour compatibilité handlers +pub use pmoaudio_ext::PlayerCommand as PipelineControl; // ─── Handle vers le pipeline ───────────────────────────────────────────────── -/// Handle partageable vers le pipeline audio d'une instance +/// Handle partageable vers le pipeline audio d'une instance. +/// +/// Expose le `PlayerHandle` pour les commandes AVTransport et le `CancellationToken` +/// pour l'arrêt complet du pipeline. #[derive(Clone)] pub struct PipelineHandle { - pub control_tx: mpsc::Sender, + pub player: PlayerHandle, pub stop_token: CancellationToken, + /// Volume courant (géré localement, pas dans PlayerSource) + state: SharedState, } impl PipelineHandle { + /// Envoie une commande de transport. Gère SetVolume/SetMute localement. pub async fn send(&self, cmd: PipelineControl) { - let _ = self.control_tx.send(cmd).await; + match cmd { + PlayerCommand::LoadUri(uri) => self.player.load_uri(uri).await, + PlayerCommand::LoadNextUri(uri) => self.player.load_next_uri(uri).await, + PlayerCommand::Play => self.player.play().await, + PlayerCommand::Pause => self.player.pause().await, + PlayerCommand::Stop => self.player.stop().await, + PlayerCommand::Seek(pos) => self.player.seek(pos).await, + } } } @@ -75,12 +71,10 @@ impl InstancePipeline { udn: String, ) -> Self { let stop_token = CancellationToken::new(); - let (control_tx, control_rx) = mpsc::channel::(32); - // Chaîne de traitement : ResamplingNode(96kHz) → ToI24Node → StreamingOggFlacSink - // Le broadcast pacé à 0.5s max d'avance, chaque subscribe() est indépendant. use pmoaudio::pipeline::AudioPipelineNode; + // Sink broadcast OGG-FLAC (multi-client, pacé à 0.5s max d'avance) let (sink, flac_handle) = StreamingOggFlacSink::new(EncoderOptions::default(), 24); // Nœud de conversion de profondeur : tout type entier → I24 @@ -91,41 +85,41 @@ impl InstancePipeline { let mut resampler = ResamplingNode::new(96_000); resampler.register(to_i24.boxed()); - // Le tx d'entrée du resampler est le point d'entrée du pipeline - let segment_tx = resampler.get_tx().expect("ResamplingNode doit avoir un sender"); + // Source avec contrôle AVTransport complet + let (mut player_source, player_handle) = PlayerSource::new(); + player_source.register(resampler.boxed()); - let pipeline_handle = PipelineHandle { - control_tx, - stop_token: stop_token.clone(), - }; - - // Lancer la chaîne resampler → to_i24 → sink en background + // Lancer le pipeline en background let sink_stop = stop_token.clone(); tokio::spawn(async move { - if let Err(e) = resampler.boxed().run(sink_stop).await { + if let Err(e) = player_source.boxed().run(sink_stop).await { warn!("Audio pipeline error: {:?}", e); } - debug!("Sink task terminated"); - }); - - // Task pipeline : reçoit les commandes UPnP et pilote les sources - let stop_token_clone = stop_token.clone(); - let control_tx_clone = pipeline_handle.control_tx.clone(); - tokio::spawn(async move { - run_pipeline( - segment_tx, - control_rx, - control_tx_clone, - stop_token_clone, - state, - udn, - #[cfg(feature = "pmoserver")] - control_point, - ) - .await; debug!("Pipeline task terminated"); }); + // Écouter les événements PlayerSource pour mettre à jour le state UPnP + let event_rx = player_handle.subscribe_events(); + let state_clone = state.clone(); + let udn_clone = udn.clone(); + #[cfg(feature = "pmoserver")] + let cp_clone = control_point.clone(); + tokio::spawn(async move { + run_event_listener( + event_rx, + state_clone, + udn_clone, + #[cfg(feature = "pmoserver")] + cp_clone, + ).await; + }); + + let pipeline_handle = PipelineHandle { + player: player_handle, + stop_token: stop_token.clone(), + state, + }; + Self { flac_handle, pipeline_handle, @@ -133,279 +127,63 @@ impl InstancePipeline { } } -// ─── Task principale du pipeline ───────────────────────────────────────────── +// ─── Listener d'événements ──────────────────────────────────────────────────── -async fn run_pipeline( - segment_tx: mpsc::Sender>, - mut control_rx: mpsc::Receiver, - control_tx: mpsc::Sender, - stop_token: CancellationToken, +async fn run_event_listener( + mut event_rx: tokio::sync::broadcast::Receiver, state: SharedState, udn: String, #[cfg(feature = "pmoserver")] control_point: Arc, ) { - let mut current_source_stop: Option = None; - let mut current_uri: Option = None; + use pmoaudio_ext::PlayerEvent; + use crate::messages::PlaybackState; loop { - tokio::select! { - _ = stop_token.cancelled() => { - info!(udn = %udn, "Pipeline stopping by cancellation"); - if let Some(src_stop) = current_source_stop.take() { - src_stop.cancel(); + match event_rx.recv().await { + Ok(event) => match event { + PlayerEvent::Playing { uri, duration_sec } => { + let mut s = state.write(); + s.playback_state = PlaybackState::Playing; + s.current_uri = Some(uri); + s.duration = duration_sec.map(seconds_to_upnp_time); + s.position = None; + // Effacer next_uri/next_metadata : la nouvelle piste est maintenant courante + s.next_uri = None; + s.next_metadata = None; } - break; - } - - cmd = control_rx.recv() => { - match cmd { - None => { - info!(udn = %udn, "Pipeline control channel closed"); - break; - } - - Some(PipelineControl::SourceEnded) => { - // La source s'est terminée (EOF ou erreur) : libérer le slot - debug!(udn = %udn, "Pipeline: SourceEnded — source slot freed"); - current_source_stop = None; - } - - Some(PipelineControl::LoadUri(uri)) => { - info!(udn = %udn, uri = %uri, "Pipeline: LoadUri"); - if let Some(src_stop) = current_source_stop.take() { - src_stop.cancel(); - } - current_uri = Some(uri.clone()); - { - let mut s = state.write(); - s.playback_state = PlaybackState::Transitioning; - s.current_uri = Some(uri.clone()); - s.position = None; - } - let src_stop = stop_token.child_token(); - current_source_stop = Some(src_stop.clone()); - let tx = segment_tx.clone(); - let notify = control_tx.clone(); - let st = state.clone(); - let udn_c = udn.clone(); - #[cfg(feature = "pmoserver")] + PlayerEvent::Paused { position_sec } => { + let mut s = state.write(); + s.playback_state = PlaybackState::Paused; + s.position = Some(seconds_to_upnp_time(position_sec)); + } + PlayerEvent::Stopped => { + let mut s = state.write(); + s.playback_state = PlaybackState::Stopped; + s.position = None; + } + PlayerEvent::TrackEnded => { + #[cfg(feature = "pmoserver")] + { let cp = control_point.clone(); + let udn_c = udn.clone(); tokio::spawn(async move { - stream_source( - uri, 0.0, tx, src_stop, st, udn_c, - #[cfg(feature = "pmoserver")] - cp, - ).await; - let _ = notify.send(PipelineControl::SourceEnded).await; + cp.advance_queue_and_prefetch(&pmocontrol::DeviceId(udn_c)); }); } - - Some(PipelineControl::LoadNextUri(uri)) => { - debug!(udn = %udn, uri = %uri, "Pipeline: LoadNextUri"); - state.write().next_uri = Some(uri); - } - - Some(PipelineControl::Play) => { - // Redémarrer la source si elle n'est pas en cours - if current_source_stop.is_none() { - if let Some(uri) = current_uri.clone() { - // Reprendre depuis la position pausée (0.0 si pas de pause) - let seek = state.read().position.as_deref() - .map(upnp_time_to_seconds) - .unwrap_or(0.0); - info!(udn = %udn, uri = %uri, seek = seek, "Pipeline: Play — restarting source"); - let src_stop = stop_token.child_token(); - current_source_stop = Some(src_stop.clone()); - let tx = segment_tx.clone(); - let notify = control_tx.clone(); - let st = state.clone(); - let udn_c = udn.clone(); - #[cfg(feature = "pmoserver")] - let cp = control_point.clone(); - tokio::spawn(async move { - stream_source( - uri, seek, tx, src_stop, st, udn_c, - #[cfg(feature = "pmoserver")] - cp, - ).await; - let _ = notify.send(PipelineControl::SourceEnded).await; - }); - } else { - debug!(udn = %udn, "Pipeline: Play — no URI loaded, ignoring"); - } - } else { - debug!(udn = %udn, "Pipeline: Play — source already running"); - } - } - - Some(PipelineControl::Pause) => { - info!(udn = %udn, "Pipeline: Pause — stopping source, position preserved"); - if let Some(src_stop) = current_source_stop.take() { - src_stop.cancel(); - } - state.write().playback_state = PlaybackState::Paused; - // position reste dans state pour la reprise via Play - } - - Some(PipelineControl::Stop) => { - info!(udn = %udn, "Pipeline: Stop"); - if let Some(src_stop) = current_source_stop.take() { - src_stop.cancel(); - } - // Conserver current_uri pour permettre un Play ultérieur - let mut s = state.write(); - s.playback_state = PlaybackState::Stopped; - s.position = None; - } - - Some(PipelineControl::Seek(pos_sec)) => { - info!(udn = %udn, pos = pos_sec, "Pipeline: Seek"); - if let Some(uri) = current_uri.clone() { - if let Some(src_stop) = current_source_stop.take() { - src_stop.cancel(); - } - let src_stop = stop_token.child_token(); - current_source_stop = Some(src_stop.clone()); - let tx = segment_tx.clone(); - let notify = control_tx.clone(); - let st = state.clone(); - let udn_c = udn.clone(); - #[cfg(feature = "pmoserver")] - let cp = control_point.clone(); - tokio::spawn(async move { - stream_source( - uri, pos_sec, tx, src_stop, st, udn_c, - #[cfg(feature = "pmoserver")] - cp, - ).await; - let _ = notify.send(PipelineControl::SourceEnded).await; - }); - } - } - - Some(PipelineControl::SetVolume(vol)) => { - state.write().volume = vol; - } - - Some(PipelineControl::SetMute(mute)) => { - state.write().mute = mute; - } } - } - } - } -} - -// ─── Task source ───────────────────────────────────────────────────────────── - -async fn stream_source( - uri: String, - seek_sec: f64, - tx: mpsc::Sender>, - stop_token: CancellationToken, - state: SharedState, - udn: String, - #[cfg(feature = "pmoserver")] - control_point: Arc, -) { - info!(udn = %udn, uri = %uri, seek = seek_sec, "Source task: opening URI"); - - match UriSource::open(&uri, seek_sec, stop_token.clone()).await { - Ok(source) => { - debug!(udn = %udn, "Source task: URI opened, duration={:?}", source.duration_sec()); - if let Some(dur) = source.duration_sec() { - state.write().duration = Some(seconds_to_upnp_time(dur)); - } - - // TrackBoundary avec métadonnées minimales - let boundary = { - let mut meta = MemoryTrackMetadata::new(); - let _ = meta.set_title(Some(uri.clone())).await; - let meta_arc = Arc::new(tokio::sync::RwLock::new(meta)); - AudioSegment::new_track_boundary(0, seek_sec, meta_arc) - }; - let _ = tx.send(boundary).await; - - // L'URI est ouverte, le flux va commencer à couler. - // Le sink bloquera naturellement si aucun client HTTP n'est connecté. - { - let mut s = state.write(); - s.playback_state = PlaybackState::Playing; - if seek_sec > 0.0 { - s.position = Some(seconds_to_upnp_time(seek_sec)); - } - } - - match source.emit_to_channel(&tx, &stop_token).await { - Ok(true) => { - debug!(udn = %udn, "Source task: EOF → TrackEnded"); - handle_track_ended( - state, udn, tx, stop_token, - #[cfg(feature = "pmoserver")] - control_point, - ).await; - } - Ok(false) => { - debug!(udn = %udn, "Source task: cancelled"); - } - Err(e) => { - warn!(udn = %udn, error = %e, "Source task error"); + PlayerEvent::Error(e) => { + tracing::warn!(udn = %udn, "PlayerSource error: {}", e); state.write().playback_state = PlaybackState::Stopped; } + }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!(udn = %udn, "Event listener lagged {} events", n); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + break; } } - Err(e) => { - warn!(udn = %udn, error = %e, "Source task: failed to open URI"); - state.write().playback_state = PlaybackState::Stopped; - } - } -} - -// ─── TrackEnded : avancer current←next ─────────────────────────────────────── - -async fn handle_track_ended( - state: SharedState, - udn: String, - tx: mpsc::Sender>, - stop_token: CancellationToken, - #[cfg(feature = "pmoserver")] - control_point: Arc, -) { - let next_uri = { - let mut s = state.write(); - let uri = s.next_uri.take(); - let meta = s.next_metadata.take(); - s.current_uri = uri.clone(); - s.current_metadata = meta; - s.next_uri = None; - s.next_metadata = None; - s.position = None; - s.duration = None; - if uri.is_some() { - s.playback_state = PlaybackState::Playing; - } else { - s.playback_state = PlaybackState::Stopped; - } - uri - }; - - #[cfg(feature = "pmoserver")] - if next_uri.is_some() { - let cp = control_point.clone(); - let udn_clone = udn.clone(); - tokio::spawn(async move { - cp.advance_queue_and_prefetch(&pmocontrol::DeviceId(udn_clone)); - }); - } - - if let Some(uri) = next_uri { - info!(udn = %udn, uri = %uri, "TrackEnded: starting next track"); - Box::pin(stream_source( - uri, 0.0, tx, stop_token, state, udn, - #[cfg(feature = "pmoserver")] - control_point, - )).await; } }