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

Gestion des notifications

Browse Source
This commit is contained in:
Eric Coissac
2025-11-29 01:56:40 +01:00
parent a559375176
commit 6ae06b38e1
7 changed files with 147 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
PMOMusic/src/main.rs
View File

@@ -81,6 +81,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.await
.expect("Failed to register MediaServer");
// Enregistrer l'instance ContentDirectory pour les notifications GENA
if let Some(cd_service) = server_instance.get_service("ContentDirectory") {
pmomediaserver::contentdirectory::state::register_instance(&cd_service);
}
// Initialiser les ProtocolInfo du MediaServer
server_instance.init_protocol_info();

4
pmomediaserver/src/contentdirectory/mod.rs
View File

@@ -77,12 +77,13 @@ use pmoupnp::define_service;
pub mod actions;
pub mod handlers;
pub mod variables;
pub mod state;
use actions::{BROWSE, GETSEARCHCAPABILITIES, GETSORTCAPABILITIES, GETSYSTEMUPDATEID, SEARCH};
use variables::{
A_ARG_TYPE_BROWSEFLAG, A_ARG_TYPE_COUNT, A_ARG_TYPE_FILTER, A_ARG_TYPE_INDEX,
A_ARG_TYPE_OBJECTID, A_ARG_TYPE_RESULT, A_ARG_TYPE_SEARCHCRITERIA, A_ARG_TYPE_SORTCRITERIA,
A_ARG_TYPE_UPDATEID, SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID,
A_ARG_TYPE_UPDATEID, SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID, CONTAINERUPDATEIDS,
};
// Service ContentDirectory:1 conforme à la spécification UPnP AV pour MediaServer
@@ -102,6 +103,7 @@ define_service! {
SEARCHCAPABILITIES,
SORTCAPABILITIES,
SYSTEMUPDATEID,
CONTAINERUPDATEIDS,
],
actions: [
BROWSE,

58
pmomediaserver/src/contentdirectory/state.rs Normal file
View File

@@ -0,0 +1,58 @@
use once_cell::sync::OnceCell;
use pmoupnp::{services::ServiceInstance, variable_types::StateValue};
use std::sync::{atomic::{AtomicU32, Ordering}, Arc, Weak, Mutex};
static CONTENTDIR_INSTANCE: OnceCell<Weak<ServiceInstance>> = OnceCell::new();
static SYSTEM_UPDATE_ID: AtomicU32 = AtomicU32::new(1);
static CONTAINER_UPDATE_IDS: Mutex<String> = Mutex::new(String::new());
/// Enregistre l'instance ContentDirectory pour pouvoir pousser des notifications GENA.
pub fn register_instance(instance: &Arc<ServiceInstance>) {
let _ = CONTENTDIR_INSTANCE.set(Arc::downgrade(instance));
// Initialiser les valeurs
set_system_update_id(1);
set_container_update_ids("");
}
/// Notifie une mise à jour en incrémentant SystemUpdateID et ContainerUpdateIDs.
/// `container_ids` doit contenir les IDs des conteneurs impactés.
pub fn notify_containers_updated(container_ids: &[&str]) {
let new_id = SYSTEM_UPDATE_ID.fetch_add(1, Ordering::Relaxed).saturating_add(1);
set_system_update_id(new_id);
if !container_ids.is_empty() {
let mut buf = String::new();
for (idx, cid) in container_ids.iter().enumerate() {
if idx > 0 {
buf.push(',');
}
buf.push_str(cid);
buf.push(',');
buf.push_str(&new_id.to_string());
}
set_container_update_ids(&buf);
}
}
fn set_system_update_id(id: u32) {
tracing::info!("ContentDirectory: SystemUpdateID -> {}", id);
if let Some(service) = CONTENTDIR_INSTANCE.get().and_then(|w| w.upgrade()) {
if let Some(var) = service.get_variable("SystemUpdateID") {
let _ = var.set_value(StateValue::UI4(id));
}
}
}
fn set_container_update_ids(value: &str) {
tracing::info!("ContentDirectory: ContainerUpdateIDs -> {}", value);
{
let mut guard = CONTAINER_UPDATE_IDS.lock().unwrap();
*guard = value.to_string();
}
if let Some(service) = CONTENTDIR_INSTANCE.get().and_then(|w| w.upgrade()) {
if let Some(var) = service.get_variable("ContainerUpdateIDs") {
let _ = var.set_value(StateValue::String(value.to_string()));
}
}
}

11
pmomediaserver/src/sources.rs
View File

@@ -173,7 +173,8 @@ impl SourcesExt for Server {
#[cfg(feature = "paradise")]
async fn register_paradise(&mut self) -> Result<()> {
use pmoparadise::{RadioParadiseClient, RadioParadiseExt, RadioParadiseSource};
use pmoparadise::{RadioParadiseExt, RadioParadiseSource};
use crate::contentdirectory::state;
tracing::info!("Initializing Radio Paradise source...");
@@ -181,7 +182,13 @@ impl SourcesExt for Server {
let base_url = self.base_url();
// Créer la source Radio Paradise (utilise le singleton PlaylistManager)
let source = Arc::new(RadioParadiseSource::new(base_url.to_string()));
let notifier = Arc::new(|containers: &[String]| {
let refs: Vec<&str> = containers.iter().map(|s| s.as_str()).collect();
state::notify_containers_updated(&refs);
});
let source = Arc::new(
RadioParadiseSource::new(base_url.to_string()).with_container_notifier(notifier),
);
// Brancher les callbacks de playlists (live/history) pour signaler les updates
source.attach_playlist_callbacks();

47
pmoparadise/src/source.rs
View File

@@ -4,6 +4,7 @@
//! exposing live streams and historical playlists for all 4 channels.
use crate::channels::{ChannelDescriptor, ALL_CHANNELS};
use std::fmt;
use pmosource::pmodidl::{Container, Item, Resource};
use pmosource::{
async_trait, AudioFormat, BrowseResult, MusicSource, MusicSourceError, Result,
@@ -32,7 +33,7 @@ const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
/// - 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)]
#[derive(Clone)]
pub struct RadioParadiseSource {
/// Base URL for streaming server (e.g., "http://localhost:8080")
base_url: String,
@@ -42,6 +43,16 @@ pub struct RadioParadiseSource {
last_change: Arc<RwLock<SystemTime>>,
/// Tokens des callbacks enregistrés auprès du PlaylistManager
callback_tokens: Arc<std::sync::Mutex<Vec<u64>>>,
/// Notifier optionnel pour signaler les mises à jour de conteneurs au ContentDirectory
container_notifier: Option<Arc<dyn Fn(&[String]) + Send + Sync + 'static>>,
}
impl fmt::Debug for RadioParadiseSource {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("RadioParadiseSource")
.field("base_url", &self.base_url)
.finish_non_exhaustive()
}
}
impl RadioParadiseSource {
@@ -61,9 +72,19 @@ impl RadioParadiseSource {
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())),
container_notifier: None,
}
}
/// Injecte un notifier pour propager les changements de playlists vers le ContentDirectory
pub fn with_container_notifier(
mut self,
notifier: Arc<dyn Fn(&[String]) + Send + Sync + 'static>,
) -> Self {
self.container_notifier = Some(notifier);
self
}
/// Build a live stream URL for a channel
fn build_live_url(&self, slug: &str) -> String {
format!("{}/radioparadise/stream/{}/flac", self.base_url, slug)
@@ -104,11 +125,35 @@ impl RadioParadiseSource {
for pid in ids {
let weak = Arc::downgrade(self);
let pid_clone = pid.clone();
let token = mgr.register_callback(move |changed_id| {
let pid = pid_clone.clone();
if changed_id == pid {
if let Some(strong) = weak.upgrade() {
tokio::spawn(async move {
strong.bump_update_counter().await;
// Notifier ContentDirectory des conteneurs concernés
let containers: Vec<String> = if pid.contains("history") {
// history playlist -> container history
ALL_CHANNELS
.iter()
.find(|ch| pid.ends_with(ch.slug))
.map(|ch| vec![format!("radio-paradise:channel:{}:history", ch.slug)])
.unwrap_or_default()
} else {
// live playlist -> container liveplaylist
ALL_CHANNELS
.iter()
.find(|ch| pid.ends_with(ch.slug))
.map(|ch| vec![format!("radio-paradise:channel:{}:liveplaylist", ch.slug)])
.unwrap_or_default()
};
if !containers.is_empty() {
if let Some(notifier) = strong.container_notifier.as_ref() {
notifier(&containers);
}
}
});
}
}

25
pmoserver/src/server.rs
View File

@@ -17,7 +17,7 @@ use crate::logs::{LogState, init_logging, log_dump, log_sse};
use axum::extract::State;
use axum::handler::Handler;
use axum::response::Redirect;
use axum::routing::{get, post};
use axum::routing::{any, get, post};
use axum::{Json, Router};
use axum_embed::ServeEmbed;
use pmoconfig::get_config;
@@ -240,6 +240,29 @@ impl Server {
};
}
/// Ajoute un handler qui accepte tous les verbes HTTP (ANY) avec état
pub async fn add_any_handler_with_state<H, T, S>(
&mut self,
path: &str,
handler: H,
state: S,
) where
H: Handler<T, S> + Clone + 'static,
T: 'static,
S: Clone + Send + Sync + 'static,
{
let route = Router::new()
.route("/", any(handler.clone()))
.with_state(state.clone());
let mut r = self.router.write().await;
*r = if path == "/" {
std::mem::take(&mut *r).merge(route)
} else {
std::mem::take(&mut *r).nest(path, route)
};
}
/// Ajoute un répertoire statique
pub async fn add_dir<E>(&mut self, path: &str)
where

4
pmoupnp/src/services/service_instance.rs
View File

@@ -579,10 +579,10 @@ impl ServiceInstance {
.add_post_handler_with_state(&self.control_route(), control_handler, instance_control)
.await;
// Handler événements
// Handler événements (SUBSCRIBE/UNSUBSCRIBE sont des verbes spécifiques, pas GET)
let instance_event = self.clone();
server
.add_handler_with_state(&self.event_route(), event_sub_handler, instance_event)
.add_any_handler_with_state(&self.event_route(), event_sub_handler, instance_event)
.await;
Ok(())