eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Migrate instance ID storage from localStorage to sessionStorage

Browse Source 
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.
This commit is contained in:
Eric Coissac
2026-02-28 14:10:38 +01:00
parent ca8702fd7f
commit de74c7b431
5 changed files with 187 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored
View File

Binary file not shown.

4
pmoapp/webapp/src/composables/useWebRenderer.ts
View File

@@ -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 {

401
pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs
View File

@@ -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<Bytes>) ← 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<Bytes> (cap=8, backpressure naturelle)
//! Arc<Mutex<Receiver<Bytes>>> → 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<Mutex<Option<mpsc::Sender<PcmChunk>>>>;
/// 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<Mutex<Option<mpsc::Sender<Bytes>>>>;
/// Handle de la task encodeur courante.
type SharedEncoderTask = Arc<Mutex<Option<JoinHandle<()>>>>;
type SharedOggRx = Arc<Mutex<mpsc::Receiver<Bytes>>>;
// ─── Handle public ────────────────────────────────────────────────────────────
/// Handle vers le sink, cloneable, reconnectable à chaque Play.
#[derive(Clone)]
pub struct DirectOggFlacHandle {
pcm_tx: SharedPcmTx,
client_connect_tx: Arc<watch::Sender<u64>>,
client_notify_internal: Arc<tokio::sync::Notify>,
first_byte_tx: Arc<watch::Sender<bool>>,
encoder_task: SharedEncoderTask,
encoder_options: EncoderOptions,
current_timestamp: Arc<tokio::sync::RwLock<f64>>,
/// 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<Bytes>,
ogg_rx: SharedOggRx,
/// Notifié quand get_stream() est appelé (premier client).
client_notify: Arc<tokio::sync::Notify>,
}
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>(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::<PcmChunk>(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<bool> {
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<Arc<watch::Sender<bool>>>,
ogg_rx: SharedOggRx,
buffer: VecDeque<u8>,
}
impl AsyncRead for DirectOggFlacStream {
@@ -188,18 +106,36 @@ impl AsyncRead for DirectOggFlacStream {
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
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<u8> = 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<tokio::sync::Notify>,
encoder_options: EncoderOptions,
encoder_task: SharedEncoderTask,
bytes_tx: SharedBytesTx,
encoder_options: EncoderOptions,
current_timestamp: Arc<tokio::sync::RwLock<f64>>,
/// 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<Bytes>,
client_notify: Arc<tokio::sync::Notify>,
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::<PcmChunk>(8);
async fn start_encoder(&mut self) {
let (pcm_tx, pcm_rx) = mpsc::channel::<PcmChunk>(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<Bytes>,
shared_bytes_tx: SharedBytesTx,
ogg_tx: mpsc::Sender<Bytes>,
options: EncoderOptions,
_current_timestamp: Arc<tokio::sync::RwLock<f64>>,
) -> 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<u8> = 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::<Bytes>(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)

9
pmowebrenderer/src/registry.rs
View File

@@ -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<pmoaudio_ext::sinks::DirectOggFlacHandle> {
) -> Option<pmoaudio_ext::sinks::DirectOggFlacStream> {
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)

44
pmowebrenderer/src/stream.rs
View File

@@ -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<Arc<RendererRegistry>>,
Path(instance_id): Path<String>,
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()