[wip] volume

Cargo.toml

@@ -14,6 +14,7 @@ default = [
     "music+all",
     "sys_info",
     "tray",
+    "volume+all",
     "workspaces+all"
 ]
 
@@ -38,6 +39,10 @@ sys_info = ["sysinfo", "regex"]
 
 tray = ["stray"]
 
+volume = []
+"volume+all" = ["volume", "volume+pulse"]
+"volume+pulse" = ["libpulse-binding", "libpulse-glib-binding"]
+
 workspaces = ["futures-util"]
 "workspaces+all" = ["workspaces", "workspaces+sway", "workspaces+hyprland"]
 "workspaces+sway" = ["workspaces", "swayipc-async"]
@@ -88,6 +93,10 @@ sysinfo = { version = "0.27.0", optional = true }
 # tray
 stray = { version = "0.1.3", optional = true }
 
+# volume
+libpulse-binding = { version = "2.27.1", optional = true }
+libpulse-glib-binding = { version = "2.27.1", optional = true }
+
 # workspaces
 swayipc-async = { version = "2.0.1", optional = true }
 hyprland = { version = "0.3.0", optional = true }

src/clients/mod.rs

@@ -7,3 +7,5 @@ pub mod music;
 #[cfg(feature = "tray")]
 pub mod system_tray;
 pub mod wayland;
+#[cfg(feature = "volume")]
+pub mod volume;

src/clients/volume/mod.rs

@@ -0,0 +1,9 @@
+#[cfg(feature = "volume+pulse")]
+pub mod pulse_bak;
+// #[cfg(feature = "volume+pulse")]
+// pub mod pulse;
+
+trait VolumeClient {
+    // TODO: Write
+}
+

src/clients/volume/pulse/callbacks.rs

@@ -0,0 +1,345 @@
+use libpulse_binding::{
+	callbacks::ListResult,
+	context::{
+		introspect::{CardInfo, SinkInfo, SinkInputInfo, SourceInfo, SourceOutputInfo},
+		subscribe::{InterestMaskSet, Operation},
+	},
+	// def::{SinkState, SourceState},
+};
+use tracing::{debug, error, info};
+
+use super::{common::*, /*pa_interface::ACTIONS_SX*/};
+// use crate::{
+// 	entry::{CardProfile, Entry},
+// 	models::EntryUpdate,
+// 	ui::Rect,
+// };
+use color_eyre::Result;
+use crate::clients::volume::pulse::CardProfile;
+
+pub fn subscribe(
+	context: &Rc<RefCell<PAContext>>,
+	info_sx: mpsc::UnboundedSender<EntryIdentifier>,
+) -> Result<()> {
+	info!("[PAInterface] Registering pulseaudio callbacks");
+
+	context.borrow_mut().subscribe(
+		InterestMaskSet::SINK
+			| InterestMaskSet::SINK_INPUT
+			| InterestMaskSet::SOURCE
+			| InterestMaskSet::CARD
+			| InterestMaskSet::SOURCE_OUTPUT
+			| InterestMaskSet::CLIENT
+			| InterestMaskSet::SERVER,
+		|success: bool| {
+			assert!(success, "subscription failed");
+		},
+	);
+
+	context.borrow_mut().set_subscribe_callback(Some(Box::new(
+		move |facility, operation, index| {
+			if let Some(facility) = facility {
+				match facility {
+					Facility::Server | Facility::Client => {
+						error!("{:?} {:?}", facility, operation);
+						return;
+					}
+					_ => {}
+				};
+
+				let entry_type: EntryType = facility.into();
+				match operation {
+					Some(Operation::New) => {
+						info!("[PAInterface] New {:?}", entry_type);
+
+						info_sx
+							.send(EntryIdentifier::new(entry_type, index))
+							.unwrap();
+					}
+					Some(Operation::Changed) => {
+						info!("[PAInterface] {:?} changed", entry_type);
+						info_sx
+							.send(EntryIdentifier::new(entry_type, index))
+							.unwrap();
+					}
+					Some(Operation::Removed) => {
+						info!("[PAInterface] {:?} removed", entry_type);
+						// (*ACTIONS_SX)
+						// 	.get()
+						// 	.send(EntryUpdate::EntryRemoved(EntryIdentifier::new(
+						// 		entry_type, index,
+						// 	)))
+						// 	.unwrap();
+					}
+					_ => {}
+				};
+			};
+		},
+	)));
+
+	Ok(())
+}
+
+pub fn request_current_state(
+	context: Rc<RefCell<PAContext>>,
+	info_sxx: mpsc::UnboundedSender<EntryIdentifier>,
+) -> Result<()> {
+	info!("[PAInterface] Requesting starting state");
+
+	let introspector = context.borrow_mut().introspect();
+
+	let info_sx = info_sxx.clone();
+	introspector.get_sink_info_list(move |x: ListResult<&SinkInfo>| {
+		if let ListResult::Item(e) = x {
+			let _ = info_sx
+				.clone()
+				.send(EntryIdentifier::new(EntryType::Sink, e.index));
+		}
+	});
+
+	let info_sx = info_sxx.clone();
+	introspector.get_sink_input_info_list(move |x: ListResult<&SinkInputInfo>| {
+		if let ListResult::Item(e) = x {
+			let _ = info_sx.send(EntryIdentifier::new(EntryType::SinkInput, e.index));
+		}
+	});
+
+	let info_sx = info_sxx.clone();
+	introspector.get_source_info_list(move |x: ListResult<&SourceInfo>| {
+		if let ListResult::Item(e) = x {
+			let _ = info_sx.send(EntryIdentifier::new(EntryType::Source, e.index));
+		}
+	});
+
+	let info_sx = info_sxx.clone();
+	introspector.get_source_output_info_list(move |x: ListResult<&SourceOutputInfo>| {
+		if let ListResult::Item(e) = x {
+			let _ = info_sx.send(EntryIdentifier::new(EntryType::SourceOutput, e.index));
+		}
+	});
+
+	introspector.get_card_info_list(move |x: ListResult<&CardInfo>| {
+		if let ListResult::Item(e) = x {
+			let _ = info_sxx.send(EntryIdentifier::new(EntryType::Card, e.index));
+		}
+	});
+
+	Ok(())
+}
+
+pub fn request_info(
+	ident: EntryIdentifier,
+	context: &Rc<RefCell<PAContext>>,
+	info_sx: mpsc::UnboundedSender<EntryIdentifier>,
+) {
+	let introspector = context.borrow_mut().introspect();
+	debug!(
+		"[PAInterface] Requesting info for {:?} {}",
+		ident.entry_type, ident.index
+	);
+	match ident.entry_type {
+		EntryType::SinkInput => {
+			introspector.get_sink_input_info(ident.index, on_sink_input_info(&info_sx));
+		}
+		EntryType::Sink => {
+			introspector.get_sink_info_by_index(ident.index, on_sink_info(&info_sx));
+		}
+		EntryType::SourceOutput => {
+			introspector.get_source_output_info(ident.index, on_source_output_info(&info_sx));
+		}
+		EntryType::Source => {
+			introspector.get_source_info_by_index(ident.index, on_source_info(&info_sx));
+		}
+		EntryType::Card => {
+			introspector.get_card_info_by_index(ident.index, on_card_info);
+		}
+	};
+}
+pub fn on_card_info(res: ListResult<&CardInfo>) {
+	if let ListResult::Item(i) = res {
+		let n = match i
+			.proplist
+			.get_str(libpulse_binding::proplist::properties::DEVICE_DESCRIPTION)
+		{
+			Some(s) => s,
+			None => String::from(""),
+		};
+		let profiles: Vec<CardProfile> = i
+			.profiles
+			.iter()
+			.filter_map(|p| {
+				p.name.clone().map(|n| CardProfile {
+					// area: Rect::default(),
+					is_selected: false,
+					name: n.to_string(),
+					description: match &p.description {
+						Some(s) => s.to_string(),
+						None => n.to_string(),
+					},
+					#[cfg(any(feature = "pa_v13"))]
+					available: p.available,
+				})
+			})
+			.collect();
+
+		let selected_profile = match &i.active_profile {
+			Some(x) => {
+				if let Some(n) = &x.name {
+					profiles.iter().position(|p| p.name == *n)
+				} else {
+					None
+				}
+			}
+			None => None,
+		};
+
+		// let ident = EntryIdentifier::new(EntryType::Card, i.index);
+		// let entry = Entry::new_card_entry(i.index, n, profiles, selected_profile);
+
+		// (*ACTIONS_SX)
+		// 	.get()
+		// 	.send(EntryUpdate::EntryUpdate(ident, Box::new(entry)))
+		// 	.unwrap();
+	}
+}
+
+pub fn on_sink_info(
+	_sx: &mpsc::UnboundedSender<EntryIdentifier>,
+) -> impl Fn(ListResult<&SinkInfo>) {
+	|res: ListResult<&SinkInfo>| {
+		if let ListResult::Item(i) = res {
+			debug!("[PADataInterface] Update {} sink info", i.index);
+			let name = match &i.description {
+				Some(name) => name.to_string(),
+				None => String::new(),
+			};
+			// let ident = EntryIdentifier::new(EntryType::Sink, i.index);
+			// let entry = Entry::new_play_entry(
+			// 	EntryType::Sink,
+			// 	i.index,
+			// 	name,
+			// 	None,
+			// 	i.mute,
+			// 	i.volume,
+			// 	Some(i.monitor_source),
+			// 	None,
+			// 	i.state == SinkState::Suspended,
+			// );
+
+			// (*ACTIONS_SX)
+			// 	.get()
+			// 	.send(EntryUpdate::EntryUpdate(ident, Box::new(entry)))
+			// 	.unwrap();
+		}
+	}
+}
+
+pub fn on_sink_input_info(
+	sx: &mpsc::UnboundedSender<EntryIdentifier>,
+) -> impl Fn(ListResult<&SinkInputInfo>) {
+	let info_sx = sx.clone();
+	move |res: ListResult<&SinkInputInfo>| {
+		if let ListResult::Item(i) = res {
+			debug!("[PADataInterface] Update {} sink input info", i.index);
+			let n = match i
+				.proplist
+				.get_str(libpulse_binding::proplist::properties::APPLICATION_NAME)
+			{
+				Some(s) => s,
+				None => match &i.name {
+					Some(s) => s.to_string(),
+					None => String::from(""),
+				},
+			};
+			// let ident = EntryIdentifier::new(EntryType::SinkInput, i.index);
+			//
+			// let entry = Entry::new_play_entry(
+			// 	EntryType::SinkInput,
+			// 	i.index,
+			// 	n,
+			// 	Some(i.sink),
+			// 	i.mute,
+			// 	i.volume,
+			// 	None,
+			// 	Some(i.sink),
+			// 	false,
+			// );
+
+			// (*ACTIONS_SX)
+			// 	.get()
+			// 	.send(EntryUpdate::EntryUpdate(ident, Box::new(entry)))
+			// 	.unwrap();
+			let _ = info_sx.send(EntryIdentifier::new(EntryType::Sink, i.sink));
+		}
+	}
+}
+
+pub fn on_source_info(
+	_sx: &mpsc::UnboundedSender<EntryIdentifier>,
+) -> impl Fn(ListResult<&SourceInfo>) {
+	move |res: ListResult<&SourceInfo>| {
+		if let ListResult::Item(i) = res {
+			debug!("[PADataInterface] Update {} source info", i.index);
+			let name = match &i.description {
+				Some(name) => name.to_string(),
+				None => String::new(),
+			};
+			// let ident = EntryIdentifier::new(EntryType::Source, i.index);
+			// let entry = Entry::new_play_entry(
+			// 	EntryType::Source,
+			// 	i.index,
+			// 	name,
+			// 	None,
+			// 	i.mute,
+			// 	i.volume,
+			// 	Some(i.index),
+			// 	None,
+			// 	i.state == SourceState::Suspended,
+			// );
+
+			// (*ACTIONS_SX)
+			// 	.get()
+			// 	.send(EntryUpdate::EntryUpdate(ident, Box::new(entry)))
+			// 	.unwrap();
+		}
+	}
+}
+
+pub fn on_source_output_info(
+	sx: &mpsc::UnboundedSender<EntryIdentifier>,
+) -> impl Fn(ListResult<&SourceOutputInfo>) {
+	let info_sx = sx.clone();
+	move |res: ListResult<&SourceOutputInfo>| {
+		if let ListResult::Item(i) = res {
+			debug!("[PADataInterface] Update {} source output info", i.index);
+			let n = match i
+				.proplist
+				.get_str(libpulse_binding::proplist::properties::APPLICATION_NAME)
+			{
+				Some(s) => s,
+				None => String::from(""),
+			};
+			if n == "RsMixerContext" {
+				return;
+			}
+			// let ident = EntryIdentifier::new(EntryType::SourceOutput, i.index);
+			// let entry = Entry::new_play_entry(
+			// 	EntryType::SourceOutput,
+			// 	i.index,
+			// 	n,
+			// 	Some(i.source),
+			// 	i.mute,
+			// 	i.volume,
+			// 	Some(i.source),
+			// 	None,
+			// 	false,
+			// );
+
+			// (*ACTIONS_SX)
+			// 	.get()
+			// 	.send(EntryUpdate::EntryUpdate(ident, Box::new(entry)))
+			// 	.unwrap();
+			let _ = info_sx.send(EntryIdentifier::new(EntryType::Source, i.index));
+		}
+	}
+}

src/clients/volume/pulse/common.rs

@@ -0,0 +1,30 @@
+pub use std::{cell::RefCell, collections::HashMap, rc::Rc};
+
+pub use libpulse_binding::{
+	context::{subscribe::Facility, Context as PAContext},
+	mainloop::{api::Mainloop as MainloopTrait, threaded::Mainloop},
+	stream::Stream,
+};
+pub use tokio::sync::mpsc;
+
+pub use super::{monitor::Monitors, PAInternal, SPEC};
+// pub use crate::{
+// 	entry::{EntryIdentifier, EntryType},
+// 	models::{EntryUpdate, PulseAudioAction},
+// 	prelude::*,
+// };
+
+pub static LOGGING_MODULE: &str = "PAInterface";
+
+impl From<Facility> for EntryType {
+	fn from(fac: Facility) -> Self {
+		match fac {
+			Facility::Sink => EntryType::Sink,
+			Facility::Source => EntryType::Source,
+			Facility::SinkInput => EntryType::SinkInput,
+			Facility::SourceOutput => EntryType::SourceOutput,
+			Facility::Card => EntryType::Card,
+			_ => EntryType::Sink,
+		}
+	}
+}

src/clients/volume/pulse/errors.rs

@@ -0,0 +1,17 @@
+// use thiserror::Error;
+//
+// use super::PAInternal;
+//
+// #[derive(Debug, Error)]
+// pub enum PAError {
+// 	#[error("cannot create pulseaudio mainloop")]
+// 	MainloopCreateError,
+// 	#[error("cannot connect pulseaudio mainloop")]
+// 	MainloopConnectError,
+// 	#[error("cannot create pulseaudio stream")]
+// 	StreamCreateError,
+// 	#[error("internal channel send error")]
+// 	ChannelError(#[from] cb_channel::SendError<PAInternal>),
+// 	#[error("pulseaudio disconnected")]
+// 	PulseAudioDisconnected,
+// }

src/clients/volume/pulse/mod.rs

@@ -0,0 +1,36 @@
+mod callbacks;
+pub mod common;
+mod errors;
+mod monitor;
+mod pa_actions;
+mod pa_interface;
+
+use common::*;
+use lazy_static::lazy_static;
+pub use pa_interface::start;
+
+#[derive(Debug)]
+pub enum PAInternal {
+	Tick,
+	Command(Box<PulseAudioAction>),
+	AskInfo(EntryIdentifier),
+}
+
+lazy_static! {
+	pub static ref SPEC: libpulse_binding::sample::Spec = libpulse_binding::sample::Spec {
+		format: libpulse_binding::sample::Format::FLOAT32NE,
+		channels: 1,
+		rate: 1024,
+	};
+}
+
+#[derive(PartialEq, Clone, Debug)]
+pub struct CardProfile {
+	pub name: String,
+	pub description: String,
+	#[cfg(any(feature = "pa_v13"))]
+	pub available: bool,
+	// pub area: Rect,
+	pub is_selected: bool,
+}
+impl Eq for CardProfile {}

src/clients/volume/pulse/monitor.rs

@@ -0,0 +1,278 @@
+use std::convert::TryInto;
+
+use libpulse_binding::stream::PeekResult;
+use tracing::{debug, error, info, warn};
+
+use super::{common::*, /*pa_interface::ACTIONS_SX*/};
+// use crate::VARIABLES;
+use color_eyre::{Report, Result};
+
+pub struct Monitor {
+    stream: Rc<RefCell<Stream>>,
+    exit_sender: mpsc::UnboundedSender<u32>,
+}
+
+pub struct Monitors {
+    monitors: HashMap<EntryIdentifier, Monitor>,
+    errors: HashMap<EntryIdentifier, usize>,
+}
+
+impl Default for Monitors {
+    fn default() -> Self {
+        Self {
+            monitors: HashMap::new(),
+            errors: HashMap::new(),
+        }
+    }
+}
+
+impl Monitors {
+    pub fn filter(
+        &mut self,
+        mainloop: &Rc<RefCell<Mainloop>>,
+        context: &Rc<RefCell<PAContext>>,
+        targets: &HashMap<EntryIdentifier, Option<u32>>,
+    ) {
+        // remove failed streams
+        // then send exit signal if stream is unwanted
+        self.monitors.retain(|ident, monitor| {
+            match monitor.stream.borrow_mut().get_state() {
+                libpulse_binding::stream::State::Terminated
+                | libpulse_binding::stream::State::Failed => {
+                    info!(
+                        "[PAInterface] Disconnecting {} sink input monitor (failed state)",
+                        ident.index
+                    );
+                    return false;
+                }
+                _ => {}
+            };
+
+            if targets.get(ident) == None {
+                let _ = monitor.exit_sender.send(0);
+            }
+
+            true
+        });
+
+        targets.iter().for_each(|(ident, monitor_src)| {
+            if self.monitors.get(ident).is_none() {
+                self.create_monitor(mainloop, context, *ident, *monitor_src);
+            }
+        });
+    }
+
+    fn create_monitor(
+        &mut self,
+        mainloop: &Rc<RefCell<Mainloop>>,
+        context: &Rc<RefCell<PAContext>>,
+        ident: EntryIdentifier,
+        monitor_src: Option<u32>,
+    ) {
+        if let Some(count) = self.errors.get(&ident) {
+            if *count >= 5 {
+                self.errors.remove(&ident);
+                // (*ACTIONS_SX)
+                //     .get()
+                //     .send(EntryUpdate::EntryRemoved(ident))
+                //     .unwrap();
+            }
+        }
+        if self.monitors.contains_key(&ident) {
+            return;
+        }
+        let (sx, rx) = mpsc::unbounded_channel();
+        if let Ok(stream) = create(
+            &mainloop,
+            &context,
+            &libpulse_binding::sample::Spec {
+                format: libpulse_binding::sample::Format::FLOAT32NE,
+                channels: 1,
+                rate: /*(*VARIABLES).get().pa_rate*/ 20,
+            },
+            ident,
+            monitor_src,
+            rx,
+        ) {
+            self.monitors.insert(
+                ident,
+                Monitor {
+                    stream,
+                    exit_sender: sx,
+                },
+            );
+            self.errors.remove(&ident);
+        } else {
+            self.error(&ident);
+        }
+    }
+
+    fn error(&mut self, ident: &EntryIdentifier) {
+        let count = match self.errors.get(&ident) {
+            Some(x) => *x,
+            None => 0,
+        };
+        self.errors.insert(*ident, count + 1);
+    }
+}
+
+fn slice_to_4_bytes(slice: &[u8]) -> [u8; 4] {
+    slice.try_into().expect("slice with incorrect length")
+}
+
+fn create(
+    p_mainloop: &Rc<RefCell<Mainloop>>,
+    p_context: &Rc<RefCell<PAContext>>,
+    p_spec: &libpulse_binding::sample::Spec,
+    ident: EntryIdentifier,
+    source_index: Option<u32>,
+    mut close_rx: mpsc::UnboundedReceiver<u32>,
+) -> Result<Rc<RefCell<Stream>>> {
+    info!("[PADataInterface] Attempting to create new monitor stream");
+
+    let stream_index = if ident.entry_type == EntryType::SinkInput {
+        Some(ident.index)
+    } else {
+        None
+    };
+
+    let stream = Rc::new(RefCell::new(
+        match Stream::new(&mut p_context.borrow_mut(), "RsMixer monitor", p_spec, None) {
+            Some(stream) => stream,
+            None => return Err(Report::msg("Error creating stream for monitoring volume")),
+        },
+    ));
+
+    // Stream state change callback
+    {
+        debug!("[PADataInterface] Registering stream state change callback");
+        let ml_ref = Rc::clone(&p_mainloop);
+        let stream_ref = Rc::downgrade(&stream);
+        stream
+            .borrow_mut()
+            .set_state_callback(Some(Box::new(move || {
+                let state = unsafe { (*(*stream_ref.as_ptr()).as_ptr()).get_state() };
+                match state {
+                    libpulse_binding::stream::State::Ready
+                    | libpulse_binding::stream::State::Failed
+                    | libpulse_binding::stream::State::Terminated => {
+                        unsafe { (*ml_ref.as_ptr()).signal(false) };
+                    }
+                    _ => {}
+                }
+            })));
+    }
+
+    // for sink inputs we want to set monitor stream to sink
+    if let Some(index) = stream_index {
+        stream.borrow_mut().set_monitor_stream(index).unwrap();
+    }
+
+    let x;
+    let mut s = None;
+    if let Some(i) = source_index {
+        x = i.to_string();
+        s = Some(x.as_str());
+    }
+
+    debug!("[PADataInterface] Connecting stream");
+    match stream.borrow_mut().connect_record(
+        s,
+        Some(&libpulse_binding::def::BufferAttr {
+            maxlength: std::u32::MAX,
+            tlength: std::u32::MAX,
+            prebuf: std::u32::MAX,
+            minreq: 0,
+            fragsize: /*(*VARIABLES).get().pa_frag_size*/ 48,
+        }),
+        libpulse_binding::stream::FlagSet::PEAK_DETECT
+            | libpulse_binding::stream::FlagSet::ADJUST_LATENCY,
+    ) {
+        Ok(_) => {}
+        Err(err) => {
+            return Err(Report::new(err).wrap_err("while connecting stream for monitoring volume"));
+        }
+    };
+
+    debug!("[PADataInterface] Waiting for stream to be ready");
+    loop {
+        match stream.borrow_mut().get_state() {
+            libpulse_binding::stream::State::Ready => {
+                break;
+            }
+            libpulse_binding::stream::State::Failed
+            | libpulse_binding::stream::State::Terminated => {
+                error!("[PADataInterface] Stream state failed/terminated");
+                return Err(Report::msg("Stream terminated"))
+            }
+            _ => {
+                p_mainloop.borrow_mut().wait();
+            }
+        }
+    }
+
+    stream.borrow_mut().set_state_callback(None);
+
+    {
+        info!("[PADataInterface] Registering stream read callback");
+        let ml_ref = Rc::clone(&p_mainloop);
+        let stream_ref = Rc::downgrade(&stream);
+        stream.borrow_mut().set_read_callback(Some(Box::new(move |_size: usize| {
+            let remove_failed = || {
+                error!("[PADataInterface] Monitor failed or terminated");
+            };
+            let disconnect_stream = || {
+                warn!("[PADataInterface] {:?} Monitor existed while the sink (input)/source (output) was already gone", ident);
+                unsafe {
+                    (*(*stream_ref.as_ptr()).as_ptr()).disconnect().unwrap();
+                    (*ml_ref.as_ptr()).signal(false);
+                };
+            };
+
+            if close_rx.try_recv().is_ok() {
+                disconnect_stream();
+                return;
+            }
+
+            match unsafe {(*(*stream_ref.as_ptr()).as_ptr()).get_state() }{
+                libpulse_binding::stream::State::Failed => {
+                    remove_failed();
+                },
+                libpulse_binding::stream::State::Terminated => {
+                    remove_failed();
+                },
+                libpulse_binding::stream::State::Ready => {
+                    match unsafe{ (*(*stream_ref.as_ptr()).as_ptr()).peek() } {
+                        Ok(res) => match res {
+                            PeekResult::Data(data) => {
+                                let count = data.len() / 4;
+                                let mut peak = 0.0;
+                                for c in 0..count {
+                                    let data_slice = slice_to_4_bytes(&data[c * 4 .. (c + 1) * 4]);
+                                    peak += f32::from_ne_bytes(data_slice).abs();
+                                }
+                                peak = peak / count as f32;
+
+                                // if (*ACTIONS_SX).get().send(EntryUpdate::PeakVolumeUpdate(ident, peak)).is_err() {
+                                //     disconnect_stream();
+                                // }
+
+                                unsafe { (*(*stream_ref.as_ptr()).as_ptr()).discard().unwrap(); };
+                            },
+                            PeekResult::Hole(_) => {
+                                unsafe { (*(*stream_ref.as_ptr()).as_ptr()).discard().unwrap(); };
+                            },
+                            _ => {},
+                        },
+                        Err(_) => {
+                            remove_failed();
+                        },
+                    }
+                },
+                _ => {},
+            };
+        })));
+    }
+
+    Ok(stream)
+}

src/clients/volume/pulse/pa_actions.rs

@@ -0,0 +1,148 @@
+use super::{callbacks, common::*};
+
+pub fn handle_command(
+	cmd: PulseAudioAction,
+	context: &Rc<RefCell<PAContext>>,
+	info_sx: &mpsc::UnboundedSender<EntryIdentifier>,
+) -> Option<()> {
+	match cmd {
+		PulseAudioAction::RequestPulseAudioState => {
+			callbacks::request_current_state(Rc::clone(&context), info_sx.clone()).unwrap();
+		}
+		PulseAudioAction::MuteEntry(ident, mute) => {
+			set_mute(ident, mute, &context);
+		}
+		PulseAudioAction::MoveEntryToParent(ident, parent) => {
+			move_entry_to_parent(ident, parent, &context, info_sx.clone());
+		}
+		PulseAudioAction::ChangeCardProfile(ident, profile) => {
+			change_card_profile(ident, profile, &context);
+		}
+		PulseAudioAction::SetVolume(ident, vol) => {
+			set_volume(ident, vol, &context);
+		}
+		PulseAudioAction::SetSuspend(ident, suspend) => {
+			set_suspend(ident, suspend, &context);
+		}
+		PulseAudioAction::KillEntry(ident) => {
+			kill_entry(ident, &context);
+		}
+		PulseAudioAction::Shutdown => {
+			//@TODO disconnect monitors
+			return None;
+		}
+		_ => {}
+	};
+	Some(())
+}
+
+fn set_volume(
+	ident: EntryIdentifier,
+	vol: libpulse_binding::volume::ChannelVolumes,
+	context: &Rc<RefCell<PAContext>>,
+) {
+	let mut introspector = context.borrow_mut().introspect();
+	match ident.entry_type {
+		EntryType::Sink => {
+			introspector.set_sink_volume_by_index(ident.index, &vol, None);
+		}
+		EntryType::SinkInput => {
+			introspector.set_sink_input_volume(ident.index, &vol, None);
+		}
+		EntryType::Source => {
+			introspector.set_source_volume_by_index(ident.index, &vol, None);
+		}
+		EntryType::SourceOutput => {
+			introspector.set_source_output_volume(ident.index, &vol, None);
+		}
+		_ => {}
+	};
+}
+
+fn change_card_profile(ident: EntryIdentifier, profile: String, context: &Rc<RefCell<PAContext>>) {
+	if ident.entry_type != EntryType::Card {
+		return;
+	}
+	context
+		.borrow_mut()
+		.introspect()
+		.set_card_profile_by_index(ident.index, &profile[..], None);
+}
+
+fn move_entry_to_parent(
+	ident: EntryIdentifier,
+	parent: EntryIdentifier,
+	context: &Rc<RefCell<PAContext>>,
+	info_sx: mpsc::UnboundedSender<EntryIdentifier>,
+) {
+	let mut introspector = context.borrow_mut().introspect();
+
+	match ident.entry_type {
+		EntryType::SinkInput => {
+			introspector.move_sink_input_by_index(
+				ident.index,
+				parent.index,
+				Some(Box::new(move |_| {
+					info_sx.send(parent).unwrap();
+					info_sx.send(ident).unwrap();
+				})),
+			);
+		}
+		EntryType::SourceOutput => {
+			introspector.move_source_output_by_index(
+				ident.index,
+				parent.index,
+				Some(Box::new(move |_| {
+					info_sx.send(parent).unwrap();
+					info_sx.send(ident).unwrap();
+				})),
+			);
+		}
+		_ => {}
+	};
+}
+
+fn set_suspend(ident: EntryIdentifier, suspend: bool, context: &Rc<RefCell<PAContext>>) {
+	let mut introspector = context.borrow_mut().introspect();
+	match ident.entry_type {
+		EntryType::Sink => {
+			introspector.suspend_sink_by_index(ident.index, suspend, None);
+		}
+		EntryType::Source => {
+			introspector.suspend_source_by_index(ident.index, suspend, None);
+		}
+		_ => {}
+	};
+}
+
+fn kill_entry(ident: EntryIdentifier, context: &Rc<RefCell<PAContext>>) {
+	let mut introspector = context.borrow_mut().introspect();
+	match ident.entry_type {
+		EntryType::SinkInput => {
+			introspector.kill_sink_input(ident.index, |_| {});
+		}
+		EntryType::SourceOutput => {
+			introspector.kill_source_output(ident.index, |_| {});
+		}
+		_ => {}
+	};
+}
+
+fn set_mute(ident: EntryIdentifier, mute: bool, context: &Rc<RefCell<PAContext>>) {
+	let mut introspector = context.borrow_mut().introspect();
+	match ident.entry_type {
+		EntryType::Sink => {
+			introspector.set_sink_mute_by_index(ident.index, mute, None);
+		}
+		EntryType::SinkInput => {
+			introspector.set_sink_input_mute(ident.index, mute, None);
+		}
+		EntryType::Source => {
+			introspector.set_source_mute_by_index(ident.index, mute, None);
+		}
+		EntryType::SourceOutput => {
+			introspector.set_source_output_mute(ident.index, mute, None);
+		}
+		_ => {}
+	};
+}

src/clients/volume/pulse/pa_interface.rs

@@ -0,0 +1,162 @@
+use std::ops::Deref;
+
+// use lazy_static::lazy_static;
+use libpulse_binding::proplist::Proplist;
+use tracing::{debug, error, info};
+// use state::Storage;
+use color_eyre::{Report, Result};
+
+use super::{callbacks, common::*, pa_actions};
+
+// lazy_static! {
+// 	pub static ref ACTIONS_SX: Storage<mpsc::UnboundedSender<EntryUpdate>> = Storage::new();
+// }
+
+pub async fn start(
+	mut internal_rx: mpsc::Receiver<PAInternal>,
+	info_sx: mpsc::UnboundedSender<EntryIdentifier>,
+	actions_sx: mpsc::UnboundedSender<EntryUpdate>,
+) -> Result<()> {
+	// (*ACTIONS_SX).set(actions_sx);
+
+	// Create new mainloop and context
+	let mut proplist = Proplist::new().unwrap();
+	proplist
+		.set_str(libpulse_binding::proplist::properties::APPLICATION_NAME, "RsMixer")
+		.unwrap();
+
+	debug!("[PAInterface] Creating new mainloop");
+	let mainloop = Rc::new(RefCell::new(match Mainloop::new() {
+		Some(ml) => ml,
+		None => {
+			error!("[PAInterface] Error while creating new mainloop");
+			return Err(Report::msg("Error while creating new mainloop"));
+		}
+	}));
+
+	debug!("[PAInterface] Creating new context");
+	let context = Rc::new(RefCell::new(
+		match PAContext::new_with_proplist(
+			mainloop.borrow_mut().deref().deref(),
+			"RsMixerContext",
+			&proplist,
+		) {
+			Some(ctx) => ctx,
+			None => {
+				error!("[PAInterface] Error while creating new context");
+				return Err(Report::msg("Error while creating new context"));
+			}
+		},
+	));
+
+	// PAContext state change callback
+	{
+		debug!("[PAInterface] Registering state change callback");
+		let ml_ref = Rc::clone(&mainloop);
+		let context_ref = Rc::clone(&context);
+		context
+			.borrow_mut()
+			.set_state_callback(Some(Box::new(move || {
+				let state = unsafe { (*context_ref.as_ptr()).get_state() };
+				if matches!(
+					state,
+					libpulse_binding::context::State::Ready
+						| libpulse_binding::context::State::Failed
+						| libpulse_binding::context::State::Terminated
+				) {
+					unsafe { (*ml_ref.as_ptr()).signal(false) };
+				}
+			})));
+	}
+
+	// Try to connect to pulseaudio
+	debug!("[PAInterface] Connecting context");
+
+	if context
+		.borrow_mut()
+		.connect(None, libpulse_binding::context::FlagSet::NOFLAGS, None)
+		.is_err()
+	{
+		error!("[PAInterface] Error while connecting context");
+		return Err(Report::msg("Error while connecting context"));
+	}
+
+	info!("[PAInterface] Starting mainloop");
+
+	// start mainloop
+	mainloop.borrow_mut().lock();
+
+	if let Err(err) = mainloop.borrow_mut().start() {
+		return Err(Report::new(err));
+	}
+
+	debug!("[PAInterface] Waiting for context to be ready...");
+	// wait for context to be ready
+	loop {
+		match context.borrow_mut().get_state() {
+			libpulse_binding::context::State::Ready => {
+				break;
+			}
+			libpulse_binding::context::State::Failed | libpulse_binding::context::State::Terminated => {
+				mainloop.borrow_mut().unlock();
+				mainloop.borrow_mut().stop();
+				error!("[PAInterface] Connection failed or context terminated");
+				return Err(Report::msg("Connection failed or context terminated"));
+			}
+			_ => {
+				mainloop.borrow_mut().wait();
+			}
+		}
+	}
+	debug!("[PAInterface] PAContext ready");
+
+	context.borrow_mut().set_state_callback(None);
+
+	callbacks::subscribe(&context, info_sx.clone())?;
+	callbacks::request_current_state(context.clone(), info_sx.clone())?;
+
+	mainloop.borrow_mut().unlock();
+
+	debug!("[PAInterface] Actually starting our mainloop");
+
+	let mut monitors = Monitors::default();
+	let mut last_targets = HashMap::new();
+
+	while let Some(msg) = internal_rx.recv().await {
+		mainloop.borrow_mut().lock();
+
+		match context.borrow_mut().get_state() {
+			libpulse_binding::context::State::Ready => {}
+			_ => {
+				mainloop.borrow_mut().unlock();
+				return Err(Report::msg("Disconnected while working"))
+			}
+		}
+
+		match msg {
+			PAInternal::AskInfo(ident) => {
+				callbacks::request_info(ident, &context, info_sx.clone());
+			}
+			PAInternal::Tick => {
+				// remove failed monitors
+				monitors.filter(&mainloop, &context, &last_targets);
+			}
+			PAInternal::Command(cmd) => {
+				let cmd = cmd.deref();
+				if pa_actions::handle_command(cmd.clone(), &context, &info_sx).is_none() {
+					monitors.filter(&mainloop, &context, &HashMap::new());
+					mainloop.borrow_mut().unlock();
+					break;
+				}
+
+				if let PulseAudioAction::CreateMonitors(mons) = cmd.clone() {
+					last_targets = mons;
+					monitors.filter(&mainloop, &context, &last_targets);
+				}
+			}
+		};
+		mainloop.borrow_mut().unlock();
+	}
+
+	Ok(())
+}

src/clients/volume/pulse_bak.rs

@@ -0,0 +1,90 @@
+use crate::clients::volume::VolumeClient;
+use libpulse_binding::context::State;
+use libpulse_binding::{
+    context::{Context, FlagSet},
+    mainloop::threaded::Mainloop,
+    proplist::{properties, Proplist},
+};
+use std::cell::RefCell;
+use std::ops::Deref;
+use std::rc::Rc;
+use tracing::{debug, error};
+
+pub fn test() {
+    let mut prop_list = Proplist::new().unwrap();
+    prop_list
+        .set_str(properties::APPLICATION_NAME, "ironbar")
+        .unwrap();
+
+    let mainloop = Rc::new(RefCell::new(Mainloop::new().unwrap()));
+
+    let context = Rc::new(RefCell::new(
+        Context::new_with_proplist(mainloop.borrow().deref(), "ironbar_context", &prop_list)
+            .unwrap(),
+    ));
+
+    // PAContext state change callback
+    {
+        debug!("[PAInterface] Registering state change callback");
+        let ml_ref = Rc::clone(&mainloop);
+        let context_ref = Rc::clone(&context);
+        context
+            .borrow_mut()
+            .set_state_callback(Some(Box::new(move || {
+                let state = unsafe { (*context_ref.as_ptr()).get_state() };
+                if matches!(state, State::Ready | State::Failed | State::Terminated) {
+                    unsafe { (*ml_ref.as_ptr()).signal(false) };
+                }
+            })));
+    }
+
+    if let Err(err) = context.borrow_mut().connect(None, FlagSet::NOFLAGS, None) {
+        error!("{err:?}");
+    }
+
+    println!("{:?}", context.borrow().get_server());
+
+    mainloop.borrow_mut().lock();
+    if let Err(err) = mainloop.borrow_mut().start() {
+        error!("{err:?}");
+    }
+
+    debug!("[PAInterface] Waiting for context to be ready...");
+    println!("[PAInterface] Waiting for context to be ready...");
+    // wait for context to be ready
+    loop {
+        match context.borrow().get_state() {
+            State::Ready => {
+                break;
+            }
+            State::Failed | State::Terminated => {
+                mainloop.borrow_mut().unlock();
+                mainloop.borrow_mut().stop();
+                error!("[PAInterface] Connection failed or context terminated");
+            }
+            _ => {
+                mainloop.borrow_mut().wait();
+            }
+        }
+    }
+    debug!("[PAInterface] PAContext ready");
+    println!("[PAInterface] PAContext ready");
+
+    context.borrow_mut().set_state_callback(None);
+
+    println!("jfgjfgg");
+
+    let introspector = context.borrow().introspect();
+
+    println!("jfgjfgg2");
+
+    introspector.get_sink_info_list(|result| {
+        println!("boo: {result:?}");
+    });
+
+    println!("fjgjfgf??");
+}
+
+struct PulseVolumeClient {}
+
+impl VolumeClient for PulseVolumeClient {}

src/main.rs

@@ -44,6 +44,8 @@ async fn main() -> Result<()> {
     info!("Ironbar version {}", VERSION);
     info!("Starting application");
 
+    clients::volume::pulse_bak::test();
+
     let wayland_client = wayland::get_client().await;
 
     let app = Application::builder().application_id(GTK_APP_ID).build();