Fix crashes

CMakeLists.txt

@@ -96,6 +96,7 @@ if(WARPPIPE_BUILD_GUI)
     gui/VolumeWidgets.cpp
     gui/AudioLevelMeter.cpp
     gui/SquareConnectionPainter.cpp
+    gui/WarpBezierConnectionPainter.cpp
   )
 
   target_link_libraries(warppipe-gui PRIVATE
@@ -114,6 +115,7 @@ if(WARPPIPE_BUILD_GUI)
       gui/VolumeWidgets.cpp
       gui/AudioLevelMeter.cpp
       gui/SquareConnectionPainter.cpp
+      gui/WarpBezierConnectionPainter.cpp
     )
 
     target_compile_definitions(warppipe-gui-tests PRIVATE WARPPIPE_TESTING)

gui/GraphEditorWidget.cpp

@@ -9,7 +9,7 @@
 #include <QtNodes/BasicGraphicsScene>
 #include <QtNodes/ConnectionStyle>
 #include <QtNodes/GraphicsView>
-#include <QtNodes/internal/DefaultConnectionPainter.hpp>
+#include "WarpBezierConnectionPainter.h"
 #include <QtNodes/internal/NodeGraphicsObject.hpp>
 #include <QtNodes/internal/ConnectionGraphicsObject.hpp>
 #include <QtNodes/internal/UndoCommands.hpp>
@@ -1927,7 +1927,7 @@ void GraphEditorWidget::setConnectionStyle(ConnectionStyleType style) {
     m_scene->setConnectionPainter(std::make_unique<SquareConnectionPainter>());
   } else {
     m_scene->setConnectionPainter(
-        std::make_unique<QtNodes::DefaultConnectionPainter>());
+        std::make_unique<WarpBezierConnectionPainter>());
   }
 
   for (auto *item : m_scene->items()) {

gui/SquareConnectionPainter.cpp

@@ -205,8 +205,7 @@ void SquareConnectionPainter::paint(
     auto *model = dynamic_cast<WarpGraphModel *>(&scene->graphModel());
     if (model) {
       auto cId = cgo.connectionId();
-      peakLevel = std::max(model->nodePeakLevel(cId.outNodeId),
-                           model->nodePeakLevel(cId.inNodeId));
+      peakLevel = model->connectionPeakLevel(cId);
     }
   }
 

gui/WarpBezierConnectionPainter.cpp

@@ -0,0 +1,117 @@
+#include "WarpBezierConnectionPainter.h"
+#include "WarpGraphModel.h"
+
+#include <QtNodes/internal/BasicGraphicsScene.hpp>
+#include <QtNodes/internal/ConnectionGraphicsObject.hpp>
+#include <QtNodes/internal/ConnectionState.hpp>
+#include <QtNodes/internal/Definitions.hpp>
+#include <QtNodes/StyleCollection>
+
+#include <QPainter>
+#include <QPainterPath>
+#include <QPainterPathStroker>
+
+#include <algorithm>
+#include <cmath>
+
+QPainterPath WarpBezierConnectionPainter::cubicPath(
+    QtNodes::ConnectionGraphicsObject const &cgo) const {
+  QPointF const &in = cgo.endPoint(QtNodes::PortType::In);
+  QPointF const &out = cgo.endPoint(QtNodes::PortType::Out);
+  auto const c1c2 = cgo.pointsC1C2();
+
+  QPainterPath cubic(out);
+  cubic.cubicTo(c1c2.first, c1c2.second, in);
+  return cubic;
+}
+
+void WarpBezierConnectionPainter::paint(
+    QPainter *painter,
+    QtNodes::ConnectionGraphicsObject const &cgo) const {
+  auto const &style = QtNodes::StyleCollection::connectionStyle();
+
+  bool const hovered = cgo.connectionState().hovered();
+  bool const selected = cgo.isSelected();
+  bool const sketch = cgo.connectionState().requiresPort();
+
+  auto path = cubicPath(cgo);
+
+  float peakLevel = 0.0f;
+  auto *scene = cgo.nodeScene();
+  if (scene) {
+    auto *model = dynamic_cast<WarpGraphModel *>(&scene->graphModel());
+    if (model) {
+      auto cId = cgo.connectionId();
+      peakLevel = model->connectionPeakLevel(cId);
+    }
+  }
+
+  auto activeColor = [&](QColor base) -> QColor {
+    if (peakLevel < 0.005f)
+      return base;
+    float t = std::min(peakLevel * 2.0f, 1.0f);
+    int r = static_cast<int>(base.red() + t * (60 - base.red()));
+    int g = static_cast<int>(base.green() + t * (210 - base.green()));
+    int b = static_cast<int>(base.blue() + t * (80 - base.blue()));
+    return QColor(std::clamp(r, 0, 255),
+                  std::clamp(g, 0, 255),
+                  std::clamp(b, 0, 255),
+                  base.alpha());
+  };
+
+  if (hovered || selected) {
+    QPen pen;
+    pen.setWidth(static_cast<int>(2 * style.lineWidth()));
+    pen.setColor(selected ? style.selectedHaloColor() : style.hoveredColor());
+    painter->setPen(pen);
+    painter->setBrush(Qt::NoBrush);
+    painter->drawPath(path);
+  }
+
+  if (sketch) {
+    QPen pen;
+    pen.setWidth(static_cast<int>(style.constructionLineWidth()));
+    pen.setColor(style.constructionColor());
+    pen.setStyle(Qt::DashLine);
+    painter->setPen(pen);
+    painter->setBrush(Qt::NoBrush);
+    painter->drawPath(path);
+  } else {
+    QColor base = selected ? style.selectedColor() : style.normalColor();
+    QColor color = selected ? base : activeColor(base);
+    float width = style.lineWidth();
+    if (!selected && peakLevel > 0.005f)
+      width += peakLevel * 1.5f;
+
+    QPen pen;
+    pen.setWidthF(width);
+    pen.setColor(color);
+    painter->setPen(pen);
+    painter->setBrush(Qt::NoBrush);
+    painter->drawPath(path);
+  }
+
+  double const pointRadius = style.pointDiameter() / 2.0;
+  painter->setPen(style.constructionColor());
+  painter->setBrush(style.constructionColor());
+  painter->drawEllipse(cgo.out(), pointRadius, pointRadius);
+  painter->drawEllipse(cgo.in(), pointRadius, pointRadius);
+}
+
+QPainterPath WarpBezierConnectionPainter::getPainterStroke(
+    QtNodes::ConnectionGraphicsObject const &cgo) const {
+  auto cubic = cubicPath(cgo);
+
+  QPointF const &out = cgo.endPoint(QtNodes::PortType::Out);
+  QPainterPath result(out);
+
+  unsigned int constexpr segments = 20;
+  for (unsigned int i = 0; i < segments; ++i) {
+    double ratio = double(i + 1) / segments;
+    result.lineTo(cubic.pointAtPercent(ratio));
+  }
+
+  QPainterPathStroker stroker;
+  stroker.setWidth(10.0);
+  return stroker.createStroke(result);
+}

gui/WarpBezierConnectionPainter.h

@@ -0,0 +1,15 @@
+#pragma once
+
+#include <QtNodes/internal/AbstractConnectionPainter.hpp>
+
+class WarpBezierConnectionPainter : public QtNodes::AbstractConnectionPainter {
+public:
+  void paint(QPainter *painter,
+             QtNodes::ConnectionGraphicsObject const &cgo) const override;
+  QPainterPath
+  getPainterStroke(QtNodes::ConnectionGraphicsObject const &cgo) const override;
+
+private:
+  QPainterPath
+  cubicPath(QtNodes::ConnectionGraphicsObject const &cgo) const;
+};

gui/WarpGraphModel.cpp

@@ -1473,6 +1473,23 @@ float WarpGraphModel::nodePeakLevel(QtNodes::NodeId nodeId) const {
   return it != m_peakLevels.end() ? it->second : 0.0f;
 }
 
+float WarpGraphModel::connectionPeakLevel(QtNodes::ConnectionId cId) const {
+  constexpr float kSourceReliableThreshold = 0.005f;
+
+  float outPeak = nodePeakLevel(cId.outNodeId);
+  if (outPeak >= kSourceReliableThreshold)
+    return outPeak;
+
+  auto outNodeIt = m_nodes.find(cId.outNodeId);
+  if (outNodeIt == m_nodes.end())
+    return outPeak;
+
+  if (classifyNode(outNodeIt->second.info) != WarpNodeType::kApplication)
+    return outPeak;
+
+  return std::max(outPeak, nodePeakLevel(cId.inNodeId));
+}
+
 void WarpGraphModel::recomputeConnectionChannels() {
   m_connectionChannels.clear();
 

gui/WarpGraphModel.h

@@ -109,6 +109,7 @@ public:
 
   void setNodePeakLevel(QtNodes::NodeId nodeId, float level);
   float nodePeakLevel(QtNodes::NodeId nodeId) const;
+  float connectionPeakLevel(QtNodes::ConnectionId cId) const;
 
   struct ConnectionChannel {
     int index = 0;

src/warppipe.cpp

@@ -124,6 +124,7 @@ struct StreamData {
 struct LinkProxy {
   pw_proxy* proxy = nullptr;
   spa_hook listener{};
+  bool listener_attached = false;
   pw_thread_loop* loop = nullptr;
   bool done = false;
   bool failed = false;
@@ -133,6 +134,20 @@ struct LinkProxy {
   uint32_t input_port = 0;
 };
 
+void DetachLinkProxy(LinkProxy* link, bool destroy_proxy) {
+  if (!link) {
+    return;
+  }
+  if (link->listener_attached) {
+    spa_hook_remove(&link->listener);
+    link->listener_attached = false;
+  }
+  if (destroy_proxy && link->proxy) {
+    pw_proxy_destroy(link->proxy);
+  }
+  link->proxy = nullptr;
+}
+
 void LinkProxyBound(void* data, uint32_t global_id) {
   auto* link = static_cast<LinkProxy*>(data);
   if (!link) {
@@ -150,6 +165,8 @@ void LinkProxyRemoved(void* data) {
   if (!link) {
     return;
   }
+  link->listener_attached = false;
+  link->proxy = nullptr;
   link->done = true;
   if (link->loop) {
     pw_thread_loop_signal(link->loop, false);
@@ -161,6 +178,8 @@ void LinkProxyError(void* data, int, int res, const char* message) {
   if (!link) {
     return;
   }
+  link->listener_attached = false;
+  link->proxy = nullptr;
   link->failed = true;
   link->error = message ? message : spa_strerror(res);
   if (link->loop) {
@@ -697,7 +716,6 @@ void Client::Impl::RegistryGlobalRemove(void* data, uint32_t id) {
   {
     std::lock_guard<std::mutex> lock(impl->cache_mutex);
     impl->virtual_streams.erase(id);
-    impl->link_proxies.erase(id);
     auto node_it = impl->nodes.find(id);
     if (node_it != impl->nodes.end()) {
       impl->nodes.erase(node_it);
@@ -1131,11 +1149,7 @@ void Client::Impl::DisconnectLocked() {
     streams.swap(virtual_streams);
   }
   for (auto& entry : links) {
-    LinkProxy* link = entry.second.get();
-    if (link) {
-      spa_hook_remove(&link->listener);
-      link->proxy = nullptr;
-    }
+    DetachLinkProxy(entry.second.get(), false);
   }
   for (auto& entry : streams) {
     StreamData* stream_data = entry.second.get();
@@ -1150,17 +1164,11 @@ void Client::Impl::DisconnectLocked() {
     }
   }
   for (auto& entry : auto_link_proxies) {
-    if (entry) {
-      spa_hook_remove(&entry->listener);
-      entry->proxy = nullptr;
-    }
+    DetachLinkProxy(entry.get(), false);
   }
   auto_link_proxies.clear();
   for (auto& entry : saved_link_proxies) {
-    if (entry) {
-      spa_hook_remove(&entry->listener);
-      entry->proxy = nullptr;
-    }
+    DetachLinkProxy(entry.get(), false);
   }
   saved_link_proxies.clear();
   for (auto& entry : node_proxies) {
@@ -1283,13 +1291,17 @@ void Client::Impl::EnforceRulesForLink(uint32_t link_id, uint32_t out_port,
 
   if (!should_destroy) return;
 
-  if (link_proxies.count(link_id)) return;
+  auto active_link_proxy = link_proxies.find(link_id);
+  if (active_link_proxy != link_proxies.end() &&
+      active_link_proxy->second && active_link_proxy->second->proxy) {
+    return;
+  }
   for (const auto& proxy : auto_link_proxies) {
-    if (proxy && proxy->output_port == out_port &&
+    if (proxy && proxy->proxy && proxy->output_port == out_port &&
         proxy->input_port == in_port) return;
   }
   for (const auto& proxy : saved_link_proxies) {
-    if (proxy && proxy->output_port == out_port &&
+    if (proxy && proxy->proxy && proxy->output_port == out_port &&
         proxy->input_port == in_port) return;
   }
   for (const auto& pair : auto_link_claimed_pairs) {
@@ -1430,12 +1442,14 @@ void Client::Impl::ProcessPendingAutoLinks() {
         if (target_in == in_port) { is_ours = true; break; }
       }
       if (!is_ours) {
-        if (link_proxies.count(link_id)) is_ours = true;
+        auto owned = link_proxies.find(link_id);
+        if (owned != link_proxies.end() && owned->second && owned->second->proxy)
+          is_ours = true;
       }
       if (!is_ours) {
         uint32_t out_port = link_entry.second.output_port.value;
         for (const auto& proxy : auto_link_proxies) {
-          if (proxy && proxy->output_port == out_port &&
+          if (proxy && proxy->proxy && proxy->output_port == out_port &&
               proxy->input_port == in_port) { is_ours = true; break; }
         }
       }
@@ -1450,7 +1464,7 @@ void Client::Impl::ProcessPendingAutoLinks() {
       if (!is_ours) {
         uint32_t out_port = link_entry.second.output_port.value;
         for (const auto& proxy : saved_link_proxies) {
-          if (proxy && proxy->output_port == out_port &&
+          if (proxy && proxy->proxy && proxy->output_port == out_port &&
               proxy->input_port == in_port) { is_ours = true; break; }
         }
       }
@@ -1498,6 +1512,7 @@ void Client::Impl::CreateAutoLinkAsync(uint32_t output_port, uint32_t input_port
   link_data->output_port = output_port;
   link_data->input_port = input_port;
   pw_proxy_add_listener(proxy, &link_data->listener, &kLinkProxyEvents, link_data.get());
+  link_data->listener_attached = true;
 
   std::lock_guard<std::mutex> lock(cache_mutex);
   auto_link_proxies.push_back(std::move(link_data));
@@ -1588,14 +1603,15 @@ void Client::Impl::ProcessSavedLinks() {
         if (saved_in == in_port) { is_ours = true; break; }
       }
       if (!is_ours) {
-        if (link_proxies.count(link_id)) {
+        auto owned = link_proxies.find(link_id);
+        if (owned != link_proxies.end() && owned->second && owned->second->proxy) {
           is_ours = true;
         }
       }
       if (!is_ours) {
         uint32_t out_port = link_entry.second.output_port.value;
         for (const auto& proxy : auto_link_proxies) {
-          if (proxy && proxy->output_port == out_port &&
+          if (proxy && proxy->proxy && proxy->output_port == out_port &&
               proxy->input_port == in_port) { is_ours = true; break; }
         }
       }
@@ -1610,7 +1626,7 @@ void Client::Impl::ProcessSavedLinks() {
       if (!is_ours) {
         uint32_t out_port = link_entry.second.output_port.value;
         for (const auto& proxy : saved_link_proxies) {
-          if (proxy && proxy->output_port == out_port &&
+          if (proxy && proxy->proxy && proxy->output_port == out_port &&
               proxy->input_port == in_port) { is_ours = true; break; }
         }
       }
@@ -1654,6 +1670,7 @@ void Client::Impl::CreateSavedLinkAsync(uint32_t output_port,
   link_data->input_port = input_port;
   pw_proxy_add_listener(proxy, &link_data->listener, &kLinkProxyEvents,
                         link_data.get());
+  link_data->listener_attached = true;
 
   std::lock_guard<std::mutex> lock(cache_mutex);
   saved_link_proxies.push_back(std::move(link_data));
@@ -1713,7 +1730,7 @@ void Client::Impl::AutoSave() {
     std::lock_guard<std::mutex> lock(cache_mutex);
     std::vector<SavedLink> live;
     for (const auto& entry : link_proxies) {
-      if (!entry.second) {
+      if (!entry.second || !entry.second->proxy) {
         continue;
       }
       auto link_it = links.find(entry.first);
@@ -1742,7 +1759,7 @@ void Client::Impl::AutoSave() {
       links_array.push_back(std::move(link_obj));
     }
     for (const auto& lp : saved_link_proxies) {
-      if (!lp || lp->id == SPA_ID_INVALID) continue;
+      if (!lp || !lp->proxy || lp->id == SPA_ID_INVALID) continue;
       auto link_it = links.find(lp->id);
       if (link_it == links.end()) continue;
       const Link& link = link_it->second;
@@ -2449,6 +2466,7 @@ Result<Link> Client::CreateLink(PortId output, PortId input, const LinkOptions&
   link_proxy->output_port = output.value;
   link_proxy->input_port = input.value;
   pw_proxy_add_listener(proxy, &link_proxy->listener, &kLinkProxyEvents, link_proxy.get());
+  link_proxy->listener_attached = true;
 
   int wait_attempts = 0;
   while (link_proxy->id == SPA_ID_INVALID && !link_proxy->failed && wait_attempts < 3) {
@@ -2469,13 +2487,13 @@ Result<Link> Client::CreateLink(PortId output, PortId input, const LinkOptions&
   if (link_proxy->failed) {
     std::string error = link_proxy->error.empty() ? "link creation failed" : link_proxy->error;
     remove_pending();
-    pw_proxy_destroy(proxy);
+    DetachLinkProxy(link_proxy.get(), true);
     pw_thread_loop_unlock(impl_->thread_loop);
     return {Status::Error(StatusCode::kUnavailable, std::move(error)), {}};
   }
   if (link_proxy->id == SPA_ID_INVALID) {
     remove_pending();
-    pw_proxy_destroy(proxy);
+    DetachLinkProxy(link_proxy.get(), true);
     pw_thread_loop_unlock(impl_->thread_loop);
     return {Status::Error(StatusCode::kTimeout, "timed out waiting for link id"), {}};
   }
@@ -2486,7 +2504,7 @@ Result<Link> Client::CreateLink(PortId output, PortId input, const LinkOptions&
   link.input_port = input;
   {
     std::lock_guard<std::mutex> lock(impl_->cache_mutex);
-    impl_->link_proxies.emplace(link_proxy->id, std::move(link_proxy));
+    impl_->link_proxies[link_proxy->id] = std::move(link_proxy);
     impl_->links[link.id.value] = link;
     std::erase_if(impl_->pending_link_pairs, [&](const auto& p) {
       return p.first == output.value && p.second == input.value;
@@ -2572,14 +2590,9 @@ Status Client::RemoveLink(LinkId link) {
     }
     auto it = impl_->link_proxies.find(link.value);
     if (it != impl_->link_proxies.end()) {
-      if (it->second && it->second->proxy) {
-        spa_hook_remove(&it->second->listener);
-        pw_proxy_destroy(it->second->proxy);
-      }
       if (impl_->registry) {
         pw_registry_destroy(impl_->registry, link.value);
       }
-      impl_->link_proxies.erase(it);
       auto link_it2 = impl_->links.find(link.value);
       if (link_it2 != impl_->links.end()) {
         uint32_t op = link_it2->second.output_port.value;
@@ -2609,26 +2622,6 @@ Status Client::RemoveLink(LinkId link) {
         impl_->links.erase(link_it);
       }
       if (out_port && in_port) {
-        for (auto& p : impl_->saved_link_proxies) {
-          if (p && p->output_port == out_port && p->input_port == in_port) {
-            spa_hook_remove(&p->listener);
-            if (p->proxy) pw_proxy_destroy(p->proxy);
-            p->proxy = nullptr;
-          }
-        }
-        std::erase_if(impl_->saved_link_proxies, [&](const auto& p) {
-          return p && p->output_port == out_port && p->input_port == in_port;
-        });
-        for (auto& p : impl_->auto_link_proxies) {
-          if (p && p->output_port == out_port && p->input_port == in_port) {
-            spa_hook_remove(&p->listener);
-            if (p->proxy) pw_proxy_destroy(p->proxy);
-            p->proxy = nullptr;
-          }
-        }
-        std::erase_if(impl_->auto_link_proxies, [&](const auto& p) {
-          return p && p->output_port == out_port && p->input_port == in_port;
-        });
         std::erase_if(impl_->auto_link_claimed_pairs, [&](const auto& pair) {
           return pair.first == out_port && pair.second == in_port;
         });
@@ -2792,18 +2785,7 @@ Status Client::RemoveRouteRule(RuleId id) {
     }
     {
       std::lock_guard<std::mutex> lock(impl_->cache_mutex);
-      for (const auto& pair : pairs_to_remove) {
-        for (auto& p : impl_->auto_link_proxies) {
-          if (p && p->output_port == pair.first &&
-              p->input_port == pair.second) {
-            spa_hook_remove(&p->listener);
-          }
-        }
-        std::erase_if(impl_->auto_link_proxies, [&](const auto& p) {
-          return p && p->output_port == pair.first &&
-                      p->input_port == pair.second;
-        });
-      }
+      (void)pairs_to_remove;
     }
     pw_thread_loop_unlock(impl_->thread_loop);
   }

tests/gui/warppipe_gui_tests.cpp

@@ -1671,6 +1671,42 @@ TEST_CASE("findPwNodeIdByName returns 0 for ghost nodes without pw mapping") {
   REQUIRE(model.findPwNodeIdByName("ghost-lookup") == 100220);
 }
 
+TEST_CASE("connectionPeakLevel uses source node activity, with application fallback") {
+  auto tc = TestClient::Create();
+  if (!tc.available()) { SUCCEED("PipeWire unavailable"); return; }
+  ensureApp();
+
+  REQUIRE(tc.client->Test_InsertNode(
+      MakeNode(100240, "app-out", "Stream/Output/Audio", "Firefox")).ok());
+  REQUIRE(tc.client->Test_InsertNode(
+      MakeNode(100241, "sink-out", "Audio/Sink")).ok());
+  REQUIRE(tc.client->Test_InsertNode(
+      MakeNode(100242, "hw-in", "Audio/Sink")).ok());
+
+  WarpGraphModel model(tc.client.get());
+  model.refreshFromClient();
+
+  auto appQt = model.qtNodeIdForPw(100240);
+  auto sinkQt = model.qtNodeIdForPw(100241);
+  auto hwQt = model.qtNodeIdForPw(100242);
+  REQUIRE(appQt != 0);
+  REQUIRE(sinkQt != 0);
+  REQUIRE(hwQt != 0);
+
+  model.setNodePeakLevel(appQt, 0.0f);
+  model.setNodePeakLevel(sinkQt, 0.0f);
+  model.setNodePeakLevel(hwQt, 0.8f);
+
+  REQUIRE(model.connectionPeakLevel(QtNodes::ConnectionId{sinkQt, 0u, hwQt, 0u}) ==
+          Catch::Approx(0.0f));
+  REQUIRE(model.connectionPeakLevel(QtNodes::ConnectionId{appQt, 0u, hwQt, 0u}) ==
+          Catch::Approx(0.8f));
+
+  model.setNodePeakLevel(appQt, 0.4f);
+  REQUIRE(model.connectionPeakLevel(QtNodes::ConnectionId{appQt, 0u, hwQt, 0u}) ==
+          Catch::Approx(0.4f));
+}
+
 TEST_CASE("saveLayout stores and loadLayout restores view state") {
   auto tc = TestClient::Create();
   if (!tc.available()) { SUCCEED("PipeWire unavailable"); return; }