Delete items from graph when they are removed by the pipewire server.

src/main.rs

@@ -1,4 +1,5 @@
 mod pipewire_connection;
+mod pipewire_state;
 mod view;
 
 use gtk::prelude::*;
@@ -19,7 +20,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Create the connection to the pipewire server and do an initial roundtrip before showing the view,
     // so that the graph is already populated when the window opens.
-    let pw_con = pipewire_connection::PipewireConnection::new(graphview.clone())
+    let pw_con = pipewire_connection::PipewireConnection::new(pipewire_state::PipewireState::new(
+        graphview.clone(),
+    ))
     .expect("Failed to initialize pipewire connection");
     pw_con.roundtrip();
     // From now on, call roundtrip() every second.

src/pipewire_connection.rs

@@ -1,50 +1,64 @@
-use crate::PipewireLink;
+use crate::pipewire_state::PipewireState;
 
+use gtk::glib::{self, clone};
 use pipewire as pw;
-use pw::{port::Direction, registry::ObjectType, PW_ID_CORE};
 
 use std::{
     cell::{Cell, RefCell},
     rc::Rc,
 };
 
+/// This struct is responsible for communication with the pipewire server.
+/// It handles new globals appearing as well as globals being removed.
+///
+/// It's `roundtrip` function must be called regularly to receive updates.
 pub struct PipewireConnection {
     mainloop: pw::MainLoop,
     _context: pw::Context<pw::MainLoop>,
-    core: pw::Core,
+    core: Rc<pw::Core>,
     _registry: pw::registry::Registry,
-    _reg_listeners: pw::registry::Listener,
+    _listeners: pw::registry::Listener,
+    _state: Rc<RefCell<PipewireState>>,
 }
 
 impl PipewireConnection {
-    pub fn new(graphview: Rc<RefCell<crate::view::GraphView>>) -> Result<Self, String> {
+    pub fn new(state: PipewireState) -> Result<Self, String> {
+        // Initialize pipewire lib and obtain needed pipewire objects.
         pw::init();
         let mainloop = pw::MainLoop::new().map_err(|_| "Failed to create pipewire mainloop!")?;
         let context =
             pw::Context::new(&mainloop).map_err(|_| "Failed to create pipewire context")?;
-        let core = context
+        let core = Rc::new(
+            context
                 .connect()
-            .map_err(|_| "Failed to connect to pipewire core")?;
+                .map_err(|_| "Failed to connect to pipewire core")?,
+        );
         let registry = core.get_registry();
 
-        let graphview = Rc::downgrade(&graphview.clone());
+        let state = Rc::new(RefCell::new(state));
-        let reg_listeners = registry
+
+        // Notify state on globals added / removed
+        let _listeners = registry
             .add_listener_local()
-            .global(move |global| {
+            .global(clone!(@weak state => @default-panic, move |global| {
-                PipewireConnection::handle_global(graphview.upgrade().unwrap(), global)
+                state.borrow_mut().global(global);
-            })
+            }))
-            .global_remove(|_| { /* TODO */ })
+            .global_remove(clone!(@weak state => @default-panic, move |id| {
+                state.borrow_mut().global_remove(id);
+            }))
             .register();
 
         Ok(Self {
-            mainloop,
+            mainloop: mainloop,
             _context: context,
             core,
             _registry: registry,
-            _reg_listeners: reg_listeners,
+            _listeners,
+            _state: state,
         })
     }
 
+    /// Receive all events from the pipewire server, sending them to the `pipewire_state` struct for processing.
     pub fn roundtrip(&self) {
         let done = Rc::new(Cell::new(false));
         let pending = self.core.sync(0);
@@ -56,7 +70,7 @@ impl PipewireConnection {
             .core
             .add_listener_local()
             .done(move |id, seq| {
-                if id == PW_ID_CORE && seq == pending {
+                if id == pw::PW_ID_CORE && seq == pending {
                     done_clone.set(true);
                     loop_clone.quit();
                 }
@@ -67,89 +81,4 @@ impl PipewireConnection {
             self.mainloop.run();
         }
     }
-
-    fn handle_global(
-        graphview: Rc<RefCell<crate::view::GraphView>>,
-        global: pw::registry::GlobalObject,
-    ) {
-        match global.type_ {
-            ObjectType::Node => {
-                let node_widget = crate::view::Node::new(&format!(
-                    "{}",
-                    global
-                        .props
-                        .map(|dict| String::from(
-                            dict.get("node.nick")
-                                .or(dict.get("node.description"))
-                                .or(dict.get("node.name"))
-                                .unwrap_or_default()
-                        ))
-                        .unwrap_or_default()
-                ));
-
-                graphview.borrow_mut().add_node(global.id, node_widget);
-            }
-            ObjectType::Port => {
-                let props = global.props.expect("Port object is missing properties");
-                let port_label = format!("{}", props.get("port.name").unwrap_or_default());
-                let node_id: u32 = props
-                    .get("node.id")
-                    .expect("Port has no node.id property!")
-                    .parse()
-                    .expect("Could not parse node.id property");
-                let port = crate::view::port::Port::new(
-                    global.id,
-                    &port_label,
-                    if matches!(props.get("port.direction"), Some("in")) {
-                        Direction::Input
-                    } else {
-                        Direction::Output
-                    },
-                );
-
-                graphview
-                    .borrow_mut()
-                    .add_port_to_node(node_id, global.id, port);
-            }
-            ObjectType::Link => {
-                let props = global.props.expect("Link object is missing properties");
-                let input_node: u32 = props
-                    .get("link.input.node")
-                    .expect("Link has no link.input.node property")
-                    .parse()
-                    .expect("Could not parse link.input.node property");
-                let input_port: u32 = props
-                    .get("link.input.port")
-                    .expect("Link has no link.input.port property")
-                    .parse()
-                    .expect("Could not parse link.input.port property");
-                let output_node: u32 = props
-                    .get("link.output.node")
-                    .expect("Link has no link.input.node property")
-                    .parse()
-                    .expect("Could not parse link.input.node property");
-                let output_port: u32 = props
-                    .get("link.output.port")
-                    .expect("Link has no link.output.port property")
-                    .parse()
-                    .expect("Could not parse link.output.port property");
-                graphview.borrow_mut().add_link(
-                    global.id,
-                    PipewireLink {
-                        node_from: output_node,
-                        port_from: output_port,
-                        node_to: input_node,
-                        port_to: input_port,
-                    },
-                );
-            }
-            _ => {}
-        }
-    }
 }
-
-/*impl Drop for PipewireConnection {
-    fn drop(&mut self) {
-        unsafe { pw::deinit() }
-    }
-}*/

src/pipewire_state.rs

@@ -0,0 +1,169 @@
+use crate::{view, PipewireLink};
+
+use pipewire::{
+    port::Direction,
+    registry::{GlobalObject, ObjectType},
+};
+
+use std::{cell::RefCell, collections::HashMap, rc::Rc};
+
+enum Item {
+    Node(view::Node),
+    Port { node_id: u32 },
+    Link,
+}
+
+/// This struct stores the state of the pipewire graph.
+///
+/// It receives updates from the [`PipewireConnection`](crate::pipewire_connection::PipewireConnection)
+/// responsible for updating it and applies them to its internal state.
+///
+/// It also keeps the view updated to always reflect this internal state.
+pub struct PipewireState {
+    graphview: Rc<RefCell<view::GraphView>>,
+    items: HashMap<u32, Item>,
+}
+
+impl PipewireState {
+    pub fn new(graphview: Rc<RefCell<view::GraphView>>) -> Self {
+        let result = Self {
+            graphview,
+            items: HashMap::new(),
+        };
+
+        result
+    }
+
+    /// This function is called from the `PipewireConnection` struct responsible for updating this struct.
+    pub fn global(&mut self, global: GlobalObject) {
+        match global.type_ {
+            ObjectType::Node => {
+                self.add_node(global);
+            }
+            ObjectType::Port => {
+                self.add_port(global);
+            }
+            ObjectType::Link => {
+                self.add_link(global);
+            }
+            _ => {}
+        }
+    }
+
+    fn add_node(&mut self, node: GlobalObject) {
+        // Update graph to contain the new node.
+        let node_widget = crate::view::Node::new(&format!(
+            "{}",
+            node.props
+                .map(|dict| String::from(
+                    dict.get("node.nick")
+                        .or(dict.get("node.description"))
+                        .or(dict.get("node.name"))
+                        .unwrap_or_default()
+                ))
+                .unwrap_or_default()
+        ));
+        self.graphview
+            .borrow_mut()
+            .add_node(node.id, node_widget.clone());
+
+        // Save the created widget so we can delete ports easier.
+        self.items.insert(node.id, Item::Node(node_widget));
+    }
+
+    fn add_port(&mut self, port: GlobalObject) {
+        // Update graph to contain the new port.
+        let props = port.props.expect("Port object is missing properties");
+        let port_label = format!("{}", props.get("port.name").unwrap_or_default());
+        let node_id: u32 = props
+            .get("node.id")
+            .expect("Port has no node.id property!")
+            .parse()
+            .expect("Could not parse node.id property");
+        let new_port = crate::view::port::Port::new(
+            port.id,
+            &port_label,
+            if matches!(props.get("port.direction"), Some("in")) {
+                Direction::Input
+            } else {
+                Direction::Output
+            },
+        );
+
+        self.graphview
+            .borrow_mut()
+            .add_port_to_node(node_id, new_port.id, new_port);
+
+        // Save node_id so we can delete this port easily.
+        self.items.insert(port.id, Item::Port { node_id });
+    }
+
+    fn add_link(&mut self, link: GlobalObject) {
+        self.items.insert(link.id, Item::Link);
+
+        // Update graph to contain the new link.
+        let props = link.props.expect("Link object is missing properties");
+        let input_node: u32 = props
+            .get("link.input.node")
+            .expect("Link has no link.input.node property")
+            .parse()
+            .expect("Could not parse link.input.node property");
+        let input_port: u32 = props
+            .get("link.input.port")
+            .expect("Link has no link.input.port property")
+            .parse()
+            .expect("Could not parse link.input.port property");
+        let output_node: u32 = props
+            .get("link.output.node")
+            .expect("Link has no link.input.node property")
+            .parse()
+            .expect("Could not parse link.input.node property");
+        let output_port: u32 = props
+            .get("link.output.port")
+            .expect("Link has no link.output.port property")
+            .parse()
+            .expect("Could not parse link.output.port property");
+        self.graphview.borrow_mut().add_link(
+            link.id,
+            PipewireLink {
+                node_from: output_node,
+                port_from: output_port,
+                node_to: input_node,
+                port_to: input_port,
+            },
+        );
+    }
+
+    /// This function is called from the `PipewireConnection` struct responsible for updating this struct.
+    pub fn global_remove(&mut self, id: u32) {
+        if let Some(item) = self.items.get(&id) {
+            match item {
+                Item::Node(_) => self.remove_node(id),
+                Item::Port { node_id } => self.remove_port(id, *node_id),
+                Item::Link => self.remove_link(id),
+            }
+
+            self.items.remove(&id);
+        } else {
+            // FIXME: Switch to log macro
+            eprintln!(
+                "Attempted to remove item with id {} that is not saved in state",
+                id
+            );
+        }
+    }
+
+    fn remove_node(&self, id: u32) {
+        self.graphview.borrow().remove_node(id);
+    }
+
+    fn remove_port(&self, id: u32, node_id: u32) {
+        if let Some(Item::Node(node)) = self.items.get(&node_id) {
+            node.remove_port(id);
+        }
+    }
+
+    fn remove_link(&self, id: u32) {
+        self.graphview.borrow().remove_link(id);
+    }
+}

src/view/graph_view.rs

@@ -205,6 +205,14 @@ impl GraphView {
         private.nodes.borrow_mut().insert(id, node);
     }
 
+    pub fn remove_node(&self, id: u32) {
+        let private = imp::GraphView::from_instance(self);
+        let mut nodes = private.nodes.borrow_mut();
+        if let Some(node) = nodes.remove(&id) {
+            node.unparent();
+        }
+    }
+
     pub fn add_port_to_node(&self, node_id: u32, port_id: u32, port: crate::view::port::Port) {
         let private = imp::GraphView::from_instance(self);
 
@@ -229,6 +237,14 @@ impl GraphView {
         self.queue_draw();
     }
 
+    pub fn remove_link(&self, id: u32) {
+        let private = imp::GraphView::from_instance(self);
+        let mut links = private.links.borrow_mut();
+        links.remove(&id);
+
+        self.queue_draw();
+    }
+
     pub fn set_dragged(&self, widget: Option<gtk::Widget>) {
         *imp::GraphView::from_instance(self).dragged.borrow_mut() = widget;
     }

src/view/node.rs

@@ -142,4 +142,16 @@ impl Node {
             .get(&id)
             .map(|port_rc| port_rc.clone())
     }
+
+    pub fn remove_port(&self, id: u32) {
+        let private = imp::Node::from_instance(self);
+        if let Some(port) = private.ports.borrow_mut().remove(&id) {
+            match port.direction {
+                Direction::Input => private.num_ports_in.set(private.num_ports_in.get() - 1),
+                Direction::Output => private.num_ports_in.set(private.num_ports_out.get() - 1),
+            }
+
+            port.widget.unparent();
+        }
+    }
 }