Release 0.2.0

This lets the user know the port can be dragged to create a link.

.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.2.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 = { git = "https://gitlab.freedesktop.org/pipewire/pipewire-rs", branch = "main" }
 
 log = "0.4.11"
 env_logger = "0.8.2"
+
+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,28 @@
-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)
 
 # 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!
 
+# Distribution packages
+
+- ArchLinux: [aur/helvum-git](https://aur.archlinux.org/packages/helvum-git)
+
 # Building
 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`.

docs/architecture.md

@@ -0,0 +1,55 @@
+# 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
+ │   ┆
+ │   ┆<─ notifies of user input
+ │   ┆    (using signals)
+ │   ┆
+ │   ┆
+ │   V           notifies of remote changes
+┌┴────────────┐         via messages          ┌───────────────────┐
+│ Application │<╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤     Seperate      │
+│   Object    ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌>│  Pipewire Thread  │
+└┬────────────┘   request changes to remote   └───────────────────┘
+ │                      via messages                    Λ
+ │                                                      ║
+ │<─── updates/reads state                              ║
+ │                                                      ║
+ V                                                      ║
+┌───────┐                                               V
+│ State │                                   [ Remote Pipewire Server ]
+└───────┘
+```
+The program is split between two threads, with most stuff happening inside the GTK thread.
+The GTK thread will sit in a GTK event processing loop, while the pipewire thread will sit in a
+pipewire event processing loop.
+
+The `Application` object inside the GTK thread is the centerpiece of this architecture.
+It 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 `Application` in the GTK thread is notified by the pipewire thread
+and updates the view to reflect those changes, and additionally memorizes anything it might need later in the state.
+
+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 request 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

src/application.rs

@@ -0,0 +1,392 @@
+use std::{cell::RefCell, collections::HashMap};
+
+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, PipewireLink, PipewireMessage,
+};
+
+#[derive(Debug, Copy, Clone)]
+pub enum MediaType {
+    Audio,
+    Video,
+    Midi,
+}
+
+// FIXME: This should be in its own .css file.
+static STYLE: &str = "
+.audio {
+    background: rgb(50,100,240);
+	color: black;
+}
+
+.video {
+    background: rgb(200,200,0);
+	color: black;
+}
+
+.midi {
+    background: rgb(200,0,50);
+    color: black;
+}
+";
+
+mod imp {
+    use super::*;
+
+    use once_cell::unsync::OnceCell;
+
+    #[derive(Default)]
+    pub struct Application {
+        pub(super) graphview: view::GraphView,
+        pub(super) state: RefCell<State>,
+        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,
+                            media_type,
+                        } => app.add_node(id, name, media_type),
+                        PipewireMessage::PortAdded {
+                            id,
+                            node_id,
+                            name,
+                            direction,
+                        } => app.add_port(id, name, node_id, direction),
+                        PipewireMessage::LinkAdded { id, link } => app.add_link(id, link),
+                        PipewireMessage::ObjectRemoved { id } => app.remove_global(id),
+                    };
+                    Continue(true)
+                }
+            ),
+        );
+
+        app
+    }
+
+    /// Add a new node to the view.
+    pub fn add_node(&self, id: u32, name: String, media_type: Option<MediaType>) {
+        info!("Adding node to graph: id {}", id);
+
+        let imp = imp::Application::from_instance(self);
+
+        imp.state.borrow_mut().insert(
+            id,
+            Item::Node {
+                // widget: node_widget,
+                media_type,
+            },
+        );
+
+        imp.graphview.add_node(id, view::Node::new(name.as_str()));
+    }
+
+    /// Add a new port to the view.
+    pub fn add_port(&self, id: u32, name: String, node_id: u32, direction: Direction) {
+        info!("Adding port to graph: id {}", id);
+
+        let imp = imp::Application::from_instance(self);
+
+        // 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) {
+                media_type.to_owned()
+            } else {
+                warn!("Node not found for Port {}", id);
+                None
+            };
+
+        // Save node_id so we can delete this port easily.
+        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::<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.
+    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 {
+                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) {
+            match item {
+                Item::Node { .. } => self.remove_node(id),
+                Item::Port { node_id } => self.remove_port(id, node_id),
+                Item::Link { .. } => self.remove_link(id),
+            }
+        } else {
+            warn!(
+                "Attempted to remove item with id {} that is not saved in state",
+                id
+            );
+        }
+    }
+
+    /// 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);
+    }
+}
+
+/// 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

@@ -1,12 +1,48 @@
+mod application;
 mod pipewire_connection;
-mod pipewire_state;
 mod view;
 
-use gtk::prelude::*;
+use application::MediaType;
+use gtk::{
+    glib::{self, PRIORITY_DEFAULT},
+    prelude::*,
+};
+use pipewire::spa::Direction;
 
-use std::{cell::RefCell, rc::Rc};
+/// Messages used GTK thread to command the pipewire thread.
+#[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,
+}
 
-#[derive(Debug)]
+/// Messages used pipewire thread to notify the GTK thread.
+#[derive(Debug, Clone)]
+enum PipewireMessage {
+    /// A new node has appeared.
+    NodeAdded {
+        id: u32,
+        name: String,
+        media_type: Option<MediaType>,
+    },
+    /// A new port has appeared.
+    PortAdded {
+        id: u32,
+        node_id: u32,
+        name: String,
+        direction: Direction,
+    },
+    /// A new link has appeared.
+    LinkAdded { id: u32, link: PipewireLink },
+    /// An object was removed
+    ObjectRemoved { id: u32 },
+}
+
+#[derive(Debug, Clone)]
 pub struct PipewireLink {
     pub node_from: u32,
     pub port_from: u32,
@@ -18,42 +54,21 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     env_logger::init();
     gtk::init()?;
 
-    let graphview = Rc::new(RefCell::new(view::GraphView::new()));
+    // 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));
 
-    // 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)
-    });
+    let app = application::Application::new(gtk_receiver, pw_sender.clone());
 
-    let app = gtk::Application::new(Some("org.freedesktop.pipewire.graphui"), Default::default())
-        .expect("Application creation failed");
+    app.run();
 
-    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();
-    });
+    pw_sender
+        .send(GtkMessage::Terminate)
+        .expect("Failed to send message");
 
-    app.run(&std::env::args().collect::<Vec<_>>());
+    pw_thread.join().expect("Pipewire thread panicked");
 
     Ok(())
 }

src/pipewire_connection.rs

@@ -1,84 +1,182 @@
-use crate::pipewire_state::PipewireState;
+use std::rc::Rc;
 
 use gtk::glib::{self, clone};
-use pipewire as pw;
-
-use std::{
-    cell::{Cell, RefCell},
-    rc::Rc,
+use log::warn;
+use pipewire::{
+    link::Link,
+    prelude::*,
+    properties,
+    registry::GlobalObject,
+    spa::{Direction, ForeignDict},
+    types::ObjectType,
+    Context, 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::{application::MediaType, GtkMessage, PipewireMessage};
 
-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 = 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));
-
-        // 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();
-
-        Ok(Self {
-            mainloop: mainloop,
-            _context: context,
-            core,
-            _registry: registry,
-            _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);
-
-        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 _receiver = pw_receiver.attach(&mainloop, {
+        let mainloop = mainloop.clone();
+        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);
                 }
-            })
-            .register();
+            }
+            GtkMessage::DestroyGlobal(id) => {
+                // FIXME: Handle error
+                registry.destroy_global(id);
+            }
+            GtkMessage::Terminate => mainloop.quit(),
+        })
+    });
 
-        while !done.get() {
-            self.mainloop.run();
-        }
-    }
+    let _listener = registry
+        .add_listener_local()
+        .global({
+            let sender = gtk_sender.clone();
+            move |global| match global.type_ {
+                ObjectType::Node => handle_node(global, &sender),
+                ObjectType::Port => handle_port(global, &sender),
+                ObjectType::Link => handle_link(global, &sender),
+                _ => {
+                    // Other objects are not interesting to us
+                }
+            }
+        })
+        .global_remove(move |id| {
+            gtk_sender
+                .send(PipewireMessage::ObjectRemoved { id })
+                .expect("Failed to send message")
+        })
+        .register();
+
+    mainloop.run();
+}
+
+/// Handle a new node being added
+fn handle_node(node: &GlobalObject<ForeignDict>, sender: &glib::Sender<PipewireMessage>) {
+    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: This relies on the node being passed to us by the pipwire server before its port.
+    let media_type = props
+        .get("media.class")
+        .map(|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
+            }
+        })
+        .flatten();
+
+    sender
+        .send(PipewireMessage::NodeAdded {
+            id: node.id,
+            name,
+            media_type,
+        })
+        .expect("Failed to send message");
+}
+
+/// Handle a new port being added
+fn handle_port(port: &GlobalObject<ForeignDict>, sender: &glib::Sender<PipewireMessage>) {
+    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
+    };
+
+    sender
+        .send(PipewireMessage::PortAdded {
+            id: port.id,
+            node_id,
+            name,
+            direction,
+        })
+        .expect("Failed to send message");
+}
+
+/// Handle a new link being added
+fn handle_link(link: &GlobalObject<ForeignDict>, sender: &glib::Sender<PipewireMessage>) {
+    let props = link
+        .props
+        .as_ref()
+        .expect("Link object is missing properties");
+    let node_from: u32 = props
+        .get("link.output.node")
+        .expect("Link has no link.input.node property")
+        .parse()
+        .expect("Could not parse link.input.node property");
+    let port_from: u32 = props
+        .get("link.output.port")
+        .expect("Link has no link.output.port property")
+        .parse()
+        .expect("Could not parse link.output.port property");
+    let node_to: u32 = props
+        .get("link.input.node")
+        .expect("Link has no link.input.node property")
+        .parse()
+        .expect("Could not parse link.input.node property");
+    let port_to: u32 = props
+        .get("link.input.port")
+        .expect("Link has no link.input.port property")
+        .parse()
+        .expect("Could not parse link.input.port property");
+
+    sender
+        .send(PipewireMessage::LinkAdded {
+            id: link.id,
+            link: crate::PipewireLink {
+                node_from,
+                port_from,
+                node_to,
+                port_to,
+            },
+        })
+        .expect("Failed to send message");
 }

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/view/graph_view.rs

@@ -1,43 +1,33 @@
 use super::Node;
 
-use gtk::{glib, graphene, gsk, prelude::*, subclass::prelude::*, WidgetExt};
+use gtk::{gdk, glib, graphene, gsk, prelude::*, subclass::prelude::*};
 
 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>>>,
     }
 
+    #[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 {
@@ -48,7 +38,7 @@ mod imp {
             let motion_controller = gtk::EventControllerMotion::new();
             motion_controller.connect_motion(|controller, x, y| {
                 let instance = controller
-                    .get_widget()
+                    .widget()
                     .unwrap()
                     .dynamic_cast::<Self::Type>()
                     .unwrap();
@@ -56,9 +46,9 @@ mod imp {
 
                 if let Some(ref widget) = *this.dragged.borrow() {
                     if controller
-                        .get_current_event()
+                        .current_event()
                         .unwrap()
-                        .get_modifier_state()
+                        .modifier_state()
                         .contains(gdk::ModifierType::BUTTON1_MASK)
                     {
                         instance.move_node(&widget, x as f32, y as f32);
@@ -81,7 +71,7 @@ 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
                 .append_cairo(&graphene::Rect::new(
@@ -93,9 +83,11 @@ 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") {
+            if let Some(rgba) = widget.style_context().lookup_color("text_view_bg") {
                 cr.set_source_rgb(rgba.red.into(), rgba.green.into(), rgba.blue.into());
-                cr.paint();
+                if let Err(e) = cr.paint() {
+                    warn!("Failed to paint graphview background: {}", e);
+                };
             } // TODO: else log colour not found
 
             // Draw a nice grid on the background.
@@ -113,7 +105,9 @@ mod imp {
                 cr.line_to(x, alloc.height as f64);
                 x += 20.0; // TODO: Change to em;
             }
-            cr.stroke();
+            if let Err(e) = cr.stroke() {
+                warn!("Failed to draw graphview grid: {}", e);
+            };
 
             // Draw all links
             cr.set_line_width(2.0);
@@ -122,7 +116,9 @@ mod imp {
                 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();
+                    if let Err(e) = cr.stroke() {
+                        warn!("Failed to draw graphview links: {}", e);
+                    };
                 } else {
                     log::warn!("Could not get allocation of ports of link: {:?}", link);
                 }
@@ -132,7 +128,7 @@ mod imp {
             self.nodes
                 .borrow()
                 .values()
-                .for_each(|node| self.get_instance().snapshot_child(node, snapshot));
+                .for_each(|node| self.instance().snapshot_child(node, snapshot));
         }
     }
 
@@ -147,31 +143,31 @@ mod imp {
             // 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);
 
@@ -212,7 +208,7 @@ impl GraphView {
         }
     }
 
-    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) {
@@ -227,6 +223,14 @@ impl GraphView {
         }
     }
 
+    pub fn remove_port(&self, id: u32, node_id: u32) {
+        let private = imp::GraphView::from_instance(self);
+        let nodes = private.nodes.borrow();
+        if let Some(node) = nodes.get(&node_id) {
+            node.remove_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,
@@ -245,23 +249,24 @@ impl GraphView {
         self.queue_draw();
     }
 
-    pub fn set_dragged(&self, widget: Option<gtk::Widget>) {
+    pub(super) fn set_dragged(&self, widget: Option<gtk::Widget>) {
         *imp::GraphView::from_instance(self).dragged.borrow_mut() = widget;
     }
 
-    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 +277,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,7 +1,7 @@
 use super::graph_view::GraphView;
 
-use gtk::{glib, prelude::*, subclass::prelude::*, WidgetExt};
-use pipewire::port::Direction;
+use gtk::{glib, prelude::*, subclass::prelude::*};
+use pipewire::spa::Direction;
 
 use std::{collections::HashMap, rc::Rc};
 
@@ -18,14 +18,11 @@ mod imp {
         pub(super) num_ports_out: Cell<u32>,
     }
 
+    #[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>();
@@ -41,12 +38,12 @@ mod imp {
             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()
+                    .widget()
                     .expect("Controller with enter event has no widget")
-                    .get_ancestor(super::Node::static_type())
+                    .ancestor(super::Node::static_type())
                     .expect("Node label does not have a node ancestor widget");
                 widget
-                    .get_ancestor(GraphView::static_type())
+                    .ancestor(GraphView::static_type())
                     .expect("Node with enter event is not on graph")
                     .dynamic_cast::<GraphView>()
                     .unwrap()
@@ -56,9 +53,9 @@ mod imp {
                 // 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()
+                    .widget()
                     .expect("Controller with leave event has no widget")
-                    .get_ancestor(GraphView::static_type())
+                    .ancestor(GraphView::static_type())
                     .expect("Node with leave event is not on graph")
                     .dynamic_cast::<GraphView>()
                     .unwrap()
@@ -111,21 +108,17 @@ 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() as i32 + 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() as i32 + 1, 1, 1);
                 private.num_ports_out.set(private.num_ports_out.get() + 1);
             }
         }
@@ -135,22 +128,18 @@ impl Node {
 
     pub fn get_port(&self, id: u32) -> Option<Rc<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,130 @@
-/// 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, subclass::Signal},
+    prelude::*,
+    subclass::prelude::*,
+};
+use log::warn;
+use pipewire::spa::Direction;
+
+use crate::application::MediaType;
+
+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) 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::Button;
+    }
+
+    impl ObjectImpl for Port {
+        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 {}
+    impl ButtonImpl for Port {}
+}
+
+glib::wrapper! {
+    pub struct Port(ObjectSubclass<imp::Port>)
+        @extends gtk::Button, 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");
+
+        res.set_child(Some(&gtk::Label::new(Some(name))));
+
+        // Add either a drag source or drop target controller depending on direction,
+        // they will be responsible for link creation by dragging an output port onto an input port.
+        //
+        // FIXME: The type used for dragging is simply a u32.
+        //   This means that anything that provides a u32 could be dragged onto a input port,
+        //   leading to that port trying to create a link to an invalid output port.
+        //   We should use a newtype instead of a plain u32.
+        //   Additionally, this does not protect against e.g. dropping an outgoing audio port on an ingoing video port.
+        match direction {
+            Direction::Input => {
+                let drop_target = gtk::DropTarget::new(u32::static_type(), gdk::DragAction::COPY);
+                let this = res.clone();
+                drop_target.connect_drop(move |drop_target, val, _, _| {
+                    if let Ok(source_id) = val.get::<u32>() {
+                        // 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
+                });
+                res.add_controller(&drop_target);
+            }
+            Direction::Output => {
+                // The port will simply provide its pipewire id to the drag target.
+                let drag_src = gtk::DragSourceBuilder::new()
+                    .content(&gdk::ContentProvider::for_value(&(id.to_value())))
+                    .build();
+                res.add_controller(&drag_src);
+
+                // 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")
     }
 }