Joren/qbz
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/stream-first-feature-flag' into pre-release

Browse Source
This commit is contained in:
Victor RH committed 2026-01-25 18:14:46 -06:00
2 parents 4b9d819a98 008c5623cb
commit 009fa8d4c5
8 files changed
+1580 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src-tauri/src/commands/playback.rs
+354 -3
View File
@@ -7,6 +7,7 @@ use tokio::sync::Mutex;
use crate::api::client::QobuzClient;
use crate::api::models::Quality;
use crate::cache::AudioCache;
use crate::config::audio_settings::AudioSettingsState;
use crate::offline_cache::OfflineCacheState;
use crate::player::PlaybackState;
use crate::queue::QueueManager;
@@ -16,10 +17,12 @@ use crate::AppState;
#[tauri::command]
pub async fn play_track(
track_id: u64,
duration_secs: Option<u64>,
state: State<'_, AppState>,
offline_cache: State<'_, OfflineCacheState>,
audio_settings: State<'_, AudioSettingsState>,
) -> 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)
{
@@ -89,8 +92,20 @@ pub async fn play_track(
}
}
// Not in any cache - download and cache in memory
log::info!("Track {} not in any cache, streaming...", track_id);
// Not in any cache - check if streaming is enabled
log::info!("Track {} not in any cache, checking streaming setting...", track_id);
// Check if stream_first_track is enabled
let (stream_first_enabled, buffer_seconds) = {
let store = audio_settings.store.lock().map_err(|e| format!("Lock error: {}", e))?;
match store.get_settings() {
Ok(settings) => (settings.stream_first_track, settings.stream_buffer_seconds),
Err(e) => {
log::warn!("Failed to get audio settings, using defaults: {}", e);
(false, 3)
}
}
};
let client = state.client.lock().await;
@@ -102,6 +117,59 @@ pub async fn play_track(
log::info!("Got stream URL for track {}", track_id);
if stream_first_enabled {
// Use streaming playback - start playing before full download
log::info!("🚀 Streaming mode enabled - starting stream-first playback for track {}", track_id);
// Get content length, audio info, and measured speed via HEAD request
let stream_info = get_stream_info(&stream_url.url).await?;
log::info!(
"📊 Stream info: {:.2} MB, {}Hz, {} channels, {:.1} MB/s",
stream_info.content_length as f64 / (1024.0 * 1024.0),
stream_info.sample_rate,
stream_info.channels,
stream_info.speed_mbps
);
// Start streaming playback with dynamic buffer based on measured speed
// The player will use from_speed_mbps internally
let buffer_writer = state.player.play_streaming_dynamic(
track_id,
stream_info.sample_rate,
stream_info.channels,
stream_info.content_length,
stream_info.speed_mbps,
duration_secs.unwrap_or(0), // Use 0 if not provided
)?;
// Release client lock before spawning background download
drop(client);
// Spawn background task to download and push data to buffer
let url = stream_url.url.clone();
let cache_clone = cache.clone();
let content_len = stream_info.content_length;
tokio::spawn(async move {
match download_and_stream(&url, buffer_writer, track_id, cache_clone, content_len).await {
Ok(()) => log::info!("🎵 Track {} ready for instant replay from cache", track_id),
Err(e) => log::error!("Streaming download failed for track {}: {}", track_id, e),
}
});
// Prefetch next track in background
spawn_prefetch(
state.client.clone(),
state.audio_cache.clone(),
&state.queue,
);
return Ok(());
}
// Standard download path (streaming disabled)
log::info!("Standard download mode for track {}", track_id);
// Download the audio
let audio_data = download_audio(&stream_url.url).await?;
let data_size = audio_data.len();
@@ -215,6 +283,289 @@ async fn download_audio(url: &str) -> Result<Vec<u8>, String> {
Ok(bytes.to_vec())
}
/// Stream info including measured download speed
pub struct StreamInfo {
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()
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|e| format!("Failed to create HTTP client: {}", e))?;
// First, do a HEAD request to get content length
let head_response = client
.head(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|e| format!("Failed HEAD request: {}", e))?;
if !head_response.status().is_success() {
return Err(format!("HEAD request failed: {}", head_response.status()));
}
let content_length = head_response
.headers()
.get("content-length")
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<u64>().ok())
.ok_or_else(|| "No content-length header".to_string())?;
// Download first ~64KB to probe audio format and measure speed
// This is enough for FLAC/M4A headers
let start_time = Instant::now();
let range_response = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.header("Range", "bytes=0-65535")
.send()
.await
.map_err(|e| format!("Failed range request: {}", e))?;
if !range_response.status().is_success() && range_response.status() != reqwest::StatusCode::PARTIAL_CONTENT {
return Err(format!("Range request failed: {}", range_response.status()));
}
let initial_bytes = range_response
.bytes()
.await
.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
let (sample_rate, channels) = extract_audio_format_from_header(&initial_bytes)?;
Ok(StreamInfo {
content_length,
sample_rate,
channels,
speed_mbps,
})
}
/// Extract sample rate and channels from audio file header
fn extract_audio_format_from_header(data: &[u8]) -> Result<(u32, u16), String> {
use std::io::{BufReader, Cursor};
use rodio::Decoder;
// Check if this is FLAC
if data.len() >= 4 && &data[0..4] == b"fLaC" {
// Parse FLAC STREAMINFO block
// FLAC format: "fLaC" + METADATA_BLOCK_HEADER (4 bytes) + STREAMINFO
if data.len() >= 26 {
// STREAMINFO starts at byte 8
// Bytes 10-12: sample rate (20 bits) + channels (3 bits) + bits per sample (5 bits)
let sr_high = ((data[18] as u32) << 12) | ((data[19] as u32) << 4) | ((data[20] as u32) >> 4);
let sample_rate = sr_high;
let channels = ((data[20] >> 1) & 0x07) + 1;
return Ok((sample_rate, channels as u16));
}
}
// Check if this is M4A/MP4 (ftyp box)
if data.len() >= 12 && &data[4..8] == b"ftyp" {
// M4A uses AAC or ALAC, try symphonia probe
use symphonia::core::formats::FormatOptions;
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
use symphonia::default::get_probe;
let cursor = Box::new(Cursor::new(data.to_vec())) as Box<dyn std::io::Read + Send + Sync>;
// For probing, we need to create a MediaSource
// Use a simple wrapper
struct ProbeSource {
inner: Cursor<Vec<u8>>,
len: u64,
}
impl std::io::Read for ProbeSource {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.inner.read(buf)
}
}
impl std::io::Seek for ProbeSource {
fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
self.inner.seek(pos)
}
}
impl symphonia::core::io::MediaSource for ProbeSource {
fn is_seekable(&self) -> bool { true }
fn byte_len(&self) -> Option<u64> { Some(self.len) }
}
let len = data.len() as u64;
let source = Box::new(ProbeSource { inner: Cursor::new(data.to_vec()), len });
let mss = MediaSourceStream::new(source, Default::default());
let mut hint = Hint::new();
hint.with_extension("m4a");
let format_opts = FormatOptions {
enable_gapless: true,
..Default::default()
};
if let Ok(probed) = get_probe().format(&hint, mss, &format_opts, &MetadataOptions::default()) {
if let Some(track) = probed.format.default_track() {
let sample_rate = track.codec_params.sample_rate.unwrap_or(44100);
let channels = track.codec_params.channels.map(|c| c.count() as u16).unwrap_or(2);
return Ok((sample_rate, channels));
}
}
// Default to common values for M4A
return Ok((44100, 2));
}
// Try rodio decoder for other formats
match Decoder::new(BufReader::new(Cursor::new(data.to_vec()))) {
Ok(decoder) => {
use rodio::Source;
Ok((decoder.sample_rate(), decoder.channels()))
}
Err(_) => {
// Default fallback
log::warn!("Could not determine audio format, using defaults (44100Hz, stereo)");
Ok((44100, 2))
}
}
}
/// Download audio chunks and stream them to the buffer writer
/// Also caches the complete data when download finishes
async fn download_and_stream(
url: &str,
writer: crate::player::BufferWriter,
track_id: u64,
cache: Arc<AudioCache>,
content_length: u64,
) -> Result<(), String> {
use std::time::{Duration, Instant};
use futures_util::StreamExt;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(300)) // Longer timeout for streaming
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|e| format!("Failed to create HTTP client: {}", e))?;
log::info!("📥 Starting streaming download for track {} ({:.2} MB total)",
track_id,
content_length as f64 / (1024.0 * 1024.0)
);
let response = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|e| format!("Failed to start stream: {}", e))?;
if !response.status().is_success() {
return Err(format!("Stream request failed: {}", response.status()));
}
let mut all_data = Vec::with_capacity(content_length as usize);
let mut stream = response.bytes_stream();
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 {
let chunk = chunk_result.map_err(|e| format!("Stream chunk error: {}", e))?;
bytes_received += chunk.len() as u64;
// Accumulate for caching
all_data.extend_from_slice(&chunk);
// Push to streaming buffer
if let Err(e) = writer.push_chunk(&chunk) {
log::error!("Failed to push chunk to buffer: {}", e);
}
// Log progress every ~500ms with speed info
let now = Instant::now();
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;
}
}
// Mark stream as complete
if let Err(e) = writer.complete() {
log::error!("Failed to mark buffer complete: {}", e);
}
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
log::info!("💾 Caching track {} for future playback", track_id);
cache.insert(track_id, all_data);
Ok(())
}
/// Number of Qobuz tracks to prefetch (not total tracks, just Qobuz)
const QOBUZ_PREFETCH_COUNT: usize = 3;
src-tauri/src/config/audio_settings.rs
+54 -2
View File
@@ -16,6 +16,10 @@ pub struct AudioSettings {
pub backend_type: Option<AudioBackendType>, // None = auto-detect
pub alsa_plugin: Option<AlsaPlugin>, // Only used when backend is ALSA
pub alsa_hardware_volume: bool, // Use ALSA mixer for volume (only with hw: devices)
/// When true, uncached tracks start playing via streaming instead of waiting for full download
pub stream_first_track: bool,
/// Initial buffer size in seconds before starting streaming playback (1-10, default 3)
pub stream_buffer_seconds: u8,
}
impl Default for AudioSettings {
@@ -28,6 +32,8 @@ impl Default for AudioSettings {
backend_type: None, // Auto-detect (PipeWire if available, else ALSA)
alsa_plugin: Some(AlsaPlugin::Hw), // Default to hw (bit-perfect)
alsa_hardware_volume: false, // Disabled by default (maximum compatibility)
stream_first_track: false, // Disabled by default (current behavior)
stream_buffer_seconds: 3, // 3 seconds initial buffer
}
}
}
@@ -58,7 +64,9 @@ impl AudioSettingsStore {
preferred_sample_rate INTEGER,
backend_type TEXT,
alsa_plugin TEXT,
alsa_hardware_volume INTEGER NOT NULL DEFAULT 0
alsa_hardware_volume INTEGER NOT NULL DEFAULT 0,
stream_first_track INTEGER NOT NULL DEFAULT 0,
stream_buffer_seconds INTEGER NOT NULL DEFAULT 3
);
INSERT OR IGNORE INTO audio_settings (id, exclusive_mode, dac_passthrough)
VALUES (1, 0, 0);"
@@ -68,6 +76,8 @@ impl AudioSettingsStore {
let _ = conn.execute("ALTER TABLE audio_settings ADD COLUMN backend_type TEXT", []);
let _ = conn.execute("ALTER TABLE audio_settings ADD COLUMN alsa_plugin TEXT", []);
let _ = conn.execute("ALTER TABLE audio_settings ADD COLUMN alsa_hardware_volume INTEGER DEFAULT 0", []);
let _ = conn.execute("ALTER TABLE audio_settings ADD COLUMN stream_first_track INTEGER DEFAULT 0", []);
let _ = conn.execute("ALTER TABLE audio_settings ADD COLUMN stream_buffer_seconds INTEGER DEFAULT 3", []);
Ok(Self { conn })
}
@@ -75,7 +85,7 @@ impl AudioSettingsStore {
pub fn get_settings(&self) -> Result<AudioSettings, String> {
self.conn
.query_row(
"SELECT output_device, exclusive_mode, dac_passthrough, preferred_sample_rate, backend_type, alsa_plugin, alsa_hardware_volume FROM audio_settings WHERE id = 1",
"SELECT output_device, exclusive_mode, dac_passthrough, preferred_sample_rate, backend_type, alsa_plugin, alsa_hardware_volume, stream_first_track, stream_buffer_seconds FROM audio_settings WHERE id = 1",
[],
|row| {
// Parse backend_type from JSON string
@@ -96,6 +106,8 @@ impl AudioSettingsStore {
backend_type,
alsa_plugin,
alsa_hardware_volume: row.get::<_, Option<i64>>(6)?.unwrap_or(0) != 0,
stream_first_track: row.get::<_, Option<i64>>(7)?.unwrap_or(0) != 0,
stream_buffer_seconds: row.get::<_, Option<i64>>(8)?.unwrap_or(3) as u8,
})
},
)
@@ -181,6 +193,28 @@ impl AudioSettingsStore {
.map_err(|e| format!("Failed to set ALSA hardware volume: {}", e))?;
Ok(())
}
pub fn set_stream_first_track(&self, enabled: bool) -> Result<(), String> {
self.conn
.execute(
"UPDATE audio_settings SET stream_first_track = ?1 WHERE id = 1",
params![enabled as i64],
)
.map_err(|e| format!("Failed to set stream first track: {}", e))?;
Ok(())
}
pub fn set_stream_buffer_seconds(&self, seconds: u8) -> Result<(), String> {
// Clamp to valid range 1-10
let clamped = seconds.clamp(1, 10);
self.conn
.execute(
"UPDATE audio_settings SET stream_buffer_seconds = ?1 WHERE id = 1",
params![clamped as i64],
)
.map_err(|e| format!("Failed to set stream buffer seconds: {}", e))?;
Ok(())
}
}
/// Thread-safe wrapper
@@ -267,3 +301,21 @@ pub fn set_audio_alsa_hardware_volume(
let store = state.store.lock().map_err(|e| format!("Lock error: {}", e))?;
store.set_alsa_hardware_volume(enabled)
}
#[tauri::command]
pub fn set_audio_stream_first_track(
state: tauri::State<'_, AudioSettingsState>,
enabled: bool,
) -> Result<(), String> {
let store = state.store.lock().map_err(|e| format!("Lock error: {}", e))?;
store.set_stream_first_track(enabled)
}
#[tauri::command]
pub fn set_audio_stream_buffer_seconds(
state: tauri::State<'_, AudioSettingsState>,
seconds: u8,
) -> Result<(), String> {
let store = state.store.lock().map_err(|e| format!("Lock error: {}", e))?;
store.set_stream_buffer_seconds(seconds)
}
src-tauri/src/lib.rs
+2
View File
@@ -699,6 +699,8 @@ pub fn run() {
config::audio_settings::set_audio_backend_type,
config::audio_settings::set_audio_alsa_plugin,
config::audio_settings::set_audio_alsa_hardware_volume,
config::audio_settings::set_audio_stream_first_track,
config::audio_settings::set_audio_stream_buffer_seconds,
// Audio backend commands
commands::get_available_backends,
commands::get_devices_for_backend,
src-tauri/src/player/mod.rs
+309
View File
@@ -11,6 +11,9 @@
//! Supports both rodio (PipeWire/Pulse) and direct ALSA (hw: devices).
mod playback_engine;
mod streaming_source;
pub use streaming_source::{BufferedMediaSource, BufferWriter, StreamingConfig, IncrementalStreamingSource};
use std::io::{BufReader, Cursor, Read, Seek, SeekFrom};
use std::panic::{self, AssertUnwindSafe};
@@ -48,6 +51,15 @@ enum AudioCommand {
sample_rate: u32,
channels: u16,
},
/// Play from streaming source (BufferedMediaSource)
/// The download task should already be running and pushing to the source
PlayStreaming {
source: Arc<BufferedMediaSource>,
track_id: u64,
sample_rate: u32,
channels: u16,
duration_secs: u64,
},
/// Pause playback
Pause,
/// Resume playback
@@ -1046,6 +1058,216 @@ impl Player {
actual_duration
);
}
AudioCommand::PlayStreaming { source, track_id, sample_rate, channels, duration_secs } => {
log::info!(
"Audio thread: starting streaming playback for track {} ({}Hz, {} channels, {}s)",
track_id,
sample_rate,
channels,
duration_secs
);
*pause_suspend_deadline = None;
// Get DAC passthrough setting
let dac_passthrough = thread_settings
.lock()
.ok()
.map(|s| s.dac_passthrough)
.unwrap_or(false);
// Check if we need to recreate the stream
let format_changed = *current_sample_rate != Some(sample_rate)
|| *current_channels != Some(channels);
let using_alsa_direct = thread_settings
.lock()
.ok()
.and_then(|s| s.backend_type)
.map(|b| b == AudioBackendType::Alsa)
.unwrap_or(false);
let needs_new_stream = stream_opt.is_none()
|| (dac_passthrough && format_changed)
|| (using_alsa_direct && format_changed);
if needs_new_stream {
if stream_opt.is_some() {
if (dac_passthrough || using_alsa_direct) && format_changed {
let mode = if using_alsa_direct { "ALSA Direct" } else { "DAC passthrough" };
log::info!(
"Streaming: Sample rate/channels changed to {}Hz/{}ch - recreating OutputStream ({})",
sample_rate,
channels,
mode
);
}
drop(stream_opt.take());
}
let stream_result = if let Some(settings) = thread_settings.lock().ok() {
match try_init_stream_with_backend(&settings, sample_rate, channels) {
Some(result) => result,
None => {
log::info!("Backend system not configured, using legacy CPAL path");
let device = if let Some(ref name) = *current_device_name {
host.output_devices()
.ok()
.and_then(|mut devices| {
devices.find(|d| d.name().ok().as_ref() == Some(name))
})
.or_else(|| host.default_output_device())
} else {
host.default_output_device()
};
let Some(device) = device else {
log::error!("No audio output device available for streaming");
thread_state.set_stream_error(true);
return;
};
if let Ok(name) = device.name() {
thread_state.set_current_device(Some(name));
}
create_output_stream_with_config(
&device,
sample_rate,
channels,
dac_passthrough,
).map(|(stream, handle)| StreamType::Rodio(stream, handle))
}
}
} else {
let device = host.default_output_device();
let Some(device) = device else {
log::error!("No audio output device available for streaming");
thread_state.set_stream_error(true);
return;
};
create_output_stream_with_config(&device, sample_rate, channels, dac_passthrough)
.map(|(stream, handle)| StreamType::Rodio(stream, handle))
};
match stream_result {
Ok(stream) => {
*stream_opt = Some(stream);
*current_sample_rate = Some(sample_rate);
*current_channels = Some(channels);
thread_state.set_stream_error(false);
log::info!("✅ Streaming audio stream ready at {}Hz", sample_rate);
std::thread::sleep(Duration::from_millis(150));
}
Err(e) => {
log::error!("❌ Failed to create stream for streaming at {}Hz: {}", sample_rate, e);
thread_state.set_stream_error(true);
return;
}
}
}
let Some(ref stream) = *stream_opt else {
log::error!("Audio thread: no audio device available for streaming");
return;
};
// Stop previous engine
if let Some(engine) = current_engine.take() {
engine.stop();
std::thread::sleep(Duration::from_millis(50));
}
// Create PlaybackEngine
let mut engine = match stream {
StreamType::Rodio(_stream, handle) => {
match PlaybackEngine::new_rodio(handle) {
Ok(e) => {
*consecutive_sink_failures = 0;
thread_state.set_stream_error(false);
e
}
Err(e) => {
log::error!("Failed to create engine for streaming: {}", e);
return;
}
}
}
#[cfg(target_os = "linux")]
StreamType::AlsaDirect(alsa_stream) => {
let hardware_volume = thread_settings
.lock()
.ok()
.map(|s| s.alsa_hardware_volume)
.unwrap_or(false);
PlaybackEngine::new_alsa_direct(alsa_stream.clone(), hardware_volume)
}
};
let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0;
engine.set_volume(volume);
// Wait for minimum buffer before starting playback
log::info!("Streaming: waiting for initial buffer...");
let start_wait = Instant::now();
let max_wait = Duration::from_secs(30);
while !source.has_min_buffer() && start_wait.elapsed() < max_wait {
std::thread::sleep(Duration::from_millis(50));
}
if !source.has_min_buffer() {
log::error!("Streaming: timeout waiting for initial buffer");
return;
}
let buffer_wait_ms = start_wait.elapsed().as_millis();
log::info!(
"Streaming: initial buffer ready in {}ms, creating incremental decoder...",
buffer_wait_ms
);
// Create incremental streaming source - this starts playback IMMEDIATELY
// while continuing to decode/download in background
let incremental_source = match IncrementalStreamingSource::new(source.clone()) {
Ok(s) => s,
Err(e) => {
log::error!("Failed to create incremental streaming source: {}", e);
return;
}
};
// Verify sample rate/channels match what we expected
let actual_sr = incremental_source.get_sample_rate();
let actual_ch = incremental_source.get_channels();
if actual_sr != sample_rate || actual_ch != channels {
log::warn!(
"Streaming: detected format {}Hz/{}ch differs from expected {}Hz/{}ch",
actual_sr, actual_ch, sample_rate, channels
);
}
// Set duration from track metadata (passed from frontend)
// This allows the seekbar to show progress even during streaming
thread_state.duration.store(duration_secs, Ordering::SeqCst);
// Box the incremental source to match the expected type
let source_to_play: Box<dyn Source<Item = i16> + Send> = Box::new(incremental_source);
if let Err(e) = engine.append(source_to_play) {
log::error!("Failed to append streaming source to engine: {}", e);
return;
}
thread_state.is_playing.store(true, Ordering::SeqCst);
thread_state.position.store(0, Ordering::SeqCst);
thread_state.current_track_id.store(track_id, Ordering::SeqCst);
thread_state.start_playback_timer(0);
*current_engine = Some(engine);
log::info!(
"Audio thread: streaming playback STARTED in {}ms (incremental decode active)",
start_wait.elapsed().as_millis()
);
}
AudioCommand::Pause => {
if let Some(ref engine) = *current_engine {
engine.pause();
@@ -1438,6 +1660,93 @@ impl Player {
Ok(())
}
/// Play from streaming source (starts playback before full download)
/// Returns the BufferWriter so caller can push data as it downloads
pub fn play_streaming(
&self,
track_id: u64,
sample_rate: u32,
channels: u16,
content_length: u64,
buffer_seconds: u8,
duration_secs: u64,
) -> Result<BufferWriter, String> {
log::info!(
"Player: Starting streaming playback for track {} ({}Hz, {}ch, {} bytes total, {}s)",
track_id,
sample_rate,
channels,
content_length,
duration_secs
);
// Use StreamingConfig::from_seconds for proper buffer sizing
let config = StreamingConfig::from_seconds(buffer_seconds);
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: Streaming playback initiated");
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
async fn download_audio(&self, url: &str) -> Result<Vec<u8>, String> {
use std::time::Duration;
src-tauri/src/player/streaming_source.rs
+771
View File
@@ -0,0 +1,771 @@
//! Buffered media source for streaming playback.
//!
//! Provides two main components:
//! 1. `BufferedMediaSource` - Wraps an async HTTP response to provide a synchronous
//! `Read + Seek` interface required by symphonia decoders.
//! 2. `IncrementalStreamingSource` - A rodio Source that decodes audio packets
//! incrementally as they become available, allowing playback to start before
//! the entire file is downloaded.
//!
//! # Design
//!
//! The source uses a growing buffer that accumulates data from the HTTP response.
//! - Reads block if requesting data not yet buffered
//! - Seek forward blocks until data is available
//! - Seek backward works within buffered data
//! - Seek beyond current buffer position blocks until data arrives
//!
//! # Thread Safety
//!
//! The buffer state is shared between:
//! - The reader (audio thread, synchronous)
//! - The writer (download task, async)
//!
//! Communication uses `Mutex` + `Condvar` for blocking synchronization.
use std::collections::VecDeque;
use std::io::{Error as IoError, ErrorKind, Read, Result as IoResult, Seek, SeekFrom};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;
use rodio::Source;
use symphonia::core::audio::SampleBuffer;
use symphonia::core::codecs::{Decoder, DecoderOptions};
use symphonia::core::errors::Error as SymphoniaError;
use symphonia::core::formats::{FormatOptions, FormatReader};
use symphonia::core::io::{MediaSource, MediaSourceStream};
use symphonia::core::meta::MetadataOptions;
use symphonia::core::probe::Hint;
use symphonia::default::{get_codecs, get_probe};
/// Configuration for the streaming buffer
#[derive(Debug, Clone)]
pub struct StreamingConfig {
/// Minimum bytes to buffer before allowing reads (for format detection)
pub initial_buffer_bytes: usize,
/// Maximum buffer size before backpressure (not enforced, just for info)
pub max_buffer_bytes: usize,
}
impl Default for StreamingConfig {
fn default() -> Self {
Self {
// 512KB default - enough for format headers and ~2-5 seconds of audio
// This allows playback to start quickly while still having enough
// buffer to handle network jitter
initial_buffer_bytes: 512 * 1024,
// 100MB max buffer
max_buffer_bytes: 100 * 1024 * 1024,
}
}
}
impl StreamingConfig {
/// Create config from buffer seconds and approximate bitrate
///
/// For Hi-Res FLAC at 192kHz/24bit stereo, bitrate is roughly 9.2 Mbps
/// We estimate ~1MB per second as a conservative approximation
pub fn from_seconds(seconds: u8) -> Self {
// Minimum 256KB to ensure format detection works
let bytes = ((seconds as usize) * 1024 * 1024).max(256 * 1024);
Self {
initial_buffer_bytes: bytes,
max_buffer_bytes: 100 * 1024 * 1024,
}
}
/// Create a minimal config for fastest startup
/// Uses smallest buffer that still allows format detection (~256KB)
pub fn fast_start() -> Self {
Self {
initial_buffer_bytes: 256 * 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
struct BufferState {
/// Accumulated data from HTTP response
data: Vec<u8>,
/// True when HTTP download is complete
download_complete: bool,
/// Error from download, if any
download_error: Option<String>,
/// Total expected size (from Content-Length), if known
total_size: Option<u64>,
}
/// A media source that buffers from an async HTTP stream.
///
/// Provides `Read + Seek` interface for decoders while data is still downloading.
/// The source is created with a `BufferWriter` that receives chunks from the
/// download task.
pub struct BufferedMediaSource {
state: Arc<(Mutex<BufferState>, Condvar)>,
config: StreamingConfig,
/// Each reader has its own read position
read_pos: std::sync::atomic::AtomicU64,
}
impl BufferedMediaSource {
/// Create a new buffered source.
///
/// Returns the source and a writer for pushing downloaded chunks.
/// The writer should be used from the async download task.
pub fn new(config: StreamingConfig, total_size: Option<u64>) -> (Self, BufferWriter) {
let state = Arc::new((
Mutex::new(BufferState {
data: Vec::with_capacity(config.initial_buffer_bytes),
download_complete: false,
download_error: None,
total_size,
}),
Condvar::new(),
));
let source = Self {
state: Arc::clone(&state),
config: config.clone(),
read_pos: std::sync::atomic::AtomicU64::new(0),
};
let writer = BufferWriter { state };
(source, writer)
}
/// Create a new reader that shares the same buffer but has its own read position.
/// This is used to pass to symphonia which needs ownership of the reader.
pub fn create_reader(&self) -> Self {
Self {
state: Arc::clone(&self.state),
config: self.config.clone(),
read_pos: std::sync::atomic::AtomicU64::new(0),
}
}
/// Wait until initial buffer is filled or download completes.
///
/// This should be called before passing the source to the decoder,
/// to ensure enough data is available for format detection.
///
/// Returns error if download fails before initial buffer is filled.
pub fn wait_for_initial_buffer(&self) -> IoResult<()> {
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| {
IoError::new(ErrorKind::Other, "Failed to acquire buffer lock")
})?;
while state.data.len() < self.config.initial_buffer_bytes
&& !state.download_complete
&& state.download_error.is_none()
{
state = cvar.wait(state).map_err(|_| {
IoError::new(ErrorKind::Other, "Condition variable wait failed")
})?;
}
if let Some(ref err) = state.download_error {
return Err(IoError::new(ErrorKind::Other, err.clone()));
}
Ok(())
}
/// Check if download is complete (full file in buffer)
pub fn is_complete(&self) -> bool {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.download_complete && state.download_error.is_none()
} else {
false
}
}
/// Get current buffer size in bytes
pub fn buffer_size(&self) -> usize {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.data.len()
} else {
0
}
}
/// Get the complete data if download finished successfully.
///
/// Used to store in cache after streaming playback completes.
/// Returns None if download is not complete or failed.
pub fn take_complete_data(&self) -> Option<Vec<u8>> {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
if state.download_complete && state.download_error.is_none() {
Some(state.data.clone())
} else {
None
}
} else {
None
}
}
/// Get download progress as a fraction (0.0 to 1.0)
///
/// Returns None if total size is unknown
pub fn progress(&self) -> Option<f32> {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.total_size.map(|total| {
if total == 0 {
1.0
} else {
state.data.len() as f32 / total as f32
}
})
} else {
None
}
}
/// Check if minimum buffer for playback is available
///
/// Returns true when initial_buffer_bytes have been buffered
/// or the download is complete.
pub fn has_min_buffer(&self) -> bool {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.data.len() >= self.config.initial_buffer_bytes || state.download_complete
} else {
false
}
}
}
impl Read for BufferedMediaSource {
fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
use std::sync::atomic::Ordering;
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| {
IoError::new(ErrorKind::Other, "Failed to acquire buffer lock")
})?;
let read_pos = self.read_pos.load(Ordering::SeqCst) as usize;
// Wait for data if we're ahead of buffer
while read_pos >= state.data.len()
&& !state.download_complete
&& state.download_error.is_none()
{
state = cvar.wait(state).map_err(|_| {
IoError::new(ErrorKind::Other, "Condition variable wait failed")
})?;
}
// Check for errors
if let Some(ref err) = state.download_error {
return Err(IoError::new(ErrorKind::Other, err.clone()));
}
// EOF if at end and download complete
if read_pos >= state.data.len() && state.download_complete {
return Ok(0);
}
// Read available data
let available = state.data.len() - read_pos;
let to_read = buf.len().min(available);
buf[..to_read].copy_from_slice(&state.data[read_pos..read_pos + to_read]);
self.read_pos.store((read_pos + to_read) as u64, Ordering::SeqCst);
Ok(to_read)
}
}
impl Seek for BufferedMediaSource {
fn seek(&mut self, pos: SeekFrom) -> IoResult<u64> {
use std::sync::atomic::Ordering;
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| {
IoError::new(ErrorKind::Other, "Failed to acquire buffer lock")
})?;
let current_pos = self.read_pos.load(Ordering::SeqCst) as i64;
let new_pos = match pos {
SeekFrom::Start(offset) => offset as i64,
SeekFrom::Current(offset) => current_pos + offset,
SeekFrom::End(offset) => {
// For End seeks, we need to know total size or have complete download
if let Some(total) = state.total_size {
total as i64 + offset
} else if state.download_complete {
state.data.len() as i64 + offset
} else {
// Can't seek from end without knowing size
return Err(IoError::new(
ErrorKind::Unsupported,
"Cannot seek from end while streaming without known size",
));
}
}
};
if new_pos < 0 {
return Err(IoError::new(
ErrorKind::InvalidInput,
"Seek position before start of stream",
));
}
let new_pos_usize = new_pos as usize;
// If seeking forward beyond buffer, wait for data
while new_pos_usize > state.data.len()
&& !state.download_complete
&& state.download_error.is_none()
{
state = cvar.wait(state).map_err(|_| {
IoError::new(ErrorKind::Other, "Condition variable wait failed")
})?;
}
if let Some(ref err) = state.download_error {
return Err(IoError::new(ErrorKind::Other, err.clone()));
}
// After download complete, check bounds
if state.download_complete && new_pos_usize > state.data.len() {
return Err(IoError::new(
ErrorKind::InvalidInput,
"Seek position beyond end of stream",
));
}
self.read_pos.store(new_pos as u64, Ordering::SeqCst);
Ok(new_pos as u64)
}
}
// Required for symphonia MediaSource trait
impl MediaSource for BufferedMediaSource {
fn is_seekable(&self) -> bool {
// We support seeking within buffered data
true
}
fn byte_len(&self) -> Option<u64> {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.total_size
} else {
None
}
}
}
/// Writer half for pushing downloaded chunks from the async download task.
///
/// This is the sender side that receives data from the HTTP response
/// and makes it available to the `BufferedMediaSource` reader.
#[derive(Clone)]
pub struct BufferWriter {
state: Arc<(Mutex<BufferState>, Condvar)>,
}
impl BufferWriter {
/// Push a chunk of downloaded data
///
/// This wakes up any readers waiting for data.
pub fn push_chunk(&self, chunk: &[u8]) -> Result<(), String> {
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| "Failed to acquire buffer lock")?;
state.data.extend_from_slice(chunk);
cvar.notify_all();
Ok(())
}
/// Mark download as complete
///
/// After this is called, readers will receive EOF after reading all buffered data.
pub fn complete(&self) -> Result<(), String> {
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| "Failed to acquire buffer lock")?;
state.download_complete = true;
cvar.notify_all();
Ok(())
}
/// Mark download as failed
///
/// After this is called, readers will receive the error on next read.
pub fn error(&self, err: String) -> Result<(), String> {
let (lock, cvar) = &*self.state;
let mut state = lock.lock().map_err(|_| "Failed to acquire buffer lock")?;
state.download_error = Some(err);
cvar.notify_all();
Ok(())
}
/// Get current buffer size in bytes
pub fn buffer_size(&self) -> usize {
let (lock, _) = &*self.state;
if let Ok(state) = lock.lock() {
state.data.len()
} else {
0
}
}
}
// =============================================================================
// IncrementalStreamingSource - A rodio Source that decodes on-demand
// =============================================================================
/// A rodio Source that decodes audio packets incrementally from a BufferedMediaSource.
///
/// This allows playback to start immediately after the initial buffer is filled,
/// while the rest of the file continues downloading in the background.
///
/// The source maintains an internal queue of decoded samples and decodes more
/// packets on-demand as samples are consumed.
pub struct IncrementalStreamingSource {
/// Sample rate of the audio
sample_rate: u32,
/// Number of channels
channels: u16,
/// Queue of decoded samples ready to play
sample_queue: VecDeque<i16>,
/// The format reader (demuxer)
format: Box<dyn FormatReader>,
/// The audio decoder
decoder: Box<dyn Decoder>,
/// Track ID we're decoding
track_id: u32,
/// Whether we've reached end of stream
finished: bool,
/// Number of packets decoded (for stats)
packets_decoded: u64,
/// Reference to the buffered source (for cache retrieval after playback)
buffered_source: Arc<BufferedMediaSource>,
}
impl IncrementalStreamingSource {
/// Create a new incremental streaming source.
///
/// This initializes the symphonia decoder and prepares for incremental decoding.
/// The BufferedMediaSource should already have its initial buffer filled.
///
/// Returns the source along with detected sample_rate and channels.
pub fn new(
buffered_source: Arc<BufferedMediaSource>,
) -> Result<Self, String> {
// Create a reader from the buffered source
let reader = buffered_source.create_reader();
let media_source = Box::new(reader) as Box<dyn MediaSource>;
let mss = MediaSourceStream::new(media_source, Default::default());
let mut hint = Hint::new();
hint.with_extension("flac"); // Most Qobuz Hi-Res is FLAC
let format_opts = FormatOptions {
enable_gapless: true,
..Default::default()
};
let metadata_opts: MetadataOptions = Default::default();
let probed = get_probe()
.format(&hint, mss, &format_opts, &metadata_opts)
.map_err(|err| format!("Symphonia probe failed for streaming: {}", err))?;
let track = probed
.format
.default_track()
.ok_or_else(|| "Symphonia: no supported audio tracks in stream".to_string())?;
let track_id = track.id;
let codec_params = track.codec_params.clone();
// Extract sample rate and channels from codec params
let sample_rate = codec_params
.sample_rate
.ok_or_else(|| "No sample rate in codec params".to_string())?;
let channels = codec_params
.channels
.map(|c| c.count() as u16)
.unwrap_or(2);
let decoder = get_codecs()
.make(&codec_params, &DecoderOptions::default())
.map_err(|err| format!("Symphonia decoder init failed for streaming: {}", err))?;
log::info!(
"IncrementalStreamingSource initialized: {}Hz, {} channels",
sample_rate,
channels
);
Ok(Self {
sample_rate,
channels,
sample_queue: VecDeque::with_capacity(sample_rate as usize * channels as usize), // ~1s buffer
format: probed.format,
decoder,
track_id,
finished: false,
packets_decoded: 0,
buffered_source,
})
}
/// Get the sample rate
pub fn get_sample_rate(&self) -> u32 {
self.sample_rate
}
/// Get the number of channels
pub fn get_channels(&self) -> u16 {
self.channels
}
/// Get reference to the buffered source for cache retrieval
pub fn buffered_source(&self) -> &Arc<BufferedMediaSource> {
&self.buffered_source
}
/// Decode more packets to fill the sample queue.
///
/// This is called when the sample queue is running low.
/// It will decode packets until the queue has at least `min_samples` or EOF is reached.
fn decode_more(&mut self, min_samples: usize) {
if self.finished {
return;
}
while self.sample_queue.len() < min_samples {
let packet = match self.format.next_packet() {
Ok(packet) => packet,
Err(SymphoniaError::IoError(ref e)) if e.kind() == std::io::ErrorKind::WouldBlock => {
// Not enough data buffered yet - wait briefly and retry
// This happens when playback catches up with download
std::thread::sleep(Duration::from_millis(5));
continue;
}
Err(SymphoniaError::IoError(_)) => {
// EOF or other IO error
log::info!(
"IncrementalStreamingSource: EOF reached after {} packets",
self.packets_decoded
);
self.finished = true;
return;
}
Err(err) => {
log::error!("Symphonia read error in stream: {}", err);
self.finished = true;
return;
}
};
if packet.track_id() != self.track_id {
continue;
}
match self.decoder.decode(&packet) {
Ok(audio_buf) => {
let spec = *audio_buf.spec();
let mut sample_buf = SampleBuffer::<i16>::new(audio_buf.frames() as u64, spec);
sample_buf.copy_interleaved_ref(audio_buf);
// Add samples to queue
self.sample_queue.extend(sample_buf.samples().iter().copied());
self.packets_decoded += 1;
}
Err(SymphoniaError::DecodeError(e)) => {
log::warn!("Decode error (skipping packet): {}", e);
continue;
}
Err(SymphoniaError::ResetRequired) => {
self.decoder.reset();
continue;
}
Err(err) => {
log::error!("Symphonia decode error: {}", err);
self.finished = true;
return;
}
}
}
}
}
impl Source for IncrementalStreamingSource {
fn current_frame_len(&self) -> Option<usize> {
// We don't know frame boundaries in the queue
None
}
fn channels(&self) -> u16 {
self.channels
}
fn sample_rate(&self) -> u32 {
self.sample_rate
}
fn total_duration(&self) -> Option<Duration> {
// We don't know total duration until download completes
// Could estimate from content-length if available
None
}
}
impl Iterator for IncrementalStreamingSource {
type Item = i16;
fn next(&mut self) -> Option<Self::Item> {
// If queue is running low, decode more
// Keep at least 0.5 seconds of audio buffered
let min_buffer = (self.sample_rate as usize * self.channels as usize) / 2;
if self.sample_queue.len() < min_buffer {
self.decode_more(min_buffer);
}
self.sample_queue.pop_front()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread;
use std::time::Duration;
#[test]
fn test_basic_read_write() {
let config = StreamingConfig {
initial_buffer_bytes: 10,
max_buffer_bytes: 100,
};
let (mut source, writer) = BufferedMediaSource::new(config, Some(20));
// Write some data
writer.push_chunk(b"Hello").unwrap();
writer.push_chunk(b"World").unwrap();
// Read it back
let mut buf = [0u8; 5];
assert_eq!(source.read(&mut buf).unwrap(), 5);
assert_eq!(&buf, b"Hello");
assert_eq!(source.read(&mut buf).unwrap(), 5);
assert_eq!(&buf, b"World");
}
#[test]
fn test_seek_within_buffer() {
let config = StreamingConfig {
initial_buffer_bytes: 5,
max_buffer_bytes: 100,
};
let (mut source, writer) = BufferedMediaSource::new(config, Some(10));
writer.push_chunk(b"0123456789").unwrap();
writer.complete().unwrap();
// Read first 5 bytes
let mut buf = [0u8; 5];
source.read(&mut buf).unwrap();
assert_eq!(&buf, b"01234");
// Seek back to start
source.seek(SeekFrom::Start(0)).unwrap();
source.read(&mut buf).unwrap();
assert_eq!(&buf, b"01234");
// Seek to middle
source.seek(SeekFrom::Start(3)).unwrap();
source.read(&mut buf).unwrap();
assert_eq!(&buf, b"34567");
}
#[test]
fn test_complete_data_retrieval() {
let config = StreamingConfig {
initial_buffer_bytes: 5,
max_buffer_bytes: 100,
};
let (source, writer) = BufferedMediaSource::new(config, Some(10));
writer.push_chunk(b"Hello").unwrap();
assert!(source.take_complete_data().is_none()); // Not complete yet
writer.push_chunk(b"World").unwrap();
writer.complete().unwrap();
let data = source.take_complete_data().unwrap();
assert_eq!(&data, b"HelloWorld");
}
#[test]
fn test_blocking_read() {
let config = StreamingConfig {
initial_buffer_bytes: 5,
max_buffer_bytes: 100,
};
let (mut source, writer) = BufferedMediaSource::new(config, None);
// Spawn thread to write after delay
thread::spawn(move || {
thread::sleep(Duration::from_millis(50));
writer.push_chunk(b"Delayed").unwrap();
writer.complete().unwrap();
});
// This should block until data arrives
let mut buf = [0u8; 7];
let n = source.read(&mut buf).unwrap();
assert_eq!(n, 7);
assert_eq!(&buf, b"Delayed");
}
}
src/lib/components/views/SettingsView.svelte
+87
View File
@@ -248,6 +248,8 @@
let selectedBackend = $state<string>('Auto');
let selectedAlsaPlugin = $state<string>('hw (Direct Hardware)');
let alsaHardwareVolume = $state(false);
let streamFirstTrack = $state(false);
let streamBufferSeconds = $state(3);
// Backend system state
let availableBackends = $state<BackendInfo[]>([]);
@@ -886,6 +888,8 @@
backend_type: 'PipeWire' | 'Alsa' | 'Pulse' | null;
alsa_plugin: 'Hw' | 'PlugHw' | 'Pcm' | null;
alsa_hardware_volume: boolean;
stream_first_track: boolean;
stream_buffer_seconds: number;
}
interface BackendInfo {
@@ -1031,6 +1035,8 @@
}
alsaHardwareVolume = settings.alsa_hardware_volume ?? false;
streamFirstTrack = settings.stream_first_track ?? false;
streamBufferSeconds = settings.stream_buffer_seconds ?? 3;
// Validate mutual exclusion: DAC Passthrough disables Gapless + Crossfade
if (dacPassthrough) {
@@ -1197,6 +1203,28 @@
}
}
async function handleStreamFirstTrackChange(enabled: boolean) {
streamFirstTrack = enabled;
try {
await invoke('set_audio_stream_first_track', { enabled });
console.log('[Audio] Stream first track changed:', enabled);
} catch (err) {
console.error('[Audio] Failed to change stream first track:', err);
}
}
async function handleStreamBufferSecondsChange(seconds: number) {
// Clamp to valid range
const clamped = Math.max(1, Math.min(10, Math.round(seconds)));
streamBufferSeconds = clamped;
try {
await invoke('set_audio_stream_buffer_seconds', { seconds: clamped });
console.log('[Audio] Stream buffer seconds changed:', clamped);
} catch (err) {
console.error('[Audio] Failed to change stream buffer seconds:', err);
}
}
async function handleBackendDeviceChange(deviceName: string) {
outputDevice = deviceName;
@@ -1651,6 +1679,30 @@
</div>
</div>
{/if}
<div class="setting-row">
<div class="setting-info">
<span class="setting-label">Stream Uncached Tracks</span>
<span class="setting-desc">Start playback faster when track is not in cache. Seeking may be limited during initial buffering.</span>
</div>
<Toggle enabled={streamFirstTrack} onchange={handleStreamFirstTrackChange} />
</div>
{#if streamFirstTrack}
<div class="setting-row">
<div class="setting-info">
<span class="setting-label">Initial Buffer Size</span>
<span class="setting-desc">Seconds to buffer before starting playback ({streamBufferSeconds}s)</span>
</div>
<input
type="range"
min="1"
max="10"
step="1"
value={streamBufferSeconds}
oninput={(e) => handleStreamBufferSecondsChange(parseInt(e.currentTarget.value))}
class="buffer-slider"
/>
</div>
{/if}
<div class="setting-row last">
<span class="setting-label">{$t('settings.audio.currentSampleRate')}</span>
<span class="setting-value" class:muted={!hardwareStatus?.is_active}>
@@ -2896,6 +2948,41 @@ flatpak override --user --filesystem=$HOME/music-nas com.blitzfc.qbz</pre>
color: rgb(251, 191, 36);
font-weight: 600;
}
/* Buffer slider styling */
.buffer-slider {
width: 120px;
height: 4px;
-webkit-appearance: none;
appearance: none;
background: var(--alpha-20);
border-radius: 2px;
cursor: pointer;
}
.buffer-slider::-webkit-slider-thumb {
-webkit-appearance: none;
appearance: none;
width: 14px;
height: 14px;
background: var(--accent-color, #3b82f6);
border-radius: 50%;
cursor: pointer;
transition: transform 0.1s ease;
}
.buffer-slider::-webkit-slider-thumb:hover {
transform: scale(1.1);
}
.buffer-slider::-moz-range-thumb {
width: 14px;
height: 14px;
background: var(--accent-color, #3b82f6);
border-radius: 50%;
cursor: pointer;
border: none;
}
</style>
<MigrationModal
src/lib/services/playbackService.ts
+1 -1
View File
@@ -114,7 +114,7 @@ export async function playTrack(
if (isLocal) {
await invoke('library_play_track', { trackId: track.id });
} else {
await invoke('play_track', { trackId: track.id });
await invoke('play_track', { trackId: track.id, durationSecs: track.duration });
}
}
src/lib/stores/playerStore.ts
+2 -2
View File
@@ -245,8 +245,8 @@ export async function togglePlay(): Promise<void> {
console.log('[Player] Restoring local track:', localTrackId);
await invoke('library_play_track', { trackId: localTrackId });
} else {
// Qobuz track - use play_track
await invoke('play_track', { trackId: currentTrack.id });
// Qobuz track - use play_track with duration for seekbar
await invoke('play_track', { trackId: currentTrack.id, durationSecs: currentTrack.duration });
}
// Seek to saved position after a short delay to let audio load