feat(streaming): add seekbar progress and dynamic buffer sizing

- Pass track duration from frontend to streaming playback for seekbar
- Add StreamInfo struct with speed measurement during probe
- Implement StreamingConfig::from_speed_mbps() for adaptive buffer sizing
- Add play_streaming_dynamic() method using speed-based buffer config
- Enhanced download logging with progress %, speed (current/avg), and ETA
- Auto-cache streamed tracks for instant replay

The seekbar now shows playback progress during streaming by using
the track's known duration from metadata rather than detecting it
from incomplete audio data.
This commit is contained in:
vicrodh committed 2026-01-25 18:06:41 -06:00
1 parent 7d0aa215a5
commit 008c5623cb
5 files changed
+201 -39

No files matched your search

+109 -28
View File
@@ -17,11 +17,12 @@ use crate::AppState;
#[tauri::command] #[tauri::command]
pub async fn play_track( pub async fn play_track(
track_id: u64, track_id: u64,
duration_secs: Option<u64>,
state: State<'_, AppState>, state: State<'_, AppState>,
offline_cache: State<'_, OfflineCacheState>, offline_cache: State<'_, OfflineCacheState>,
audio_settings: State<'_, AudioSettingsState>, audio_settings: State<'_, AudioSettingsState>,
) -> Result<(), String> { ) -> Result<(), String> {
log::info!("Command: play_track {}", track_id); log::info!("Command: play_track {} (duration: {:?}s)", track_id, duration_secs);
// First check offline cache (persistent disk cache) // First check offline cache (persistent disk cache)
{ {
@@ -118,25 +119,28 @@ pub async fn play_track(
if stream_first_enabled { if stream_first_enabled {
// Use streaming playback - start playing before full download // Use streaming playback - start playing before full download
log::info!("Streaming mode enabled - starting stream-first playback for track {}", track_id); log::info!("🚀 Streaming mode enabled - starting stream-first playback for track {}", track_id);
// Get content length and audio info via HEAD request // Get content length, audio info, and measured speed via HEAD request
let (content_length, sample_rate, channels) = get_stream_info(&stream_url.url).await?; let stream_info = get_stream_info(&stream_url.url).await?;
log::info!( log::info!(
"Stream info: {} bytes, {}Hz, {} channels", "📊 Stream info: {:.2} MB, {}Hz, {} channels, {:.1} MB/s",
content_length, stream_info.content_length as f64 / (1024.0 * 1024.0),
sample_rate, stream_info.sample_rate,
channels stream_info.channels,
stream_info.speed_mbps
); );
// Start streaming playback and get buffer writer // Start streaming playback with dynamic buffer based on measured speed
let buffer_writer = state.player.play_streaming( // The player will use from_speed_mbps internally
let buffer_writer = state.player.play_streaming_dynamic(
track_id, track_id,
sample_rate, stream_info.sample_rate,
channels, stream_info.channels,
content_length, stream_info.content_length,
buffer_seconds, stream_info.speed_mbps,
duration_secs.unwrap_or(0), // Use 0 if not provided
)?; )?;
// Release client lock before spawning background download // Release client lock before spawning background download
@@ -145,9 +149,10 @@ pub async fn play_track(
// Spawn background task to download and push data to buffer // Spawn background task to download and push data to buffer
let url = stream_url.url.clone(); let url = stream_url.url.clone();
let cache_clone = cache.clone(); let cache_clone = cache.clone();
let content_len = stream_info.content_length;
tokio::spawn(async move { tokio::spawn(async move {
match download_and_stream(&url, buffer_writer, track_id, cache_clone).await { match download_and_stream(&url, buffer_writer, track_id, cache_clone, content_len).await {
Ok(()) => log::info!("Streaming download complete for track {}", track_id), Ok(()) => log::info!("🎵 Track {} ready for instant replay from cache", track_id),
Err(e) => log::error!("Streaming download failed for track {}: {}", track_id, e), Err(e) => log::error!("Streaming download failed for track {}: {}", track_id, e),
} }
}); });
@@ -278,9 +283,17 @@ async fn download_audio(url: &str) -> Result<Vec<u8>, String> {
Ok(bytes.to_vec()) Ok(bytes.to_vec())
} }
/// Get stream info (content length, sample rate, channels) via HEAD request and initial bytes /// Stream info including measured download speed
async fn get_stream_info(url: &str) -> Result<(u64, u32, u16), String> { pub struct StreamInfo {
use std::time::Duration; pub content_length: u64,
pub sample_rate: u32,
pub channels: u16,
pub speed_mbps: f64,
}
/// Get stream info (content length, sample rate, channels, speed) via HEAD request and initial bytes
async fn get_stream_info(url: &str) -> Result<StreamInfo, String> {
use std::time::{Duration, Instant};
let client = reqwest::Client::builder() let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30)) .timeout(Duration::from_secs(30))
@@ -307,8 +320,10 @@ async fn get_stream_info(url: &str) -> Result<(u64, u32, u16), String> {
.and_then(|s| s.parse::<u64>().ok()) .and_then(|s| s.parse::<u64>().ok())
.ok_or_else(|| "No content-length header".to_string())?; .ok_or_else(|| "No content-length header".to_string())?;
// Download first ~64KB to probe audio format // Download first ~64KB to probe audio format and measure speed
// This is enough for FLAC/M4A headers // This is enough for FLAC/M4A headers
let start_time = Instant::now();
let range_response = client let range_response = client
.get(url) .get(url)
.header("User-Agent", "Mozilla/5.0") .header("User-Agent", "Mozilla/5.0")
@@ -326,10 +341,30 @@ async fn get_stream_info(url: &str) -> Result<(u64, u32, u16), String> {
.await .await
.map_err(|e| format!("Failed to read initial bytes: {}", e))?; .map_err(|e| format!("Failed to read initial bytes: {}", e))?;
let elapsed = start_time.elapsed();
let bytes_downloaded = initial_bytes.len() as f64;
let speed_mbps = if elapsed.as_secs_f64() > 0.0 {
(bytes_downloaded / elapsed.as_secs_f64()) / (1024.0 * 1024.0)
} else {
10.0 // Assume fast if instant
};
log::info!(
"📶 Probe: {}KB in {:.0}ms = {:.1} MB/s",
initial_bytes.len() / 1024,
elapsed.as_millis(),
speed_mbps
);
// Try to extract audio format from initial bytes // Try to extract audio format from initial bytes
let (sample_rate, channels) = extract_audio_format_from_header(&initial_bytes)?; let (sample_rate, channels) = extract_audio_format_from_header(&initial_bytes)?;
Ok((content_length, sample_rate, channels)) Ok(StreamInfo {
content_length,
sample_rate,
channels,
speed_mbps,
})
} }
/// Extract sample rate and channels from audio file header /// Extract sample rate and channels from audio file header
@@ -427,8 +462,9 @@ async fn download_and_stream(
writer: crate::player::BufferWriter, writer: crate::player::BufferWriter,
track_id: u64, track_id: u64,
cache: Arc<AudioCache>, cache: Arc<AudioCache>,
content_length: u64,
) -> Result<(), String> { ) -> Result<(), String> {
use std::time::Duration; use std::time::{Duration, Instant};
use futures_util::StreamExt; use futures_util::StreamExt;
let client = reqwest::Client::builder() let client = reqwest::Client::builder()
@@ -437,7 +473,10 @@ async fn download_and_stream(
.build() .build()
.map_err(|e| format!("Failed to create HTTP client: {}", e))?; .map_err(|e| format!("Failed to create HTTP client: {}", e))?;
log::info!("Starting streaming download for track {}", track_id); log::info!("📥 Starting streaming download for track {} ({:.2} MB total)",
track_id,
content_length as f64 / (1024.0 * 1024.0)
);
let response = client let response = client
.get(url) .get(url)
@@ -450,9 +489,12 @@ async fn download_and_stream(
return Err(format!("Stream request failed: {}", response.status())); return Err(format!("Stream request failed: {}", response.status()));
} }
let mut all_data = Vec::new(); let mut all_data = Vec::with_capacity(content_length as usize);
let mut stream = response.bytes_stream(); let mut stream = response.bytes_stream();
let mut bytes_received = 0u64; let mut bytes_received = 0u64;
let start_time = Instant::now();
let mut last_log_time = Instant::now();
let mut last_log_bytes = 0u64;
while let Some(chunk_result) = stream.next().await { while let Some(chunk_result) = stream.next().await {
let chunk = chunk_result.map_err(|e| format!("Stream chunk error: {}", e))?; let chunk = chunk_result.map_err(|e| format!("Stream chunk error: {}", e))?;
@@ -466,9 +508,39 @@ async fn download_and_stream(
log::error!("Failed to push chunk to buffer: {}", e); log::error!("Failed to push chunk to buffer: {}", e);
} }
// Log progress every ~1MB // Log progress every ~500ms with speed info
if bytes_received % (1024 * 1024) < chunk.len() as u64 { let now = Instant::now();
log::debug!("Streaming download progress: {} bytes", bytes_received); if now.duration_since(last_log_time) >= Duration::from_millis(500) {
let elapsed_total = start_time.elapsed().as_secs_f64();
let elapsed_interval = now.duration_since(last_log_time).as_secs_f64();
let bytes_this_interval = bytes_received - last_log_bytes;
// Current speed (MB/s)
let current_speed = (bytes_this_interval as f64 / elapsed_interval) / (1024.0 * 1024.0);
// Average speed
let avg_speed = (bytes_received as f64 / elapsed_total) / (1024.0 * 1024.0);
// Progress percentage
let progress = (bytes_received as f64 / content_length as f64) * 100.0;
// ETA
let remaining_bytes = content_length.saturating_sub(bytes_received);
let eta_secs = if avg_speed > 0.0 {
remaining_bytes as f64 / (avg_speed * 1024.0 * 1024.0)
} else {
0.0
};
log::info!(
"📊 Download: {:.1}% ({:.2}/{:.2} MB) | Speed: {:.2} MB/s (avg: {:.2}) | ETA: {:.1}s",
progress,
bytes_received as f64 / (1024.0 * 1024.0),
content_length as f64 / (1024.0 * 1024.0),
current_speed,
avg_speed,
eta_secs
);
last_log_time = now;
last_log_bytes = bytes_received;
} }
} }
@@ -477,9 +549,18 @@ async fn download_and_stream(
log::error!("Failed to mark buffer complete: {}", e); log::error!("Failed to mark buffer complete: {}", e);
} }
log::info!("Streaming download complete: {} bytes total", bytes_received); let total_time = start_time.elapsed();
let avg_speed = (bytes_received as f64 / total_time.as_secs_f64()) / (1024.0 * 1024.0);
log::info!(
"✅ Streaming download complete: {:.2} MB in {:.1}s ({:.2} MB/s avg)",
bytes_received as f64 / (1024.0 * 1024.0),
total_time.as_secs_f64(),
avg_speed
);
// Cache the complete file for future plays // Cache the complete file for future plays
log::info!("💾 Caching track {} for future playback", track_id);
cache.insert(track_id, all_data); cache.insert(track_id, all_data);
Ok(()) Ok(())
+57 -8
View File
@@ -58,6 +58,7 @@ enum AudioCommand {
track_id: u64, track_id: u64,
sample_rate: u32, sample_rate: u32,
channels: u16, channels: u16,
duration_secs: u64,
}, },
/// Pause playback /// Pause playback
Pause, Pause,
@@ -1057,12 +1058,13 @@ impl Player {
actual_duration actual_duration
); );
} }
AudioCommand::PlayStreaming { source, track_id, sample_rate, channels } => { AudioCommand::PlayStreaming { source, track_id, sample_rate, channels, duration_secs } => {
log::info!( log::info!(
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels)", "Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s)",
track_id, track_id,
sample_rate, sample_rate,
channels channels,
duration_secs
); );
*pause_suspend_deadline = None; *pause_suspend_deadline = None;
@@ -1244,9 +1246,9 @@ impl Player {
); );
} }
// Duration unknown until download completes - estimate from content-length if available // Set duration from track metadata (passed from frontend)
// For now, set to 0 and update when known // This allows the seekbar to show progress even during streaming
thread_state.duration.store(0, Ordering::SeqCst); thread_state.duration.store(duration_secs, Ordering::SeqCst);
// Box the incremental source to match the expected type // Box the incremental source to match the expected type
let source_to_play: Box<dyn Source<Item = i16> + Send> = Box::new(incremental_source); let source_to_play: Box<dyn Source<Item = i16> + Send> = Box::new(incremental_source);
@@ -1667,13 +1669,15 @@ impl Player {
channels: u16, channels: u16,
content_length: u64, content_length: u64,
buffer_seconds: u8, buffer_seconds: u8,
duration_secs: u64,
) -> Result<BufferWriter, String> { ) -> Result<BufferWriter, String> {
log::info!( log::info!(
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total)", "Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s)",
track_id, track_id,
sample_rate, sample_rate,
channels, channels,
content_length content_length,
duration_secs
); );
// Use StreamingConfig::from_seconds for proper buffer sizing // Use StreamingConfig::from_seconds for proper buffer sizing
@@ -1688,6 +1692,7 @@ impl Player {
track_id, track_id,
sample_rate, sample_rate,
channels, channels,
duration_secs,
}) })
.map_err(|e| { .map_err(|e| {
log::error!("Player: Failed to send streaming command: {}", e); log::error!("Player: Failed to send streaming command: {}", e);
@@ -1698,6 +1703,50 @@ impl Player {
Ok(writer) Ok(writer)
} }
/// Play from streaming source with dynamic buffer based on measured speed
/// Returns the BufferWriter so caller can push data as it downloads
pub fn play_streaming_dynamic(
&self,
track_id: u64,
sample_rate: u32,
channels: u16,
content_length: u64,
speed_mbps: f64,
duration_secs: u64,
) -> Result<BufferWriter, String> {
log::info!(
"Player: Starting dynamic streaming for track {} ({}Hz, {}ch, {:.2} MB, {:.1} MB/s, {}s)",
track_id,
sample_rate,
channels,
content_length as f64 / (1024.0 * 1024.0),
speed_mbps,
duration_secs
);
// Use StreamingConfig::from_speed_mbps for dynamic buffer sizing
let config = StreamingConfig::from_speed_mbps(speed_mbps);
let (source, writer) = BufferedMediaSource::new(config, Some(content_length));
let source = Arc::new(source);
self.tx
.send(AudioCommand::PlayStreaming {
source: source.clone(),
track_id,
sample_rate,
channels,
duration_secs,
})
.map_err(|e| {
log::error!("Player: Failed to send streaming command: {}", e);
format!("Failed to send streaming play command: {}", e)
})?;
log::info!("Player: Dynamic streaming playback initiated");
Ok(writer)
}
/// Download audio from URL with timeout /// Download audio from URL with timeout
async fn download_audio(&self, url: &str) -> Result<Vec<u8>, String> { async fn download_audio(&self, url: &str) -> Result<Vec<u8>, String> {
use std::time::Duration; use std::time::Duration;
+32
View File
@@ -82,6 +82,38 @@ impl StreamingConfig {
max_buffer_bytes: 100 * 1024 * 1024, max_buffer_bytes: 100 * 1024 * 1024,
} }
} }
/// Create config dynamically based on measured download speed
///
/// - Very fast (>10 MB/s): 256KB (instant start)
/// - Fast (5-10 MB/s): 384KB
/// - Normal (2-5 MB/s): 512KB
/// - Slow (1-2 MB/s): 1MB (more buffer to prevent stutter)
/// - Very slow (<1 MB/s): 2MB
pub fn from_speed_mbps(speed_mbps: f64) -> Self {
let initial_buffer = if speed_mbps >= 10.0 {
256 * 1024 // 256KB - instant start for very fast connections
} else if speed_mbps >= 5.0 {
384 * 1024 // 384KB
} else if speed_mbps >= 2.0 {
512 * 1024 // 512KB - default
} else if speed_mbps >= 1.0 {
1024 * 1024 // 1MB - more buffer for slower connections
} else {
2 * 1024 * 1024 // 2MB - maximum buffer for very slow connections
};
log::info!(
"📶 Dynamic buffer: {:.1} MB/s detected → {}KB initial buffer",
speed_mbps,
initial_buffer / 1024
);
Self {
initial_buffer_bytes: initial_buffer,
max_buffer_bytes: 100 * 1024 * 1024,
}
}
} }
/// Internal state shared between reader and writer /// Internal state shared between reader and writer
+1 -1
View File
@@ -114,7 +114,7 @@ export async function playTrack(
if (isLocal) { if (isLocal) {
await invoke('library_play_track', { trackId: track.id }); await invoke('library_play_track', { trackId: track.id });
} else { } else {
await invoke('play_track', { trackId: track.id }); await invoke('play_track', { trackId: track.id, durationSecs: track.duration });
} }
} }
+2 -2
View File
@@ -245,8 +245,8 @@ export async function togglePlay(): Promise<void> {
console.log('[Player] Restoring local track:', localTrackId); console.log('[Player] Restoring local track:', localTrackId);
await invoke('library_play_track', { trackId: localTrackId }); await invoke('library_play_track', { trackId: localTrackId });
} else { } else {
// Qobuz track - use play_track // Qobuz track - use play_track with duration for seekbar
await invoke('play_track', { trackId: currentTrack.id }); await invoke('play_track', { trackId: currentTrack.id, durationSecs: currentTrack.duration });
} }
// Seek to saved position after a short delay to let audio load // Seek to saved position after a short delay to let audio load