diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 779544a..6eeda14 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,6 +52,23 @@ jobs: - name: Tests run: cargo test --workspace + - name: GTK application identity + run: | + export XDG_RUNTIME_DIR=$(mktemp -d) + export XDG_CONFIG_HOME=$(mktemp -d) + export GDK_BACKEND=broadway BROADWAY_DISPLAY=:5 + export PIPEWIRE_REMOTE=pipedeck-ui-no-server + gtk4-broadwayd :5 > /tmp/pipedeck-broadway.log 2>&1 & + broadway_pid=$! + trap 'kill "$broadway_pid" 2>/dev/null || true' EXIT + for attempt in $(seq 50); do + [ -S "$XDG_RUNTIME_DIR/broadway6.socket" ] && break + sleep 0.1 + done + dbus-run-session cargo test -p pipedeck --bin pipedeck \ + channel_dialog::tests::legacy_assignments_follow_late_identity \ + -- --ignored --test-threads=1 + # What software centres read of Pipedeck, as they read it. - name: Software centre listing run: appstreamcli validate --no-net crates/pipedeck/data/dev._2c2t.Pipedeck.metainfo.xml diff --git a/CHANGELOG.md b/CHANGELOG.md index c7879c5..71d24f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ follow [Semantic Versioning](https://semver.org/). ### Fixed +- Match Flatpak playback streams using client portal ids, retaining legacy + aliases for routing, focus matching, playing status and picker deduplication. +- Preserve remaining application assignments when an alias or row is removed. +- Explicitly retire loopbacks when dock devices disappear and restore them + on reconnection, including a replacement with the same name. + - Pipedeck crashed when a device it played to or read from went away, a dock unplugged or a USB card waking up again, then a mix's output was switched off or a channel removed. The loopbacks that went with the diff --git a/README.md b/README.md index 9d9cd45..41e2bd5 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,8 @@ with its icon, under `~/.local/share` on its first start. virtual output, which any application can pick in its audio settings, or a capture device such as a microphone. You can attach applications to a channel: Pipedeck moves their audio there whenever they start playing. +Flatpak applications are matched by their application id, so an app picked +from the installed list follows the same channel when it starts playing. **Mixes** (columns) are where sound goes. Each mix is an input device named after it: OBS, Discord or a browser list it with the microphones. A mix can diff --git a/crates/pipedeck-engine/examples/smoke.rs b/crates/pipedeck-engine/examples/smoke.rs index 43ae04a..152f980 100644 --- a/crates/pipedeck-engine/examples/smoke.rs +++ b/crates/pipedeck-engine/examples/smoke.rs @@ -4,6 +4,7 @@ //! the mix to a real device, then tears everything down and checks that no //! Pipedeck node is left behind. Run with //! `cargo run -p pipedeck-engine --example smoke` (needs a running PipeWire). +//! Pass `-- --routing-regressions` to run only overlapping-alias cases. use std::process::{Command as Process, ExitCode}; use std::sync::mpsc; @@ -421,7 +422,212 @@ const PREFIX: &str = "pipedeck-smoke"; /// nor the user's moved by this one. const VOICE_APP: &str = "pipedeck-smoke-vesktop"; +/// A silent application with identity on its client, as Flatpak supplies +/// it. Keeping it off the node reproduces the installed-app routing bug. +fn identity_player(args: &[String]) -> Result<(), pipewire::Error> { + use libspa::param::audio::{AudioFormat, AudioInfoRaw, MAX_CHANNELS}; + use libspa::pod::{serialize::PodSerializer, Object, Pod, Value}; + use libspa::utils::{Direction, SpaTypes}; + use pipewire::properties::properties; + use pipewire::stream::{StreamBox, StreamFlags}; + + pipewire::init(); + let mainloop = pipewire::main_loop::MainLoopRc::new(None)?; + let context = pipewire::context::ContextRc::new(&mainloop, None)?; + let core = context.connect_rc(Some(properties! { + "pipewire.access.portal.app_id" => args[0].as_str(), + "application.name" => args[1].as_str(), + "application.process.binary" => args[2].as_str(), + }))?; + let stream = StreamBox::new( + &core, + "identity-player", + properties! { + "media.type" => "Audio", + "media.category" => "Playback", + "media.role" => "Music", + "application.name" => args[1].as_str(), + "application.process.binary" => args[2].as_str(), + "node.name" => args[3].as_str(), + }, + )?; + let _listener = stream + .add_local_listener_with_user_data(()) + .process(|stream, _| { + let Some(mut buffer) = stream.dequeue_buffer() else { + return; + }; + let Some(data) = buffer.datas_mut().first_mut() else { + return; + }; + let size = data + .data() + .map(|bytes| { + bytes.fill(0); + bytes.len() + }) + .unwrap_or(0); + let chunk = data.chunk_mut(); + *chunk.offset_mut() = 0; + *chunk.stride_mut() = 4; + *chunk.size_mut() = size as u32; + }) + .register()?; + let mut audio = AudioInfoRaw::new(); + audio.set_format(AudioFormat::S16LE); + audio.set_rate(48_000); + audio.set_channels(2); + let mut position = [0; MAX_CHANNELS]; + position[0] = libspa::sys::SPA_AUDIO_CHANNEL_FL; + position[1] = libspa::sys::SPA_AUDIO_CHANNEL_FR; + audio.set_position(position); + let bytes = PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &Value::Object(Object { + type_: SpaTypes::ObjectParamFormat.as_raw(), + id: libspa::param::ParamType::EnumFormat.as_raw(), + properties: audio.into(), + }), + ) + .unwrap() + .0 + .into_inner(); + stream.connect( + Direction::Output, + None, + StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS, + &mut [Pod::from_bytes(&bytes).unwrap()], + )?; + mainloop.run(); + Ok(()) +} + +/// Overlapping app identities must preserve whichever assignment remains. +fn assignment_regressions( + engine: &pipedeck_engine::EngineHandle, + rx: &mpsc::Receiver, + dir: &std::path::Path, + failures: &mut u32, +) { + std::fs::create_dir_all(dir).unwrap(); + let fixture = dir.join("routing-regressions.toml"); + std::fs::write( + &fixture, + r#" +latency = "512/48000" +[[mix]] +id = 101 +name = "Routing test" +[[source]] +id = 101 +name = "Legacy assignment" +apps = ["Spotify"] +[[source]] +id = 102 +name = "Canonical assignment" +apps = ["com.spotify.Client"] +"#, + ) + .unwrap(); + let reload = || { + drain(rx); + engine.send(Command::ImportConfig(fixture.clone())).unwrap(); + wait_state(rx, "review fixture imported", |s| { + s.sources.iter().any(|source| source.id == SourceId(102)) + }); + settle(); + }; + reload(); + let mut player = Process::new(std::env::current_exe().unwrap()) + .args([ + "--identity-player", + "com.spotify.Client", + "Spotify", + "spotify", + "review-alias-player", + ]) + .spawn() + .unwrap(); + let apps = wait_for( + rx, + "canonical identity and legacy aliases", + |event| matches!(event, Event::Apps { running } if running.iter().any(|app| app.key == "com.spotify.Client" && app.matches("Spotify"))), + ); + check( + matches!(apps, Event::Apps { running } if running.iter().any(|app| app.matches("Spotify") && app.matches("spotify"))), + "the UI receives legacy aliases with the canonical app key", + failures, + ); + let routed = |row: u32| { + let dump = pw_dump(); + any_node_id(&dump, "review-alias-player") + .zip(node_id(&dump, &format!("pipedeck-smoke.src.{row}"))) + .is_some_and(|pair| links(&dump).contains(&pair)) + }; + settle(); + check( + routed(102), + "the canonical assignment takes precedence", + failures, + ); + engine + .send(Command::ReleaseApp { + id: SourceId(101), + app: "Spotify".into(), + }) + .unwrap(); + settle(); + check( + routed(102), + "releasing a legacy alias preserves the canonical route", + failures, + ); + reload(); + engine + .send(Command::ReleaseApp { + id: SourceId(102), + app: "com.spotify.Client".into(), + }) + .unwrap(); + settle(); + check( + routed(101), + "releasing the canonical id falls back to the remaining legacy route", + failures, + ); + engine + .send(Command::ReleaseApp { + id: SourceId(101), + app: "Spotify".into(), + }) + .unwrap(); + settle(); + check( + !routed(101) && !routed(102), + "removing the last alias releases the stream", + failures, + ); + reload(); + engine.send(Command::RemoveSource(SourceId(101))).unwrap(); + wait_state(rx, "legacy row removed", |s| { + s.sources.iter().all(|source| source.id != SourceId(101)) + }); + settle(); + check( + routed(102), + "removing a legacy row preserves the canonical route", + failures, + ); + let _ = player.kill(); + let _ = player.wait(); +} + fn main() -> ExitCode { + let args: Vec<_> = std::env::args().collect(); + if args.get(1).is_some_and(|arg| arg == "--identity-player") { + identity_player(&args[2..]).expect("the silent identity player runs"); + return ExitCode::SUCCESS; + } std::env::set_var("PIPEDECK_NODE_PREFIX", PREFIX); std::env::set_var("PIPEDECK_VOICE_APP", VOICE_APP); // A control socket of its own, beside the user's mixer's. @@ -448,6 +654,21 @@ fn main() -> ExitCode { // A fresh config starts with one mix and no source. let state = wait_state(&rx, "initial state", |_| true); + if args + .get(1) + .is_some_and(|arg| arg == "--routing-regressions") + { + assignment_regressions(&engine, &rx, &dir, &mut failures); + engine.shutdown(); + wait_for(&rx, "Stopped", |event| matches!(event, Event::Stopped)); + let _ = std::fs::remove_dir_all(&dir); + let _ = std::fs::remove_dir_all(&control_dir); + return if failures == 0 { + ExitCode::SUCCESS + } else { + ExitCode::FAILURE + }; + } let mix: MixId = state.mixes.first().expect("one mix by default").id; // The first Devices event is empty: the registry has not replied yet, and // the real list follows on the next housekeeping tick. @@ -899,6 +1120,90 @@ fn main() -> ExitCode { let _ = player.kill(); let _ = player.wait(); + // Installed Flatpak entries are assigned by id, while registry nodes + // may only give a name or binary. The id is in the full client info. + // Exercise both an assignment made before playback and one made after. + for (suffix, name, binary, before) in [ + ("Spotify", "Spotify", "spotify", true), + ("Browser", "Microsoft Edge", "msedge", false), + ] { + let id = format!("dev.pipedeck.Smoke{suffix}"); + if before { + engine + .send(Command::AssignApp { + id: source, + app: id.clone(), + }) + .unwrap(); + } + let node_name = format!("identity-check-player-{suffix}"); + let mut player = Process::new(std::env::current_exe().unwrap()) + .args(["--identity-player", &id, name, binary, &node_name]) + .spawn() + .expect("the silent identity player starts"); + wait_for( + &rx, + "a Flatpak application's identity", + |event| matches!(event, Event::Apps { running } if running.iter().any(|app| app.key == id)), + ); + if !before { + engine + .send(Command::AssignApp { + id: source, + app: id.clone(), + }) + .unwrap(); + } + settle(); + let dump = pw_dump(); + let joined = any_node_id(&dump, &node_name) + .zip(sink_node) + .is_some_and(|pair| links(&dump).contains(&pair)); + check( + joined, + &format!("{name} routes by its Flatpak client id (assigned before playback: {before})"), + &mut failures, + ); + if suffix == "Browser" { + // Sandboxed stream pids need not relate to the host window pid. + // OpenDeck's "In front" action must still identify the binary. + engine + .send(Command::SetFocus { + pid: u32::MAX, + class: "MsEdGe".into(), + }) + .unwrap(); + let focused = wait_for( + &rx, + "the focused Flatpak browser", + |event| matches!(event, Event::Focused { app: Some(app) } if app.key == id), + ); + check( + matches!(focused, Event::Focused { app: Some(app) } if app.key == id && app.matches("msedge")), + "a window-class alias resolves the canonical Flatpak app for In front", + &mut failures, + ); + } + engine + .send(Command::ReleaseApp { + id: source, + app: id, + }) + .unwrap(); + settle(); + let dump = pw_dump(); + let joined = any_node_id(&dump, &node_name) + .zip(sink_node) + .is_some_and(|pair| links(&dump).contains(&pair)); + check( + !joined, + &format!("releasing {name}'s Flatpak id releases its stream"), + &mut failures, + ); + let _ = player.kill(); + let _ = player.wait(); + } + let cfg = std::fs::read_to_string(&config_path).unwrap_or_default(); check( cfg.contains("[[link]]") && cfg.contains("gain = 0.5") && cfg.contains(&device.name), @@ -2370,10 +2675,9 @@ fn main() -> ExitCode { s.sources.iter().all(|row| row.id != board) && s.mixes.is_empty() }); - // A dock unplugged takes its devices away, and the loopbacks reading or - // playing to them destroy themselves. The mixer takes them off without - // destroying them again, which crashed it, and makes them again once - // the dock is back. + // A dock unplugged takes its devices away. Whether PipeWire destroys + // their loopbacks or leaves them running, the mixer takes them off + // safely and makes them again once the dock is back. engine.send(Command::AddMix).unwrap(); let state = wait_state(&rx, "a mix for the dock", |s| !s.mixes.is_empty()); let dock_mix = state.mixes[0].id; @@ -2461,6 +2765,15 @@ fn main() -> ExitCode { &format!("unplugged, the dock's loopbacks went: {nodes:?}"), &mut failures, ); + // Longer than the repair delay: an absent dock must not cause a loop + // of destroying and recreating streams against a device still away. + std::thread::sleep(Duration::from_secs(3)); + let nodes = node_names(&pw_dump()); + check( + !nodes.contains(&out_node) && !nodes.contains(&cell_node), + "the dock's loopbacks stay absent while its devices are away", + &mut failures, + ); // Plugged in again: both come back by themselves. plug(dock_out, "Audio/Sink"); plug(dock_in, "Audio/Source"); @@ -2475,6 +2788,29 @@ fn main() -> ExitCode { &format!("plugged in again, the dock's output and cell came back: {nodes:?}"), &mut failures, ); + // A quick replug may announce the replacement under the same name + // before the next engine tick. The old streams must still be replaced. + let old_output = node_serial(&pw_dump(), &out_node); + let old_cell = node_serial(&pw_dump(), &cell_node); + unplug(dock_out); + unplug(dock_in); + plug(dock_out, "Audio/Sink"); + plug(dock_in, "Audio/Source"); + let deadline = Instant::now() + Duration::from_secs(10); + let replaced = || { + let dump = pw_dump(); + let output = node_serial(&dump, &out_node); + let cell = node_serial(&dump, &cell_node); + output.is_some() && cell.is_some() && output != old_output && cell != old_cell + }; + while !replaced() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(300)); + } + check( + replaced(), + "a same-name replug replaces both dock loopbacks", + &mut failures, + ); // Unplugged again, then what crashed the mixer: its output switched // off, its row removed, each over a loopback that went by itself. unplug(dock_out); @@ -2585,6 +2921,8 @@ fn main() -> ExitCode { &mut failures, ); + assignment_regressions(&engine, &rx, &dir, &mut failures); + engine.shutdown(); wait_for(&rx, "Stopped", |e| matches!(e, Event::Stopped)); settle(); diff --git a/crates/pipedeck-engine/src/pw/apps.rs b/crates/pipedeck-engine/src/pw/apps.rs new file mode 100644 index 0000000..89eb89a --- /dev/null +++ b/crates/pipedeck-engine/src/pw/apps.rs @@ -0,0 +1,154 @@ +//! Match desktop launchers to streams, whose client may hold their identity. + +use libspa::utils::dict::DictRef; + +use crate::types::{SourceConfig, SourceId}; + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct Identity { + pub key: String, + pub aliases: Vec, +} + +impl Identity { + pub fn read(node: &DictRef, client: Option<&DictRef>) -> Option { + let client_prop = |key| client.and_then(|props| props.get(key)); + // PulseAudio streams may share a server client. Its executable is + // then pipewire, not the application named by the stream. + let same_application = node.get("application.name").is_none_or(|name| { + client_prop("application.name").is_some_and(|client_name| client_name == name) + }); + let app_client_prop = |key| same_application.then(|| client_prop(key)).flatten(); + // The portal's id agrees with Flatpak desktop entries. It lives on + // the client, not necessarily on any of that client's audio nodes. + let candidates = [ + client_prop("pipewire.access.portal.app_id"), + node.get("pipewire.access.portal.app_id"), + node.get("application.process.binary"), + app_client_prop("application.process.binary"), + node.get("application.id"), + app_client_prop("application.id"), + node.get("application.name"), + app_client_prop("application.name"), + node.get("node.name"), + ]; + let mut aliases = Vec::new(); + for value in candidates.into_iter().flatten().filter(|s| !s.is_empty()) { + if !aliases.iter().any(|alias| alias == value) { + aliases.push(value.to_owned()); + } + } + Some(Self { + key: aliases.first()?.clone(), + aliases, + }) + } + + pub fn matches(&self, key: &str) -> bool { + self.aliases.iter().any(|alias| alias == key) + } + + pub fn source(&self, sources: &[SourceConfig]) -> Option { + self.source_except(sources, None) + } + + pub fn source_except( + &self, + sources: &[SourceConfig], + except: Option, + ) -> Option { + // An explicit assignment of the stable id beats an older assignment + // of the same application's display name or binary on another row. + self.aliases.iter().find_map(|key| { + sources + .iter() + .find(|source| Some(source.id) != except && source.apps.contains(key)) + .map(|source| source.id) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pipewire::properties::properties; + + #[test] + fn spotify_uses_its_flatpak_id_from_the_client() { + let node = properties! { "application.name" => "Spotify", "node.name" => "spotify" }; + let client = properties! { "pipewire.access.portal.app_id" => "com.spotify.Client" }; + let identity = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + assert_eq!(identity.key, "com.spotify.Client"); + assert!(identity.matches("Spotify")); + assert!(identity.matches("spotify")); + } + + #[test] + fn flatpak_browsers_keep_their_binary_as_a_legacy_alias() { + for (id, binary, name) in [ + ("com.microsoft.Edge", "msedge", "Microsoft Edge"), + ("org.mozilla.firefox", "firefox", "Firefox"), + ("com.brave.Browser", "brave", "Brave"), + ] { + let node = + properties! { "application.process.binary" => binary, "application.name" => name }; + let client = properties! { "pipewire.access.portal.app_id" => id }; + let identity = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + assert_eq!(identity.key, id); + assert!(identity.matches(binary)); + assert!(identity.matches(name)); + } + } + + #[test] + fn native_apps_use_the_binary_even_when_the_client_supplies_it() { + let node = properties! { "application.name" => "Firefox" }; + let client = properties! { "application.name" => "Firefox", "application.process.binary" => "firefox", "application.id" => "org.mozilla.firefox" }; + let identity = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + assert_eq!(identity.key, "firefox"); + assert!(identity.matches("org.mozilla.firefox")); + } + + #[test] + fn a_shared_audio_server_client_does_not_become_the_application() { + let node = properties! { "application.name" => "Firefox", "node.name" => "Firefox" }; + let client = properties! { "application.name" => "PipeWire", "application.process.binary" => "pipewire" }; + let identity = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + assert_eq!(identity.key, "Firefox"); + assert!(!identity.matches("pipewire")); + assert!(!identity.matches("PipeWire")); + } + + #[test] + fn late_client_identity_preserves_assignments_and_prefers_the_stable_id() { + let node = properties! { "application.name" => "Spotify" }; + let client = properties! { "pipewire.access.portal.app_id" => "com.spotify.Client" }; + let before = Identity::read(node.as_ref(), None).unwrap(); + let after = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + let mut config = crate::config::Config { + sources: vec![ + SourceConfig::virtual_sink(SourceId(1), "System"), + SourceConfig::virtual_sink(SourceId(2), "Music"), + ], + ..Default::default() + }; + config.sources[0].apps = vec!["Spotify".into()]; + let old_source = config.sources[0].id; + assert_eq!(before.source(&config.sources), Some(old_source)); + assert_eq!(after.source(&config.sources), Some(old_source)); + config.sources[1].apps = vec!["com.spotify.Client".into()]; + assert_eq!(after.source(&config.sources), Some(config.sources[1].id)); + } + + #[test] + fn empty_ids_fall_back_without_matching_unrelated_apps() { + let node = properties! { "application.name" => "", "node.name" => "player" }; + let client = properties! { "pipewire.access.portal.app_id" => "" }; + let identity = Identity::read(node.as_ref(), Some(client.as_ref())).unwrap(); + assert_eq!(identity.key, "player"); + assert!(!identity.matches("com.spotify.Client")); + assert!( + Identity::read(pipewire::properties::PropertiesBox::new().as_ref(), None).is_none() + ); + } +} diff --git a/crates/pipedeck-engine/src/pw/focus.rs b/crates/pipedeck-engine/src/pw/focus.rs index d07cf86..f317437 100644 --- a/crates/pipedeck-engine/src/pw/focus.rs +++ b/crates/pipedeck-engine/src/pw/focus.rs @@ -73,12 +73,22 @@ pub(super) fn focused<'a>( playing .into_iter() .find(|(app, _)| { - let (key, name) = (app.key.to_lowercase(), app.name.to_lowercase()); - let base = key.rsplit('/').next().unwrap_or(&key).to_owned(); - (!class.is_empty() && (class == key || class == name || class == base)) - || comm - .as_deref() - .is_some_and(|comm| comm == base || comm == name) + let name = app.name.to_lowercase(); + if (!class.is_empty() && class == name) || comm.as_deref() == Some(name.as_str()) { + return true; + } + // A portal id can replace the executable as the canonical key. + // Keep window classes and host commands matching its older keys. + std::iter::once(&app.key) + .chain(app.aliases.iter()) + .any(|key| { + let key = key.to_lowercase(); + let base = key.rsplit('/').next().unwrap_or(&key); + (!class.is_empty() && (class == key || class == base)) + || comm + .as_deref() + .is_some_and(|comm| comm == key || comm == base) + }) }) .map(|(app, _)| app.clone()) } @@ -117,6 +127,7 @@ mod tests { fn app(key: &str, name: &str) -> App { App { key: key.into(), + aliases: vec![], name: name.into(), icon: None, } @@ -158,6 +169,43 @@ mod tests { assert_eq!(focused(&focus, [(&app("spotify", "Spotify"), None)]), None); } + #[test] + fn flatpak_focus_falls_back_to_binary_aliases() { + for alias in ["msedge", "/app/bin/msedge"] { + let mut edge = app("com.microsoft.Edge", "Microsoft Edge"); + edge.aliases = vec![alias.into()]; + let focus = Focus { + pid: u32::MAX, + class: "MsEdGe".into(), + }; + assert_eq!(focused(&focus, [(&edge, None)]), Some(edge.clone())); + let unrelated = Focus { + pid: u32::MAX, + class: "org.kde.dolphin".into(), + }; + assert_eq!(focused(&unrelated, [(&edge, None)]), None); + } + } + + #[test] + fn process_command_aliases_work_without_overriding_pid_matches() { + let me = std::process::id(); + let mut edge = app("com.microsoft.Edge", "Microsoft Edge"); + edge.aliases = vec![command(me).unwrap().to_uppercase()]; + let focus = Focus { + pid: me, + class: "unrelated-window-class".into(), + }; + // No stream pid: only the host process command can identify the app. + assert_eq!(focused(&focus, [(&edge, None)]), Some(edge.clone())); + let spotify = app("spotify", "Spotify"); + // A known stream pid takes precedence over another app's alias. + assert_eq!( + focused(&focus, [(&edge, None), (&spotify, Some(me))]), + Some(spotify) + ); + } + #[test] fn a_process_name_with_brackets_is_read_past() { assert_eq!(parent_in("4242 (Web (Content)) S 17 4242 0"), Some(17)); diff --git a/crates/pipedeck-engine/src/pw/mod.rs b/crates/pipedeck-engine/src/pw/mod.rs index b51256f..7e255ec 100644 --- a/crates/pipedeck-engine/src/pw/mod.rs +++ b/crates/pipedeck-engine/src/pw/mod.rs @@ -14,6 +14,7 @@ //! Everything here belongs to this process' client connection: when the //! process dies, the server drops all of it. Nothing lingers. +mod apps; pub mod args; pub mod filter_chain; mod focus; @@ -34,6 +35,7 @@ use std::time::{Duration, Instant}; use libspa::param::ParamType; use libspa::pod::Pod; use libspa::utils::dict::DictRef; +use pipewire::client::{Client, ClientListener}; use pipewire::context::ContextRc; use pipewire::core::CoreRc; use pipewire::link::Link; @@ -54,6 +56,7 @@ use crate::types::{ }; use crate::vst3::Plugin; +use apps::Identity; use filter_chain::ChainSpec; use loopback::{LoopbackSpec, AUDIO_POSITION, CHANNELS}; use meter::Meter; @@ -254,6 +257,9 @@ struct DeviceEntry { /// A playback stream belonging to some application. struct AppStream { app: App, + identity: Identity, + /// Retained so a client identity arriving after the node can resolve it. + global: GlobalObject, /// The process playing it, as the application says. pid: Option, /// Sent by the application to one of our voice sinks: it is already @@ -270,16 +276,6 @@ struct AppStream { _watch: Option<(Node, NodeListener)>, } -/// The key an assignment matches on: the binary when the server knows it, -/// the application name otherwise, and the node name as a last resort. -fn app_key(props: &DictRef) -> Option { - props - .get("application.process.binary") - .or_else(|| props.get("application.name")) - .or_else(|| props.get("node.name")) - .map(str::to_owned) -} - /// Which of a row's hosted plug-ins the effect at `index` is: the chain /// counts only the effects the mixer hosts, the user sees all of them. fn hosted_index(effects: &[Effect], index: usize) -> usize { @@ -344,6 +340,10 @@ pub struct Graph { /// Proxies of nodes we are about to destroy. See [`Quarantine`]. retired: Quarantine, devices: HashMap, + /// Device removals are handled on the tick: some PipeWire versions + /// leave their loopbacks alive, and modules cannot be dropped in a + /// registry callback. + lost_devices: Vec, devices_dirty: bool, /// One measurement per row and per column. A cell needs none: it carries /// its channel's signal scaled by its own fader. @@ -395,6 +395,11 @@ pub struct Graph { /// Which process each client is, as the server says. In a Flatpak, /// that is how this process's own clients are found. client_pids: HashMap, + /// Full client properties include the Flatpak portal id, which the + /// registry omits. Listeners only queue changes for the next tick. + clients: HashMap, + client_props: HashMap, + client_updates: Rc>>, /// The client that made each of the mixer's nodes, by the node's id. node_clients: HashMap, /// This process as the server knows it: its own number outside a @@ -452,6 +457,7 @@ impl Graph { written: HashMap::new(), retired: Quarantine::new(), devices: HashMap::new(), + lost_devices: Vec::new(), devices_dirty: false, plugins: crate::vst3::installed(), stereotool: crate::stereotool::Status::Absent, @@ -470,6 +476,9 @@ impl Graph { voice_link_owner: HashMap::new(), own_clients: std::collections::HashSet::new(), client_pids: HashMap::new(), + clients: HashMap::new(), + client_props: HashMap::new(), + client_updates: Rc::new(RefCell::new(Vec::new())), node_clients: HashMap::new(), own_pid: flatpak_instance() .is_none() @@ -1218,7 +1227,7 @@ impl Graph { .map(|cfg| cfg.apps.clone()) .unwrap_or_default(); for key in &apps { - self.move_app(key, None); + self.reroute_app(key, Some(id)); } // The voices play into the row's sink, so they go before it. let voices: Vec<(SourceId, String)> = self @@ -2026,16 +2035,88 @@ impl Graph { } } - /// Keep the clients of this process, which made the nodes that are ours. + /// Keep ownership from the registry and bind the full application identity. fn remember_client(&mut self, global: &GlobalObject<&DictRef>) { // The server fills this in from the socket, so it cannot be claimed // by anyone else. - let Some(pid) = global.props.and_then(|p| p.get("pipewire.sec.pid")) else { + if let Some(props) = global.props { + if let Some(pid) = props.get("pipewire.sec.pid") { + self.client_pids.insert(global.id, pid.to_owned()); + if self.own_pid.as_deref() == Some(pid) { + self.own_clients.insert(global.id); + } + } + self.client_props + .insert(global.id, PropertiesBox::from_dict(props)); + } + match self.registry.bind::(global) { + Ok(client) => { + let updates = self.client_updates.clone(); + let id = global.id; + let listener = client + .add_listener_local() + .info(move |info| { + if let Some(props) = info.props() { + updates + .borrow_mut() + .push((id, PropertiesBox::from_dict(props))); + } + }) + .register(); + self.clients.insert(id, (client, listener)); + } + Err(e) => log::warn!( + "cannot read application identity for client {}: {e}", + global.id + ), + } + } + + /// Full client info is asynchronous: resolve and reroute nodes already + /// announced too, rather than waiting for an application to play again. + fn absorb_clients(&mut self) { + let updates: Vec<_> = self.client_updates.borrow_mut().drain(..).collect(); + if updates.is_empty() { return; - }; - self.client_pids.insert(global.id, pid.to_owned()); - if self.own_pid.as_deref() == Some(pid) { - self.own_clients.insert(global.id); + } + for (id, props) in updates { + if self.clients.contains_key(&id) { + self.client_props.insert(id, props); + } + } + let mut watches = Vec::new(); + let mut changed = false; + for (id, stream) in &mut self.streams { + let Some(props) = stream.global.props.as_ref() else { + continue; + }; + let client = props + .get("client.id") + .and_then(|id| id.parse::().ok()) + .and_then(|id| self.client_props.get(&id)); + let Some(identity) = Identity::read(props.as_ref(), client.map(AsRef::as_ref)) else { + continue; + }; + if stream.identity == identity { + continue; + } + stream.app.key = identity.key.clone(); + stream.app.aliases = identity.aliases.clone(); + stream.identity = identity; + if stream.identity.matches(voice_app()) && stream._watch.is_none() { + watches.push((*id, stream.global.to_owned())); + } + changed = true; + } + for (id, global) in watches { + let watch = self.watch_stream_target(&global); + if let Some(stream) = self.streams.get_mut(&id) { + stream._watch = watch; + } + } + if changed { + self.streams_dirty = true; + self.reassign_apps(); } } @@ -2145,11 +2226,16 @@ impl Graph { /// An application playing: sent to its channel when it has one. fn remember_stream(&mut self, global: &GlobalObject<&DictRef>, props: &DictRef) { - let Some(key) = app_key(props) else { + let client = props + .get("client.id") + .and_then(|id| id.parse::().ok()) + .and_then(|id| self.client_props.get(&id)); + let Some(identity) = Identity::read(props, client.map(AsRef::as_ref)) else { return; }; let app = App { - key, + key: identity.key.clone(), + aliases: identity.aliases.clone(), name: app_name(props), icon: props .get("application.icon-name") @@ -2160,12 +2246,12 @@ impl Graph { // theirs, and its own mix wherever the user assigned it. Which is // which is in the stream's own properties, which the registry leaves // out: the node is bound to read them before it is moved. - let watch = if app.key == voice_app() { + let watch = if identity.matches(voice_app()) { self.watch_stream_target(global) } else { // An application the user has assigned lands on its row's sink // as soon as it starts playing. - if let Some(source) = self.source_for_app(&app.key) { + if let Some(source) = identity.source(&self.config.sources) { self.move_stream(global.id, &app.name, Some(source)); } None @@ -2177,6 +2263,8 @@ impl Graph { global.id, AppStream { app, + identity, + global: global.to_owned(), pid, pinned: false, voice: None, @@ -2243,9 +2331,9 @@ impl Graph { for (id, stream) in &self.streams { if let Some(voice) = &stream.voice { self.pin_stream(*id, &stream.app.name, voice); - } else if stream.app.key == voice_app() && !stream.placed { + } else if stream.identity.matches(voice_app()) && !stream.placed { // Where it was sent is not known yet; it is placed then. - } else if let Some(source) = self.source_for_app(&stream.app.key) { + } else if let Some(source) = stream.identity.source(&self.config.sources) { self.move_stream(*id, &stream.app.name, Some(source)); } } @@ -2261,7 +2349,7 @@ impl Graph { }; let mut any = false; for (id, stream) in &self.streams { - if stream.voice.is_some() || self.source_for_app(&stream.app.key).is_some() { + if stream.voice.is_some() || stream.identity.source(&self.config.sources).is_some() { metadata.set_property(*id, "target.object", None, None); any = true; } @@ -2399,6 +2487,11 @@ impl Graph { pub fn on_global_remove(&mut self, global_id: u32) { self.own_clients.remove(&global_id); self.client_pids.remove(&global_id); + self.clients.remove(&global_id); + self.client_props.remove(&global_id); + self.client_updates + .borrow_mut() + .retain(|(id, _)| *id != global_id); self.node_clients.remove(&global_id); self.waiting_stages .retain(|(_, global)| global.id != global_id); @@ -2443,7 +2536,8 @@ impl Graph { } } self.ports.remove(&global_id); - if self.devices.remove(&global_id).is_some() { + if let Some(device) = self.devices.remove(&global_id) { + self.lost_devices.push(device); self.devices_dirty = true; } if self.streams.remove(&global_id).is_some() { @@ -2472,6 +2566,12 @@ impl Graph { .iter() .find(|source| source.apps.iter().any(|app| app == key)) .map(|source| source.id) + .or_else(|| { + self.streams + .values() + .find(|stream| stream.identity.matches(key)) + .and_then(|stream| stream.identity.source(&self.config.sources)) + }) } /// Point a stream at a row's sink, the way a session manager does it. @@ -2502,19 +2602,39 @@ impl Graph { /// Move every running stream of an application, or release them. fn move_app(&self, key: &str, source: Option) { for (id, stream) in self.streams.iter().filter(|(_, s)| !s.pinned) { - if stream.app.key == key { + if stream.identity.matches(key) { self.move_stream(*id, &stream.app.name, source); } } } + /// Removing one spelling or row need not remove another valid assignment. + fn reroute_app(&self, key: &str, except: Option) { + for (id, stream) in self + .streams + .iter() + .filter(|(_, stream)| !stream.pinned && stream.identity.matches(key)) + { + let source = stream.identity.source_except(&self.config.sources, except); + self.move_stream(*id, &stream.app.name, source); + } + } + /// Assign an application to a row, taking it from whichever row had it. pub fn assign_app(&mut self, id: SourceId, key: String) -> Result<(), EngineError> { if self.config.source(id).is_none() { return Err(EngineError::UnknownSource(id)); } + let identities: Vec<_> = self + .streams + .values() + .filter(|stream| stream.identity.matches(&key)) + .map(|stream| &stream.identity) + .collect(); for source in &mut self.config.sources { - source.apps.retain(|app| app != &key); + source.apps.retain(|app| { + app != &key && !identities.iter().any(|identity| identity.matches(app)) + }); } let source = self .config @@ -2534,7 +2654,7 @@ impl Graph { .ok_or(EngineError::UnknownSource(id))?; source.apps.retain(|app| app != key); self.dirty = true; - self.move_app(key, None); + self.reroute_app(key, None); self.sync_voices(); Ok(()) } @@ -2543,7 +2663,13 @@ impl Graph { fn app_list(&self) -> Vec { let mut apps: Vec = Vec::new(); for stream in self.streams.values() { - if !apps.iter().any(|app| app.key == stream.app.key) { + if let Some(app) = apps.iter_mut().find(|app| app.key == stream.app.key) { + for alias in &stream.app.aliases { + if !app.aliases.contains(alias) { + app.aliases.push(alias.clone()); + } + } + } else { apps.push(stream.app.clone()); } } @@ -2674,6 +2800,7 @@ impl Graph { /// Periodic housekeeping from the engine timer: debounced config saves /// and device list updates. pub fn tick(&mut self) { + self.absorb_clients(); self.mend_stages(); self.tend_sounds(); self.hook_up_links(); @@ -2703,15 +2830,24 @@ impl Graph { impl Graph { // --- loopbacks that went ------------------------------------------------ - /// Take off the cells and outputs whose loopback destroyed itself, as a - /// loopback does when its device goes, and make them again once the - /// device is back: a dock unplugged and plugged in again, a card that - /// went to sleep. + /// Take off the cells and outputs whose device went, whether or not + /// PipeWire destroyed their loopbacks, and make them again once the + /// device is back. Queued registry removals also cover a device replaced + /// by one with the same name between two ticks. fn mend_stages(&mut self) { + let lost = std::mem::take(&mut self.lost_devices); let fallen: Vec<(SourceId, MixId)> = self .links .iter() - .filter(|(_, stage)| stage._module.is_gone()) + .filter(|(cell, stage)| { + stage._module.is_gone() + || self.config.source(cell.0).is_some_and(|source| { + source.device.as_ref().is_some_and(|name| { + lost.iter() + .any(|device| !device.is_output && device.device.name == *name) + }) + }) + }) .map(|(cell, _)| *cell) .collect(); for cell in fallen { @@ -2723,6 +2859,8 @@ impl Graph { self.drop_link(cell); self.fallen_cells.insert(cell, Instant::now()); } + let config = &self.config; + let lost = &lost; let fallen: Vec<(MixId, usize)> = self .mixes .iter() @@ -2730,7 +2868,19 @@ impl Graph { mix.outputs .iter() .enumerate() - .filter(|(_, stage)| stage.as_ref().is_some_and(|s| s._module.is_gone())) + .filter(move |(index, stage)| { + stage.as_ref().is_some_and(|stage| { + stage._module.is_gone() + || config + .mix(*id) + .and_then(|mix| mix.outputs.get(*index)) + .is_some_and(|output| { + lost.iter().any(|device| { + device.is_output && device.device.name == output.device + }) + }) + }) + }) .map(move |(index, _)| (*id, index)) }) .collect(); @@ -2821,6 +2971,13 @@ impl Graph { let Some(cfg) = self.config.source(id) else { return false; }; + if cfg + .device + .as_ref() + .is_some_and(|device| !self.device_present(device, false)) + { + return false; + } let (target, from_sink) = channel_output(cfg); if from_sink && (target.starts_with(NODE_PREFIX) || target.starts_with(node_prefix())) { self.sink_ids.contains_key(&target) diff --git a/crates/pipedeck-engine/src/pw/voices.rs b/crates/pipedeck-engine/src/pw/voices.rs index 17c8007..09c239b 100644 --- a/crates/pipedeck-engine/src/pw/voices.rs +++ b/crates/pipedeck-engine/src/pw/voices.rs @@ -111,9 +111,9 @@ fn meet(voices: &mut Vec, member: &CallMember, today: u64) -> bool impl Graph { /// Bind a stream to read where it was sent, and queue that for the tick. - pub(super) fn watch_stream_target( + pub(super) fn watch_stream_target>( &self, - global: &GlobalObject<&DictRef>, + global: &GlobalObject

, ) -> Option<(Node, NodeListener)> { let node = match self.registry.bind::(global) { Ok(node) => node, diff --git a/crates/pipedeck-engine/src/types.rs b/crates/pipedeck-engine/src/types.rs index 465dd77..cacea3c 100644 --- a/crates/pipedeck-engine/src/types.rs +++ b/crates/pipedeck-engine/src/types.rs @@ -661,8 +661,11 @@ pub struct Effect { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct App { /// What an assignment matches on, stable across runs of the application: - /// its binary when the server knows it, its name otherwise. + /// its Flatpak id when available, otherwise its binary or name. pub key: String, + /// Older keys the same application may have been assigned under. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub aliases: Vec, /// What to show. pub name: String, /// Icon name the application advertises, if it does. @@ -670,6 +673,12 @@ pub struct App { pub icon: Option, } +impl App { + pub fn matches(&self, key: &str) -> bool { + self.key == key || self.aliases.iter().any(|alias| alias == key) + } +} + /// An audio device the user can attach to a mix or turn into a source. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Device { @@ -683,6 +692,22 @@ pub struct Device { mod tests { use super::*; + #[test] + fn app_aliases_match_legacy_assignments_and_old_events_still_decode() { + let mut app: App = + serde_json::from_str(r#"{"key":"com.spotify.Client","name":"Spotify"}"#).unwrap(); + assert!(app.matches("com.spotify.Client")); + assert!(!app.matches("Spotify")); + app.aliases = vec!["Spotify".into(), "spotify".into()]; + assert!(app.matches("Spotify")); + assert!(app.matches("spotify")); + assert!(!app.matches("Firefox")); + assert_eq!( + serde_json::from_str::(&serde_json::to_string(&app).unwrap()).unwrap(), + app + ); + } + #[test] fn linear_volume_is_cubic_and_clamped() { let half = ChainState { diff --git a/crates/pipedeck/src/channel_dialog.rs b/crates/pipedeck/src/channel_dialog.rs index 98702ca..03b81b4 100644 --- a/crates/pipedeck/src/channel_dialog.rs +++ b/crates/pipedeck/src/channel_dialog.rs @@ -54,7 +54,7 @@ pub struct ChannelDialog { gone: Rc>, /// The applications drawn, and the ones playing the picker was made /// with: drawn again only when these change, not under the pointer. - drawn_apps: RefCell, Vec)>>, + drawn_apps: RefCell, Vec)>>, /// The people of past calls on the row carrying them, each with a way /// to forget them, and who was drawn there. past: gtk::Box, @@ -434,7 +434,7 @@ impl ChannelDialog { // Drawn again only when what is assigned or what plays changed: an // application starting anywhere says so, and the list and its // picker made again each time would close the picker as it is used. - let playing: Vec = running.iter().map(|app| app.key.clone()).collect(); + let playing = running.to_vec(); let now = (source.apps.clone(), playing); if self.drawn_apps.borrow().as_ref() == Some(&now) { self.syncing.set(false); @@ -457,7 +457,7 @@ impl ChannelDialog { for key in &source.apps { // An assigned application that is not playing right now is still // listed, since the assignment is what outlives the stream. - let row = self.app_row(key, running.iter().find(|app| &app.key == key)); + let row = self.app_row(key, running.iter().find(|app| app.matches(key))); self.apps.append(&row); } @@ -592,7 +592,14 @@ impl ChannelDialog { // rebuilds the list. let mut rows: Vec<(String, gtk::Widget)> = Vec::new(); - let held = |key: &str| source.apps.iter().any(|app| app == key); + let held = |key: &str| { + source.apps.iter().any(|assigned| { + assigned == key + || running + .iter() + .any(|app| app.matches(key) && app.matches(assigned)) + }) + }; let playing: Vec<&App> = running.iter().filter(|app| !held(&app.key)).collect(); if !playing.is_empty() { rows.push((String::new(), section("Playing now"))); @@ -607,7 +614,7 @@ impl ChannelDialog { let installed: Vec<&DesktopApp> = self .installed .iter() - .filter(|app| !held(&app.key) && !running.iter().any(|r| r.key == app.key)) + .filter(|app| !held(&app.key) && !running.iter().any(|r| r.matches(&app.key))) .collect(); if !installed.is_empty() { rows.push((String::new(), section("Installed"))); @@ -699,3 +706,67 @@ fn section(label: &str) -> gtk::Widget { heading.set_margin_start(6); heading.upcast() } + +#[cfg(test)] +mod tests { + use super::*; + #[test] + #[ignore = "requires a GTK display; run on the Broadway backend"] + fn legacy_assignments_follow_late_identity() { + adw::init().expect("GTK display available"); + let app = adw::Application::builder() + .application_id("dev._2c2t.PipedeckAliasTest") + .flags(gtk::gio::ApplicationFlags::NON_UNIQUE) + .build(); + app.register(None::<>k::gio::Cancellable>).unwrap(); + let parent = adw::ApplicationWindow::builder().application(&app).build(); + let path = + std::env::temp_dir().join(format!("pipedeck-alias-ui-{}.toml", std::process::id())); + let engine = EngineLink::new(pipedeck_engine::spawn(path, |_| {})); + fn labels(root: >k::Widget) -> Vec { + let mut text = Vec::new(); + if let Some(label) = root.downcast_ref::() { + text.push(label.text().to_string()); + } + let mut child = root.first_child(); + while let Some(widget) = child { + text.extend(labels(&widget)); + child = widget.next_sibling(); + } + text + } + let mut source = SourceConfig::virtual_sink(SourceId(99), "Legacy music"); + source.apps = vec!["Spotify".into()]; + let mut running = vec![App { + key: "com.spotify.Client".into(), + aliases: vec![], + name: "Spotify".into(), + icon: None, + }]; + let dialog = ChannelDialog::present( + &parent, + &engine, + &source, + &[], + &running, + &[], + &Status::Absent, + ); + assert!(labels(dialog.apps.upcast_ref()) + .iter() + .any(|text| text == "Not running")); + // The canonical key remains the same when late client aliases arrive. + running[0].aliases = vec!["Spotify".into(), "spotify".into()]; + dialog.refresh(&source, &[], &running, &[], &Status::Absent); + let text = labels(dialog.apps.upcast_ref()); + assert!(text.iter().any(|text| text == "Playing")); + assert!(!text.iter().any(|text| text == "Not running")); + let popover = dialog.app_popover(&source, &running); + assert!(!labels(popover.upcast_ref()) + .iter() + .any(|text| text == "Playing now" || text == "Spotify")); + dialog.close_gone(); + engine.shutdown(); + parent.close(); + } +} diff --git a/crates/pipedeck/src/desktop.rs b/crates/pipedeck/src/desktop.rs index fc22336..b38c774 100644 --- a/crates/pipedeck/src/desktop.rs +++ b/crates/pipedeck/src/desktop.rs @@ -12,8 +12,8 @@ use std::path::{Path, PathBuf}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct DesktopApp { /// The key an assignment matches on, which must line up with what - /// PipeWire reports for the streams this application opens: the basename - /// of its executable. + /// PipeWire reports for the streams this application opens: its Flatpak + /// id, or the basename of its executable for a native application. pub key: String, pub name: String, /// Icon name, or an absolute path when the entry gives one. @@ -42,8 +42,7 @@ fn entry_dirs() -> Vec { /// Turn an `Exec=` line into the key PipeWire will report. /// /// Field codes such as `%U` are dropped, as is a leading `env VAR=value`. -/// A Flatpak launcher runs the application from its own binary, so the -/// application id is the closer match. +/// Flatpak ids match the portal identity read from the PipeWire client. fn key_from_exec(exec: &str) -> Option { let mut words = exec .split_whitespace()