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

Bump version to 0.3.12 and enhance Radio France source integration

Browse Source 
Bump version from 0.3.11 to 0.3.12

- Update Cargo.toml and version.txt to 0.3.12
- Refactor Radio France source registration to share Arc with API
- Enhance API routes to start/stop metadata refresh on stream access
- Add cleanup logic for stream disconnection
- Improve logging for metadata refresh operations
- Update Radio France source methods to be public for external access
This commit is contained in:
Eric Coissac
2026-01-23 22:41:54 +01:00
parent d3d2f24a6a
commit a0c1ab4228
8 changed files with 209 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -4,7 +4,7 @@ version = 4
[[package]]
name = "PMOMusic"
version = "0.3.11"
version = "0.3.12"
dependencies = [
"axum 0.8.7",
"console-subscriber",

2
PMOMusic/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "PMOMusic"
version = "0.3.11"
version = "0.3.12"
edition = "2024"
[dependencies]

15
pmomediaserver/src/sources.rs
View File

@@ -270,13 +270,16 @@ impl SourcesExt for Server {
});
let source = source.with_container_notifier(notifier);
// Enregistrer la source
self.register_music_source(Arc::new(source)).await;
// Enregistrer la source (Arc pour partage avec l'API)
let source_arc = Arc::new(source);
self.register_music_source(source_arc.clone()).await;
// Initialiser les routes API Radio France
self.init_radiofrance().await.map_err(|e| {
SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e))
})?;
// Initialiser les routes API Radio France avec la source
self.init_radiofrance_with_source(source_arc)
.await
.map_err(|e| {
SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e))
})?;
tracing::info!("✅ Radio France source registered successfully");

60
pmoradiofrance/src/api_rest.rs
View File

@@ -16,6 +16,7 @@ use axum::{
};
use futures::StreamExt;
use serde_json;
use std::sync::Arc;
// ============ Gestion des erreurs ============
@@ -95,6 +96,25 @@ async fn proxy_stream(
State(state): State<RadioFranceState>,
Path(slug): Path<String>,
) -> Result<Response, AppError> {
// Start metadata refresh when stream is accessed
#[cfg(feature = "logging")]
tracing::info!("Stream proxy accessed for station: {}", slug);
if let Some(ref source) = state.source {
// Spawn refresh task (non-blocking)
let source_clone = Arc::clone(source);
let slug_clone = slug.clone();
tokio::spawn(async move {
if let Err(e) = source_clone.start_metadata_refresh(&slug_clone).await {
#[cfg(feature = "logging")]
tracing::error!("Failed to start metadata refresh for {}: {}", slug_clone, e);
}
});
} else {
#[cfg(feature = "logging")]
tracing::warn!("No source available to start metadata refresh");
}
// Get the stream URL
let stream_url = state
.client
@@ -116,12 +136,48 @@ async fn proxy_stream(
headers.insert("content-type", "audio/aac".parse().unwrap());
headers.insert("cache-control", "no-cache".parse().unwrap());
// Create streaming body
// Create streaming body with cleanup on disconnect
let stream = response
.bytes_stream()
.map(|chunk| chunk.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
let body = Body::from_stream(stream);
// Wrap the stream to detect when client disconnects
let source_for_cleanup = state.source.clone();
let slug_for_cleanup = slug.clone();
let monitored_stream =
futures::stream::unfold((stream, false), move |(mut stream, mut done)| {
let source = source_for_cleanup.clone();
let slug = slug_for_cleanup.clone();
async move {
if done {
return None;
}
match stream.next().await {
Some(Ok(chunk)) => Some((Ok(chunk), (stream, false))),
Some(Err(e)) => {
// Error occurred, stop refresh
#[cfg(feature = "logging")]
tracing::info!("Stream error for {}, stopping refresh", slug);
if let Some(src) = source {
src.stop_metadata_refresh(&slug).await;
}
Some((Err(e), (stream, true)))
}
None => {
// Stream ended normally, stop refresh
#[cfg(feature = "logging")]
tracing::info!("Stream ended for {}, stopping refresh", slug);
if let Some(src) = source {
src.stop_metadata_refresh(&slug).await;
}
None
}
}
}
});
let body = Body::from_stream(monitored_stream);
Ok((headers, body).into_response())
}

17
pmoradiofrance/src/pmoserver_ext.rs
View File

@@ -12,14 +12,21 @@ use crate::stateful_client::RadioFranceStatefulClient;
#[derive(Clone)]
pub struct RadioFranceState {
pub client: Arc<RadioFranceStatefulClient>,
pub source: Option<Arc<crate::source::RadioFranceSource>>,
}
impl RadioFranceState {
pub fn new(client: RadioFranceStatefulClient) -> Self {
Self {
client: Arc::new(client),
source: None,
}
}
pub fn with_source(mut self, source: Arc<crate::source::RadioFranceSource>) -> Self {
self.source = Some(source);
self
}
}
/// Trait pour étendre pmoserver avec les fonctionnalités Radio France
@@ -80,6 +87,16 @@ pub trait RadioFranceExt {
/// server.init_radiofrance().await?;
/// ```
async fn init_radiofrance(&mut self) -> Result<Arc<RadioFranceState>>;
/// Initialise l'extension Radio France avec une source existante
///
/// Cette méthode est similaire à `init_radiofrance()` mais utilise une source
/// déjà créée et enregistrée, permettant de partager la même instance entre
/// le MediaServer UPnP et les routes API REST.
async fn init_radiofrance_with_source(
&mut self,
source: Arc<crate::source::RadioFranceSource>,
) -> Result<Arc<RadioFranceState>>;
}
// L'implémentation du trait sera dans un module séparé (pmoserver_impl.rs)

25
pmoradiofrance/src/pmoserver_impl.rs
View File

@@ -56,4 +56,29 @@ impl RadioFranceExt for Server {
Ok(Arc::new(state))
}
async fn init_radiofrance_with_source(
&mut self,
source: Arc<crate::source::RadioFranceSource>,
) -> Result<Arc<RadioFranceState>> {
info!("Initializing Radio France API with existing source...");
// Créer le client stateful
let config = pmoconfig::get_config();
let client = RadioFranceStatefulClient::new(config)
.await
.map_err(|e| anyhow::anyhow!("Failed to create Radio France client: {}", e))?;
// Créer l'état partagé avec la source
let state = RadioFranceState::new(client).with_source(source);
// Créer et enregistrer le router
let router = create_router(state.clone());
self.add_router("/api/radiofrance", router).await;
info!("Radio France API initialized with source");
info!("API endpoints available at /api/radiofrance/*");
Ok(Arc::new(state))
}
}

99
pmoradiofrance/src/source.rs
View File

@@ -151,7 +151,7 @@ impl RadioFranceSource {
}
/// Start metadata refresh task for a station
async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> {
pub async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> {
let mut handles = self.refresh_handles.write().await;
// If already running, do nothing
@@ -177,11 +177,51 @@ impl RadioFranceSource {
Ok(metadata) => {
let delay = std::time::Duration::from_millis(metadata.delay_to_refresh);
#[cfg(feature = "logging")]
{
let artist = metadata
.now
.song
.as_ref()
.and_then(|s| {
if s.interpreters.is_empty() {
None
} else {
Some(s.artists_display())
}
})
.unwrap_or_else(|| "".to_string());
tracing::debug!(
"Refreshed metadata for {}: title='{}' artist='{}' delay={}ms",
slug,
metadata.now.first_line.title.as_deref().unwrap_or(""),
artist,
metadata.delay_to_refresh
);
}
// Update the playlist metadata
#[cfg(feature = "cache")]
{
let mut pls = playlists.write().await;
#[cfg(feature = "logging")]
tracing::debug!(
"Looking for playlist '{}' in cache, found: {}",
slug,
pls.contains_key(&slug)
);
if let Some(playlist) = pls.get_mut(&slug) {
let old_title = playlist.stream_item.title.clone();
#[cfg(feature = "logging")]
tracing::debug!(
"Updating playlist for {}: current title = '{}'",
slug,
old_title
);
let _: Result<()> = playlist
.update_metadata(
&metadata,
@@ -190,6 +230,18 @@ impl RadioFranceSource {
)
.await;
let new_title = playlist.stream_item.title.clone();
#[cfg(feature = "logging")]
if old_title != new_title {
tracing::info!(
"Metadata updated for {}: {} -> {}",
slug,
old_title,
new_title
);
}
// Update change tracking
*update_id.write().await = update_id.read().await.wrapping_add(1);
*last_change.write().await = Some(SystemTime::now());
@@ -198,6 +250,13 @@ impl RadioFranceSource {
if let Some(ref notifier) = container_notifier {
// Notify the station's stream item container
let container_id = format!("radiofrance:{}", slug);
#[cfg(feature = "logging")]
tracing::info!(
"Notifying UPnP container update: {}",
container_id
);
notifier(&[container_id]);
}
}
@@ -206,12 +265,41 @@ impl RadioFranceSource {
#[cfg(not(feature = "cache"))]
{
let mut pls = playlists.write().await;
#[cfg(feature = "logging")]
tracing::debug!(
"Looking for playlist '{}' in cache, found: {}",
slug,
pls.contains_key(&slug)
);
if let Some(playlist) = pls.get_mut(&slug) {
let old_title = playlist.stream_item.title.clone();
#[cfg(feature = "logging")]
tracing::debug!(
"Updating playlist for {}: current title = '{}'",
slug,
old_title
);
let _: Result<()> = playlist.update_metadata_no_cache(
&metadata,
server_base_url.as_deref(),
);
let new_title = playlist.stream_item.title.clone();
#[cfg(feature = "logging")]
if old_title != new_title {
tracing::info!(
"Metadata updated for {}: {} -> {}",
slug,
old_title,
new_title
);
}
// Update change tracking
*update_id.write().await = update_id.read().await.wrapping_add(1);
*last_change.write().await = Some(SystemTime::now());
@@ -220,6 +308,13 @@ impl RadioFranceSource {
if let Some(ref notifier) = container_notifier {
// Notify the station's stream item container
let container_id = format!("radiofrance:{}", slug);
#[cfg(feature = "logging")]
tracing::info!(
"Notifying UPnP container update: {}",
container_id
);
notifier(&[container_id]);
}
}
@@ -246,7 +341,7 @@ impl RadioFranceSource {
}
/// Stop metadata refresh task for a station
async fn stop_metadata_refresh(&self, station_slug: &str) {
pub async fn stop_metadata_refresh(&self, station_slug: &str) {
let mut handles = self.refresh_handles.write().await;
if let Some(handle) = handles.remove(station_slug) {
handle.abort();

2
version.txt
View File

@@ -1 +1 @@
0.3.11
0.3.12