Toggle links on and off when ports are connected by the user.

This extends the `Application` struct to keep more advanced state.
This state is then used to determine the needed information to create
or delete a link between the two connected ports.

A message to create/delete the link is then send to the pipewire thread,
which executed the request.

src/application.rs

@@ -7,11 +7,11 @@ use gtk::{
     subclass::prelude::*,
 };
 use log::{info, warn};
-use pipewire::spa::Direction;
+use pipewire::{channel::Sender, spa::Direction};
 
 use crate::{
     view::{self},
-    PipewireMessage,
+    GtkMessage, PipewireLink, PipewireMessage,
 };
 
 #[derive(Debug, Copy, Clone)]
@@ -21,25 +21,6 @@ pub enum MediaType {
     Midi,
 }
 
-/// Any pipewire item we need to keep track of.
-/// These will be saved in the controllers `state` map associated with their id.
-enum Item {
-    Node {
-        // Keep track of the widget to easily remove ports on it later.
-        // widget: view::Node,
-        // Keep track of the nodes media type to color ports on it.
-        media_type: Option<MediaType>,
-    },
-    Port {
-        // Save the id of the node this is on so we can remove the port from it
-        // when it is deleted.
-        node_id: u32,
-    },
-    // We don't need to memorize anything about links right now, but we need to
-    // be able to find out an id is a link.
-    Link,
-}
-
 // FIXME: This should be in its own .css file.
 static STYLE: &str = "
 .audio {
@@ -61,10 +42,13 @@ static STYLE: &str = "
 mod imp {
     use super::*;
 
+    use once_cell::unsync::OnceCell;
+
     #[derive(Default)]
     pub struct Application {
         pub(super) graphview: view::GraphView,
-        pub(super) state: RefCell<HashMap<u32, Item>>,
+        pub(super) state: RefCell<State>,
+        pub(super) pw_sender: OnceCell<RefCell<Sender<GtkMessage>>>,
     }
 
     #[glib::object_subclass]
@@ -118,11 +102,21 @@ glib::wrapper! {
 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>) -> Self {
+    pub(super) fn new(
+        gtk_receiver: Receiver<PipewireMessage>,
+        pw_sender: Sender<GtkMessage>,
+    ) -> Self {
         let app: Application =
             glib::Object::new(&[("application-id", &"org.freedesktop.ryuukyu.helvum")])
                 .expect("Failed to create new Application");
 
+        let imp = imp::Application::from_instance(&app);
+        imp.pw_sender
+            .set(RefCell::new(pw_sender))
+            // Discard the returned sender, as it does not implement `Debug`.
+            .map_err(|_| ())
+            .expect("pw_sender field was already set");
+
         // Add <Control-Q> shortcut for quitting the application.
         let quit = gtk::gio::SimpleAction::new("quit", None);
         quit.connect_activate(clone!(@weak app => move |_, _| {
@@ -185,7 +179,7 @@ impl Application {
 
         // Find out the nodes media type so that the port can be colored.
         let media_type =
-            if let Some(Item::Node { media_type, .. }) = imp.state.borrow().get(&node_id) {
+            if let Some(Item::Node { media_type, .. }) = imp.state.borrow().get(node_id) {
                 media_type.to_owned()
             } else {
                 warn!("Node not found for Port {}", id);
@@ -196,32 +190,92 @@ impl Application {
         imp.state.borrow_mut().insert(id, Item::Port { node_id });
 
         let port = view::Port::new(id, name.as_str(), direction, media_type);
+
+        // Create or delete a link if the widget emits the "port-toggled" signal.
+        if let Err(e) = port.connect_local(
+            "port_toggled",
+            false,
+            clone!(@weak self as app => @default-return None, move |args| {
+                // Args always look like this: &[widget, id_port_from, id_port_to]
+                let port_from = args[1].get_some::<u32>().unwrap();
+                let port_to = args[2].get_some::<u32>().unwrap();
+
+                app.toggle_link(port_from, port_to);
+
+                None
+            }),
+        ) {
+            warn!("Failed to connect to \"port-toggled\" signal: {}", e);
+        }
+
         imp.graphview.add_port(node_id, id, port);
     }
 
     /// Add a new link to the view.
-    pub fn add_link(&self, id: u32, link: crate::PipewireLink) {
+    pub fn add_link(&self, id: u32, link: PipewireLink) {
         info!("Adding link to graph: id {}", id);
 
         let imp = imp::Application::from_instance(self);
 
         // FIXME: Links should be colored depending on the data they carry (video, audio, midi) like ports are.
 
-        imp.state.borrow_mut().insert(id, Item::Link);
+        imp.state.borrow_mut().insert(
+            id,
+            Item::Link {
+                output_port: link.port_from,
+                input_port: link.port_to,
+            },
+        );
 
         // Update graph to contain the new link.
         imp.graphview.add_link(id, link);
     }
 
+    // Toggle a link between the two specified ports on the remote pipewire server.
+    fn toggle_link(&self, port_from: u32, port_to: u32) {
+        let imp = imp::Application::from_instance(self);
+        let sender = imp.pw_sender.get().expect("pw_sender not set").borrow_mut();
+        let state = imp.state.borrow_mut();
+
+        if let Some(id) = state.get_link_id(port_from, port_to) {
+            info!("Requesting removal of link with id {}", id);
+
+            sender
+                .send(GtkMessage::DestroyGlobal(id))
+                .expect("Failed to send message");
+        } else {
+            info!(
+                "Requesting creation of link from port id:{} to port id:{}",
+                port_from, port_to
+            );
+
+            let node_from = state
+                .get_node_of_port(port_from)
+                .expect("Requested port not in state");
+            let node_to = state
+                .get_node_of_port(port_to)
+                .expect("Requested port not in state");
+
+            sender
+                .send(GtkMessage::CreateLink(PipewireLink {
+                    node_from,
+                    port_from,
+                    node_to,
+                    port_to,
+                }))
+                .expect("Failed to send message");
+        }
+    }
+
     /// Handle a global object being removed.
     pub fn remove_global(&self, id: u32) {
         let imp = imp::Application::from_instance(self);
 
-        if let Some(item) = imp.state.borrow_mut().remove(&id) {
+        if let Some(item) = imp.state.borrow_mut().remove(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),
+                Item::Link { .. } => self.remove_link(id),
             }
         } else {
             warn!(
@@ -256,3 +310,83 @@ impl Application {
         imp.graphview.remove_link(id);
     }
 }
+
+/// Any pipewire item we need to keep track of.
+/// These will be saved in the [`Application`]s `state` struct associated with their id.
+enum Item {
+    Node {
+        // Keep track of the nodes media type to color ports on it.
+        media_type: Option<MediaType>,
+    },
+    Port {
+        // Save the id of the node this is on so we can remove the port from it
+        // when it is deleted.
+        node_id: u32,
+    },
+    // We don't need to memorize anything about links right now, but we need to
+    // be able to find out an id is a link.
+    Link {
+        output_port: u32,
+        input_port: u32,
+    },
+}
+
+/// This struct keeps track of any relevant items and stores them under their IDs.
+///
+/// Given two port ids, it can also efficiently find the id of the link that connects them.
+#[derive(Default)]
+struct State {
+    /// Map pipewire ids to items.
+    items: HashMap<u32, Item>,
+    /// Map `(output port id, input port id)` tuples to the id of the link that connects them.
+    links: HashMap<(u32, u32), u32>,
+}
+
+impl State {
+    /// Add a new item under the specified id.
+    fn insert(&mut self, id: u32, item: Item) {
+        if let Item::Link {
+            output_port,
+            input_port,
+        } = item
+        {
+            self.links.insert((output_port, input_port), id);
+        }
+
+        self.items.insert(id, item);
+    }
+
+    /// Get the item that has the specified id.
+    fn get(&self, id: u32) -> Option<&Item> {
+        self.items.get(&id)
+    }
+
+    /// Get the id of the link that links the two specified ports.
+    fn get_link_id(&self, output_port: u32, input_port: u32) -> Option<u32> {
+        self.links.get(&(output_port, input_port)).copied()
+    }
+
+    /// Remove the item with the specified id, returning it if it exists.
+    fn remove(&mut self, id: u32) -> Option<Item> {
+        let removed = self.items.remove(&id);
+
+        if let Some(Item::Link {
+            output_port,
+            input_port,
+        }) = removed
+        {
+            self.links.remove(&(output_port, input_port));
+        }
+
+        removed
+    }
+
+    /// Convenience function: Get the id of the node a port is on
+    fn get_node_of_port(&self, port: u32) -> Option<u32> {
+        if let Some(Item::Port { node_id }) = self.get(port) {
+            Some(*node_id)
+        } else {
+            None
+        }
+    }
+}

src/main.rs

@@ -10,14 +10,18 @@ use gtk::{
 use pipewire::spa::Direction;
 
 /// Messages used GTK thread to command the pipewire thread.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 enum GtkMessage {
+    /// Create a new link.
+    CreateLink(PipewireLink),
+    /// Destroy the global with the specified id.
+    DestroyGlobal(u32),
     /// Quit the event loop and let the thread finish.
     Terminate,
 }
 
 /// Messages used pipewire thread to notify the GTK thread.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 enum PipewireMessage {
     /// A new node has appeared.
     NodeAdded {
@@ -38,7 +42,7 @@ enum PipewireMessage {
     ObjectRemoved { id: u32 },
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct PipewireLink {
     pub node_from: u32,
     pub port_from: u32,
@@ -56,7 +60,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pw_thread =
         std::thread::spawn(move || pipewire_connection::thread_main(gtk_sender, pw_receiver));
 
-    let app = application::Application::new(gtk_receiver);
+    let app = application::Application::new(gtk_receiver, pw_sender.clone());
 
     app.run(&std::env::args().collect::<Vec<_>>());
 

src/pipewire_connection.rs

@@ -1,6 +1,11 @@
-use gtk::glib;
+use std::rc::Rc;
+
+use gtk::glib::{self, clone};
+use log::warn;
 use pipewire::{
+    link::Link,
     prelude::*,
+    properties,
     registry::GlobalObject,
     spa::{Direction, ForeignDict},
     types::ObjectType,
@@ -17,13 +22,31 @@ pub(super) fn thread_main(
     let mainloop = MainLoop::new().expect("Failed to create mainloop");
     let context = Context::new(&mainloop).expect("Failed to create context");
     let core = context.connect(None).expect("Failed to connect to remote");
-    let registry = core.get_registry().expect("Failed to get registry");
+    let registry = Rc::new(core.get_registry().expect("Failed to get registry"));
 
     let _receiver = pw_receiver.attach(&mainloop, {
         let mainloop = mainloop.clone();
-        move |msg| match msg {
+        clone!(@weak registry => move |msg| match msg {
+            GtkMessage::CreateLink(link) => {
+                if let Err(e) = core.create_object::<Link, _>(
+                    "link-factory",
+                    &properties! {
+                        "link.output.node" => link.node_from.to_string(),
+                        "link.output.port" => link.port_from.to_string(),
+                        "link.input.node" => link.node_to.to_string(),
+                        "link.input.port" => link.port_to.to_string(),
+                        "object.linger" => "1"
+                    },
+                ) {
+                    warn!("Failed to create link: {}", e);
+                }
+            }
+            GtkMessage::DestroyGlobal(id) => {
+                // FIXME: Handle error
+                registry.destroy_global(id);
+            }
             GtkMessage::Terminate => mainloop.quit(),
-        }
+        })
     });
 
     let _listener = registry