diff --git a/Cargo.lock b/Cargo.lock index 14d3f466..4c19cb72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3897,6 +3897,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "ureq 2.12.1", ] [[package]] @@ -4016,7 +4017,7 @@ dependencies = [ "tracing", "tracing-log 0.1.4", "tracing-subscriber", - "ureq", + "ureq 3.1.4", "url", "urlencoding", "utoipa", @@ -6513,6 +6514,22 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "ureq" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d1a66277ed75f640d608235660df48c8e3c19f3b4edb6a263315626cc3c01d" +dependencies = [ + "base64 0.22.1", + "flate2", + "log", + "once_cell", + "rustls", + "rustls-pki-types", + "url", + "webpki-roots 0.26.11", +] + [[package]] name = "ureq" version = "3.1.4" @@ -6527,7 +6544,7 @@ dependencies = [ "rustls-pki-types", "ureq-proto", "utf-8", - "webpki-roots", + "webpki-roots 1.0.4", ] [[package]] @@ -6820,6 +6837,15 @@ dependencies = [ "libwebp-sys", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.4", +] + [[package]] name = "webpki-roots" version = "1.0.4" diff --git a/pmoaudio-ext/Cargo.toml b/pmoaudio-ext/Cargo.toml index f36fdc6b..debf6720 100644 --- a/pmoaudio-ext/Cargo.toml +++ b/pmoaudio-ext/Cargo.toml @@ -32,10 +32,11 @@ serde = { workspace = true, optional = true } serde_json = { workspace = true, optional = true } reqwest = { workspace = true, features = ["stream"], optional = true } futures = { version = "0.3", optional = true } +ureq = { version = "2", optional = true } [features] default = [] cache-sink = ["dep:pmoaudiocache", "dep:pmoflac", "dep:pmometadata", "dep:serde_json"] playlist = ["cache-sink", "dep:pmoplaylist", "dep:pmocache"] -http-stream = ["dep:pmoflac", "dep:pmometadata", "dep:bytes", "dep:serde", "dep:reqwest", "dep:futures"] +http-stream = ["dep:pmoflac", "dep:pmometadata", "dep:bytes", "dep:serde", "dep:reqwest", "dep:futures", "dep:ureq"] all = ["cache-sink", "playlist", "http-stream"] diff --git a/pmoaudio-ext/src/sources/player_source.rs b/pmoaudio-ext/src/sources/player_source.rs index 051421ed..acdd56c7 100644 --- a/pmoaudio-ext/src/sources/player_source.rs +++ b/pmoaudio-ext/src/sources/player_source.rs @@ -165,6 +165,7 @@ impl NodeLogic for PlayerSourceLogic { let mut current_uri: Option = None; let mut next_uri: Option = None; let mut paused_at_sec: f64 = 0.0; + let mut is_continuous: bool = false; info!("PlayerSource: started"); @@ -182,7 +183,7 @@ impl NodeLogic for PlayerSourceLogic { None => break, Some(cmd) => { self.handle_command( - cmd, &mut state, &mut current_uri, + cmd, is_continuous, &mut state, &mut current_uri, &mut next_uri, &mut paused_at_sec, &output, &stop_token, ).await?; @@ -213,15 +214,17 @@ impl NodeLogic for PlayerSourceLogic { }; let duration_sec = source.duration_sec(); + let is_continuous = source.is_continuous(); let _ = self.event_tx.send(PlayerEvent::Playing { uri: uri.clone(), duration_sec, }); - info!("PlayerSource: playing {:?} from {:.1}s", uri, paused_at_sec); + info!("PlayerSource: playing {:?} from {:.1}s continuous={}", uri, paused_at_sec, is_continuous); // Pompe audio — s'arrête sur EOF, Pause, Stop, ou commande let result = self.pump( source, + is_continuous, &mut state, &mut current_uri, &mut next_uri, @@ -256,6 +259,7 @@ impl PlayerSourceLogic { async fn handle_command( &mut self, cmd: PlayerCommand, + is_continuous: bool, state: &mut TransportState, current_uri: &mut Option, next_uri: &mut Option, @@ -282,14 +286,16 @@ impl PlayerSourceLogic { 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?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, *paused_at_sec, stream_type, 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?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await?; *state = TransportState::Playing; } TransportState::Idle => { @@ -321,7 +327,8 @@ impl PlayerSourceLogic { 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?; + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + send_track_boundary(current_uri.as_deref(), output, pos, stream_type, stop_token).await?; *state = TransportState::Playing; } } @@ -336,6 +343,7 @@ impl PlayerSourceLogic { async fn pump( &mut self, source: UriSource, + is_continuous: bool, state: &mut TransportState, current_uri: &mut Option, next_uri: &mut Option, @@ -395,7 +403,8 @@ impl PlayerSourceLogic { *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 { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await { result = Err(e); } *state = TransportState::Playing; @@ -409,7 +418,8 @@ impl PlayerSourceLogic { 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 { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, pos, stream_type, stop_token).await { result = Err(e); break; } @@ -435,7 +445,8 @@ impl PlayerSourceLogic { 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 { + let stream_type = if is_continuous { StreamType::Continuous } else { StreamType::Finite }; + if let Err(e) = send_track_boundary(current_uri.as_deref(), output, 0.0, stream_type, stop_token).await { result = Err(e); } *state = TransportState::Playing; @@ -495,6 +506,7 @@ async fn send_track_boundary( uri: Option<&str>, output: &[mpsc::Sender>], timestamp_sec: f64, + stream_type: StreamType, stop_token: &CancellationToken, ) -> Result<(), AudioError> { if output.is_empty() || stop_token.is_cancelled() { @@ -506,7 +518,7 @@ async fn send_track_boundary( 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, StreamType::Finite); + let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc, stream_type); send_to_children("PlayerSource", output, boundary).await } diff --git a/pmoaudio-ext/src/sources/uri_source.rs b/pmoaudio-ext/src/sources/uri_source.rs index ecf7c196..8f1e91b7 100644 --- a/pmoaudio-ext/src/sources/uri_source.rs +++ b/pmoaudio-ext/src/sources/uri_source.rs @@ -49,6 +49,8 @@ pub struct UriSource { reader: Box, stream_info: StreamInfo, frames_to_skip: u64, + /// true si c'est un flux continu (radio, stream) sans durée définie + pub is_continuous: bool, } impl UriSource { @@ -88,6 +90,11 @@ impl UriSource { }) } + /// Retourne true si c'est un flux continu (radio, stream) sans durée définie. + pub fn is_continuous(&self) -> bool { + self.is_continuous + } + /// Émet les chunks audio vers `tx`. /// /// Retourne `Ok(true)` si EOF naturel, `Ok(false)` si annulé ou receiver fermé. @@ -201,7 +208,7 @@ impl UriSource { ); let (_, reader) = stream.into_reader(); - Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip }) + Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip, is_continuous: false }) } async fn open_http( @@ -209,6 +216,9 @@ impl UriSource { seek_sec: f64, stop_token: &CancellationToken, ) -> Result { + // Détecter si c'est un flux continu (radio, stream) basé sur l'URL + let is_continuous = detect_continuous_stream(url); + let response = tokio::select! { _ = stop_token.cancelled() => { return Err(AudioError::IoError("Cancelled before HTTP connect".into())); @@ -244,11 +254,89 @@ impl UriSource { let frames_to_skip = (seek_sec * stream_info.sample_rate as f64) as u64; info!( - "UriSource: opened HTTP {} Hz {} ch {} bps", - stream_info.sample_rate, stream_info.channels, stream_info.bits_per_sample, + "UriSource: opened HTTP {} Hz {} ch {} bps continuous={}", + stream_info.sample_rate, stream_info.channels, stream_info.bits_per_sample, is_continuous, ); let (_, reader) = stream.into_reader(); - Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip }) + Ok(Self { + reader: Box::new(reader), + stream_info, + frames_to_skip, + is_continuous, + }) } } + +/// Détecte si une URL correspond à un flux continu (radio, stream) sans durée définie. +/// +/// Cette fonction: +/// 1. Vérifie les patterns d'URL connus (stream, live, radio, etc.) +/// 2. Fait une requête HTTP HEAD pour vérifier les headers (Content-Length, ICY, etc.) +fn detect_continuous_stream(url: &str) -> bool { + let url_lower = url.to_lowercase(); + + // 1. Quick check sur les patterns d'URL très explicites + // Ces patterns indiquent clairement un stream live + if url_lower.contains("/live") + || url_lower.contains("/radiolar") + || url_lower.contains(".pls") + || url_lower.contains(".m3u") + || url_lower.contains("icy") + { + return true; + } + + // 2. Vérification HTTP headers (le plus fiable) + if url.starts_with("http://") || url.starts_with("https://") { + if let Ok(is_stream) = check_http_stream_headers(url) { + if is_stream { + return true; + } + } + } + + false +} + +/// Vérifie les headers HTTP pour déterminer si c'est un stream +fn check_http_stream_headers(url: &str) -> Result { + use std::time::Duration; + + let agent = ureq::AgentBuilder::new() + .timeout(Duration::from_secs(3)) + .build(); + + let response = agent + .head(url) + .call() + .map_err(|e| format!("HTTP HEAD failed: {}", e))?; + + // Headers ICY (Icecast/Shoutcast) = toujours un stream + if response.header("icy-name").is_some() + || response.header("icy-metaint").is_some() + { + return Ok(true); + } + + // Pas de Content-Length = stream potentiel + let has_content_length = response.header("content-length").is_some(); + + // Transfer-Encoding: chunked = stream potentiel + let is_chunked = response + .header("transfer-encoding") + .map(|v| v.to_lowercase().contains("chunked")) + .unwrap_or(false); + + // Decision: pas de Content-Length + (chunked ou content-type streaming) + let content_type = response + .header("content-type") + .unwrap_or("") + .to_lowercase(); + + let is_streaming_mime = content_type.contains("audio/mpeg") + || content_type.contains("audio/aac") + || content_type.contains("application/ogg"); + + Ok(!has_content_length && (is_streaming_mime || is_chunked)) +} diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index ee7b548a..6eaf720a 100755 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -178,7 +178,7 @@ impl NodeLogic for HttpSourceLogic { // Émettre TrackBoundary avec les métadonnées HTTP let track_boundary = - AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), StreamType::Continuous); + AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), StreamType::Finite); send_to_children(std::any::type_name::(), &output, track_boundary).await?; // Préparer la lecture des chunks audio