eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

refactor: Remove obsolete streaming.rs and deprecated examples

Browse Source 
Further cleanup of unused code after paradise/ removal.

## Removed Files

### Module (217 lines)
- **streaming.rs**: FLAC streaming decoder using claxon
  - `ChannelReader`: Async Stream → sync Read adapter
  - `StreamingPCMDecoder`: Claxon-based FLAC decoder
  - `PCMChunk`: PCM data container
  - **Reason**: Was only used by paradise/worker.rs (deleted)
  - **Replacement**: RadioParadiseStreamSource uses pmoflac directly

### Examples (3 files)
- **show_source_image.rs**: Used deprecated RadioParadiseSource
- **with_cache.rs**: Used deprecated RadioParadiseSource
- **test_streaming.rs**: Used deleted streaming.rs module
  - **Replacement**: radio_paradise_stream.rs example shows modern approach

## Updated
- **lib.rs**: Removed `pub mod streaming;`

## Remaining Examples
Valid examples using current API:
- ✅ now_playing.rs - API metadata access
- ✅ stream_block.rs - HTTP block streaming
- ✅ extract_track.rs - Per-track extraction (feature: per-track)
- ✅ radio_paradise_stream.rs - Modern pmoaudio integration

## Statistics
- Before: 3566 lines (after paradise/ removal)
- After: 3348 lines
- This cleanup: -218 lines (-6%)
- **Total removed since start: 3048 lines (-48%)**

## Testing
- ✅ All 23 tests pass
- ✅ Compilation successful with all features
- ✅ Examples compile (except per-track which requires feature)
This commit is contained in:
Claude
2025-11-05 07:49:19 +00:00
parent f80ebd9f3d
commit 74d8788463
5 changed files with 0 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
pmoparadise/examples/show_source_image.rs
View File

@@ -1,71 +0,0 @@
//! Example showing how to access and save the Radio Paradise source image
//!
//! This example demonstrates:
//! - Getting source information via the MusicSource trait
//! - Accessing the embedded WebP image
//! - Optionally saving it to a file
use pmoaudiocache::cache as audio_cache;
use pmocovers::cache as covers_cache;
use pmoparadise::{RadioParadiseClient, RadioParadiseSource};
use pmosource::MusicSource;
use std::fs;
use std::io::Write;
use std::sync::Arc;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Create the client and source
let client = RadioParadiseClient::new().await?;
// Build lightweight caches under the system temp dir for this example
let base_dir = std::env::temp_dir().join(format!(
"pmoparadise_show_source_image_{}",
std::process::id()
));
let covers_dir = base_dir.join("covers");
let audio_dir = base_dir.join("audio");
std::fs::create_dir_all(&covers_dir)?;
std::fs::create_dir_all(&audio_dir)?;
let cover_cache = Arc::new(covers_cache::new_cache(
covers_dir.to_string_lossy().as_ref(),
32,
)?);
let audio_cache = Arc::new(audio_cache::new_cache(
audio_dir.to_string_lossy().as_ref(),
32,
)?);
let source = RadioParadiseSource::new_default(client, cover_cache, audio_cache);
// Display source information
println!("Music Source Information");
println!("========================");
println!("Name: {}", source.name());
println!("ID: {}", source.id());
println!("Image MIME type: {}", source.default_image_mime_type());
// Get the embedded image
let image_data = source.default_image();
println!("Embedded image size: {} bytes", image_data.len());
// Verify WebP format
if image_data.len() >= 12 {
let is_webp = &image_data[0..4] == b"RIFF" && &image_data[8..12] == b"WEBP";
println!("Valid WebP format: {}", is_webp);
}
// Optional: save to file
if std::env::args().any(|arg| arg == "--save") {
let filename = format!("{}_default.webp", source.id());
let mut file = fs::File::create(&filename)?;
file.write_all(image_data)?;
println!("\nImage saved to: {}", filename);
println!("You can view it with: open {}", filename);
} else {
println!("\nTo save the image to disk, run with: --save");
}
Ok(())
}

129
pmoparadise/examples/test_streaming.rs
View File

@@ -1,129 +0,0 @@
//! Test progressive streaming implementation
//!
//! This example tests the streaming implementation and measures performance
//!
//! Run with:
//! ```bash
//! RUST_LOG=info cargo run --example test_streaming
//! ```
use pmoparadise::RadioParadiseClient;
use std::time::Instant;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize tracing with timestamps
tracing_subscriber::fmt()
.with_target(false)
.with_thread_ids(false)
.with_level(true)
.init();
println!("🎵 Testing Progressive FLAC Streaming");
println!("=====================================\n");
// Create the Radio Paradise client
println!("📡 Connecting to Radio Paradise...");
let client = RadioParadiseClient::new().await?;
println!("✅ Connected!\n");
// Get current block
println!("🎧 Fetching current block metadata...");
let block = client.get_block(None).await?;
println!("\n📊 Block Information:");
println!(" Event ID: {}", block.event);
println!(" Songs: {}", block.song_count());
println!(" Duration: ~{} seconds\n", block.length / 1000);
// List songs
println!("🎵 Songs in this block:");
for (idx, song) in block.songs_ordered() {
println!(
" {}. {} - {} ({}s at {}s)",
idx + 1,
song.artist,
song.title,
song.duration / 1000,
song.elapsed / 1000
);
}
println!();
// Now test the streaming decoder
println!("⚡ Starting progressive streaming test...");
println!(" (This will download and decode the block progressively)");
println!();
let start_time = Instant::now();
let block_url = block.url.parse()?;
let http_stream = client.stream_block(&block_url).await?;
use pmoparadise::streaming::StreamingPCMDecoder;
// Decode in a blocking task
let decode_task = tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<(u64, usize)>> {
let mut decoder = StreamingPCMDecoder::new(http_stream)?;
println!(
" 🎼 Stream info: {}Hz, {} channels, {} bits",
decoder.sample_rate(),
decoder.channels(),
decoder.bits_per_sample()
);
let mut chunk_times = Vec::new();
let mut chunk_count = 0;
while let Some(chunk) = decoder.decode_chunk()? {
chunk_count += 1;
chunk_times.push((chunk.position_ms, chunk.samples.len()));
if chunk_count % 50 == 0 {
println!(
" 📦 Chunk {} at {}ms ({} samples)",
chunk_count,
chunk.position_ms,
chunk.samples.len()
);
}
}
Ok(chunk_times)
});
let chunk_times = decode_task
.await
.map_err(|e| anyhow::anyhow!("Join error: {}", e))??;
let total_time = start_time.elapsed();
println!("\n✅ Streaming Complete!");
println!("\n📈 Performance Metrics:");
println!(" Total chunks decoded: {}", chunk_times.len());
println!(" Total time: {:.2}s", total_time.as_secs_f64());
if let Some((first_pos, _)) = chunk_times.first() {
println!(" First chunk at: {}ms", first_pos);
}
if let Some((last_pos, _)) = chunk_times.last() {
println!(
" Last chunk at: {}ms (~{:.1}s)",
last_pos,
last_pos / 1000
);
}
println!("\n💡 Analysis:");
println!(" With the old approach (download all first):");
println!(" - Would need to wait for full download (~12-16s)");
println!(" - Then decode all samples");
println!(" - Total: ~15-20s before first track");
println!();
println!(" With progressive streaming:");
println!(" - First chunks arrive in ~2-3s");
println!(" - First track (3min) ready in ~6-8s");
println!(" - Improvement: ~2x faster! ⚡");
Ok(())
}

107
pmoparadise/examples/with_cache.rs
View File

@@ -1,107 +0,0 @@
//! Example demonstrating Radio Paradise with cache support
//!
//! This example shows how to use the RadioParadiseSource with pmocovers
//! and pmoaudiocache to cache both cover images and audio tracks.
//!
//! Run with:
//! ```bash
//! cargo run --example with_cache --features cache
//! ```
use pmoaudiocache::AudioCache;
use pmocovers::Cache as CoverCache;
use pmoparadise::{RadioParadiseClient, RadioParadiseSource};
use pmosource::MusicSource;
use std::sync::Arc;
use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize tracing
tracing_subscriber::fmt::init();
println!("🎵 Radio Paradise with Cache Support");
println!("=====================================\n");
// Create the Radio Paradise client
println!("📡 Connecting to Radio Paradise...");
let client = RadioParadiseClient::new().await?;
println!("✅ Connected!\n");
// Initialize caches
println!("💾 Initializing caches...");
let cover_cache = Arc::new(CoverCache::new("./cache/covers", 500)?);
let audio_cache = Arc::new(AudioCache::new("./cache/audio", 100)?);
println!("✅ Caches initialized!\n");
// Create the source with caching enabled
let source = RadioParadiseSource::new_with_cache(
client.clone(),
"http://localhost:8080",
50,
Some(cover_cache.clone()),
Some(audio_cache.clone()),
);
println!("📻 Source: {}", source.name());
println!("🆔 ID: {}", source.id());
println!("📝 Supports FIFO: {}\n", source.supports_fifo());
// Fetch current playing information
println!("🎧 Fetching current track information...");
let now_playing = client.now_playing().await?;
let block = Arc::new(now_playing.block.clone());
println!("\n🎵 Now Playing:");
println!(" Event: {}", block.event);
if let Some(song) = &now_playing.current_song {
println!(" Title: {}", song.title);
println!(" Artist: {}", song.artist);
println!(" Album: {}", song.album);
}
println!();
// Add current song to the source
println!("➕ Adding current track to FIFO with caching...");
if let Some(song) = &now_playing.current_song {
source
.add_song(
block.clone(),
song,
now_playing.current_song_index.unwrap_or(0),
)
.await?;
println!("✅ Track added and caching started!");
println!(" - Cover image will be cached to: ./cache/covers/");
println!(" - Audio will be cached to: ./cache/audio/\n");
}
// Wait a bit for caching to start
println!("⏳ Waiting for cache operations to complete...");
sleep(Duration::from_secs(5)).await;
// Get items from FIFO
println!("\n📋 Items in FIFO:");
let items = source.get_items(0, 10).await?;
for (i, item) in items.iter().enumerate() {
println!(
" {}. {} - {}",
i + 1,
item.artist.as_deref().unwrap_or("Unknown"),
item.title
);
// Show resolved URI (will use cached version if available)
if let Ok(uri) = source.resolve_uri(&item.id).await {
println!(" URI: {}", uri);
}
}
println!("\n✨ Example complete!");
println!("\n💡 Tips:");
println!(" - Run the example again to see faster loading from cache");
println!(" - Check ./cache/covers/ for cached cover images");
println!(" - Check ./cache/audio/ for cached FLAC files");
Ok(())
}

1
pmoparadise/src/lib.rs
View File

@@ -216,7 +216,6 @@ pub mod error;
pub mod models;
pub mod source;
pub mod stream;
pub mod streaming;
#[cfg(feature = "per-track")]
pub mod track;

217
pmoparadise/src/streaming.rs
View File

@@ -1,217 +0,0 @@
use anyhow::Result;
use bytes::Bytes;
use futures::stream::Stream;
use std::io::{self, Read};
use std::pin::Pin;
use std::sync::mpsc::{sync_channel, Receiver, RecvError, SyncSender};
use std::time::{Duration, Instant};
const CHANNEL_BUFFER_SIZE: usize = 64; // Augmenté de 16 à 64 pour réduire les warnings "buffer plein"
pub const CHUNK_SIZE_FRAMES: usize = 4096;
pub struct ChannelReader {
receiver: Receiver<Result<Bytes, String>>,
current_chunk: Option<Bytes>,
position: usize,
}
impl ChannelReader {
pub fn new(
stream: Pin<Box<dyn Stream<Item = Result<Bytes, crate::error::Error>> + Send>>,
) -> Self {
let (tx, rx) = sync_channel(CHANNEL_BUFFER_SIZE);
tokio::spawn(Self::stream_feeder(stream, tx));
Self {
receiver: rx,
current_chunk: None,
position: 0,
}
}
async fn stream_feeder(
mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, crate::error::Error>> + Send>>,
tx: SyncSender<Result<Bytes, String>>,
) {
use futures::StreamExt;
while let Some(result) = stream.next().await {
let start = Instant::now();
let to_send = result.map_err(|e| e.to_string());
match tx.try_send(to_send) {
Ok(_) => { /* message envoyé sans attente */ }
Err(std::sync::mpsc::TrySendError::Full(value)) => {
tracing::warn!("stream_feeder: buffer plein");
// Revenir à l’envoi bloquant pour ne pas perdre le message
if tx.send(value).is_err() {
break;
}
}
Err(std::sync::mpsc::TrySendError::Disconnected(_)) => break,
}
let waited = start.elapsed();
tracing::trace!("stream_feeder send {:?}", waited);
if waited > Duration::from_millis(200) {
tracing::warn!("stream_feeder wait {:?}", waited);
}
}
}
}
impl Read for ChannelReader {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let start = Instant::now();
loop {
if let Some(chunk) = &self.current_chunk {
if self.position < chunk.len() {
let available = chunk.len() - self.position;
let to_copy = available.min(buf.len());
buf[..to_copy].copy_from_slice(&chunk[self.position..self.position + to_copy]);
self.position += to_copy;
tracing::trace!(
"ChannelReader copied {} bytes (elapsed {:?})",
to_copy,
start.elapsed()
);
return Ok(to_copy);
}
}
match self.receiver.recv() {
Ok(Ok(bytes)) => {
tracing::trace!(
"ChannelReader received chunk of {} bytes after {:?}",
bytes.len(),
start.elapsed()
);
self.current_chunk = Some(bytes);
self.position = 0;
}
Ok(Err(e)) => {
tracing::warn!("ChannelReader received error chunk: {}", e);
return Err(io::Error::new(io::ErrorKind::Other, e));
}
Err(RecvError) => {
tracing::trace!("ChannelReader stream closed after {:?}", start.elapsed());
return Ok(0);
}
}
}
}
}
#[derive(Debug, Clone)]
pub struct PCMChunk {
pub samples: Vec<i32>,
pub position_ms: u64,
pub sample_rate: u32,
pub channels: u32,
}
pub struct StreamingPCMDecoder<R: Read> {
reader: claxon::FlacReader<std::io::BufReader<R>>,
sample_rate: u32,
channels: u32,
bits_per_sample: u32,
total_samples_decoded: u64,
done: bool,
}
impl StreamingPCMDecoder<ChannelReader> {
/// Create a new decoder from an HTTP stream with default chunk size
pub fn new(http_stream: crate::stream::BlockStream) -> anyhow::Result<Self> {
Self::with_chunk_size(http_stream, CHUNK_SIZE_FRAMES)
}
pub fn with_chunk_size(
http_stream: crate::stream::BlockStream,
_chunk_size: usize,
) -> anyhow::Result<Self> {
let channel_reader = ChannelReader::new(http_stream.into_inner());
let buffered = std::io::BufReader::new(channel_reader);
let reader = claxon::FlacReader::new(buffered)
.map_err(|e| anyhow::anyhow!("FLAC reader error: {}", e))?;
let info = reader.streaminfo();
Ok(Self {
reader,
sample_rate: info.sample_rate,
channels: info.channels,
bits_per_sample: info.bits_per_sample,
total_samples_decoded: 0,
done: false,
})
}
/// Get the sample rate (e.g., 44100 Hz)
pub fn sample_rate(&self) -> u32 {
self.sample_rate
}
/// Get the number of channels (e.g., 2 for stereo)
pub fn channels(&self) -> u32 {
self.channels
}
/// Get bits per sample (e.g., 16)
pub fn bits_per_sample(&self) -> u32 {
self.bits_per_sample
}
pub fn decode_chunk(&mut self) -> anyhow::Result<Option<PCMChunk>> {
if self.done {
return Ok(None);
}
// Crée le FrameReader à la volée (emprunt de self.reader)
let mut frames = self.reader.blocks();
// API claxon 0.6.x : il FAUT fournir un Vec<i32> par valeur
let buf: Vec<i32> = Vec::new();
let frame = match frames.read_next_or_eof(buf) {
Ok(None) => {
self.done = true;
return Ok(None);
}
Ok(Some(f)) => f,
Err(e) => return Err(anyhow::anyhow!("FLAC decode error: {}", e)),
};
let planar_samples: Vec<i32> = frame.into_buffer();
if planar_samples.is_empty() {
self.done = true;
return Ok(None);
}
// IMPORTANT: Claxon retourne les samples en format PLANAR (tous les L, puis tous les R)
// Mais nous avons besoin du format INTERLEAVED (L, R, L, R, ...) pour l'encodage
let block_size = planar_samples.len() / self.channels as usize;
let mut samples = Vec::with_capacity(planar_samples.len());
for i in 0..block_size {
for ch in 0..self.channels as usize {
samples.push(planar_samples[ch * block_size + i]);
}
}
let position_ms = {
let frames = self.total_samples_decoded / self.channels as u64;
(frames * 1000) / self.sample_rate as u64
};
self.total_samples_decoded += samples.len() as u64;
Ok(Some(PCMChunk {
samples,
position_ms,
sample_rate: self.sample_rate,
channels: self.channels,
}))
}
}
pub fn ms_to_frames(ms: u64, sample_rate: u32) -> usize {
((ms as u128 * sample_rate as u128) / 1000) as usize
}
pub fn frames_to_ms(frames: usize, sample_rate: u32) -> u64 {
((frames as u128 * 1000) / sample_rate as u128) as u64
}