Joren/qobuz-qt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: qbz-1 streaming, gapless prefetch, accurate scrobbling, Range-seek

Browse Source 
Port proven playback architecture from qbqt fork:
- Bounded VecDeque buffer with condvar backpressure (4MB cap)
- decrypt_and_extract_frames for clean FLAC frame extraction from ISOBMFF
- Cancel+restart seeking with sub-segment sample skipping
- start_prefetch / QueueNext for gapless transitions with pre-started downloads
- track_transitioned signaling for scrobbler during gapless playback
- Range-request HTTP seeking for non-segmented (MP3) tracks
- OnceLock HTTP client singleton with cancel-aware chunked downloads
- Accumulated listening time scrobbling (prevents false scrobbles from seeking)
- Array-format Last.fm scrobble params (artist[0], track[0], etc.)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
joren
2026-03-30 20:12:23 +02:00
parent 6c8d032ce9
commit 70810bd4b1
16 changed files with 1539 additions and 382 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
rust/Cargo.toml
View File

@@ -28,6 +28,15 @@ md5 = "0.7"
dirs = "5"
toml = "0.8"
# Crypto — qbz-1 key derivation + segment decryption
aes = "0.8"
cbc = { version = "0.1", features = ["alloc"] }
ctr = "0.9"
hkdf = "0.12"
sha2 = "0.10"
hex = "0.4"
base64 = "0.22"
[profile.release]
lto = "thin"
opt-level = 3

6
rust/include/qobuz_backend.h
View File

@@ -75,17 +75,13 @@ uint64_t qobuz_backend_get_duration(const QobuzBackendOpaque *backend);
uint8_t qobuz_backend_get_volume(const QobuzBackendOpaque *backend);
int qobuz_backend_get_state(const QobuzBackendOpaque *backend);
int qobuz_backend_take_track_finished(QobuzBackendOpaque *backend);
int qobuz_backend_take_track_transitioned(QobuzBackendOpaque *backend);
// ReplayGain / Gapless
void qobuz_backend_set_replaygain(QobuzBackendOpaque *backend, bool enabled);
void qobuz_backend_set_gapless(QobuzBackendOpaque *backend, bool enabled);
void qobuz_backend_prefetch_track(QobuzBackendOpaque *backend, int64_t track_id, int32_t format_id);
// Visualizer PCM access
uint32_t qobuz_backend_viz_read(QobuzBackendOpaque *backend, float *buf, uint32_t max_samples);
uint32_t qobuz_backend_viz_sample_rate(const QobuzBackendOpaque *backend);
uint32_t qobuz_backend_viz_channels(const QobuzBackendOpaque *backend);
// Artist releases (auto-paginates to fetch all)
void qobuz_backend_get_artist_releases(QobuzBackendOpaque *backend, int64_t artist_id, const char *release_type, uint32_t limit, uint32_t offset);

334
rust/src/api/client.rs
View File

@@ -1,6 +1,10 @@
use anyhow::{bail, Result};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
use cbc::cipher::{block_padding::NoPadding, BlockDecryptMut, KeyIvInit};
use hkdf::Hkdf;
use reqwest::Client;
use serde_json::Value;
use sha2::Sha256;
use std::time::{SystemTime, UNIX_EPOCH};
use super::models::*;
@@ -15,10 +19,66 @@ pub const DEFAULT_APP_SECRET: &str = "e79f8b9be485692b0e5f9dd895826368";
pub struct QobuzClient {
http: Client,
pub auth_token: Option<String>,
/// Playback session ID from POST /session/start — sent as X-Session-Id for /file/url.
session_id: Option<String>,
session_expires_at: Option<u64>,
/// `infos` field from the session/start response — used to derive the KEK for track key unwrapping.
session_infos: Option<String>,
app_id: String,
app_secret: String,
}
// ── qbz-1 key derivation helpers ─────────────────────────────────────────────
type Aes128CbcDec = cbc::Decryptor<aes::Aes128>;
/// Decode a base64url string (with or without padding).
fn b64url_decode(s: &str) -> Result<Vec<u8>> {
Ok(URL_SAFE_NO_PAD.decode(s.trim_end_matches('='))?)
}
/// Full qbz-1 key derivation:
/// Phase 1: HKDF-SHA256(ikm=hex(app_secret), salt=b64url(infos[0]), info=b64url(infos[1])) → 16-byte KEK
/// Phase 2: AES-128-CBC/NoPadding(key=KEK, iv=b64url(key_field[2])).decrypt(b64url(key_field[1]))[..16]
fn derive_track_key(session_infos: &str, app_secret_hex: &str, key_field: &str) -> Result<[u8; 16]> {
// Phase 1: HKDF
let infos_parts: Vec<&str> = session_infos.splitn(2, '.').collect();
if infos_parts.len() != 2 {
bail!("session_infos must be '<salt_b64>.<info_b64>', got: {session_infos}");
}
let salt = b64url_decode(infos_parts[0])?;
let info = b64url_decode(infos_parts[1])?;
let ikm = hex::decode(app_secret_hex)?;
let hk = Hkdf::<Sha256>::new(Some(&salt), &ikm);
let mut kek = [0u8; 16];
hk.expand(&info, &mut kek)
.map_err(|e| anyhow::anyhow!("HKDF expand failed: {e:?}"))?;
// Phase 2: AES-128-CBC/NoPadding
let key_parts: Vec<&str> = key_field.splitn(3, '.').collect();
if key_parts.len() != 3 || key_parts[0] != "qbz-1" {
bail!("unexpected key field format: {key_field}");
}
let ct = b64url_decode(key_parts[1])?;
let iv_bytes = b64url_decode(key_parts[2])?;
if ct.len() < 16 || iv_bytes.len() < 16 {
bail!("key field ciphertext/iv too short ({} / {} bytes)", ct.len(), iv_bytes.len());
}
let iv: [u8; 16] = iv_bytes[..16].try_into()?;
let mut buf = ct;
let decrypted = Aes128CbcDec::new(&kek.into(), &iv.into())
.decrypt_padded_mut::<NoPadding>(&mut buf)
.map_err(|e| anyhow::anyhow!("AES-CBC decrypt failed: {e:?}"))?;
let mut track_key = [0u8; 16];
track_key.copy_from_slice(&decrypted[..16]);
Ok(track_key)
}
// ─────────────────────────────────────────────────────────────────────────────
impl QobuzClient {
pub fn new() -> Result<Self> {
Self::new_with_config(None, None)
@@ -44,6 +104,9 @@ impl QobuzClient {
Ok(Self {
http,
auth_token: None,
session_id: None,
session_expires_at: None,
session_infos: None,
app_id,
app_secret,
})
@@ -106,20 +169,16 @@ impl QobuzClient {
if let Some(token) = &self.auth_token {
builder = builder.header("Authorization", format!("Bearer {}", token));
}
if let Some(sid) = &self.session_id {
builder = builder.header("X-Session-Id", sid.as_str());
}
builder
}
// --- Auth ---
pub async fn login(&mut self, email: &str, password: &str) -> Result<OAuthLoginResponse> {
match self.oauth2_login(email, password).await {
Ok(r) => Ok(r),
Err(_) => self.legacy_login(email, password).await,
}
}
/// NOTE: Qobuz API requires credentials as GET query params — not our choice.
async fn oauth2_login(&mut self, email: &str, password: &str) -> Result<OAuthLoginResponse> {
pub async fn login(&mut self, email: &str, password: &str) -> Result<OAuthLoginResponse> {
let ts = Self::ts();
let mut sign_params: Vec<(&str, String)> = vec![
("password", password.to_string()),
@@ -144,44 +203,14 @@ impl QobuzClient {
let body: Value = resp.json().await?;
if !status.is_success() {
let msg = body.get("message").and_then(|m| m.as_str()).unwrap_or("login failed");
bail!("oauth2 login failed ({}): {}", status, msg);
}
self.extract_and_store_token(serde_json::from_value(body)?)
}
async fn legacy_login(&mut self, email: &str, password: &str) -> Result<OAuthLoginResponse> {
let ts = Self::ts();
let mut sign_params: Vec<(&str, String)> = vec![
("email", email.to_string()),
("password", password.to_string()),
];
let sig = self.request_sig("userlogin", &mut sign_params, ts);
let resp = self
.http
.get(self.url("user/login"))
.query(&[
("app_id", self.app_id.as_str()),
("email", email),
("password", password),
("request_ts", ts.to_string().as_str()),
("request_sig", sig.as_str()),
])
.send()
.await?;
let status = resp.status();
let body: Value = resp.json().await?;
if !status.is_success() {
let msg = body.get("message").and_then(|m| m.as_str()).unwrap_or("login failed");
bail!("user login failed ({}): {}", status, msg);
bail!("login failed ({}): {}", status, msg);
}
self.extract_and_store_token(serde_json::from_value(body)?)
}
fn extract_and_store_token(&mut self, login: OAuthLoginResponse) -> Result<OAuthLoginResponse> {
// auth_token = OAuth2 bearer (preferred) or legacy session token
if let Some(token) = login
.oauth2
.as_ref()
@@ -190,9 +219,57 @@ impl QobuzClient {
{
self.auth_token = Some(token);
}
// Reset any cached playback session — it belongs to a different auth context.
self.session_id = None;
self.session_expires_at = None;
self.session_infos = None;
Ok(login)
}
/// Start a playback session via POST /session/start.
/// The returned session_id is required as X-Session-Id when calling /file/url.
/// Sessions expire; we cache and reuse until 60s before expiry.
async fn ensure_session(&mut self) -> Result<()> {
let now = Self::ts();
if let (Some(_), Some(exp)) = (&self.session_id, self.session_expires_at) {
if now + 60 < exp {
return Ok(()); // still valid
}
}
let ts = Self::ts();
let mut sign_params: Vec<(&str, String)> = vec![
("profile", "qbz-1".to_string()),
];
let sig = self.request_sig("sessionstart", &mut sign_params, ts);
let resp = self
.http
.post(self.url("session/start"))
.query(&[
("app_id", self.app_id.as_str()),
("request_ts", ts.to_string().as_str()),
("request_sig", sig.as_str()),
])
.header("Authorization", format!("Bearer {}", self.auth_token.as_deref().unwrap_or("")))
.form(&[("profile", "qbz-1")])
.send()
.await?;
let body = Self::check_response(resp).await?;
let session_id = body["session_id"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("session/start: no session_id in response"))?
.to_string();
let expires_at = body["expires_at"].as_u64().unwrap_or(now + 3600);
let infos = body["infos"].as_str().map(|s| s.to_string());
eprintln!("[session] started session_id={}... expires_at={} infos={:?}", &session_id[..session_id.len().min(8)], expires_at, infos);
self.session_id = Some(session_id);
self.session_expires_at = Some(expires_at);
self.session_infos = infos;
Ok(())
}
// --- User ---
pub async fn get_user(&self) -> Result<UserDto> {
@@ -215,7 +292,9 @@ impl QobuzClient {
Ok(serde_json::from_value(body)?)
}
pub async fn get_track_url(&self, track_id: i64, format: Format) -> Result<TrackFileUrlDto> {
pub async fn get_track_url(&mut self, track_id: i64, format: Format) -> Result<TrackFileUrlDto> {
self.ensure_session().await?;
let ts = Self::ts();
let intent = "stream";
let mut sign_params: Vec<(&str, String)> = vec![
@@ -223,10 +302,10 @@ impl QobuzClient {
("intent", intent.to_string()),
("track_id", track_id.to_string()),
];
let sig = self.request_sig("trackgetFileUrl", &mut sign_params, ts);
let sig = self.request_sig("fileurl", &mut sign_params, ts);
let resp = self
.get_request("track/getFileUrl")
.get_request("file/url")
.query(&[
("track_id", track_id.to_string()),
("format_id", format.id().to_string()),
@@ -238,7 +317,24 @@ impl QobuzClient {
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body)?)
eprintln!("[file/url] response: {}", serde_json::to_string(&body).unwrap_or_default());
let mut url_dto: TrackFileUrlDto = serde_json::from_value(body)?;
// Unwrap the per-track key: decrypt the CBC-wrapped key using HKDF-derived KEK.
if let (Some(key_field), Some(infos)) = (url_dto.key.clone(), self.session_infos.as_deref()) {
match derive_track_key(infos, &self.app_secret, &key_field) {
Ok(track_key) => {
url_dto.key = Some(hex::encode(track_key));
eprintln!("[key] track key unwrapped OK");
}
Err(e) => {
eprintln!("[key] track key unwrap failed: {e}");
url_dto.key = None; // disable decryption rather than play garbage
}
}
}
Ok(url_dto)
}
// --- Album ---
@@ -289,17 +385,48 @@ impl QobuzClient {
// --- Search ---
pub async fn search(&self, query: &str, offset: u32, limit: u32) -> Result<SearchCatalogDto> {
let (tracks, albums, artists) = tokio::try_join!(
self.search_tracks(query, offset, limit),
self.search_albums(query, offset, limit),
self.search_artists(query, offset, limit),
)?;
Ok(SearchCatalogDto {
query: Some(query.to_string()),
albums: Some(albums),
tracks: Some(tracks),
artists: Some(artists),
playlists: None,
})
}
async fn search_tracks(&self, query: &str, offset: u32, limit: u32) -> Result<SearchResultItems<TrackDto>> {
let resp = self
.get_request("catalog/search")
.query(&[
("query", query),
("offset", &offset.to_string()),
("limit", &limit.to_string()),
])
.get_request("track/search")
.query(&[("query", query), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body)?)
Ok(serde_json::from_value(body["tracks"].clone())?)
}
async fn search_albums(&self, query: &str, offset: u32, limit: u32) -> Result<SearchResultItems<AlbumDto>> {
let resp = self
.get_request("album/search")
.query(&[("query", query), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body["albums"].clone())?)
}
async fn search_artists(&self, query: &str, offset: u32, limit: u32) -> Result<SearchResultItems<ArtistDto>> {
let resp = self
.get_request("artist/search")
.query(&[("query", query), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body["artists"].clone())?)
}
// --- Favorites / Library ---
@@ -329,34 +456,103 @@ impl QobuzClient {
Ok(serde_json::from_value(body)?)
}
pub async fn get_fav_tracks(&self, offset: u32, limit: u32) -> Result<SearchResultItems<TrackDto>> {
/// Fetch all favorite IDs (tracks, albums, artists) in one call.
async fn get_fav_ids(&self) -> Result<FavIdsDto> {
let resp = self
.get_request("favorite/getUserFavorites")
.query(&[("type", "tracks"), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.get_request("favorite/getUserFavoriteIds")
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body["tracks"].clone())?)
Ok(serde_json::from_value(body)?)
}
/// Batch-fetch tracks by ID via POST /track/getList.
async fn get_tracks_by_ids(&self, ids: &[i64]) -> Result<Vec<TrackDto>> {
if ids.is_empty() {
return Ok(vec![]);
}
let resp = self
.post_request("track/getList")
.json(&serde_json::json!({ "tracks_id": ids }))
.send()
.await?;
let body = Self::check_response(resp).await?;
let items: Vec<TrackDto> = serde_json::from_value(
body["tracks"]["items"].clone(),
)
.unwrap_or_default();
Ok(items)
}
pub async fn get_fav_tracks(&self, offset: u32, limit: u32) -> Result<SearchResultItems<TrackDto>> {
let ids = self.get_fav_ids().await?;
let all_ids = ids.tracks.unwrap_or_default();
let total = all_ids.len() as i32;
let page: Vec<i64> = all_ids
.into_iter()
.skip(offset as usize)
.take(limit as usize)
.collect();
let items = self.get_tracks_by_ids(&page).await?;
Ok(SearchResultItems {
items: Some(items),
total: Some(total),
offset: Some(offset as i32),
limit: Some(limit as i32),
})
}
pub async fn get_fav_albums(&self, offset: u32, limit: u32) -> Result<SearchResultItems<AlbumDto>> {
let resp = self
.get_request("favorite/getUserFavorites")
.query(&[("type", "albums"), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body["albums"].clone())?)
let ids = self.get_fav_ids().await?;
let all_ids = ids.albums.unwrap_or_default();
let total = all_ids.len() as i32;
let page: Vec<&str> = all_ids
.iter()
.skip(offset as usize)
.take(limit as usize)
.map(|s| s.as_str())
.collect();
let mut items = Vec::with_capacity(page.len());
for album_id in page {
match self.get_album(album_id).await {
Ok(a) => items.push(a),
Err(e) => eprintln!("[fav] failed to fetch album {}: {}", album_id, e),
}
}
Ok(SearchResultItems {
items: Some(items),
total: Some(total),
offset: Some(offset as i32),
limit: Some(limit as i32),
})
}
pub async fn get_fav_artists(&self, offset: u32, limit: u32) -> Result<SearchResultItems<FavArtistDto>> {
let resp = self
.get_request("favorite/getUserFavorites")
.query(&[("type", "artists"), ("offset", &offset.to_string()), ("limit", &limit.to_string())])
.send()
.await?;
let body = Self::check_response(resp).await?;
Ok(serde_json::from_value(body["artists"].clone())?)
let ids = self.get_fav_ids().await?;
let all_ids = ids.artists.unwrap_or_default();
let total = all_ids.len() as i32;
let page: Vec<i64> = all_ids
.into_iter()
.skip(offset as usize)
.take(limit as usize)
.collect();
let mut items = Vec::with_capacity(page.len());
for artist_id in page {
match self.get_artist_page(artist_id).await {
Ok(v) => {
if let Ok(a) = serde_json::from_value::<FavArtistDto>(v) {
items.push(a);
}
}
Err(e) => eprintln!("[fav] failed to fetch artist {}: {}", artist_id, e),
}
}
Ok(SearchResultItems {
items: Some(items),
total: Some(total),
offset: Some(offset as i32),
limit: Some(limit as i32),
})
}
// --- Playlist management ---

15
rust/src/api/models.rs
View File

@@ -74,12 +74,16 @@ pub struct AudioInfoDto {
#[derive(Debug, Deserialize, Clone, Serialize)]
pub struct TrackFileUrlDto {
pub track_id: Option<i64>,
pub duration: Option<i64>,
pub duration: Option<f64>,
pub url: Option<String>,
pub url_template: Option<String>,
pub n_segments: Option<u32>,
pub format_id: Option<i32>,
pub mime_type: Option<String>,
pub sampling_rate: Option<f64>,
pub bit_depth: Option<i32>,
/// qbz-1 encryption key: "qbz-1.<base64url_aes_key>.<base64url_iv>"
pub key: Option<String>,
}
// --- Album ---
@@ -219,6 +223,15 @@ pub struct FavArtistDto {
pub image: Option<ImageDto>,
}
// --- Favorite IDs ---
#[derive(Debug, Deserialize, Serialize)]
pub struct FavIdsDto {
pub tracks: Option<Vec<i64>>,
pub albums: Option<Vec<String>>,
pub artists: Option<Vec<i64>>,
}
// --- Format ---
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]

124
rust/src/lib.rs
View File

@@ -81,6 +81,9 @@ struct PrefetchedTrack {
track_id: i64,
track: api::models::TrackDto,
url: String,
n_segments: u32,
encryption_key: Option<String>,
prefetch_data: Option<player::decoder::PrefetchData>,
}
struct BackendInner {
@@ -476,8 +479,9 @@ pub unsafe extern "C" fn qobuz_backend_play_track(
}
};
let (track, url) = if let Some(pf) = cached {
(pf.track, pf.url)
// Extract prefetch_data to embed directly into TrackInfo
let (track, url, n_segments, encryption_key, prefetch_data) = if let Some(pf) = cached {
(pf.track, pf.url, pf.n_segments, pf.encryption_key, pf.prefetch_data)
} else {
// Fetch track metadata
let track = match client.lock().await.get_track(track_id).await {
@@ -489,11 +493,24 @@ pub unsafe extern "C" fn qobuz_backend_play_track(
Ok(u) => u,
Err(e) => { call_cb(cb, ud, EV_TRACK_URL_ERR, &err_json(&e.to_string())); return; }
};
let url = match url_dto.url {
Some(u) => u,
None => { call_cb(cb, ud, EV_TRACK_URL_ERR, &err_json("no stream URL")); return; }
let encryption_key = url_dto.key.clone();
// Prefer segmented url_template (reliable CDN path), fall back to plain url
eprintln!("[lib] url_dto: url={:?}, url_template={:?}, n_segments={:?}, mime={:?}, key_present={}",
url_dto.url.as_deref().map(|u| &u[..u.len().min(60)]),
url_dto.url_template.as_deref().map(|u| &u[..u.len().min(60)]),
url_dto.n_segments,
url_dto.mime_type,
encryption_key.is_some());
let (url, n_segments) = if let (Some(tmpl), Some(n)) = (url_dto.url_template, url_dto.n_segments) {
(tmpl, n)
} else if let Some(u) = url_dto.url {
(u, 0u32)
} else {
call_cb(cb, ud, EV_TRACK_URL_ERR, &err_json("no stream URL"));
return;
};
(track, url)
eprintln!("[lib] resolved: n_segments={n_segments}, url_prefix={}", &url[..url.len().min(80)]);
(track, url, n_segments, encryption_key, None)
};
// 2. Notify track change
@@ -513,7 +530,7 @@ pub unsafe extern "C" fn qobuz_backend_play_track(
if let Some(dur) = track.duration {
status.duration_secs.store(dur as u64, std::sync::atomic::Ordering::Relaxed);
}
let _ = cmd_tx.send(player::PlayerCommand::Play(player::TrackInfo { track, url, replaygain_db }));
let _ = cmd_tx.send(player::PlayerCommand::Play(player::TrackInfo { track, url, n_segments, encryption_key, replaygain_db, prefetch_data }));
// 5. State notification
call_cb(cb, ud, EV_STATE_CHANGED, r#"{"state":"playing"}"#);
@@ -582,6 +599,28 @@ pub unsafe extern "C" fn qobuz_backend_take_track_finished(ptr: *mut Backend) ->
if finished { 1 } else { 0 }
}
#[no_mangle]
pub unsafe extern "C" fn qobuz_backend_take_track_transitioned(ptr: *mut Backend) -> c_int {
let inner = &(*ptr).0;
let transitioned = inner
.player
.status
.track_transitioned
.swap(false, std::sync::atomic::Ordering::SeqCst);
if transitioned {
// Emit track changed so the Qt UI and Scrobbler automatically pick up the new song
if let Some(track) = inner.player.status.current_track.lock().unwrap().as_ref() {
if let Ok(j) = serde_json::to_string(track) {
call_cb(inner.cb, inner.ud, EV_TRACK_CHANGED, &j);
}
}
1
} else {
0
}
}
// ---------- ReplayGain / Gapless ----------
#[no_mangle]
@@ -600,10 +639,11 @@ pub unsafe extern "C" fn qobuz_backend_prefetch_track(
track_id: i64,
format_id: i32,
) {
let inner = &(*ptr).0;
let inner = &(*ptr).0;
let client = inner.client.clone();
let prefetch = inner.prefetch.clone();
let format = Format::from_id(format_id);
let cmd_tx = inner.player.cmd_tx.clone();
let rg_enabled = inner.replaygain_enabled.clone();
spawn(inner, async move {
let track = match client.lock().await.get_track(track_id).await {
@@ -614,11 +654,41 @@ pub unsafe extern "C" fn qobuz_backend_prefetch_track(
Ok(u) => u,
Err(_) => return,
};
let url = match url_dto.url {
Some(u) => u,
None => return,
let encryption_key = url_dto.key.clone();
let (url, n_segments) = if let (Some(tmpl), Some(n)) = (url_dto.url_template, url_dto.n_segments) {
(tmpl, n)
} else if let Some(u) = url_dto.url {
(u, 0u32)
} else {
return;
};
*prefetch.lock().await = Some(PrefetchedTrack { track_id, track, url });
// KICKSTART DOWNLOADING IMMEDIATELY
let prefetch_data = if n_segments > 0 {
Some(player::decoder::start_prefetch(
url.clone(),
n_segments,
encryption_key.as_deref(),
1,
))
} else {
None
};
let replaygain_db = if rg_enabled.load(std::sync::atomic::Ordering::Relaxed) {
track.audio_info.as_ref().and_then(|ai| ai.replaygain_track_gain)
} else {
None
};
let _ = cmd_tx.send(player::PlayerCommand::QueueNext(player::TrackInfo {
track,
url,
n_segments,
encryption_key,
replaygain_db,
prefetch_data,
}));
});
}
@@ -717,34 +787,6 @@ pub unsafe extern "C" fn qobuz_backend_get_user(ptr: *mut Backend) {
});
}
// ---------- Visualizer PCM access ----------
/// Read up to `max_samples` f32 PCM values into `buf`.
/// Returns the number of samples actually read.
#[no_mangle]
pub unsafe extern "C" fn qobuz_backend_viz_read(
ptr: *mut Backend,
buf: *mut f32,
max_samples: u32,
) -> u32 {
let consumer = &(*ptr).0.player.status.viz_consumer;
let Ok(mut lock) = consumer.try_lock() else { return 0 };
let slice = std::slice::from_raw_parts_mut(buf, max_samples as usize);
rb::RbConsumer::read(&mut *lock, slice).unwrap_or(0) as u32
}
/// Returns current sample rate of the audio stream (0 if idle).
#[no_mangle]
pub unsafe extern "C" fn qobuz_backend_viz_sample_rate(ptr: *const Backend) -> u32 {
(*ptr).0.player.status.viz_sample_rate.load(std::sync::atomic::Ordering::Relaxed)
}
/// Returns current channel count (0 if idle).
#[no_mangle]
pub unsafe extern "C" fn qobuz_backend_viz_channels(ptr: *const Backend) -> u32 {
(*ptr).0.player.status.viz_channels.load(std::sync::atomic::Ordering::Relaxed)
}
// ---------- Playlist management ----------
pub const EV_PLAYLIST_CREATED: c_int = 20;

1069
rust/src/player/decoder.rs
View File

File diff suppressed because it is too large Load Diff

83
rust/src/player/mod.rs
View File

@@ -1,33 +1,33 @@
mod decoder;
pub mod decoder;
pub mod output;
use rb::{SpscRb, RB};
use std::sync::{
atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicU8, Ordering},
atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering},
Arc,
};
use std::time::Duration;
use crate::api::TrackDto;
use decoder::NextAction;
/// Size of the visualizer ring buffer in f32 samples (~180ms at 44.1kHz stereo).
const VIZ_RING_SIZE: usize = 16 * 1024;
#[derive(Debug, Clone)]
#[derive(Clone)]
pub enum PlayerCommand {
Play(TrackInfo),
QueueNext(TrackInfo),
Pause,
Resume,
Stop,
SetVolume(u8),
}
#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct TrackInfo {
pub track: TrackDto,
pub url: String,
/// ReplayGain track gain in dB, if enabled and available.
pub n_segments: u32,
pub encryption_key: Option<String>,
pub replaygain_db: Option<f64>,
pub prefetch_data: Option<decoder::PrefetchData>,
}
#[derive(Debug, Clone, PartialEq)]
@@ -45,26 +45,16 @@ pub struct PlayerStatus {
pub duration_secs: Arc<AtomicU64>,
pub volume: Arc<AtomicU8>,
pub current_track: Arc<std::sync::Mutex<Option<TrackDto>>>,
/// Set to true by the decode thread when a track finishes naturally.
pub track_finished: Arc<AtomicBool>,
/// Set by the player loop when a seek command arrives; cleared by the decode thread.
pub track_transitioned: Arc<AtomicBool>,
pub seek_requested: Arc<AtomicBool>,
pub seek_target_secs: Arc<AtomicU64>,
/// Linear gain factor to apply (1.0 = unity). Updated each time a new track starts.
pub replaygain_gain: Arc<std::sync::Mutex<f32>>,
/// When false the audio output is torn down after each track, producing a gap.
pub gapless: Arc<AtomicBool>,
/// Visualizer ring buffer (consumer side, read by FFI).
pub viz_ring: Arc<SpscRb<f32>>,
pub viz_consumer: Arc<std::sync::Mutex<rb::Consumer<f32>>>,
pub viz_sample_rate: Arc<AtomicU32>,
pub viz_channels: Arc<AtomicU32>,
}
impl PlayerStatus {
pub fn new() -> Self {
let viz_ring = Arc::new(SpscRb::new(VIZ_RING_SIZE));
let viz_consumer = Arc::new(std::sync::Mutex::new(viz_ring.consumer()));
Self {
state: Arc::new(std::sync::Mutex::new(PlayerState::Idle)),
position_secs: Arc::new(AtomicU64::new(0)),
@@ -72,33 +62,26 @@ impl PlayerStatus {
volume: Arc::new(AtomicU8::new(80)),
current_track: Arc::new(std::sync::Mutex::new(None)),
track_finished: Arc::new(AtomicBool::new(false)),
track_transitioned: Arc::new(AtomicBool::new(false)),
seek_requested: Arc::new(AtomicBool::new(false)),
seek_target_secs: Arc::new(AtomicU64::new(0)),
replaygain_gain: Arc::new(std::sync::Mutex::new(1.0)),
gapless: Arc::new(AtomicBool::new(false)),
viz_ring,
viz_consumer,
viz_sample_rate: Arc::new(AtomicU32::new(0)),
viz_channels: Arc::new(AtomicU32::new(0)),
}
}
pub fn get_state(&self) -> PlayerState {
self.state.lock().unwrap().clone()
}
pub fn get_position(&self) -> u64 {
self.position_secs.load(Ordering::Relaxed)
}
pub fn get_duration(&self) -> u64 {
self.duration_secs.load(Ordering::Relaxed)
}
pub fn get_volume(&self) -> u8 {
self.volume.load(Ordering::Relaxed)
}
}
pub struct Player {
@@ -122,51 +105,45 @@ impl Player {
pub fn send(&self, cmd: PlayerCommand) {
self.cmd_tx.send(cmd).ok();
}
pub fn pause(&self) {
self.send(PlayerCommand::Pause);
}
pub fn resume(&self) {
self.send(PlayerCommand::Resume);
}
pub fn stop(&self) {
self.send(PlayerCommand::Stop);
}
pub fn set_volume(&self, vol: u8) {
self.status.volume.store(vol, Ordering::Relaxed);
self.send(PlayerCommand::SetVolume(vol));
}
pub fn seek(&self, secs: u64) {
self.status.seek_target_secs.store(secs, Ordering::Relaxed);
self.status.seek_requested.store(true, Ordering::SeqCst);
}
}
/// The player loop runs on a single dedicated OS thread.
/// It owns the `AudioOutput` locally so there are no Send constraints.
/// Decoding is performed inline; the command channel is polled via try_recv
/// inside the decode loop to handle Pause/Resume/Seek/Stop/Play without
/// tearng down and re-opening the audio device between tracks.
fn player_loop(rx: std::sync::mpsc::Receiver<PlayerCommand>, status: PlayerStatus) {
use std::sync::mpsc::RecvTimeoutError;
let mut audio_output: Option<output::AudioOutput> = None;
let paused = Arc::new(AtomicBool::new(false));
// pending_info holds a Play command that interrupted an ongoing decode
let mut pending_info: Option<TrackInfo> = None;
let mut pending_action: Option<NextAction> = None;
'outer: loop {
// Wait for a Play command (or use one that was interrupted)
let info = if let Some(p) = pending_info.take() {
p
let info = if let Some(action) = pending_action.take() {
match action {
NextAction::Play(info) | NextAction::Transition(info) => info,
}
} else {
loop {
match rx.recv_timeout(Duration::from_millis(100)) {
Ok(PlayerCommand::Play(info)) => break info,
Ok(PlayerCommand::QueueNext(info)) => {
// If completely idle and get QueueNext, treat as Play
break info;
}
Ok(PlayerCommand::Stop) => {
audio_output = None;
paused.store(false, Ordering::SeqCst);
@@ -178,15 +155,15 @@ fn player_loop(rx: std::sync::mpsc::Receiver<PlayerCommand>, status: PlayerStatu
Ok(PlayerCommand::SetVolume(v)) => {
status.volume.store(v, Ordering::Relaxed);
}
Ok(_) => {} // Pause/Resume ignored when idle
Ok(_) => {}
Err(RecvTimeoutError::Timeout) => {}
Err(RecvTimeoutError::Disconnected) => break 'outer,
}
}
};
// Compute ReplayGain factor
let rg_factor = info.replaygain_db
let rg_factor = info
.replaygain_db
.map(|db| 10f32.powf(db as f32 / 20.0))
.unwrap_or(1.0);
*status.replaygain_gain.lock().unwrap() = rg_factor;
@@ -199,13 +176,15 @@ fn player_loop(rx: std::sync::mpsc::Receiver<PlayerCommand>, status: PlayerStatu
status.position_secs.store(0, Ordering::Relaxed);
paused.store(false, Ordering::SeqCst);
match decoder::play_track_inline(&info.url, &status, &paused, &mut audio_output, &rx) {
Ok(Some(next_info)) => {
// Interrupted by a new Play — loop immediately with reused audio output
pending_info = Some(next_info);
match decoder::play_track_inline(info, &status, &paused, &mut audio_output, &rx) {
Ok(Some(NextAction::Play(next_track))) => {
pending_action = Some(NextAction::Play(next_track));
}
Ok(Some(NextAction::Transition(next_track))) => {
pending_action = Some(NextAction::Play(next_track));
status.track_transitioned.store(true, Ordering::SeqCst);
}
Ok(None) => {
// Track finished naturally — tear down audio output if gapless is off
if !status.gapless.load(Ordering::Relaxed) {
audio_output = None;
}
@@ -215,8 +194,6 @@ fn player_loop(rx: std::sync::mpsc::Receiver<PlayerCommand>, status: PlayerStatu
Err(e) => {
eprintln!("playback error: {e}");
*status.state.lock().unwrap() = PlayerState::Error(e.to_string());
// Signal track end so the queue advances to the next track
// instead of stalling on an unplayable track.
status.track_finished.store(true, Ordering::SeqCst);
}
}

18
rust/src/player/output.rs
View File

@@ -14,7 +14,6 @@ const RING_BUFFER_SIZE: usize = 32 * 1024;
pub struct AudioOutput {
ring_buf_producer: rb::Producer<f32>,
viz_producer: Option<rb::Producer<f32>>,
_stream: cpal::Stream,
pub sample_rate: u32,
pub channels: usize,
@@ -52,17 +51,12 @@ impl AudioOutput {
Ok(Self {
ring_buf_producer: producer,
viz_producer: None,
_stream: stream,
sample_rate,
channels,
})
}
pub fn set_viz_producer(&mut self, producer: rb::Producer<f32>) {
self.viz_producer = Some(producer);
}
pub fn write(
&mut self,
decoded: AudioBufferRef<'_>,
@@ -75,13 +69,13 @@ impl AudioOutput {
);
sample_buf.copy_interleaved_ref(decoded);
let samples: Vec<f32> = sample_buf.samples().iter().map(|s| s * volume).collect();
self.write_samples(&samples, stop)
}
// Best-effort copy for visualizer (non-blocking, ok to drop samples)
if let Some(ref mut viz) = self.viz_producer {
let _ = viz.write(&samples);
}
let mut remaining = &samples[..];
/// Write pre-converted interleaved f32 samples directly to the ring buffer.
/// Returns early (without error) if `stop` is set.
pub fn write_samples(&mut self, samples: &[f32], stop: &Arc<AtomicBool>) -> Result<()> {
let mut remaining = samples;
while !remaining.is_empty() {
if stop.load(Ordering::SeqCst) {
return Ok(());