leyoda/cosmic-applets
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

audio: Don't recreate mpris proxies and streams

Browse source 
Fixes https://github.com/pop-os/cosmic-applets/issues/306 and should
overall be more correct and performant.
This commit is contained in:
Ian Douglas Scott 2024-04-03 20:45:30 -07:00 • committed by Michael Murphy
parent 56d3a754e2
commit 1475f8f32b
3 changed files with 168 additions and 172 deletions
Download patch file Download diff file Expand all files Collapse all files

13
Cargo.lock generated
View file

@ -1179,7 +1179,7 @@ dependencies = [
[[package]]
name = "cosmic-dbus-networkmanager"
version = "0.1.0"
source = "git+https://github.com/pop-os/dbus-settings-bindings#8b9767f6cedede2def12941ce89e14bfcd913aeb"
source = "git+https://github.com/pop-os/dbus-settings-bindings#c81f428acec4c8633efb62b4f8284202dad9a492"
dependencies = [
"bitflags 2.5.0",
"derive_builder",
@ -3534,8 +3534,9 @@ dependencies = [
[[package]]
name = "mpris2-zbus"
version = "0.1.0"
source = "git+https://github.com/pop-os/dbus-settings-bindings#8b9767f6cedede2def12941ce89e14bfcd913aeb"
source = "git+https://github.com/pop-os/dbus-settings-bindings#c81f428acec4c8633efb62b4f8284202dad9a492"
dependencies = [
"futures-util",
"serde",
"thiserror",
"time",
@ -4943,7 +4944,7 @@ dependencies = [
[[package]]
name = "switcheroo-control"
version = "0.1.0"
source = "git+https://github.com/pop-os/dbus-settings-bindings#8b9767f6cedede2def12941ce89e14bfcd913aeb"
source = "git+https://github.com/pop-os/dbus-settings-bindings#c81f428acec4c8633efb62b4f8284202dad9a492"
dependencies = [
"zbus",
]
@ -5842,7 +5843,7 @@ dependencies = [
"js-sys",
"log",
"naga",
"parking_lot 0.12.1",
"parking_lot 0.11.2",
"profiling",
"raw-window-handle 0.6.0",
"smallvec",
@ -5869,7 +5870,7 @@ dependencies = [
"log",
"naga",
"once_cell",
"parking_lot 0.12.1",
"parking_lot 0.11.2",
"profiling",
"raw-window-handle 0.6.0",
"rustc-hash",
@ -5909,7 +5910,7 @@ dependencies = [
"naga",
"objc",
"once_cell",
"parking_lot 0.12.1",
"parking_lot 0.11.2",
"profiling",
"range-alloc",
"raw-window-handle 0.6.0",

2
cosmic-applet-audio/src/lib.rs
View file

@ -50,7 +50,7 @@ const PLAY: &str = "media-playback-start-symbolic";
pub fn run() -> cosmic::iced::Result {
localize();
cosmic::applet::run::<Audio>(true, ())
cosmic::applet::run::<Audio>(false, ())
}
#[derive(Default)]

325
cosmic-applet-audio/src/mpris_subscription.rs
View file

@ -1,16 +1,21 @@
use std::{borrow::Cow, fmt::Debug, hash::Hash, path::PathBuf};
use std::{borrow::Cow, collections::HashMap, fmt::Debug, hash::Hash, path::PathBuf};
use cosmic::{
iced::{self, subscription},
iced_futures::futures::{self, SinkExt, StreamExt},
iced_futures::futures::{self, future::OptionFuture, FutureExt, SinkExt, StreamExt},
};
use mpris2_zbus::{
enumerator,
media_player::MediaPlayer,
player::{PlaybackStatus, Player},
};
use tokio::join;
use urlencoding::decode;
use zbus::{fdo::DBusProxy, Connection};
use zbus::{
names::{BusName, OwnedBusName},
zvariant::OwnedValue,
Connection,
};
#[derive(Clone, Debug)]
pub struct PlayerStatus {
@ -77,19 +82,30 @@ pub fn mpris_subscription<I: 'static + Hash + Copy + Send + Sync + Debug>(
id: I,
) -> iced::Subscription<MprisUpdate> {
subscription::channel(id, 50, move |mut output| async move {
let mut state = State::Setup;
loop {
state = update(state, &mut output).await;
}
run(&mut output).await;
let _ = output.send(MprisUpdate::Finished).await;
futures::future::pending().await
})
}
#[derive(Debug)]
pub enum State {
Setup,
Player(Player, DBusProxy<'static>),
Finished,
#[derive(Clone, Debug)]
struct MprisPlayer {
player: Player,
#[allow(dead_code)]
media_player: MediaPlayer,
}
impl MprisPlayer {
async fn new(conn: &Connection, name: OwnedBusName) -> mpris2_zbus::error::Result<Self> {
Ok(Self {
player: Player::new(conn, name.clone()).await?,
media_player: MediaPlayer::new(conn, name).await?,
})
}
fn name(&self) -> &BusName {
self.player.destination()
}
}
#[derive(Clone, Debug)]
@ -107,162 +123,145 @@ pub enum MprisRequest {
Previous,
}
async fn update(state: State, output: &mut futures::channel::mpsc::Sender<MprisUpdate>) -> State {
match state {
State::Setup => {
let Ok(conn) = Connection::session().await else {
tracing::error!("Failed to connect to session bus.");
_ = output.send(MprisUpdate::Finished).await;
return State::Finished;
};
let mut players = mpris2_zbus::media_player::MediaPlayer::new_all(&conn)
.await
.unwrap_or_else(|_| Vec::new());
let Ok(dbus_proxy) = zbus::fdo::DBusProxy::builder(&conn)
.path("/org/freedesktop/DBus")
.unwrap()
.build()
.await
else {
tracing::error!("Failed to create dbus proxy.");
return State::Finished;
};
if players.is_empty() {
let Ok(mut stream) = dbus_proxy.receive_name_owner_changed().await else {
tracing::error!("Failed to receive name owner changed signal.");
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
// restart from the beginning
return State::Setup;
};
while let Some(c) = stream.next().await {
if let Ok(args) = c.args() {
if args.name.contains("org.mpris.MediaPlayer2") {
if let Ok(p) =
MediaPlayer::new(&conn, args.name().to_owned().into()).await
{
players.push(p);
}
break;
}
}
struct State {
conn: Connection,
enumerator_stream:
Box<dyn futures::Stream<Item = zbus::Result<enumerator::Event>> + Unpin + Send>,
players: Vec<MprisPlayer>,
active_player: Option<MprisPlayer>,
active_player_metadata_stream:
Option<zbus::PropertyStream<'static, HashMap<String, OwnedValue>>>,
any_player_state_stream: futures::stream::SelectAll<zbus::PropertyStream<'static, String>>,
}
impl State {
async fn new() -> Result<Self, zbus::Error> {
let conn = Connection::session().await?;
let enumerator = enumerator::Enumerator::new(&conn).await?;
let enumerator_stream = enumerator.receive_changes().await?;
let player_names = enumerator.players().await?;
let mut players = Vec::with_capacity(player_names.len());
for name in player_names {
match MprisPlayer::new(&conn, name).await {
Ok(player) => {
players.push(player);
}
Err(err) => {
tracing::error!("Failed to add player: {}", err);
}
}
let Some(player) = find_active(players).await else {
tracing::error!("Failed to find active media player.");
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
return State::Setup;
};
let Some(player_status) = PlayerStatus::new(player.clone()).await else {
tracing::error!("Failed to get player status.");
return State::Setup;
};
_ = output.send(MprisUpdate::Player(player_status)).await;
State::Player(player, dbus_proxy)
}
State::Player(player, dbus_proxy) => {
let Ok(mut name_owner_changed) = player.receive_owner_changed().await else {
tracing::error!("Failed to receive owner changed signal.");
// restart from the beginning
return State::Setup;
};
let mut metadata_changed = player.receive_metadata_changed().await;
let Ok(mut new_mpris) = dbus_proxy.receive_name_owner_changed().await else {
tracing::error!("Failed to receive name owner changed signal.");
// restart from the beginning
return State::Setup;
};
let conn = player.connection();
let media_players = mpris2_zbus::media_player::MediaPlayer::new_all(conn)
.await
.unwrap_or_else(|_| Vec::new());
let mut players = Vec::with_capacity(media_players.len());
for p in media_players {
if let Ok(p) = p.player().await {
players.push(p);
}
// pre-sort by path so that the same player is always selected
players.sort_by(|a, b| a.name().cmp(&b.name()));
let mut state = Self {
conn,
enumerator_stream: Box::new(enumerator_stream),
players,
active_player: None,
active_player_metadata_stream: None,
any_player_state_stream: futures::stream::select_all(Vec::new()),
};
state.update_active_player().await;
state.update_any_player_state_stream().await;
Ok(state)
}
async fn add_player(&mut self, name: OwnedBusName) {
let player = match MprisPlayer::new(&self.conn, name).await {
Ok(player) => player,
Err(err) => {
tracing::error!("Failed to add player: {}", err);
return;
}
};
self.players.push(player);
self.players.sort_by(|a, b| a.name().cmp(&b.name()));
self.update_any_player_state_stream().await;
}
loop {
let mut listeners = Vec::with_capacity(players.len());
for p in &players {
listeners.push(p.receive_playback_status_changed().await);
}
let mut player_state_changed_list = Vec::with_capacity(listeners.len());
for l in &mut listeners {
player_state_changed_list.push(Box::pin(async move {
let changed = l.next().await;
if let Some(c) = changed {
c.get().await.ok()
} else {
tracing::error!("Failed to receive playback status changed signal.");
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
None
}
}));
}
let any_player_state_changed =
futures::future::select_all(player_state_changed_list);
let keep_going = tokio::select! {
m = metadata_changed.next() => {
m.is_some()
},
n = name_owner_changed.next() => {
n.map(|n| n.is_some()).unwrap_or_default()
},
_ = new_mpris.next() => {
true
},
_ = any_player_state_changed => {
true
},
};
if !keep_going {
break;
}
if let Some(update) = PlayerStatus::new(player.clone()).await {
if matches!(update.status, PlaybackStatus::Stopped) {
break;
}
// if paused check if any players are playing
// if they are, break
if !matches!(update.status, PlaybackStatus::Playing) {
let conn = player.connection();
let players = mpris2_zbus::media_player::MediaPlayer::new_all(conn)
.await
.unwrap_or_else(|_| Vec::new());
if let Some(active) = find_active(players).await {
if active.destination() != player.destination() {
break;
}
}
}
_ = output.send(MprisUpdate::Player(update)).await;
} else {
break;
}
}
_ = output.send(MprisUpdate::Setup).await;
State::Setup
async fn remove_player(&mut self, name: OwnedBusName) {
if let Some(idx) = self.players.iter().position(|p| p.name() == &name) {
self.players.remove(idx);
}
State::Finished => iced::futures::future::pending().await,
self.update_any_player_state_stream().await;
}
async fn update_active_player(&mut self) {
let new_active_player = find_active(&self.players).await;
if self.active_player.as_ref().map(|p| p.name()) != new_active_player.map(|p| p.name()) {
self.active_player = new_active_player.cloned();
if let Some(player) = new_active_player {
self.active_player_metadata_stream =
Some(player.player.receive_metadata_changed().await);
} else {
self.active_player_metadata_stream = None;
}
}
}
async fn update_any_player_state_stream(&mut self) {
let mut listeners = Vec::with_capacity(self.players.len());
for p in &self.players {
listeners.push(p.player.receive_playback_status_changed().await);
}
self.any_player_state_stream = futures::stream::select_all(listeners);
}
}
async fn find_active(mut players: Vec<MediaPlayer>) -> Option<Player> {
// pre-sort by path so that the same player is always selected
players.sort_by(|a, b| {
let a = a.destination();
let b = b.destination();
a.cmp(b)
});
let mut best = (0, None);
async fn run(output: &mut futures::channel::mpsc::Sender<MprisUpdate>) {
let mut state = match State::new().await {
Ok(state) => state,
Err(err) => {
tracing::error!("Faile do monitor for mpris clients: {}", err);
return;
}
};
loop {
if let Some(player) = &state.active_player {
if let Some(player_status) = PlayerStatus::new(player.player.clone()).await {
_ = output.send(MprisUpdate::Player(player_status)).await;
} else {
tracing::error!("Failed to get player status.");
};
} else {
let _ = output.send(MprisUpdate::Setup).await;
}
let metadata_changed_next = OptionFuture::from(
state
.active_player_metadata_stream
.as_mut()
.map(|s| s.next()),
);
tokio::select! {
_ = metadata_changed_next, if state.active_player.is_some() => {
},
event = state.enumerator_stream.next() => {
match dbg!(event) {
Some(Ok(enumerator::Event::Add(name))) => state.add_player(name).await,
Some(Ok(enumerator::Event::Remove(name))) => state.remove_player(name).await,
Some(Err(err)) => {
tracing::error!("Error listening for mpris clients: {:?}", err);
return;
}
None => {}
}
state.update_active_player().await;
}
_ = state.any_player_state_stream.next(), if !state.players.is_empty() => {
state.update_active_player().await;
},
};
}
}
async fn find_active<'a>(players: &'a Vec<MprisPlayer>) -> Option<&'a MprisPlayer> {
let mut best = (0, None::<&'a MprisPlayer>);
let eval = |p: Player| async move {
let v = {
let status = p.playback_status().await;
@ -277,12 +276,8 @@ async fn find_active(mut players: Vec<MediaPlayer>) -> Option<Player> {
v + p.metadata().await.is_ok() as i32
};
for p in players {
let p = match p.player().await {
Ok(p) => p,
Err(_) => continue,
};
let v = eval(p.clone()).await;
for p in players.iter() {
let v = eval(p.player.clone()).await;
if v > best.0 {
best = (v, Some(p));
}