yetangitu/librespot
1
0
Fork 0
mirror of https://github.com/librespot-org/librespot.git synced 2025-10-02 17:29:22 +02:00
Code Issues Releases Wiki Activity

refactor: remove parking_lot dependency and refine feature selections (#1543)

Browse source 
- refine dependency features and versions in Cargo.toml files
- switch from parking_lot to std sync primitives
- remove dashmap dependency and use DefaultKeyedStateStore
- update crates

Replace parking_lot with std::sync::{Mutex, RwLock, Condvar} throughout the
codebase. Update dependencies and code to use poisoning-aware locks, adding
explicit panic messages where necessary. Update governor to use DashMapStateStore
for rate limiting.
This commit is contained in:
Roderick van Domburg 2025-09-21 22:43:50 +02:00 • committed by GitHub
parent df5f957bdd
commit 6f6cd04874
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 637 additions and 340 deletions
Download patch file Download diff file Expand all files Collapse all files

514
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

1
Cargo.toml
View file

@ -179,7 +179,6 @@ tokio = { version = "1", features = [
"macros",
"signal",
"sync",
"parking_lot",
"process",
] }
url = "2.2"

5
audio/Cargo.toml
View file

@ -23,12 +23,11 @@ librespot-core = { version = "0.7.1", path = "../core", default-features = false
aes = "0.8"
bytes = "1"
ctr = "0.9"
futures-util = "0.3"
futures-util = { version = "0.3", default-features = false, features = ["std"] }
http-body-util = "0.1"
hyper = { version = "1.6", features = ["http1", "http2"] }
hyper-util = { version = "0.1", features = ["client", "http2"] }
log = "0.4"
parking_lot = { version = "0.12", features = ["deadlock_detection"] }
tempfile = "3"
thiserror = "2"
tokio = { version = "1", features = ["macros", "parking_lot", "sync"] }
tokio = { version = "1", features = ["macros", "sync"] }

42
audio/src/fetch/mod.rs
View file

@ -8,13 +8,14 @@ use std::{
Arc, OnceLock,
atomic::{AtomicBool, AtomicUsize, Ordering},
},
sync::{Condvar, Mutex},
time::Duration,
};
use futures_util::{StreamExt, TryFutureExt, future::IntoStream};
use hyper::{Response, StatusCode, body::Incoming, header::CONTENT_RANGE};
use hyper_util::client::legacy::ResponseFuture;
use parking_lot::{Condvar, Mutex};
use tempfile::NamedTempFile;
use thiserror::Error;
use tokio::sync::{Semaphore, mpsc, oneshot};
@ -27,6 +28,8 @@ use crate::range_set::{Range, RangeSet};
pub type AudioFileResult = Result<(), librespot_core::Error>;
const DOWNLOAD_STATUS_POISON_MSG: &str = "audio download status mutex should not be poisoned";
#[derive(Error, Debug)]
pub enum AudioFileError {
#[error("other end of channel disconnected")]
@ -163,7 +166,10 @@ impl StreamLoaderController {
pub fn range_available(&self, range: Range) -> bool {
if let Some(ref shared) = self.stream_shared {
let download_status = shared.download_status.lock();
let download_status = shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
range.length
<= download_status
@ -214,7 +220,10 @@ impl StreamLoaderController {
self.fetch(range);
if let Some(ref shared) = self.stream_shared {
let mut download_status = shared.download_status.lock();
let mut download_status = shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
let download_timeout = AudioFetchParams::get().download_timeout;
while range.length
@ -222,11 +231,13 @@ impl StreamLoaderController {
.downloaded
.contained_length_from_value(range.start)
{
if shared
let (new_download_status, wait_result) = shared
.cond
.wait_for(&mut download_status, download_timeout)
.timed_out()
{
.wait_timeout(download_status, download_timeout)
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status = new_download_status;
if wait_result.timed_out() {
return Err(AudioFileError::WaitTimeout.into());
}
@ -558,7 +569,11 @@ impl Read for AudioFileStreaming {
let mut ranges_to_request = RangeSet::new();
ranges_to_request.add_range(&Range::new(offset, length_to_request));
let mut download_status = self.shared.download_status.lock();
let mut download_status = self
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
ranges_to_request.subtract_range_set(&download_status.downloaded);
ranges_to_request.subtract_range_set(&download_status.requested);
@ -571,12 +586,14 @@ impl Read for AudioFileStreaming {
let download_timeout = AudioFetchParams::get().download_timeout;
while !download_status.downloaded.contains(offset) {
if self
let (new_download_status, wait_result) = self
.shared
.cond
.wait_for(&mut download_status, download_timeout)
.timed_out()
{
.wait_timeout(download_status, download_timeout)
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status = new_download_status;
if wait_result.timed_out() {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
Error::deadline_exceeded(AudioFileError::WaitTimeout),
@ -619,6 +636,7 @@ impl Seek for AudioFileStreaming {
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG)
.downloaded
.contains(requested_pos as usize);

35
audio/src/fetch/receive.rs
View file

@ -33,6 +33,7 @@ enum ReceivedData {
}
const ONE_SECOND: Duration = Duration::from_secs(1);
const DOWNLOAD_STATUS_POISON_MSG: &str = "audio download status mutex should not be poisoned";
async fn receive_data(
shared: Arc<AudioFileShared>,
@ -124,7 +125,10 @@ async fn receive_data(
if bytes_remaining > 0 {
{
let missing_range = Range::new(offset, bytes_remaining);
let mut download_status = shared.download_status.lock();
let mut download_status = shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status.requested.subtract_range(&missing_range);
shared.cond.notify_all();
}
@ -189,7 +193,11 @@ impl AudioFileFetch {
// The iteration that follows spawns streamers fast, without awaiting them,
// so holding the lock for the entire scope of this function should be faster
// then locking and unlocking multiple times.
let mut download_status = self.shared.download_status.lock();
let mut download_status = self
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
ranges_to_request.subtract_range_set(&download_status.downloaded);
ranges_to_request.subtract_range_set(&download_status.requested);
@ -227,7 +235,11 @@ impl AudioFileFetch {
let mut missing_data = RangeSet::new();
missing_data.add_range(&Range::new(0, self.shared.file_size));
{
let download_status = self.shared.download_status.lock();
let download_status = self
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
missing_data.subtract_range_set(&download_status.downloaded);
missing_data.subtract_range_set(&download_status.requested);
}
@ -349,7 +361,11 @@ impl AudioFileFetch {
let received_range = Range::new(data.offset, data.data.len());
let full = {
let mut download_status = self.shared.download_status.lock();
let mut download_status = self
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status.downloaded.add_range(&received_range);
self.shared.cond.notify_all();
@ -415,7 +431,10 @@ pub(super) async fn audio_file_fetch(
initial_request.offset + initial_request.length,
);
let mut download_status = shared.download_status.lock();
let mut download_status = shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status.requested.add_range(&requested_range);
}
@ -466,7 +485,11 @@ pub(super) async fn audio_file_fetch(
if fetch.shared.is_download_streaming() && fetch.has_download_slots_available() {
let bytes_pending: usize = {
let download_status = fetch.shared.download_status.lock();
let download_status = fetch
.shared
.download_status
.lock()
.expect(DOWNLOAD_STATUS_POISON_MSG);
download_status
.requested

8
connect/Cargo.toml
View file

@ -22,12 +22,12 @@ librespot-core = { version = "0.7.1", path = "../core", default-features = false
librespot-playback = { version = "0.7.1", path = "../playback", default-features = false }
librespot-protocol = { version = "0.7.1", path = "../protocol", default-features = false }
futures-util = "0.3"
futures-util = { version = "0.3", default-features = false, features = ["std"] }
log = "0.4"
protobuf = "3.7"
rand = { version = "0.9", default-features = false, features = ["small_rng"] }
serde_json = "1.0"
thiserror = "2"
tokio = { version = "1", features = ["macros", "parking_lot", "sync"] }
tokio-stream = "0.1"
uuid = { version = "1.18", features = ["v4"] }
tokio = { version = "1", features = ["macros", "sync"] }
tokio-stream = { version = "0.1", default-features = false }
uuid = { version = "1.18", default-features = false, features = ["v4"] }

20
core/Cargo.toml
View file

@ -51,16 +51,12 @@ data-encoding = "2.9"
flate2 = "1.1"
form_urlencoded = "1.2"
futures-core = "0.3"
futures-util = { version = "0.3", features = [
futures-util = { version = "0.3", default-features = false, features = [
"alloc",
"bilock",
"sink",
"unstable",
] }
governor = { version = "0.10", default-features = false, features = [
"std",
"jitter",
] }
governor = { version = "0.10", default-features = false, features = ["std"] }
hmac = "0.12"
httparse = "1.10"
http = "1.3"
@ -84,14 +80,13 @@ num-bigint = "0.4"
num-derive = "0.4"
num-integer = "0.1"
num-traits = "0.2"
parking_lot = { version = "0.12", features = ["deadlock_detection"] }
pbkdf2 = { version = "0.12", default-features = false, features = ["hmac"] }
pin-project-lite = "0.2"
priority-queue = "2.5"
protobuf = "3.7"
protobuf-json-mapping = "3.7"
quick-xml = { version = "0.38", features = ["serialize"] }
rand = "0.9"
rand = { version = "0.9", default-features = false, features = ["thread_rng"] }
rsa = "0.9"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
@ -104,23 +99,22 @@ tokio = { version = "1", features = [
"io-util",
"macros",
"net",
"parking_lot",
"rt",
"sync",
"time",
] }
tokio-stream = "0.1"
tokio-stream = { version = "0.1", default-features = false }
tokio-tungstenite = { version = "0.27", default-features = false }
tokio-util = { version = "0.7", features = ["codec"] }
tokio-util = { version = "0.7", default-features = false }
url = "2"
uuid = { version = "1", default-features = false, features = ["v4"] }
[build-dependencies]
rand = "0.9"
rand = { version = "0.9", default-features = false, features = ["thread_rng"] }
rand_distr = "0.5"
vergen-gitcl = { version = "1.0", default-features = false, features = [
"build",
] }
[dev-dependencies]
tokio = { version = "1", features = ["macros", "parking_lot"] }
tokio = { version = "1", features = ["macros"] }

22
core/src/cache.rs
View file

@ -4,16 +4,17 @@ use std::{
fs::{self, File},
io::{self, Read, Write},
path::{Path, PathBuf},
sync::Arc,
sync::{Arc, Mutex},
time::SystemTime,
};
use parking_lot::Mutex;
use priority_queue::PriorityQueue;
use thiserror::Error;
use crate::{Error, FileId, authentication::Credentials, error::ErrorKind};
const CACHE_LIMITER_POISON_MSG: &str = "cache limiter mutex should not be poisoned";
#[derive(Debug, Error)]
pub enum CacheError {
#[error("audio cache location is not configured")]
@ -189,15 +190,24 @@ impl FsSizeLimiter {
}
fn add(&self, file: &Path, size: u64) {
self.limiter.lock().add(file, size, SystemTime::now())
self.limiter
.lock()
.expect(CACHE_LIMITER_POISON_MSG)
.add(file, size, SystemTime::now())
}
fn touch(&self, file: &Path) -> bool {
self.limiter.lock().update(file, SystemTime::now())
self.limiter
.lock()
.expect(CACHE_LIMITER_POISON_MSG)
.update(file, SystemTime::now())
}
fn remove(&self, file: &Path) -> bool {
self.limiter.lock().remove(file)
self.limiter
.lock()
.expect(CACHE_LIMITER_POISON_MSG)
.remove(file)
}
fn prune_internal<F: FnMut() -> Option<PathBuf>>(mut pop: F) -> Result<(), Error> {
@ -232,7 +242,7 @@ impl FsSizeLimiter {
}
fn prune(&self) -> Result<(), Error> {
Self::prune_internal(|| self.limiter.lock().pop())
Self::prune_internal(|| self.limiter.lock().expect(CACHE_LIMITER_POISON_MSG).pop())
}
fn new(path: &Path, limit: u64) -> Result<Self, Error> {

9
core/src/component.rs
View file

@ -1,20 +1,23 @@
pub(crate) const COMPONENT_POISON_MSG: &str = "component mutex should not be poisoned";
macro_rules! component {
($name:ident : $inner:ident { $($key:ident : $ty:ty = $value:expr,)* }) => {
#[derive(Clone)]
pub struct $name(::std::sync::Arc<($crate::session::SessionWeak, ::parking_lot::Mutex<$inner>)>);
pub struct $name(::std::sync::Arc<($crate::session::SessionWeak, ::std::sync::Mutex<$inner>)>);
impl $name {
#[allow(dead_code)]
pub(crate) fn new(session: $crate::session::SessionWeak) -> $name {
debug!(target:"librespot::component", "new {}", stringify!($name));
$name(::std::sync::Arc::new((session, ::parking_lot::Mutex::new($inner {
$name(::std::sync::Arc::new((session, ::std::sync::Mutex::new($inner {
$($key : $value,)*
}))))
}
#[allow(dead_code)]
fn lock<F: FnOnce(&mut $inner) -> R, R>(&self, f: F) -> R {
let mut inner = (self.0).1.lock();
let mut inner = (self.0).1.lock()
.expect($crate::component::COMPONENT_POISON_MSG);
f(&mut inner)
}

54
core/src/dealer/mod.rs
View file

@ -6,7 +6,7 @@ use std::{
iter,
pin::Pin,
sync::{
Arc,
Arc, Mutex,
atomic::{self, AtomicBool},
},
task::Poll,
@ -15,7 +15,6 @@ use std::{
use futures_core::{Future, Stream};
use futures_util::{SinkExt, StreamExt, future::join_all};
use parking_lot::Mutex;
use thiserror::Error;
use tokio::{
select,
@ -57,6 +56,11 @@ const PING_TIMEOUT: Duration = Duration::from_secs(3);
const RECONNECT_INTERVAL: Duration = Duration::from_secs(10);
const DEALER_REQUEST_HANDLERS_POISON_MSG: &str =
"dealer request handlers mutex should not be poisoned";
const DEALER_MESSAGE_HANDLERS_POISON_MSG: &str =
"dealer message handlers mutex should not be poisoned";
struct Response {
pub success: bool,
}
@ -350,6 +354,7 @@ impl DealerShared {
if self
.message_handlers
.lock()
.expect(DEALER_MESSAGE_HANDLERS_POISON_MSG)
.retain(split, &mut |tx| tx.send(msg.clone()).is_ok())
{
return;
@ -387,7 +392,10 @@ impl DealerShared {
return;
};
let handler_map = self.request_handlers.lock();
let handler_map = self
.request_handlers
.lock()
.expect(DEALER_REQUEST_HANDLERS_POISON_MSG);
if let Some(handler) = handler_map.get(split) {
handler.handle_request(payload_request, responder);
@ -425,21 +433,51 @@ impl Dealer {
where
H: RequestHandler,
{
add_handler(&mut self.shared.request_handlers.lock(), uri, handler)
add_handler(
&mut self
.shared
.request_handlers
.lock()
.expect(DEALER_REQUEST_HANDLERS_POISON_MSG),
uri,
handler,
)
}
pub fn remove_handler(&self, uri: &str) -> Option<Box<dyn RequestHandler>> {
remove_handler(&mut self.shared.request_handlers.lock(), uri)
remove_handler(
&mut self
.shared
.request_handlers
.lock()
.expect(DEALER_REQUEST_HANDLERS_POISON_MSG),
uri,
)
}
pub fn subscribe(&self, uris: &[&str]) -> Result<Subscription, Error> {
subscribe(&mut self.shared.message_handlers.lock(), uris)
subscribe(
&mut self
.shared
.message_handlers
.lock()
.expect(DEALER_MESSAGE_HANDLERS_POISON_MSG),
uris,
)
}
pub fn handles(&self, uri: &str) -> bool {
handles(
&self.shared.request_handlers.lock(),
&self.shared.message_handlers.lock(),
&self
.shared
.request_handlers
.lock()
.expect(DEALER_REQUEST_HANDLERS_POISON_MSG),
&self
.shared
.message_handlers
.lock()
.expect(DEALER_MESSAGE_HANDLERS_POISON_MSG),
uri,
)
}

9
core/src/http_client.rs
View file

@ -1,5 +1,4 @@
use std::{
collections::HashMap,
sync::OnceLock,
time::{Duration, Instant},
};
@ -7,7 +6,8 @@ use std::{
use bytes::Bytes;
use futures_util::{FutureExt, future::IntoStream};
use governor::{
Quota, RateLimiter, clock::MonotonicClock, middleware::NoOpMiddleware, state::InMemoryState,
Quota, RateLimiter, clock::MonotonicClock, middleware::NoOpMiddleware,
state::keyed::DefaultKeyedStateStore,
};
use http::{Uri, header::HeaderValue};
use http_body_util::{BodyExt, Full};
@ -18,7 +18,6 @@ use hyper_util::{
rt::TokioExecutor,
};
use nonzero_ext::nonzero;
use parking_lot::Mutex;
use thiserror::Error;
use url::Url;
@ -100,10 +99,8 @@ pub struct HttpClient {
proxy_url: Option<Url>,
hyper_client: OnceLock<HyperClient>,
// while the DashMap variant is more performant, our level of concurrency
// is pretty low so we can save pulling in that extra dependency
rate_limiter:
RateLimiter<String, Mutex<HashMap<String, InMemoryState>>, MonotonicClock, NoOpMiddleware>,
RateLimiter<String, DefaultKeyedStateStore<String>, MonotonicClock, NoOpMiddleware>,
}
impl HttpClient {

197
core/src/session.rs
View file

@ -4,8 +4,7 @@ use std::{
io,
pin::Pin,
process::exit,
sync::OnceLock,
sync::{Arc, Weak},
sync::{Arc, OnceLock, RwLock, Weak},
task::{Context, Poll},
time::{Duration, SystemTime, UNIX_EPOCH},
};
@ -34,7 +33,6 @@ use futures_core::TryStream;
use futures_util::StreamExt;
use librespot_protocol::authentication::AuthenticationType;
use num_traits::FromPrimitive;
use parking_lot::RwLock;
use pin_project_lite::pin_project;
use quick_xml::events::Event;
use thiserror::Error;
@ -45,6 +43,8 @@ use tokio::{
use tokio_stream::wrappers::UnboundedReceiverStream;
use uuid::Uuid;
const SESSION_DATA_POISON_MSG: &str = "session data rwlock should not be poisoned";
#[derive(Debug, Error)]
pub enum SessionError {
#[error(transparent)]
@ -338,7 +338,11 @@ impl Session {
}
pub fn time_delta(&self) -> i64 {
self.0.data.read().time_delta
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.time_delta
}
pub fn spawn<T>(&self, task: T)
@ -388,15 +392,32 @@ impl Session {
// you need more fields at once, in which case this can spare multiple `read`
// locks.
pub fn user_data(&self) -> UserData {
self.0.data.read().user_data.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.clone()
}
pub fn session_id(&self) -> String {
self.0.data.read().session_id.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.session_id
.clone()
}
pub fn set_session_id(&self, session_id: &str) {
session_id.clone_into(&mut self.0.data.write().session_id);
session_id.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.session_id,
);
}
pub fn device_id(&self) -> &str {
@ -404,63 +425,155 @@ impl Session {
}
pub fn client_id(&self) -> String {
self.0.data.read().client_id.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.client_id
.clone()
}
pub fn set_client_id(&self, client_id: &str) {
client_id.clone_into(&mut self.0.data.write().client_id);
client_id.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.client_id,
);
}
pub fn client_name(&self) -> String {
self.0.data.read().client_name.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.client_name
.clone()
}
pub fn set_client_name(&self, client_name: &str) {
client_name.clone_into(&mut self.0.data.write().client_name);
client_name.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.client_name,
);
}
pub fn client_brand_name(&self) -> String {
self.0.data.read().client_brand_name.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.client_brand_name
.clone()
}
pub fn set_client_brand_name(&self, client_brand_name: &str) {
client_brand_name.clone_into(&mut self.0.data.write().client_brand_name);
client_brand_name.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.client_brand_name,
);
}
pub fn client_model_name(&self) -> String {
self.0.data.read().client_model_name.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.client_model_name
.clone()
}
pub fn set_client_model_name(&self, client_model_name: &str) {
client_model_name.clone_into(&mut self.0.data.write().client_model_name);
client_model_name.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.client_model_name,
);
}
pub fn connection_id(&self) -> String {
self.0.data.read().connection_id.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.connection_id
.clone()
}
pub fn set_connection_id(&self, connection_id: &str) {
connection_id.clone_into(&mut self.0.data.write().connection_id);
connection_id.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.connection_id,
);
}
pub fn username(&self) -> String {
self.0.data.read().user_data.canonical_username.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.canonical_username
.clone()
}
pub fn set_username(&self, username: &str) {
username.clone_into(&mut self.0.data.write().user_data.canonical_username);
username.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.canonical_username,
);
}
pub fn auth_data(&self) -> Vec<u8> {
self.0.data.read().auth_data.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.auth_data
.clone()
}
pub fn set_auth_data(&self, auth_data: &[u8]) {
auth_data.clone_into(&mut self.0.data.write().auth_data);
auth_data.clone_into(
&mut self
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.auth_data,
);
}
pub fn country(&self) -> String {
self.0.data.read().user_data.country.clone()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.country
.clone()
}
pub fn filter_explicit_content(&self) -> bool {
@ -489,6 +602,7 @@ impl Session {
self.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.attributes
.insert(key.to_owned(), value.to_owned())
@ -497,11 +611,24 @@ impl Session {
pub fn set_user_attributes(&self, attributes: UserAttributes) {
Self::check_catalogue(&attributes);
self.0.data.write().user_data.attributes.extend(attributes)
self.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.attributes
.extend(attributes)
}
pub fn get_user_attribute(&self, key: &str) -> Option<String> {
self.0.data.read().user_data.attributes.get(key).cloned()
self.0
.data
.read()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.attributes
.get(key)
.cloned()
}
fn weak(&self) -> SessionWeak {
@ -510,13 +637,13 @@ impl Session {
pub fn shutdown(&self) {
debug!("Shutdown: Invalidating session");
self.0.data.write().invalid = true;
self.0.data.write().expect(SESSION_DATA_POISON_MSG).invalid = true;
self.mercury().shutdown();
self.channel().shutdown();
}
pub fn is_invalid(&self) -> bool {
self.0.data.read().invalid
self.0.data.read().expect(SESSION_DATA_POISON_MSG).invalid
}
}
@ -643,7 +770,7 @@ where
.unwrap_or(Duration::ZERO)
.as_secs() as i64;
{
let mut data = session.0.data.write();
let mut data = session.0.data.write().expect(SESSION_DATA_POISON_MSG);
data.time_delta = server_timestamp.saturating_sub(timestamp);
}
@ -668,7 +795,13 @@ where
Some(CountryCode) => {
let country = String::from_utf8(data.as_ref().to_owned())?;
info!("Country: {country:?}");
session.0.data.write().user_data.country = country;
session
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.country = country;
Ok(())
}
Some(StreamChunkRes) | Some(ChannelError) => session.channel().dispatch(cmd, data),
@ -713,7 +846,13 @@ where
trace!("Received product info: {user_attributes:#?}");
Session::check_catalogue(&user_attributes);
session.0.data.write().user_data.attributes = user_attributes;
session
.0
.data
.write()
.expect(SESSION_DATA_POISON_MSG)
.user_data
.attributes = user_attributes;
Ok(())
}
Some(SecretBlock)

8
discovery/Cargo.toml
View file

@ -32,7 +32,7 @@ ctr = "0.9"
dns-sd = { version = "0.1", optional = true }
form_urlencoded = "1.2"
futures-core = "0.3"
futures-util = "0.3"
futures-util = { version = "0.3", default-features = false, features = ["std"] }
hmac = "0.12"
http-body-util = "0.1"
hyper = { version = "1.6", features = ["http1"] }
@ -43,7 +43,7 @@ hyper-util = { version = "0.1", features = [
] }
libmdns = { version = "0.10", optional = true }
log = "0.4"
rand = "0.9"
rand = { version = "0.9", default-features = false, features = ["thread_rng"] }
serde = { version = "1", default-features = false, features = [
"derive",
], optional = true }
@ -51,7 +51,7 @@ serde_repr = "0.1"
serde_json = "1.0"
sha1 = "0.10"
thiserror = "2"
tokio = { version = "1", features = ["parking_lot", "sync", "rt"] }
tokio = { version = "1", features = ["sync", "rt"] }
zbus = { version = "5", default-features = false, features = [
"tokio",
], optional = true }
@ -59,4 +59,4 @@ zbus = { version = "5", default-features = false, features = [
[dev-dependencies]
futures = "0.3"
hex = "0.4"
tokio = { version = "1", features = ["macros", "parking_lot", "rt"] }
tokio = { version = "1", features = ["macros", "rt"] }

14
playback/Cargo.toml
View file

@ -51,18 +51,12 @@ librespot-audio = { version = "0.7.1", path = "../audio", default-features = fal
librespot-core = { version = "0.7.1", path = "../core", default-features = false }
librespot-metadata = { version = "0.7.1", path = "../metadata", default-features = false }
portable-atomic = "1"
futures-util = "0.3"
futures-util = { version = "0.3", default-features = false, features = ["std"] }
log = "0.4"
parking_lot = { version = "0.12", features = ["deadlock_detection"] }
portable-atomic = "1"
shell-words = "1.1"
thiserror = "2"
tokio = { version = "1", features = [
"parking_lot",
"rt",
"rt-multi-thread",
"sync",
] }
tokio = { version = "1", features = ["rt-multi-thread", "sync"] }
zerocopy = { version = "0.8", features = ["derive"] }
# Backends
@ -97,5 +91,5 @@ symphonia = { version = "0.5", default-features = false, features = [
ogg = { version = "0.9", optional = true }
# Dithering
rand = { version = "0.9", features = ["small_rng"] }
rand = { version = "0.9", default-features = false, features = ["small_rng"] }
rand_distr = "0.5"

29
playback/src/audio_backend/gstreamer.rs
View file

@ -1,3 +1,5 @@
use std::sync::{Arc, Mutex};
use gstreamer::{
State,
event::{FlushStart, FlushStop},
@ -8,8 +10,7 @@ use gstreamer as gst;
use gstreamer_app as gst_app;
use gstreamer_audio as gst_audio;
use parking_lot::Mutex;
use std::sync::Arc;
const GSTREAMER_ASYNC_ERROR_POISON_MSG: &str = "gstreamer async error mutex should not be poisoned";
use super::{Open, Sink, SinkAsBytes, SinkError, SinkResult};
@ -97,7 +98,9 @@ impl Open for GstreamerSink {
gst::MessageView::Eos(_) => {
println!("gst signaled end of stream");
let mut async_error_storage = async_error_clone.lock();
let mut async_error_storage = async_error_clone
.lock()
.expect(GSTREAMER_ASYNC_ERROR_POISON_MSG);
*async_error_storage = Some(String::from("gst signaled end of stream"));
}
gst::MessageView::Error(err) => {
@ -108,7 +111,9 @@ impl Open for GstreamerSink {
err.debug()
);
let mut async_error_storage = async_error_clone.lock();
let mut async_error_storage = async_error_clone
.lock()
.expect(GSTREAMER_ASYNC_ERROR_POISON_MSG);
*async_error_storage = Some(format!(
"Error from {:?}: {} ({:?})",
err.src().map(|s| s.path_string()),
@ -138,7 +143,10 @@ impl Open for GstreamerSink {
impl Sink for GstreamerSink {
fn start(&mut self) -> SinkResult<()> {
*self.async_error.lock() = None;
*self
.async_error
.lock()
.expect(GSTREAMER_ASYNC_ERROR_POISON_MSG) = None;
self.appsrc.send_event(FlushStop::new(true));
self.bufferpool
.set_active(true)
@ -150,7 +158,10 @@ impl Sink for GstreamerSink {
}
fn stop(&mut self) -> SinkResult<()> {
*self.async_error.lock() = None;
*self
.async_error
.lock()
.expect(GSTREAMER_ASYNC_ERROR_POISON_MSG) = None;
self.appsrc.send_event(FlushStart::new());
self.pipeline
.set_state(State::Paused)
@ -173,7 +184,11 @@ impl Drop for GstreamerSink {
impl SinkAsBytes for GstreamerSink {
#[inline]
fn write_bytes(&mut self, data: &[u8]) -> SinkResult<()> {
if let Some(async_error) = &*self.async_error.lock() {
if let Some(async_error) = &*self
.async_error
.lock()
.expect(GSTREAMER_ASYNC_ERROR_POISON_MSG)
{
return Err(SinkError::OnWrite(async_error.to_string()));
}

10
playback/src/player.rs
View file

@ -6,6 +6,7 @@ use std::{
mem,
pin::Pin,
process::exit,
sync::Mutex,
sync::{
Arc,
atomic::{AtomicUsize, Ordering},
@ -32,7 +33,6 @@ use futures_util::{
stream::futures_unordered::FuturesUnordered,
};
use librespot_metadata::track::Tracks;
use parking_lot::Mutex;
use symphonia::core::io::MediaSource;
use tokio::sync::{mpsc, oneshot};
@ -46,6 +46,8 @@ pub const PCM_AT_0DBFS: f64 = 1.0;
// otherwise expect in Vorbis comments. This packet isn't well-formed and players may balk at it.
const SPOTIFY_OGG_HEADER_END: u64 = 0xa7;
const LOAD_HANDLES_POISON_MSG: &str = "load handles mutex should not be poisoned";
pub type PlayerResult = Result<(), Error>;
pub struct Player {
@ -2281,11 +2283,11 @@ impl PlayerInternal {
let _ = result_tx.send(data);
}
let mut load_handles = load_handles_clone.lock();
let mut load_handles = load_handles_clone.lock().expect(LOAD_HANDLES_POISON_MSG);
load_handles.remove(&thread::current().id());
});
let mut load_handles = self.load_handles.lock();
let mut load_handles = self.load_handles.lock().expect(LOAD_HANDLES_POISON_MSG);
load_handles.insert(load_handle.thread().id(), load_handle);
result_rx.map_err(|_| ())
@ -2320,7 +2322,7 @@ impl Drop for PlayerInternal {
let handles: Vec<thread::JoinHandle<()>> = {
// waiting for the thread while holding the mutex would result in a deadlock
let mut load_handles = self.load_handles.lock();
let mut load_handles = self.load_handles.lock().expect(LOAD_HANDLES_POISON_MSG);
load_handles
.drain()