fix(networkmanager): notify upon new device from watch_device()

Additionally:
- Pass the device proxy to watch_device_state() now that we've had to
  create one anyway
- Improve event received logging in module widget

src/clients/networkmanager/event.rs

@@ -1,13 +1,18 @@
 use crate::clients::networkmanager::dbus::{DeviceState, DeviceType};
 
 #[derive(Debug, Clone)]
-pub enum Event {
+pub enum ClientToModuleEvent {
-    DeviceAdded {
+    DeviceChanged {
-        interface: String,
+        number: u32,
-    },
-    DeviceStateChanged {
-        interface: String,
         r#type: DeviceType,
-        state: DeviceState,
+        new_state: DeviceState,
+    },
+    DeviceRemoved {
+        number: u32,
     },
 }
+
+#[derive(Debug, Clone)]
+pub enum ModuleToClientEvent {
+    NewController,
+}

src/clients/networkmanager/mod.rs

@@ -1,15 +1,17 @@
 use color_eyre::Result;
 use color_eyre::eyre::Ok;
 use futures_lite::StreamExt;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
-use tokio::sync::broadcast;
+use tokio::sync::{RwLock, broadcast};
+use tokio::task::JoinHandle;
+use tracing::debug;
 use zbus::Connection;
-use zbus::zvariant::{ObjectPath, Str};
+use zbus::zvariant::ObjectPath;
 
 use crate::clients::ClientResult;
 use crate::clients::networkmanager::dbus::{DbusProxy, DeviceDbusProxy};
-use crate::clients::networkmanager::event::Event;
+use crate::clients::networkmanager::event::{ClientToModuleEvent, ModuleToClientEvent};
 use crate::{register_fallible_client, spawn};
 
 pub mod dbus;
@@ -17,102 +19,221 @@ pub mod event;
 
 #[derive(Debug)]
 pub struct Client {
-    tx: broadcast::Sender<Event>,
+    inner: &'static ClientInner,
 }
 
 impl Client {
-    fn new() -> Result<Client> {
+    fn new() -> Client {
-        let (tx, _) = broadcast::channel(64);
+        let inner = Box::leak(Box::new(ClientInner::new()));
-        Ok(Client { tx })
+        Client { inner }
     }
 
     fn run(&self) -> Result<()> {
-        let tx = self.tx.clone();
+        self.inner.run()
-        spawn(async move {
+    }
-            let dbus_connection = Connection::system().await?;
-            let root = DbusProxy::new(&dbus_connection).await?;
 
-            let mut devices = HashSet::new();
+    pub fn subscribe(&self) -> broadcast::Receiver<ClientToModuleEvent> {
+        self.inner.subscribe()
+    }
+
+    pub fn get_sender(&self) -> broadcast::Sender<ModuleToClientEvent> {
+        self.inner.get_sender()
+    }
+}
+
+#[derive(Debug)]
+struct ClientInner {
+    controller_sender: broadcast::Sender<ClientToModuleEvent>,
+    sender: broadcast::Sender<ModuleToClientEvent>,
+    device_watchers: RwLock<HashMap<ObjectPath<'static>, DeviceWatcher>>,
+    dbus_connection: RwLock<Option<Connection>>,
+}
+
+#[derive(Clone, Debug)]
+struct DeviceWatcher {
+    state_watcher: Arc<JoinHandle<Result<()>>>,
+}
+
+impl ClientInner {
+    fn new() -> ClientInner {
+        let (controller_sender, _) = broadcast::channel(64);
+        let (sender, _) = broadcast::channel(8);
+        let device_watchers = RwLock::new(HashMap::new());
+        let dbus_connection = RwLock::new(None);
+        ClientInner {
+            controller_sender,
+            sender,
+            device_watchers,
+            dbus_connection,
+        }
+    }
+
+    fn run(&'static self) -> Result<()> {
+        debug!("Client running");
+
+        spawn(self.watch_devices_list());
+
+        let receiver = self.sender.subscribe();
+        spawn(self.handle_received_events(receiver));
+
+        Ok(())
+    }
+
+    fn subscribe(&self) -> broadcast::Receiver<ClientToModuleEvent> {
+        self.controller_sender.subscribe()
+    }
+
+    fn get_sender(&self) -> broadcast::Sender<ModuleToClientEvent> {
+        self.sender.clone()
+    }
+
+    async fn watch_devices_list(&'static self) -> Result<()> {
+        debug!("D-Bus devices list watcher starting");
+
+        let root = DbusProxy::new(&self.dbus_connection().await?).await?;
 
         let mut devices_changes = root.receive_all_devices_changed().await;
         while let Some(devices_change) = devices_changes.next().await {
             // The new list of devices from dbus, not to be confused with the added devices below
-                let new_devices = HashSet::from_iter(devices_change.get().await?);
+            let new_device_paths = devices_change
+                .get()
+                .await?
+                .iter()
+                .map(ObjectPath::to_owned)
+                .collect::<HashSet<_>>();
 
-                let added_devices = new_devices.difference(&devices);
+            let mut watchers = self.device_watchers.write().await;
-                for added_device in added_devices {
+            let device_paths = watchers.keys().cloned().collect::<HashSet<_>>();
-                    spawn(watch_device(added_device.to_owned(), tx.clone()));
+
+            let added_device_paths = new_device_paths.difference(&device_paths);
+            for added_device_path in added_device_paths {
+                debug_assert!(!watchers.contains_key(added_device_path));
+
+                let watcher = self.watch_device(added_device_path.clone()).await?;
+                watchers.insert(added_device_path.clone(), watcher);
             }
 
-                let _removed_devices = devices.difference(&new_devices);
+            let removed_device_paths = device_paths.difference(&new_device_paths);
-                // TODO: Cook up some way to notify closures for removed devices to exit
+            for removed_device_path in removed_device_paths {
+                let watcher = watchers
+                    .get(removed_device_path)
+                    .expect("Device to be removed should be present in watchers");
+                watcher.state_watcher.abort();
+                watchers.remove(removed_device_path);
 
-                devices = new_devices;
+                let number = get_number_from_dbus_path(removed_device_path);
+                self.controller_sender
+                    .send(ClientToModuleEvent::DeviceRemoved { number })?;
+
+                debug!("D-bus device watchers for {} stopped", removed_device_path);
+            }
         }
 
         Ok(())
-        });
+    }
+
+    async fn handle_received_events(
+        &'static self,
+        mut receiver: broadcast::Receiver<ModuleToClientEvent>,
+    ) -> Result<()> {
+        while let Result::Ok(event) = receiver.recv().await {
+            match event {
+                ModuleToClientEvent::NewController => {
+                    debug!("Client received NewController event");
+
+                    for device_path in self.device_watchers.read().await.keys() {
+                        let dbus_connection = &self.dbus_connection().await?;
+                        let device = DeviceDbusProxy::new(dbus_connection, device_path).await?;
+
+                        let number = get_number_from_dbus_path(device_path);
+                        let r#type = device.device_type().await?;
+                        let new_state = device.state().await?;
+                        self.controller_sender
+                            .send(ClientToModuleEvent::DeviceChanged {
+                                number,
+                                r#type,
+                                new_state,
+                            })?;
+                    }
+                }
+            }
+        }
 
         Ok(())
     }
 
-    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
+    async fn watch_device(&'static self, path: ObjectPath<'_>) -> Result<DeviceWatcher> {
-        self.tx.subscribe()
+        let dbus_connection = &self.dbus_connection().await?;
+        let proxy = DeviceDbusProxy::new(dbus_connection, path.to_owned()).await?;
+
+        let number = get_number_from_dbus_path(&path);
+        let r#type = proxy.device_type().await?;
+        let new_state = proxy.state().await?;
+
+        // Notify modules that the device exists even if its properties don't change
+        self.controller_sender
+            .send(ClientToModuleEvent::DeviceChanged {
+                number,
+                r#type: r#type.clone(),
+                new_state,
+            })?;
+
+        let state_watcher = Arc::new(spawn(self.watch_device_state(proxy)));
+
+        Ok(DeviceWatcher { state_watcher })
+    }
+
+    async fn watch_device_state(&'static self, proxy: DeviceDbusProxy<'_>) -> Result<()> {
+        let path = proxy.inner().path();
+
+        debug!("D-Bus device state watcher for {} starting", path);
+
+        let number = get_number_from_dbus_path(path);
+        let r#type = proxy.device_type().await?;
+
+        let mut changes = proxy.receive_state_changed().await;
+        while let Some(change) = changes.next().await {
+            let new_state = change.get().await?;
+            self.controller_sender
+                .send(ClientToModuleEvent::DeviceChanged {
+                    number,
+                    r#type: r#type.clone(),
+                    new_state,
+                })?;
+        }
+
+        Ok(())
+    }
+
+    async fn dbus_connection(&self) -> Result<Connection> {
+        let dbus_connection_guard = self.dbus_connection.read().await;
+        if let Some(dbus_connection) = &*dbus_connection_guard {
+            Ok(dbus_connection.clone())
+        } else {
+            // Yes it's a bit awkward to first obtain a read lock and then a write lock but it
+            //  needs to happen only once, and after that all read lock acquisitions will be
+            //  instant
+            drop(dbus_connection_guard);
+            let dbus_connection = Connection::system().await?;
+            *self.dbus_connection.write().await = Some(dbus_connection.clone());
+            Ok(dbus_connection)
+        }
     }
 }
 
 pub fn create_client() -> ClientResult<Client> {
-    let client = Arc::new(Client::new()?);
+    let client = Arc::new(Client::new());
     client.run()?;
     Ok(client)
 }
 
-async fn watch_device(device_path: ObjectPath<'_>, tx: broadcast::Sender<Event>) -> Result<()> {
+fn get_number_from_dbus_path(path: &ObjectPath) -> u32 {
-    let dbus_connection = Connection::system().await?;
+    let (_, number_str) = path
-    let device = DeviceDbusProxy::new(&dbus_connection, device_path.to_owned()).await?;
+        .rsplit_once('/')
-
+        .expect("Path must have at least two segments to contain an object number");
-    let interface = device.interface().await?;
+    number_str
-    tx.send(Event::DeviceAdded {
+        .parse()
-        interface: interface.to_string(),
+        .expect("Last segment was not a positive integer")
-    })?;
-
-    spawn(watch_device_state(
-        device_path.to_owned(),
-        interface.to_owned(),
-        tx.clone(),
-    ));
-
-    Ok(())
-}
-
-async fn watch_device_state(
-    device_path: ObjectPath<'_>,
-    interface: Str<'_>,
-    tx: broadcast::Sender<Event>,
-) -> Result<()> {
-    let dbus_connection = Connection::system().await?;
-    let device = DeviceDbusProxy::new(&dbus_connection, &device_path).await?;
-    let r#type = device.device_type().await?;
-
-    // Send an event communicating the initial state
-    let state = device.state().await?;
-    tx.send(Event::DeviceStateChanged {
-        interface: interface.to_string(),
-        r#type: r#type.clone(),
-        state,
-    })?;
-
-    let mut state_changes = device.receive_state_changed().await;
-    while let Some(state_change) = state_changes.next().await {
-        let state = state_change.get().await?;
-        tx.send(Event::DeviceStateChanged {
-            interface: interface.to_string(),
-            r#type: r#type.clone(),
-            state,
-        })?;
-    }
-
-    Ok(())
 }
 
 register_fallible_client!(Client, network_manager);

src/clients/volume/mod.rs

@@ -3,7 +3,7 @@ mod sink_input;
 
 use crate::{APP_ID, arc_mut, lock, register_client, spawn_blocking};
 use libpulse_binding::callbacks::ListResult;
-use libpulse_binding::context::introspect::ServerInfo;
+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};
@@ -24,11 +24,11 @@ type ArcMutVec<T> = Arc<Mutex<Vec<T>>>;
 pub enum Event {
     AddSink(Sink),
     UpdateSink(Sink),
-    RemoveSink,
+    RemoveSink(String),
 
-    AddInput,
+    AddInput(SinkInput),
-    UpdateInput,
+    UpdateInput(SinkInput),
-    RemoveInput,
+    RemoveInput(u32),
 }
 
 #[derive(Debug)]
@@ -51,7 +51,10 @@ struct Data {
 
 pub enum ConnectionState {
     Disconnected,
-    Connected,
+    Connected {
+        context: Arc<Mutex<Context>>,
+        introspector: Introspector,
+    },
 }
 
 impl Debug for ConnectionState {
@@ -61,7 +64,7 @@ impl Debug for ConnectionState {
             "{}",
             match self {
                 Self::Disconnected => "Disconnected",
-                Self::Connected => "Connected",
+                Self::Connected { .. } => "Connected",
             }
         )
     }
@@ -117,9 +120,14 @@ impl Client {
             error!("{err:?}");
         }
 
+        let introspector = lock!(context).introspect();
+
         {
             let mut inner = lock!(self.connection);
-            *inner = ConnectionState::Connected;
+            *inner = ConnectionState::Connected {
+                context,
+                introspector,
+            };
         }
 
         loop {
@@ -283,4 +291,22 @@ fn volume_to_percent(volume: ChannelVolumes) -> f64 {
     ((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

@@ -1,4 +1,4 @@
-use super::{ArcMutVec, Client, Event, volume_to_percent};
+use super::{ArcMutVec, Client, ConnectionState, Event, percent_to_volume, volume_to_percent};
 use crate::channels::SyncSenderExt;
 use crate::lock;
 use libpulse_binding::callbacks::ListResult;
@@ -6,7 +6,7 @@ use libpulse_binding::context::Context;
 use libpulse_binding::context::introspect::SinkInfo;
 use libpulse_binding::context::subscribe::Operation;
 use libpulse_binding::def::SinkState;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, mpsc};
 use tokio::sync::broadcast;
 use tracing::{debug, error, instrument, trace};
 
@@ -14,6 +14,7 @@ use tracing::{debug, error, instrument, trace};
 pub struct Sink {
     index: u32,
     pub name: String,
+    pub description: String,
     pub volume: f64,
     pub muted: bool,
     pub active: bool,
@@ -28,6 +29,11 @@ impl From<&SinkInfo<'_>> for Sink {
                 .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,
@@ -40,6 +46,43 @@ impl Client {
     pub fn sinks(&self) -> Arc<Mutex<Vec<Sink>>> {
         self.data.sinks.clone()
     }
+
+    #[instrument(level = "trace")]
+    pub fn set_default_sink(&self, name: &str) {
+        if let ConnectionState::Connected { context, .. } = &*lock!(self.connection) {
+            lock!(context).set_default_sink(name, |_| {});
+        }
+    }
+
+    #[instrument(level = "trace")]
+    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;
+                };
+                tx.send_expect(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);
+        }
+    }
+
+    #[instrument(level = "trace")]
+    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(
@@ -134,9 +177,10 @@ fn update(
 fn remove(index: u32, sinks: &ArcMutVec<Sink>, tx: &broadcast::Sender<Event>) {
     trace!("removing {index}");
 
-    let sinks = lock!(sinks);
+    let mut sinks = lock!(sinks);
 
-    if let Some(_pos) = sinks.iter().position(|s| s.index == index) {
+    if let Some(pos) = sinks.iter().position(|s| s.index == index) {
-        tx.send_expect(Event::RemoveSink);
+        let info = sinks.remove(pos);
+        tx.send_expect(Event::RemoveSink(info.name));
     }
 }

src/clients/volume/sink_input.rs

@@ -1,22 +1,37 @@
-use super::{ArcMutVec, Client, Event};
+use super::{ArcMutVec, Client, ConnectionState, Event, percent_to_volume, volume_to_percent};
 use crate::channels::SyncSenderExt;
 use crate::lock;
 use libpulse_binding::callbacks::ListResult;
 use libpulse_binding::context::Context;
 use libpulse_binding::context::introspect::SinkInputInfo;
 use libpulse_binding::context::subscribe::Operation;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, mpsc};
 use tokio::sync::broadcast;
 use tracing::{debug, error, instrument, trace};
 
 #[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 }
+        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,
+        }
     }
 }
 
@@ -25,6 +40,36 @@ impl Client {
     pub fn sink_inputs(&self) -> Arc<Mutex<Vec<SinkInput>>> {
         self.data.sink_inputs.clone()
     }
+
+    #[instrument(level = "trace")]
+    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;
+                };
+                tx.send_expect(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);
+        }
+    }
+
+    #[instrument(level = "trace")]
+    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(
@@ -74,7 +119,7 @@ pub fn add(
     trace!("adding {info:?}");
 
     lock!(inputs).push(info.into());
-    tx.send_expect(Event::AddInput);
+    tx.send_expect(Event::AddInput(info.into()));
 }
 
 fn update(
@@ -98,15 +143,16 @@ fn update(
         inputs[pos] = info.into();
     }
 
-    tx.send_expect(Event::UpdateInput);
+    tx.send_expect(Event::UpdateInput(info.into()));
 }
 
 fn remove(index: u32, inputs: &ArcMutVec<SinkInput>, tx: &broadcast::Sender<Event>) {
-    let inputs = lock!(inputs);
+    let mut inputs = lock!(inputs);
 
     trace!("removing {index}");
 
-    if let Some(_pos) = inputs.iter().position(|s| s.index == index) {
+    if let Some(pos) = inputs.iter().position(|s| s.index == index) {
-        tx.send_expect(Event::RemoveInput);
+        let info = inputs.remove(pos);
+        tx.send_expect(Event::RemoveInput(info.index));
     }
 }

src/modules/networkmanager.rs

@@ -1,6 +1,6 @@
 use crate::clients::networkmanager::Client;
 use crate::clients::networkmanager::dbus::{DeviceState, DeviceType};
-use crate::clients::networkmanager::event::Event;
+use crate::clients::networkmanager::event::{ClientToModuleEvent, ModuleToClientEvent};
 use crate::config::CommonConfig;
 use crate::gtk_helpers::IronbarGtkExt;
 use crate::image::Provider;
@@ -13,6 +13,7 @@ use gtk::{Image, Orientation};
 use serde::Deserialize;
 use std::collections::HashMap;
 use tokio::sync::{broadcast, mpsc};
+use tracing::debug;
 
 #[derive(Debug, Deserialize, Clone)]
 #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
@@ -29,7 +30,7 @@ const fn default_icon_size() -> i32 {
 }
 
 impl Module<gtk::Box> for NetworkManagerModule {
-    type SendMessage = Event;
+    type SendMessage = ClientToModuleEvent;
     type ReceiveMessage = ();
 
     module_impl!("network_manager");
@@ -37,19 +38,24 @@ impl Module<gtk::Box> for NetworkManagerModule {
     fn spawn_controller(
         &self,
         _info: &ModuleInfo,
-        context: &WidgetContext<Event, ()>,
+        context: &WidgetContext<ClientToModuleEvent, ()>,
-        _rx: mpsc::Receiver<()>,
+        _widget_receiver: mpsc::Receiver<()>,
     ) -> Result<()> {
         let client = context.try_client::<Client>()?;
         // Should we be using context.tx with ModuleUpdateEvent::Update instead?
-        let tx = context.update_tx.clone();
+        let widget_sender = context.update_tx.clone();
-        // Must be done here synchronously to avoid race condition
-        let mut client_rx = client.subscribe();
-        spawn(async move {
-            while let Result::Ok(event) = client_rx.recv().await {
-                tx.send(event)?;
-            }
 
+        // Must be done here otherwise we miss the response to our `NewController` event
+        let mut client_receiver = client.subscribe();
+
+        client
+            .get_sender()
+            .send(ModuleToClientEvent::NewController)?;
+
+        spawn(async move {
+            while let Result::Ok(event) = client_receiver.recv().await {
+                widget_sender.send(event)?;
+            }
             Ok(())
         });
 
@@ -58,16 +64,17 @@ impl Module<gtk::Box> for NetworkManagerModule {
 
     fn into_widget(
         self,
-        context: WidgetContext<Event, ()>,
+        context: WidgetContext<ClientToModuleEvent, ()>,
         _info: &ModuleInfo,
     ) -> Result<ModuleParts<gtk::Box>> {
+        // Must be done here otherwise we miss the response to our `NewController` event
+        let receiver = context.subscribe();
+
         let container = gtk::Box::new(Orientation::Horizontal, 0);
 
-        // Must be done here synchronously to avoid race condition
-        let rx = context.subscribe();
         // We cannot use recv_glib_async here because the lifetimes don't work out
         spawn_future_local(handle_update_events(
-            rx,
+            receiver,
             container.clone(),
             self.icon_size,
             context.ironbar.image_provider(),
@@ -78,31 +85,37 @@ impl Module<gtk::Box> for NetworkManagerModule {
 }
 
 async fn handle_update_events(
-    mut rx: broadcast::Receiver<Event>,
+    mut widget_receiver: broadcast::Receiver<ClientToModuleEvent>,
     container: gtk::Box,
     icon_size: i32,
     image_provider: Provider,
-) {
+) -> Result<()> {
-    let mut icons = HashMap::<String, Image>::new();
+    // TODO: Ensure the visible icons are always in the same order
+    let mut icons = HashMap::<u32, Image>::new();
 
-    while let Result::Ok(event) = rx.recv().await {
+    while let Result::Ok(event) = widget_receiver.recv().await {
         match event {
-            Event::DeviceAdded { interface, .. } => {
+            ClientToModuleEvent::DeviceChanged {
+                number,
+                r#type,
+                new_state,
+            } => {
+                debug!(
+                    "Module widget received DeviceChanged event for number {}",
+                    number
+                );
+
+                let icon: &_ = icons.entry(number).or_insert_with(|| {
+                    debug!("Adding icon for device {}", number);
+
                     let icon = Image::new();
                     icon.add_class("icon");
                     container.add(&icon);
-                icons.insert(interface, icon);
+                    icon
-            }
+                });
-            Event::DeviceStateChanged {
+
-                interface,
-                r#type,
-                state,
-            } => {
-                let icon = icons
-                    .get(&interface)
-                    .expect("the icon for the interface to be present");
                 // TODO: Make this configurable at runtime
-                let icon_name = get_icon_for_device_state(&r#type, &state);
+                let icon_name = get_icon_for_device_state(&r#type, &new_state);
                 match icon_name {
                     Some(icon_name) => {
                         image_provider
@@ -115,8 +128,22 @@ async fn handle_update_events(
                     }
                 }
             }
-        };
+            ClientToModuleEvent::DeviceRemoved { number } => {
+                debug!(
+                    "Module widget received DeviceRemoved event for number {}",
+                    number
+                );
+
+                let icon = icons
+                    .get(&number)
+                    .expect("The icon for {} was about to be removed but was not present");
+                container.remove(icon);
+                icons.remove(&number);
             }
+        }
+    }
+
+    Ok(())
 }
 
 fn get_icon_for_device_state(r#type: &DeviceType, state: &DeviceState) -> Option<&'static str> {

src/modules/volume.rs

@@ -1,35 +1,140 @@
 use crate::channels::{AsyncSenderExt, BroadcastReceiverExt};
 use crate::clients::volume::{self, Event};
-use crate::config::CommonConfig;
+use crate::config::{CommonConfig, LayoutConfig, TruncateMode};
-use crate::gtk_helpers::IronbarGtkExt;
+use crate::gtk_helpers::{IronbarGtkExt, IronbarLabelExt};
 use crate::modules::{
-    Module, ModuleInfo, ModuleParts, ModuleUpdateEvent, PopupButton, WidgetContext,
+    Module, ModuleInfo, ModuleParts, ModulePopup, ModuleUpdateEvent, PopupButton, WidgetContext,
 };
 use crate::{lock, module_impl, spawn};
+use glib::Propagation;
+use gtk::pango::EllipsizeMode;
 use gtk::prelude::*;
-use gtk::{Button, Image};
+use gtk::{Button, CellRendererText, ComboBoxText, Label, Orientation, Scale, ToggleButton};
 use serde::Deserialize;
+use std::collections::HashMap;
 use tokio::sync::mpsc;
 use tracing::trace;
 
 #[derive(Debug, Clone, Deserialize)]
 #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
 pub struct VolumeModule {
-    #[serde(default = "default_icon_size")]
+    /// The format string to use for the widget button label.
-    icon_size: i32,
+    /// For available tokens, see [below](#formatting-tokens).
+    ///
+    /// **Default**: `{icon} {percentage}%`
+    #[serde(default = "default_format")]
+    format: String,
+
+    /// Maximum value to allow volume sliders to reach.
+    /// Pulse supports values > 100 but this may result in distortion.
+    ///
+    /// **Default**: `100`
+    #[serde(default = "default_max_volume")]
+    max_volume: f64,
+
+    /// Volume state icons.
+    ///
+    /// See [icons](#icons).
+    #[serde(default)]
+    icons: Icons,
 
     // -- Common --
+    /// See [truncate options](module-level-options#truncate-mode).
+    ///
+    /// **Default**: `null`
+    pub(crate) truncate: Option<TruncateMode>,
+
+    /// See [layout options](module-level-options#layout)
+    #[serde(default, flatten)]
+    layout: LayoutConfig,
+
     /// See [common options](module-level-options#common-options).
     #[serde(flatten)]
     pub common: Option<CommonConfig>,
 }
 
-const fn default_icon_size() -> i32 {
+fn default_format() -> String {
-    24
+    String::from("{icon} {percentage}%")
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
+pub struct Icons {
+    /// Icon to show for high volume levels.
+    ///
+    /// **Default**: `󰕾`
+    #[serde(default = "default_icon_volume_high")]
+    volume_high: String,
+
+    /// Icon to show for medium volume levels.
+    ///
+    /// **Default**: `󰖀`
+    #[serde(default = "default_icon_volume_medium")]
+    volume_medium: String,
+
+    /// Icon to show for low volume levels.
+    ///
+    /// **Default**: `󰕿`
+    #[serde(default = "default_icon_volume_low")]
+    volume_low: String,
+
+    /// Icon to show for muted outputs.
+    ///
+    /// **Default**: `󰝟`
+    #[serde(default = "default_icon_muted")]
+    muted: String,
+}
+
+impl Icons {
+    fn volume_icon(&self, volume_percent: f64) -> &str {
+        match volume_percent as u32 {
+            0..=33 => &self.volume_low,
+            34..=66 => &self.volume_medium,
+            67.. => &self.volume_high,
+        }
+    }
+}
+
+impl Default for Icons {
+    fn default() -> Self {
+        Self {
+            volume_high: default_icon_volume_high(),
+            volume_medium: default_icon_volume_medium(),
+            volume_low: default_icon_volume_low(),
+            muted: default_icon_muted(),
+        }
+    }
+}
+
+const fn default_max_volume() -> f64 {
+    100.0
+}
+
+fn default_icon_volume_high() -> String {
+    String::from("󰕾")
+}
+
+fn default_icon_volume_medium() -> String {
+    String::from("󰖀")
+}
+
+fn default_icon_volume_low() -> String {
+    String::from("󰕿")
+}
+
+fn default_icon_muted() -> String {
+    String::from("󰝟")
 }
 
 #[derive(Debug, Clone)]
-pub enum Update {}
+pub enum Update {
+    SinkChange(String),
+    SinkVolume(String, f64),
+    SinkMute(String, bool),
+
+    InputVolume(u32, f64),
+    InputMute(u32, bool),
+}
 
 impl Module<Button> for VolumeModule {
     type SendMessage = Event;
@@ -41,7 +146,7 @@ impl Module<Button> for VolumeModule {
         &self,
         _info: &ModuleInfo,
         context: &WidgetContext<Self::SendMessage, Self::ReceiveMessage>,
-        _rx: mpsc::Receiver<Self::ReceiveMessage>,
+        mut rx: mpsc::Receiver<Self::ReceiveMessage>,
     ) -> color_eyre::Result<()>
     where
         <Self as Module<Button>>::SendMessage: Clone,
@@ -75,8 +180,8 @@ impl Module<Button> for VolumeModule {
                     tx.send_update(Event::AddSink(sink)).await;
                 }
 
-                for _input in inputs {
+                for input in inputs {
-                    tx.send_update(Event::AddInput).await;
+                    tx.send_update(Event::AddInput(input)).await;
                 }
 
                 // recv loop
@@ -87,18 +192,38 @@ impl Module<Button> for VolumeModule {
             });
         }
 
+        // ui events
+        spawn(async move {
+            while let Some(update) = rx.recv().await {
+                match update {
+                    Update::SinkChange(name) => client.set_default_sink(&name),
+                    Update::SinkVolume(name, volume) => client.set_sink_volume(&name, volume),
+                    Update::SinkMute(name, muted) => client.set_sink_muted(&name, muted),
+                    Update::InputVolume(index, volume) => client.set_input_volume(index, volume),
+                    Update::InputMute(index, muted) => client.set_input_muted(index, muted),
+                }
+            }
+        });
+
         Ok(())
     }
 
     fn into_widget(
         self,
         context: WidgetContext<Self::SendMessage, Self::ReceiveMessage>,
-        _info: &ModuleInfo,
+        info: &ModuleInfo,
     ) -> color_eyre::Result<ModuleParts<Button>>
     where
         <Self as Module<Button>>::SendMessage: Clone,
     {
+        let button_label = Label::builder()
+            .use_markup(true)
+            .angle(self.layout.angle(info))
+            .justify(self.layout.justify.into())
+            .build();
+
         let button = Button::new();
+        button.add(&button_label);
 
         {
             let tx = context.tx.clone();
@@ -110,44 +235,261 @@ impl Module<Button> for VolumeModule {
 
         let rx = context.subscribe();
 
-        let image_icon = Image::new();
+        rx.recv_glib(
-        image_icon.add_class("icon");
+            (&self.icons, &self.format),
-        button.set_image(Some(&image_icon));
+            move |(icons, format), event| match event {
-
-        rx.recv_glib_async((), move |(), event| {
-            let image_icon = image_icon.clone();
-            let image_provider = context.ironbar.image_provider();
-
-            async move {
-                match event {
                 Event::AddSink(sink) | Event::UpdateSink(sink) if sink.active => {
-                        image_provider
+                    let label = format
-                            .load_into_image_silent(
+                        .replace(
-                                &determine_volume_icon(sink.muted, sink.volume),
+                            "{icon}",
-                                self.icon_size,
+                            if sink.muted {
-                                false,
+                                &icons.muted
-                                &image_icon,
+                            } else {
+                                icons.volume_icon(sink.volume)
+                            },
                         )
-                            .await;
+                        .replace("{percentage}", &sink.volume.to_string())
+                        .replace("{name}", &sink.description);
+
+                    button_label.set_label_escaped(&label);
                 }
                 _ => {}
+            },
+        );
+
+        let popup = self
+            .into_popup(context, info)
+            .into_popup_parts(vec![&button]);
+
+        Ok(ModuleParts::new(button, popup))
+    }
+
+    fn into_popup(
+        self,
+        context: WidgetContext<Self::SendMessage, Self::ReceiveMessage>,
+        _info: &ModuleInfo,
+    ) -> Option<gtk::Box>
+    where
+        Self: Sized,
+    {
+        let container = gtk::Box::new(Orientation::Horizontal, 10);
+
+        let sink_container = gtk::Box::new(Orientation::Vertical, 5);
+        sink_container.add_class("device-box");
+
+        let input_container = gtk::Box::new(Orientation::Vertical, 5);
+        input_container.add_class("apps-box");
+
+        container.add(&sink_container);
+        container.add(&input_container);
+
+        let sink_selector = ComboBoxText::new();
+        sink_selector.add_class("device-selector");
+
+        let renderer = sink_selector
+            .cells()
+            .first()
+            .expect("to exist")
+            .clone()
+            .downcast::<CellRendererText>()
+            .expect("to be valid cast");
+
+        renderer.set_width_chars(20);
+        renderer.set_ellipsize(EllipsizeMode::End);
+
+        {
+            let tx = context.controller_tx.clone();
+            sink_selector.connect_changed(move |selector| {
+                if let Some(name) = selector.active_id() {
+                    tx.send_spawn(Update::SinkChange(name.into()));
+                }
+            });
+        }
+
+        sink_container.add(&sink_selector);
+
+        let slider = Scale::builder()
+            .orientation(Orientation::Vertical)
+            .height_request(100)
+            .inverted(true)
+            .build();
+
+        slider.add_class("slider");
+
+        slider.set_range(0.0, self.max_volume);
+        slider.set_value(50.0);
+        sink_container.add(&slider);
+
+        {
+            let tx = context.controller_tx.clone();
+            let selector = sink_selector.clone();
+
+            slider.connect_button_release_event(move |scale, _| {
+                if let Some(sink) = selector.active_id() {
+                    // GTK will send values outside min/max range
+                    let val = scale.value().clamp(0.0, self.max_volume);
+                    tx.send_spawn(Update::SinkVolume(sink.into(), val));
+                }
+
+                Propagation::Proceed
+            });
+        }
+
+        let btn_mute = ToggleButton::new();
+        btn_mute.add_class("btn-mute");
+        sink_container.add(&btn_mute);
+
+        {
+            let tx = context.controller_tx.clone();
+            let selector = sink_selector.clone();
+
+            btn_mute.connect_toggled(move |btn| {
+                if let Some(sink) = selector.active_id() {
+                    let muted = btn.is_active();
+                    tx.send_spawn(Update::SinkMute(sink.into(), muted));
+                }
+            });
+        }
+
+        container.show_all();
+
+        let mut inputs = HashMap::new();
+        let mut sinks = vec![];
+
+        context
+            .subscribe()
+            .recv_glib(&input_container, move |input_container, event| {
+                match event {
+                    Event::AddSink(info) => {
+                        sink_selector.append(Some(&info.name), &info.description);
+
+                        if info.active {
+                            sink_selector.set_active(Some(sinks.len() as u32));
+                            slider.set_value(info.volume);
+
+                            btn_mute.set_active(info.muted);
+                            btn_mute.set_label(if info.muted {
+                                &self.icons.muted
+                            } else {
+                                self.icons.volume_icon(info.volume)
+                            });
+                        }
+
+                        sinks.push(info);
+                    }
+                    Event::UpdateSink(info) => {
+                        if info.active {
+                            if let Some(pos) = sinks.iter().position(|s| s.name == info.name) {
+                                sink_selector.set_active(Some(pos as u32));
+                                slider.set_value(info.volume);
+
+                                btn_mute.set_active(info.muted);
+                                btn_mute.set_label(if info.muted {
+                                    &self.icons.muted
+                                } else {
+                                    self.icons.volume_icon(info.volume)
+                                });
+                            }
+                        }
+                    }
+                    Event::RemoveSink(name) => {
+                        if let Some(pos) = sinks.iter().position(|s| s.name == name) {
+                            ComboBoxTextExt::remove(&sink_selector, pos as i32);
+                            sinks.remove(pos);
+                        }
+                    }
+
+                    Event::AddInput(info) => {
+                        let index = info.index;
+
+                        let item_container = gtk::Box::new(Orientation::Vertical, 0);
+                        item_container.add_class("app-box");
+
+                        let label = Label::new(Some(&info.name));
+                        label.add_class("title");
+
+                        if let Some(truncate) = self.truncate {
+                            label.truncate(truncate);
+                        };
+
+                        let slider = Scale::builder().sensitive(info.can_set_volume).build();
+                        slider.set_range(0.0, self.max_volume);
+                        slider.set_value(info.volume);
+                        slider.add_class("slider");
+
+                        {
+                            let tx = context.controller_tx.clone();
+                            slider.connect_button_release_event(move |scale, _| {
+                                // GTK will send values outside min/max range
+                                let val = scale.value().clamp(0.0, self.max_volume);
+                                tx.send_spawn(Update::InputVolume(index, val));
+
+                                Propagation::Proceed
+                            });
+                        }
+
+                        let btn_mute = ToggleButton::new();
+                        btn_mute.add_class("btn-mute");
+
+                        btn_mute.set_active(info.muted);
+                        btn_mute.set_label(if info.muted {
+                            &self.icons.muted
+                        } else {
+                            self.icons.volume_icon(info.volume)
+                        });
+
+                        {
+                            let tx = context.controller_tx.clone();
+                            btn_mute.connect_toggled(move |btn| {
+                                let muted = btn.is_active();
+                                tx.send_spawn(Update::InputMute(index, muted));
+                            });
+                        }
+
+                        item_container.add(&label);
+                        item_container.add(&slider);
+                        item_container.add(&btn_mute);
+                        item_container.show_all();
+
+                        input_container.add(&item_container);
+
+                        inputs.insert(
+                            info.index,
+                            InputUi {
+                                container: item_container,
+                                label,
+                                slider,
+                                btn_mute,
+                            },
+                        );
+                    }
+                    Event::UpdateInput(info) => {
+                        if let Some(ui) = inputs.get(&info.index) {
+                            ui.label.set_label(&info.name);
+                            ui.slider.set_value(info.volume);
+                            ui.slider.set_sensitive(info.can_set_volume);
+                            ui.btn_mute.set_label(if info.muted {
+                                &self.icons.muted
+                            } else {
+                                self.icons.volume_icon(info.volume)
+                            });
+                        }
+                    }
+                    Event::RemoveInput(index) => {
+                        if let Some(ui) = inputs.remove(&index) {
+                            input_container.remove(&ui.container);
+                        }
                     }
                 }
             });
 
-        Ok(ModuleParts::new(button, None))
+        Some(container)
     }
 }
 
-fn determine_volume_icon(muted: bool, volume: f64) -> String {
+struct InputUi {
-    let icon_variant = if muted {
+    container: gtk::Box,
-        "muted"
+    label: Label,
-    } else if volume <= 33.3333 {
+    slider: Scale,
-        "low"
+    btn_mute: ToggleButton,
-    } else if volume <= 66.6667 {
-        "medium"
-    } else {
-        "high"
-    };
-    format!("audio-volume-{icon_variant}-symbolic")
 }