mpris: Add helper for enumerating players, watching for new players (#12)

Cargo.toml

@@ -13,6 +13,7 @@ members = [
 ]
 
 [workspace.dependencies]
+futures-util = "0.3.30"
 serde = { version = "1.0", features = ["derive"] }
 thiserror = "1.0"
 time = { version = "0.3", features = ["parsing"] }

mpris2/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 license = "MPL-2.0"
 
 [dependencies]
+futures-util.workspace = true
 serde.workspace = true
 thiserror.workspace = true
 time.workspace = true

mpris2/examples/enumerate.rs

@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use futures_util::stream::StreamExt;
+use mpris2_zbus::enumerator::Enumerator;
+
+#[tokio::main]
+async fn main() -> zbus::Result<()> {
+	let connection = zbus::Connection::session().await?;
+	let enumerator = Enumerator::new(&connection).await?;
+	let mut stream = enumerator.receive_changes().await?;
+	println!("players: {:?}", enumerator.players().await?);
+	while let Some(event) = stream.next().await {
+		println!("change: {:?}", event?);
+	}
+	Ok(())
+}

mpris2/src/enumerator.rs

@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use futures_util::{
+	stream::{self, StreamExt},
+	Stream,
+};
+use zbus::{fdo::DBusProxy, names::OwnedBusName, Connection};
+
+#[derive(Clone, Debug)]
+pub enum Event {
+	Remove(OwnedBusName),
+	Add(OwnedBusName),
+}
+
+/// Helper to list mpris players on DBus bus, and watch for addition/removal.
+///
+/// Uses `org.freedesktop.DBus` to watch for clients that claim names starting with
+/// `org.mpris.MediaPlayer2.`
+pub struct Enumerator {
+	proxy: DBusProxy<'static>,
+}
+
+impl Enumerator {
+	pub async fn new(connection: &Connection) -> zbus::Result<Self> {
+		Ok(Self {
+			proxy: DBusProxy::builder(connection)
+				.path("/org/freedesktop/DBus")?
+				.build()
+				.await?,
+		})
+	}
+
+	/// Returns a stream that is signalled when an mpris client is added or removed
+	pub async fn receive_changes(
+		&self,
+	) -> zbus::Result<impl Stream<Item = zbus::Result<Event>> + Unpin> {
+		let stream = self.proxy.receive_name_owner_changed().await?;
+		Ok(stream
+			.filter_map(|signal| {
+				Box::pin(async move {
+					let args = match signal.args() {
+						Ok(args) => args,
+						Err(err) => {
+							return Some(stream::iter(Some(Err(err)).into_iter().chain(None)));
+						}
+					};
+					if args.name().contains("org.mpris.MediaPlayer2.") {
+						let remove = args
+							.old_owner
+							.as_ref()
+							.map(|_| Ok(Event::Remove(args.name().to_owned().into())));
+						let add = args
+							.new_owner
+							.as_ref()
+							.map(|_| Ok(Event::Add(args.name().to_owned().into())));
+						Some(stream::iter(remove.into_iter().chain(add)))
+					} else {
+						None
+					}
+				})
+			})
+			.flatten())
+	}
+
+	/// Get names of all mpris players currently on the bus
+	pub async fn players(&self) -> zbus::Result<Vec<OwnedBusName>> {
+		let mut players = self.proxy.list_names().await?;
+		players.retain(|name| name.starts_with("org.mpris.MediaPlayer2."));
+		Ok(players)
+	}
+}

mpris2/src/lib.rs

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: MPL-2.0
 pub mod bindings;
+pub mod enumerator;
 pub mod error;
 pub mod media_player;
 pub mod metadata;

mpris2/src/media_player.rs

@@ -4,13 +4,14 @@ use crate::{
 		media_player::MediaPlayer2Proxy, player::PlayerProxy, playlist::PlaylistsProxy,
 		track_list::TrackListProxy,
 	},
+	enumerator::Enumerator,
 	error::{Error, Result},
 	player::Player,
 	playlists::Playlists,
 	track_list::TrackList,
 };
 use std::ops::Deref;
-use zbus::{fdo::DBusProxy, names::OwnedBusName, Connection};
+use zbus::{names::OwnedBusName, Connection};
 
 #[derive(Debug, Clone)]
 pub struct MediaPlayer {
@@ -30,17 +31,7 @@ impl MediaPlayer {
 
 	/// Gets the names of all the MPRIS players that are available on the current session.
 	pub async fn available_players(connection: &Connection) -> Result<Vec<OwnedBusName>> {
-		let dbus = DBusProxy::builder(connection)
-			.path("/org/freedesktop/DBus")?
-			.build()
-			.await?;
-		let mut players = Vec::new();
-		for name in dbus.list_names().await? {
-			if name.starts_with("org.mpris.MediaPlayer2.") {
-				players.push(name);
-			}
-		}
-		Ok(players)
+		Ok(Enumerator::new(connection).await?.players().await?)
 	}
 
 	/// Gets a new instance of all the MPRIS players that are available on the current session.