From de74c7b431ec12ecf9840daca02756ed53dcff14 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 28 Feb 2026 14:10:38 +0100 Subject: [PATCH] Migrate instance ID storage from localStorage to sessionStorage Switch instance ID storage from localStorage to sessionStorage in useWebRenderer composable to ensure better session management and prevent data persistence across browser sessions. Also update the DirectOggFlacSink implementation to improve backpressure handling, streamline the OGG chaining logic, and add proper Safari range header support for the stream endpoint. --- .DS_Store | Bin 18436 -> 18436 bytes .../webapp/src/composables/useWebRenderer.ts | 4 +- .../src/sinks/direct_ogg_flac_sink.rs | 401 +++++++----------- pmowebrenderer/src/registry.rs | 9 +- pmowebrenderer/src/stream.rs | 44 +- 5 files changed, 187 insertions(+), 271 deletions(-) diff --git a/.DS_Store b/.DS_Store index 8c67cf49f9d59ef73bc232093ccfd0f87fcec91d..39adbca77b0359db4eefe16c8f2a8b64fb334f54 100644 GIT binary patch delta 72 zcmZpfz}PZ@ae_Z%@5X@L{ETLsc?6O;*i0>rbQBCH`x(hjKB}Oyd8>pz8z%=t1sLj0 Z)-w}hV=7``V7<5brW_OFW^RR_>Hzym6f*z- delta 55 zcmV-70LcG@kO72{0gz7tjImI=4*@l^3=m=n2rV->EFdv6GLuj;8`F9wqo NANP@fFtZCF`6uEU5zqht diff --git a/pmoapp/webapp/src/composables/useWebRenderer.ts b/pmoapp/webapp/src/composables/useWebRenderer.ts index 246fcca6..93bb6a2a 100644 --- a/pmoapp/webapp/src/composables/useWebRenderer.ts +++ b/pmoapp/webapp/src/composables/useWebRenderer.ts @@ -34,10 +34,10 @@ function generateUUID(): string { function getOrCreateInstanceId(): string { try { - let id = localStorage.getItem(INSTANCE_ID_KEY); + let id = sessionStorage.getItem(INSTANCE_ID_KEY); if (!id) { id = generateUUID(); - localStorage.setItem(INSTANCE_ID_KEY, id); + sessionStorage.setItem(INSTANCE_ID_KEY, id); } return id; } catch { diff --git a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs index 343567f4..92b4409b 100644 --- a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs @@ -1,35 +1,28 @@ //! DirectOggFlacSink — nœud puits OGG-FLAC pour un seul client HTTP. //! -//! Combine la logique de backpressure/reconnexion de `DirectFlacSink` -//! avec l'encodage OGG-FLAC de `StreamingOggFlacSink`. -//! -//! # Cycle de vie -//! -//! - **Play** : le navigateur appelle `GET /stream`. `connect()` crée un channel -//! Bytes (bytes_tx/rx), une task de forwarding qui copie les Bytes dans un -//! DuplexStream, et lance le premier encodeur OGG-FLAC. Le `bytes_tx` est -//! stocké dans le sink pour le chaining TrackBoundary. -//! - **Stop** : le navigateur ferme la connexion. Le DuplexStream se rompt, -//! la task de forwarding se termine, le bytes_tx devient invalide. Le sink -//! voit `pcm_tx.send()` échouer et bloque sur `client_notify`. -//! - **TrackBoundary** : le sink ferme le `pcm_tx` courant (EOF → encodeur écrit -//! EOS OGG), attend la fin de l'encodeur, puis relance un nouvel encodeur -//! dans le même `bytes_tx` (OGG chaining : nouvelle BOS OGG dans le même flux HTTP). -//! - **Play suivant** : `connect()` → nouveau DuplexStream + channel → nouveau pipe. -//! //! # Architecture //! //! ```text //! AudioSegment I24 @ 96 kHz -//! ↓ NodeLogic::process() [bloque si pas de client] +//! ↓ NodeLogic::process() //! chunk_to_pcm_bytes() → PCM 24-bit LE -//! ↓ SharedPcmTx -//! ByteStreamReader → encode_flac_stream() → OGG pages -//! ↓ SharedBytesTx (mpsc::Sender) ← persistant entre encodeurs -//! [forwarding task] → tokio::io::DuplexStream -//! ↓ DirectOggFlacStream (AsyncRead) → Body HTTP +//! ↓ pcm_tx (cap=2) +//! ByteStreamReader → encode_flac_stream() → OGG pages (Bytes) +//! ↓ ogg_tx mpsc::Sender (cap=8, backpressure naturelle) +//! Arc>> → DirectOggFlacStream (AsyncRead) → HTTP → Safari //! ``` +//! +//! # Backpressure +//! +//! Safari lent → ogg_rx plein → ogg_tx.send() bloque → encodeur bloque +//! → pcm_tx.send() bloque → pipeline audio bloque. +//! +//! # OGG chaining (TrackBoundary) +//! +//! Fermer pcm_tx → encodeur termine (EOS) → lancer nouvel encodeur. +//! Le même ogg_tx est réutilisé : le stream HTTP est continu. +use std::collections::VecDeque; use std::io; use std::pin::Pin; use std::sync::Arc; @@ -46,7 +39,7 @@ use tokio::io::{AsyncRead, ReadBuf}; use tokio::sync::{mpsc, watch, Mutex}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; -use tracing::{debug, warn}; +use tracing::{debug, trace, warn}; use crate::sinks::byte_stream_reader::{ByteStreamReader, PcmChunk}; use crate::sinks::chunk_to_pcm::chunk_to_pcm_bytes; @@ -57,129 +50,54 @@ pub const DIRECT_OGG_FLAC_SAMPLE_RATE: u32 = 96_000; pub const DIRECT_OGG_FLAC_CHANNELS: u8 = 2; pub const DIRECT_OGG_FLAC_BITS_PER_SAMPLE: u8 = 24; -/// Capacité du pipe duplex (~256 KB). -const PIPE_CAPACITY: usize = 256 * 1024; -/// Capacité du channel Bytes intermédiaire. -const BYTES_CHANNEL_CAPACITY: usize = 64; +/// Capacité du canal OGG → HTTP (en chunks de ~8 KB). +/// Fournit un petit buffer sans casser la backpressure TCP. +const OGG_CHANNEL_CAPACITY: usize = 8; -// ─── Shared state ───────────────────────────────────────────────────────────── +// ─── Types partagés ─────────────────────────────────────────────────────────── type SharedPcmTx = Arc>>>; -/// Canal Bytes persistant entre les encodeurs successifs (OGG chaining). -/// Le sink y envoie les pages OGG ; une task de forwarding les copie dans le DuplexStream. -type SharedBytesTx = Arc>>>; -/// Handle de la task encodeur courante. type SharedEncoderTask = Arc>>>; +type SharedOggRx = Arc>>; // ─── Handle public ──────────────────────────────────────────────────────────── -/// Handle vers le sink, cloneable, reconnectable à chaque Play. #[derive(Clone)] pub struct DirectOggFlacHandle { pcm_tx: SharedPcmTx, - client_connect_tx: Arc>, - client_notify_internal: Arc, - first_byte_tx: Arc>, + encoder_task: SharedEncoderTask, encoder_options: EncoderOptions, current_timestamp: Arc>, - /// Canal Bytes persistant partagé avec la logic du sink pour le OGG chaining. - bytes_tx: SharedBytesTx, - /// Task encodeur courante partagée avec la logic du sink. - encoder_task: SharedEncoderTask, + /// Canal OGG partagé avec le stream HTTP. + ogg_tx: mpsc::Sender, + ogg_rx: SharedOggRx, + /// Notifié quand get_stream() est appelé (premier client). + client_notify: Arc, } impl DirectOggFlacHandle { - /// Crée un nouveau flux OGG-FLAC pour le client HTTP. - /// Remplace toute connexion précédente. - pub async fn connect(&self) -> DirectOggFlacStream { - let connect_count_before = *self.client_connect_tx.borrow(); - debug!("DirectOggFlacHandle::connect() called, connect_count={}", connect_count_before); - - // Annuler l'encodeur précédent s'il tourne encore - if let Some(old_task) = self.encoder_task.lock().await.take() { - old_task.abort(); - } - - // Créer le channel Bytes persistant (OGG chaining) - let (bytes_tx, bytes_rx) = mpsc::channel::(BYTES_CHANNEL_CAPACITY); - *self.bytes_tx.lock().await = Some(bytes_tx.clone()); - - // Créer le DuplexStream vers le client HTTP - let (mut pipe_writer, pipe_reader) = tokio::io::duplex(PIPE_CAPACITY); - - // Task de forwarding : Bytes → DuplexStream - // Se termine quand bytes_rx est fermé (bytes_tx droppé) ou pipe cassé - tokio::spawn(async move { - use tokio::io::AsyncWriteExt; - let mut rx = bytes_rx; - while let Some(bytes) = rx.recv().await { - if pipe_writer.write_all(&bytes).await.is_err() { - debug!("DirectOggFlacStream forwarder: pipe broken, client disconnected"); - break; - } - } - debug!("DirectOggFlacStream forwarder: done"); - }); - - // Réinitialiser les signaux - let _ = self.first_byte_tx.send(false); - *self.current_timestamp.write().await = 0.0; - - // Créer le premier pcm_tx + encodeur - let (pcm_tx, pcm_rx) = mpsc::channel::(8); - let current_dur = Arc::new(tokio::sync::RwLock::new(0.0f64)); - let pcm_reader = ByteStreamReader::new(pcm_rx, self.current_timestamp.clone(), current_dur); - - *self.pcm_tx.lock().await = Some(pcm_tx); - - let new_count = connect_count_before.wrapping_add(1); - let _ = self.client_connect_tx.send(new_count); - self.client_notify_internal.notify_one(); - debug!("DirectOggFlacHandle::connect() client_connect_count -> {}", new_count); - - let options = self.encoder_options.clone(); - let current_timestamp = self.current_timestamp.clone(); - let shared_bytes_tx = self.bytes_tx.clone(); - - let handle = tokio::spawn(async move { - debug!("DirectOggFlacHandle: initial encoder task started"); - if let Err(e) = run_ogg_encoder(pcm_reader, bytes_tx, shared_bytes_tx, options, current_timestamp).await { - debug!("DirectOggFlacHandle: initial encoder stopped: {}", e); - } - debug!("DirectOggFlacHandle: initial encoder task ended"); - }); - - *self.encoder_task.lock().await = Some(handle); - - debug!("DirectOggFlacHandle::connect() returning DirectOggFlacStream"); + /// Retourne le stream OGG-FLAC pour le handler HTTP. + /// Lance le premier encodeur au premier appel. + pub fn get_stream(&self) -> DirectOggFlacStream { + debug!("DirectOggFlacHandle::get_stream()"); + self.client_notify.notify_one(); DirectOggFlacStream { - inner: pipe_reader, - first_byte_tx: Some(self.first_byte_tx.clone()), + ogg_rx: self.ogg_rx.clone(), + buffer: VecDeque::new(), } } - pub fn first_byte_ready(&self) -> watch::Receiver { - self.first_byte_tx.subscribe() - } - pub async fn current_position_sec(&self) -> f64 { *self.current_timestamp.read().await } - - pub async fn wait_for_client(&self) { - let seen = *self.client_connect_tx.borrow(); - debug!("DirectOggFlacHandle::wait_for_client() called, seen connect_count={}", seen); - let mut rx = self.client_connect_tx.subscribe(); - let _ = rx.wait_for(|v| *v > seen).await; - debug!("DirectOggFlacHandle::wait_for_client() unblocked"); - } } // ─── Stream public ──────────────────────────────────────────────────────────── +/// AsyncRead sur le canal OGG-FLAC. pub struct DirectOggFlacStream { - inner: tokio::io::DuplexStream, - first_byte_tx: Option>>, + ogg_rx: SharedOggRx, + buffer: VecDeque, } impl AsyncRead for DirectOggFlacStream { @@ -188,18 +106,36 @@ impl AsyncRead for DirectOggFlacStream { cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll> { - let filled_before = buf.filled().len(); - let result = Pin::new(&mut self.inner).poll_read(cx, buf); - if let Poll::Ready(Ok(())) = &result { - let filled_after = buf.filled().len(); - if filled_after > filled_before { - if let Some(tx) = self.first_byte_tx.take() { - debug!("DirectOggFlacStream: first {} bytes sent to HTTP client", filled_after - filled_before); - let _ = tx.send(true); - } - } + // Vider le buffer interne en premier + if !self.buffer.is_empty() { + let to_copy = self.buffer.len().min(buf.remaining()); + let chunk: Vec = self.buffer.drain(..to_copy).collect(); + buf.put_slice(&chunk); + return Poll::Ready(Ok(())); + } + + // Tenter de recevoir le prochain chunk OGG + let mut guard = match self.ogg_rx.try_lock() { + Ok(g) => g, + Err(_) => { + cx.waker().wake_by_ref(); + return Poll::Pending; + } + }; + + match guard.poll_recv(cx) { + Poll::Ready(Some(bytes)) => { + drop(guard); + let to_copy = bytes.len().min(buf.remaining()); + buf.put_slice(&bytes[..to_copy]); + if to_copy < bytes.len() { + self.buffer.extend(&bytes[to_copy..]); + } + Poll::Ready(Ok(())) + } + Poll::Ready(None) => Poll::Ready(Ok(())), // EOF + Poll::Pending => Poll::Pending, } - result } } @@ -207,13 +143,11 @@ impl AsyncRead for DirectOggFlacStream { struct DirectOggFlacSinkLogic { pcm_tx: SharedPcmTx, - client_notify: Arc, - encoder_options: EncoderOptions, encoder_task: SharedEncoderTask, - bytes_tx: SharedBytesTx, + encoder_options: EncoderOptions, current_timestamp: Arc>, - /// Vrai dès qu'au moins un chunk audio a été encodé dans le stream courant. - /// Empêche le OGG chaining sur le TrackBoundary initial (avant tout audio). + ogg_tx: mpsc::Sender, + client_notify: Arc, has_encoded_frames: bool, } @@ -229,6 +163,15 @@ impl NodeLogic for DirectOggFlacSinkLogic { AudioError::ProcessingError("DirectOggFlacSink requires an input".into()) })?; + // Attendre le premier client Safari avant de démarrer l'encodeur + debug!("DirectOggFlacSink: waiting for first client..."); + tokio::select! { + _ = stop_token.cancelled() => return Ok(()), + _ = self.client_notify.notified() => {} + } + debug!("DirectOggFlacSink: first client connected, starting encoder"); + self.start_encoder().await; + loop { tokio::select! { _ = stop_token.cancelled() => { @@ -244,25 +187,9 @@ impl NodeLogic for DirectOggFlacSinkLogic { } Some(seg) => match &seg.segment { _AudioSegment::Chunk(chunk) => { - // Attendre un client si nécessaire (backpressure quand pas de Play) - loop { - let tx_opt = self.pcm_tx.lock().await.clone(); - if tx_opt.is_some() { - break; - } - debug!("DirectOggFlacSink: no pcm_tx, waiting for client_notify..."); - tokio::select! { - _ = stop_token.cancelled() => { - debug!("DirectOggFlacSink: cancelled while waiting for client"); - return Ok(()); - } - _ = self.client_notify.notified() => { - debug!("DirectOggFlacSink: client_notify received, rechecking pcm_tx"); - } - } - } + let tx = self.pcm_tx.lock().await.clone(); + let Some(tx) = tx else { continue; }; - let tx = self.pcm_tx.lock().await.clone().unwrap(); let pcm_bytes = chunk_to_pcm_bytes(chunk, DIRECT_OGG_FLAC_BITS_PER_SAMPLE)?; let duration_sec = chunk.len() as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64; let pcm_chunk = PcmChunk { @@ -271,11 +198,7 @@ impl NodeLogic for DirectOggFlacSinkLogic { duration_sec, }; if tx.send(pcm_chunk).await.is_err() { - warn!( - ts = seg.timestamp_sec, - "DirectOggFlacSink: chunk dropped (client disconnected at {:.3}s), waiting for reconnect", - seg.timestamp_sec, - ); + warn!("DirectOggFlacSink: encoder gone at {:.3}s", seg.timestamp_sec); *self.pcm_tx.lock().await = None; self.has_encoded_frames = false; } else { @@ -289,11 +212,12 @@ impl NodeLogic for DirectOggFlacSinkLogic { self.has_encoded_frames = false; self.do_track_boundary().await; } else { - debug!("DirectOggFlacSink: TrackBoundary ignored (no frames encoded yet)"); + debug!("DirectOggFlacSink: TrackBoundary ignored (no frames)"); } } SyncMarker::EndOfStream => { debug!("DirectOggFlacSink: EndOfStream"); + self.stop_encoder().await; } _ => {} }, @@ -303,6 +227,7 @@ impl NodeLogic for DirectOggFlacSinkLogic { } } + self.stop_encoder().await; Ok(()) } @@ -312,71 +237,56 @@ impl NodeLogic for DirectOggFlacSinkLogic { } impl DirectOggFlacSinkLogic { - /// OGG chaining : ferme l'encodeur courant (EOF → EOS OGG), attend sa fin, - /// puis relance un nouvel encodeur dans le même channel Bytes (nouvelle BOS OGG). - async fn do_track_boundary(&mut self) { - // 1. Fermer le pcm_tx courant → EOF dans ByteStreamReader → encodeur écrit EOS OGG - { - let mut guard = self.pcm_tx.lock().await; - *guard = None; - } - - // 2. Attendre la fin de la task encodeur courante - let old_task = self.encoder_task.lock().await.take(); - if let Some(handle) = old_task { - let _ = handle.await; - debug!("DirectOggFlacSink: previous encoder task joined"); - } - - // 3. Vérifier que le bytes_tx est encore valide (client pas déconnecté) - let bytes_tx = { - let guard = self.bytes_tx.lock().await; - guard.clone() - }; - let Some(bytes_tx) = bytes_tx else { - debug!("DirectOggFlacSink: bytes_tx gone (client disconnected), skip OGG chaining"); - return; - }; - - // 4. Nouveau pcm_tx + ByteStreamReader - let (pcm_tx, pcm_rx) = mpsc::channel::(8); + async fn start_encoder(&mut self) { + let (pcm_tx, pcm_rx) = mpsc::channel::(2); let current_dur = Arc::new(tokio::sync::RwLock::new(0.0f64)); - let pcm_reader = ByteStreamReader::new( - pcm_rx, - self.current_timestamp.clone(), - current_dur, - ); + let pcm_reader = ByteStreamReader::new(pcm_rx, self.current_timestamp.clone(), current_dur); *self.pcm_tx.lock().await = Some(pcm_tx); - // 5. Relancer l'encodeur dans le même bytes_tx (OGG chaining : nouvelle BOS OGG) + let ogg_tx = self.ogg_tx.clone(); let options = self.encoder_options.clone(); let current_timestamp = self.current_timestamp.clone(); - let shared_bytes_tx = self.bytes_tx.clone(); - let handle = tokio::spawn(async move { - debug!("DirectOggFlacSink: chained encoder task started"); - if let Err(e) = run_ogg_encoder(pcm_reader, bytes_tx, shared_bytes_tx, options, current_timestamp).await { - debug!("DirectOggFlacSink: chained encoder stopped: {}", e); + debug!("DirectOggFlacSink: encoder task started"); + if let Err(e) = run_ogg_encoder(pcm_reader, ogg_tx, options, current_timestamp).await { + debug!("DirectOggFlacSink: encoder stopped: {}", e); } - debug!("DirectOggFlacSink: chained encoder task ended"); + debug!("DirectOggFlacSink: encoder task ended"); }); - *self.encoder_task.lock().await = Some(handle); - debug!("DirectOggFlacSink: OGG chaining complete, new encoder started"); + } + + async fn stop_encoder(&mut self) { + *self.pcm_tx.lock().await = None; + if let Some(handle) = self.encoder_task.lock().await.take() { + let _ = handle.await; + } + self.has_encoded_frames = false; + } + + async fn do_track_boundary(&mut self) { + // Fermer pcm_tx → encodeur écrit EOS, se termine + *self.pcm_tx.lock().await = None; + if let Some(handle) = self.encoder_task.lock().await.take() { + let _ = handle.await; + debug!("DirectOggFlacSink: previous encoder joined"); + } + // Démarrer le nouvel encodeur sur le même ogg_tx + self.start_encoder().await; + debug!("DirectOggFlacSink: OGG chaining complete"); } } // ─── Encodeur FLAC + wrapper OGG ───────────────────────────────────────────── -/// Encode PCM → FLAC → OGG et envoie les pages OGG dans `bytes_tx`. -/// Quand le channel devient invalide (client déconnecté), nettoie `shared_bytes_tx`. async fn run_ogg_encoder( pcm_reader: ByteStreamReader, - bytes_tx: mpsc::Sender, - shared_bytes_tx: SharedBytesTx, + ogg_tx: mpsc::Sender, options: EncoderOptions, _current_timestamp: Arc>, ) -> Result<(), AudioError> { + use tokio::io::AsyncReadExt; + let format = PcmFormat { sample_rate: DIRECT_OGG_FLAC_SAMPLE_RATE, channels: DIRECT_OGG_FLAC_CHANNELS, @@ -394,36 +304,34 @@ async fn run_ogg_encoder( let mut ogg = OggPageWriter::new(stream_serial); let ogg_flac_id = create_ogg_flac_identification(&flac_header)?; - let bos_page = Bytes::from(ogg.create_page(&ogg_flac_id, true, false, false)); + let bos_page = ogg.create_page(&ogg_flac_id, true, false, false); let vorbis_comment = create_empty_vorbis_comment(); - let comment_page = Bytes::from(ogg.create_page(&vorbis_comment, false, false, false)); + let comment_page = ogg.create_page(&vorbis_comment, false, false, false); - macro_rules! send_or_cleanup { - ($page:expr) => { - if bytes_tx.send($page).await.is_err() { - debug!("DirectOggFlacSink: bytes_tx broken, client disconnected"); - *shared_bytes_tx.lock().await = None; - return Ok(()); - } - }; + let mut header = Vec::new(); + header.extend_from_slice(&bos_page); + header.extend_from_slice(&comment_page); + if ogg_tx.send(Bytes::from(header)).await.is_err() { + debug!("DirectOggFlacSink: ogg_tx closed on headers"); + return Ok(()); } - send_or_cleanup!(bos_page); - send_or_cleanup!(comment_page); - - use tokio::io::AsyncReadExt; use crate::sinks::flac_frame_utils::{validate_frame_header_crc, parse_flac_block_size}; let mut encoded_samples = 0u64; - let mut read_buffer = vec![0u8; 16384]; + let mut read_buffer = vec![0u8; 65536]; let mut accumulator: Vec = Vec::with_capacity(32768); loop { match flac_stream.read(&mut read_buffer).await { Ok(0) => { - // EOF : page EOS finale - let eos_page = Bytes::from(ogg.create_page(&accumulator, false, true, false)); - let _ = bytes_tx.send(eos_page).await; + trace!( + "OggEncoder: EOF — accum={} B, encoded={:.3}s", + accumulator.len(), + encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64, + ); + let eos_page = ogg.create_page(&accumulator, false, true, false); + let _ = ogg_tx.send(Bytes::from(eos_page)).await; break; } Ok(n) => { @@ -460,14 +368,16 @@ async fn run_ogg_encoder( encoded_samples = encoded_samples.saturating_add(first_samples as u64); ogg.add_samples(first_samples as u64); - let ogg_page = Bytes::from(ogg.create_page(&frame, false, false, false)); - if bytes_tx.send(ogg_page).await.is_err() { - warn!( - "DirectOggFlacSink: bytes_tx broken after {} samples ({:.3}s), client disconnected", - encoded_samples, - encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64, - ); - *shared_bytes_tx.lock().await = None; + trace!( + "OggEncoder: frame {} B, {:.3}s total", + frame.len(), + encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64, + ); + + let ogg_page = ogg.create_page(&frame, false, false, false); + if ogg_tx.send(Bytes::from(ogg_page)).await.is_err() { + warn!("DirectOggFlacSink: ogg_tx closed after {:.3}s", + encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64); return Ok(()); } } @@ -624,22 +534,20 @@ pub struct DirectOggFlacSink { impl DirectOggFlacSink { pub fn new(encoder_options: EncoderOptions) -> (Self, DirectOggFlacHandle) { let pcm_tx: SharedPcmTx = Arc::new(Mutex::new(None)); - let client_notify_internal = Arc::new(tokio::sync::Notify::new()); - let (client_connect_tx, _) = watch::channel(0u64); - let client_connect_tx = Arc::new(client_connect_tx); - let (first_byte_tx, _) = watch::channel(false); - let first_byte_tx = Arc::new(first_byte_tx); - let current_timestamp = Arc::new(tokio::sync::RwLock::new(0.0f64)); - let bytes_tx: SharedBytesTx = Arc::new(Mutex::new(None)); let encoder_task: SharedEncoderTask = Arc::new(Mutex::new(None)); + let current_timestamp = Arc::new(tokio::sync::RwLock::new(0.0f64)); + let client_notify = Arc::new(tokio::sync::Notify::new()); + + let (ogg_tx, ogg_rx) = mpsc::channel::(OGG_CHANNEL_CAPACITY); + let ogg_rx = Arc::new(Mutex::new(ogg_rx)); let logic = DirectOggFlacSinkLogic { pcm_tx: pcm_tx.clone(), - client_notify: client_notify_internal.clone(), - encoder_options: encoder_options.clone(), encoder_task: encoder_task.clone(), - bytes_tx: bytes_tx.clone(), + encoder_options: encoder_options.clone(), current_timestamp: current_timestamp.clone(), + ogg_tx: ogg_tx.clone(), + client_notify: client_notify.clone(), has_encoded_frames: false, }; @@ -649,13 +557,12 @@ impl DirectOggFlacSink { let handle = DirectOggFlacHandle { pcm_tx, - client_connect_tx, - client_notify_internal, - first_byte_tx, + encoder_task, encoder_options, current_timestamp, - bytes_tx, - encoder_task, + ogg_tx, + ogg_rx, + client_notify, }; (sink, handle) diff --git a/pmowebrenderer/src/registry.rs b/pmowebrenderer/src/registry.rs index 40cbc95e..7e1c81eb 100644 --- a/pmowebrenderer/src/registry.rs +++ b/pmowebrenderer/src/registry.rs @@ -116,15 +116,16 @@ impl RendererRegistry { Ok((stream_url, udn)) } - /// Retourne le DirectOggFlacHandle pour l'endpoint /stream (clonable). - pub fn get_flac_handle( + /// Retourne un DirectOggFlacStream pour l'endpoint /stream. + /// Chaque appel retourne un wrapper sur le même reader persistant. + pub fn get_stream( &self, instance_id: &str, - ) -> Option { + ) -> Option { self.instances .read() .get(instance_id) - .map(|i| i.flac_handle.clone()) + .map(|i| i.flac_handle.get_stream()) } /// Retourne le PipelineHandle par UDN (pour les handlers UPnP) diff --git a/pmowebrenderer/src/stream.rs b/pmowebrenderer/src/stream.rs index b147c986..2e759ebb 100644 --- a/pmowebrenderer/src/stream.rs +++ b/pmowebrenderer/src/stream.rs @@ -1,15 +1,17 @@ //! Handler HTTP GET /api/webrenderer/{id}/stream //! -//! Sert le flux FLAC d'une instance WebRenderer via DirectFlacSink. +//! Sert le flux OGG-FLAC d'une instance WebRenderer. //! -//! Reconnectable : appelé à chaque Play, crée un nouveau pipe + encodeur. +//! Safari fait systématiquement une requête Range: bytes=0-1 avant de jouer. +//! On répond 206 avec 2 octets factices pour satisfaire la sonde, +//! puis la vraie requête (sans Range) reçoit le stream persistant. use axum::{ body::Body, extract::{Path, State}, http::{ - StatusCode, - header::{CACHE_CONTROL, CONNECTION, CONTENT_TYPE}, + HeaderMap, StatusCode, + header::{CACHE_CONTROL, CONNECTION, CONTENT_TYPE, CONTENT_RANGE, CONTENT_LENGTH, ACCEPT_RANGES}, }, response::{IntoResponse, Response}, }; @@ -20,21 +22,31 @@ use tracing::info; use crate::registry::RendererRegistry; /// GET /api/webrenderer/{id}/stream -/// -/// Crée un nouveau pipe FLAC à chaque connexion (chaque Play). -/// Le flux reste ouvert jusqu'à ce que le client se déconnecte (Stop). -/// Les morceaux s'enchaînent en gapless dans le même flux. -/// -/// Safari envoie un Range header (bytes=0-) et exige Accept-Ranges: bytes. -/// On répond 206 Partial Content si un Range header est présent, sinon 200. pub async fn stream_handler( State(registry): State>, Path(instance_id): Path, + headers: HeaderMap, ) -> impl IntoResponse { info!(instance_id = %instance_id, "FLAC stream client connecting"); - let handle = match registry.get_flac_handle(&instance_id) { - Some(h) => h, + // Détecter la sonde Range: bytes=0-1 de Safari + if let Some(range) = headers.get("range") { + if range.as_bytes() == b"bytes=0-1" { + info!(instance_id = %instance_id, "Safari range probe — responding 206"); + return Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(CONTENT_TYPE, "audio/ogg; codecs=flac") + .header(ACCEPT_RANGES, "bytes") + .header(CONTENT_RANGE, "bytes 0-1/*") + .header(CONTENT_LENGTH, "2") + .body(Body::from(vec![0u8, 0u8])) + .unwrap() + .into_response(); + } + } + + let stream = match registry.get_stream(&instance_id) { + Some(s) => s, None => { return ( StatusCode::NOT_FOUND, @@ -44,18 +56,14 @@ pub async fn stream_handler( } }; - let stream = handle.connect().await; - info!(instance_id = %instance_id, "FLAC stream started"); - // Toujours 200 OK pour un flux live de taille inconnue. - // Les navigateurs (Safari inclus) acceptent 200 pour l'audio streaming. - // Un Content-Range invalide (bytes 0-*/*) ferait rejeter le flux. Response::builder() .status(StatusCode::OK) .header(CONTENT_TYPE, "audio/ogg; codecs=flac") .header(CACHE_CONTROL, "no-store, no-transform") .header(CONNECTION, "keep-alive") + .header(ACCEPT_RANGES, "bytes") .header("X-Content-Type-Options", "nosniff") .body(Body::from_stream(ReaderStream::new(stream))) .unwrap()