things
This commit is contained in:
@@ -16,4 +16,6 @@ chrono = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
reqwest = { version = "0.13.5", features = ["json"] }
|
||||
pipewire-native = "0.1.4"
|
||||
pipewire-native-spa = "0.1.4"
|
||||
hyprland = { git = "https://github.com/hyprland-community/hyprland-rs", branch = "master", default-features = false, features = ["listener", "tokio"] }
|
||||
|
||||
@@ -8,6 +8,7 @@ use common::{Inbox, Service, Wire};
|
||||
|
||||
pub mod compositors;
|
||||
pub mod datetime;
|
||||
pub mod pipewire;
|
||||
pub mod weather;
|
||||
|
||||
/// Conversion from a route key + its inbox to the subscription backing it.
|
||||
@@ -28,6 +29,7 @@ impl IntoSubscription for Service {
|
||||
compositors::hyprland::HyprlandService::run(inbox)
|
||||
}
|
||||
k if k.starts_with(weather::NAMESPACE) => weather::WeatherService::run(inbox),
|
||||
k if k.starts_with(pipewire::NAMESPACE) => pipewire::PipewireService::run(inbox),
|
||||
_ => {
|
||||
tracing::warn!(service = key, "no service impl for route key");
|
||||
Subscription::none()
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
//! PipeWire service: follows the default sink and source — WirePlumber names
|
||||
//! them in its `default` metadata — and republishes their volume/mute
|
||||
//! whenever PipeWire reports a change.
|
||||
//!
|
||||
//! PipeWire drives its own main loop, so the connection lives on a dedicated
|
||||
//! thread; that thread hands state to the async side over a channel.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use common::{AudioNode, Endpoint, Inbox, PipewireState, Wire};
|
||||
use iced::futures::{channel::mpsc, SinkExt};
|
||||
use iced::Subscription;
|
||||
use pipewire_native as pw;
|
||||
use pipewire_native::context::Context;
|
||||
use pipewire_native::core::Core;
|
||||
use pipewire_native::main_loop::MainLoop;
|
||||
use pipewire_native::properties::Properties;
|
||||
use pipewire_native::proxy::metadata::{Metadata, MetadataEvents};
|
||||
use pipewire_native::proxy::node::{Node, NodeEvents};
|
||||
use pipewire_native::proxy::registry::{Registry, RegistryEvents};
|
||||
use pipewire_native::proxy::{HasProxy, ProxyEvents};
|
||||
use pipewire_native::some_closure;
|
||||
use pipewire_native::types;
|
||||
use pipewire_native_spa as spa;
|
||||
use tokio::sync::mpsc as tokio_mpsc;
|
||||
|
||||
/// Route-key namespace owned by this service; keys are `"pipewire.<kind>"`.
|
||||
pub const NAMESPACE: &str = "pipewire.";
|
||||
|
||||
/// Node classes the service follows.
|
||||
const SINK: &str = "Audio/Sink";
|
||||
const SOURCE: &str = "Audio/Source";
|
||||
|
||||
/// WirePlumber's metadata object holding the default nodes.
|
||||
const DEFAULTS: &str = "default";
|
||||
const DEFAULT_SINK: &str = "default.audio.sink";
|
||||
const DEFAULT_SOURCE: &str = "default.audio.source";
|
||||
|
||||
pub struct PipewireService;
|
||||
|
||||
impl PipewireService {
|
||||
pub fn run(inbox: Inbox) -> Subscription<Wire> {
|
||||
Subscription::run_with(inbox, |inbox| {
|
||||
let key = inbox.key();
|
||||
// We dont need to rx any msgs from modules
|
||||
let mut _rx = inbox.take().expect("inbox receiver is taken once");
|
||||
|
||||
iced::stream::channel(0, move |mut sender: mpsc::Sender<Wire>| async move {
|
||||
// The PipeWire thread owns the connection and drops it once
|
||||
// this channel goes away, which stops its loop.
|
||||
let (tx, mut rx) = tokio_mpsc::unbounded_channel();
|
||||
std::thread::spawn(move || connect(tx));
|
||||
while let Some(state) = rx.recv().await {
|
||||
tracing::debug!(?state, "pipewire state changed");
|
||||
let wire = Wire::topic(Endpoint::service(key), key, state);
|
||||
if sender.send(wire).await.is_err() {
|
||||
tracing::warn!(key, "pipewire send failed; stopping service");
|
||||
return;
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Connects, follows the default nodes, and runs PipeWire's loop until the
|
||||
/// subscriber (or PipeWire itself) goes away.
|
||||
fn connect(tx: tokio_mpsc::UnboundedSender<PipewireState>) {
|
||||
pw::init();
|
||||
let conn = match Connection::open() {
|
||||
Ok(conn) => conn,
|
||||
Err(err) => {
|
||||
tracing::warn!(%err, "pipewire unavailable; audio state stays unknown");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let track = Arc::new(Mutex::new(Track::new(conn, tx)));
|
||||
watch(&track);
|
||||
let main_loop = track.lock().unwrap().conn.main_loop.clone();
|
||||
|
||||
tracing::info!("pipewire connected");
|
||||
main_loop.run();
|
||||
tracing::info!("pipewire disconnected");
|
||||
}
|
||||
|
||||
/// The connection chain: PipeWire tears the session down if any link of it
|
||||
/// goes away, so all of it lives for as long as the loop runs.
|
||||
struct Connection {
|
||||
main_loop: MainLoop,
|
||||
_context: Context,
|
||||
_core: Core,
|
||||
registry: Registry,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
fn open() -> std::io::Result<Self> {
|
||||
let main_loop = MainLoop::new(&Properties::new())
|
||||
.ok_or_else(|| std::io::Error::other("pipewire main loop unavailable"))?;
|
||||
let context = Context::new(&main_loop, Properties::new())?;
|
||||
let core = context.connect(None)?;
|
||||
let registry = core.registry()?;
|
||||
Ok(Self {
|
||||
main_loop,
|
||||
_context: context,
|
||||
_core: core,
|
||||
registry,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribes to every sink/source node and to the `default` metadata, so
|
||||
/// PipeWire pushes volume/mute changes instead of us polling for them.
|
||||
fn watch(track: &Arc<Mutex<Track>>) {
|
||||
let registry = track.lock().unwrap().conn.registry.clone();
|
||||
registry.add_listener(RegistryEvents {
|
||||
global: some_closure!([registry ^(track)] id, _perms, type_, version, props, {
|
||||
match type_ {
|
||||
types::interface::NODE => {
|
||||
let class = props.get("media.class").unwrap_or_default();
|
||||
if class != SINK && class != SOURCE {
|
||||
return;
|
||||
}
|
||||
if let Some(name) = props.get("node.name") {
|
||||
bind_node(track, ®istry, id, type_, version, name.to_string());
|
||||
}
|
||||
}
|
||||
types::interface::METADATA if props.get("metadata.name") == Some(DEFAULTS) => {
|
||||
bind_defaults(track, ®istry, id, type_, version);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}),
|
||||
global_remove: some_closure!([^(track)] id, {
|
||||
let mut track = track.lock().unwrap();
|
||||
if let Some(name) = track.ids.remove(&id) {
|
||||
track.nodes.remove(&name);
|
||||
track.publish();
|
||||
}
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
/// Binds a sink/source node and follows its `Props` param (volume + mute).
|
||||
fn bind_node(
|
||||
track: &Arc<Mutex<Track>>,
|
||||
registry: &Registry,
|
||||
id: pw::Id,
|
||||
type_: &str,
|
||||
version: u32,
|
||||
name: String,
|
||||
) {
|
||||
let Ok(object) = registry.bind(id, type_, version) else {
|
||||
return;
|
||||
};
|
||||
let Some(proxy) = object.downcast_proxy::<Node>() else {
|
||||
return;
|
||||
};
|
||||
let Some(node) = proxy.object() else {
|
||||
return;
|
||||
};
|
||||
|
||||
node.add_listener(NodeEvents {
|
||||
param: some_closure!([^(track, name)] _seq, id, _index, _next, pod, {
|
||||
if id != spa::param::ParamType::Props {
|
||||
return;
|
||||
}
|
||||
let mut track = track.lock().unwrap();
|
||||
merge(track.nodes.entry(name.clone()).or_default(), pod);
|
||||
track.publish();
|
||||
}),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
// A local proxy is only usable once the server has bound it, and asking
|
||||
// for params is a method call.
|
||||
proxy.add_listener(ProxyEvents {
|
||||
bound_props: some_closure!([^(node)] _id, _props, {
|
||||
let _ = node.subscribe_params(&[spa::param::ParamType::Props]);
|
||||
let _ = node.enum_params(0, Some(spa::param::ParamType::Props), 0, u32::MAX, None);
|
||||
}),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
tracing::debug!(name, "following pipewire node");
|
||||
let mut track = track.lock().unwrap();
|
||||
track.ids.insert(id, name);
|
||||
track.keep.push(object);
|
||||
}
|
||||
|
||||
/// Binds WirePlumber's `default` metadata and follows which nodes are the
|
||||
/// current sink and source.
|
||||
fn bind_defaults(
|
||||
track: &Arc<Mutex<Track>>,
|
||||
registry: &Registry,
|
||||
id: pw::Id,
|
||||
type_: &str,
|
||||
version: u32,
|
||||
) {
|
||||
let Ok(object) = registry.bind(id, type_, version) else {
|
||||
return;
|
||||
};
|
||||
let Some(proxy) = object.downcast_proxy::<Metadata>() else {
|
||||
return;
|
||||
};
|
||||
let Some(metadata) = proxy.object() else {
|
||||
return;
|
||||
};
|
||||
|
||||
metadata.add_listener(MetadataEvents {
|
||||
property: some_closure!([^(track)] _subject, key, _type, value, {
|
||||
let (Some(key), Some(value)) = (key, value) else {
|
||||
return;
|
||||
};
|
||||
let Some(name) = default_node(value) else {
|
||||
return;
|
||||
};
|
||||
let mut track = track.lock().unwrap();
|
||||
match key {
|
||||
DEFAULT_SINK => track.sink = Some(name),
|
||||
DEFAULT_SOURCE => track.source = Some(name),
|
||||
_ => return,
|
||||
}
|
||||
track.publish();
|
||||
}),
|
||||
});
|
||||
|
||||
tracing::debug!("following pipewire default metadata");
|
||||
track.lock().unwrap().keep.push(object);
|
||||
}
|
||||
|
||||
/// Merges the volume/mute carried by a `SPA_PARAM_Props` pod into `state`;
|
||||
/// anything the pod leaves out keeps its previous value.
|
||||
fn merge(state: &mut AudioNode, pod: &spa::pod::RawPodOwned) {
|
||||
let mut parser = spa::pod::parser::Parser::new(pod.data());
|
||||
let _ = parser.pop_object::<spa::param::props::Prop, spa::param::ParamType, _>(|props, _id| {
|
||||
for (key, _flags, value) in props {
|
||||
match key {
|
||||
spa::param::props::Prop::Volume => {
|
||||
if let Ok(volume) = value.decode::<f32>() {
|
||||
state.volume = volume;
|
||||
}
|
||||
}
|
||||
spa::param::props::Prop::Mute => {
|
||||
if let Ok(muted) = value.decode::<bool>() {
|
||||
state.muted = muted;
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
|
||||
/// WirePlumber stores a default as `{"name":"<node.name>"}`.
|
||||
fn default_node(value: &str) -> Option<String> {
|
||||
#[derive(serde::Deserialize)]
|
||||
struct Default {
|
||||
name: String,
|
||||
}
|
||||
serde_json::from_str::<Default>(value).ok().map(|d| d.name)
|
||||
}
|
||||
|
||||
/// What the PipeWire callbacks share: the connection, the audio state they
|
||||
/// observe, and the channel home.
|
||||
struct Track {
|
||||
conn: Connection,
|
||||
/// Volume/mute of every sink and source, keyed by `node.name`.
|
||||
nodes: HashMap<String, AudioNode>,
|
||||
/// Registry id -> `node.name`, so a removed node can be forgotten.
|
||||
ids: HashMap<pw::Id, String>,
|
||||
/// `node.name` of the default sink/source, per the metadata.
|
||||
sink: Option<String>,
|
||||
source: Option<String>,
|
||||
/// Bound proxies, kept alive so their listeners stay registered.
|
||||
keep: Vec<Box<dyn HasProxy>>,
|
||||
tx: tokio_mpsc::UnboundedSender<PipewireState>,
|
||||
/// Last state published, so redundant updates are dropped.
|
||||
sent: Option<PipewireState>,
|
||||
}
|
||||
|
||||
impl Track {
|
||||
fn new(conn: Connection, tx: tokio_mpsc::UnboundedSender<PipewireState>) -> Self {
|
||||
Self {
|
||||
conn,
|
||||
nodes: HashMap::new(),
|
||||
ids: HashMap::new(),
|
||||
sink: None,
|
||||
source: None,
|
||||
keep: Vec::new(),
|
||||
tx,
|
||||
sent: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes the default sink/source, if the state actually changed.
|
||||
fn publish(&mut self) {
|
||||
let state = PipewireState {
|
||||
sink: self.node(&self.sink),
|
||||
source: self.node(&self.source),
|
||||
};
|
||||
if self.sent == Some(state) {
|
||||
return;
|
||||
}
|
||||
self.sent = Some(state);
|
||||
if self.tx.send(state).is_err() {
|
||||
tracing::info!("pipewire subscriber gone; disconnecting");
|
||||
self.conn.main_loop.quit();
|
||||
}
|
||||
}
|
||||
|
||||
/// State of the named node; zeroes while the default is unknown.
|
||||
fn node(&self, name: &Option<String>) -> AudioNode {
|
||||
name.as_ref()
|
||||
.and_then(|name| self.nodes.get(name))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user