WIP using libpulse_binding directly in audio applet

Cargo.lock

@@ -315,13 +315,14 @@ version = "0.1.0"
 dependencies = [
  "async-io",
  "freedesktop-desktop-entry",
+ "futures",
  "futures-util",
  "gtk4",
  "libcosmic-widgets",
  "libpulse-binding",
+ "libpulse-glib-binding",
  "mpris2-zbus",
  "once_cell",
- "pulsectl-rs",
  "relm4-macros 0.4.4",
  "tokio",
  "tracker",
@@ -1541,6 +1542,29 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "libpulse-glib-binding"
+version = "2.25.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df0e7a964c9f7e95d4f073affc19adfda009fa0d55e8831dbb66c78be1d0e6e5"
+dependencies = [
+ "glib",
+ "glib-sys",
+ "libpulse-binding",
+ "libpulse-mainloop-glib-sys",
+]
+
+[[package]]
+name = "libpulse-mainloop-glib-sys"
+version = "1.19.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36f61c4064926cc77ea14bb206a21ce1d5a06e175e5c0ce078804bb6c4527b28"
+dependencies = [
+ "glib-sys",
+ "libpulse-sys",
+ "pkg-config",
+]
+
 [[package]]
 name = "libpulse-sys"
 version = "1.19.3"
@@ -2046,15 +2070,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "pulsectl-rs"
-version = "0.3.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "06a988bceed1981b2c5fc4a3da0e4e073fdaff8e6bd022b089f54bc573dc3cfc"
-dependencies = [
- "libpulse-binding",
-]
-
 [[package]]
 name = "quick-error"
 version = "1.2.3"

applets/cosmic-applet-audio/Cargo.toml

@@ -5,10 +5,11 @@ edition = "2021"
 license = "GPL-3.0-or-later"
 
 [dependencies]
+futures = "0.3.21"
 futures-util = "0.3.21"
 libcosmic-widgets = { git = "https://github.com/pop-os/libcosmic" }
 libpulse-binding = "2.26.0"
-pulsectl-rs = "0.3.2"
+libpulse-glib-binding = "2.25.0"
 tracker = "0.1.1"
 freedesktop-desktop-entry = "0.5.0"
 mpris2-zbus = { git = "https://github.com/pop-os/mpris2-zbus" }

applets/cosmic-applet-audio/src/input.rs

@@ -1,19 +1,21 @@
 use gtk4::{prelude::*, Button, Label, ListBox};
 use libcosmic_widgets::{relm4::RelmContainerExt, LabeledItem};
-use pulsectl::controllers::{types::DeviceInfo, DeviceControl, SourceController};
 use std::rc::Rc;
 
-fn get_inputs() -> Vec<DeviceInfo> {
+use crate::pa::{Source, PA};
-    SourceController::create()
+
-        .expect("failed to create input controller")
+pub async fn get_inputs(pa: &PA) -> Vec<Source> {
-        .list_devices()
+    // XXX handle error
+    pa.get_source_info_list()
+        .await
         .expect("failed to list input devices")
 }
 
-pub fn refresh_default_input(label: &Label) -> DeviceInfo {
+pub async fn refresh_default_input(pa: &PA, label: &Label) -> Source {
-    let default_input = SourceController::create()
+    // XXX handle error
-        .expect("failed to create input controller")
+    let default_input = pa
-        .get_default_device()
+        .get_default_source()
+        .await
         .expect("failed to get default input");
     label.set_text(match &default_input.description {
         Some(name) => name.as_str(),
@@ -22,12 +24,11 @@ pub fn refresh_default_input(label: &Label) -> DeviceInfo {
     default_input
 }
 
-pub fn refresh_input_widgets(inputs: &ListBox) {
+pub async fn refresh_input_widgets(pa: &PA, inputs: &ListBox) {
     while let Some(row) = inputs.row_at_index(0) {
         inputs.remove(&row);
     }
-    for input in get_inputs() {
+    for input in get_inputs(pa).await {
-        let input = Rc::new(input.clone());
         let name = match &input.name {
             Some(name) => name.to_owned(),
             None => continue, // Why doesn't this have a name? Whatever, it's invalid.
@@ -40,10 +41,13 @@ pub fn refresh_input_widgets(inputs: &ListBox) {
                 set_child: set_current_input_device = &Button {
                     set_label: "Switch",
                     connect_clicked: move |_| {
+                        // XXX Need mutable borrow? Is this a problem for async?
+                        /*
                         SourceController::create()
                             .expect("failed to create input controller")
                             .set_default_device(&name)
                             .expect("failed to set default device");
+                        */
                     }
                 }
             }

applets/cosmic-applet-audio/src/main.rs

@@ -8,9 +8,11 @@ mod input;
 mod now_playing;
 mod output;
 mod pa;
+use pa::PA;
 mod task;
 mod volume;
 
+use futures::{channel::mpsc, stream::StreamExt};
 use gtk4::{
     gio::ApplicationFlags,
     glib::{self, clone, MainContext, PRIORITY_DEFAULT},
@@ -24,7 +26,7 @@ use libpulse_binding::{
 };
 use mpris2_zbus::metadata::Metadata;
 use once_cell::sync::Lazy;
-use pulsectl::Handler;
+use std::rc::Rc;
 use tokio::runtime::Runtime;
 
 static RT: Lazy<Runtime> = Lazy::new(|| Runtime::new().expect("failed to build tokio runtime"));
@@ -39,35 +41,29 @@ fn main() {
 }
 
 fn app(application: &Application) {
-    let handler =
+    // XXX handle no pulseaudio daemon?
-        Handler::connect("com.system76.cosmic.applets.audio").expect("failed to connect to pulse");
+    let mut pa = PA::new().unwrap();
-    task::spawn_local(clone!(@strong handler.mainloop as main_loop => async move {
+    let (refresh_output_tx, mut refresh_output_rx) = mpsc::unbounded();
-        pa::drive_main_loop(main_loop).await
+    let (refresh_input_tx, mut refresh_input_rx) = mpsc::unbounded();
-    }));
+    let (now_playing_tx, mut now_playing_rx) = mpsc::unbounded::<Vec<Metadata>>();
-    let (refresh_output_tx, refresh_output_rx) = MainContext::channel::<()>(PRIORITY_DEFAULT);
+    pa.context
-    let (refresh_input_tx, refresh_input_rx) = MainContext::channel::<()>(PRIORITY_DEFAULT);
-    let (now_playing_tx, now_playing_rx) = MainContext::channel::<Vec<Metadata>>(PRIORITY_DEFAULT);
-    handler
-        .context
-        .borrow_mut()
         .set_subscribe_callback(Some(Box::new(clone!(@strong refresh_output_tx, @strong refresh_input_tx => move |facility, operation, _idx| {
             if !matches!(operation, Some(Operation::Changed)) {
                 return;
             }
             match facility {
                 Some(Facility::Sink) => {
-                    refresh_output_tx.send(()).expect("failed to send output refresh message");
+                    refresh_output_tx.unbounded_send(()).expect("failed to send output refresh message");
                 }
                 Some(Facility::Source) => {
-                    refresh_input_tx.send(()).expect("failed to send output refresh message");
+                    refresh_input_tx.unbounded_send(()).expect("failed to send output refresh message");
                 }
                 _ => {}
             }
         }))));
-    handler
+    pa.context
-        .context
-        .borrow_mut()
         .subscribe(InterestMaskSet::SINK | InterestMaskSet::SOURCE, |_| {});
+    let pa = Rc::new(pa);
     view! {
         window = ApplicationWindow {
             set_application: Some(application),
@@ -153,29 +149,27 @@ fn app(application: &Application) {
             }
         }
     }
-    refresh_input_rx.attach(
+    glib::MainContext::default().spawn_local(
-        None,
+        clone!(@weak inputs, @weak current_input, @weak input_volume, @strong pa => async move {
-        clone!(@weak inputs, @weak current_input, @weak input_volume => @default-return Continue(true), move |_| {
+            while let Some(()) = refresh_input_rx.next().await {
-            input::refresh_input_widgets(&inputs);
+                input::refresh_input_widgets(&pa, &inputs);
-            let default_input = input::refresh_default_input(&current_input);
+                let default_input = input::refresh_default_input(&pa, &current_input);
-            volume::update_volume(&default_input, &input_volume);
+                // XXX volume::update_volume(&default_input, &input_volume);
-            Continue(true)
+            }
         }),
     );
-    refresh_output_rx.attach(
+    glib::MainContext::default().spawn_local(
-        None,
+        clone!(@weak outputs, @weak current_output, @weak output_volume, @strong pa => async move {
-        clone!(@weak outputs, @weak current_output, @weak output_volume => @default-return Continue(true), move |_| {
+            while let Some(()) = refresh_output_rx.next().await {
-            output::refresh_output_widgets(&outputs);
+                output::refresh_output_widgets(&pa, &outputs);
-            let default_output = output::refresh_default_output(&current_output);
+                let default_output = output::refresh_default_output(&pa, &current_output);
-            volume::update_volume(&default_output, &output_volume);
+                // XXX volume::update_volume(&default_output, &output_volume);
-            Continue(true)
+            }
-        }),
-    );
-    now_playing_rx.attach(
-        None,
-        clone!(@weak playing_apps => @default-return Continue(true), move |all_metadata| {
-            Continue(true)
         }),
     );
+    glib::MainContext::default().spawn_local(clone!(@weak playing_apps => async move {
+        while let Some(all_metadata) = now_playing_rx.next().await {
+        }
+    }));
     window.show();
 }

applets/cosmic-applet-audio/src/output.rs

@@ -1,19 +1,21 @@
 use gtk4::{prelude::*, Button, Label, ListBox};
 use libcosmic_widgets::{relm4::RelmContainerExt, LabeledItem};
-use pulsectl::controllers::{types::DeviceInfo, DeviceControl, SinkController};
 use std::rc::Rc;
 
-fn get_outputs() -> Vec<DeviceInfo> {
+use crate::pa::{Sink, PA};
-    SinkController::create()
+
-        .expect("failed to create output controller")
+pub async fn get_outputs(pa: &PA) -> Vec<Sink> {
-        .list_devices()
+    // XXX handle error
+    pa.get_sink_info_list()
+        .await
         .expect("failed to list output devices")
 }
 
-pub fn refresh_default_output(label: &Label) -> DeviceInfo {
+pub async fn refresh_default_output(pa: &PA, label: &Label) -> Sink {
-    let default_output = SinkController::create()
+    // XXX handle error
-        .expect("failed to create output controller")
+    let default_output = pa
-        .get_default_device()
+        .get_default_sink()
+        .await
         .expect("failed to get default output");
     label.set_text(match &default_output.description {
         Some(name) => name.as_str(),
@@ -22,12 +24,11 @@ pub fn refresh_default_output(label: &Label) -> DeviceInfo {
     default_output
 }
 
-pub fn refresh_output_widgets(outputs: &ListBox) {
+pub async fn refresh_output_widgets(pa: &PA, outputs: &ListBox) {
     while let Some(row) = outputs.row_at_index(0) {
         outputs.remove(&row);
     }
-    for output in get_outputs() {
+    for output in get_outputs(pa).await {
-        let output = Rc::new(output.clone());
         let name = match &output.name {
             Some(name) => name.to_owned(),
             None => continue, // Why doesn't this have a name? Whatever, it's invalid.
@@ -40,10 +41,13 @@ pub fn refresh_output_widgets(outputs: &ListBox) {
                 set_child: set_current_input_device = &Button {
                     set_label: "Switch",
                     connect_clicked: move |_| {
+                        // XXX Need mutable borrow? Is this a problem for async?
+                        /*
                         SinkController::create()
                             .expect("failed to create output controller")
                             .set_default_device(&name)
                             .expect("failed to set default device");
+                        */
                     }
                 }
             }

applets/cosmic-applet-audio/src/pa.rs

@@ -1,12 +1,154 @@
-use async_io::Timer;
+use futures::{channel::oneshot, future::poll_fn, task::Poll};
-use futures_util::StreamExt;
+use libpulse_binding::{
-use libpulse_binding::mainloop::standard::Mainloop;
+    callbacks::ListResult,
-use std::{cell::RefCell, rc::Rc, time::Duration};
+    context::{introspect::SinkInfo, Context},
+};
+use libpulse_glib_binding::Mainloop;
+use std::rc::Rc;
 
-pub async fn drive_main_loop(main_loop: Rc<RefCell<Mainloop>>) {
+pub struct Sink {
-    let mut timer = Timer::interval(Duration::from_millis(100));
+    pub name: Option<String>,
-    loop {
+    pub description: Option<String>,
-        main_loop.borrow_mut().iterate(false);
+}
-        timer.next().await;
+
+pub struct Source {
+    pub name: Option<String>,
+    pub description: Option<String>,
+}
+
+pub struct ServerInfo {
+    pub default_sink_name: Option<String>,
+    pub default_source_name: Option<String>,
+}
+
+pub struct PA {
+    main_loop: Mainloop,
+    pub context: Context,
+}
+
+impl PA {
+    pub fn new() -> Option<Self> {
+        let main_loop = Mainloop::new(None)?;
+        let context = Context::new(&main_loop, "com.system76.cosmic.applets.audio")?;
+        Some(Self { main_loop, context })
+    }
+
+    pub async fn get_server_info(&self) -> ServerInfo {
+        let (sender, receiver) = oneshot::channel();
+        let mut sender = Some(sender);
+        self.context.introspect().get_server_info(move |info| {
+            sender.take().unwrap().send(ServerInfo {
+                default_sink_name: info.default_sink_name.clone().map(|x| x.into_owned()),
+                default_source_name: info.default_source_name.clone().map(|x| x.into_owned()),
+            });
+        });
+        receiver.await.unwrap()
+    }
+
+    pub async fn get_sink_info_list(&self) -> Result<Vec<Sink>, ()> {
+        let (sender, receiver) = oneshot::channel();
+        let mut sender = Some(sender);
+        let mut items = Some(Vec::new());
+        self.context
+            .introspect()
+            .get_sink_info_list(move |result| match result {
+                ListResult::Item(item) => items.as_mut().unwrap().push(Sink {
+                    name: item.name.clone().map(|x| x.into_owned()),
+                    description: item.description.clone().map(|x| x.into_owned()),
+                }),
+                ListResult::End => {
+                    sender.take().unwrap().send(Ok(items.take().unwrap()));
+                }
+                ListResult::Error => {
+                    sender.take().unwrap().send(Err(()));
+                }
+            });
+        receiver.await.unwrap()
+    }
+
+    pub async fn get_default_sink(&self) -> Result<Sink, ()> {
+        let name = match self.get_server_info().await.default_sink_name {
+            Some(name) => name,
+            None => {
+                return Err(());
+            }
+        };
+        let (sender, receiver) = oneshot::channel();
+        let mut sender = Some(sender);
+        let mut sink = None;
+        self.context
+            .introspect()
+            .get_sink_info_by_name(&name, move |result| match result {
+                ListResult::Item(item) => {
+                    sink = Some(Sink {
+                        name: item.name.clone().map(|x| x.into_owned()),
+                        description: item.description.clone().map(|x| x.into_owned()),
+                    });
+                }
+                ListResult::End => {
+                    sender.take().unwrap().send(sink.take().ok_or(()));
+                }
+                ListResult::Error => {
+                    sender.take().unwrap().send(Err(()));
+                }
+            });
+        receiver.await.unwrap()
+    }
+
+    /*
+    // XXX async wait and handle error
+    pub fn set_default_sink(&mut self, name: &str) {
+        self.context.set_default_sink(name, |_| {});
+    }
+    */
+
+    pub async fn get_source_info_list(&self) -> Result<Vec<Source>, ()> {
+        let (sender, receiver) = oneshot::channel();
+        let mut sender = Some(sender);
+        let mut items = Some(Vec::new());
+        self.context
+            .introspect()
+            .get_source_info_list(move |result| match result {
+                ListResult::Item(item) => items.as_mut().unwrap().push(Source {
+                    name: item.name.clone().map(|x| x.into_owned()),
+                    description: item.description.clone().map(|x| x.into_owned()),
+                }),
+                ListResult::End => {
+                    sender.take().unwrap().send(Ok(items.take().unwrap()));
+                }
+                ListResult::Error => {
+                    sender.take().unwrap().send(Err(()));
+                }
+            });
+        receiver.await.unwrap()
+    }
+
+    pub async fn get_default_source(&self) -> Result<Source, ()> {
+        let name = match self.get_server_info().await.default_source_name {
+            Some(name) => name,
+            None => {
+                return Err(());
+            }
+        };
+        let (sender, receiver) = oneshot::channel();
+        let mut sender = Some(sender);
+        let mut source = None;
+        self.context
+            .introspect()
+            .get_source_info_by_name(&name, move |result| match result {
+                ListResult::Item(item) => {
+                    source = Some(Source {
+                        name: item.name.clone().map(|x| x.into_owned()),
+                        description: item.description.clone().map(|x| x.into_owned()),
+                    });
+                }
+                ListResult::End => {
+                    sender.take().unwrap().send(source.take().ok_or(()));
+                }
+                ListResult::Error => {
+                    sender.take().unwrap().send(Err(()));
+                }
+            });
+        receiver.await.unwrap()
     }
 }

applets/cosmic-applet-audio/src/volume.rs

@@ -1,7 +1,8 @@
 use gtk4::{prelude::*, Scale};
 use libpulse_binding::volume::Volume;
-use pulsectl::controllers::types::DeviceInfo;
 
+/*
 pub fn update_volume(device: &DeviceInfo, scale: &Scale) {
     scale.set_value((device.volume.avg().0 as f64 / Volume::NORMAL.0 as f64) * 100.);
 }
+*/