feat: add new volume module

src/clients/mod.rs

@@ -10,6 +10,8 @@ pub mod music;
 pub mod system_tray;
 #[cfg(feature = "upower")]
 pub mod upower;
+#[cfg(feature = "volume")]
+pub mod volume;
 pub mod wayland;
 
 /// Singleton wrapper consisting of
@@ -27,6 +29,8 @@ pub struct Clients {
     tray: Option<Arc<system_tray::TrayEventReceiver>>,
     #[cfg(feature = "upower")]
     upower: Option<Arc<zbus::fdo::PropertiesProxy<'static>>>,
+    #[cfg(feature = "volume")]
+    volume: Option<Arc<volume::Client>>,
 }
 
 impl Clients {
@@ -86,6 +90,13 @@ impl Clients {
             })
             .clone()
     }
+
+    #[cfg(feature = "volume")]
+    pub fn volume(&mut self) -> Arc<volume::Client> {
+        self.volume
+            .get_or_insert_with(volume::create_client)
+            .clone()
+    }
 }
 
 /// Types implementing this trait

src/clients/volume/mod.rs

@@ -0,0 +1,312 @@
+mod sink;
+mod sink_input;
+
+use crate::{arc_mut, lock, register_client, send, spawn_blocking};
+use libpulse_binding::callbacks::ListResult;
+use libpulse_binding::context::introspect::{Introspector, ServerInfo};
+use libpulse_binding::context::subscribe::{Facility, InterestMaskSet, Operation};
+use libpulse_binding::context::{Context, FlagSet, State};
+use libpulse_binding::mainloop::standard::{IterateResult, Mainloop};
+use libpulse_binding::proplist::Proplist;
+use libpulse_binding::volume::{ChannelVolumes, Volume};
+use std::fmt::{Debug, Formatter};
+use std::sync::{Arc, Mutex};
+use tokio::sync::broadcast;
+use tracing::{debug, error, info, warn};
+
+pub use sink::Sink;
+pub use sink_input::SinkInput;
+
+type ArcMutVec<T> = Arc<Mutex<Vec<T>>>;
+
+#[derive(Debug, Clone)]
+pub enum Event {
+    AddSink(Sink),
+    UpdateSink(Sink),
+    RemoveSink(String),
+
+    AddInput(SinkInput),
+    UpdateInput(SinkInput),
+    RemoveInput(u32),
+}
+
+#[derive(Debug)]
+pub struct Client {
+    connection: Arc<Mutex<ConnectionState>>,
+
+    data: Data,
+
+    tx: broadcast::Sender<Event>,
+    _rx: broadcast::Receiver<Event>,
+}
+
+#[derive(Debug, Default, Clone)]
+struct Data {
+    sinks: ArcMutVec<Sink>,
+    sink_inputs: ArcMutVec<SinkInput>,
+
+    default_sink_name: Arc<Mutex<Option<String>>>,
+}
+
+pub enum ConnectionState {
+    Disconnected,
+    Connected {
+        context: Arc<Mutex<Context>>,
+        introspector: Introspector,
+    },
+}
+
+impl Debug for ConnectionState {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::Disconnected => "Disconnected",
+                Self::Connected { .. } => "Connected",
+            }
+        )
+    }
+}
+
+impl Client {
+    pub fn new() -> Self {
+        let (tx, rx) = broadcast::channel(32);
+
+        Self {
+            connection: arc_mut!(ConnectionState::Disconnected),
+            data: Data::default(),
+            tx,
+            _rx: rx,
+        }
+    }
+
+    /// Starts the client.
+    fn run(&self) {
+        let Some(mut proplist) = Proplist::new() else {
+            error!("Failed to create PA proplist");
+            return;
+        };
+
+        if proplist
+            .set_str("APPLICATION_NAME", "dev.jstanger.ironbar")
+            .is_err()
+        {
+            error!("Failed to update PA proplist");
+        }
+
+        let Some(mut mainloop) = Mainloop::new() else {
+            error!("Failed to create PA mainloop");
+            return;
+        };
+
+        let Some(context) = Context::new_with_proplist(&mainloop, "Ironbar Context", &proplist)
+        else {
+            error!("Failed to create PA context");
+            return;
+        };
+
+        let context = arc_mut!(context);
+
+        let state_callback = Box::new({
+            let context = context.clone();
+            let data = self.data.clone();
+            let tx = self.tx.clone();
+
+            move || on_state_change(&context, &data, &tx)
+        });
+
+        lock!(context).set_state_callback(Some(state_callback));
+
+        if let Err(err) = lock!(context).connect(None, FlagSet::NOAUTOSPAWN, None) {
+            error!("{err:?}");
+        }
+
+        let introspector = lock!(context).introspect();
+
+        {
+            let mut inner = lock!(self.connection);
+            *inner = ConnectionState::Connected {
+                context,
+                introspector,
+            };
+        }
+
+        loop {
+            match mainloop.iterate(true) {
+                IterateResult::Success(_) => {}
+                IterateResult::Err(err) => error!("{err:?}"),
+                IterateResult::Quit(_) => break,
+            }
+        }
+    }
+
+    /// Gets an event receiver.
+    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
+        self.tx.subscribe()
+    }
+}
+
+/// Creates a new Pulse volume client.
+pub fn create_client() -> Arc<Client> {
+    let client = Arc::new(Client::new());
+
+    {
+        let client = client.clone();
+        spawn_blocking(move || {
+            client.run();
+        });
+    }
+
+    client
+}
+
+fn on_state_change(context: &Arc<Mutex<Context>>, data: &Data, tx: &broadcast::Sender<Event>) {
+    let Ok(state) = context.try_lock().map(|lock| lock.get_state()) else {
+        return;
+    };
+
+    match state {
+        State::Ready => {
+            info!("connected to server");
+
+            let introspect = lock!(context).introspect();
+            let introspect2 = lock!(context).introspect();
+
+            introspect.get_sink_info_list({
+                let sinks = data.sinks.clone();
+                let default_sink = data.default_sink_name.clone();
+
+                let tx = tx.clone();
+
+                move |info| match info {
+                    ListResult::Item(_) => sink::add(info, &sinks, &tx),
+                    ListResult::End => {
+                        introspect2.get_server_info({
+                            let sinks = sinks.clone();
+                            let default_sink = default_sink.clone();
+                            let tx = tx.clone();
+
+                            move |info| set_default_sink(info, &sinks, &default_sink, &tx)
+                        });
+                    }
+                    ListResult::Error => error!("Error while receiving sinks"),
+                }
+            });
+
+            introspect.get_sink_input_info_list({
+                let inputs = data.sink_inputs.clone();
+                let tx = tx.clone();
+
+                move |info| sink_input::add(info, &inputs, &tx)
+            });
+
+            let subscribe_callback = Box::new({
+                let context = context.clone();
+                let data = data.clone();
+                let tx = tx.clone();
+
+                move |facility, op, i| on_event(&context, &data, &tx, facility, op, i)
+            });
+
+            lock!(context).set_subscribe_callback(Some(subscribe_callback));
+            lock!(context).subscribe(
+                InterestMaskSet::SERVER | InterestMaskSet::SINK_INPUT | InterestMaskSet::SINK,
+                |_| (),
+            );
+        }
+        State::Failed => error!("Failed to connect to audio server"),
+        State::Terminated => error!("Connection to audio server terminated"),
+        _ => {}
+    }
+}
+
+fn on_event(
+    context: &Arc<Mutex<Context>>,
+    data: &Data,
+    tx: &broadcast::Sender<Event>,
+    facility: Option<Facility>,
+    op: Option<Operation>,
+    i: u32,
+) {
+    let (Some(facility), Some(op)) = (facility, op) else {
+        return;
+    };
+
+    match facility {
+        Facility::Server => on_server_event(context, &data.sinks, &data.default_sink_name, tx),
+        Facility::Sink => sink::on_event(context, &data.sinks, &data.default_sink_name, tx, op, i),
+        Facility::SinkInput => sink_input::on_event(context, &data.sink_inputs, tx, op, i),
+        _ => error!("Received unhandled facility: {facility:?}"),
+    }
+}
+
+fn on_server_event(
+    context: &Arc<Mutex<Context>>,
+    sinks: &ArcMutVec<Sink>,
+    default_sink: &Arc<Mutex<Option<String>>>,
+    tx: &broadcast::Sender<Event>,
+) {
+    lock!(context).introspect().get_server_info({
+        let sinks = sinks.clone();
+        let default_sink = default_sink.clone();
+        let tx = tx.clone();
+
+        move |info| set_default_sink(info, &sinks, &default_sink, &tx)
+    });
+}
+
+fn set_default_sink(
+    info: &ServerInfo,
+    sinks: &ArcMutVec<Sink>,
+    default_sink: &Arc<Mutex<Option<String>>>,
+    tx: &broadcast::Sender<Event>,
+) {
+    let default_sink_name = info.default_sink_name.as_ref().map(ToString::to_string);
+
+    if default_sink_name != *lock!(default_sink) {
+        if let Some(ref default_sink_name) = default_sink_name {
+            if let Some(sink) = lock!(sinks)
+                .iter_mut()
+                .find(|s| s.name.as_str() == default_sink_name.as_str())
+            {
+                sink.active = true;
+                debug!("Set sink active: {}", sink.name);
+                send!(tx, Event::UpdateSink(sink.clone()));
+            } else {
+                warn!("Couldn't find sink: {}", default_sink_name);
+            }
+        }
+    }
+
+    *lock!(default_sink) = default_sink_name;
+}
+
+/// Converts a Pulse `ChannelVolumes` struct into a single percentage value,
+/// representing the average value across all channels.
+fn volume_to_percent(volume: ChannelVolumes) -> f64 {
+    let avg = volume.avg().0;
+    let base_delta = (Volume::NORMAL.0 - Volume::MUTED.0) as f64 / 100.0;
+
+    ((avg - Volume::MUTED.0) as f64 / base_delta).round()
+}
+
+/// Converts a percentage volume into a Pulse volume value,
+/// which can be used for setting channel volumes.
+pub fn percent_to_volume(target_percent: f64) -> u32 {
+    let base_delta = (Volume::NORMAL.0 as f32 - Volume::MUTED.0 as f32) / 100.0;
+
+    if target_percent < 0.0 {
+        Volume::MUTED.0
+    } else if target_percent == 100.0 {
+        Volume::NORMAL.0
+    } else if target_percent >= 150.0 {
+        (Volume::NORMAL.0 as f32 * 1.5) as u32
+    } else if target_percent < 100.0 {
+        Volume::MUTED.0 + target_percent as u32 * base_delta as u32
+    } else {
+        Volume::NORMAL.0 + (target_percent - 100.0) as u32 * base_delta as u32
+    }
+}
+
+register_client!(Client, volume);

src/clients/volume/sink.rs

@@ -0,0 +1,175 @@
+use super::{percent_to_volume, volume_to_percent, ArcMutVec, Client, ConnectionState, Event};
+use crate::{lock, send};
+use libpulse_binding::callbacks::ListResult;
+use libpulse_binding::context::introspect::SinkInfo;
+use libpulse_binding::context::subscribe::Operation;
+use libpulse_binding::context::Context;
+use libpulse_binding::def::SinkState;
+use std::sync::{mpsc, Arc, Mutex};
+use tokio::sync::broadcast;
+use tracing::{debug, error};
+
+#[derive(Debug, Clone)]
+pub struct Sink {
+    index: u32,
+    pub name: String,
+    pub description: String,
+    pub volume: f64,
+    pub muted: bool,
+    pub active: bool,
+}
+
+impl From<&SinkInfo<'_>> for Sink {
+    fn from(value: &SinkInfo) -> Self {
+        Self {
+            index: value.index,
+            name: value
+                .name
+                .as_ref()
+                .map(ToString::to_string)
+                .unwrap_or_default(),
+            description: value
+                .description
+                .as_ref()
+                .map(ToString::to_string)
+                .unwrap_or_default(),
+            muted: value.mute,
+            volume: volume_to_percent(value.volume),
+            active: value.state == SinkState::Running,
+        }
+    }
+}
+
+impl Client {
+    pub fn sinks(&self) -> Arc<Mutex<Vec<Sink>>> {
+        self.data.sinks.clone()
+    }
+
+    pub fn set_default_sink(&self, name: &str) {
+        if let ConnectionState::Connected { context, .. } = &*lock!(self.connection) {
+            lock!(context).set_default_sink(name, |_| {});
+        }
+    }
+
+    pub fn set_sink_volume(&self, name: &str, volume_percent: f64) {
+        if let ConnectionState::Connected { introspector, .. } = &mut *lock!(self.connection) {
+            let (tx, rx) = mpsc::channel();
+
+            introspector.get_sink_info_by_name(name, move |info| {
+                let ListResult::Item(info) = info else {
+                    return;
+                };
+                send!(tx, info.volume);
+            });
+
+            let new_volume = percent_to_volume(volume_percent);
+
+            let mut volume = rx.recv().expect("to receive info");
+            for v in volume.get_mut() {
+                v.0 = new_volume;
+            }
+
+            introspector.set_sink_volume_by_name(name, &volume, None);
+        }
+    }
+
+    pub fn set_sink_muted(&self, name: &str, muted: bool) {
+        if let ConnectionState::Connected { introspector, .. } = &mut *lock!(self.connection) {
+            introspector.set_sink_mute_by_name(name, muted, None);
+        }
+    }
+}
+
+pub fn on_event(
+    context: &Arc<Mutex<Context>>,
+    sinks: &ArcMutVec<Sink>,
+    default_sink: &Arc<Mutex<Option<String>>>,
+    tx: &broadcast::Sender<Event>,
+    op: Operation,
+    i: u32,
+) {
+    let introspect = lock!(context).introspect();
+
+    match op {
+        Operation::New => {
+            debug!("new sink");
+            introspect.get_sink_info_by_index(i, {
+                let sinks = sinks.clone();
+                let tx = tx.clone();
+
+                move |info| add(info, &sinks, &tx)
+            });
+        }
+        Operation::Changed => {
+            debug!("sink changed");
+            introspect.get_sink_info_by_index(i, {
+                let sinks = sinks.clone();
+                let default_sink = default_sink.clone();
+                let tx = tx.clone();
+
+                move |info| update(info, &sinks, &default_sink, &tx)
+            });
+        }
+        Operation::Removed => {
+            debug!("sink removed");
+            remove(i, sinks, tx);
+        }
+    }
+}
+
+pub fn add(info: ListResult<&SinkInfo>, sinks: &ArcMutVec<Sink>, tx: &broadcast::Sender<Event>) {
+    let ListResult::Item(info) = info else {
+        return;
+    };
+
+    lock!(sinks).push(info.into());
+    send!(tx, Event::AddSink(info.into()));
+}
+
+fn update(
+    info: ListResult<&SinkInfo>,
+    sinks: &ArcMutVec<Sink>,
+    default_sink: &Arc<Mutex<Option<String>>>,
+    tx: &broadcast::Sender<Event>,
+) {
+    let ListResult::Item(info) = info else {
+        return;
+    };
+
+    {
+        let mut sinks = lock!(sinks);
+        let Some(pos) = sinks.iter().position(|sink| sink.index == info.index) else {
+            error!("received update to untracked sink input");
+            return;
+        };
+
+        sinks[pos] = info.into();
+
+        // update in local copy
+        if !sinks[pos].active {
+            if let Some(default_sink) = &*lock!(default_sink) {
+                sinks[pos].active = &sinks[pos].name == default_sink;
+            }
+        }
+    }
+
+    let mut sink: Sink = info.into();
+
+    // update in broadcast copy
+    if !sink.active {
+        if let Some(default_sink) = &*lock!(default_sink) {
+            sink.active = &sink.name == default_sink;
+        }
+    }
+
+    send!(tx, Event::UpdateSink(sink));
+}
+
+fn remove(index: u32, sinks: &ArcMutVec<Sink>, tx: &broadcast::Sender<Event>) {
+    let mut sinks = lock!(sinks);
+
+    if let Some(pos) = sinks.iter().position(|s| s.index == index) {
+        let info = sinks.remove(pos);
+        send!(tx, Event::RemoveSink(info.name));
+    }
+}

src/clients/volume/sink_input.rs

@@ -0,0 +1,148 @@
+use super::{percent_to_volume, volume_to_percent, ArcMutVec, Client, ConnectionState, Event};
+use crate::{lock, send};
+use libpulse_binding::callbacks::ListResult;
+use libpulse_binding::context::introspect::SinkInputInfo;
+use libpulse_binding::context::subscribe::Operation;
+use libpulse_binding::context::Context;
+use std::sync::{mpsc, Arc, Mutex};
+use tokio::sync::broadcast;
+use tracing::{debug, error};
+
+#[derive(Debug, Clone)]
+pub struct SinkInput {
+    pub index: u32,
+    pub name: String,
+    pub volume: f64,
+    pub muted: bool,
+
+    pub can_set_volume: bool,
+}
+
+impl From<&SinkInputInfo<'_>> for SinkInput {
+    fn from(value: &SinkInputInfo) -> Self {
+        Self {
+            index: value.index,
+            name: value
+                .name
+                .as_ref()
+                .map(ToString::to_string)
+                .unwrap_or_default(),
+            muted: value.mute,
+            volume: volume_to_percent(value.volume),
+            can_set_volume: value.has_volume && value.volume_writable,
+        }
+    }
+}
+
+impl Client {
+    pub fn sink_inputs(&self) -> Arc<Mutex<Vec<SinkInput>>> {
+        self.data.sink_inputs.clone()
+    }
+
+    pub fn set_input_volume(&self, index: u32, volume_percent: f64) {
+        if let ConnectionState::Connected { introspector, .. } = &mut *lock!(self.connection) {
+            let (tx, rx) = mpsc::channel();
+
+            introspector.get_sink_input_info(index, move |info| {
+                let ListResult::Item(info) = info else {
+                    return;
+                };
+                send!(tx, info.volume);
+            });
+
+            let new_volume = percent_to_volume(volume_percent);
+
+            let mut volume = rx.recv().expect("to receive info");
+            for v in volume.get_mut() {
+                v.0 = new_volume;
+            }
+
+            introspector.set_sink_input_volume(index, &volume, None);
+        }
+    }
+
+    pub fn set_input_muted(&self, index: u32, muted: bool) {
+        if let ConnectionState::Connected { introspector, .. } = &mut *lock!(self.connection) {
+            introspector.set_sink_input_mute(index, muted, None);
+        }
+    }
+}
+
+pub fn on_event(
+    context: &Arc<Mutex<Context>>,
+    inputs: &ArcMutVec<SinkInput>,
+    tx: &broadcast::Sender<Event>,
+    op: Operation,
+    i: u32,
+) {
+    let introspect = lock!(context).introspect();
+
+    match op {
+        Operation::New => {
+            debug!("new sink input");
+            introspect.get_sink_input_info(i, {
+                let inputs = inputs.clone();
+                let tx = tx.clone();
+
+                move |info| add(info, &inputs, &tx)
+            });
+        }
+        Operation::Changed => {
+            debug!("sink input changed");
+            introspect.get_sink_input_info(i, {
+                let inputs = inputs.clone();
+                let tx = tx.clone();
+
+                move |info| update(info, &inputs, &tx)
+            });
+        }
+        Operation::Removed => {
+            debug!("sink input removed");
+            remove(i, inputs, tx);
+        }
+    }
+}
+
+pub fn add(
+    info: ListResult<&SinkInputInfo>,
+    inputs: &ArcMutVec<SinkInput>,
+    tx: &broadcast::Sender<Event>,
+) {
+    let ListResult::Item(info) = info else {
+        return;
+    };
+
+    lock!(inputs).push(info.into());
+    send!(tx, Event::AddInput(info.into()));
+}
+
+fn update(
+    info: ListResult<&SinkInputInfo>,
+    inputs: &ArcMutVec<SinkInput>,
+    tx: &broadcast::Sender<Event>,
+) {
+    let ListResult::Item(info) = info else {
+        return;
+    };
+
+    {
+        let mut inputs = lock!(inputs);
+        let Some(pos) = inputs.iter().position(|input| input.index == info.index) else {
+            error!("received update to untracked sink input");
+            return;
+        };
+
+        inputs[pos] = info.into();
+    }
+
+    send!(tx, Event::UpdateInput(info.into()));
+}
+
+fn remove(index: u32, inputs: &ArcMutVec<SinkInput>, tx: &broadcast::Sender<Event>) {
+    let mut inputs = lock!(inputs);
+
+    if let Some(pos) = inputs.iter().position(|s| s.index == index) {
+        let info = inputs.remove(pos);
+        send!(tx, Event::RemoveInput(info.index));
+    }
+}