From 4884fddf0e90b31da33b80af2af8c58117a406ea Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Nov 2025 19:33:25 +0000 Subject: [PATCH] Add stream_block example for testing HTTP streaming with VLC Creates a new example demonstrating StreamingFlacSink usage with pmoserver for real-world HTTP streaming testing with media players like VLC. Features: - Uses pmoserver instead of raw Axum for realistic testing - Streams a single Radio Paradise block over HTTP - Supports both pure FLAC and ICY metadata modes - Provides /test/stream endpoint for streaming - Provides /test/metadata endpoint for JSON metadata queries - Includes health check endpoint Usage: cargo run --example stream_block --features full -- Testing with VLC: # Pure FLAC mode vlc http://localhost:8080/test/stream # ICY metadata mode (Now Playing) vlc --http-continuous --icy-metadata http://localhost:8080/test/stream Dependencies: - Requires pmoserver for HTTP server - Requires StreamingFlacSink from pmoaudio-ext (http-stream feature) - Integrated with full feature set (pmoaudio + pmoaudio-ext + pmoserver) --- pmoparadise/Cargo.toml | 9 +- pmoparadise/examples/stream_block.rs | 281 +++++++++++++++++++++++++++ 2 files changed, 288 insertions(+), 2 deletions(-) create mode 100644 pmoparadise/examples/stream_block.rs diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index baf25782..220a19cc 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -52,7 +52,7 @@ symphonia = { version = "0.5", features = ["all"] } claxon = "0.4" # pmoaudio-ext with playlist support (optional for examples) -pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist"] } +pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist", "http-stream"] } # Common music source traits pmosource = { path = "../pmosource" } @@ -91,7 +91,7 @@ cache = [] # Active le support pmoaudio node (RadioParadiseStreamSource) pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util"] # Active le support complet avec playlist (pour les exemples avancés) -full = ["pmoaudio", "dep:pmoaudio-ext", "pmoconfig"] +full = ["pmoaudio", "dep:pmoaudio-ext", "pmoconfig", "dep:pmoserver"] [dev-dependencies] # Tests @@ -106,3 +106,8 @@ pmoaudiocache = { path = "../pmoaudiocache" } [[example]] name = "now_playing" path = "examples/now_playing.rs" + +[[example]] +name = "stream_block" +path = "examples/stream_block.rs" +required-features = ["full"] diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs new file mode 100644 index 00000000..f0409173 --- /dev/null +++ b/pmoparadise/examples/stream_block.rs @@ -0,0 +1,281 @@ +//! Streams a Radio Paradise block via HTTP using pmoserver +//! +//! This example demonstrates streaming a single Radio Paradise block +//! using the StreamingFlacSink over HTTP via pmoserver. Perfect for +//! testing with VLC or other media players that support HTTP streaming. +//! +//! Architecture: +//! ```text +//! RadioParadiseStreamSource → StreamingFlacSink +//! ↓ +//! StreamHandle +//! ↓ +//! pmoserver (Axum) +//! ↓ +//! VLC / Media Player Client +//! ``` +//! +//! Usage: +//! cargo run --example stream_block --features full -- +//! +//! Example: +//! cargo run --example stream_block --features full -- 0 # Main Mix +//! +//! Then open in VLC: +//! vlc http://localhost:8080/test/stream +//! +//! For ICY metadata (Now Playing): +//! vlc --http-continuous --icy-metadata http://localhost:8080/test/stream + +use axum::{ + body::Body, + extract::State, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, +}; +use pmoaudio::AudioPipelineNode; +use pmoaudio_ext::StreamingFlacSink; +use pmoflac::EncoderOptions; +use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +use pmoserver::{ServerBuilder, init_logging, LoggingOptions}; +use std::env; +use std::sync::Arc; +use tokio_util::io::ReaderStream; +use tokio_util::sync::CancellationToken; + +/// Shared application state +struct AppState { + stream_handle: pmoaudio_ext::StreamHandle, +} + +/// Main HTTP handler for streaming +async fn stream_handler( + State(state): State>, + headers: HeaderMap, +) -> Result { + tracing::info!("New client connected"); + + // Check if client wants ICY metadata (VLC with --icy-metadata flag) + let want_icy = headers + .get("Icy-MetaData") + .and_then(|v| v.to_str().ok()) + .map(|v| v == "1") + .unwrap_or(false); + + if want_icy { + tracing::info!("Client requested ICY metadata mode"); + + // Subscribe to ICY stream + let icy_stream = state.stream_handle.subscribe_icy(); + + // Build response with ICY headers + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header("icy-metaint", "16000") + .header("icy-name", "Radio Paradise Stream Test") + .header("Cache-Control", "no-cache, no-store") + .body(Body::from_stream(ReaderStream::new(icy_stream))) + .unwrap()) + } else { + tracing::info!("Client requested pure FLAC mode"); + + // Subscribe to pure FLAC stream + let flac_stream = state.stream_handle.subscribe_flac(); + + // Build response + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header("Cache-Control", "no-cache, no-store") + .body(Body::from_stream(ReaderStream::new(flac_stream))) + .unwrap()) + } +} + +/// Metadata endpoint (JSON) +async fn metadata_handler(State(state): State>) -> impl IntoResponse { + let metadata = state.stream_handle.get_metadata().await; + axum::Json(metadata) +} + +/// Health check endpoint +async fn health_handler() -> &'static str { + "OK" +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Initialize logging via pmoserver + init_logging(LoggingOptions { + console_json: false, + console_level: "info", + file_json: false, + file_level: "debug", + file_path: None, + directives: vec![ + "pmoaudio=debug".to_string(), + "pmoaudio_ext=debug".to_string(), + "pmoparadise=debug".to_string(), + ], + })?; + + tracing::info!("=== Radio Paradise HTTP Streaming Test ==="); + + // Parse arguments + let args: Vec = env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: {} ", args[0]); + eprintln!(); + eprintln!("Streams a Radio Paradise block via HTTP for testing."); + 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!("After starting, open in VLC:"); + eprintln!(" vlc http://localhost:8080/stream"); + eprintln!(); + eprintln!("For ICY metadata (Now Playing):"); + eprintln!(" vlc --http-continuous --icy-metadata http://localhost:8080/stream"); + std::process::exit(1); + } + + let channel_id: u8 = 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); + } + }; + + tracing::info!("Channel ID: {}", channel_id); + + // ═══════════════════════════════════════════════════════════════════════════ + // Fetch block metadata + // ═══════════════════════════════════════════════════════════════════════════ + + 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!(""); + + // ═══════════════════════════════════════════════════════════════════════════ + // Create streaming pipeline + // ═══════════════════════════════════════════════════════════════════════════ + + tracing::info!("Creating streaming pipeline..."); + + // Create Radio Paradise source + let mut source = RadioParadiseStreamSource::new(client); + source.push_block_id(block.event); + tracing::debug!("RadioParadiseStreamSource created with block {}", block.event); + + // Create streaming FLAC sink + let encoder_options = EncoderOptions { + compression_level: 5, + verify: false, + ..Default::default() + }; + + let (streaming_sink, stream_handle) = StreamingFlacSink::new(encoder_options, 16); + tracing::debug!("StreamingFlacSink created"); + + // Connect source → sink + source.register(Box::new(streaming_sink)); + tracing::info!("Pipeline connected: RadioParadiseStreamSource → StreamingFlacSink"); + + // ═══════════════════════════════════════════════════════════════════════════ + // Setup pmoserver with streaming routes + // ═══════════════════════════════════════════════════════════════════════════ + + tracing::info!("Setting up pmoserver..."); + + let mut server = ServerBuilder::new("RadioParadiseStreamTest", "http://localhost", 8080) + .build(); + + let app_state = Arc::new(AppState { stream_handle }); + + // Add streaming route + server.add_handler_with_state("/test/stream", stream_handler, app_state.clone()).await; + + // Add metadata route + server.add_handler_with_state("/test/metadata", metadata_handler, app_state.clone()).await; + + // Add health check + server.add_handler("/test/health", health_handler).await; + + tracing::info!(""); + tracing::info!("========================================"); + tracing::info!("Ready to stream!"); + tracing::info!(""); + tracing::info!("Open in VLC:"); + tracing::info!(" vlc http://localhost:8080/test/stream"); + tracing::info!(""); + tracing::info!("For ICY metadata:"); + tracing::info!(" vlc --http-continuous --icy-metadata http://localhost:8080/test/stream"); + tracing::info!(""); + tracing::info!("Metadata endpoint:"); + tracing::info!(" curl http://localhost:8080/test/metadata"); + tracing::info!("========================================"); + tracing::info!(""); + + // ═══════════════════════════════════════════════════════════════════════════ + // Start pipeline and server + // ═══════════════════════════════════════════════════════════════════════════ + + let stop_token = CancellationToken::new(); + let stop_token_pipeline = stop_token.clone(); + + // Start pipeline in background + let pipeline_handle = tokio::spawn(async move { + tracing::info!("[PIPELINE] Starting..."); + let result = Box::new(source).run(stop_token_pipeline).await; + match &result { + Ok(()) => tracing::info!("[PIPELINE] Completed successfully"), + Err(e) => tracing::error!("[PIPELINE] Error: {}", e), + } + result + }); + + // Start pmoserver (blocks until Ctrl+C) + tracing::info!("[SERVER] Starting pmoserver..."); + server.start().await; + + // Server stopped, cancel pipeline + tracing::info!("Server stopped, canceling pipeline..."); + stop_token.cancel(); + + // Wait for pipeline to finish + match pipeline_handle.await { + Ok(Ok(())) => tracing::info!("Pipeline completed successfully"), + Ok(Err(e)) => tracing::error!("Pipeline error: {}", e), + Err(e) => tracing::error!("Pipeline task error: {}", e), + } + + tracing::info!("Shutdown complete"); + Ok(()) +}