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

feat(slint): Plex progressive streaming (fast play) + fix re-sync wiping hydrated quality

Browse Source 
Two Plex-port fixes (both in the Plex play/data path):

1. Progressive streaming — play starts in ~1s instead of ~10s. play_plex_audible
   no longer downloads the whole file before play_data; it resolves only the
   part URL (new additive qbz_plex::plex_resolve_part_url, no body GET) and feeds
   it into the player's streaming source (play_streaming_dynamic) via a shared
   feeder module (remote_stream.rs, extracted from the QConnect renderer's HTTP
   feeder — both now share it). Bit-perfect preserved: the original /file bytes
   are streamed (no transcode), device init derives from the decoded stream; the
   protected audio backend is untouched. Cast/DLNA keep the whole-file path.
   Falls back to whole-file play_data on a non-direct part / streaming-setup error.

2. Re-sync no longer wipes hydrated quality — the sync did plex_cache_clear()
   (full wipe) BEFORE the per-section save_tracks, defeating save_tracks' COALESCE
   hydration carry-over, so already-hydrated albums reverted to a bare FLAC badge
   on every settings-open / background-refresh / section-toggle. Replaced the full
   wipe with a new additive plex_cache_prune_sections(keep) that removes only the
   DE-selected sections, keeping the selected sections (and their hydration);
   save_tracks then re-applies its per-section carry-over.

cc
This commit is contained in:
vicrodh committed 2026-06-08 14:39:05 -06:00
1 parent ed7891c697
commit b37c29af14
6 files changed
+457 -242
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/qbz-slint/src/main.rs
+1
View File
@@ -44,6 +44,7 @@ mod qconnect_event_sink;
mod qconnect_service;
mod qconnect_transport;
mod queue;
mod remote_stream;
mod drag;
mod ephemeral;
mod folders;
crates/qbz-slint/src/playback.rs
+87 -14
View File
@@ -245,7 +245,8 @@ async fn play_audible(runtime: &Runtime, weak: &slint::Weak<AppWindow>, track_id
.source_item_id_hint
.clone()
.unwrap_or_else(|| track_id.to_string());
play_plex_audible(runtime, weak, track_id, rating_key).await;
play_plex_audible(runtime, weak, track_id, rating_key, qt.duration_secs)
.await;
return;
}
_ => {}
@@ -346,19 +347,30 @@ async fn play_local_file_audible(
}
}
/// Audible step for a Plex track: resolve the original part bytes from the
/// user's LAN Plex server and hand them to the player's `play_data` seam —
/// exactly like the local-file path, but the bytes come from the network
/// instead of disk. `play_data` extracts the sample rate and drives the
/// PROTECTED device init from the DECODED bytes (bit-perfect), so the Plex
/// Audible step for a Plex track: PROGRESSIVE STREAMING.
///
/// Resolves ONLY the direct-play part URL (no body download — that was the
/// ~10s stall) and feeds it into the player's progressive streaming sink via
/// the shared `remote_stream` feeder (the same one QConnect uses). Playback
/// starts as soon as the initial buffer fills (~1s), not after the whole FLAC
/// lands. The feeder decodes the same original bytes and drives the PROTECTED
/// device init from the DECODED stream (bit-perfect), so the Plex
/// `sampling_rate_hz`/`bit_depth` are display-only and never touched here.
///
/// On any streaming-setup failure (resolve / probe / sink open) it falls back
/// to the old whole-file `plex_resolve_track_media` + `play_data` so a server
/// that breaks streaming still plays. The loading spinner is cleared only on
/// the hard-error paths; the poll loop clears it on the first decoded-audio
/// edge (same as Qobuz / QConnect).
///
/// `play_id` is the queue id (numeric rating key); `rating_key` is the string
/// key the resolve needs.
/// key the resolve needs; `duration_secs` comes from the queue track.
async fn play_plex_audible(
runtime: &Runtime,
weak: &slint::Weak<AppWindow>,
play_id: u64,
rating_key: String,
duration_secs: u64,
) {
let cfg = crate::plex_settings::get();
if cfg.base_url.is_empty() || cfg.token.is_empty() {
@@ -366,23 +378,84 @@ async fn play_plex_audible(
clear_loading(weak, play_id);
return;
}
let resolved = match qbz_plex::plex_resolve_track_media(
cfg.base_url,
cfg.token,
// 1. Resolve JUST the direct-play part URL — no body download.
let loc = match qbz_plex::plex_resolve_part_url(
cfg.base_url.clone(),
cfg.token.clone(),
rating_key.clone(),
)
.await
{
Ok(r) => r,
Ok(l) => l,
Err(e) => {
log::error!("[qbz-slint] plex play: resolve {rating_key} failed: {e}");
clear_loading(weak, play_id);
return;
}
};
if let Err(e) = runtime.core().player().play_data(resolved.bytes, play_id) {
log::error!("[qbz-slint] plex play: play_data {play_id} failed: {e}");
clear_loading(weak, play_id);
if !loc.direct_play_confirmed {
// Not a direct `/library/parts/.../file` part: the server may force a
// transcode and the streamed bytes would not be bit-perfect. Fall back
// to the whole-file resolve so the track still plays.
log::warn!(
"[qbz-slint] plex play: {rating_key} part is not a direct /file part ({}); \
full-download fallback",
loc.part_key
);
plex_full_download_fallback(runtime, weak, play_id, rating_key).await;
return;
}
// 2. Stream the part URL progressively (same feeder QConnect uses). On
// setup failure (probe / sink open), fall back to whole-file download.
let player = runtime.core().player();
match crate::remote_stream::stream_remote_track_into_player(
&player,
play_id,
duration_secs,
0, // Plex callers don't resume mid-track; start at 0 (like QConnect).
&loc.part_url,
"Plex",
)
.await
{
Ok(()) => {
// Buffering started — the poll loop clears the spinner on the first
// decoded-audio edge. Nothing else to do here.
}
Err(e) => {
log::warn!(
"[qbz-slint] plex play: streaming setup for {play_id} failed ({e}); \
full-download fallback"
);
plex_full_download_fallback(runtime, weak, play_id, rating_key).await;
}
}
}
/// Whole-file Plex fallback: resolve + download the entire part body and hand
/// it to `play_data`. Slow (the original ~10s path), but keeps a track playable
/// when progressive streaming setup fails or the part is not direct-play.
async fn plex_full_download_fallback(
runtime: &Runtime,
weak: &slint::Weak<AppWindow>,
play_id: u64,
rating_key: String,
) {
let cfg = crate::plex_settings::get();
match qbz_plex::plex_resolve_track_media(cfg.base_url, cfg.token, rating_key.clone()).await {
Ok(r) => {
if let Err(e) = runtime.core().player().play_data(r.bytes, play_id) {
log::error!("[qbz-slint] plex play: fallback play_data {play_id} failed: {e}");
clear_loading(weak, play_id);
}
}
Err(e) => {
log::error!("[qbz-slint] plex play: fallback resolve {rating_key} failed: {e}");
clear_loading(weak, play_id);
}
}
}
crates/qbz-slint/src/plex_auth.rs
+9 -2
View File
@@ -757,12 +757,19 @@ async fn sync_selected_libraries(
let server_id = (!machine_id.is_empty()).then_some(machine_id.clone());
let selected = plex_settings::get().selected_section_keys;
// cache_clear + re-save sections (blocking).
// Prune ONLY the de-selected sections + re-save sections (blocking). Do
// NOT cache_clear here: a full wipe deletes the hydrated rows BEFORE the
// per-section save_tracks can carry their quality over, so already-hydrated
// albums would revert to a bare FLAC badge on every re-sync (settings open /
// background refresh / section toggle). Pruning keeps the selected sections
// (and their hydration); save_tracks below re-applies the per-section
// carry-over.
{
let sections = sections.clone();
let server_id = server_id.clone();
let keep = selected.clone();
let _ = tokio::task::spawn_blocking(move || {
let _ = qbz_plex::plex_cache_clear();
let _ = qbz_plex::plex_cache_prune_sections(&keep);
if !sections.is_empty() {
let _ = qbz_plex::plex_cache_save_sections(server_id, sections);
}
crates/qbz-slint/src/qconnect_engine.rs
+5 -226
View File
@@ -22,13 +22,12 @@
#![allow(dead_code)]
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use qbz_app::shell::AppRuntime;
use qbz_core::QbzCore;
use qbz_models::{Quality, QueueTrack, RepeatMode, Track};
use qbz_player::{BufferWriter, PlaybackState};
use qbz_player::PlaybackState;
use qconnect_app::QconnectRendererEngine;
use crate::adapter::SlintAdapter;
@@ -138,12 +137,14 @@ impl QconnectRendererEngine for SlintRendererEngine {
.await
.map_err(|err| format!("resolve stream url for remote track {track_id}: {err}"))?;
match stream_remote_track_into_player(
self.core(),
let player = self.core().player();
match crate::remote_stream::stream_remote_track_into_player(
&player,
track_id,
duration_secs,
start_position_secs,
&stream_url.url,
"QConnect",
)
.await
{
@@ -173,228 +174,6 @@ impl QconnectRendererEngine for SlintRendererEngine {
}
}
// ===================== HTTP streaming feeder =====================
//
// reqwest + BufferWriter bound, so it stays frontend-side and never crosses the
// qconnect-app boundary. Verbatim port of the Tauri `track_loading.rs` feeder.
struct QconnectRemoteStreamInfo {
content_length: u64,
sample_rate: u32,
channels: u16,
bit_depth: u32,
speed_mbps: f64,
}
async fn stream_remote_track_into_player(
core: &QbzCore<SlintAdapter>,
track_id: u64,
duration_secs: u64,
start_position_secs: u64,
url: &str,
) -> Result<(), String> {
let stream_info = probe_remote_stream_info(url).await?;
log::info!(
"[QConnect/STREAMING] Track {} - {:.2} MB, {}Hz, {} ch, {}-bit, {:.1} MB/s",
track_id,
stream_info.content_length as f64 / (1024.0 * 1024.0),
stream_info.sample_rate,
stream_info.channels,
stream_info.bit_depth,
stream_info.speed_mbps
);
let writer = core
.player()
.play_streaming_dynamic(
track_id,
stream_info.sample_rate,
stream_info.channels,
stream_info.bit_depth,
stream_info.content_length,
stream_info.speed_mbps,
duration_secs,
start_position_secs, // QConnect callers pass 0; resume is local-only
)
.map_err(|err| format!("start streaming remote track {track_id}: {err}"))?;
let url = url.to_string();
let content_length = stream_info.content_length;
tokio::spawn(async move {
if let Err(err) =
download_and_stream_remote_track(&url, writer, track_id, content_length).await
{
log::error!(
"[QConnect/STREAMING] Track {} failed while streaming: {}",
track_id,
err
);
}
});
Ok(())
}
async fn probe_remote_stream_info(url: &str) -> Result<QconnectRemoteStreamInfo, String> {
use std::time::Instant;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|err| format!("create stream probe client: {err}"))?;
let head_response = client
.head(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("probe HEAD request failed: {err}"))?;
if !head_response.status().is_success() {
return Err(format!(
"probe HEAD request failed with status {}",
head_response.status()
));
}
let content_length = head_response
.headers()
.get("content-length")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<u64>().ok())
.ok_or_else(|| "probe missing content-length header".to_string())?;
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(|err| format!("probe range request failed: {err}"))?;
if !range_response.status().is_success() {
return Err(format!(
"probe range request failed with status {}",
range_response.status()
));
}
let initial_bytes = range_response
.bytes()
.await
.map_err(|err| format!("read probe bytes failed: {err}"))?;
let elapsed = start_time.elapsed();
let speed_mbps = if elapsed.as_secs_f64() > 0.0 {
(initial_bytes.len() as f64 / elapsed.as_secs_f64()) / (1024.0 * 1024.0)
} else {
10.0
};
let (sample_rate, channels, bit_depth) =
if initial_bytes.len() >= 26 && initial_bytes.starts_with(b"fLaC") {
let sample_rate = ((initial_bytes[18] as u32) << 12)
| ((initial_bytes[19] as u32) << 4)
| ((initial_bytes[20] as u32) >> 4);
let channels = ((initial_bytes[20] >> 1) & 0x07) + 1;
let bit_depth = ((initial_bytes[20] & 0x01) << 4) | ((initial_bytes[21] >> 4) & 0x0F);
(sample_rate, channels as u16, (bit_depth + 1) as u32)
} else {
log::warn!("[QConnect/STREAMING] Non-FLAC probe for remote handoff, using defaults");
(44_100, 2, 16)
};
Ok(QconnectRemoteStreamInfo {
content_length,
sample_rate,
channels,
bit_depth,
speed_mbps,
})
}
async fn download_and_stream_remote_track(
url: &str,
writer: BufferWriter,
track_id: u64,
content_length: u64,
) -> Result<(), String> {
use futures_util::StreamExt;
use std::time::Instant;
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(300))
.build()
.map_err(|err| format!("create remote streaming client: {err}"))?;
let response = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("start remote streaming request failed: {err}"))?;
if !response.status().is_success() {
return Err(format!(
"remote streaming request failed with status {}",
response.status()
));
}
let mut bytes_received = 0u64;
let mut stream = response.bytes_stream();
let start_time = Instant::now();
let mut last_log_time = Instant::now();
while let Some(chunk_result) = stream.next().await {
let chunk = chunk_result.map_err(|err| format!("remote streaming chunk failed: {err}"))?;
bytes_received += chunk.len() as u64;
if let Err(err) = writer.push_chunk(&chunk) {
log::error!(
"[QConnect/STREAMING] Failed to push chunk for track {}: {}",
track_id,
err
);
}
let now = Instant::now();
if now.duration_since(last_log_time) >= Duration::from_secs(2) && content_length > 0 {
let progress = (bytes_received as f64 / content_length as f64) * 100.0;
let avg_speed =
(bytes_received as f64 / start_time.elapsed().as_secs_f64()) / (1024.0 * 1024.0);
log::info!(
"[QConnect/STREAMING] Track {} {:.1}% ({:.2}/{:.2} MB) @ {:.2} MB/s",
track_id,
progress,
bytes_received as f64 / (1024.0 * 1024.0),
content_length as f64 / (1024.0 * 1024.0),
avg_speed
);
last_log_time = now;
}
}
if let Err(err) = writer.complete() {
log::error!(
"[QConnect/STREAMING] Failed to mark stream complete for track {}: {}",
track_id,
err
);
}
log::info!(
"[QConnect/STREAMING] Track {} complete: {:.2} MB in {:.1}s",
track_id,
bytes_received as f64 / (1024.0 * 1024.0),
start_time.elapsed().as_secs_f64()
);
Ok(())
}
async fn download_remote_audio(url: &str) -> Result<Vec<u8>, String> {
let response = reqwest::Client::new()
crates/qbz-slint/src/remote_stream.rs
+262
View File
@@ -0,0 +1,262 @@
//! Shared HTTP streaming feeder.
//!
//! Ports the Tauri `track_loading.rs` progressive feeder verbatim: probe a
//! remote audio URL for size + FLAC format, open the player's progressive
//! streaming sink (`Player::play_streaming_dynamic`), then push the body to the
//! returned `BufferWriter` chunk-by-chunk as it arrives. Playback starts as soon
//! as the initial buffer fills — not after the whole file lands.
//!
//! `reqwest + BufferWriter` bound only, so it stays frontend-side and never
//! crosses the qconnect-app boundary. Used by BOTH the QConnect renderer
//! (`qconnect_engine.rs`) and the Plex playback path (`playback.rs`), so there
//! is exactly one feeder.
//!
//! BIT-PERFECT: `play_streaming_dynamic` decodes the same original bytes and
//! drives the PROTECTED device init from the decoded stream. The
//! `sample_rate`/`bit_depth` parsed here are only the streaming-config hints;
//! the audio backend (`pipewire_backend.rs`, `init_device`, `audio_settings.rs`)
//! is untouched.
use std::time::Duration;
use qbz_player::{BufferWriter, Player};
/// Format/size facts sniffed from a remote audio URL before streaming.
pub struct RemoteStreamInfo {
pub content_length: u64,
pub sample_rate: u32,
pub channels: u16,
pub bit_depth: u32,
pub speed_mbps: f64,
}
/// Probe + open the progressive sink + spawn the background feeder.
///
/// On success the player has begun buffering and `play_streaming_dynamic` will
/// start audio once the initial buffer fills; the body download runs in a
/// spawned task. Errors here mean the caller should fall back to a full
/// download (the probe or the sink open failed).
pub async fn stream_remote_track_into_player(
player: &Player,
track_id: u64,
duration_secs: u64,
start_position_secs: u64,
url: &str,
log_tag: &str,
) -> Result<(), String> {
let stream_info = probe_remote_stream_info(url).await?;
log::info!(
"[{}/STREAMING] Track {} - {:.2} MB, {}Hz, {} ch, {}-bit, {:.1} MB/s",
log_tag,
track_id,
stream_info.content_length as f64 / (1024.0 * 1024.0),
stream_info.sample_rate,
stream_info.channels,
stream_info.bit_depth,
stream_info.speed_mbps
);
let writer = player
.play_streaming_dynamic(
track_id,
stream_info.sample_rate,
stream_info.channels,
stream_info.bit_depth,
stream_info.content_length,
stream_info.speed_mbps,
duration_secs,
start_position_secs,
)
.map_err(|err| format!("start streaming remote track {track_id}: {err}"))?;
let url = url.to_string();
let content_length = stream_info.content_length;
let log_tag = log_tag.to_string();
tokio::spawn(async move {
if let Err(err) =
download_and_stream_remote_track(&url, writer, track_id, content_length, &log_tag).await
{
log::error!(
"[{}/STREAMING] Track {} failed while streaming: {}",
log_tag,
track_id,
err
);
}
});
Ok(())
}
/// HEAD for content-length, then a small `Range: bytes=0-65535` GET to (a)
/// measure throughput and (b) parse the FLAC `STREAMINFO` block for the real
/// sample rate / channels / bit depth. Never defaults silently for FLAC (a
/// wrong sample rate would silently resample hi-res).
pub async fn probe_remote_stream_info(url: &str) -> Result<RemoteStreamInfo, String> {
use std::time::Instant;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|err| format!("create stream probe client: {err}"))?;
let head_response = client
.head(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("probe HEAD request failed: {err}"))?;
if !head_response.status().is_success() {
return Err(format!(
"probe HEAD request failed with status {}",
head_response.status()
));
}
let content_length = head_response
.headers()
.get("content-length")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<u64>().ok())
.ok_or_else(|| "probe missing content-length header".to_string())?;
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(|err| format!("probe range request failed: {err}"))?;
if !range_response.status().is_success() {
return Err(format!(
"probe range request failed with status {}",
range_response.status()
));
}
let initial_bytes = range_response
.bytes()
.await
.map_err(|err| format!("read probe bytes failed: {err}"))?;
let elapsed = start_time.elapsed();
let speed_mbps = if elapsed.as_secs_f64() > 0.0 {
(initial_bytes.len() as f64 / elapsed.as_secs_f64()) / (1024.0 * 1024.0)
} else {
10.0
};
let (sample_rate, channels, bit_depth) =
if initial_bytes.len() >= 26 && initial_bytes.starts_with(b"fLaC") {
let sample_rate = ((initial_bytes[18] as u32) << 12)
| ((initial_bytes[19] as u32) << 4)
| ((initial_bytes[20] as u32) >> 4);
let channels = ((initial_bytes[20] >> 1) & 0x07) + 1;
let bit_depth = ((initial_bytes[20] & 0x01) << 4) | ((initial_bytes[21] >> 4) & 0x0F);
(sample_rate, channels as u16, (bit_depth + 1) as u32)
} else {
log::warn!("[remote-stream] Non-FLAC probe for remote handoff, using defaults");
(44_100, 2, 16)
};
Ok(RemoteStreamInfo {
content_length,
sample_rate,
channels,
bit_depth,
speed_mbps,
})
}
/// Plain full-body GET → `bytes_stream()` loop → `writer.push_chunk` →
/// `writer.complete()`. No HTTP Range on the main GET (the `BufferedMediaSource`
/// buffers every pushed byte and serves seeks from the growing buffer).
pub async fn download_and_stream_remote_track(
url: &str,
writer: BufferWriter,
track_id: u64,
content_length: u64,
log_tag: &str,
) -> Result<(), String> {
use futures_util::StreamExt;
use std::time::Instant;
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(300))
.build()
.map_err(|err| format!("create remote streaming client: {err}"))?;
let response = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|err| format!("start remote streaming request failed: {err}"))?;
if !response.status().is_success() {
return Err(format!(
"remote streaming request failed with status {}",
response.status()
));
}
let mut bytes_received = 0u64;
let mut stream = response.bytes_stream();
let start_time = Instant::now();
let mut last_log_time = Instant::now();
while let Some(chunk_result) = stream.next().await {
let chunk = chunk_result.map_err(|err| format!("remote streaming chunk failed: {err}"))?;
bytes_received += chunk.len() as u64;
if let Err(err) = writer.push_chunk(&chunk) {
log::error!(
"[{}/STREAMING] Failed to push chunk for track {}: {}",
log_tag,
track_id,
err
);
}
let now = Instant::now();
if now.duration_since(last_log_time) >= Duration::from_secs(2) && content_length > 0 {
let progress = (bytes_received as f64 / content_length as f64) * 100.0;
let avg_speed =
(bytes_received as f64 / start_time.elapsed().as_secs_f64()) / (1024.0 * 1024.0);
log::info!(
"[{}/STREAMING] Track {} {:.1}% ({:.2}/{:.2} MB) @ {:.2} MB/s",
log_tag,
track_id,
progress,
bytes_received as f64 / (1024.0 * 1024.0),
content_length as f64 / (1024.0 * 1024.0),
avg_speed
);
last_log_time = now;
}
}
if let Err(err) = writer.complete() {
log::error!(
"[{}/STREAMING] Failed to mark stream complete for track {}: {}",
log_tag,
track_id,
err
);
}
log::info!(
"[{}/STREAMING] Track {} complete: {:.2} MB in {:.1}s",
log_tag,
track_id,
bytes_received as f64 / (1024.0 * 1024.0),
start_time.elapsed().as_secs_f64()
);
Ok(())
}