diff --git a/pmoapp/webapp/src/components/unified/RendererTabContent.vue b/pmoapp/webapp/src/components/unified/RendererTabContent.vue index 334db22f..a5d4f2ac 100644 --- a/pmoapp/webapp/src/components/unified/RendererTabContent.vue +++ b/pmoapp/webapp/src/components/unified/RendererTabContent.vue @@ -56,6 +56,10 @@ async function handleQueueItemClick(item: QueueItem) { try { await api.seekQueueIndex(props.rendererId, item.index) console.log('[RendererTabContent] Jumped to queue index:', item.index, item.title) + + // Force un refetch immédiat pour synchroniser la cover affichée + // sans attendre l'événement SSE qui peut avoir un délai + await refresh(true) } catch (error) { console.error('[RendererTabContent] Error seeking to queue index:', error) uiStore.notifyError(`Erreur: ${error instanceof Error ? error.message : 'Impossible de sauter à cet item'}`) diff --git a/pmocontrol/examples/chromecast_info.rs b/pmocontrol/examples/chromecast_info.rs new file mode 100644 index 00000000..3f32982c --- /dev/null +++ b/pmocontrol/examples/chromecast_info.rs @@ -0,0 +1,120 @@ +//! Simple Chromecast info retrieval test +//! +//! This is the most minimal test - just connects and gets device status. +//! +//! Usage: +//! cargo run --example chromecast_info -- + +use std::env; +use std::sync::Once; + +use rust_cast::CastDevice; + +const DEFAULT_DESTINATION_ID: &str = "receiver-0"; +const DEFAULT_PORT: u16 = 8009; + +/// Ensures the Rustls CryptoProvider is initialized exactly once. +fn ensure_crypto_provider_initialized() { + static INIT: Once = Once::new(); + + INIT.call_once(|| { + let _ = rustls::crypto::CryptoProvider::install_default( + rustls::crypto::aws_lc_rs::default_provider() + ); + }); +} + +fn main() { + // Parse command line arguments + let args: Vec = env::args().collect(); + + if args.len() < 2 { + eprintln!("Usage: {} ", args[0]); + eprintln!("\nExample:"); + eprintln!(" {} 192.168.1.100", args[0]); + std::process::exit(1); + } + + let chromecast_ip = &args[1]; + + println!("═══════════════════════════════════════════════════════"); + println!(" Chromecast Info Test"); + println!("═══════════════════════════════════════════════════════"); + println!(); + + // Initialize crypto provider + ensure_crypto_provider_initialized(); + + // Connect to the device + println!("→ Connecting to {}:{}...", chromecast_ip, DEFAULT_PORT); + let cast_device = match CastDevice::connect_without_host_verification(chromecast_ip, DEFAULT_PORT) { + Ok(device) => { + println!(" ✓ Connected"); + device + } + Err(e) => { + eprintln!(" ✗ Failed: {}", e); + std::process::exit(1); + } + }; + + // Connect to receiver channel + println!(); + println!("→ Connecting to receiver channel..."); + if let Err(e) = cast_device.connection.connect(DEFAULT_DESTINATION_ID.to_string()) { + eprintln!(" ✗ Failed: {}", e); + std::process::exit(1); + } + println!(" ✓ Channel connected"); + + // Send initial ping + println!(); + println!("→ Sending initial ping..."); + if let Err(e) = cast_device.heartbeat.ping() { + eprintln!(" ✗ Failed: {}", e); + std::process::exit(1); + } + println!(" ✓ Ping sent"); + + // Get receiver status + println!(); + println!("→ Getting receiver status..."); + match cast_device.receiver.get_status() { + Ok(status) => { + println!(" ✓ Status retrieved\n"); + + // Volume + if let Some(level) = status.volume.level { + println!(" Volume: {:.0}%", level * 100.0); + } + if let Some(muted) = status.volume.muted { + println!(" Muted: {}", muted); + } + + // Applications + println!("\n Running applications: {}", status.applications.len()); + for (i, app) in status.applications.iter().enumerate() { + println!("\n App #{}:", i + 1); + println!(" Display Name: {}", app.display_name); + println!(" App ID: {}", app.app_id); + println!(" Session ID: {}", app.session_id); + println!(" Transport ID: {}", app.transport_id); + println!(" Status: {}", app.status_text); + println!(" Namespaces: {}", app.namespaces.len()); + } + + if status.applications.is_empty() { + println!(" (no apps currently running)"); + } + } + Err(e) => { + eprintln!(" ✗ Failed: {}", e); + std::process::exit(1); + } + } + + println!(); + println!("═══════════════════════════════════════════════════════"); + println!(" Test completed successfully!"); + println!("═══════════════════════════════════════════════════════"); +} diff --git a/pmocontrol/examples/chromecast_playback_test.rs b/pmocontrol/examples/chromecast_playback_test.rs new file mode 100644 index 00000000..528a872f --- /dev/null +++ b/pmocontrol/examples/chromecast_playback_test.rs @@ -0,0 +1,284 @@ +//! Simple Chromecast playback test +//! +//! This example tests basic Chromecast functionality by: +//! 1. Connecting to a Chromecast device +//! 2. Launching the DefaultMediaReceiver app +//! 3. Loading and playing a test media URL +//! 4. Maintaining the heartbeat loop +//! +//! Usage: +//! cargo run --example chromecast_playback_test -- [media_url] +//! +//! Example: +//! cargo run --example chromecast_playback_test -- 192.168.1.100 + +use std::env; +use std::sync::Once; + +use rust_cast::{ + CastDevice, ChannelMessage, + channels::{ + heartbeat::HeartbeatResponse, + media::{Media, StreamType}, + receiver::CastDeviceApp, + }, +}; + +const DEFAULT_DESTINATION_ID: &str = "receiver-0"; +const DEFAULT_PORT: u16 = 8009; + +// Test media URLs (public domain audio files) +const TEST_MEDIA_URL: &str = "https://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3"; + +/// Ensures the Rustls CryptoProvider is initialized exactly once. +fn ensure_crypto_provider_initialized() { + static INIT: Once = Once::new(); + + INIT.call_once(|| { + let _ = rustls::crypto::CryptoProvider::install_default( + rustls::crypto::aws_lc_rs::default_provider() + ); + println!("✓ Rustls CryptoProvider initialized"); + }); +} + +/// Detects content type from URL path +fn detect_content_type(url: &str) -> String { + // Detect from URL path - check if path contains /flac/, /mp3/, etc. + if url.contains("/flac/") || url.contains(".flac") { + println!(" ✓ Detected FLAC from URL path"); + return "audio/flac".to_string(); + } + if url.contains("/mp3/") || url.contains(".mp3") { + println!(" ✓ Detected MP3 from URL path"); + return "audio/mpeg".to_string(); + } + if url.contains(".m4a") || url.contains(".mp4") || url.contains(".aac") { + println!(" ✓ Detected AAC/M4A from URL path"); + return "audio/mp4".to_string(); + } + if url.contains("/ogg/") || url.contains(".ogg") { + println!(" ✓ Detected OGG from URL path"); + return "audio/ogg".to_string(); + } + if url.contains(".opus") { + println!(" ✓ Detected Opus from URL path"); + return "audio/opus".to_string(); + } + if url.contains(".wav") { + println!(" ✓ Detected WAV from URL path"); + return "audio/wav".to_string(); + } + + // Fallback + println!(" ⚠ Could not detect type, using audio/mpeg as fallback"); + "audio/mpeg".to_string() +} + +fn main() { + // Setup logging + tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .init(); + + // Parse command line arguments + let args: Vec = env::args().collect(); + + if args.len() < 2 { + eprintln!("Usage: {} [media_url]", args[0]); + eprintln!("\nExample:"); + eprintln!(" {} 192.168.1.100", args[0]); + eprintln!("\nIf no media URL is provided, will use: {}", TEST_MEDIA_URL); + std::process::exit(1); + } + + let chromecast_ip = &args[1]; + let media_url = if args.len() > 2 { + &args[2] + } else { + TEST_MEDIA_URL + }; + + println!("╔════════════════════════════════════════════════════════════╗"); + println!("║ Chromecast Playback Test (rust_cast) ║"); + println!("╚════════════════════════════════════════════════════════════╝"); + println!(); + println!("Target: {}:{}", chromecast_ip, DEFAULT_PORT); + println!("Media URL: {}", media_url); + println!(); + println!("Detecting media content type..."); + let media_type = detect_content_type(media_url); + println!(); + + // Initialize crypto provider + ensure_crypto_provider_initialized(); + + // Step 1: Connect to the device + println!("──────────────────────────────────────────────────────────"); + println!("STEP 1: Connecting to Chromecast..."); + let cast_device = match CastDevice::connect_without_host_verification(chromecast_ip, DEFAULT_PORT) { + Ok(device) => { + println!("✓ Connected to Chromecast"); + device + } + Err(e) => { + eprintln!("✗ Failed to connect: {}", e); + std::process::exit(1); + } + }; + + // Step 2: Connect to the default receiver channel + println!(); + println!("STEP 2: Connecting to receiver channel..."); + if let Err(e) = cast_device.connection.connect(DEFAULT_DESTINATION_ID.to_string()) { + eprintln!("✗ Failed to connect channel: {}", e); + std::process::exit(1); + } + println!("✓ Channel connected"); + + // Step 3: Send initial ping (CRITICAL per rust_caster.rs) + println!(); + println!("STEP 3: Sending initial heartbeat ping..."); + if let Err(e) = cast_device.heartbeat.ping() { + eprintln!("✗ Failed to send initial ping: {}", e); + std::process::exit(1); + } + println!("✓ Initial ping sent"); + + // Step 4: Get receiver status + println!(); + println!("STEP 4: Getting receiver status..."); + let status = match cast_device.receiver.get_status() { + Ok(status) => { + println!("✓ Receiver status obtained"); + println!(" - Volume: {:.0}%", status.volume.level.unwrap_or(0.5) * 100.0); + println!(" - Muted: {}", status.volume.muted.unwrap_or(false)); + println!(" - Running apps: {}", status.applications.len()); + status + } + Err(e) => { + eprintln!("✗ Failed to get status: {}", e); + std::process::exit(1); + } + }; + + // Step 5: Launch DefaultMediaReceiver + println!(); + println!("STEP 5: Launching DefaultMediaReceiver app..."); + let app = match cast_device.receiver.launch_app(&CastDeviceApp::DefaultMediaReceiver) { + Ok(app) => { + println!("✓ App launched successfully"); + println!(" - App ID: {}", app.app_id); + println!(" - Display Name: {}", app.display_name); + println!(" - Session ID: {}", app.session_id); + println!(" - Transport ID: {}", app.transport_id); + app + } + Err(e) => { + eprintln!("✗ Failed to launch app: {}", e); + std::process::exit(1); + } + }; + + // Step 6: Connect to the app's transport + println!(); + println!("STEP 6: Connecting to app transport..."); + if let Err(e) = cast_device.connection.connect(app.transport_id.as_str()) { + eprintln!("✗ Failed to connect to app transport: {}", e); + std::process::exit(1); + } + println!("✓ Connected to app transport: {}", app.transport_id); + + // Step 7: Load the media + println!(); + println!("STEP 7: Loading media..."); + println!(" Content-Type: {}", media_type); + let media = Media { + content_id: media_url.to_string(), + content_type: media_type, + stream_type: StreamType::Buffered, + duration: None, + metadata: None, + }; + + match cast_device.media.load( + app.transport_id.as_str(), + app.session_id.as_str(), + &media, + ) { + Ok(status) => { + println!("✓ Media loaded successfully!"); + println!(" - Media status entries: {}", status.entries.len()); + + if let Some(entry) = status.entries.first() { + println!(" - Player state: {:?}", entry.player_state); + println!(" - Media session ID: {}", entry.media_session_id); + + if let Some(ref media) = entry.media { + println!(" - Content ID: {}", media.content_id); + println!(" - Stream type: {:?}", media.stream_type); + } + } + } + Err(e) => { + eprintln!("✗ Failed to load media: {}", e); + std::process::exit(1); + } + } + + // Step 8: Enter heartbeat loop + println!(); + println!("──────────────────────────────────────────────────────────"); + println!("STEP 8: Entering heartbeat loop (Ctrl+C to exit)..."); + println!("──────────────────────────────────────────────────────────"); + println!(); + + let mut heartbeat_count = 0; + let mut media_message_count = 0; + + loop { + match cast_device.receive() { + Ok(ChannelMessage::Heartbeat(response)) => { + if let HeartbeatResponse::Ping = response { + heartbeat_count += 1; + println!("[Heartbeat #{:3}] Received Ping, sending Pong...", heartbeat_count); + + if let Err(e) = cast_device.heartbeat.pong() { + eprintln!("✗ Failed to send pong: {}", e); + break; + } + } else { + println!("[Heartbeat] {:?}", response); + } + } + + Ok(ChannelMessage::Media(response)) => { + media_message_count += 1; + println!("[Media #{:3}] {:?}", media_message_count, response); + } + + Ok(ChannelMessage::Receiver(response)) => { + println!("[Receiver] {:?}", response); + } + + Ok(ChannelMessage::Connection(response)) => { + println!("[Connection] {:?}", response); + } + + Ok(ChannelMessage::Raw(response)) => { + println!("[Raw] Unsupported message type: {:?}", response); + } + + Err(e) => { + eprintln!("✗ Error receiving message: {}", e); + break; + } + } + } + + println!(); + println!("──────────────────────────────────────────────────────────"); + println!("Test completed."); + println!("Total heartbeats: {}", heartbeat_count); + println!("Total media messages: {}", media_message_count); +} diff --git a/pmocontrol/rust-cast-async-analysis.md b/pmocontrol/rust-cast-async-analysis.md new file mode 100644 index 00000000..b71937da --- /dev/null +++ b/pmocontrol/rust-cast-async-analysis.md @@ -0,0 +1,409 @@ +# Analyse : Rendre rust-cast asynchrone vs autres options + +**Date :** 2025-12-27 +**Question :** Est-il plus simple d'intégrer rust-cast et le modifier pour le rendre asynchrone ? + +--- + +## 1. Analyse de la codebase rust-cast + +### Taille et complexité + +```bash +Total : ~5800 lignes de code Rust +Structure modulaire : +├── src/lib.rs (~570 lignes) +├── src/message_manager.rs (~300 lignes) +├── src/channels/ +│ ├── media.rs (~800 lignes) +│ ├── receiver.rs (~400 lignes) +│ ├── heartbeat.rs (~100 lignes) +│ └── connection.rs (~100 lignes) +├── src/cast/ +│ ├── cast_channel.rs (généré par protobuf) +│ └── proxies.rs (~500 lignes) +└── src/errors.rs, utils.rs (~200 lignes) +``` + +**Conclusion :** Codebase de taille **modeste et bien structurée**. + +--- + +## 2. Points bloquants identifiés + +### 2.1 I/O synchrone bloquant + +Tous les I/O passent par `MessageManager` où `S: Read + Write` : + +```rust +// message_manager.rs:246-253 +fn read(&self) -> Result { + let mut buffer: [u8; 4] = [0; 4]; + let reader = &mut *self.stream.borrow_mut(); + + reader.read_exact(&mut buffer)?; // ❌ BLOQUANT + let length = utils::read_u32_from_buffer(&buffer)?; + + let mut buffer: Vec = Vec::with_capacity(length as usize); + let mut limited_reader = reader.take(u64::from(length)); + limited_reader.read_to_end(&mut buffer)?; // ❌ BLOQUANT + ... +} +``` + +```rust +// message_manager.rs:138-141 +pub fn send(&self, message: CastMessage) -> Result<(), Error> { + ... + let writer = &mut *self.stream.borrow_mut(); + writer.write_all(&message_length_buffer)?; // ❌ BLOQUANT + writer.write_all(&message_content_buffer)?; // ❌ BLOQUANT + ... +} +``` + +### 2.2 Connexion TLS + +```rust +// lib.rs:125 +let stream = StreamOwned::new( + conn, + TcpStream::connect((host.as_ref(), port))? // ❌ BLOQUANT +); +``` + +**Total : 5 points bloquants critiques** (connect, read_exact, read_to_end, 2x write_all) + +--- + +## 3. Effort pour rendre rust-cast asynchrone + +### 3.1 Modifications requises + +#### A. Remplacer la stack réseau + +**Avant (sync) :** +```rust +use std::net::TcpStream; +use rustls::{ClientConnection, StreamOwned}; + +type TlsStream = StreamOwned; +``` + +**Après (async) :** +```rust +use async_io::Async; +use std::net::TcpStream; +use async_rustls::{TlsConnector, client::TlsStream}; + +// OU avec tokio : +use tokio::net::TcpStream; +use tokio_rustls::{TlsConnector, client::TlsStream}; +``` + +⚠️ **PROBLÈME :** `rustls::StreamOwned` n'existe pas en version async native. Il faut utiliser : +- `async-rustls` (pour async-std/smol) +- `tokio-rustls` (pour tokio) + +Ces crates ont une **API différente** de `rustls::StreamOwned`. + +#### B. Modifier `MessageManager` + +```diff +- pub struct MessageManager where S: Write + Read { ++ pub struct MessageManager where S: AsyncWrite + AsyncRead + Unpin { + +- pub fn send(&self, message: CastMessage) -> Result<(), Error> { ++ pub async fn send(&self, message: CastMessage) -> Result<(), Error> { + ... +- writer.write_all(&message_length_buffer)?; ++ writer.write_all(&message_length_buffer).await?; + } + +- pub fn receive(&self) -> Result { ++ pub async fn receive(&self) -> Result { + ... + } + +- fn read(&self) -> Result { ++ async fn read(&self) -> Result { +- reader.read_exact(&mut buffer)?; ++ reader.read_exact(&mut buffer).await?; +- limited_reader.read_to_end(&mut buffer)?; ++ limited_reader.read_to_end(&mut buffer).await?; + } +} +``` + +#### C. Propager `async` dans tous les channels + +**Avant :** +```rust +// channels/media.rs +impl<'a, S> MediaChannel<'a, S> where S: Write + Read { + pub fn play(&self, ...) -> Result<(), Error> { + self.message_manager.send(...)?; + self.message_manager.receive_find_map(...) + } +} +``` + +**Après :** +```rust +impl<'a, S> MediaChannel<'a, S> where S: AsyncWrite + AsyncRead + Unpin { + pub async fn play(&self, ...) -> Result<(), Error> { + self.message_manager.send(...).await?; + self.message_manager.receive_find_map(...).await + } +} +``` + +**Impact :** TOUS les channels (media, receiver, heartbeat, connection) deviennent `async`. + +#### D. Modifier `CastDevice` + +```diff +impl<'a> CastDevice<'a> { +- pub fn connect(host: S, port: u16) -> Result, Error> ++ pub async fn connect(host: S, port: u16) -> Result, Error> + { + ... +- let stream = TcpStream::connect((host.as_ref(), port))?; ++ let stream = TcpStream::connect((host.as_ref(), port)).await?; + ... + } + +- pub fn receive(&self) -> Result { ++ pub async fn receive(&self) -> Result { +- let cast_message = self.message_manager.receive()?; ++ let cast_message = self.message_manager.receive().await?; + ... + } +} +``` + +### 3.2 Estimation de l'effort + +| Tâche | Fichiers touchés | Complexité | Temps estimé | +|-------|------------------|------------|--------------| +| Choisir stack async (smol vs tokio) | - | Faible | 1h | +| Migrer vers async-rustls/tokio-rustls | lib.rs | **MOYENNE** | 4-6h | +| Rendre MessageManager async | message_manager.rs | **MOYENNE** | 4-6h | +| Rendre tous les channels async | 4 fichiers | **MOYENNE-ÉLEVÉE** | 8-12h | +| Mettre à jour CastDevice | lib.rs | Moyenne | 2-4h | +| Tests et debug | Tous | **ÉLEVÉE** | 8-16h | +| **TOTAL** | **~10 fichiers** | **ÉLEVÉE** | **27-45 heures** | + +⚠️ **RISQUES :** +- API `async-rustls` différente de `rustls::StreamOwned` → peut nécessiter refactoring profond +- Gestion des locks async (`Mutex` → `async_lock::Mutex` ou `tokio::sync::Mutex`) +- Bugs subtils liés à la concurrence async +- Tests nécessaires pour valider la stabilité + +--- + +## 4. Comparaison des 4 options + +### Option 1 : ✅ **Rester avec rust-cast sync et corriger le TLS** + +**Effort :** FAIBLE (2-8 heures) + +**Actions :** +- Investiguer les erreurs TLS prématurées +- Ajouter retry logic sur les reconnexions +- Améliorer la gestion d'erreur dans [chromecast_renderer.rs](pmocontrol/src/chromecast_renderer.rs:86-104) +- Peut-être ajuster les timeouts de lecture + +**Avantages :** +- ✅ Garde l'API sync compatible avec PMOMusic +- ✅ Risque minimal +- ✅ Solution rapide + +**Inconvénients :** +- ⚠️ Ne résout peut-être pas tous les problèmes TLS + +--- + +### Option 2 : 🔧 **Forker rust-cast et moderniser le TLS (reste sync)** + +**Effort :** MOYEN (8-16 heures) + +**Actions :** +- Forker rust-cast sur GitHub/GitLab +- Améliorer la gestion TLS (retry, reconnexion automatique) +- Ajouter logs détaillés +- Corriger les bugs TLS identifiés +- Maintenir un fork privé + +**Avantages :** +- ✅ Garde l'API sync +- ✅ Contrôle total sur les correctifs +- ✅ Peut merger les améliorations de upstream + +**Inconvénients :** +- ⚠️ Maintenance du fork à long terme +- ⚠️ Doit suivre les mises à jour de rustls + +--- + +### Option 3 : 🔄 **Rendre rust-cast asynchrone** + +**Effort :** ÉLEVÉ (27-45 heures) + +**Actions :** +- Migrer vers async-rustls ou tokio-rustls +- Rendre tout le code async (MessageManager, channels, CastDevice) +- Adapter PMOMusic pour wrapper les appels async + +**Avantages :** +- ✅ Architecture moderne +- ✅ Potentiellement meilleure performance pour gérer plusieurs devices +- ✅ Résout probablement les problèmes TLS via stack moderne + +**Inconvénients :** +- ❌ Effort très élevé +- ❌ Risque de bugs subtils +- ❌ PMOMusic doit wrapper tous les appels avec `smol::block_on()` +- ❌ Overhead de conversion sync→async→sync + +**⚠️ PARADOXE :** Rendre rust-cast async pour ensuite le wrapper en sync dans PMOMusic = **surcharge inutile** + +--- + +### Option 4 : ❌ **Migrer vers cast-sender (déjà async)** + +**Effort :** TRÈS ÉLEVÉ (40-80 heures) + +**Problèmes critiques :** +- ❌ API incomplète (pas de get_status, pas de seek) +- ❌ Nécessite architecture stateful complexe +- ❌ Documentation insuffisante (23%) + +**Voir :** [cast-sender-evaluation.md](cast-sender-evaluation.md) + +--- + +## 5. Analyse détaillée : Async est-il vraiment utile ? + +### 5.1 Cas d'usage PMOMusic + +**Architecture actuelle :** +- 1 thread par Chromecast actif (pour le heartbeat) +- Opérations de contrôle (play, pause, volume) : sporadiques +- Pas de gestion massive de connexions simultanées + +**Bénéfice de async :** +- ❌ **FAIBLE** : PMOMusic n'a pas besoin de gérer 100+ connexions simultanées +- ❌ **OVERHEAD** : Wrapping sync→async→sync ajoute de la complexité + +### 5.2 Vraie cause des problèmes TLS ? + +Les problèmes de "fermeture TLS prématurée" sont probablement dus à : +- Timeout réseau trop court +- Gestion d'erreur insuffisante lors des reconnexions +- Bugs spécifiques de certaines versions de rustls + +**Async ne résout PAS directement ces problèmes !** + +--- + +## 6. Recommandation finale + +### 🏆 **Option recommandée : Option 1 (Corriger rust-cast sync)** + +**Raisons :** + +1. **Effort minimal** : 2-8 heures vs 27-45h pour async +2. **Risque minimal** : Garde l'architecture validée +3. **Compatibilité** : Pas de changement dans PMOMusic +4. **Pragmatique** : Résout le problème réel (TLS) sans over-engineering + +**Plan d'action concret :** + +```rust +// Améliorer la fonction connect_to_device +fn connect_to_device(host: &str, port: u16) -> Result { + const MAX_RETRIES: u32 = 3; + const RETRY_DELAY_MS: u64 = 1000; + + for attempt in 1..=MAX_RETRIES { + match try_connect(host, port) { + Ok(device) => return Ok(device), + Err(e) if attempt < MAX_RETRIES => { + tracing::warn!( + "Connection attempt {} failed: {}. Retrying in {}ms...", + attempt, e, RETRY_DELAY_MS + ); + std::thread::sleep(Duration::from_millis(RETRY_DELAY_MS)); + } + Err(e) => return Err(e), + } + } + unreachable!() +} + +// Ajouter timeout configurable pour les read operations +// Ajouter meilleure gestion d'erreur dans le heartbeat loop +``` + +--- + +### 🥈 **Alternative : Option 2 (Fork rust-cast)** + +Si l'Option 1 ne suffit pas après investigation, forker permet : +- Corrections TLS plus profondes +- Ajout de fonctionnalités manquantes +- Contrôle total + +**Pas besoin de rendre async !** + +--- + +### 🚫 **Options déconseillées :** + +- ❌ **Option 3** (Async rust-cast) : Effort 5-10x supérieur pour bénéfice marginal +- ❌ **Option 4** (cast-sender) : API incomplète, effort encore plus élevé + +--- + +## 7. Conclusion + +**NON, rendre rust-cast asynchrone n'est PAS plus simple.** + +**Comparaison des efforts :** + +| Option | Effort (heures) | Complexité | Risque | +|--------|----------------|------------|--------| +| 1. Corriger rust-cast sync | 2-8 | Faible | Minimal | +| 2. Forker rust-cast | 8-16 | Moyenne | Faible | +| 3. **Async rust-cast** | **27-45** | **Élevée** | **Élevé** | +| 4. Migrer cast-sender | 40-80 | Très élevée | Très élevé | + +**Le ratio effort/bénéfice de l'option async est défavorable :** +- **5-10x plus d'effort** que corriger le code sync +- **Bénéfice minimal** pour l'architecture actuelle de PMOMusic +- **Risques élevés** de bugs de concurrence async + +**Recommandation :** Commencer par l'**Option 1**, investiguer les vrais problèmes TLS, et envisager l'**Option 2** (fork) uniquement si nécessaire. Éviter absolument l'**Option 3** (async) sauf changement radical d'architecture de PMOMusic. + +--- + +## Annexe : Si vous vouliez quand même faire async... + +### Stack recommandée + +**Pour PMOMusic (déjà avec smol) :** +```toml +[dependencies] +async-io = "2.3" +async-rustls = "0.4" +futures-lite = "2.1" +``` + +**Points d'attention :** +- Remplacer tous les `Mutex` par `async_lock::Mutex` +- Gérer correctement le `Unpin` trait pour les streams +- Tester intensivement la gestion des erreurs async +- Prévoir 2-3 semaines de développement + tests + +**Mais encore une fois : le jeu n'en vaut pas la chandelle !** diff --git a/pmocontrol/src/chromecast_renderer.rs b/pmocontrol/src/chromecast_renderer.rs index 9e3a6f56..7f7b5d2b 100644 --- a/pmocontrol/src/chromecast_renderer.rs +++ b/pmocontrol/src/chromecast_renderer.rs @@ -94,6 +94,12 @@ fn connect_to_device<'a>(host: &'a str, port: u16) -> Result> { .connect(DEFAULT_DESTINATION_ID.to_string()) .map_err(|e| anyhow!("Failed to connect channel: {}", e))?; + // Send initial gre to establish heartbeat communication + // This is critical per rust_caster.rs example + device.heartbeat + .ping() + .map_err(|e| anyhow!("Failed to send initial heartbeat ping: {}", e))?; + Ok(device) } diff --git a/pmocontrol/src/control_point/openhome_queue.rs b/pmocontrol/src/control_point/openhome_queue.rs index 415f31ed..0c9dcb33 100644 --- a/pmocontrol/src/control_point/openhome_queue.rs +++ b/pmocontrol/src/control_point/openhome_queue.rs @@ -58,56 +58,15 @@ impl OpenHomeQueue { track_ids.push(entry.id); } - // Try multiple methods to determine the currently playing track, from most to least reliable: - // 1. Info.Id() - Direct ID query (fastest, but fails if track no longer in playlist) - // 2. Info.Track() - Returns URI, which we can search for (works even if track removed) - // 3. None - No current track can be determined - let current_id = self.playlist.id()?; - // { - // // Try Info.Id() first - // if let Ok(id) = client.id() { - // debug!( - // renderer = self.renderer_id.0.as_str(), - // track_id = id, - // "Detected current track via Info.Id()" - // ); - // return Some(id); - // } + // Get the currently playing track ID from the renderer (may be None if no track is playing) + let current_id = self.playlist.id().ok(); - // // If Id() fails, try Track() to get the URI and search for it - // if let Ok(track_info) = client.track() { - // debug!( - // renderer = self.renderer_id.0.as_str(), - // track_uri = track_info.uri.as_str(), - // "Info.Id() failed, searching for current track by URI from Info.Track()" - // ); - // return entries - // .iter() - // .find(|entry| entry.uri == track_info.uri) - // .map(|entry| { - // debug!( - // renderer = self.renderer_id.0.as_str(), - // found_id = entry.id, - // found_uri = entry.uri.as_str(), - // "Found current track ID by matching URI" - // ); - // entry.id - // }); - // } - - // debug!( - // renderer = self.renderer_id.0.as_str(), - // "Both Info.Id() and Info.Track() failed, cannot determine current track" - // ); - // None - // }); - - // let current_index = current_id - // .and_then(|id| track_ids.iter().position(|entry_id| *entry_id == id)); + // Find the index of the current track in the playlist + let current_index = current_id.and_then(|id| track_ids.iter().position(|entry_id| *entry_id == id)); self.items = items; self.track_ids = track_ids; - self.current_index = Some(current_id as usize); + self.current_index = current_index; Ok(()) } @@ -339,34 +298,10 @@ impl OpenHomeQueue { playing_idx: usize, playing_id: u32, ) -> Result<()> { - // Re-read the current playlist state to get fresh IDs - // This minimizes race conditions where IDs become invalid between our last refresh - // and now (due to UPnP events from the server) - let current_entries = self.playlist.read_all_tracks()?; - let current_ids: Vec = current_entries.iter().map(|e| e.id).collect(); - - debug!( - renderer = self.renderer_id.0.as_str(), - fresh_id_count = current_ids.len(), - cached_id_count = self.track_ids.len(), - "Re-read playlist before deletions to avoid stale ID errors" - ); - - // Find the playing track in the fresh list - let fresh_playing_idx = current_ids.iter().position(|&id| id == playing_id); - - if fresh_playing_idx.is_none() { - debug!( - renderer = self.renderer_id.0.as_str(), - playing_id, - "Playing track not found in fresh playlist - renderer state may have changed, aborting modification" - ); - // The playing track is gone - don't try to manipulate the playlist - return Ok(()); - } - - // Delete everything except the currently playing item (using fresh IDs) - for &track_id in current_ids.iter().rev() { + // Delete everything except the currently playing item + // Using delete_id_if_exists() to handle cases where another control point + // may have already modified the playlist + for &track_id in self.track_ids.iter().rev() { if track_id != playing_id { self.playlist.delete_id_if_exists(track_id)?; } @@ -409,53 +344,15 @@ impl OpenHomeQueue { pivot_idx_new: usize, pivot_id: u32, ) -> Result<()> { - debug!( - renderer = self.renderer_id.0.as_str(), - pivot_id, - pivot_idx_new, - new_playlist_len = new_items.len(), - "Starting replace_queue_with_pivot - will re-read from OpenHome" - ); + // Find the pivot index in our current state + let pivot_idx = self.track_ids.iter().position(|&id| id == pivot_id) + .ok_or_else(|| anyhow!("Pivot track ID {} not found in playlist", pivot_id))?; - // Re-read the current playlist state from OpenHome (the ONLY source of truth) - // This is CRITICAL to avoid deleting IDs that no longer exist, which can - // put the renderer (upmpdcli) into a degraded state where Info.TransportState() - // starts returning HTTP 500 errors. - let current_entries = self.playlist.read_all_tracks()?; - - // Convert entries to PlaybackItems - this is the REAL current state - let mut fresh_items = Vec::with_capacity(current_entries.len()); - let mut fresh_ids = Vec::with_capacity(current_entries.len()); - for entry in ¤t_entries { - fresh_items.push(self.playback_item_from_entry(entry)); - fresh_ids.push(entry.id); - } - - debug!( - renderer = self.renderer_id.0.as_str(), - fresh_count = fresh_items.len(), - "Re-read playlist from OpenHome (source of truth)" - ); - - // Find the pivot in the fresh list - let fresh_pivot_idx = fresh_ids.iter().position(|&id| id == pivot_id); - - if fresh_pivot_idx.is_none() { - debug!( - renderer = self.renderer_id.0.as_str(), - pivot_id, - "Pivot track not found in fresh playlist - renderer state changed, aborting" - ); - return Ok(()); - } - - let fresh_pivot_idx = fresh_pivot_idx.unwrap(); - - // Split fresh data at the pivot - use ONLY fresh data, ignore cache - let old_before: Vec = fresh_items[..fresh_pivot_idx].to_vec(); - let old_after: Vec = fresh_items[fresh_pivot_idx + 1..].to_vec(); - let old_ids_before: Vec = fresh_ids[..fresh_pivot_idx].to_vec(); - let old_ids_after: Vec = fresh_ids[fresh_pivot_idx + 1..].to_vec(); + // Split current data at the pivot + let old_before: Vec = self.items[..pivot_idx].to_vec(); + let old_after: Vec = self.items[pivot_idx + 1..].to_vec(); + let old_ids_before: Vec = self.track_ids[..pivot_idx].to_vec(); + let old_ids_after: Vec = self.track_ids[pivot_idx + 1..].to_vec(); let new_before = &new_items[..pivot_idx_new]; let new_after = &new_items[pivot_idx_new + 1..]; @@ -855,43 +752,17 @@ impl QueueBackend for OpenHomeQueue { // (e.g., manual edits from another control point) would keep the stale items. self.refresh_from_openhome()?; - // Try to get the currently playing track ID from the renderer. - // Note: Some OpenHome renderers (like upmpdcli) don't reliably support Info.Id(), - // so we fall back to using our internal current_index pointer. - let currently_playing_id_from_renderer = self - .playlist.id().ok(); - - // Find the currently playing item in our local state. - // Priority: 1) Renderer-reported ID, 2) Our internal current_index - let playing_info = if let Some(id) = currently_playing_id_from_renderer { - // CASE: Renderer explicitly reported the playing track ID + // Get the currently playing track ID from the renderer + let playing_info = self.playlist.id().ok().and_then(|id| { self.track_ids .iter() .position(|&tid| tid == id) .map(|idx| (idx, id, self.items[idx].uri.clone())) - } else if let Some(idx) = self.current_index { - // CASE: Use our internal pointer (fallback for renderers without Info.Id() support) - if idx < self.track_ids.len() && idx < self.items.len() { - let id = self.track_ids[idx]; - let uri = self.items[idx].uri.clone(); - debug!( - renderer = self.renderer_id.0.as_str(), - current_index = idx, - track_id = id, - "Using internal current_index as fallback (renderer didn't report playing ID)" - ); - Some((idx, id, uri)) - } else { - None - } - } else { - None - }; + }); debug!( renderer = self.renderer_id.0.as_str(), actual_items = self.items.len(), - currently_playing_id_from_renderer = ?currently_playing_id_from_renderer, playing_info_detected = playing_info.is_some(), "OpenHome playlist state refreshed before replace_queue" ); diff --git a/pmocontrol/src/openhome_client.rs b/pmocontrol/src/openhome_client.rs index b866c6fb..5f06f3b7 100644 --- a/pmocontrol/src/openhome_client.rs +++ b/pmocontrol/src/openhome_client.rs @@ -146,10 +146,10 @@ impl OhPlaylistClient { let envelope = ensure_success("Id", &call_result)?; let response = find_child_with_suffix(&envelope.body.content, "IdResponse") .ok_or_else(|| anyhow!("Missing IdResponse element in SOAP body"))?; - let id_text = extract_child_text(response, "Id")?; + let id_text = extract_child_text(response, "Value")?; let id = id_text .parse::() - .map_err(|_| anyhow!("Invalid Info.Id value: {}", id_text))?; + .map_err(|_| anyhow!("Invalid Playlist.Id value: {}", id_text))?; Ok(id) } diff --git a/pmocontrol/src/openhome_renderer.rs b/pmocontrol/src/openhome_renderer.rs index a78a7f3e..32fe20b1 100644 --- a/pmocontrol/src/openhome_renderer.rs +++ b/pmocontrol/src/openhome_renderer.rs @@ -118,63 +118,8 @@ impl OpenHomeRenderer { let playlist = self.playlist_client_for("snapshot_openhome_playlist")?; let entries = playlist.read_all_tracks()?; - // Essayer d'obtenir current_id depuis Info.Id() - let mut current_id = self - .playlist - .as_ref() - .and_then(|client| { - match client.id() { - Ok(id) => { - debug!( - renderer = self.info.id.0.as_str(), - current_id = id, - "OpenHome Info service returned current_id" - ); - Some(id) - } - Err(err) => { - debug!( - renderer = self.info.id.0.as_str(), - error = %err, - "OpenHome Info.Id() failed, will try Info.Track()" - ); - None - } - } - }); - - // Fallback: Si Info.Id() échoue, essayer Info.Track() et matcher l'URI - if current_id.is_none() { - if let Some(client) = self.info_client.as_ref() { - match client.track() { - Ok(track_info) => { - debug!( - renderer = self.info.id.0.as_str(), - track_uri = track_info.uri.as_str(), - "OpenHome Info.Track() returned, searching by URI" - ); - // Trouver l'entry qui matche cet URI - current_id = entries.iter() - .find(|entry| entry.uri == track_info.uri) - .map(|entry| { - debug!( - renderer = self.info.id.0.as_str(), - found_id = entry.id, - "Found current_id by matching URI" - ); - entry.id - }); - } - Err(err) => { - debug!( - renderer = self.info.id.0.as_str(), - error = %err, - "OpenHome Info.Track() also failed" - ); - } - } - } - } + // Get current track ID from the playlist service + let current_id = playlist.id().ok(); let current_index = current_id.and_then(|id| entries.iter().position(|entry| entry.id == id)); diff --git a/pmocontrol/src/sse.rs b/pmocontrol/src/sse.rs index 06439a4c..d07b9146 100644 --- a/pmocontrol/src/sse.rs +++ b/pmocontrol/src/sse.rs @@ -19,6 +19,8 @@ use crate::control_point::ControlPoint; #[cfg(feature = "pmoserver")] use crate::model::{MediaServerEvent, RendererEvent}; #[cfg(feature = "pmoserver")] +use crate::registry::DeviceRegistryRead; +#[cfg(feature = "pmoserver")] use async_stream::stream; #[cfg(feature = "pmoserver")] use axum::{ @@ -168,6 +170,32 @@ pub async fn renderer_events_sse( }); let stream = stream! { + // INITIAL SNAPSHOT: Send Online events for all currently discovered renderers + // This ensures clients see devices that were discovered before they connected + let initial_renderers = { + let registry = control_point.registry(); + let reg = registry.read().unwrap(); + reg.list_renderers() + }; + + for info in initial_renderers { + if info.online { + let timestamp = chrono::Utc::now(); + let payload = RendererEventPayload::Online { + renderer_id: info.id.0.clone(), + friendly_name: info.friendly_name.clone(), + model_name: info.model_name.clone(), + manufacturer: info.manufacturer.clone(), + timestamp, + }; + + if let Ok(json) = serde_json::to_string(&payload) { + yield Ok::<_, axum::Error>(Event::default().event("renderer").data(json)); + } + } + } + + // Then stream future events while let Some(event) = rx_tokio.recv().await { let timestamp = chrono::Utc::now(); @@ -285,6 +313,32 @@ pub async fn media_server_events_sse( }); let stream = stream! { + // INITIAL SNAPSHOT: Send Online events for all currently discovered media servers + // This ensures clients see servers that were discovered before they connected + let initial_servers = { + let registry = control_point.registry(); + let reg = registry.read().unwrap(); + reg.list_servers() + }; + + for info in initial_servers { + if info.online { + let timestamp = chrono::Utc::now(); + let payload = MediaServerEventPayload::Online { + server_id: info.id.0.clone(), + friendly_name: info.friendly_name.clone(), + model_name: info.model_name.clone(), + manufacturer: info.manufacturer.clone(), + timestamp, + }; + + if let Ok(json) = serde_json::to_string(&payload) { + yield Ok::<_, axum::Error>(Event::default().event("media_server").data(json)); + } + } + } + + // Then stream future events while let Some(event) = rx_tokio.recv().await { let timestamp = chrono::Utc::now(); @@ -370,6 +424,53 @@ pub async fn all_events_sse(State(control_point): State>) -> i }); let stream = stream! { + // INITIAL SNAPSHOT: Send Online events for all currently discovered devices + // This ensures clients see devices that were discovered before they connected + let (initial_renderers, initial_servers) = { + let registry = control_point.registry(); + let reg = registry.read().unwrap(); + (reg.list_renderers(), reg.list_servers()) + }; + + // Send renderer Online events + for info in initial_renderers { + if info.online { + let timestamp = chrono::Utc::now(); + let renderer_payload = RendererEventPayload::Online { + renderer_id: info.id.0.clone(), + friendly_name: info.friendly_name.clone(), + model_name: info.model_name.clone(), + manufacturer: info.manufacturer.clone(), + timestamp, + }; + let payload = UnifiedEventPayload::Renderer(renderer_payload); + + if let Ok(json) = serde_json::to_string(&payload) { + yield Ok::<_, axum::Error>(Event::default().event("control").data(json)); + } + } + } + + // Send server Online events + for info in initial_servers { + if info.online { + let timestamp = chrono::Utc::now(); + let server_payload = MediaServerEventPayload::Online { + server_id: info.id.0.clone(), + friendly_name: info.friendly_name.clone(), + model_name: info.model_name.clone(), + manufacturer: info.manufacturer.clone(), + timestamp, + }; + let payload = UnifiedEventPayload::MediaServer(server_payload); + + if let Ok(json) = serde_json::to_string(&payload) { + yield Ok::<_, axum::Error>(Event::default().event("control").data(json)); + } + } + } + + // Then stream future events loop { tokio::select! { Some(event) = renderer_rx_tokio.recv() => {