vsink volume works

gui/VolumeWidgets.cpp

@@ -94,6 +94,8 @@ int NodeVolumeWidget::volume() const { return m_slider->value(); }
 
 bool NodeVolumeWidget::isMuted() const { return m_muteBtn->isChecked(); }
 
+bool NodeVolumeWidget::isSliderDown() const { return m_slider->isSliderDown(); }
+
 void NodeVolumeWidget::setVolume(int value) {
   QSignalBlocker blocker(m_slider);
   m_slider->setValue(value);

gui/VolumeWidgets.h

@@ -20,6 +20,7 @@ public:
 
   int volume() const;
   bool isMuted() const;
+  bool isSliderDown() const;
 
   void setVolume(int value);
   void setMuted(bool muted);

gui/WarpGraphModel.cpp

@@ -627,6 +627,38 @@ void WarpGraphModel::refreshFromClient() {
     }
   }
 
+  for (const auto &[pwId, qtId] : m_pwToQt) {
+    auto volResult = m_client->GetNodeVolume(warppipe::NodeId{pwId});
+    if (!volResult.ok()) continue;
+
+    float vol = volResult.value.volume;
+    bool mute = volResult.value.mute;
+    int sliderVal = static_cast<int>(std::round(vol * 100.0f));
+    sliderVal = std::clamp(sliderVal, 0, 150);
+
+    auto stateIt = m_volumeStates.find(qtId);
+    if (stateIt == m_volumeStates.end()) continue;
+
+    NodeVolumeState &cached = stateIt->second;
+    bool changed = (std::abs(cached.volume - vol) > 1e-4f) || (cached.mute != mute);
+    if (!changed) continue;
+
+    NodeVolumeState previous = cached;
+    cached.volume = vol;
+    cached.mute = mute;
+
+    auto wIt = m_volumeWidgets.find(qtId);
+    if (wIt != m_volumeWidgets.end()) {
+      auto *vw = static_cast<NodeVolumeWidget *>(wIt->second);
+      if (!vw->isSliderDown()) {
+        vw->setVolume(sliderVal);
+        vw->setMuted(mute);
+      }
+    }
+
+    Q_EMIT nodeVolumeChanged(qtId, previous, cached);
+  }
+
   m_refreshing = false;
 }
 

src/warppipe.cpp

@@ -19,6 +19,7 @@
 #include <spa/param/audio/format-utils.h>
 #include <spa/param/props.h>
 #include <spa/pod/builder.h>
+#include <spa/pod/parser.h>
 #include <spa/utils/defs.h>
 
 #include <spa/utils/result.h>
@@ -292,6 +293,14 @@ static const pw_stream_events kNodeMeterEvents = {
   .process = NodeMeterProcess,
 };
 
+struct NodeProxyData {
+  pw_proxy* proxy = nullptr;
+  spa_hook object_listener{};
+  uint32_t node_id = 0;
+  void* impl_ptr = nullptr;
+  bool params_subscribed = false;
+};
+
 }  // namespace
 
 Status Status::Ok() {
@@ -331,6 +340,8 @@ struct Client::Impl {
   std::unordered_map<uint32_t, std::unique_ptr<LinkProxy>> link_proxies;
 
   std::unordered_map<uint32_t, VolumeState> volume_states;
+  std::unordered_map<uint32_t, std::unique_ptr<NodeProxyData>> node_proxies;
+  std::unordered_map<uint32_t, uint32_t> node_channel_counts;
 
   std::unordered_map<uint32_t, MeterState> meter_states;
   std::unordered_set<uint32_t> metered_nodes;
@@ -391,6 +402,8 @@ struct Client::Impl {
   void SetupMasterMeter();
   void TeardownMasterMeter();
   void TeardownAllLiveMeters();
+  void BindNodeForParams(uint32_t id);
+  void UnbindNodeForParams(uint32_t id);
 
   static void RegistryGlobal(void* data,
                              uint32_t id,
@@ -403,8 +416,136 @@ struct Client::Impl {
   static void CoreError(void* data, uint32_t id, int seq, int res, const char* message);
   static int MetadataProperty(void* data, uint32_t subject, const char* key,
                               const char* type, const char* value);
+  static void NodeInfoChanged(void* data, const struct pw_node_info* info);
+  static void NodeParamChanged(void* data, int seq, uint32_t id,
+                               uint32_t index, uint32_t next,
+                               const struct spa_pod* param);
 };
 
+void Client::Impl::NodeInfoChanged(void* data, const struct pw_node_info* info) {
+  auto* np = static_cast<NodeProxyData*>(data);
+  if (!np || !info || np->params_subscribed) return;
+
+  for (uint32_t i = 0; i < info->n_params; ++i) {
+    if (info->params[i].id == SPA_PARAM_Props &&
+        (info->params[i].flags & SPA_PARAM_INFO_READ)) {
+      np->params_subscribed = true;
+      uint32_t params[] = {SPA_PARAM_Props};
+      pw_node_subscribe_params(
+          reinterpret_cast<pw_node*>(np->proxy), params, 1);
+      break;
+    }
+  }
+}
+
+void Client::Impl::NodeParamChanged(void* data, int, uint32_t id,
+                                    uint32_t, uint32_t,
+                                    const struct spa_pod* param) {
+  auto* np = static_cast<NodeProxyData*>(data);
+  if (!np || !np->impl_ptr || !param) return;
+  if (id != SPA_PARAM_Props) return;
+  if (!spa_pod_is_object(param)) return;
+
+  auto* impl = static_cast<Client::Impl*>(np->impl_ptr);
+
+  float volume = -1.0f;
+  bool mute = false;
+  bool found_mute = false;
+  uint32_t n_channels = 0;
+
+  const auto* obj = reinterpret_cast<const spa_pod_object*>(param);
+  const spa_pod_prop* prop;
+  SPA_POD_OBJECT_FOREACH(obj, prop) {
+    switch (prop->key) {
+    case SPA_PROP_channelVolumes: {
+      float vols[64];
+      uint32_t n = spa_pod_copy_array(&prop->value, SPA_TYPE_Float, vols, 64);
+      if (n > 0) {
+        volume = vols[0];
+        n_channels = n;
+      }
+      break;
+    }
+    case SPA_PROP_mute: {
+      bool m = false;
+      if (spa_pod_get_bool(&prop->value, &m) >= 0) {
+        mute = m;
+        found_mute = true;
+      }
+      break;
+    }
+    }
+  }
+
+  bool changed = false;
+  {
+    std::lock_guard<std::mutex> lock(impl->cache_mutex);
+    auto& vs = impl->volume_states[np->node_id];
+    if (volume >= 0.0f && std::abs(vs.volume - volume) > 0.0001f) {
+      vs.volume = volume;
+      changed = true;
+    }
+    if (found_mute && vs.mute != mute) {
+      vs.mute = mute;
+      changed = true;
+    }
+    if (n_channels > 0) {
+      impl->node_channel_counts[np->node_id] = n_channels;
+    }
+  }
+
+  if (changed) {
+    impl->NotifyChange();
+  }
+}
+
+void Client::Impl::BindNodeForParams(uint32_t id) {
+  if (!registry) return;
+
+  {
+    std::lock_guard<std::mutex> lock(cache_mutex);
+    if (node_proxies.count(id)) return;
+  }
+
+  static const pw_node_events node_param_events = {
+    .version = PW_VERSION_NODE_EVENTS,
+    .info = NodeInfoChanged,
+    .param = NodeParamChanged,
+  };
+
+  auto np = std::make_unique<NodeProxyData>();
+  np->proxy = static_cast<pw_proxy*>(
+      pw_registry_bind(registry, id,
+                       PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0));
+  if (!np->proxy) return;
+
+  np->node_id = id;
+  np->impl_ptr = this;
+
+  pw_proxy_add_object_listener(np->proxy,
+      &np->object_listener, &node_param_events, np.get());
+
+  std::lock_guard<std::mutex> lock(cache_mutex);
+  node_proxies[id] = std::move(np);
+}
+
+void Client::Impl::UnbindNodeForParams(uint32_t id) {
+  std::unique_ptr<NodeProxyData> np;
+  {
+    std::lock_guard<std::mutex> lock(cache_mutex);
+    auto it = node_proxies.find(id);
+    if (it != node_proxies.end()) {
+      np = std::move(it->second);
+      node_proxies.erase(it);
+    }
+    node_channel_counts.erase(id);
+  }
+  if (np) {
+    spa_hook_remove(&np->object_listener);
+    pw_proxy_destroy(np->proxy);
+  }
+}
+
 void Client::Impl::RegistryGlobal(void* data,
                                   uint32_t id,
                                   uint32_t,
@@ -417,6 +558,7 @@ void Client::Impl::RegistryGlobal(void* data,
   }
 
   bool notify = false;
+  bool bind_node = false;
 
   {
     std::lock_guard<std::mutex> lock(impl->cache_mutex);
@@ -435,6 +577,7 @@ void Client::Impl::RegistryGlobal(void* data,
       impl->nodes[id] = info;
       impl->CheckRulesForNode(info);
       notify = true;
+      bind_node = true;
     } else if (IsPortType(type)) {
       PortInfo info;
       info.id = PortId{id};
@@ -472,6 +615,10 @@ void Client::Impl::RegistryGlobal(void* data,
     }
   }
 
+  if (bind_node) {
+    impl->BindNodeForParams(id);
+  }
+
   if (notify) {
     impl->NotifyChange();
     return;
@@ -548,6 +695,7 @@ void Client::Impl::RegistryGlobalRemove(void* data, uint32_t id) {
       impl->links.erase(id);
     }
   }
+  impl->UnbindNodeForParams(id);
   impl->NotifyChange();
 }
 
@@ -568,17 +716,19 @@ void Client::Impl::CoreDone(void* data, uint32_t, int seq) {
   }
 }
 
-void Client::Impl::CoreError(void* data, uint32_t, int, int res, const char* message) {
+void Client::Impl::CoreError(void* data, uint32_t id, int seq, int res, const char* message) {
   auto* impl = static_cast<Client::Impl*>(data);
   if (!impl) {
     return;
   }
+  if (id == PW_ID_CORE) {
     impl->connected = false;
     impl->last_error = Status::Error(StatusCode::kUnavailable,
                                      message ? message : spa_strerror(res));
     if (impl->thread_loop) {
       pw_thread_loop_signal(impl->thread_loop, false);
     }
+  }
 }
 
 Status Client::Impl::SyncLocked() {
@@ -876,6 +1026,14 @@ void Client::Impl::DisconnectLocked() {
     }
   }
   saved_link_proxies.clear();
+  for (auto& entry : node_proxies) {
+    if (entry.second) {
+      spa_hook_remove(&entry.second->object_listener);
+      entry.second->proxy = nullptr;
+    }
+  }
+  node_proxies.clear();
+  node_channel_counts.clear();
   if (metadata_listener_attached) {
     spa_hook_remove(&metadata_listener);
     metadata_listener_attached = false;
@@ -1765,24 +1923,58 @@ Status Client::SetNodeVolume(NodeId node, float volume, bool mute) {
     }
   }
 
-  auto* proxy = static_cast<pw_node*>(
+  uint32_t n_channels;
-      pw_registry_bind(impl_->registry, node.value,
+  pw_stream* own_stream = nullptr;
-                       PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0));
+  pw_proxy* proxy = nullptr;
-  if (!proxy) {
+  {
+    std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+    auto streamIt = impl_->virtual_streams.find(node.value);
+    if (streamIt != impl_->virtual_streams.end() && streamIt->second->stream) {
+      own_stream = streamIt->second->stream;
+      n_channels = streamIt->second->channels;
+    } else {
+      auto proxyIt = impl_->node_proxies.find(node.value);
+      if (proxyIt == impl_->node_proxies.end() || !proxyIt->second->proxy) {
         pw_thread_loop_unlock(impl_->thread_loop);
-    return Status::Error(StatusCode::kInternal, "failed to bind node proxy");
+        return Status::Error(StatusCode::kNotFound, "no proxy bound for node");
       }
+      proxy = proxyIt->second->proxy;
+      auto chIt = impl_->node_channel_counts.find(node.value);
+      n_channels = (chIt != impl_->node_channel_counts.end()) ? chIt->second : 2;
+    }
+  }
+  if (n_channels == 0) n_channels = 2;
 
-  uint8_t buffer[128];
+  uint8_t buffer[512];
   spa_pod_builder builder = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
-  auto* param = reinterpret_cast<const spa_pod*>(spa_pod_builder_add_object(
-      &builder,
-      SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
-      SPA_PROP_volume, SPA_POD_Float(volume),
-      SPA_PROP_mute, SPA_POD_Bool(mute)));
 
-  pw_node_set_param(proxy, SPA_PARAM_Props, 0, param);
+  uint32_t vol_prop = own_stream ? SPA_PROP_monitorVolumes : SPA_PROP_channelVolumes;
-  pw_proxy_destroy(reinterpret_cast<pw_proxy*>(proxy));
+  uint32_t mute_prop = own_stream ? SPA_PROP_monitorMute : SPA_PROP_mute;
+
+  spa_pod_frame obj_frame;
+  spa_pod_builder_push_object(&builder, &obj_frame,
+                              SPA_TYPE_OBJECT_Props, SPA_PARAM_Props);
+
+  spa_pod_builder_prop(&builder, vol_prop, 0);
+  spa_pod_frame arr_frame;
+  spa_pod_builder_push_array(&builder, &arr_frame);
+  for (uint32_t ch = 0; ch < n_channels; ++ch) {
+    spa_pod_builder_float(&builder, volume);
+  }
+  spa_pod_builder_pop(&builder, &arr_frame);
+
+  spa_pod_builder_prop(&builder, mute_prop, 0);
+  spa_pod_builder_bool(&builder, mute);
+
+  auto* param = static_cast<const spa_pod*>(
+      spa_pod_builder_pop(&builder, &obj_frame));
+
+  if (own_stream) {
+    pw_stream_set_param(own_stream, SPA_PARAM_Props, param);
+  } else {
+    pw_node_set_param(reinterpret_cast<pw_node*>(proxy),
+                      SPA_PARAM_Props, 0, param);
+  }
 
   {
     std::lock_guard<std::mutex> lock(impl_->cache_mutex);