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 -- <channel_id> 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)
This commit is contained in:
Claude
281
pmoparadise/examples/stream_block.rs
Normal file
281
pmoparadise/examples/stream_block.rs
Normal file
@@ -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 -- <channel_id>
|
||||
//!
|
||||
//! 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<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
) -> Result<Response, StatusCode> {
|
||||
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<Arc<AppState>>) -> 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<dyn std::error::Error>> {
|
||||
// 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<String> = env::args().collect();
|
||||
if args.len() < 2 {
|
||||
eprintln!("Usage: {} <channel_id>", 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(())
|
||||
}
|
||||
Reference in New Issue
Block a user