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

Ajout la playlist live

Browse Source
This commit is contained in:
Eric Coissac
2025-11-29 01:08:07 +01:00
parent 503b8bc4ff
commit a559375176
6 changed files with 350 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
pmomediaserver/src/contentdirectory/variables/containerupdateids.rs Normal file
View File

@@ -0,0 +1,9 @@
use pmoupnp::define_variable;
// Liste des conteneurs modifiés (format "id,updateId,id,updateId,...")
define_variable! {
pub static CONTAINERUPDATEIDS: String = "ContainerUpdateIDs" {
evented: true,
// valeur initiale vide
}
}

2
pmomediaserver/src/contentdirectory/variables/mod.rs
View File

@@ -7,6 +7,7 @@ mod a_arg_type_result;
mod a_arg_type_searchcriteria;
mod a_arg_type_sortcriteria;
mod a_arg_type_updateid;
mod containerupdateids;
mod searchcapabilities;
mod sortcapabilities;
mod systemupdateid;
@@ -20,6 +21,7 @@ pub use a_arg_type_result::A_ARG_TYPE_RESULT;
pub use a_arg_type_searchcriteria::A_ARG_TYPE_SEARCHCRITERIA;
pub use a_arg_type_sortcriteria::A_ARG_TYPE_SORTCRITERIA;
pub use a_arg_type_updateid::A_ARG_TYPE_UPDATEID;
pub use containerupdateids::CONTAINERUPDATEIDS;
pub use searchcapabilities::SEARCHCAPABILITIES;
pub use sortcapabilities::SORTCAPABILITIES;
pub use systemupdateid::SYSTEMUPDATEID;

7
pmomediaserver/src/sources.rs
View File

@@ -181,10 +181,13 @@ impl SourcesExt for Server {
let base_url = self.base_url();
// Créer la source Radio Paradise (utilise le singleton PlaylistManager)
let source = RadioParadiseSource::new(base_url.to_string());
let source = Arc::new(RadioParadiseSource::new(base_url.to_string()));
// Brancher les callbacks de playlists (live/history) pour signaler les updates
source.attach_playlist_callbacks();
// Enregistrer la source
self.register_music_source(Arc::new(source)).await;
self.register_music_source(source.clone()).await;
tracing::info!("✅ Radio Paradise source registered successfully");

284
pmoparadise/src/source.rs
View File

@@ -28,6 +28,8 @@ const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
/// - Root: `radio-paradise`
/// - Channel container: `radio-paradise:channel:{slug}`
/// - Live stream item: `radio-paradise:channel:{slug}:live`
/// - Live playlist container: `radio-paradise:channel:{slug}:liveplaylist`
/// - Live playlist track: `radio-paradise:channel:{slug}:liveplaylist:track:{pk}`
/// - History container: `radio-paradise:channel:{slug}:history`
/// - History track: `radio-paradise:channel:{slug}:history:track:{pk}`
#[derive(Debug, Clone)]
@@ -38,6 +40,8 @@ pub struct RadioParadiseSource {
update_counter: Arc<RwLock<u32>>,
/// Last change timestamp
last_change: Arc<RwLock<SystemTime>>,
/// Tokens des callbacks enregistrés auprès du PlaylistManager
callback_tokens: Arc<std::sync::Mutex<Vec<u64>>>,
}
impl RadioParadiseSource {
@@ -56,6 +60,7 @@ impl RadioParadiseSource {
base_url: base_url.into(),
update_counter: Arc::new(RwLock::new(0)),
last_change: Arc::new(RwLock::new(SystemTime::now())),
callback_tokens: Arc::new(std::sync::Mutex::new(Vec::new())),
}
}
@@ -69,6 +74,49 @@ impl RadioParadiseSource {
format!("{}/radioparadise/stream/{}/ogg", self.base_url, slug)
}
/// Incrémente l'update_counter et met à jour last_change
async fn bump_update_counter(&self) {
{
let mut c = self.update_counter.write().await;
*c = c.wrapping_add(1).max(1);
}
let mut lc = self.last_change.write().await;
*lc = SystemTime::now();
}
/// Enregistre des callbacks sur les playlists live/historique pour notifier les changements
pub fn attach_playlist_callbacks(self: &Arc<Self>) {
use pmoplaylist::PlaylistManager;
// Préparer les IDs de playlists à surveiller (live + history pour chaque canal)
let ids: Vec<String> = ALL_CHANNELS
.iter()
.flat_map(|ch| {
vec![
Self::live_playlist_id(ch.slug),
Self::history_playlist_id(ch.slug),
]
})
.collect();
let mgr = PlaylistManager();
let mut tokens = self.callback_tokens.lock().unwrap();
for pid in ids {
let weak = Arc::downgrade(self);
let token = mgr.register_callback(move |changed_id| {
if changed_id == pid {
if let Some(strong) = weak.upgrade() {
tokio::spawn(async move {
strong.bump_update_counter().await;
});
}
}
});
tokens.push(token);
}
}
/// URL de fallback pour l'image par défaut de la source
fn default_cover_url(&self) -> String {
format!("{}/api/sources/{}/image", self.base_url, self.id())
@@ -152,6 +200,11 @@ impl RadioParadiseSource {
format!("radio-paradise-history-{}", slug)
}
/// Live playlist id for a channel
fn live_playlist_id(slug: &str) -> String {
format!("radio-paradise-live-{}", slug)
}
/// Get channel descriptor by slug
fn get_channel_by_slug(slug: &str) -> Option<&'static ChannelDescriptor> {
ALL_CHANNELS.iter().find(|ch| ch.slug == slug)
@@ -168,6 +221,15 @@ impl RadioParadiseSource {
["radio-paradise", "channel", slug, "live"] => ObjectIdType::LiveStream {
slug: (*slug).to_string(),
},
["radio-paradise", "channel", slug, "liveplaylist"] => ObjectIdType::LivePlaylist {
slug: (*slug).to_string(),
},
["radio-paradise", "channel", slug, "liveplaylist", "track", pk] => {
ObjectIdType::LivePlaylistTrack {
slug: (*slug).to_string(),
pk: (*pk).to_string(),
}
}
["radio-paradise", "channel", slug, "history"] => ObjectIdType::History {
slug: (*slug).to_string(),
},
@@ -196,6 +258,21 @@ impl RadioParadiseSource {
}
}
/// Build the live playlist container for a channel
fn build_live_playlist_container(&self, descriptor: &ChannelDescriptor) -> Container {
Container {
id: format!("radio-paradise:channel:{}:liveplaylist", descriptor.slug),
parent_id: format!("radio-paradise:channel:{}", descriptor.slug),
restricted: Some("1".to_string()),
child_count: None,
searchable: Some("0".to_string()),
title: format!("{} - Live Playlist", descriptor.display_name),
class: "object.container.playlistContainer".to_string(),
containers: vec![],
items: vec![],
}
}
/// Build a live stream item for a channel
fn build_live_stream_item(&self, descriptor: &ChannelDescriptor) -> Item {
let stream_url = self.build_live_url(descriptor.slug);
@@ -332,6 +409,82 @@ impl RadioParadiseSource {
Ok(items)
}
/// Get items from live playlist (current stream queue)
#[cfg(feature = "playlist")]
async fn get_live_playlist_items(
&self,
slug: &str,
_offset: usize,
count: usize,
) -> Result<Vec<Item>> {
let playlist_id = Self::live_playlist_id(slug);
let manager = pmoplaylist::PlaylistManager();
let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| {
MusicSourceError::BrowseError(format!("Failed to get live playlist {}: {}", playlist_id, e))
})?;
let mut items = reader.to_items(count).await.map_err(|e| {
MusicSourceError::BrowseError(format!(
"Failed to read live playlist entries: {}",
e
))
})?;
for item in items.iter_mut() {
// Ajuster id/parent/url pour coller au schéma Radio Paradise
if let Some(resource) = item.resources.first_mut() {
if let Some(pk) = resource.url.split('/').last() {
item.id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
item.parent_id = format!(
"radio-paradise:channel:{}:liveplaylist",
slug
);
if resource.url.starts_with('/') {
resource.url = format!("{}{}", self.base_url, resource.url);
}
}
}
if item.genre.is_none() {
item.genre = Some("Radio Paradise".to_string());
}
if let Some(art) = item.album_art.as_mut() {
if art.starts_with('/') {
*art = format!("{}{}", self.base_url, art);
}
} else {
item.album_art = Some(self.default_cover_url());
}
}
Ok(items)
}
/// Get a single item from the live playlist by pk
#[cfg(feature = "playlist")]
async fn get_live_playlist_item(&self, slug: &str, pk: &str) -> Result<Item> {
let items = self.get_live_playlist_items(slug, 0, 1000).await?;
let expected_id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
for item in items {
if item.id == expected_id {
return Ok(item);
}
}
Err(MusicSourceError::ObjectNotFound(format!(
"Track with pk {} not found in live playlist",
pk
)))
}
}
/// Types of object IDs in the Radio Paradise source
@@ -340,6 +493,8 @@ enum ObjectIdType {
Root,
Channel { slug: String },
LiveStream { slug: String },
LivePlaylist { slug: String },
LivePlaylistTrack { slug: String, pk: String },
History { slug: String },
HistoryTrack { slug: String, pk: String },
Unknown,
@@ -393,6 +548,7 @@ impl MusicSource for RadioParadiseSource {
})?;
let live_item = self.build_live_stream_item(descriptor);
let live_playlist_container = self.build_live_playlist_container(descriptor);
#[cfg(feature = "playlist")]
let history_container = self.build_history_container_with_count(descriptor).await;
@@ -400,7 +556,7 @@ impl MusicSource for RadioParadiseSource {
let history_container = self.build_history_container(descriptor);
Ok(BrowseResult::Mixed {
containers: vec![history_container],
containers: vec![live_playlist_container, history_container],
items: vec![live_item],
})
}
@@ -439,12 +595,52 @@ impl MusicSource for RadioParadiseSource {
Ok(BrowseResult::Items(vec![item]))
}
ObjectIdType::LivePlaylist { slug } => {
// Playlist du live : container + items
let descriptor = Self::get_channel_by_slug(&slug).ok_or_else(|| {
MusicSourceError::ObjectNotFound(format!("Unknown channel: {}", slug))
})?;
#[cfg(feature = "playlist")]
{
let container = self.build_live_playlist_container(descriptor);
let items = self.get_live_playlist_items(&slug, 0, 100).await?;
Ok(BrowseResult::Mixed {
containers: vec![container],
items,
})
}
#[cfg(not(feature = "playlist"))]
{
let container = self.build_live_playlist_container(descriptor);
Ok(BrowseResult::Containers(vec![container]))
}
}
ObjectIdType::HistoryTrack { slug, pk } => {
// Return metadata for the history track item
let item = self.get_item(object_id).await?;
Ok(BrowseResult::Items(vec![item]))
}
ObjectIdType::LivePlaylistTrack { slug, pk } => {
// Détails d'un titre du live (playlist live)
#[cfg(feature = "playlist")]
{
let item = self.get_live_playlist_item(&slug, &pk).await?;
Ok(BrowseResult::Items(vec![item]))
}
#[cfg(not(feature = "playlist"))]
{
let _ = (slug, pk);
Err(MusicSourceError::NotSupported(
"Playlist feature not enabled".to_string(),
))
}
}
ObjectIdType::Unknown => Err(MusicSourceError::ObjectNotFound(format!(
"Unknown object ID: {}",
object_id
@@ -464,6 +660,11 @@ impl MusicSource for RadioParadiseSource {
Ok(format!("{}/cache/audio/{}", self.base_url, pk))
}
ObjectIdType::LivePlaylistTrack { pk, .. } => {
// Return cached audio URL
Ok(format!("{}/cache/audio/{}", self.base_url, pk))
}
_ => Err(MusicSourceError::ObjectNotFound(format!(
"Cannot resolve URI for object: {}",
object_id
@@ -514,6 +715,14 @@ impl MusicSource for RadioParadiseSource {
bitrate: None,
channels: Some(2),
}]),
ObjectIdType::LivePlaylistTrack { .. } => Ok(vec![AudioFormat {
format_id: "flac".to_string(),
mime_type: "audio/flac".to_string(),
sample_rate: Some(44100),
bit_depth: Some(16),
bitrate: None,
channels: Some(2),
}]),
_ => Err(MusicSourceError::ObjectNotFound(format!(
"Cannot list formats for object: {}",
object_id
@@ -603,6 +812,79 @@ impl MusicSource for RadioParadiseSource {
}
}
ObjectIdType::LivePlaylistTrack { slug, pk } => {
#[cfg(feature = "playlist")]
{
let playlist_id = Self::live_playlist_id(&slug);
let manager = pmoplaylist::PlaylistManager();
let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| {
MusicSourceError::BrowseError(format!(
"Failed to get live playlist {}: {}",
playlist_id, e
))
})?;
let items = reader.to_items(1000).await.map_err(|e| {
MusicSourceError::BrowseError(format!(
"Failed to read live playlist entries: {}",
e
))
})?;
for mut item in items {
if let Some(resource) = item.resources.first_mut() {
if let Some(pk2) = resource.url.split('/').last() {
item.id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk2
);
item.parent_id = format!(
"radio-paradise:channel:{}:liveplaylist",
slug
);
if resource.url.starts_with('/') {
resource.url = format!("{}{}", self.base_url, resource.url);
}
}
}
if item.genre.is_none() {
item.genre = Some("Radio Paradise".to_string());
}
if let Some(art) = item.album_art.as_mut() {
if art.starts_with('/') {
*art = format!("{}{}", self.base_url, art);
}
} else {
item.album_art = Some(self.default_cover_url());
}
let expected_id = format!(
"radio-paradise:channel:{}:liveplaylist:track:{}",
slug, pk
);
if item.id == expected_id {
return Ok(item);
}
}
Err(MusicSourceError::ObjectNotFound(format!(
"Track with pk {} not found in live playlist",
pk
)))
}
#[cfg(not(feature = "playlist"))]
{
let _ = (slug, pk);
Err(MusicSourceError::NotSupported(
"Playlist feature not enabled".to_string(),
))
}
}
_ => Err(MusicSourceError::ObjectNotFound(format!(
"Cannot get item for object: {}",
object_id

18
pmoplaylist/src/handle/write.rs
View File

@@ -47,6 +47,9 @@ impl WriteHandle {
self.save_to_db().await?;
}
// Notifier le manager
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -75,6 +78,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -107,6 +112,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -126,6 +133,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -141,6 +150,9 @@ impl WriteHandle {
// Supprimer du manager
crate::manager::delete_playlist_internal(&self.playlist.id).await?;
// Notifier la suppression
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -156,6 +168,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -175,6 +189,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}
@@ -194,6 +210,8 @@ impl WriteHandle {
self.save_to_db().await?;
}
crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id);
Ok(())
}

34
pmoplaylist/src/manager.rs
View File

@@ -8,9 +8,10 @@ use crate::Result;
use once_cell::sync::OnceCell;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::{Arc, atomic::{AtomicU64, Ordering}};
use std::time::Duration;
use tokio::sync::RwLock;
use std::sync::RwLock as StdRwLock;
/// Singleton PlaylistManager
static PLAYLIST_MANAGER: OnceCell<PlaylistManager> = OnceCell::new();
@@ -22,6 +23,8 @@ static AUDIO_CACHE: OnceCell<Arc<pmoaudiocache::Cache>> = OnceCell::new();
struct ManagerInner {
playlists: RwLock<HashMap<String, Arc<Playlist>>>,
persistence: Option<Arc<PersistenceManager>>,
callbacks: StdRwLock<HashMap<u64, Arc<dyn Fn(&str) + Send + Sync>>>,
cb_counter: AtomicU64,
}
/// Gestionnaire central de playlists
@@ -47,6 +50,8 @@ impl PlaylistManager {
inner: Arc::new(ManagerInner {
playlists: RwLock::new(HashMap::new()),
persistence: Some(persistence.clone()),
callbacks: StdRwLock::new(HashMap::new()),
cb_counter: AtomicU64::new(1),
}),
};
@@ -123,6 +128,33 @@ impl PlaylistManager {
Ok(WriteHandle::new(playlist, write_token))
}
/// Enregistre un callback de modification de playlist.
///
/// Retourne un jeton (u64) pour désenregistrer plus tard.
pub fn register_callback<F>(&self, cb: F) -> u64
where
F: Fn(&str) + Send + Sync + 'static,
{
let token = self.inner.cb_counter.fetch_add(1, Ordering::Relaxed);
let mut guard = self.inner.callbacks.write().unwrap();
guard.insert(token, Arc::new(cb));
token
}
/// Désenregistre un callback via son jeton.
pub fn unregister_callback(&self, token: u64) {
let mut guard = self.inner.callbacks.write().unwrap();
guard.remove(&token);
}
/// Notifie tous les callbacks qu'une playlist a changé.
pub(crate) fn notify_playlist_changed(&self, id: &str) {
let guard = self.inner.callbacks.read().unwrap();
for cb in guard.values() {
cb(id);
}
}
/// R<>cup<75>re un write handle (cr<63>e <20>ph<70>m<EFBFBD>re si n'existe pas)
pub async fn get_write_handle(&self, id: String) -> Result<WriteHandle> {
let mut playlists = self.inner.playlists.write().await;