pmomusic/pmoparadise/examples/play_and_cache.rs

//! Télécharge un bloc Radio Paradise, le cache, et le joue en même temps
//!
//! Ce programme démontre l'utilisation complète de la chaîne :
//! 1. RadioParadiseStreamSource - Télécharge et décode un bloc FLAC
//! 2. FlacCacheSink - Cache chaque piste en FLAC et alimente une playlist
//! 3. PlaylistSource - Lit la playlist pendant le téléchargement
//! 4. TimerNode - Régule le débit pour éviter EOF prématurés (progressive cache)
//! 5. AudioSink - Joue l'audio sur la sortie standard
//!
//! Architecture :
//! ```text
//! Pipeline 1 (Download & Cache):
//!   RadioParadiseStreamSource → FlacCacheSink (avec playlist abonnée)
//!
//! Pipeline 2 (Playback):
//!   PlaylistSource → TimerNode (rate limiting) → AudioSink
//!                      ↓
//!                 Prévention EOF
//!                 (3s max lead)
//! ```
//!
//! Usage:
//!   cargo run --example play_and_cache --features full -- <channel_id>
//!
//! Exemple:
//!   cargo run --example play_and_cache --features full -- 0    # Main Mix
//!   cargo run --example play_and_cache --features full -- 2    # Rock Mix

use pmoaudio::{AudioPipelineNode, AudioSink, TimerNode};
use pmoaudio_ext::{FlacCacheSink, PlaylistSource};
use pmoaudiocache::{
    new_cache_with_consolidation as new_audio_cache,
    register_audio_cache as register_global_audio_cache,
};
use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache};
use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource};
use pmoplaylist::{register_audio_cache as register_playlist_audio_cache, PlaylistRole};
use std::env;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialiser tracing avec beaucoup de logs
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive(tracing::Level::DEBUG.into())
                .add_directive("pmoaudio=debug".parse()?)
                .add_directive("pmoaudio_ext=debug".parse()?)
                .add_directive("pmoplaylist=debug".parse()?)
                .add_directive("pmoparadise=debug".parse()?)
                .add_directive("pmoaudiocache=debug".parse()?),
        )
        .init();

    tracing::info!("=== Radio Paradise Play & Cache ===");

    // Récupérer les arguments
    let args: Vec<String> = env::args().collect();
    if args.len() < 2 {
        eprintln!("Usage: {} <channel_id> [--null-audio]", args[0]);
        eprintln!();
        eprintln!("Downloads a Radio Paradise block, caches it, and plays it simultaneously.");
        eprintln!();
        eprintln!("Channel IDs:");
        eprintln!("  0 - Main Mix (eclectic, diverse mix)");
        eprintln!("  1 - Mellow Mix (smooth, chilled music)");
        eprintln!("  2 - Rock Mix (classic & modern rock)");
        eprintln!("  3 - World/Etc Mix (global sounds)");
        eprintln!();
        eprintln!("Options:");
        eprintln!("  --null-audio    Don't play audio (for testing without audio device)");
        std::process::exit(1);
    }

    let channel_id: u16 = match args[1].parse() {
        Ok(id) if id <= 3 => id,
        _ => {
            eprintln!("Error: channel_id must be a number between 0 and 3");
            std::process::exit(1);
        }
    };

    let use_null_audio = args.len() > 2 && args[2] == "--null-audio";

    tracing::info!("Channel ID: {}", channel_id);
    if use_null_audio {
        tracing::info!("Using null audio output (no playback)");
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // Initialiser les caches et le gestionnaire de playlist
    // ═══════════════════════════════════════════════════════════════════════════

    let base_dir =
        std::env::var("PMO_CONFIG_DIR").unwrap_or_else(|_| "/tmp/pmomusic_test".to_string());
    std::fs::create_dir_all(&base_dir)?;

    tracing::info!("Initializing caches in: {}", base_dir);

    // Créer le cache audio
    let audio_cache_dir = format!("{}/audio_cache", base_dir);
    std::fs::create_dir_all(&audio_cache_dir)?;
    let audio_cache = new_audio_cache(&audio_cache_dir, 1000).await?;
    tracing::debug!("Audio cache initialized at: {}", audio_cache_dir);

    // Créer le cache de covers
    let cover_cache_dir = format!("{}/cover_cache", base_dir);
    std::fs::create_dir_all(&cover_cache_dir)?;
    let cover_cache = new_cover_cache(&cover_cache_dir, 100).await?;
    tracing::debug!("Cover cache initialized at: {}", cover_cache_dir);

    // Enregistrer le cache audio dans pmoplaylist
    // (requis par pmoplaylist pour valider les pks)
    register_global_audio_cache(audio_cache.clone());
    register_playlist_audio_cache(audio_cache.clone());
    register_cover_cache(cover_cache.clone());
    tracing::debug!("Audio cache registered in pmoplaylist");

    // Utiliser le gestionnaire de playlist singleton
    tracing::info!("Getting playlist manager...");
    let playlist_manager = pmoplaylist::PlaylistManager();
    tracing::debug!("Playlist manager obtained");

    // ═══════════════════════════════════════════════════════════════════════════
    // Créer la playlist pour ce channel
    // ═══════════════════════════════════════════════════════════════════════════

    let playlist_id = format!("radio-paradise-ch{}", channel_id);
    tracing::info!("Creating playlist: {}", playlist_id);

    // Créer une playlist éphémère (non persistante) pour cet exemple
    let writer = playlist_manager
        .get_write_handle(playlist_id.clone())
        .await?;
    writer
        .set_title(format!("Radio Paradise - Channel {}", channel_id))
        .await?;
    writer.set_role(PlaylistRole::Radio).await?;
    writer.flush().await?; // Vider la playlist si elle existait
    tracing::debug!("Playlist created and flushed");

    // Créer le reader pour la lecture
    let reader = playlist_manager.get_read_handle(&playlist_id).await?;
    tracing::debug!("Read handle created");

    // ═══════════════════════════════════════════════════════════════════════════
    // Récupérer les infos du bloc à télécharger
    // ═══════════════════════════════════════════════════════════════════════════

    tracing::info!("Fetching current block metadata...");
    let client = RadioParadiseClient::builder()
        .channel(channel_id)
        .build()
        .await?;

    let block = client.get_block(None).await?;

    tracing::info!("Block Information:");
    tracing::info!("  Event ID: {}", block.event);
    tracing::info!("  Songs: {}", block.song_count());
    tracing::info!("  Duration: {:.1} minutes", block.length as f64 / 60000.0);
    tracing::info!("");

    tracing::info!("Tracklist:");
    for (index, song) in block.songs_ordered() {
        tracing::info!(
            "  {:2}. {} - {} ({})",
            index + 1,
            song.artist,
            song.title,
            song.album.as_deref().unwrap_or("Unknown Album")
        );
    }
    tracing::info!("");

    // ═══════════════════════════════════════════════════════════════════════════
    // Pipeline 1: Téléchargement et cache
    // ═══════════════════════════════════════════════════════════════════════════

    tracing::info!("Creating download pipeline...");

    // Créer la source Radio Paradise
    let mut download_source = RadioParadiseStreamSource::new(client);
    download_source.push_block_id(block.event);
    tracing::debug!(
        "RadioParadiseStreamSource created with block {}",
        block.event
    );

    // Créer le sink de cache FLAC
    let mut cache_sink = FlacCacheSink::new(audio_cache.clone(), cover_cache.clone());
    cache_sink.register_playlist(writer);
    tracing::debug!("FlacCacheSink created and registered with playlist");

    // Connecter source → sink
    download_source.register(Box::new(cache_sink));
    tracing::info!("Download pipeline connected: RadioParadiseStreamSource → FlacCacheSink");

    // ═══════════════════════════════════════════════════════════════════════════
    // Pipeline 2: Lecture depuis la playlist
    // ═══════════════════════════════════════════════════════════════════════════

    tracing::info!("Creating playback pipeline...");

    // Créer la source playlist
    let mut playlist_source = PlaylistSource::new(reader, audio_cache.clone());
    tracing::debug!("PlaylistSource created");

    // Créer le timer node pour réguler le débit (empêche EOF prématurés)
    // Tolère 3 secondes d'avance max pour permettre le buffering
    let mut timer = TimerNode::new(3.0);
    tracing::debug!("TimerNode created (max_lead_time=3.0s)");

    // Créer le sink audio
    let audio_sink = if use_null_audio {
        AudioSink::with_null_output()
    } else {
        AudioSink::make()
    };
    tracing::debug!("AudioSink created");

    // Connecter timer → audio (AVANT de mettre timer dans une Box)
    timer.register(audio_sink);

    // Connecter playlist → timer
    playlist_source.register(timer.boxed());
    tracing::info!("Playback pipeline connected: PlaylistSource → TimerNode → AudioSink");

    // ═══════════════════════════════════════════════════════════════════════════
    // Lancer les deux pipelines en parallèle
    // ═══════════════════════════════════════════════════════════════════════════

    tracing::info!("");
    tracing::info!("========================================");
    tracing::info!("Starting both pipelines...");
    tracing::info!("Pipeline 1: Downloading and caching");
    tracing::info!("Pipeline 2: Playing from playlist");
    tracing::info!("========================================");
    tracing::info!("");

    let stop_token = CancellationToken::new();
    let stop_token_download = stop_token.clone();
    let stop_token_playback = stop_token.clone();

    // Gérer Ctrl+C
    let stop_token_ctrl_c = stop_token.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.ok();
        tracing::warn!("Received Ctrl+C, stopping...");
        stop_token_ctrl_c.cancel();
    });

    let start = std::time::Instant::now();

    // Lancer les deux pipelines en parallèle
    let download_handle = tokio::spawn(async move {
        tracing::info!("[DOWNLOAD] Pipeline starting...");
        let result = Box::new(download_source).run(stop_token_download).await;
        match &result {
            Ok(()) => tracing::info!("[DOWNLOAD] Pipeline completed successfully"),
            Err(e) => tracing::error!("[DOWNLOAD] Pipeline error: {}", e),
        }
        result
    });

    let playback_handle = tokio::spawn(async move {
        // Pas de sleep - le cache progressif permet de démarrer immédiatement
        // dès que le prebuffer (512 KB) est atteint
        tracing::info!("[PLAYBACK] Pipeline starting (will wait for prebuffer)...");
        let result = Box::new(playlist_source).run(stop_token_playback).await;
        match &result {
            Ok(()) => tracing::info!("[PLAYBACK] Pipeline completed successfully"),
            Err(e) => tracing::error!("[PLAYBACK] Pipeline error: {}", e),
        }
        result
    });

    // Attendre les deux pipelines
    let (download_result, playback_result) = tokio::join!(download_handle, playback_handle);

    let elapsed = start.elapsed();

    // Vérifier les résultats
    match (download_result, playback_result) {
        (Ok(Ok(())), Ok(Ok(()))) => {
            tracing::info!("");
            tracing::info!("========================================");
            tracing::info!("✓ Both pipelines completed successfully");
            tracing::info!("  Total time: {:.2}s", elapsed.as_secs_f64());
            tracing::info!("========================================");
        }
        (download_res, playback_res) => {
            tracing::error!("");
            tracing::error!("========================================");
            if let Err(e) = download_res {
                tracing::error!("✗ Download pipeline error: {:?}", e);
            } else if let Ok(Err(e)) = download_res {
                tracing::error!("✗ Download pipeline error: {}", e);
            }
            if let Err(e) = playback_res {
                tracing::error!("✗ Playback pipeline error: {:?}", e);
            } else if let Ok(Err(e)) = playback_res {
                tracing::error!("✗ Playback pipeline error: {}", e);
            }
            tracing::error!("========================================");
            return Err("Pipeline error".into());
        }
    }

    Ok(())
}