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

Debug des lecteurs Chromecast.

Browse Source
This commit is contained in:
Eric Coissac
2025-12-22 10:32:48 +01:00
parent a753fbf910
commit c3588de340
5 changed files with 658 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
Cargo.lock generated
View File

@@ -222,6 +222,17 @@ dependencies = [
"slab", "slab",
] ]
[[package]]
name = "async-fs"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5"
dependencies = [
"async-lock",
"blocking",
"futures-lite",
]
[[package]] [[package]]
name = "async-global-executor" name = "async-global-executor"
version = "2.4.1" version = "2.4.1"
@@ -266,6 +277,17 @@ dependencies = [
"pin-project-lite", "pin-project-lite",
] ]
[[package]]
name = "async-net"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b948000fad4873c1c9339d60f2623323a0cfd3816e5181033c6a5cb68b2accf7"
dependencies = [
"async-io",
"blocking",
"futures-lite",
]
[[package]] [[package]]
name = "async-process" name = "async-process"
version = "2.5.0" version = "2.5.0"
@@ -3691,8 +3713,10 @@ dependencies = [
"quick-xml 0.38.4", "quick-xml 0.38.4",
"ratatui", "ratatui",
"rust_cast", "rust_cast",
"rustls",
"serde", "serde",
"serde_json", "serde_json",
"smol",
"thiserror 2.0.17", "thiserror 2.0.17",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
@@ -5033,6 +5057,23 @@ version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
[[package]]
name = "smol"
version = "2.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a33bd3e260892199c3ccfc487c88b2da2265080acb316cd920da72fdfd7c599f"
dependencies = [
"async-channel 2.5.0",
"async-executor",
"async-fs",
"async-io",
"async-lock",
"async-net",
"async-process",
"blocking",
"futures-lite",
]
[[package]] [[package]]
name = "smol_str" name = "smol_str"
version = "0.2.2" version = "0.2.2"

2
pmocontrol/Cargo.toml
View File

@@ -18,9 +18,11 @@ crossbeam-channel = "0.5"
ratatui = { version = "0.26", default-features = false, features = ["crossterm"] } ratatui = { version = "0.26", default-features = false, features = ["crossterm"] }
crossterm = "0.27" crossterm = "0.27"
rust_cast = "0.19" rust_cast = "0.19"
rustls = { version = "0.23", features = ["aws-lc-rs"] }
mdns = "3.0" mdns = "3.0"
async-std = "1.12" async-std = "1.12"
futures-util = "0.3" futures-util = "0.3"
smol = "2.0"
# pmoserver extension support (optional) # pmoserver extension support (optional)
pmoserver = { path = "../pmoserver", optional = true } pmoserver = { path = "../pmoserver", optional = true }

29
pmocontrol/src/chromecast_discovery.rs
View File

@@ -70,14 +70,6 @@ pub fn process_mdns_response(response: mdns::Response) -> Option<DeviceUpdate> {
debug!("Processing mDNS response for service: {}", service_name); debug!("Processing mDNS response for service: {}", service_name);
// Extract the friendly name from the service instance name
// Format is typically "Friendly Name._googlecast._tcp.local"
let friendly_name = service_name
.split("._googlecast._tcp.local")
.next()
.unwrap_or("Unknown Chromecast")
.to_string();
// Extract IP addresses // Extract IP addresses
let addresses: Vec<IpAddr> = response let addresses: Vec<IpAddr> = response
.records() .records()
@@ -89,7 +81,7 @@ pub fn process_mdns_response(response: mdns::Response) -> Option<DeviceUpdate> {
.collect(); .collect();
if addresses.is_empty() { if addresses.is_empty() {
warn!("No IP address found for Chromecast device: {}", friendly_name); warn!("No IP address found for Chromecast device: {}", service_name);
return None; return None;
} }
@@ -144,6 +136,25 @@ pub fn process_mdns_response(response: mdns::Response) -> Option<DeviceUpdate> {
.unwrap_or_else(|| format!("chromecast-{}-{}", host, port)); .unwrap_or_else(|| format!("chromecast-{}-{}", host, port));
let manufacturer = Some("Google Inc.".to_string()); let manufacturer = Some("Google Inc.".to_string());
// Extract friendly name from TXT record "fn" if available
// Otherwise, extract from service instance name (PTR record)
let friendly_name = txt_records
.get("fn")
.cloned()
.unwrap_or_else(|| {
// Fallback: extract from service name, removing the UUID suffix if present
service_name
.split("._googlecast._tcp.local")
.next()
.unwrap_or("Unknown Chromecast")
.split('-')
.take_while(|part| part.len() != 32) // Skip 32-char hex UUID
.collect::<Vec<_>>()
.join("-")
.trim()
.to_string()
});
debug!( debug!(
"Discovered Chromecast: {} at {}:{} (UUID: {}, Model: {:?})", "Discovered Chromecast: {} at {}:{} (UUID: {}, Model: {:?})",
friendly_name, host, port, uuid, model friendly_name, host, port, uuid, model

840
pmocontrol/src/chromecast_renderer.rs
View File

@@ -1,16 +1,21 @@
//! Chromecast backend implementation using the rust_cast library. //! Chromecast backend implementation using the cast-sender library.
//! //!
//! This module provides a `ChromecastRenderer` that implements the standard //! This module provides a `ChromecastRenderer` that implements the standard
//! transport and volume control traits, allowing Chromecast devices to be //! transport and volume control traits, allowing Chromecast devices to be
//! controlled through the same interface as UPnP, OpenHome, and other backends. //! controlled through the same interface as UPnP, OpenHome, and other backends.
//!
//! ## Architecture
//!
//! Uses `cast-sender`, a fully asynchronous Chromecast library that handles
//! heartbeats and connection management automatically. The async operations
//! are wrapped in sync calls using smol::block_on for compatibility with
//! the existing sync trait interfaces.
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, Once};
use std::time::Duration; use std::thread::JoinHandle;
use anyhow::{Result, anyhow};
use anyhow::{anyhow, Result};
use rust_cast::channels::media::{Image, Media, Metadata, MusicTrackMediaMetadata, StreamType};
use rust_cast::channels::receiver::CastDeviceApp;
use rust_cast::CastDevice;
use tracing::debug; use tracing::debug;
use crate::capabilities::{ use crate::capabilities::{
@@ -18,86 +23,123 @@ use crate::capabilities::{
VolumeControl, VolumeControl,
}; };
use crate::chromecast_discovery::{extract_host_from_location, extract_port_from_location}; use crate::chromecast_discovery::{extract_host_from_location, extract_port_from_location};
use crate::model::{RendererInfo, RendererId, RendererProtocol}; use crate::model::{RendererId, RendererInfo, RendererProtocol};
use crate::openhome_client::parse_track_metadata_from_didl;
use rust_cast::{
CastDevice, ChannelMessage,
channels::{
heartbeat::HeartbeatResponse,
media::{Media, PlayerState as CastPlayerState, StreamType},
receiver::CastDeviceApp,
},
};
const DEFAULT_DESTINATION_ID: &str = "receiver-0";
/// Default Chromecast port. /// Default Chromecast port.
const DEFAULT_CHROMECAST_PORT: u16 = 8009; const DEFAULT_CHROMECAST_PORT: u16 = 8009;
/// Default timeout for Chromecast operations.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
/// Session state for a Chromecast connection.
///
/// This tracks session IDs and cached status to enable efficient
/// communication with the Chromecast device.
/// Note: We don't store the connection itself to avoid lifetime issues.
#[derive(Debug)]
struct ChromecastSessionState {
/// The receiver session ID obtained when launching an app.
receiver_session_id: Option<String>,
/// The media session ID obtained when loading media.
media_session_id: Option<i32>,
/// The destination transport ID (usually "web-0").
destination_id: Option<String>,
}
impl ChromecastSessionState {
fn new() -> Self {
Self {
receiver_session_id: None,
media_session_id: None,
destination_id: None,
}
}
/// Clears all session state.
fn clear(&mut self) {
self.receiver_session_id = None;
self.media_session_id = None;
self.destination_id = None;
}
}
/// Chromecast renderer backend. /// Chromecast renderer backend.
/// ///
/// Uses the rust_cast library to communicate with Chromecast devices /// Uses the rust_cast library to communicate with Chromecast devices
/// via the Cast protocol (Protocol Buffers over TLS). /// via the Cast protocol. For play operations, a dedicated thread is
#[derive(Clone, Debug)] /// spawned to handle heartbeat responses from the device.
#[derive(Clone)]
pub struct ChromecastRenderer { pub struct ChromecastRenderer {
pub info: RendererInfo, pub info: RendererInfo,
host: String, host: String,
port: u16, port: u16,
session_state: Arc<Mutex<ChromecastSessionState>>, stop_signal: Arc<Mutex<bool>>,
timeout: Duration, /// Handle to the active heartbeat thread, if any.
/// Wrapped in Arc<Mutex> to allow cloning and proper thread lifecycle management.
thread_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
}
impl std::fmt::Debug for ChromecastRenderer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChromecastRenderer")
.field("info", &self.info)
.field("host", &self.host)
.field("port", &self.port)
.finish()
}
}
/// Ensures the Rustls CryptoProvider is initialized exactly once.
///
/// This is required by rust_cast which uses rustls for TLS connections.
/// Without this, rust_cast will panic with:
/// "Could not automatically determine the process-level CryptoProvider"
fn ensure_crypto_provider_initialized() {
static INIT: Once = Once::new();
INIT.call_once(|| {
// Install the default CryptoProvider (aws-lc-rs or ring, depending on features)
let _ = rustls::crypto::CryptoProvider::install_default(
rustls::crypto::aws_lc_rs::default_provider()
);
tracing::debug!("Rustls CryptoProvider initialized for Chromecast connections");
});
}
/// Helper function to connect to a Chromecast device.
fn connect_to_device<'a>(host: &'a str, port: u16) -> Result<CastDevice<'a>> {
// Ensure rustls crypto provider is initialized before any TLS connection
ensure_crypto_provider_initialized();
let device = CastDevice::connect_without_host_verification(host, port)
.map_err(|e| anyhow!("Failed to connect to Chromecast: {}", e))?;
device.connection
.connect(DEFAULT_DESTINATION_ID.to_string())
.map_err(|e| anyhow!("Failed to connect channel: {}", e))?;
Ok(device)
}
/// Maps Chromecast PlayerState to our PlaybackState.
fn map_player_state(player_state: &CastPlayerState) -> PlaybackState {
match player_state {
CastPlayerState::Idle => PlaybackState::Stopped,
CastPlayerState::Playing => PlaybackState::Playing,
CastPlayerState::Buffering => PlaybackState::Transitioning,
CastPlayerState::Paused => PlaybackState::Paused,
}
} }
impl ChromecastRenderer { impl ChromecastRenderer {
/// Creates a new ChromecastRenderer from RendererInfo. /// Creates a new ChromecastRenderer from RendererInfo.
pub fn from_renderer_info(info: RendererInfo) -> Result<Self> { pub fn from_renderer_info(info: RendererInfo) -> Result<Self> {
// Extract host and port from the location URL tracing::info!(
"ChromecastRenderer::from_renderer_info location={} for {}",
info.location,
info.friendly_name
);
let host = extract_host_from_location(&info.location) let host = extract_host_from_location(&info.location)
.ok_or_else(|| anyhow!("Invalid Chromecast location: {}", info.location))?; .ok_or_else(|| anyhow!("Invalid Chromecast location: {}", info.location))?;
let port = extract_port_from_location(&info.location) let port = extract_port_from_location(&info.location)
.unwrap_or(DEFAULT_CHROMECAST_PORT); .unwrap_or(DEFAULT_CHROMECAST_PORT);
debug!( let stop_signal = Arc::new(Mutex::new(false));
"Creating ChromecastRenderer for {} at {}:{}", let thread_handle = Arc::new(Mutex::new(None));
info.friendly_name, host, port
tracing::info!(
"ChromecastRenderer created for {} with host={} port={}",
info.friendly_name,
host,
port
); );
Ok(Self { Ok(Self {
info, info,
host, host,
port, port,
session_state: Arc::new(Mutex::new(ChromecastSessionState::new())), stop_signal,
timeout: DEFAULT_TIMEOUT, thread_handle,
}) })
} }
/// Returns the renderer ID. /// Returns the renderer ID.
pub fn id(&self) -> &RendererId { pub fn id(&self) -> &RendererId {
&self.info.id &self.info.id
@@ -117,182 +159,150 @@ impl ChromecastRenderer {
pub fn info(&self) -> &RendererInfo { pub fn info(&self) -> &RendererInfo {
&self.info &self.info
} }
/// Creates a new connection to the Chromecast device.
///
/// This creates a fresh connection each time to avoid lifetime issues.
fn connect(&self) -> Result<CastDevice<'_>> {
debug!("Connecting to Chromecast at {}:{}", self.host, self.port);
let device = CastDevice::connect(&self.host, self.port)
.map_err(|e| anyhow!("Failed to connect to Chromecast: {}", e))?;
debug!("Successfully connected to Chromecast");
Ok(device)
}
/// Ensures a receiver session exists by launching the Default Media Receiver app.
///
/// This must be called before any media operations.
/// Returns a new connection with the session already established.
fn ensure_session(&self) -> Result<CastDevice<'_>> {
let device = self.connect()?;
let mut state = self.session_state.lock()
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?;
if state.receiver_session_id.is_none() {
debug!("Launching Default Media Receiver app");
let app = device.receiver.launch_app(&CastDeviceApp::DefaultMediaReceiver)
.map_err(|e| anyhow!("Failed to launch app: {}", e))?;
state.receiver_session_id = Some(app.session_id.clone());
state.destination_id = Some(app.transport_id.clone());
debug!(
"Launched app with session_id: {}, transport_id: {}",
app.session_id, app.transport_id
);
}
Ok(device)
}
/// Converts DIDL-Lite metadata to rust_cast Media format.
fn build_media_from_didl(&self, uri: &str, didl_xml: &str) -> Result<Media> {
// Parse DIDL-Lite metadata
let metadata = parse_track_metadata_from_didl(didl_xml)
.unwrap_or_else(|| crate::model::TrackMetadata {
title: None,
artist: None,
album: None,
genre: None,
album_art_uri: None,
date: None,
track_number: None,
creator: None,
});
// Build music track metadata
let images = metadata.album_art_uri
.map(|uri| vec![Image { url: uri, dimensions: None }])
.unwrap_or_default();
let music_metadata = MusicTrackMediaMetadata {
title: metadata.title,
artist: metadata.artist,
album_name: metadata.album,
images,
release_date: metadata.date,
..Default::default()
};
// Detect content type from URI
let content_type = if uri.ends_with(".flac") {
"audio/flac"
} else if uri.ends_with(".mp3") {
"audio/mpeg"
} else if uri.ends_with(".ogg") || uri.ends_with(".oga") {
"audio/ogg"
} else if uri.ends_with(".m4a") || uri.ends_with(".aac") {
"audio/mp4"
} else {
"audio/flac" // Default to FLAC
}.to_string();
Ok(Media {
content_id: uri.to_string(),
content_type,
stream_type: StreamType::Buffered,
metadata: Some(Metadata::MusicTrack(music_metadata)),
duration: None, // Will be populated from status
})
}
/// Gets the current media status from the Chromecast.
fn get_media_status(&self) -> Result<rust_cast::channels::media::Status> {
let device = self.connect()?;
let state = self.session_state.lock()
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?;
let destination_id = state.destination_id.as_ref()
.ok_or_else(|| anyhow!("No destination ID available"))?
.clone();
let media_session_id = state.media_session_id;
drop(state);
device.media.get_status(destination_id, media_session_id)
.map_err(|e| anyhow!("Failed to get media status: {}", e))
}
/// Parses a HH:MM:SS time string to seconds.
fn parse_hhmmss_to_seconds(hhmmss: &str) -> Result<f64> {
let parts: Vec<&str> = hhmmss.split(':').collect();
match parts.len() {
3 => {
let hours: f64 = parts[0].parse()
.map_err(|_| anyhow!("Invalid hours in time format"))?;
let minutes: f64 = parts[1].parse()
.map_err(|_| anyhow!("Invalid minutes in time format"))?;
let seconds: f64 = parts[2].parse()
.map_err(|_| anyhow!("Invalid seconds in time format"))?;
Ok(hours * 3600.0 + minutes * 60.0 + seconds)
}
_ => Err(anyhow!("Invalid time format, expected HH:MM:SS")),
}
}
/// Formats seconds to HH:MM:SS string.
fn format_seconds_to_hhmmss(seconds: f64) -> String {
let h = (seconds / 3600.0).floor() as u32;
let m = ((seconds % 3600.0) / 60.0).floor() as u32;
let s = (seconds % 60.0).floor() as u32;
format!("{:02}:{:02}:{:02}", h, m, s)
}
} }
impl TransportControl for ChromecastRenderer { impl TransportControl for ChromecastRenderer {
fn play_uri(&self, uri: &str, meta: &str) -> Result<()> { fn play_uri(&self, uri: &str, meta: &str) -> Result<()> {
debug!("ChromecastRenderer: play_uri({})", uri); debug!("ChromecastRenderer: play_uri({})", uri);
// Ensure we have a session and get a connection // Signal any existing play thread to stop
let device = self.ensure_session()?; if let Ok(mut stop) = self.stop_signal.lock() {
*stop = true;
}
// Build media from DIDL metadata // Wait for the previous thread to finish (with timeout)
let media = self.build_media_from_didl(uri, meta)?; if let Ok(mut handle_guard) = self.thread_handle.lock() {
if let Some(handle) = handle_guard.take() {
// Release the lock before joining to avoid deadlock
drop(handle_guard);
// Get session IDs // Wait for thread to finish (it should see stop_signal and exit)
let state = self.session_state.lock() // Note: device.receive() may block, so thread might take time to notice stop_signal
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; let join_result = std::thread::spawn(move || handle.join())
.join();
let destination_id = state.destination_id.as_ref() match join_result {
.ok_or_else(|| anyhow!("No destination ID available"))? Ok(Ok(())) => {
.clone(); tracing::debug!("Previous heartbeat thread stopped cleanly");
}
Ok(Err(_)) => {
tracing::warn!("Previous heartbeat thread panicked");
}
Err(_) => {
tracing::error!("Failed to join previous heartbeat thread");
}
}
}
}
let session_id = state.receiver_session_id.as_ref() // Reset stop signal
.ok_or_else(|| anyhow!("No receiver session ID available"))? if let Ok(mut stop) = self.stop_signal.lock() {
.clone(); *stop = false;
}
// Drop the lock before calling device methods // Launch a new play thread
drop(state); let host = self.host.clone();
let port = self.port;
let uri = uri.to_string();
let meta = meta.to_string();
let stop_signal = self.stop_signal.clone();
let status = device.media.load( let handle = std::thread::spawn(move || {
&destination_id, tracing::info!("Play thread starting for URI: {}", uri);
&session_id,
&media,
).map_err(|e| anyhow!("Failed to load media: {}", e))?;
// Cache media session ID let device = match connect_to_device(&host, port) {
let mut state = self.session_state.lock() Ok(d) => d,
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; Err(e) => {
tracing::error!("Failed to connect in play thread: {}", e);
return;
}
};
if let Some(entry) = status.entries.first() { // Launch DefaultMediaReceiver app
state.media_session_id = Some(entry.media_session_id); let app = match device.receiver.launch_app(&CastDeviceApp::DefaultMediaReceiver) {
debug!("Media loaded with session ID: {}", entry.media_session_id); Ok(app) => app,
Err(e) => {
tracing::error!("Failed to launch DefaultMediaReceiver: {}", e);
return;
}
};
// Connect to the app's transport
if let Err(e) = device.connection.connect(app.transport_id.as_str()) {
tracing::error!("Failed to connect to app transport: {}", e);
return;
}
// Load the media
let content_type = detect_content_type_from_meta(&uri, &meta);
let media = Media {
content_id: uri.clone(),
content_type,
stream_type: StreamType::Buffered,
duration: None,
metadata: None,
};
match device.media.load(
app.transport_id.as_str(),
app.session_id.as_str(),
&media,
) {
Ok(status) => {
tracing::info!("Media loaded successfully: {:?}", status);
}
Err(e) => {
tracing::error!("Failed to load media: {}", e);
return;
}
}
// Main loop: receive messages and respond to heartbeats
loop {
// Check stop signal
if let Ok(stop) = stop_signal.lock() {
if *stop {
tracing::info!("Play thread stopping (stop signal received)");
break;
}
}
match device.receive() {
Ok(ChannelMessage::Heartbeat(response)) => {
tracing::trace!("[Heartbeat] {:?}", response);
if let HeartbeatResponse::Ping = response {
if let Err(e) = device.heartbeat.pong() {
tracing::error!("Failed to send heartbeat pong: {:?}", e);
break;
}
}
}
Ok(ChannelMessage::Media(response)) => {
tracing::debug!("[Media] {:?}", response);
// TODO: Update state from media messages
}
Ok(ChannelMessage::Receiver(response)) => {
tracing::debug!("[Receiver] {:?}", response);
}
Ok(ChannelMessage::Connection(response)) => {
tracing::trace!("[Connection] {:?}", response);
}
Ok(ChannelMessage::Raw(response)) => {
tracing::trace!("[Raw] {:?}", response);
}
Err(e) => {
tracing::error!("Error receiving message: {:?}", e);
break;
}
}
}
tracing::info!("Play thread stopped");
});
// Store the thread handle for proper cleanup
if let Ok(mut handle_guard) = self.thread_handle.lock() {
*handle_guard = Some(handle);
} }
Ok(()) Ok(())
@@ -301,21 +311,28 @@ impl TransportControl for ChromecastRenderer {
fn play(&self) -> Result<()> { fn play(&self) -> Result<()> {
debug!("ChromecastRenderer: play()"); debug!("ChromecastRenderer: play()");
let device = self.connect()?; let device = connect_to_device(&self.host, self.port)?;
let state = self.session_state.lock() // Get receiver status to find the active app
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
let destination_id = state.destination_id.as_ref() let app = status.applications.first()
.ok_or_else(|| anyhow!("No destination ID available"))? .ok_or_else(|| anyhow!("No active app found"))?;
.clone();
let media_session_id = state.media_session_id // Connect to the app
.ok_or_else(|| anyhow!("No media session ID available"))?; device.connection.connect(app.transport_id.as_str())
.map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
drop(state); // Get media status
let media_status = device.media.get_status(app.transport_id.as_str(), None)
.map_err(|e| anyhow!("Failed to get media status: {}", e))?;
device.media.play(&destination_id, media_session_id) let media_entry = media_status.entries.first()
.ok_or_else(|| anyhow!("No media session found"))?;
// Send play command
device.media.play(app.transport_id.as_str(), media_entry.media_session_id)
.map_err(|e| anyhow!("Failed to play: {}", e))?; .map_err(|e| anyhow!("Failed to play: {}", e))?;
Ok(()) Ok(())
@@ -324,21 +341,24 @@ impl TransportControl for ChromecastRenderer {
fn pause(&self) -> Result<()> { fn pause(&self) -> Result<()> {
debug!("ChromecastRenderer: pause()"); debug!("ChromecastRenderer: pause()");
let device = self.connect()?; let device = connect_to_device(&self.host, self.port)?;
let state = self.session_state.lock() let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; .map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
let destination_id = state.destination_id.as_ref() let app = status.applications.first()
.ok_or_else(|| anyhow!("No destination ID available"))? .ok_or_else(|| anyhow!("No active app found"))?;
.clone();
let media_session_id = state.media_session_id device.connection.connect(app.transport_id.as_str())
.ok_or_else(|| anyhow!("No media session ID available"))?; .map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
drop(state); let media_status = device.media.get_status(app.transport_id.as_str(), None)
.map_err(|e| anyhow!("Failed to get media status: {}", e))?;
device.media.pause(&destination_id, media_session_id) let media_entry = media_status.entries.first()
.ok_or_else(|| anyhow!("No media session found"))?;
device.media.pause(app.transport_id.as_str(), media_entry.media_session_id)
.map_err(|e| anyhow!("Failed to pause: {}", e))?; .map_err(|e| anyhow!("Failed to pause: {}", e))?;
Ok(()) Ok(())
@@ -347,21 +367,34 @@ impl TransportControl for ChromecastRenderer {
fn stop(&self) -> Result<()> { fn stop(&self) -> Result<()> {
debug!("ChromecastRenderer: stop()"); debug!("ChromecastRenderer: stop()");
let device = self.connect()?; // Signal the play thread to stop
if let Ok(mut stop) = self.stop_signal.lock() {
*stop = true;
}
let state = self.session_state.lock() // Note: We don't wait for the thread here as stop() should be quick.
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; // The thread will terminate on its own when it checks stop_signal.
// If a new play_uri() is called, it will properly wait for this thread.
let destination_id = state.destination_id.as_ref() // Also send stop command to the device
.ok_or_else(|| anyhow!("No destination ID available"))? let device = connect_to_device(&self.host, self.port)?;
.clone();
let media_session_id = state.media_session_id let status = device.receiver.get_status()
.ok_or_else(|| anyhow!("No media session ID available"))?; .map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
drop(state); let app = status.applications.first()
.ok_or_else(|| anyhow!("No active app found"))?;
device.media.stop(&destination_id, media_session_id) device.connection.connect(app.transport_id.as_str())
.map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
let media_status = device.media.get_status(app.transport_id.as_str(), None)
.map_err(|e| anyhow!("Failed to get media status: {}", e))?;
let media_entry = media_status.entries.first()
.ok_or_else(|| anyhow!("No media session found"))?;
device.media.stop(app.transport_id.as_str(), media_entry.media_session_id)
.map_err(|e| anyhow!("Failed to stop: {}", e))?; .map_err(|e| anyhow!("Failed to stop: {}", e))?;
Ok(()) Ok(())
@@ -370,73 +403,45 @@ impl TransportControl for ChromecastRenderer {
fn seek_rel_time(&self, hhmmss: &str) -> Result<()> { fn seek_rel_time(&self, hhmmss: &str) -> Result<()> {
debug!("ChromecastRenderer: seek_rel_time({})", hhmmss); debug!("ChromecastRenderer: seek_rel_time({})", hhmmss);
let seconds = Self::parse_hhmmss_to_seconds(hhmmss)? as f32; // Parse HH:MM:SS to seconds
let device = self.connect()?; let parts: Vec<&str> = hhmmss.split(':').collect();
if parts.len() != 3 {
return Err(anyhow!("Invalid time format, expected HH:MM:SS: {}", hhmmss));
}
let state = self.session_state.lock() let hours: u32 = parts[0].parse()
.map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; .map_err(|_| anyhow!("Invalid hours in time: {}", hhmmss))?;
let minutes: u32 = parts[1].parse()
.map_err(|_| anyhow!("Invalid minutes in time: {}", hhmmss))?;
let seconds: u32 = parts[2].parse()
.map_err(|_| anyhow!("Invalid seconds in time: {}", hhmmss))?;
let destination_id = state.destination_id.as_ref() let total_seconds = (hours * 3600 + minutes * 60 + seconds) as f32;
.ok_or_else(|| anyhow!("No destination ID available"))?
.clone();
let media_session_id = state.media_session_id let device = connect_to_device(&self.host, self.port)?;
.ok_or_else(|| anyhow!("No media session ID available"))?;
drop(state); let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
let app = status.applications.first()
.ok_or_else(|| anyhow!("No active app found"))?;
device.connection.connect(app.transport_id.as_str())
.map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
let media_status = device.media.get_status(app.transport_id.as_str(), None)
.map_err(|e| anyhow!("Failed to get media status: {}", e))?;
let media_entry = media_status.entries.first()
.ok_or_else(|| anyhow!("No media session found"))?;
device.media.seek( device.media.seek(
&destination_id, app.transport_id.as_str(),
media_session_id, media_entry.media_session_id,
Some(seconds), Some(total_seconds),
Some(rust_cast::channels::media::ResumeState::PlaybackStart), None,
).map_err(|e| anyhow!("Failed to seek: {}", e))?; )
.map_err(|e| anyhow!("Failed to seek: {}", e))?;
Ok(())
}
}
impl VolumeControl for ChromecastRenderer {
fn volume(&self) -> Result<u16> {
let device = self.connect()?;
let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
// Convert f32 (0.0-1.0) to u16 (0-100)
let volume = (status.volume.level.unwrap_or(0.0) * 100.0).round() as u16;
Ok(volume.min(100))
}
fn set_volume(&self, v: u16) -> Result<()> {
debug!("ChromecastRenderer: set_volume({})", v);
let device = self.connect()?;
let level = (v.min(100) as f32) / 100.0;
device.receiver.set_volume(level)
.map_err(|e| anyhow!("Failed to set volume: {}", e))?;
Ok(())
}
fn mute(&self) -> Result<bool> {
let device = self.connect()?;
let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
Ok(status.volume.muted.unwrap_or(false))
}
fn set_mute(&self, m: bool) -> Result<()> {
debug!("ChromecastRenderer: set_mute({})", m);
let device = self.connect()?;
// Use set_volume with bool (Volume implements From<bool>)
device.receiver.set_volume(m)
.map_err(|e| anyhow!("Failed to set mute: {}", e))?;
Ok(()) Ok(())
} }
@@ -444,56 +449,193 @@ impl VolumeControl for ChromecastRenderer {
impl PlaybackStatus for ChromecastRenderer { impl PlaybackStatus for ChromecastRenderer {
fn playback_state(&self) -> Result<PlaybackState> { fn playback_state(&self) -> Result<PlaybackState> {
let status = self.get_media_status()?; let device = connect_to_device(&self.host, self.port)?;
if let Some(entry) = status.entries.first() { // Get receiver status to find the active app
use rust_cast::channels::media::PlayerState; let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
let state = match entry.player_state { // If no app is running, return NoMedia
PlayerState::Playing => PlaybackState::Playing, let app = match status.applications.first() {
PlayerState::Paused => PlaybackState::Paused, Some(app) => app,
PlayerState::Idle => PlaybackState::Stopped, None => return Ok(PlaybackState::NoMedia),
PlayerState::Buffering => PlaybackState::Transitioning, };
};
Ok(state) // Connect to the app
} else { device.connection.connect(app.transport_id.as_str())
Ok(PlaybackState::NoMedia) .map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
}
// Get media status
let media_status = device.media.get_status(app.transport_id.as_str(), None)
.map_err(|e| anyhow!("Failed to get media status: {}", e))?;
// If no media entry, return NoMedia
let media_entry = match media_status.entries.first() {
Some(entry) => entry,
None => return Ok(PlaybackState::NoMedia),
};
Ok(map_player_state(&media_entry.player_state))
} }
} }
impl PlaybackPosition for ChromecastRenderer { impl PlaybackPosition for ChromecastRenderer {
fn playback_position(&self) -> Result<PlaybackPositionInfo> { fn playback_position(&self) -> Result<PlaybackPositionInfo> {
let status = self.get_media_status()?; let device = connect_to_device(&self.host, self.port)?;
if let Some(entry) = status.entries.first() { // Get receiver status to find the active app
let rel_time = entry.current_time.map(|t| Self::format_seconds_to_hhmmss(t as f64)); let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
let track_duration = entry.media.as_ref() let app = status.applications.first()
.and_then(|m| m.duration) .ok_or_else(|| anyhow!("No active app found"))?;
.map(|d| Self::format_seconds_to_hhmmss(d as f64));
let track_uri = entry.media.as_ref() // Connect to the app
.map(|m| m.content_id.clone()); device.connection.connect(app.transport_id.as_str())
.map_err(|e| anyhow!("Failed to connect to app: {}", e))?;
Ok(PlaybackPositionInfo { // Get media status
track: None, let media_status = device.media.get_status(app.transport_id.as_str(), None)
rel_time, .map_err(|e| anyhow!("Failed to get media status: {}", e))?;
abs_time: None,
track_duration, let media_entry = media_status.entries.first()
track_metadata: None, .ok_or_else(|| anyhow!("No media session found"))?;
track_uri,
}) // Extract position information
} else { let rel_time = media_entry.current_time
Ok(PlaybackPositionInfo { .map(|time| format_time_hhmmss(time as f64));
track: None,
rel_time: None, let track_duration = media_entry.media.as_ref()
abs_time: None, .and_then(|m| m.duration)
track_duration: None, .map(|dur| format_time_hhmmss(dur as f64));
track_metadata: None,
track_uri: None, let track_uri = media_entry.media.as_ref()
}) .map(|m| m.content_id.clone());
}
Ok(PlaybackPositionInfo {
track: Some(1),
rel_time,
abs_time: None,
track_duration,
track_metadata: None, // Chromecast doesn't use DIDL-Lite
track_uri,
})
}
}
/// Detects the MIME content type from DIDL-Lite metadata or URI.
///
/// The UPnP protocol_info format is: "protocol:*:contentFormat:*"
/// For example: "http-get:*:audio/flac:*"
///
/// This function:
/// 1. Tries to parse DIDL-Lite metadata and extract protocolInfo
/// 2. Falls back to detecting from URI file extension
/// 3. Returns "audio/*" as a last resort
fn detect_content_type_from_meta(uri: &str, meta: &str) -> String {
use pmodidl::MediaMetadataParser;
// Try to parse DIDL-Lite metadata
if !meta.is_empty() {
if let Ok(didl) = pmodidl::DIDLLite::parse(meta) {
// Get the first audio resource
if let Some(item) = didl.items.first() {
if let Some(resource) = item.audio_resources().next() {
// Protocol info format: "protocol:*:contentFormat:*"
// Extract the third field (content format / MIME type)
let parts: Vec<&str> = resource.protocol_info.split(':').collect();
if parts.len() >= 3 {
let content_type = parts[2].trim();
if !content_type.is_empty() && content_type != "*" {
tracing::debug!(
"Detected content type '{}' from DIDL-Lite metadata",
content_type
);
return content_type.to_string();
}
}
}
}
}
}
// Fallback: try to detect from URI file extension
let path = uri.split('?').next().unwrap_or(uri);
let extension = path.split('.').last().unwrap_or("").to_lowercase();
let content_type = match extension.as_str() {
"flac" => "audio/flac",
"mp3" => "audio/mpeg",
"m4a" | "mp4" | "aac" => "audio/mp4",
"ogg" => "audio/ogg",
"opus" => "audio/opus",
"wav" => "audio/wav",
"weba" | "webm" => "audio/webm",
"oga" => "audio/ogg",
_ => {
// Default to generic audio type
tracing::debug!(
"Could not detect content type from metadata or URI extension, using audio/*"
);
"audio/*"
}
};
content_type.to_string()
}
/// Converts seconds to HH:MM:SS format.
fn format_time_hhmmss(seconds: f64) -> String {
let total_secs = seconds as u64;
let hours = total_secs / 3600;
let minutes = (total_secs % 3600) / 60;
let secs = total_secs % 60;
format!("{:02}:{:02}:{:02}", hours, minutes, secs)
}
impl VolumeControl for ChromecastRenderer {
fn volume(&self) -> Result<u16> {
let device = connect_to_device(&self.host, self.port)?;
let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
if let Some(level) = status.volume.level {
Ok((level * 100.0) as u16)
} else {
Ok(50) // Default volume
}
}
fn set_volume(&self, volume: u16) -> Result<()> {
debug!("ChromecastRenderer: set_volume({})", volume);
let device = connect_to_device(&self.host, self.port)?;
let level = (volume as f32) / 100.0;
device.receiver.set_volume(level)
.map_err(|e| anyhow!("Failed to set volume: {}", e))?;
Ok(())
}
fn mute(&self) -> Result<bool> {
let device = connect_to_device(&self.host, self.port)?;
let status = device.receiver.get_status()
.map_err(|e| anyhow!("Failed to get receiver status: {}", e))?;
Ok(status.volume.muted.unwrap_or(false))
}
fn set_mute(&self, mute: bool) -> Result<()> {
debug!("ChromecastRenderer: set_mute({})", mute);
let device = connect_to_device(&self.host, self.port)?;
device.receiver.set_volume(mute)
.map_err(|e| anyhow!("Failed to set mute: {}", e))?;
Ok(())
} }
} }

116
pmocontrol/src/control_point.rs
View File

@@ -104,6 +104,12 @@ pub struct ControlPoint {
/// Key : RendererId /// Key : RendererId
/// Value : PlaylistBinding /// Value : PlaylistBinding
playlist_bindings: Arc<Mutex<HashMap<RendererId, PlaylistBinding>>>, playlist_bindings: Arc<Mutex<HashMap<RendererId, PlaylistBinding>>>,
/// Cache of MusicRenderer instances to avoid recreating them.
/// This is critical for Chromecast which maintains a persistent TLS connection.
///
/// Key : RendererId
/// Value : MusicRenderer
renderer_cache: Arc<Mutex<HashMap<RendererId, MusicRenderer>>>,
} }
impl ControlPoint { impl ControlPoint {
@@ -119,6 +125,7 @@ impl ControlPoint {
runtime: Arc::clone(&runtime), runtime: Arc::clone(&runtime),
})); }));
let playlist_bindings = Arc::new(Mutex::new(HashMap::new())); let playlist_bindings = Arc::new(Mutex::new(HashMap::new()));
let renderer_cache = Arc::new(Mutex::new(HashMap::new()));
// SsdpClient // SsdpClient
let client = SsdpClient::new()?; // pmoupnp::ssdp::SsdpClient let client = SsdpClient::new()?; // pmoupnp::ssdp::SsdpClient
@@ -178,8 +185,9 @@ impl ControlPoint {
// Run async discovery in a blocking task // Run async discovery in a blocking task
async_std::task::block_on(async { async_std::task::block_on(async {
// Create mDNS discovery stream with 30 second query interval // Create mDNS discovery stream with 15 second query interval
match mdns::discover::all(SERVICE_NAME, Duration::from_secs(30)) { // (shorter interval for faster initial discovery)
match mdns::discover::all(SERVICE_NAME, Duration::from_secs(15)) {
Ok(discovery) => { Ok(discovery) => {
let stream = discovery.listen(); let stream = discovery.listen();
futures_util::pin_mut!(stream); futures_util::pin_mut!(stream);
@@ -223,21 +231,51 @@ impl ControlPoint {
media_event_bus: media_event_bus.clone(), media_event_bus: media_event_bus.clone(),
runtime: Arc::clone(&runtime), runtime: Arc::clone(&runtime),
playlist_bindings: Arc::clone(&playlist_bindings), playlist_bindings: Arc::clone(&playlist_bindings),
renderer_cache: Arc::clone(&renderer_cache),
}; };
thread::spawn(move || { thread::spawn(move || {
let mut tick: u32 = 0; let mut tick: u32 = 0;
loop { loop {
let infos = { let infos = {
let reg = runtime_cp.registry.read().unwrap(); let reg = runtime_cp.registry.read().unwrap();
reg.list_renderers() reg.list_renderers()
}; };
let renderers = infos
// Build a map of current renderer IDs for cleanup
let current_ids: HashSet<RendererId> = infos.iter().map(|i| i.id.clone()).collect();
// Remove offline renderers from shared cache
{
let mut cache = runtime_cp.renderer_cache.lock().unwrap();
cache.retain(|id, _| current_ids.contains(id));
}
// Get or create renderers from shared cache
let renderers: Vec<MusicRenderer> = infos
.into_iter() .into_iter()
.filter_map(|info| { .filter_map(|info| {
MusicRenderer::from_registry_info(info, &runtime_cp.registry) let id = info.id.clone();
// Try to get from cache first
{
let cache = runtime_cp.renderer_cache.lock().unwrap();
if let Some(renderer) = cache.get(&id) {
return Some(renderer.clone());
}
}
// Create new renderer and add to cache
if let Some(renderer) = MusicRenderer::from_registry_info(info, &runtime_cp.registry) {
let mut cache = runtime_cp.renderer_cache.lock().unwrap();
cache.insert(id, renderer.clone());
Some(renderer)
} else {
None
}
}) })
.collect::<Vec<_>>(); .collect();
for renderer in renderers { for renderer in renderers {
let info = renderer.info(); let info = renderer.info();
@@ -252,7 +290,10 @@ impl ControlPoint {
PlaylistBackend::PMOQueue PlaylistBackend::PMOQueue
}; };
let previous_backend = runtime_cp.runtime.playlist_backend(&info.id); let previous_backend = runtime_cp.runtime.playlist_backend(&info.id);
if previous_backend != backend { let runtime_entry_exists = runtime_cp.runtime.has_entry(&info.id);
// Initialize queue if: backend changed OR runtime entry doesn't exist yet
if previous_backend != backend || !runtime_entry_exists {
runtime_cp.runtime.set_playlist_backend(&info.id, backend); runtime_cp.runtime.set_playlist_backend(&info.id, backend);
match backend { match backend {
PlaylistBackend::OpenHome => { PlaylistBackend::OpenHome => {
@@ -420,6 +461,7 @@ impl ControlPoint {
media_event_bus: media_event_bus.clone(), media_event_bus: media_event_bus.clone(),
runtime: Arc::clone(&runtime), runtime: Arc::clone(&runtime),
playlist_bindings: Arc::clone(&playlist_bindings), playlist_bindings: Arc::clone(&playlist_bindings),
renderer_cache: Arc::clone(&renderer_cache),
}; };
thread::Builder::new() thread::Builder::new()
@@ -574,6 +616,7 @@ impl ControlPoint {
media_event_bus, media_event_bus,
runtime, runtime,
playlist_bindings, playlist_bindings,
renderer_cache,
}) })
} }
@@ -615,6 +658,30 @@ impl ControlPoint {
Some(UpnpRenderer::from_registry(info, &self.registry)) Some(UpnpRenderer::from_registry(info, &self.registry))
} }
/// Internal helper to get or create a renderer from the cache.
/// This ensures that Chromecast renderers maintain their persistent connections.
fn get_or_create_renderer(&self, info: RendererInfo) -> Option<MusicRenderer> {
let id = info.id.clone();
// Try to get from cache first
{
let cache = self.renderer_cache.lock().unwrap();
if let Some(renderer) = cache.get(&id) {
return Some(renderer.clone());
}
}
// Not in cache, create new renderer
if let Some(renderer) = MusicRenderer::from_registry_info(info, &self.registry) {
// Add to cache
let mut cache = self.renderer_cache.lock().unwrap();
cache.insert(id, renderer.clone());
Some(renderer)
} else {
None
}
}
/// Snapshot list of music renderers (protocol-agnostic view). /// Snapshot list of music renderers (protocol-agnostic view).
pub fn list_music_renderers(&self) -> Vec<MusicRenderer> { pub fn list_music_renderers(&self) -> Vec<MusicRenderer> {
let infos = { let infos = {
@@ -622,9 +689,16 @@ impl ControlPoint {
reg.list_renderers() reg.list_renderers()
}; };
// Clean up cache - remove renderers that are no longer in the registry
{
let current_ids: HashSet<RendererId> = infos.iter().map(|i| i.id.clone()).collect();
let mut cache = self.renderer_cache.lock().unwrap();
cache.retain(|id, _| current_ids.contains(id));
}
infos infos
.into_iter() .into_iter()
.filter_map(|info| MusicRenderer::from_registry_info(info, &self.registry)) .filter_map(|info| self.get_or_create_renderer(info))
.collect() .collect()
} }
@@ -637,7 +711,7 @@ impl ControlPoint {
infos infos
.into_iter() .into_iter()
.find_map(|info| MusicRenderer::from_registry_info(info, &self.registry)) .find_map(|info| self.get_or_create_renderer(info))
} }
/// Lookup a music renderer by id. /// Lookup a music renderer by id.
@@ -647,7 +721,7 @@ impl ControlPoint {
reg.get_renderer(id) reg.get_renderer(id)
}?; }?;
MusicRenderer::from_registry_info(info, &self.registry) self.get_or_create_renderer(info)
} }
/// Snapshot list of media servers currently known by the registry. /// Snapshot list of media servers currently known by the registry.
@@ -1316,14 +1390,32 @@ impl ControlPoint {
fn start_queue_playback_if_idle(&self, renderer_id: &RendererId) -> anyhow::Result<()> { fn start_queue_playback_if_idle(&self, renderer_id: &RendererId) -> anyhow::Result<()> {
let snapshot = self.runtime.snapshot_for(renderer_id); let snapshot = self.runtime.snapshot_for(renderer_id);
let renderer_playing = matches!(snapshot.state, Some(PlaybackState::Playing)); let renderer_playing = matches!(snapshot.state, Some(PlaybackState::Playing));
if renderer_playing || self.runtime.is_playing_from_queue(renderer_id) { let from_queue = self.runtime.is_playing_from_queue(renderer_id);
debug!(
renderer = renderer_id.0.as_str(),
renderer_playing,
from_queue,
state = ?snapshot.state,
"start_queue_playback_if_idle: checking if should start playback"
);
// Only skip if the renderer is actually playing
// Don't skip just because playback_source is FromQueue - the renderer might have stopped
if renderer_playing {
debug!(
renderer = renderer_id.0.as_str(),
"start_queue_playback_if_idle: skipping because renderer is already playing"
);
return Ok(()); return Ok(());
} }
// Check if queue has ANY items (not just upcoming items after current)
// This is important for newly attached playlists with current_index set
let has_items = self let has_items = self
.runtime .runtime
.queue_snapshot(renderer_id) .queue_full_snapshot(renderer_id)
.map(|items| !items.is_empty()) .map(|(items, _)| !items.is_empty())
.unwrap_or(false); .unwrap_or(false);
if !has_items { if !has_items {
debug!( debug!(