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

Gerer les souscription aux variables

Browse Source
This commit is contained in:
Eric Coissac
2025-10-09 07:00:48 +02:00
parent e5c83d01c0
commit fd1a44b6fa
3 changed files with 140 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pmoapp/src/lib.rs
View File

@@ -298,7 +298,7 @@ pub trait WebAppExt {
/// # Type Parameter
///
/// * `W` - Type RustEmbed contenant les fichiers de la webapp
fn add_webapp<W>(&mut self, path: &str) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>
async fn add_webapp<W>(&mut self, path: &str)
where
W: RustEmbed + Clone + Send + Sync + 'static;
@@ -311,7 +311,7 @@ pub trait WebAppExt {
/// # Type Parameter
///
/// * `W` - Type RustEmbed contenant les fichiers de la webapp
fn add_webapp_with_redirect<W>(&mut self, path: &str) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>
async fn add_webapp_with_redirect<W>(&mut self, path: &str)
where
W: RustEmbed + Clone + Send + Sync + 'static;
}

18
pmoapp/src/pmoserver_impl.rs
View File

@@ -30,28 +30,24 @@
use crate::WebAppExt;
use pmoserver::Server;
use rust_embed::RustEmbed;
use std::future::Future;
use std::pin::Pin;
impl WebAppExt for Server {
fn add_webapp<W>(&mut self, path: &str) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>
async fn add_webapp<W>(&mut self, path: &str)
where
W: RustEmbed + Clone + Send + Sync + 'static,
{
let path = path.to_string();
Box::pin(async move {
self.add_spa::<W>(&path).await;
})
self.add_spa::<W>(&path).await;
}
fn add_webapp_with_redirect<W>(&mut self, path: &str) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>
async fn add_webapp_with_redirect<W>(&mut self, path: &str)
where
W: RustEmbed + Clone + Send + Sync + 'static,
{
let path = path.to_string();
Box::pin(async move {
self.add_spa::<W>(&path).await;
self.add_redirect("/", &path).await;
})
self.add_spa::<W>(&path).await;
self.add_redirect("/", &path).await;
}
}

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

@@ -1,28 +1,28 @@
//! Implémentation de ServiceInstance.
use std::{
collections::HashMap,
sync::{Arc, Mutex, RwLock},
time::Duration,
pin::Pin,
future::Future,
};
use axum::{
body::Body,
extract::{Request, State},
http::{HeaderMap, StatusCode},
response::{IntoResponse, Response},
body::Body,
};
use std::{
collections::HashMap,
future::Future,
pin::Pin,
sync::{Arc, Mutex, RwLock},
time::Duration,
};
use tokio::time;
use tracing::{info, warn, error};
use xmltree::{Element, XMLNode, EmitterConfig};
use tracing::{error, info, warn};
use xmltree::{Element, EmitterConfig, XMLNode};
use crate::{
services::{Service, ServiceError},
UpnpInstance, UpnpObject, UpnpObjectType, UpnpTyped, UpnpTypedInstance,
actions::{ActionInstance, ActionInstanceSet},
state_variables::{StateVarInstance, StateVarInstanceSet, UpnpVariable},
devices::DeviceInstance,
UpnpObject, UpnpInstance, UpnpTyped, UpnpTypedInstance, UpnpObjectType,
services::{Service, ServiceError},
state_variables::{StateVarInstance, StateVarInstanceSet, UpnpVariable},
};
/// Méthodes HTTP pour les événements UPnP.
@@ -96,7 +96,6 @@ pub struct ServiceInstance {
seqid: Arc<Mutex<HashMap<String, u32>>>,
}
impl std::fmt::Debug for ServiceInstance {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ServiceInstance")
@@ -194,7 +193,9 @@ impl UpnpObject for ServiceInstance {
let mut elem = Element::new("service");
let mut service_type = Element::new("serviceType");
service_type.children.push(XMLNode::Text(self.service_type()));
service_type
.children
.push(XMLNode::Text(self.service_type()));
elem.children.push(XMLNode::Element(service_type));
let mut service_id = Element::new("serviceId");
@@ -206,11 +207,15 @@ impl UpnpObject for ServiceInstance {
elem.children.push(XMLNode::Element(scpd_url));
let mut control_url = Element::new("controlURL");
control_url.children.push(XMLNode::Text(self.control_route()));
control_url
.children
.push(XMLNode::Text(self.control_route()));
elem.children.push(XMLNode::Element(control_url));
let mut event_sub_url = Element::new("eventSubURL");
event_sub_url.children.push(XMLNode::Text(self.event_route()));
event_sub_url
.children
.push(XMLNode::Text(self.event_route()));
elem.children.push(XMLNode::Element(event_sub_url));
elem
@@ -306,8 +311,14 @@ impl ServiceInstance {
/// Retourne une erreur si l'enregistrement des routes échoue.
pub async fn register_urls(&self, server: &mut pmoserver::Server) -> Result<(), ServiceError> {
let device = self.device.read().unwrap();
let device_name = device.as_ref().map(|d| d.get_name().clone()).unwrap_or_else(|| "unknown".to_string());
let server_url = device.as_ref().map(|d| d.base_url().to_string()).unwrap_or_default();
let device_name = device
.as_ref()
.map(|d| d.get_name().clone())
.unwrap_or_else(|| "unknown".to_string());
let server_url = device
.as_ref()
.map(|d| d.base_url().to_string())
.unwrap_or_default();
drop(device);
info!(
@@ -320,26 +331,24 @@ impl ServiceInstance {
// Handler SCPD
let instance_scpd = self.clone();
server.add_handler(&self.scpd_route(), move || {
let instance = instance_scpd.clone();
async move { instance.scpd_handler().await }
}).await;
server
.add_handler(&self.scpd_route(), move || {
let instance = instance_scpd.clone();
async move { instance.scpd_handler().await }
})
.await;
// Handler control
let instance_control = self.clone();
server.add_post_handler_with_state(
&self.control_route(),
control_handler,
instance_control,
).await;
server
.add_post_handler_with_state(&self.control_route(), control_handler, instance_control)
.await;
// Handler événements
let instance_event = self.clone();
server.add_handler_with_state(
&self.event_route(),
event_sub_handler,
instance_event,
).await;
server
.add_handler_with_state(&self.event_route(), event_sub_handler, instance_event)
.await;
Ok(())
}
@@ -366,16 +375,14 @@ impl ServiceInstance {
// actionList
if !self.actions.all().is_empty() {
elem.children.push(XMLNode::Element(
self.actions.to_xml_element()
));
elem.children
.push(XMLNode::Element(self.actions.to_xml_element()));
}
// serviceStateTable
if !self.statevariables.all().is_empty() {
elem.children.push(XMLNode::Element(
self.statevariables.to_xml_element()
));
elem.children
.push(XMLNode::Element(self.statevariables.to_xml_element()));
}
elem
@@ -399,9 +406,13 @@ impl ServiceInstance {
(
StatusCode::OK,
[(axum::http::header::CONTENT_TYPE, "text/xml; charset=\"utf-8\"")],
[(
axum::http::header::CONTENT_TYPE,
"text/xml; charset=\"utf-8\"",
)],
xml,
).into_response()
)
.into_response()
}
/// Ajoute un abonné aux événements.
@@ -443,9 +454,13 @@ impl ServiceInstance {
tokio::spawn(async move {
let callback = callback.trim().trim_matches(|c| c == '<' || c == '>');
let mut body = r#"<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">"#.to_string();
let mut body =
r#"<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">"#.to_string();
for (name, val) in changed {
body.push_str(&format!("<e:property><{0}>{1}</{0}></e:property>", name, val));
body.push_str(&format!(
"<e:property><{0}>{1}</{0}></e:property>",
name, val
));
}
body.push_str("</e:propertyset>");
@@ -462,7 +477,11 @@ impl ServiceInstance {
.await
{
Ok(resp) => {
info!("✅ Initial event sent to {}, status={}", callback, resp.status());
info!(
"✅ Initial event sent to {}, status={}",
callback,
resp.status()
);
}
Err(e) => {
error!("Failed to send initial event to {}: {}", callback, e);
@@ -511,9 +530,13 @@ impl ServiceInstance {
tokio::spawn(async move {
let callback = callback.trim().trim_matches(|c| c == '<' || c == '>');
let mut body = r#"<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">"#.to_string();
let mut body =
r#"<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">"#.to_string();
for (name, val) in changed_clone {
body.push_str(&format!("<e:property><{0}>{1}</{0}></e:property>", name, val));
body.push_str(&format!(
"<e:property><{0}>{1}</{0}></e:property>",
name, val
));
}
body.push_str("</e:propertyset>");
@@ -565,18 +588,26 @@ impl ServiceInstance {
}
/// Handler Axum pour les événements (SUBSCRIBE/UNSUBSCRIBE).
fn event_sub_handler(
async fn event_sub_handler(
State(instance): State<ServiceInstance>,
headers: HeaderMap,
req: Request<Body>,
) -> Pin<Box<dyn Future<Output = Response> + Send>> {
Box::pin(async move {
) -> Response {
info!("📡 Event Subscription request for {}", instance.get_name());
let method = req.method().as_str();
let sid = headers.get("SID").and_then(|v| v.to_str().ok()).unwrap_or("");
let timeout = headers.get("Timeout").and_then(|v| v.to_str().ok()).unwrap_or("");
let callback = headers.get("Callback").and_then(|v| v.to_str().ok()).unwrap_or("");
let sid = headers
.get("SID")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let timeout = headers
.get("Timeout")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let callback = headers
.get("Callback")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
match method {
METHOD_SUBSCRIBE => {
@@ -584,14 +615,19 @@ fn event_sub_handler(
// Nouvelle souscription
let new_sid = format!("uuid:{}", uuid::Uuid::new_v4());
if !callback.is_empty() {
instance.add_subscriber(new_sid.clone(), callback.to_string()).await;
instance
.add_subscriber(new_sid.clone(), callback.to_string())
.await;
}
let timeout_val = if timeout.is_empty() {
"Second-1800"
} else {
timeout
};
info!("🔒 New subscription: SID={}, Callback={}, Timeout={}", new_sid, callback, timeout_val);
info!(
"🔒 New subscription: SID={}, Callback={}, Timeout={}",
new_sid, callback, timeout_val
);
let sid_clone = new_sid.clone();
let instance_clone = instance.clone();
@@ -612,14 +648,15 @@ fn event_sub_handler(
[
(
axum::http::header::HeaderName::from_static("sid"),
axum::http::HeaderValue::from_str(&response_sid).unwrap()
axum::http::HeaderValue::from_str(&response_sid).unwrap(),
),
(
axum::http::header::HeaderName::from_static("timeout"),
axum::http::HeaderValue::from_str(&response_timeout).unwrap()
axum::http::HeaderValue::from_str(&response_timeout).unwrap(),
),
],
).into_response()
)
.into_response()
}
METHOD_UNSUBSCRIBE => {
if !sid.is_empty() {
@@ -633,15 +670,10 @@ fn event_sub_handler(
StatusCode::METHOD_NOT_ALLOWED.into_response()
}
}
})
}
/// Handler Axum pour le contrôle SOAP.
fn control_handler(
State(instance): State<ServiceInstance>,
_body: String,
) -> Pin<Box<dyn Future<Output = Response> + Send>> {
Box::pin(async move {
async fn control_handler(State(instance): State<ServiceInstance>, _body: String) -> Response {
info!("📡 Control request for {}", instance.get_name());
// TODO: Parser le SOAP et appeler l'action correspondante
@@ -660,10 +692,13 @@ fn control_handler(
(
StatusCode::OK,
[(axum::http::header::CONTENT_TYPE, "text/xml; charset=\"utf-8\"")],
[(
axum::http::header::CONTENT_TYPE,
"text/xml; charset=\"utf-8\"",
)],
response_xml,
).into_response()
})
)
.into_response()
}
#[cfg(test)]