Use the extracted version of pa_volume_interface

handlers/home_assistant/src/main.rs

@@ -6,7 +6,6 @@ use crate::handler::Handler;
 mod config;
 mod ha_client;
 mod handler;
-mod util;
 
 #[derive(Debug, Parser)]
 #[command(name = "home_assistant")]

handlers/home_assistant/src/util.rs

@@ -1,69 +0,0 @@
-use std::time::Duration;
-use tokio::sync::mpsc;
-use tokio::sync::mpsc::error::TrySendError;
-use tokio::sync::mpsc::{Receiver, Sender};
-use tokio::time::{timeout, MissedTickBehavior};
-
-/// Sends a message into the output channel after a message in the input channel was received, with a delay of `duration`.
-/// The delay is reset when a new message is reset.
-pub fn spawn_debouncer(duration: Duration) -> (Sender<()>, Receiver<()>) {
-    let (input_sender, mut input_receiver) = mpsc::channel::<()>(1);
-    let (output_sender, output_receiver) = mpsc::channel::<()>(1);
-
-    tokio::spawn(async move {
-        'outer: loop {
-            if input_receiver.recv().await.is_none() {
-                break 'outer;
-            }
-
-            'inner: loop {
-                match timeout(duration, input_receiver.recv()).await {
-                    Ok(None) => break 'outer,
-                    Ok(Some(_)) => continue 'inner,
-                    Err(_) => {
-                        if let Err(TrySendError::Closed(_)) = output_sender.try_send(()) {
-                            break 'outer;
-                        } else {
-                            break 'inner;
-                        }
-                    }
-                }
-            }
-        }
-    });
-
-    (input_sender, output_receiver)
-}
-
-/// Sends messages from the input channel into the output channel, but only if the time since the last message is greater than duration.
-/// The last message that was not sent yet will be sent after duration.
-pub fn spawn_throttler<T: Send + 'static>(duration: Duration) -> (Sender<T>, Receiver<T>) {
-    let (input_sender, mut input_receiver) = mpsc::channel::<T>(25);
-    let (output_sender, output_receiver) = mpsc::channel::<T>(25);
-
-    tokio::spawn(async move {
-        let mut pending_value: Option<T> = None;
-        let mut interval = tokio::time::interval(duration);
-        interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
-
-        'outer: loop {
-            tokio::select! {
-                value = input_receiver.recv() => {
-                    match value {
-                        None => break 'outer,
-                        Some(value) => {
-                            pending_value = Some(value);
-                        }
-                    };
-                }
-                _ = interval.tick() => {
-                    if let Some(value) = pending_value.take() {
-                        output_sender.send(value).await.unwrap();
-                    }
-                }
-            }
-        }
-    });
-
-    (input_sender, output_receiver)
-}

handlers/pa_volume/Cargo.toml

@@ -5,13 +5,13 @@ edition = "2021"
 
 [dependencies]
 deckster_mode = { path = "../../crates/deckster_mode" }
-pa_volume_interface = { path = "../../crates/pa_volume_interface" }
+pulseaudio-volume-interface = { git = "https://git.moritzruth.de/moritzruth/pulseaudio-volume-interface.git" }
 clap = { version = "4.5.7", features = ["derive"] }
 color-eyre = "0.6.3"
 serde = { version = "1.0.203", features = ["derive"] }
 serde_regex = "1.1.0"
 regex = "1.10.5"
-parse-display = "0.9.1"
+parse-display = "0.10.0"
 env_logger = "0.11.3"
 log = "0.4.21"
-tokio = { version = "1.38.0", features = ["macros", "parking_lot", "rt-multi-thread", "sync"] }
+tokio = { version = "1.38.0", features = ["macros", "parking_lot", "rt-multi-thread"] }

handlers/pa_volume/src/handler.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use parse_display::Display;
+use pulseaudio_volume_interface::PulseAudioVolumeInterface;
 use regex::Regex;
 use serde::Deserialize;
 use tokio::sync::broadcast;
@@ -11,7 +12,7 @@ use deckster_mode::shared::handler_communication::{
 use deckster_mode::shared::path::KnobPath;
 use deckster_mode::shared::state::KnobStyleByStateMap;
 use deckster_mode::{send_command, DecksterHandler};
-use pa_volume_interface::{PaEntityKind, PaEntityMetadata, PaEntityState, PaVolumeInterface};
+use pulseaudio_volume_interface::{EntityKind as PaEntityKind, EntityMetadata as PaEntityMetadata, EntityState as PaEntityState};
 
 #[derive(Debug, Clone, Deserialize)]
 pub struct KnobConfig {
@@ -113,7 +114,7 @@ fn state_matches(target: &Target, state: &PaEntityState) -> bool {
 
     static EMPTY_STRING: String = String::new();
 
-    return target.predicates.iter().all(|p| {
+    target.predicates.iter().all(|p| {
         let v = match (&p.property, state.metadata()) {
             (TargetPredicateProperty::InternalName, PaEntityMetadata::Sink { name, .. }) => Some(name),
             (TargetPredicateProperty::InternalName, PaEntityMetadata::Source { name, .. }) => Some(name),
@@ -141,10 +142,10 @@ fn state_matches(target: &Target, state: &PaEntityState) -> bool {
 
             false
         }
-    });
+    })
 }
 
-async fn manage_knob(path: KnobPath, config: KnobConfig, mut events: broadcast::Receiver<(KnobPath, KnobEvent)>, pa_volume_interface: Arc<PaVolumeInterface>) {
+async fn manage_knob(path: KnobPath, config: KnobConfig, mut events: broadcast::Receiver<(KnobPath, KnobEvent)>, pa_volume_interface: PulseAudioVolumeInterface) {
     let (initial_state, mut volume_states) = pa_volume_interface.subscribe_to_state();
 
     let mut entity_state: Option<Arc<PaEntityState>> = initial_state
@@ -273,13 +274,13 @@ pub struct Handler {
 impl Handler {
     pub fn new(data: InitialHandlerMessage<(), (), KnobConfig>) -> Result<Self, HandlerInitializationError> {
         let (events_sender, _) = broadcast::channel::<(KnobPath, KnobEvent)>(5);
-        let pa_volume_interface = Arc::new(PaVolumeInterface::spawn_thread("deckster handler".to_owned()));
+        let pa_volume_interface = PulseAudioVolumeInterface::new("deckster handler".to_owned());
 
         let runtime = tokio::runtime::Builder::new_multi_thread().worker_threads(1).build().unwrap();
 
         for (path, config) in data.knob_configs {
             let events_receiver = events_sender.subscribe();
-            runtime.spawn(manage_knob(path, config, events_receiver, Arc::clone(&pa_volume_interface)));
+            runtime.spawn(manage_knob(path, config, events_receiver, pa_volume_interface.clone()));
         }
 
         Ok(Handler { events_sender, runtime })

handlers/pa_volume/src/main.rs

@@ -1,10 +1,9 @@
 use clap::Parser;
 use color_eyre::Result;
-use pa_volume_interface::PaVolumeInterface;
 use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
-
+use pulseaudio_volume_interface::PulseAudioVolumeInterface;
 use crate::handler::Handler;
 
 mod handler;
@@ -33,8 +32,10 @@ fn main() -> Result<()> {
 }
 
 fn print_entities() {
-    let pa_volume_interface = Arc::new(PaVolumeInterface::spawn_thread("deckster handler".to_owned()));
-    sleep(Duration::from_secs(1)); // wait for the data to load
+    let pa_volume_interface = Arc::new(PulseAudioVolumeInterface::new("deckster handler".to_owned()));
+    // Wait for the data to load.
+    // Because of PulseAudio, there is no way to know when everything is ready, so we have to just hope a second is enough.
+    sleep(Duration::from_secs(1));
 
     let full_state = pa_volume_interface.current_state();
     let entities_by_id = full_state.entities_by_id();