Update to latest gtk-rs crates

src/application.rs

@@ -16,7 +16,7 @@
 
 use adw::{
     gio,
-    glib::{self, clone, Receiver},
+    glib::{self, clone},
     gtk,
     prelude::*,
     subclass::prelude::*,
@@ -33,8 +33,9 @@ static AUTHORS: &str = env!("CARGO_PKG_AUTHORS");
 mod imp {
     use super::*;
 
+    use std::cell::OnceCell;
+
     use adw::subclass::prelude::AdwApplicationImpl;
-    use once_cell::unsync::OnceCell;
 
     #[derive(Default)]
     pub struct Application {
@@ -143,7 +144,7 @@ impl Application {
     /// Create the view.
     /// This will set up the entire user interface and prepare it for being run.
     pub(super) fn new(
-        gtk_receiver: Receiver<PipewireMessage>,
+        gtk_receiver: async_channel::Receiver<PipewireMessage>,
         pw_sender: Sender<GtkMessage>,
     ) -> Self {
         let app: Application = glib::Object::builder()

src/graph_manager.rs

@@ -23,9 +23,7 @@ use crate::{ui::graph::GraphView, GtkMessage, PipewireMessage};
 mod imp {
     use super::*;
 
-    use std::{cell::RefCell, collections::HashMap};
-
-    use once_cell::unsync::OnceCell;
+    use std::{cell::OnceCell, cell::RefCell, collections::HashMap};
 
     use crate::{ui::graph, MediaType, NodeType};
 
@@ -53,36 +51,58 @@ mod imp {
     impl ObjectImpl for GraphManager {}
 
     impl GraphManager {
-        pub fn attach_receiver(&self, receiver: glib::Receiver<crate::PipewireMessage>) {
-            receiver.attach(None, glib::clone!(
-                @weak self as imp => @default-return glib::ControlFlow::Continue,
-                move |msg| {
-                    match msg {
-                        PipewireMessage::NodeAdded { id, name, node_type } => imp.add_node(id, name.as_str(), node_type),
-                        PipewireMessage::NodeNameChanged { id, name, media_name } => imp.node_name_changed(id, &name, &media_name),
-                        PipewireMessage::PortAdded { id, node_id, name, direction } => imp.add_port(id, name.as_str(), node_id, direction),
-                        PipewireMessage::PortFormatChanged { id, media_type } => imp.port_media_type_changed(id, media_type),
-                        PipewireMessage::LinkAdded {
-                            id, port_from, port_to, active, media_type
-                        } => imp.add_link(id, port_from, port_to, active, media_type),
-                        PipewireMessage::LinkStateChanged { id, active } => imp.link_state_changed(id, active),
-                        PipewireMessage::LinkFormatChanged { id, media_type } => imp.link_format_changed(id, media_type),
-                        PipewireMessage::NodeRemoved { id } => imp.remove_node(id),
-                        PipewireMessage::PortRemoved { id, node_id } => imp.remove_port(id, node_id),
-                        PipewireMessage::LinkRemoved { id } => imp.remove_link(id),
-                        PipewireMessage::Connecting => {
-                            imp.obj().connection_banner().set_revealed(true);
-                        }
-                        PipewireMessage::Connected => {
-                            imp.obj().connection_banner().set_revealed(false);
-                        },
-                        PipewireMessage::Disconnected => {
-                            imp.clear();
-                        },
-                    };
-                    glib::ControlFlow::Continue
-                }
-            ));
+        pub async fn receive(&self, receiver: async_channel::Receiver<crate::PipewireMessage>) {
+            loop {
+                let Ok(msg) = receiver.recv().await else {
+                    continue;
+                };
+                match msg {
+                    PipewireMessage::NodeAdded {
+                        id,
+                        name,
+                        node_type,
+                    } => self.add_node(id, name.as_str(), node_type),
+                    PipewireMessage::NodeNameChanged {
+                        id,
+                        name,
+                        media_name,
+                    } => self.node_name_changed(id, &name, &media_name),
+                    PipewireMessage::PortAdded {
+                        id,
+                        node_id,
+                        name,
+                        direction,
+                    } => self.add_port(id, name.as_str(), node_id, direction),
+                    PipewireMessage::PortFormatChanged { id, media_type } => {
+                        self.port_media_type_changed(id, media_type)
+                    }
+                    PipewireMessage::LinkAdded {
+                        id,
+                        port_from,
+                        port_to,
+                        active,
+                        media_type,
+                    } => self.add_link(id, port_from, port_to, active, media_type),
+                    PipewireMessage::LinkStateChanged { id, active } => {
+                        self.link_state_changed(id, active)
+                    }
+                    PipewireMessage::LinkFormatChanged { id, media_type } => {
+                        self.link_format_changed(id, media_type)
+                    }
+                    PipewireMessage::NodeRemoved { id } => self.remove_node(id),
+                    PipewireMessage::PortRemoved { id, node_id } => self.remove_port(id, node_id),
+                    PipewireMessage::LinkRemoved { id } => self.remove_link(id),
+                    PipewireMessage::Connecting => {
+                        self.obj().connection_banner().set_revealed(true);
+                    }
+                    PipewireMessage::Connected => {
+                        self.obj().connection_banner().set_revealed(false);
+                    }
+                    PipewireMessage::Disconnected => {
+                        self.clear();
+                    }
+                };
+            }
         }
 
         /// Add a new node to the view.
@@ -332,19 +352,23 @@ glib::wrapper! {
     pub struct GraphManager(ObjectSubclass<imp::GraphManager>);
 }
 
+async fn receive(graph_manager: GraphManager, receiver: async_channel::Receiver<PipewireMessage>) {
+    graph_manager.imp().receive(receiver).await
+}
+
 impl GraphManager {
     pub fn new(
         graph: &GraphView,
         connection_banner: &adw::Banner,
         sender: PwSender<GtkMessage>,
-        receiver: glib::Receiver<PipewireMessage>,
+        receiver: async_channel::Receiver<PipewireMessage>,
     ) -> Self {
         let res: Self = glib::Object::builder()
             .property("graph", graph)
             .property("connection-banner", connection_banner)
             .build();
 
-        res.imp().attach_receiver(receiver);
+        glib::MainContext::default().spawn_local(receive(res.clone(), receiver));
         assert!(
             res.imp().pw_sender.set(sender).is_ok(),
             "Should be able to set pw_sender)"

src/main.rs

@@ -120,7 +120,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Start the pipewire thread with channels in both directions.
 
-    let (gtk_sender, gtk_receiver) = glib::MainContext::channel(glib::Priority::DEFAULT);
+    let (gtk_sender, gtk_receiver) = async_channel::unbounded();
     let (pw_sender, pw_receiver) = pipewire::channel::channel();
     let pw_thread =
         std::thread::spawn(move || pipewire_connection::thread_main(gtk_sender, pw_receiver));

src/pipewire_connection/mod.rs

@@ -63,7 +63,7 @@ enum ProxyItem {
 
 /// The "main" function of the pipewire thread.
 pub(super) fn thread_main(
-    gtk_sender: glib::Sender<PipewireMessage>,
+    gtk_sender: async_channel::Sender<PipewireMessage>,
     mut pw_receiver: pipewire::channel::Receiver<GtkMessage>,
 ) {
     let mainloop = MainLoop::new(None).expect("Failed to create mainloop");
@@ -81,7 +81,7 @@ pub(super) fn thread_main(
                 if !is_connecting {
                     is_connecting = true;
                     gtk_sender
-                        .send(PipewireMessage::Connecting)
+                        .send_blocking(PipewireMessage::Connecting)
                         .expect("Failed to send message");
                 }
 
@@ -116,7 +116,7 @@ pub(super) fn thread_main(
         if is_connecting {
             is_connecting = false;
             gtk_sender
-                .send(PipewireMessage::Connected)
+                .send_blocking(PipewireMessage::Connected)
                 .expect("Failed to send message");
         }
 
@@ -145,7 +145,7 @@ pub(super) fn thread_main(
                 }
 
                 if res == -libc::EPIPE {
-                    gtk_sender.send(PipewireMessage::Disconnected)
+                    gtk_sender.send_blocking(PipewireMessage::Disconnected)
                         .expect("Failed to send message");
                     mainloop.quit();
                 } else {
@@ -169,7 +169,7 @@ pub(super) fn thread_main(
             ))
             .global_remove(clone!(@strong proxies, @strong state => move |id| {
                 if let Some(item) = state.borrow_mut().remove(id) {
-                    gtk_sender.send(match item {
+                    gtk_sender.send_blocking(match item {
                         Item::Node { .. } => PipewireMessage::NodeRemoved {id},
                         Item::Port { node_id } => PipewireMessage::PortRemoved {id, node_id},
                         Item::Link { .. } => PipewireMessage::LinkRemoved {id},
@@ -202,7 +202,7 @@ fn get_node_name(props: &DictRef) -> &str {
 /// Handle a new node being added
 fn handle_node(
     node: &GlobalObject<&DictRef>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
     registry: &Rc<Registry>,
     proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
     state: &Rc<RefCell<State>>,
@@ -237,7 +237,7 @@ fn handle_node(
     state.borrow_mut().insert(node.id, Item::Node);
 
     sender
-        .send(PipewireMessage::NodeAdded {
+        .send_blocking(PipewireMessage::NodeAdded {
             id: node.id,
             name,
             node_type,
@@ -263,7 +263,7 @@ fn handle_node(
 
 fn handle_node_info(
     info: &NodeInfoRef,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
     proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
 ) {
     debug!("Received node info: {:?}", info);
@@ -280,7 +280,7 @@ fn handle_node_info(
         let name = get_node_name(props).to_string();
 
         sender
-            .send(PipewireMessage::NodeNameChanged {
+            .send_blocking(PipewireMessage::NodeNameChanged {
                 id,
                 name,
                 media_name: media_name.to_string(),
@@ -292,7 +292,7 @@ fn handle_node_info(
 /// Handle a new port being added
 fn handle_port(
     port: &GlobalObject<&DictRef>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
     registry: &Rc<Registry>,
     proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
     state: &Rc<RefCell<State>>,
@@ -326,7 +326,7 @@ fn handle_port_info(
     info: &PortInfoRef,
     proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
     state: &Rc<RefCell<State>>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
 ) {
     debug!("Received port info: {:?}", info);
 
@@ -367,7 +367,7 @@ fn handle_port_info(
         }
 
         sender
-            .send(PipewireMessage::PortAdded {
+            .send_blocking(PipewireMessage::PortAdded {
                 id,
                 node_id,
                 name,
@@ -380,7 +380,7 @@ fn handle_port_info(
 fn handle_port_enum_format(
     port_id: u32,
     param: Option<&pipewire::spa::pod::Pod>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
 ) {
     let media_type = param
         .and_then(|param| pipewire::spa::param::format_utils::parse_format(param).ok())
@@ -388,7 +388,7 @@ fn handle_port_enum_format(
         .unwrap_or(MediaType::Unknown);
 
     sender
-        .send(PipewireMessage::PortFormatChanged {
+        .send_blocking(PipewireMessage::PortFormatChanged {
             id: port_id,
             media_type,
         })
@@ -398,7 +398,7 @@ fn handle_port_enum_format(
 /// Handle a new link being added
 fn handle_link(
     link: &GlobalObject<&DictRef>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
     registry: &Rc<Registry>,
     proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
     state: &Rc<RefCell<State>>,
@@ -428,7 +428,7 @@ fn handle_link(
 fn handle_link_info(
     info: &LinkInfoRef,
     state: &Rc<RefCell<State>>,
-    sender: &glib::Sender<PipewireMessage>,
+    sender: &async_channel::Sender<PipewireMessage>,
 ) {
     debug!("Received link info: {:?}", info);
 
@@ -439,7 +439,7 @@ fn handle_link_info(
         // Info was an update - figure out if we should notify the gtk thread
         if info.change_mask().contains(LinkChangeMask::STATE) {
             sender
-                .send(PipewireMessage::LinkStateChanged {
+                .send_blocking(PipewireMessage::LinkStateChanged {
                     id,
                     active: matches!(info.state(), LinkState::Active),
                 })
@@ -447,7 +447,7 @@ fn handle_link_info(
         }
         if info.change_mask().contains(LinkChangeMask::FORMAT) {
             sender
-                .send(PipewireMessage::LinkFormatChanged {
+                .send_blocking(PipewireMessage::LinkFormatChanged {
                     id,
                     media_type: get_link_media_type(info),
                 })
@@ -461,7 +461,7 @@ fn handle_link_info(
         state.insert(id, Item::Link { port_from, port_to });
 
         sender
-            .send(PipewireMessage::LinkAdded {
+            .send_blocking(PipewireMessage::LinkAdded {
                 id,
                 port_from,
                 port_to,

src/ui/graph/port.rs

@@ -28,9 +28,9 @@ use super::PortHandle;
 mod imp {
     use super::*;
 
-    use std::cell::Cell;
+    use std::cell::{Cell, OnceCell};
 
-    use once_cell::{sync::Lazy, unsync::OnceCell};
+    use once_cell::sync::Lazy;
     use pipewire::spa::{param::format::MediaType, utils::Direction};
 
     /// Graphical representation of a pipewire port.