Release 0.3.0

to the left side of the screen. So by adding a constant offset,
we can avoid having possible links that are partly out of the view
just after starting the program

.gitignore

@@ -1,2 +1,3 @@
+/.flatpak-builder
 /.vscode
 /target

.gitlab-ci.yml

@@ -0,0 +1,75 @@
+include:
+  - project: 'freedesktop/ci-templates'             # the project to include from
+    ref: '98f557799157ebb0395cf11d40f01f61fbbace20' # git ref of that project
+    file: '/templates/fedora.yml'                   # the actual file to include
+
+stages:
+  - prepare
+  - lint
+  - test
+  - extras
+
+variables:
+  FDO_UPSTREAM_REPO: 'ryuukyu/helvum'
+
+# Version and tag for our current container
+.fedora:
+  variables:
+    FDO_DISTRIBUTION_VERSION: '34'
+    # Update this to trigger a container rebuild
+    FDO_DISTRIBUTION_TAG: '2021-05-06.0'
+
+build-fedora-container:
+  extends:
+  - .fedora                            # our template job above
+  - .fdo.container-build@fedora@x86_64 # the CI template
+  stage: prepare
+  variables:
+    # clang-devel: required by rust bindgen
+    FDO_DISTRIBUTION_PACKAGES: >-
+      rust
+      cargo
+      rustfmt
+      clippy
+      pipewire-devel
+      gtk4-devel
+      clang-devel
+
+rustfmt:
+  extends:
+    - .fedora
+    - .fdo.distribution-image@fedora
+  stage: lint
+  script:
+    - cargo fmt --version
+    - cargo fmt -- --color=always --check
+
+test-stable:
+  extends:
+    - .fedora
+    - .fdo.distribution-image@fedora
+  stage: test
+  script:
+    - rustc --version
+    - cargo build --color=always --all-targets
+    - cargo test --color=always
+
+rustdoc:
+  extends:
+    - .fedora
+    - .fdo.distribution-image@fedora
+  stage: extras
+  variables:
+    RUSTDOCFLAGS: '-Dwarnings'
+  script:
+    - rustdoc --version
+    - cargo doc --no-deps
+
+clippy:
+  extends:
+    - .fedora
+    - .fdo.distribution-image@fedora
+  stage: extras
+  script:
+    - cargo clippy --version
+    - cargo clippy --color=always --all-targets -- -D warnings

Cargo.toml

@@ -1,14 +1,22 @@
 [package]
-name = "graphui"
-version = "0.1.0"
+name = "helvum"
+version = "0.3.0"
 authors = ["Tom A. Wagner <tom.a.wagner@protonmail.com>"]
 edition = "2018"
+license = "GPL-3.0-only"
+description = "A GTK patchbay for pipewire"
+repository = "https://gitlab.freedesktop.org/ryuukyu/helvum"
+readme = "README.md"
+keywords = ["pipewire", "gtk", "patchbay", "gui", "utility"]
+categories = ["gui", "multimedia"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-pipewire = { git = "https://gitlab.freedesktop.org/gdesmott/pipewire-rs", branch = "proxies"}
-gtk = { git = "https://github.com/gtk-rs/gtk4-rs/", package = "gtk4" }
+pipewire = "0.4"
+gtk = { version = "0.2", package = "gtk4" }
 
 log = "0.4.11"
-env_logger = "0.8.2"
+env_logger = "0.9.0"
+
+once_cell = "1.7.2"

LICENSE

@@ -631,7 +631,7 @@ to attach them to the start of each source file to most effectively
 state the exclusion of warranty; and each file should have at least
 the "copyright" line and a pointer to where the full notice is found.
 
-    Pipewire Graphui
+    Helvum
     Copyright (C) 2020  Ryuukyu
 
     This program is free software: you can redistribute it and/or modify
@@ -652,7 +652,7 @@ Also add information on how to contact you by electronic and paper mail.
   If the program does terminal interaction, make it output a short
 notice like this when it starts in an interactive mode:
 
-    Pipewire Graphui  Copyright (C) 2020  Ryuukyu
+    Helvum  Copyright (C) 2020  Ryuukyu
     This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
     This is free software, and you are welcome to redistribute it
     under certain conditions; type `show c' for details.

README.md

@@ -1,23 +1,51 @@
-A graphical interface for viewing the current pipewire graph (name pending, suggestions welcome), inspired by the JACK tool [catia](https://kx.studio/Applications:Catia).
+Helvum is a GTK-based patchbay for pipewire, inspired by the JACK tool [catia](https://kx.studio/Applications:Catia).
 
 ![Screenshot](screenshot.png)
 
+[![Packaging status](https://repology.org/badge/vertical-allrepos/helvum.svg)](https://repology.org/project/helvum/versions)
+
+
 # Features planned
 
-- Allow creation of links from one port to another.
-- Color ports and links based on whether they carry audio, video or midi data.
+- Volume control
+- "Debug mode" that lets you view advanced information for nodes and ports
 
 More suggestions are welcome!
 
 # Building
+
+## Via flatpak (recommended)
+The recommended way to build is using flatpak, which will take care of all dependencies and avoid any problems that may come from different system configurations.
+
+First, install the required flatpak platform and SDK, if you dont have them already:
+```shell
+$ flatpak install org.gnome.{Platform,Sdk}//40 org.freedesktop.Sdk.Extension.rust-stable//20.08
+```
+
+To compile and install as a flatpak, run
+```shell
+$ flatpak-builder --install flatpak-build/ org.freedesktop.ryuukyu.Helvum.json
+```
+
+You can then run the app via
+```shell
+flatpak run org.freedesktop.ryuukyu.Helvum
+```
+
+## Manually
 For compilation, you will need:
 
 - An up-to-date rust toolchain
-- gtk-4.0 and pipewire-0.3 development headers
+- `libclang-3.7` or higher
+- `gtk-4.0` and `pipewire-0.3` development headers
 
 To compile, run
 
     $ cargo build --release
 
 in the repository root.
-The resulting binary will be at `target/release/graphui`.
+The resulting binary will be at `target/release/helvum`.
+
+# License
+Helvum is distributed under the terms of the GPL3 license.
+See LICENSE for more information.

docs/architecture.md

@@ -0,0 +1,51 @@
+# Architecture
+If you want to understand the high-level architecture of helvum,
+this document is the right place.
+
+It provides a birds-eye view of the general architecture, and also goes into details on some
+components like the view.
+
+# Top Level Architecture
+Helvum uses an architecture with the components laid out like this:
+
+```
+┌──────┐
+│ GTK  │
+│ View │
+└────┬─┘
+ Λ   ┆
+ │<───── updates view                               ┌───────┐
+ │   ┆                                              │ State │
+ │   ┆<─ notifies of user input                     └───────┘
+ │   ┆      (using signals)                             Λ
+ │   ┆                                                  │
+ │   ┆                                                  │<─── updates/reads state
+ │   V           notifies of remote changes             │
+┌┴────────────┐         via messages          ┌─────────┴─────────┐
+│ Application │<╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤     Seperate      │
+│   Object    ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌>│  Pipewire Thread  │
+└─────────────┘   request changes to remote   └───────────────────┘
+                        via messages                    Λ
+                                                        ║
+                                                        ║
+                                                        V
+                                            [ Remote Pipewire Server ]
+```
+The program is split between two cooperating threads.
+The GTK thread (displayed on the left side) will sit in a GTK event processing loop, while the pipewire thread (displayed on the right side) will sit in a pipewire event processing loop.
+
+The `Application` object inside the GTK thread communicates with the pipewire thread using two channels,
+where each message sent by one thread will trigger the loop of the other thread to invoke a callback
+with the received message.
+
+For each change on the remote pipewire server, the pipewire thread updates the state and notifies
+the `Application` in the GTK thread if changes are needed.
+The `Application` then updates the view to reflect those changes.
+
+Additionally, a user may also make changes using the view.
+For each change, the view notifies the `Application` by emitting a matching signal.
+The `Application` will then ask the pipewire thread to make those changes on the remote. \
+These changes will then be applied to the view like any other remote changes as explained above.
+
+# View Architecture
+TODO

org.freedesktop.ryuukyu.Helvum.json

@@ -0,0 +1,36 @@
+{
+    "app-id": "org.freedesktop.ryuukyu.Helvum",
+    "runtime": "org.gnome.Platform",
+    "runtime-version": "40",
+    "sdk": "org.gnome.Sdk",
+    "sdk-extensions": ["org.freedesktop.Sdk.Extension.rust-stable"],
+    "command": "helvum",
+    "finish-args" : [
+        "--socket=fallback-x11",
+        "--socket=wayland",
+        "--device=dri",
+        "--share=ipc",
+        "--filesystem=xdg-run/pipewire-0"
+    ],
+    "build-options" : {
+        "append-path" : "/usr/lib/sdk/rust-stable/bin",
+        "build-args" : [
+            "--share=network"
+        ]
+    },
+    "modules": [
+        {
+            "name": "Helvum",
+            "buildsystem": "simple",
+            "build-commands": [
+                "cargo install --path . --root /app --no-track"
+            ],
+            "sources": [
+                {
+                    "type": "dir",
+                    "path": "./"
+                }
+            ]
+        }
+    ]
+}

src/application.rs

@@ -0,0 +1,243 @@
+use std::cell::RefCell;
+
+use gtk::{
+    gio,
+    glib::{self, clone, Continue, Receiver},
+    prelude::*,
+    subclass::prelude::*,
+};
+use log::{info, warn};
+use pipewire::{channel::Sender, spa::Direction};
+
+use crate::{
+    view::{self},
+    GtkMessage, MediaType, PipewireLink, PipewireMessage,
+};
+
+static STYLE: &str = include_str!("style.css");
+
+mod imp {
+    use super::*;
+
+    use once_cell::unsync::OnceCell;
+
+    #[derive(Default)]
+    pub struct Application {
+        pub(super) graphview: view::GraphView,
+        pub(super) pw_sender: OnceCell<RefCell<Sender<GtkMessage>>>,
+    }
+
+    #[glib::object_subclass]
+    impl ObjectSubclass for Application {
+        const NAME: &'static str = "HelvumApplication";
+        type Type = super::Application;
+        type ParentType = gtk::Application;
+    }
+
+    impl ObjectImpl for Application {}
+    impl ApplicationImpl for Application {
+        fn activate(&self, app: &Self::Type) {
+            let scrollwindow = gtk::ScrolledWindowBuilder::new()
+                .child(&self.graphview)
+                .build();
+            let window = gtk::ApplicationWindowBuilder::new()
+                .application(app)
+                .default_width(1280)
+                .default_height(720)
+                .title("Helvum - Pipewire Patchbay")
+                .child(&scrollwindow)
+                .build();
+            window
+                .settings()
+                .set_gtk_application_prefer_dark_theme(true);
+            window.show();
+        }
+
+        fn startup(&self, app: &Self::Type) {
+            self.parent_startup(app);
+
+            // Load CSS from the STYLE variable.
+            let provider = gtk::CssProvider::new();
+            provider.load_from_data(STYLE.as_bytes());
+            gtk::StyleContext::add_provider_for_display(
+                &gtk::gdk::Display::default().expect("Error initializing gtk css provider."),
+                &provider,
+                gtk::STYLE_PROVIDER_PRIORITY_APPLICATION,
+            );
+        }
+    }
+    impl GtkApplicationImpl for Application {}
+}
+
+glib::wrapper! {
+    pub struct Application(ObjectSubclass<imp::Application>)
+        @extends gio::Application, gtk::Application,
+        @implements gio::ActionGroup, gio::ActionMap;
+}
+
+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>,
+        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 |_, _| {
+            app.quit();
+        }));
+        app.set_accels_for_action("app.quit", &["<Control>Q"]);
+        app.add_action(&quit);
+
+        // React to messages received from the pipewire thread.
+        gtk_receiver.attach(
+            None,
+            clone!(
+                @weak app => @default-return Continue(true),
+                move |msg| {
+                    match msg {
+                        PipewireMessage::NodeAdded{ id, name } => app.add_node(id, name.as_str()),
+                        PipewireMessage::PortAdded{ id, node_id, name, direction, media_type } => app.add_port(id, name.as_str(), node_id, direction, media_type),
+                        PipewireMessage::LinkAdded{ id, node_from, port_from, node_to, port_to, active} => app.add_link(id, node_from, port_from, node_to, port_to, active),
+                        PipewireMessage::LinkStateChanged { id, active } => app.link_state_changed(id, active), // TODO
+                        PipewireMessage::NodeRemoved { id } => app.remove_node(id),
+                        PipewireMessage::PortRemoved { id, node_id } => app.remove_port(id, node_id),
+                        PipewireMessage::LinkRemoved { id } => app.remove_link(id)
+                    };
+                    Continue(true)
+                }
+            ),
+        );
+
+        app
+    }
+
+    /// Add a new node to the view.
+    fn add_node(&self, id: u32, name: &str) {
+        info!("Adding node to graph: id {}", id);
+
+        imp::Application::from_instance(self)
+            .graphview
+            .add_node(id, view::Node::new(name));
+    }
+
+    /// Add a new port to the view.
+    fn add_port(
+        &self,
+        id: u32,
+        name: &str,
+        node_id: u32,
+        direction: Direction,
+        media_type: Option<MediaType>,
+    ) {
+        info!("Adding port to graph: id {}", id);
+
+        let imp = imp::Application::from_instance(self);
+
+        let port = view::Port::new(id, name, 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::<u32>().unwrap();
+                let port_to = args[2].get::<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.
+    fn add_link(
+        &self,
+        id: u32,
+        node_from: u32,
+        port_from: u32,
+        node_to: u32,
+        port_to: u32,
+        active: bool,
+    ) {
+        info!("Adding link to graph: id {}", id);
+
+        // FIXME: Links should be colored depending on the data they carry (video, audio, midi) like ports are.
+
+        // Update graph to contain the new link.
+        imp::Application::from_instance(self).graphview.add_link(
+            id,
+            PipewireLink {
+                node_from,
+                port_from,
+                node_to,
+                port_to,
+            },
+            active,
+        );
+    }
+
+    fn link_state_changed(&self, id: u32, active: bool) {
+        info!(
+            "Link state changed: Link (id={}) is now {}",
+            id,
+            if active { "active" } else { "inactive" }
+        );
+
+        imp::Application::from_instance(self)
+            .graphview
+            .set_link_state(id, active);
+    }
+
+    // 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();
+        sender
+            .send(GtkMessage::ToggleLink { port_from, port_to })
+            .expect("Failed to send message");
+    }
+
+    /// Remove the node with the specified id from the view.
+    fn remove_node(&self, id: u32) {
+        info!("Removing node from graph: id {}", id);
+
+        let imp = imp::Application::from_instance(self);
+        imp.graphview.remove_node(id);
+    }
+
+    /// Remove the port with the id `id` from the node with the id `node_id`
+    /// from the view.
+    fn remove_port(&self, id: u32, node_id: u32) {
+        info!("Removing port from graph: id {}, node_id: {}", id, node_id);
+
+        let imp = imp::Application::from_instance(self);
+        imp.graphview.remove_port(id, node_id);
+    }
+
+    /// Remove the link with the specified id from the view.
+    fn remove_link(&self, id: u32) {
+        info!("Removing link from graph: id {}", id);
+
+        let imp = imp::Application::from_instance(self);
+        imp.graphview.remove_link(id);
+    }
+}

src/main.rs

@@ -1,12 +1,68 @@
+mod application;
 mod pipewire_connection;
-mod pipewire_state;
 mod view;
 
-use gtk::prelude::*;
+use gtk::{
+    glib::{self, PRIORITY_DEFAULT},
+    prelude::*,
+};
+use pipewire::spa::Direction;
 
-use std::{cell::RefCell, rc::Rc};
+/// Messages sent by the GTK thread to notify the pipewire thread.
+#[derive(Debug, Clone)]
+enum GtkMessage {
+    /// Toggle a link between the two specified ports.
+    ToggleLink { port_from: u32, port_to: u32 },
+    /// Quit the event loop and let the thread finish.
+    Terminate,
+}
 
-#[derive(Debug)]
+/// Messages sent by the pipewire thread to notify the GTK thread.
+#[derive(Debug, Clone)]
+enum PipewireMessage {
+    NodeAdded {
+        id: u32,
+        name: String,
+    },
+    PortAdded {
+        id: u32,
+        node_id: u32,
+        name: String,
+        direction: Direction,
+        media_type: Option<MediaType>,
+    },
+    LinkAdded {
+        id: u32,
+        node_from: u32,
+        port_from: u32,
+        node_to: u32,
+        port_to: u32,
+        active: bool,
+    },
+    LinkStateChanged {
+        id: u32,
+        active: bool,
+    },
+    NodeRemoved {
+        id: u32,
+    },
+    PortRemoved {
+        id: u32,
+        node_id: u32,
+    },
+    LinkRemoved {
+        id: u32,
+    },
+}
+
+#[derive(Debug, Copy, Clone)]
+pub enum MediaType {
+    Audio,
+    Video,
+    Midi,
+}
+
+#[derive(Debug, Clone)]
 pub struct PipewireLink {
     pub node_from: u32,
     pub port_from: u32,
@@ -18,42 +74,25 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     env_logger::init();
     gtk::init()?;
 
-    let graphview = Rc::new(RefCell::new(view::GraphView::new()));
+    // Aquire main context so that we can attach the gtk channel later.
+    let ctx = glib::MainContext::default();
+    let _guard = ctx.acquire().unwrap();
 
-    // 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(pipewire_state::PipewireState::new(
-        graphview.clone(),
-    ))
-    .expect("Failed to initialize pipewire connection");
-    pw_con.roundtrip();
-    // From now on, call roundtrip() every second.
-    gtk::glib::timeout_add_seconds_local(1, move || {
-        pw_con.roundtrip();
-        Continue(true)
-    });
+    // Start the pipewire thread with channels in both directions.
+    let (gtk_sender, gtk_receiver) = glib::MainContext::channel(PRIORITY_DEFAULT);
+    let (pw_sender, pw_receiver) = pipewire::channel::channel();
+    let pw_thread =
+        std::thread::spawn(move || pipewire_connection::thread_main(gtk_sender, pw_receiver));
 
-    let app = gtk::Application::new(Some("org.freedesktop.pipewire.graphui"), Default::default())
-        .expect("Application creation failed");
+    let app = application::Application::new(gtk_receiver, pw_sender.clone());
 
-    app.connect_activate(move |app| {
-        let scrollwindow = gtk::ScrolledWindowBuilder::new()
-            .child(&*graphview.borrow())
-            .build();
-        let window = gtk::ApplicationWindowBuilder::new()
-            .application(app)
-            .default_width(1280)
-            .default_height(720)
-            .title("Pipewire Graph Editor")
-            .child(&scrollwindow)
-            .build();
-        window
-            .get_settings()
-            .set_property_gtk_application_prefer_dark_theme(true);
-        window.show();
-    });
+    app.run();
 
-    app.run(&std::env::args().collect::<Vec<_>>());
+    pw_sender
+        .send(GtkMessage::Terminate)
+        .expect("Failed to send message");
+
+    pw_thread.join().expect("Pipewire thread panicked");
 
     Ok(())
 }

src/pipewire_connection.rs

@@ -1,84 +1,277 @@
-use crate::pipewire_state::PipewireState;
+mod state;
+
+use std::{cell::RefCell, collections::HashMap, rc::Rc};
 
 use gtk::glib::{self, clone};
-use pipewire as pw;
-
-use std::{
-    cell::{Cell, RefCell},
-    rc::Rc,
+use log::{debug, info, warn};
+use pipewire::{
+    link::{Link, LinkChangeMask, LinkListener, LinkState},
+    prelude::*,
+    properties,
+    registry::{GlobalObject, Registry},
+    spa::{Direction, ForeignDict},
+    types::ObjectType,
+    Context, Core, MainLoop,
 };
 
-/// 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: Rc<pw::Core>,
-    _registry: pw::registry::Registry,
-    _listeners: pw::registry::Listener,
-    _state: Rc<RefCell<PipewireState>>,
+use crate::{GtkMessage, MediaType, PipewireMessage};
+use state::{Item, State};
+
+enum ProxyItem {
+    Link {
+        _proxy: Link,
+        _listener: LinkListener,
+    },
 }
 
-impl PipewireConnection {
-    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 = Rc::new(
-            context
-                .connect()
-                .map_err(|_| "Failed to connect to pipewire core")?,
-        );
-        let registry = core.get_registry();
+/// The "main" function of the pipewire thread.
+pub(super) fn thread_main(
+    gtk_sender: glib::Sender<PipewireMessage>,
+    pw_receiver: pipewire::channel::Receiver<GtkMessage>,
+) {
+    let mainloop = MainLoop::new().expect("Failed to create mainloop");
+    let context = Context::new(&mainloop).expect("Failed to create context");
+    let core = Rc::new(context.connect(None).expect("Failed to connect to remote"));
+    let registry = Rc::new(core.get_registry().expect("Failed to get registry"));
 
-        let state = Rc::new(RefCell::new(state));
+    // Keep proxies and their listeners alive so that we can receive info events.
+    let proxies = Rc::new(RefCell::new(HashMap::new()));
 
-        // Notify state on globals added / removed
-        let _listeners = registry
-            .add_listener_local()
-            .global(clone!(@weak state => @default-panic, move |global| {
-                state.borrow_mut().global(global);
-            }))
-            .global_remove(clone!(@weak state => @default-panic, move |id| {
-                state.borrow_mut().global_remove(id);
-            }))
-            .register();
+    let state = Rc::new(RefCell::new(State::new()));
 
-        Ok(Self {
-            mainloop: mainloop,
-            _context: context,
-            core,
-            _registry: registry,
-            _listeners,
-            _state: state,
+    let _receiver = pw_receiver.attach(&mainloop, {
+        clone!(@strong mainloop, @weak core, @weak registry, @strong state => move |msg| match msg {
+            GtkMessage::ToggleLink { port_from, port_to } => toggle_link(port_from, port_to, &core, &registry, &state),
+            GtkMessage::Terminate => mainloop.quit(),
         })
-    }
+    });
 
-    /// 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);
-
-        let done_clone = done.clone();
-        let loop_clone = self.mainloop.clone();
-
-        let _listener = self
-            .core
-            .add_listener_local()
-            .done(move |id, seq| {
-                if id == pw::PW_ID_CORE && seq == pending {
-                    done_clone.set(true);
-                    loop_clone.quit();
+    let _listener = registry
+        .add_listener_local()
+        .global(clone!(@strong gtk_sender, @weak registry, @strong proxies, @strong state =>
+            move |global| match global.type_ {
+                ObjectType::Node => handle_node(global, &gtk_sender, &state),
+                ObjectType::Port => handle_port(global, &gtk_sender, &state),
+                ObjectType::Link => handle_link(global, &gtk_sender, &registry, &proxies, &state),
+                _ => {
+                    // Other objects are not interesting to us
                 }
-            })
-            .register();
+            }
+        ))
+        .global_remove(clone!(@strong proxies, @strong state => move |id| {
+            if let Some(item) = state.borrow_mut().remove(id) {
+                gtk_sender.send(match item {
+                    Item::Node { .. } => PipewireMessage::NodeRemoved {id},
+                    Item::Port { node_id } => PipewireMessage::PortRemoved {id, node_id},
+                    Item::Link { .. } => PipewireMessage::LinkRemoved {id},
+                }).expect("Failed to send message");
+            } else {
+                warn!(
+                    "Attempted to remove item with id {} that is not saved in state",
+                    id
+                );
+            }
 
-        while !done.get() {
-            self.mainloop.run();
+            proxies.borrow_mut().remove(&id);
+        }))
+        .register();
+
+    mainloop.run();
+}
+
+/// Handle a new node being added
+fn handle_node(
+    node: &GlobalObject<ForeignDict>,
+    sender: &glib::Sender<PipewireMessage>,
+    state: &Rc<RefCell<State>>,
+) {
+    let props = node
+        .props
+        .as_ref()
+        .expect("Node object is missing properties");
+
+    // Get the nicest possible name for the node, using a fallback chain of possible name attributes.
+    let name = String::from(
+        props
+            .get("node.nick")
+            .or_else(|| props.get("node.description"))
+            .or_else(|| props.get("node.name"))
+            .unwrap_or_default(),
+    );
+
+    // FIXME: Instead of checking these props, the "EnumFormat" parameter should be checked instead.
+    let media_type = props.get("media.class").and_then(|class| {
+        if class.contains("Audio") {
+            Some(MediaType::Audio)
+        } else if class.contains("Video") {
+            Some(MediaType::Video)
+        } else if class.contains("Midi") {
+            Some(MediaType::Midi)
+        } else {
+            None
+        }
+    });
+
+    state.borrow_mut().insert(
+        node.id,
+        Item::Node {
+            // widget: node_widget,
+            media_type,
+        },
+    );
+
+    sender
+        .send(PipewireMessage::NodeAdded { id: node.id, name })
+        .expect("Failed to send message");
+}
+
+/// Handle a new port being added
+fn handle_port(
+    port: &GlobalObject<ForeignDict>,
+    sender: &glib::Sender<PipewireMessage>,
+    state: &Rc<RefCell<State>>,
+) {
+    let props = port
+        .props
+        .as_ref()
+        .expect("Port object is missing properties");
+    let name = props.get("port.name").unwrap_or_default().to_string();
+    let node_id: u32 = props
+        .get("node.id")
+        .expect("Port has no node.id property!")
+        .parse()
+        .expect("Could not parse node.id property");
+    let direction = if matches!(props.get("port.direction"), Some("in")) {
+        Direction::Input
+    } else {
+        Direction::Output
+    };
+
+    // Find out the nodes media type so that the port can be colored.
+    let media_type = if let Some(Item::Node { media_type, .. }) = state.borrow().get(node_id) {
+        media_type.to_owned()
+    } else {
+        warn!("Node not found for Port {}", port.id);
+        None
+    };
+
+    // Save node_id so we can delete this port easily.
+    state.borrow_mut().insert(port.id, Item::Port { node_id });
+
+    sender
+        .send(PipewireMessage::PortAdded {
+            id: port.id,
+            node_id,
+            name,
+            direction,
+            media_type,
+        })
+        .expect("Failed to send message");
+}
+
+/// Handle a new link being added
+fn handle_link(
+    link: &GlobalObject<ForeignDict>,
+    sender: &glib::Sender<PipewireMessage>,
+    registry: &Rc<Registry>,
+    proxies: &Rc<RefCell<HashMap<u32, ProxyItem>>>,
+    state: &Rc<RefCell<State>>,
+) {
+    debug!(
+        "New link (id:{}) appeared, setting up info listener.",
+        link.id
+    );
+
+    let proxy: Link = registry.bind(link).expect("Failed to bind to link proxy");
+    let listener = proxy
+        .add_listener_local()
+        .info(clone!(@strong state, @strong sender => move |info| {
+            debug!("Received link info: {:?}", info);
+
+            let id = info.id();
+
+            let mut state = state.borrow_mut();
+            if let Some(Item::Link { .. }) = state.get(id) {
+                // Info was an update - figure out if we should notify the gtk thread
+                if info.change_mask().contains(LinkChangeMask::STATE) {
+                    sender.send(PipewireMessage::LinkStateChanged {
+                        id,
+                        active: matches!(info.state(), LinkState::Active)
+                    }).expect("Failed to send message");
+                }
+                // TODO -- check other values that might have changed
+            } else {
+                // First time we get info. We can now notify the gtk thread of a new link.
+                let node_from = info.output_node_id();
+                let port_from = info.output_port_id();
+                let node_to = info.input_node_id();
+                let port_to = info.input_port_id();
+
+                state.insert(id, Item::Link {
+                    port_from, port_to
+                });
+
+                sender.send(PipewireMessage::LinkAdded {
+                    id,
+                    node_from,
+                    port_from,
+                    node_to,
+                    port_to,
+                    active: matches!(info.state(), LinkState::Active)
+                }).expect(
+                    "Failed to send message"
+                );
+            }
+        }))
+        .register();
+
+    proxies.borrow_mut().insert(
+        link.id,
+        ProxyItem::Link {
+            _proxy: proxy,
+            _listener: listener,
+        },
+    );
+}
+
+/// Toggle a link between the two specified ports.
+fn toggle_link(
+    port_from: u32,
+    port_to: u32,
+    core: &Rc<Core>,
+    registry: &Rc<Registry>,
+    state: &Rc<RefCell<State>>,
+) {
+    let state = state.borrow_mut();
+    if let Some(id) = state.get_link_id(port_from, port_to) {
+        info!("Requesting removal of link with id {}", id);
+
+        // FIXME: Handle error
+        registry.destroy_global(id);
+    } 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");
+
+        if let Err(e) = core.create_object::<Link, _>(
+            "link-factory",
+            &properties! {
+                "link.output.node" => node_from.to_string(),
+                "link.output.port" => port_from.to_string(),
+                "link.input.node" => node_to.to_string(),
+                "link.input.port" => port_to.to_string(),
+                "object.linger" => "1"
+            },
+        ) {
+            warn!("Failed to create link: {}", e);
         }
     }
 }

src/pipewire_connection/state.rs

@@ -0,0 +1,81 @@
+use std::collections::HashMap;
+
+use crate::MediaType;
+
+/// Any pipewire item we need to keep track of.
+/// These will be saved in the `State` struct associated with their id.
+pub(super) 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,
+    },
+    Link {
+        port_from: u32,
+        port_to: 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)]
+pub(super) 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 {
+    /// Create a new, empty state.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add a new item under the specified id.
+    pub fn insert(&mut self, id: u32, item: Item) {
+        if let Item::Link {
+            port_from, port_to, ..
+        } = item
+        {
+            self.links.insert((port_from, port_to), id);
+        }
+
+        self.items.insert(id, item);
+    }
+
+    /// Get the item that has the specified id.
+    pub fn get(&self, id: u32) -> Option<&Item> {
+        self.items.get(&id)
+    }
+
+    /// Get the id of the link that links the two specified ports.
+    pub 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.
+    pub fn remove(&mut self, id: u32) -> Option<Item> {
+        let removed = self.items.remove(&id);
+
+        if let Some(Item::Link { port_from, port_to }) = removed {
+            self.links.remove(&(port_from, port_to));
+        }
+
+        removed
+    }
+
+    /// Convenience function: Get the id of the node a port is on
+    pub 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/pipewire_state.rs

@@ -1,168 +0,0 @@
-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 {
-            log::warn!(
-                "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/style.css

@@ -0,0 +1,14 @@
+.audio {
+    background: rgb(50,100,240);
+    color: black;
+}
+
+.video {
+    background: rgb(200,200,0);
+    color: black;
+}
+
+.midi {
+    background: rgb(200,0,50);
+    color: black;
+}

src/view/graph_view.rs

@@ -1,71 +1,86 @@
-use super::Node;
+use super::{Node, Port};
 
-use gtk::{glib, graphene, gsk, prelude::*, subclass::prelude::*, WidgetExt};
+use gtk::{
+    glib::{self, clone},
+    graphene, gsk,
+    prelude::*,
+    subclass::prelude::*,
+};
+use log::{error, warn};
 
 use std::collections::HashMap;
 
 mod imp {
     use super::*;
 
-    use gtk::{gdk, WidgetExt};
-
     use std::{cell::RefCell, rc::Rc};
 
+    use log::warn;
+
+    #[derive(Default)]
     pub struct GraphView {
         pub(super) nodes: RefCell<HashMap<u32, Node>>,
-        pub(super) links: RefCell<HashMap<u32, crate::PipewireLink>>,
-        pub(super) dragged: Rc<RefCell<Option<gtk::Widget>>>,
+        /// Stores the link and whether it is currently active.
+        pub(super) links: RefCell<HashMap<u32, (crate::PipewireLink, bool)>>,
     }
 
+    #[glib::object_subclass]
     impl ObjectSubclass for GraphView {
         const NAME: &'static str = "GraphView";
         type Type = super::GraphView;
         type ParentType = gtk::Widget;
-        type Instance = glib::subclass::simple::InstanceStruct<Self>;
-        type Class = glib::subclass::simple::ClassStruct<Self>;
-
-        glib::object_subclass!();
 
         fn class_init(klass: &mut Self::Class) {
             // The layout manager determines how child widgets are laid out.
             klass.set_layout_manager_type::<gtk::FixedLayout>();
         }
-
-        fn new() -> Self {
-            Self {
-                nodes: RefCell::new(HashMap::new()),
-                links: RefCell::new(HashMap::new()),
-                dragged: Rc::new(RefCell::new(None)),
-            }
-        }
     }
 
     impl ObjectImpl for GraphView {
         fn constructed(&self, obj: &Self::Type) {
             self.parent_constructed(obj);
 
-            // Move the Node that is currently being dragged to the cursor position as long as Mouse Button 1 is held.
-            let motion_controller = gtk::EventControllerMotion::new();
-            motion_controller.connect_motion(|controller, x, y| {
-                let instance = controller
-                    .get_widget()
-                    .unwrap()
-                    .dynamic_cast::<Self::Type>()
-                    .unwrap();
-                let this = imp::GraphView::from_instance(&instance);
+            let drag_state = Rc::new(RefCell::new(None));
+            let drag_controller = gtk::GestureDrag::new();
 
-                if let Some(ref widget) = *this.dragged.borrow() {
-                    if controller
-                        .get_current_event()
-                        .unwrap()
-                        .get_modifier_state()
-                        .contains(gdk::ModifierType::BUTTON1_MASK)
-                    {
-                        instance.move_node(&widget, x as f32, y as f32);
+            drag_controller.connect_drag_begin(
+                clone!(@strong drag_state => move |drag_controller, x, y| {
+                let mut drag_state = drag_state.borrow_mut();
+                let widget = drag_controller
+                    .widget()
+                    .expect("drag-begin event has no widget")
+                    .dynamic_cast::<Self::Type>()
+                    .expect("drag-begin event is not on the GraphView");
+                // pick() should at least return the widget itself.
+                let target = widget.pick(x, y, gtk::PickFlags::DEFAULT).expect("drag-begin pick() did not return a widget");
+                *drag_state = if target.ancestor(Port::static_type()).is_some() {
+                    // The user targeted a port, so the dragging should be handled by the Port
+                    // component instead of here.
+                    None
+                } else if let Some(target) = target.ancestor(Node::static_type()) {
+                    // The user targeted a Node without targeting a specific Port.
+                    // Drag the Node around the screen.
+                    let (x, y) = widget.get_node_position(&target);
+                    Some((target, x, y))
+                } else {
+                    None
+                }
+            }));
+            drag_controller.connect_drag_update(
+                clone!(@strong drag_state => move |drag_controller, x, y| {
+                    let widget = drag_controller
+                        .widget()
+                        .expect("drag-update event has no widget")
+                        .dynamic_cast::<Self::Type>()
+                        .expect("drag-update event is not on the GraphView");
+                    let drag_state = drag_state.borrow();
+                    if let Some((ref node, x1, y1)) = *drag_state {
+                        widget.move_node(node, x1 + x as f32, y1 + y as f32);
                     }
-                };
-            });
-            obj.add_controller(&motion_controller);
+                }
+                ),
+            );
+            obj.add_controller(&drag_controller);
         }
 
         fn dispose(&self, _obj: &Self::Type) {
@@ -81,9 +96,9 @@ mod imp {
             /* FIXME: A lot of hardcoded values in here.
             Try to use relative units (em) and colours from the theme as much as possible. */
 
-            let alloc = widget.get_allocation();
+            let alloc = widget.allocation();
 
-            let cr = snapshot
+            let background_cr = snapshot
                 .append_cairo(&graphene::Rect::new(
                     0.0,
                     0.0,
@@ -93,46 +108,80 @@ mod imp {
                 .expect("Failed to get cairo context");
 
             // Try to replace the background color with a darker one from the theme.
-            if let Some(rgba) = widget.get_style_context().lookup_color("text_view_bg") {
-                cr.set_source_rgb(rgba.red.into(), rgba.green.into(), rgba.blue.into());
-                cr.paint();
+            if let Some(rgba) = widget.style_context().lookup_color("text_view_bg") {
+                background_cr.set_source_rgb(rgba.red.into(), rgba.green.into(), rgba.blue.into());
+                if let Err(e) = background_cr.paint() {
+                    warn!("Failed to paint graphview background: {}", e);
+                };
             } // TODO: else log colour not found
 
             // Draw a nice grid on the background.
-            cr.set_source_rgb(0.18, 0.18, 0.18);
-            cr.set_line_width(0.2); // TODO: Set to 1px
+            background_cr.set_source_rgb(0.18, 0.18, 0.18);
+            background_cr.set_line_width(0.2); // TODO: Set to 1px
             let mut y = 0.0;
             while y < alloc.height.into() {
-                cr.move_to(0.0, y);
-                cr.line_to(alloc.width as f64, y);
+                background_cr.move_to(0.0, y);
+                background_cr.line_to(alloc.width.into(), y);
                 y += 20.0; // TODO: Change to em;
             }
             let mut x = 0.0;
-            while x < alloc.width as f64 {
-                cr.move_to(x, 0.0);
-                cr.line_to(x, alloc.height as f64);
+            while x < alloc.width.into() {
+                background_cr.move_to(x, 0.0);
+                background_cr.line_to(x, alloc.height.into());
                 x += 20.0; // TODO: Change to em;
             }
-            cr.stroke();
-
-            // Draw all links
-            cr.set_line_width(2.0);
-            cr.set_source_rgb(0.0, 0.0, 0.0);
-            for link in self.links.borrow().values() {
-                if let Some((from_x, from_y, to_x, to_y)) = self.get_link_coordinates(link) {
-                    cr.move_to(from_x, from_y);
-                    cr.curve_to(from_x + 75.0, from_y, to_x - 75.0, to_y, to_x, to_y);
-                    cr.stroke();
-                } else {
-                    log::warn!("Could not get allocation of ports of link: {:?}", link);
-                }
-            }
+            if let Err(e) = background_cr.stroke() {
+                warn!("Failed to draw graphview grid: {}", e);
+            };
 
             // Draw all children
             self.nodes
                 .borrow()
                 .values()
-                .for_each(|node| self.get_instance().snapshot_child(node, snapshot));
+                .for_each(|node| self.instance().snapshot_child(node, snapshot));
+
+            // Draw all links
+            let link_cr = snapshot
+                .append_cairo(&graphene::Rect::new(
+                    0.0,
+                    0.0,
+                    alloc.width as f32,
+                    alloc.height as f32,
+                ))
+                .expect("Failed to get cairo context");
+            link_cr.set_line_width(2.0);
+            link_cr.set_source_rgb(0.0, 0.0, 0.0);
+            for (link, active) in self.links.borrow().values() {
+                if let Some((from_x, from_y, to_x, to_y)) = self.get_link_coordinates(link) {
+                    link_cr.move_to(from_x, from_y);
+
+                    // Use dashed line for inactive links, full line otherwise.
+                    if *active {
+                        link_cr.set_dash(&[], 0.0);
+                    } else {
+                        link_cr.set_dash(&[10.0, 5.0], 0.0);
+                    }
+
+                    // Place curve control offset by half the x distance between the two points.
+                    // This makes the curve scale well for varying distances between the two ports,
+                    // especially when the output port is farther right than the input port.
+                    let half_x_dist = f64::abs(from_x - to_x) / 2.0;
+                    link_cr.curve_to(
+                        from_x + half_x_dist,
+                        from_y,
+                        to_x - half_x_dist,
+                        to_y,
+                        to_x,
+                        to_y,
+                    );
+
+                    if let Err(e) = link_cr.stroke() {
+                        warn!("Failed to draw graphview links: {}", e);
+                    };
+                } else {
+                    warn!("Could not get allocation of ports of link: {:?}", link);
+                }
+            }
         }
     }
 
@@ -140,42 +189,42 @@ mod imp {
         /// Get coordinates for the drawn link to start at and to end at.
         ///
         /// # Returns
-        /// Some((from_x, from_y, to_x, to_y)) if all objects the links refers to exist as widgets.
+        /// `Some((from_x, from_y, to_x, to_y))` if all objects the links refers to exist as widgets.
         fn get_link_coordinates(&self, link: &crate::PipewireLink) -> Option<(f64, f64, f64, f64)> {
             let nodes = self.nodes.borrow();
 
             // For some reason, gtk4::WidgetExt::translate_coordinates gives me incorrect values,
             // so we manually calculate the needed offsets here.
 
-            let from_port = &nodes.get(&link.node_from)?.get_port(link.port_from)?.widget;
+            let from_port = &nodes.get(&link.node_from)?.get_port(link.port_from)?;
             let gtk::Allocation {
                 x: mut fx,
                 y: mut fy,
                 width: fw,
                 height: fh,
-            } = from_port.get_allocation();
+            } = from_port.allocation();
             let from_node = from_port
-                .get_ancestor(Node::static_type())
+                .ancestor(Node::static_type())
                 .expect("Port is not a child of a node");
-            let gtk::Allocation { x: fnx, y: fny, .. } = from_node.get_allocation();
+            let gtk::Allocation { x: fnx, y: fny, .. } = from_node.allocation();
             fx += fnx + fw;
             fy += fny + (fh / 2);
 
-            let to_port = &nodes.get(&link.node_to)?.get_port(link.port_to)?.widget;
+            let to_port = &nodes.get(&link.node_to)?.get_port(link.port_to)?;
             let gtk::Allocation {
                 x: mut tx,
                 y: mut ty,
                 height: th,
                 ..
-            } = to_port.get_allocation();
+            } = to_port.allocation();
             let to_node = to_port
-                .get_ancestor(Node::static_type())
+                .ancestor(Node::static_type())
                 .expect("Port is not a child of a node");
-            let gtk::Allocation { x: tnx, y: tny, .. } = to_node.get_allocation();
+            let gtk::Allocation { x: tnx, y: tny, .. } = to_node.allocation();
             tx += tnx;
             ty += tny + (th / 2);
 
-            Some((fx as f64, fy as f64, tx as f64, ty as f64))
+            Some((fx.into(), fy.into(), tx.into(), ty.into()))
         }
     }
 }
@@ -196,8 +245,8 @@ impl GraphView {
 
         // Place widgets in colums of 4, growing down, then right.
         // TODO: Make a better positioning algorithm.
-        let x = (private.nodes.borrow().len() / 4) as f32 * 400.0; // This relies on integer division rounding down.
-        let y = private.nodes.borrow().len() as f32 % 4.0 * 100.0;
+        let x = ((private.nodes.borrow().len() / 4) as f32 * 400.0) + 20.0; // This relies on integer division rounding down.
+        let y = (private.nodes.borrow().len() as f32 % 4.0 * 100.0) + 20.0;
 
         self.move_node(&node.clone().upcast(), x, y);
 
@@ -209,34 +258,48 @@ impl GraphView {
         let mut nodes = private.nodes.borrow_mut();
         if let Some(node) = nodes.remove(&id) {
             node.unparent();
+        } else {
+            warn!("Tried to remove non-existant node (id={}) from graph", id);
         }
     }
 
-    pub fn add_port_to_node(&self, node_id: u32, port_id: u32, port: crate::view::port::Port) {
+    pub fn add_port(&self, node_id: u32, port_id: u32, port: crate::view::port::Port) {
         let private = imp::GraphView::from_instance(self);
 
         if let Some(node) = private.nodes.borrow_mut().get_mut(&node_id) {
             node.add_port(port_id, port);
         } else {
-            // FIXME: Log this instead
-            log::error!(
+            error!(
                 "Node with id {} not found when trying to add port with id {} to graph",
-                node_id,
-                port_id
+                node_id, port_id
             );
         }
     }
 
-    /// Add a link to the graph.
-    ///
-    /// `add_link` takes three arguments: `link_id` is the id of the link as assigned by the pipewire server,
-    /// `from` and `to` are the id's of the ingoing and outgoing port, respectively.
-    pub fn add_link(&self, link_id: u32, link: crate::PipewireLink) {
+    pub fn remove_port(&self, id: u32, node_id: u32) {
         let private = imp::GraphView::from_instance(self);
-        private.links.borrow_mut().insert(link_id, link);
+        let nodes = private.nodes.borrow();
+        if let Some(node) = nodes.get(&node_id) {
+            node.remove_port(id);
+        }
+    }
+
+    pub fn add_link(&self, link_id: u32, link: crate::PipewireLink, active: bool) {
+        let private = imp::GraphView::from_instance(self);
+        private.links.borrow_mut().insert(link_id, (link, active));
         self.queue_draw();
     }
 
+    pub fn set_link_state(&self, link_id: u32, active: bool) {
+        let private = imp::GraphView::from_instance(self);
+        if let Some((_, state)) = private.links.borrow_mut().get_mut(&link_id) {
+            *state = active;
+            self.queue_draw();
+        } else {
+            warn!("Link state changed on unknown link (id={})", link_id);
+        }
+    }
+
     pub fn remove_link(&self, id: u32) {
         let private = imp::GraphView::from_instance(self);
         let mut links = private.links.borrow_mut();
@@ -245,23 +308,36 @@ impl GraphView {
         self.queue_draw();
     }
 
-    pub fn set_dragged(&self, widget: Option<gtk::Widget>) {
-        *imp::GraphView::from_instance(self).dragged.borrow_mut() = widget;
+    pub(super) fn get_node_position(&self, node: &gtk::Widget) -> (f32, f32) {
+        let layout_manager = self
+            .layout_manager()
+            .expect("Failed to get layout manager")
+            .dynamic_cast::<gtk::FixedLayout>()
+            .expect("Failed to cast to FixedLayout");
+
+        let node = layout_manager
+            .layout_child(node)
+            .expect("Could not get layout child")
+            .dynamic_cast::<gtk::FixedLayoutChild>()
+            .expect("Could not cast to FixedLayoutChild");
+        let transform = node.transform().unwrap_or_default();
+        transform.to_translate()
     }
 
-    pub fn move_node(&self, node: &gtk::Widget, x: f32, y: f32) {
+    pub(super) fn move_node(&self, node: &gtk::Widget, x: f32, y: f32) {
         let layout_manager = self
-            .get_layout_manager()
+            .layout_manager()
             .expect("Failed to get layout manager")
             .dynamic_cast::<gtk::FixedLayout>()
             .expect("Failed to cast to FixedLayout");
 
         let transform = gsk::Transform::new()
-            .translate(&graphene::Point::new(x, y))
+            // Nodes should not be able to be dragged out of the view, so we use `max(coordinate, 0.0)` to prevent that.
+            .translate(&graphene::Point::new(f32::max(x, 0.0), f32::max(y, 0.0)))
             .unwrap();
 
         layout_manager
-            .get_layout_child(node)
+            .layout_child(node)
             .expect("Could not get layout child")
             .dynamic_cast::<gtk::FixedLayoutChild>()
             .expect("Could not cast to FixedLayoutChild")
@@ -272,3 +348,9 @@ impl GraphView {
         self.queue_draw();
     }
 }
+
+impl Default for GraphView {
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/view/mod.rs

@@ -1,6 +1,11 @@
+//! The view presented to the user.
+//!
+//! This module contains gtk widgets needed to present the graphical user interface.
+
 mod graph_view;
 mod node;
-pub mod port;
+mod port;
 
 pub use graph_view::GraphView;
 pub use node::Node;
+pub use port::Port;

src/view/node.rs

@@ -1,9 +1,7 @@
-use super::graph_view::GraphView;
+use gtk::{glib, prelude::*, subclass::prelude::*};
+use pipewire::spa::Direction;
 
-use gtk::{glib, prelude::*, subclass::prelude::*, WidgetExt};
-use pipewire::port::Direction;
-
-use std::{collections::HashMap, rc::Rc};
+use std::collections::HashMap;
 
 mod imp {
     use super::*;
@@ -13,19 +11,16 @@ mod imp {
     pub struct Node {
         pub(super) grid: gtk::Grid,
         pub(super) label: gtk::Label,
-        pub(super) ports: RefCell<HashMap<u32, Rc<crate::view::port::Port>>>,
-        pub(super) num_ports_in: Cell<u32>,
-        pub(super) num_ports_out: Cell<u32>,
+        pub(super) ports: RefCell<HashMap<u32, crate::view::port::Port>>,
+        pub(super) num_ports_in: Cell<i32>,
+        pub(super) num_ports_out: Cell<i32>,
     }
 
+    #[glib::object_subclass]
     impl ObjectSubclass for Node {
         const NAME: &'static str = "Node";
         type Type = super::Node;
         type ParentType = gtk::Widget;
-        type Instance = glib::subclass::simple::InstanceStruct<Self>;
-        type Class = glib::subclass::simple::ClassStruct<Self>;
-
-        glib::object_subclass!();
 
         fn class_init(klass: &mut Self::Class) {
             klass.set_layout_manager_type::<gtk::BinLayout>();
@@ -37,35 +32,6 @@ mod imp {
 
             grid.attach(&label, 0, 0, 2, 1);
 
-            let motion_controller = gtk::EventControllerMotion::new();
-            motion_controller.connect_enter(|controller, _, _| {
-                // Tell the graphview that the Node is the target of a drag when the mouse enters its label
-                let widget = controller
-                    .get_widget()
-                    .expect("Controller with enter event has no widget")
-                    .get_ancestor(super::Node::static_type())
-                    .expect("Node label does not have a node ancestor widget");
-                widget
-                    .get_ancestor(GraphView::static_type())
-                    .expect("Node with enter event is not on graph")
-                    .dynamic_cast::<GraphView>()
-                    .unwrap()
-                    .set_dragged(Some(widget));
-            });
-            motion_controller.connect_leave(|controller| {
-                // Tell the graphview that the Node is no longer the target of a drag when the mouse leaves.
-                // FIXME: Check that we are the current target before setting none.
-                controller
-                    .get_widget()
-                    .expect("Controller with leave event has no widget")
-                    .get_ancestor(GraphView::static_type())
-                    .expect("Node with leave event is not on graph")
-                    .dynamic_cast::<GraphView>()
-                    .unwrap()
-                    .set_dragged(None);
-            });
-            label.add_controller(&motion_controller);
-
             // Display a grab cursor when the mouse is over the label so the user knows the node can be dragged.
             label.set_cursor(gtk::gdk::Cursor::from_name("grab", None).as_ref());
 
@@ -111,46 +77,38 @@ impl Node {
     pub fn add_port(&mut self, id: u32, port: super::port::Port) {
         let private = imp::Node::from_instance(self);
 
-        match port.direction {
+        match port.direction() {
             Direction::Input => {
                 private
                     .grid
-                    .attach(&port.widget, 0, private.num_ports_in.get() as i32 + 1, 1, 1);
+                    .attach(&port, 0, private.num_ports_in.get() + 1, 1, 1);
                 private.num_ports_in.set(private.num_ports_in.get() + 1);
             }
             Direction::Output => {
-                private.grid.attach(
-                    &port.widget,
-                    1,
-                    private.num_ports_out.get() as i32 + 1,
-                    1,
-                    1,
-                );
+                private
+                    .grid
+                    .attach(&port, 1, private.num_ports_out.get() + 1, 1, 1);
                 private.num_ports_out.set(private.num_ports_out.get() + 1);
             }
         }
 
-        private.ports.borrow_mut().insert(id, Rc::new(port));
+        private.ports.borrow_mut().insert(id, port);
     }
 
-    pub fn get_port(&self, id: u32) -> Option<Rc<super::port::Port>> {
+    pub fn get_port(&self, id: u32) -> Option<super::port::Port> {
         let private = imp::Node::from_instance(self);
-        private
-            .ports
-            .borrow_mut()
-            .get(&id)
-            .map(|port_rc| port_rc.clone())
+        private.ports.borrow_mut().get(&id).cloned()
     }
 
     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 {
+            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();
+            port.unparent();
         }
     }
 }

src/view/port.rs

@@ -1,16 +1,194 @@
-/// Graphical representation of a pipewire port.
-pub struct Port {
-    pub(super) widget: gtk::Button,
-    pub id: u32,
-    pub direction: pipewire::port::Direction,
+use gtk::{
+    gdk,
+    glib::{self, clone, subclass::Signal},
+    prelude::*,
+    subclass::prelude::*,
+};
+use log::{trace, warn};
+use pipewire::spa::Direction;
+
+use crate::MediaType;
+
+/// A helper struct for linking a output port to an input port.
+/// It carries the output ports id.
+#[derive(Clone, Debug, glib::GBoxed)]
+#[gboxed(type_name = "HelvumForwardLink")]
+struct ForwardLink(u32);
+
+/// A helper struct for linking an input to an output port.
+/// It carries the input ports id.
+#[derive(Clone, Debug, glib::GBoxed)]
+#[gboxed(type_name = "HelvumReversedLink")]
+struct ReversedLink(u32);
+
+mod imp {
+    use once_cell::{sync::Lazy, unsync::OnceCell};
+    use pipewire::spa::Direction;
+
+    use super::*;
+
+    /// Graphical representation of a pipewire port.
+    #[derive(Default)]
+    pub struct Port {
+        pub(super) label: OnceCell<gtk::Label>,
+        pub(super) id: OnceCell<u32>,
+        pub(super) direction: OnceCell<Direction>,
+    }
+
+    #[glib::object_subclass]
+    impl ObjectSubclass for Port {
+        const NAME: &'static str = "Port";
+        type Type = super::Port;
+        type ParentType = gtk::Widget;
+
+        fn class_init(klass: &mut Self::Class) {
+            klass.set_layout_manager_type::<gtk::BinLayout>();
+
+            // Make it look like a GTK button.
+            klass.set_css_name("button");
+        }
+    }
+
+    impl ObjectImpl for Port {
+        fn dispose(&self, _obj: &Self::Type) {
+            if let Some(label) = self.label.get() {
+                label.unparent()
+            }
+        }
+
+        fn signals() -> &'static [Signal] {
+            static SIGNALS: Lazy<Vec<Signal>> = Lazy::new(|| {
+                vec![Signal::builder(
+                    "port-toggled",
+                    // Provide id of output port and input port to signal handler.
+                    &[<u32>::static_type().into(), <u32>::static_type().into()],
+                    // signal handler sends back nothing.
+                    <()>::static_type().into(),
+                )
+                .build()]
+            });
+
+            SIGNALS.as_ref()
+        }
+    }
+    impl WidgetImpl for Port {}
+}
+
+glib::wrapper! {
+    pub struct Port(ObjectSubclass<imp::Port>)
+        @extends gtk::Widget;
 }
 
 impl Port {
-    pub fn new(id: u32, name: &str, direction: pipewire::port::Direction) -> Self {
-        Self {
-            widget: gtk::Button::with_label(name),
-            id,
-            direction,
+    pub fn new(id: u32, name: &str, direction: Direction, media_type: Option<MediaType>) -> Self {
+        // Create the widget and initialize needed fields
+        let res: Self = glib::Object::new(&[]).expect("Failed to create Port");
+
+        let private = imp::Port::from_instance(&res);
+        private.id.set(id).expect("Port id already set");
+        private
+            .direction
+            .set(direction)
+            .expect("Port direction already set");
+
+        let label = gtk::Label::new(Some(name));
+        label.set_parent(&res);
+        private
+            .label
+            .set(label)
+            .expect("Port label was already set");
+
+        // Add a drag source and drop target controller with the type depending on direction,
+        // they will be responsible for link creation by dragging an output port onto an input port or the other way around.
+
+        // FIXME: We should protect against different media types, e.g. it should not be possible to drop a video port on an audio port.
+
+        // The port will simply provide its pipewire id to the drag target.
+        let drag_src = gtk::DragSourceBuilder::new()
+            .content(&gdk::ContentProvider::for_value(&match direction {
+                Direction::Input => ReversedLink(id).to_value(),
+                Direction::Output => ForwardLink(id).to_value(),
+            }))
+            .build();
+        drag_src.connect_drag_begin(move |_, _| {
+            trace!("Drag started from port {}", id);
+        });
+        drag_src.connect_drag_cancel(move |_, _, _| {
+            trace!("Drag from port {} was cancelled", id);
+            false
+        });
+        res.add_controller(&drag_src);
+
+        // The drop target will accept either a `ForwardLink` or `ReversedLink` depending in its own direction,
+        // and use it to emit its `port-toggled` signal.
+        let drop_target = gtk::DropTarget::new(
+            match direction {
+                Direction::Input => ForwardLink::static_type(),
+                Direction::Output => ReversedLink::static_type(),
+            },
+            gdk::DragAction::COPY,
+        );
+        match direction {
+            Direction::Input => {
+                drop_target.connect_drop(
+                    clone!(@weak res as this => @default-panic, move |drop_target, val, _, _| {
+                        if let Ok(ForwardLink(source_id)) = val.get::<ForwardLink>() {
+                            // Get the callback registered in the widget and call it
+                            drop_target
+                                .widget()
+                                .expect("Drop target has no widget")
+                                .emit_by_name("port-toggled", &[&source_id, &this.id()])
+                                .expect("Failed to send signal");
+                        } else {
+                            warn!("Invalid type dropped on ingoing port");
+                        }
+
+                        true
+                    }),
+                );
+            }
+            Direction::Output => {
+                drop_target.connect_drop(
+                    clone!(@weak res as this => @default-panic, move |drop_target, val, _, _| {
+                        if let Ok(ReversedLink(target_id)) = val.get::<ReversedLink>() {
+                            // Get the callback registered in the widget and call it
+                            drop_target
+                                .widget()
+                                .expect("Drop target has no widget")
+                                .emit_by_name("port-toggled", &[&this.id(), &target_id])
+                                .expect("Failed to send signal");
+                        } else {
+                            warn!("Invalid type dropped on outgoing port");
+                        }
+
+                        true
+                    }),
+                );
+            }
         }
+        res.add_controller(&drop_target);
+
+        // Display a grab cursor when the mouse is over the port so the user knows it can be dragged to another port.
+        res.set_cursor(gtk::gdk::Cursor::from_name("grab", None).as_ref());
+
+        // Color the port according to its media type.
+        match media_type {
+            Some(MediaType::Video) => res.add_css_class("video"),
+            Some(MediaType::Audio) => res.add_css_class("audio"),
+            Some(MediaType::Midi) => res.add_css_class("midi"),
+            None => {}
+        }
+
+        res
+    }
+
+    pub fn id(&self) -> u32 {
+        let private = imp::Port::from_instance(self);
+        private.id.get().copied().expect("Port id is not set")
+    }
+
+    pub fn direction(&self) -> &Direction {
+        let private = imp::Port::from_instance(self);
+        private.direction.get().expect("Port direction is not set")
     }
 }