diff --git a/Cargo.lock b/Cargo.lock index 71a31c0e..3736cee7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -222,6 +222,17 @@ dependencies = [ "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]] name = "async-global-executor" version = "2.4.1" @@ -266,6 +277,17 @@ dependencies = [ "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]] name = "async-process" version = "2.5.0" @@ -3691,8 +3713,10 @@ dependencies = [ "quick-xml 0.38.4", "ratatui", "rust_cast", + "rustls", "serde", "serde_json", + "smol", "thiserror 2.0.17", "tokio", "tokio-stream", @@ -5033,6 +5057,23 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" 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]] name = "smol_str" version = "0.2.2" diff --git a/pmocontrol/Cargo.toml b/pmocontrol/Cargo.toml index 98100d54..68e7d90a 100644 --- a/pmocontrol/Cargo.toml +++ b/pmocontrol/Cargo.toml @@ -18,9 +18,11 @@ crossbeam-channel = "0.5" ratatui = { version = "0.26", default-features = false, features = ["crossterm"] } crossterm = "0.27" rust_cast = "0.19" +rustls = { version = "0.23", features = ["aws-lc-rs"] } mdns = "3.0" async-std = "1.12" futures-util = "0.3" +smol = "2.0" # pmoserver extension support (optional) pmoserver = { path = "../pmoserver", optional = true } diff --git a/pmocontrol/src/chromecast_discovery.rs b/pmocontrol/src/chromecast_discovery.rs index bc308020..bd67a255 100644 --- a/pmocontrol/src/chromecast_discovery.rs +++ b/pmocontrol/src/chromecast_discovery.rs @@ -70,14 +70,6 @@ pub fn process_mdns_response(response: mdns::Response) -> Option { 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 let addresses: Vec = response .records() @@ -89,7 +81,7 @@ pub fn process_mdns_response(response: mdns::Response) -> Option { .collect(); 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; } @@ -144,6 +136,25 @@ pub fn process_mdns_response(response: mdns::Response) -> Option { .unwrap_or_else(|| format!("chromecast-{}-{}", host, port)); 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::>() + .join("-") + .trim() + .to_string() + }); + debug!( "Discovered Chromecast: {} at {}:{} (UUID: {}, Model: {:?})", friendly_name, host, port, uuid, model diff --git a/pmocontrol/src/chromecast_renderer.rs b/pmocontrol/src/chromecast_renderer.rs index 1a98b4c2..9e3a6f56 100644 --- a/pmocontrol/src/chromecast_renderer.rs +++ b/pmocontrol/src/chromecast_renderer.rs @@ -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 //! transport and volume control traits, allowing Chromecast devices to be //! 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::time::Duration; +use std::sync::{Arc, Mutex, Once}; +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 crate::capabilities::{ @@ -18,86 +23,123 @@ use crate::capabilities::{ VolumeControl, }; use crate::chromecast_discovery::{extract_host_from_location, extract_port_from_location}; -use crate::model::{RendererInfo, RendererId, RendererProtocol}; -use crate::openhome_client::parse_track_metadata_from_didl; +use crate::model::{RendererId, RendererInfo, RendererProtocol}; + +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. 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, - - /// The media session ID obtained when loading media. - media_session_id: Option, - - /// The destination transport ID (usually "web-0"). - destination_id: Option, -} - -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. /// /// Uses the rust_cast library to communicate with Chromecast devices -/// via the Cast protocol (Protocol Buffers over TLS). -#[derive(Clone, Debug)] +/// via the Cast protocol. For play operations, a dedicated thread is +/// spawned to handle heartbeat responses from the device. +#[derive(Clone)] pub struct ChromecastRenderer { pub info: RendererInfo, host: String, port: u16, - session_state: Arc>, - timeout: Duration, + stop_signal: Arc>, + /// Handle to the active heartbeat thread, if any. + /// Wrapped in Arc to allow cloning and proper thread lifecycle management. + thread_handle: Arc>>>, +} + +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> { + // 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 { /// Creates a new ChromecastRenderer from RendererInfo. pub fn from_renderer_info(info: RendererInfo) -> Result { - // 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) .ok_or_else(|| anyhow!("Invalid Chromecast location: {}", info.location))?; let port = extract_port_from_location(&info.location) .unwrap_or(DEFAULT_CHROMECAST_PORT); - debug!( - "Creating ChromecastRenderer for {} at {}:{}", - info.friendly_name, host, port + let stop_signal = Arc::new(Mutex::new(false)); + let thread_handle = Arc::new(Mutex::new(None)); + + tracing::info!( + "ChromecastRenderer created for {} with host={} port={}", + info.friendly_name, + host, + port ); Ok(Self { info, host, port, - session_state: Arc::new(Mutex::new(ChromecastSessionState::new())), - timeout: DEFAULT_TIMEOUT, + stop_signal, + thread_handle, }) } - /// Returns the renderer ID. pub fn id(&self) -> &RendererId { &self.info.id @@ -117,182 +159,150 @@ impl ChromecastRenderer { pub fn info(&self) -> &RendererInfo { &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> { - 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> { - 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 { - // 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 { - 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 { - 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 { fn play_uri(&self, uri: &str, meta: &str) -> Result<()> { debug!("ChromecastRenderer: play_uri({})", uri); - // Ensure we have a session and get a connection - let device = self.ensure_session()?; + // Signal any existing play thread to stop + if let Ok(mut stop) = self.stop_signal.lock() { + *stop = true; + } - // Build media from DIDL metadata - let media = self.build_media_from_didl(uri, meta)?; + // Wait for the previous thread to finish (with timeout) + 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 - let state = self.session_state.lock() - .map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; + // Wait for thread to finish (it should see stop_signal and exit) + // Note: device.receive() may block, so thread might take time to notice stop_signal + let join_result = std::thread::spawn(move || handle.join()) + .join(); - let destination_id = state.destination_id.as_ref() - .ok_or_else(|| anyhow!("No destination ID available"))? - .clone(); + match join_result { + Ok(Ok(())) => { + 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() - .ok_or_else(|| anyhow!("No receiver session ID available"))? - .clone(); + // Reset stop signal + if let Ok(mut stop) = self.stop_signal.lock() { + *stop = false; + } - // Drop the lock before calling device methods - drop(state); + // Launch a new play thread + 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( - &destination_id, - &session_id, - &media, - ).map_err(|e| anyhow!("Failed to load media: {}", e))?; + let handle = std::thread::spawn(move || { + tracing::info!("Play thread starting for URI: {}", uri); - // Cache media session ID - let mut state = self.session_state.lock() - .map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; + let device = match connect_to_device(&host, port) { + Ok(d) => d, + Err(e) => { + tracing::error!("Failed to connect in play thread: {}", e); + return; + } + }; - if let Some(entry) = status.entries.first() { - state.media_session_id = Some(entry.media_session_id); - debug!("Media loaded with session ID: {}", entry.media_session_id); + // Launch DefaultMediaReceiver app + let app = match device.receiver.launch_app(&CastDeviceApp::DefaultMediaReceiver) { + 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(()) @@ -301,21 +311,28 @@ impl TransportControl for ChromecastRenderer { fn play(&self) -> Result<()> { debug!("ChromecastRenderer: play()"); - let device = self.connect()?; + let device = connect_to_device(&self.host, self.port)?; - let state = self.session_state.lock() - .map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; + // Get receiver status to find the active app + let status = device.receiver.get_status() + .map_err(|e| anyhow!("Failed to get receiver status: {}", e))?; - let destination_id = state.destination_id.as_ref() - .ok_or_else(|| anyhow!("No destination ID available"))? - .clone(); + let app = status.applications.first() + .ok_or_else(|| anyhow!("No active app found"))?; - let media_session_id = state.media_session_id - .ok_or_else(|| anyhow!("No media session ID available"))?; + // Connect to the app + 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))?; Ok(()) @@ -324,21 +341,24 @@ impl TransportControl for ChromecastRenderer { fn pause(&self) -> Result<()> { debug!("ChromecastRenderer: pause()"); - let device = self.connect()?; + let device = connect_to_device(&self.host, self.port)?; - let state = self.session_state.lock() - .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() - .ok_or_else(|| anyhow!("No destination ID available"))? - .clone(); + let app = status.applications.first() + .ok_or_else(|| anyhow!("No active app found"))?; - let media_session_id = state.media_session_id - .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); + 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))?; Ok(()) @@ -347,21 +367,34 @@ impl TransportControl for ChromecastRenderer { fn stop(&self) -> Result<()> { 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() - .map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; + // Note: We don't wait for the thread here as stop() should be quick. + // 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() - .ok_or_else(|| anyhow!("No destination ID available"))? - .clone(); + // Also send stop command to the device + let device = connect_to_device(&self.host, self.port)?; - let media_session_id = state.media_session_id - .ok_or_else(|| anyhow!("No media session ID available"))?; + let status = device.receiver.get_status() + .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))?; Ok(()) @@ -370,73 +403,45 @@ impl TransportControl for ChromecastRenderer { fn seek_rel_time(&self, hhmmss: &str) -> Result<()> { debug!("ChromecastRenderer: seek_rel_time({})", hhmmss); - let seconds = Self::parse_hhmmss_to_seconds(hhmmss)? as f32; - let device = self.connect()?; + // Parse HH:MM:SS to seconds + 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() - .map_err(|e| anyhow!("Failed to acquire session state lock: {}", e))?; + let hours: u32 = parts[0].parse() + .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() - .ok_or_else(|| anyhow!("No destination ID available"))? - .clone(); + let total_seconds = (hours * 3600 + minutes * 60 + seconds) as f32; - let media_session_id = state.media_session_id - .ok_or_else(|| anyhow!("No media session ID available"))?; + let device = connect_to_device(&self.host, self.port)?; - 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( - &destination_id, - media_session_id, - Some(seconds), - Some(rust_cast::channels::media::ResumeState::PlaybackStart), - ).map_err(|e| anyhow!("Failed to seek: {}", e))?; - - Ok(()) - } -} - -impl VolumeControl for ChromecastRenderer { - fn volume(&self) -> Result { - 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 { - 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) - device.receiver.set_volume(m) - .map_err(|e| anyhow!("Failed to set mute: {}", e))?; + app.transport_id.as_str(), + media_entry.media_session_id, + Some(total_seconds), + None, + ) + .map_err(|e| anyhow!("Failed to seek: {}", e))?; Ok(()) } @@ -444,56 +449,193 @@ impl VolumeControl for ChromecastRenderer { impl PlaybackStatus for ChromecastRenderer { fn playback_state(&self) -> Result { - let status = self.get_media_status()?; + let device = connect_to_device(&self.host, self.port)?; - if let Some(entry) = status.entries.first() { - use rust_cast::channels::media::PlayerState; + // Get receiver status to find the active app + let status = device.receiver.get_status() + .map_err(|e| anyhow!("Failed to get receiver status: {}", e))?; - let state = match entry.player_state { - PlayerState::Playing => PlaybackState::Playing, - PlayerState::Paused => PlaybackState::Paused, - PlayerState::Idle => PlaybackState::Stopped, - PlayerState::Buffering => PlaybackState::Transitioning, - }; + // If no app is running, return NoMedia + let app = match status.applications.first() { + Some(app) => app, + None => return Ok(PlaybackState::NoMedia), + }; - Ok(state) - } else { - Ok(PlaybackState::NoMedia) - } + // Connect to the app + device.connection.connect(app.transport_id.as_str()) + .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 { fn playback_position(&self) -> Result { - let status = self.get_media_status()?; + let device = connect_to_device(&self.host, self.port)?; - if let Some(entry) = status.entries.first() { - let rel_time = entry.current_time.map(|t| Self::format_seconds_to_hhmmss(t as f64)); + // Get receiver status to find the active app + let status = device.receiver.get_status() + .map_err(|e| anyhow!("Failed to get receiver status: {}", e))?; - let track_duration = entry.media.as_ref() - .and_then(|m| m.duration) - .map(|d| Self::format_seconds_to_hhmmss(d as f64)); + let app = status.applications.first() + .ok_or_else(|| anyhow!("No active app found"))?; - let track_uri = entry.media.as_ref() - .map(|m| m.content_id.clone()); + // Connect to the app + device.connection.connect(app.transport_id.as_str()) + .map_err(|e| anyhow!("Failed to connect to app: {}", e))?; - Ok(PlaybackPositionInfo { - track: None, - rel_time, - abs_time: None, - track_duration, - track_metadata: None, - track_uri, - }) - } else { - Ok(PlaybackPositionInfo { - track: None, - rel_time: None, - abs_time: None, - track_duration: None, - track_metadata: None, - track_uri: None, - }) - } + // 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))?; + + let media_entry = media_status.entries.first() + .ok_or_else(|| anyhow!("No media session found"))?; + + // Extract position information + let rel_time = media_entry.current_time + .map(|time| format_time_hhmmss(time as f64)); + + let track_duration = media_entry.media.as_ref() + .and_then(|m| m.duration) + .map(|dur| format_time_hhmmss(dur as f64)); + + 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 { + 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 { + 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(()) } } diff --git a/pmocontrol/src/control_point.rs b/pmocontrol/src/control_point.rs index 5fb7a75a..073ed397 100644 --- a/pmocontrol/src/control_point.rs +++ b/pmocontrol/src/control_point.rs @@ -104,6 +104,12 @@ pub struct ControlPoint { /// Key : RendererId /// Value : PlaylistBinding playlist_bindings: Arc>>, + /// 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>>, } impl ControlPoint { @@ -119,6 +125,7 @@ impl ControlPoint { runtime: Arc::clone(&runtime), })); let playlist_bindings = Arc::new(Mutex::new(HashMap::new())); + let renderer_cache = Arc::new(Mutex::new(HashMap::new())); // SsdpClient let client = SsdpClient::new()?; // pmoupnp::ssdp::SsdpClient @@ -178,8 +185,9 @@ impl ControlPoint { // Run async discovery in a blocking task async_std::task::block_on(async { - // Create mDNS discovery stream with 30 second query interval - match mdns::discover::all(SERVICE_NAME, Duration::from_secs(30)) { + // Create mDNS discovery stream with 15 second query interval + // (shorter interval for faster initial discovery) + match mdns::discover::all(SERVICE_NAME, Duration::from_secs(15)) { Ok(discovery) => { let stream = discovery.listen(); futures_util::pin_mut!(stream); @@ -223,21 +231,51 @@ impl ControlPoint { media_event_bus: media_event_bus.clone(), runtime: Arc::clone(&runtime), playlist_bindings: Arc::clone(&playlist_bindings), + renderer_cache: Arc::clone(&renderer_cache), }; thread::spawn(move || { let mut tick: u32 = 0; + loop { let infos = { let reg = runtime_cp.registry.read().unwrap(); reg.list_renderers() }; - let renderers = infos + + // Build a map of current renderer IDs for cleanup + let current_ids: HashSet = 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 = infos .into_iter() .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::>(); + .collect(); for renderer in renderers { let info = renderer.info(); @@ -252,7 +290,10 @@ impl ControlPoint { PlaylistBackend::PMOQueue }; 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); match backend { PlaylistBackend::OpenHome => { @@ -420,6 +461,7 @@ impl ControlPoint { media_event_bus: media_event_bus.clone(), runtime: Arc::clone(&runtime), playlist_bindings: Arc::clone(&playlist_bindings), + renderer_cache: Arc::clone(&renderer_cache), }; thread::Builder::new() @@ -574,6 +616,7 @@ impl ControlPoint { media_event_bus, runtime, playlist_bindings, + renderer_cache, }) } @@ -615,6 +658,30 @@ impl ControlPoint { 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 { + 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). pub fn list_music_renderers(&self) -> Vec { let infos = { @@ -622,9 +689,16 @@ impl ControlPoint { reg.list_renderers() }; + // Clean up cache - remove renderers that are no longer in the registry + { + let current_ids: HashSet = infos.iter().map(|i| i.id.clone()).collect(); + let mut cache = self.renderer_cache.lock().unwrap(); + cache.retain(|id, _| current_ids.contains(id)); + } + infos .into_iter() - .filter_map(|info| MusicRenderer::from_registry_info(info, &self.registry)) + .filter_map(|info| self.get_or_create_renderer(info)) .collect() } @@ -637,7 +711,7 @@ impl ControlPoint { infos .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. @@ -647,7 +721,7 @@ impl ControlPoint { 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. @@ -1316,14 +1390,32 @@ impl ControlPoint { fn start_queue_playback_if_idle(&self, renderer_id: &RendererId) -> anyhow::Result<()> { let snapshot = self.runtime.snapshot_for(renderer_id); 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(()); } + // 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 .runtime - .queue_snapshot(renderer_id) - .map(|items| !items.is_empty()) + .queue_full_snapshot(renderer_id) + .map(|(items, _)| !items.is_empty()) .unwrap_or(false); if !has_items { debug!(