modules: port to new WpProxy API

lib/wp/remote-pipewire.c

@@ -389,8 +389,8 @@ wp_remote_pipewire_create_object (WpRemotePipewire *self,
   g_return_val_if_fail (self->core_proxy, NULL);
 
   pw_proxy = pw_core_proxy_create_object (self->core_proxy, factory_name,
-      interface_type, interface_version, wp_properties_peek_dict (properties),
+      interface_type, interface_version,
-      0);
+      properties ? wp_properties_peek_dict (properties) : NULL, 0);
   return wp_proxy_new_wrap (WP_REMOTE (self), pw_proxy, interface_type,
       interface_version);
 }

meson.build

@@ -33,6 +33,6 @@ gnome = import('gnome')
 wp_lib_include_dir = include_directories('lib')
 
 subdir('lib')
-#subdir('modules')
+subdir('modules')
 subdir('src')
 subdir('tests')

modules/module-client-permissions.c

@@ -9,111 +9,22 @@
 #include <wp/wp.h>
 #include <pipewire/pipewire.h>
 
-struct client_data
-{
-  union {
-    struct pw_proxy *proxy;
-    struct pw_client_proxy *client_proxy;
-  };
-  struct spa_hook proxy_listener;
-  struct spa_hook client_listener;
-  gboolean done;
-};
-
-static gboolean
-do_free_client_data (gpointer data)
-{
-  g_rc_box_release (data);
-  return G_SOURCE_REMOVE;
-}
-
 static void
-proxy_destroy (void *data)
+client_added (WpRemote * remote, WpProxyClient *client, gpointer data)
 {
-  struct client_data *d = data;
+  g_autoptr (WpProperties) properties = NULL;
-  d->proxy = NULL;
-  /* destroy later because we can't free the memory of the proxy_listener
-   * while we are running in one of its callbacks */
-  g_idle_add (do_free_client_data, data);
-}
-
-static gboolean
-do_destroy_proxy (gpointer data)
-{
-  struct client_data *d = data;
-  if (d->proxy) {
-    g_debug ("Destroying client proxy %p", d->proxy);
-    pw_proxy_destroy (d->proxy);
-  }
-  return G_SOURCE_REMOVE;
-}
-
-static void
-proxy_done (void *data, int seq)
-{
-  struct client_data *d = data;
-
-  /* the proxy is not useful to keep around once we have changed permissions.
-   * take an extra ref on the client data because the proxy may
-   * disappear on its own if the client disconnects in the meantime */
-  if (d->done)
-    g_idle_add_full (G_PRIORITY_DEFAULT_IDLE, do_destroy_proxy,
-        g_rc_box_acquire (d), g_rc_box_release);
-}
-
-static const struct pw_proxy_events proxy_events = {
-  PW_VERSION_PROXY_EVENTS,
-  .destroy = proxy_destroy,
-  .done = proxy_done,
-};
-
-static void
-client_info (void *object, const struct pw_client_info *info)
-{
-  struct client_data *d = object;
   const char *access;
+  guint32 id = wp_proxy_get_global_id (WP_PROXY (client));
 
-  if (!(info->change_mask & PW_CLIENT_CHANGE_MASK_PROPS))
+  g_debug ("Client added: %d", id);
-    return;
 
-  g_return_if_fail (info->props);
+  properties = wp_proxy_client_get_properties (client);
-  access = spa_dict_lookup (info->props, "pipewire.access");
+  access = wp_properties_get (properties, PW_KEY_ACCESS);
 
-  /* grant full permissions to restricted or security confined apps
-     TODO: we should eventually build a system where we can use the role
-     and the client's security label to grant access only to specific nodes
-     and endpoints in the graph */
   if (!g_strcmp0 (access, "flatpak") || !g_strcmp0 (access, "restricted")) {
-    const struct pw_permission perm = PW_PERMISSION_INIT(-1, PW_PERM_RWX);
+    g_debug ("Granting full access to client %d", id);
-
+    wp_proxy_client_update_permissions (client, 1, -1, PW_PERM_RWX);
-    g_debug ("Granting full access to client %d (%p)", info->id, d->proxy);
-    pw_client_proxy_update_permissions (d->client_proxy, 1, &perm);
   }
-
-  d->done = TRUE;
-  pw_proxy_sync (d->proxy, 123456);
-}
-
-static const struct pw_client_proxy_events client_events = {
-  PW_VERSION_CLIENT_PROXY_EVENTS,
-  .info = client_info,
-};
-
-static void
-client_added (WpRemotePipewire * remote, guint32 id,
-    const struct spa_dict *properties, gpointer data)
-{
-  struct client_data *d;
-
-  d = g_rc_box_new0 (struct client_data);
-
-  d->proxy = wp_remote_pipewire_proxy_bind (remote, id,
-      PW_TYPE_INTERFACE_Client);
-  pw_proxy_add_listener (d->proxy, &d->proxy_listener, &proxy_events, d);
-  pw_client_proxy_add_listener (d->client_proxy, &d->client_listener,
-      &client_events, d);
-
-  g_debug ("Bound to client %d (%p)", id, d->proxy);
 }
 
 void
@@ -122,6 +33,9 @@ wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
   WpRemote *remote = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
   g_return_if_fail (remote != NULL);
 
+  wp_remote_pipewire_set_default_features (WP_REMOTE_PIPEWIRE (remote),
+      WP_TYPE_PROXY_CLIENT, WP_PROXY_FEATURE_PW_PROXY | WP_PROXY_FEATURE_INFO);
+
   g_signal_connect(remote, "global-added::client", (GCallback) client_added,
       NULL);
 }

modules/module-pipewire/simple-endpoint-link.c

@@ -102,16 +102,8 @@ simple_endpoint_link_finalize (GObject * object)
 {
   WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(object);
 
-  /* Destroy the init task */
+  g_clear_object (&self->init_task);
-  g_clear_object(&self->init_task);
+  g_clear_pointer (&self->link_proxies, g_ptr_array_unref);
-
-  /* Destroy the proxies port */
-  if (self->link_proxies) {
-    g_ptr_array_free(self->link_proxies, TRUE);
-    self->link_proxies = NULL;
-  }
-
-  /* Clear the core weak reference */
   g_weak_ref_clear (&self->core);
 }
 
@@ -150,34 +142,23 @@ simple_endpoint_link_get_property (GObject * object, guint property_id,
 }
 
 static void
-finish_simple_endpoint_link_creation(WpPipewireSimpleEndpointLink *self)
+on_proxy_link_augmented (WpProxy *proxy, GAsyncResult *res, gpointer data)
-{
-  /* Don't do anything if the link has already been initialized */
-  if (!self->init_task)
-    return;
-
-  /* Finish the creation of the audio dsp */
-  g_task_return_boolean (self->init_task, TRUE);
-  g_clear_object(&self->init_task);
-}
-
-static void
-on_proxy_link_created(GObject *initable, GAsyncResult *res, gpointer data)
 {
   WpPipewireSimpleEndpointLink *self = data;
-  WpProxyLink *proxy_link = NULL;
+  g_autoptr (GError) error = NULL;
 
-  /* Get the link */
+  wp_proxy_augment_finish (proxy, res, &error);
-  proxy_link = wp_proxy_link_new_finish(initable, res, NULL);
+  if (error && self->init_task) {
-  g_return_if_fail (proxy_link);
+    g_task_return_error (self->init_task, g_steal_pointer (&error));
-
+    g_clear_object (&self->init_task);
-  /* Add the proxy link to the array */
+    return;
-  g_ptr_array_add(self->link_proxies, proxy_link);
+  }
-  self->link_count--;
 
   /* Finish the simple endpoint link creation if all links have been created */
-  if (self->link_count == 0)
+  if (--self->link_count == 0 && self->init_task) {
-    finish_simple_endpoint_link_creation (self);
+    g_task_return_boolean (self->init_task, TRUE);
+    g_clear_object(&self->init_task);
+  }
 }
 
 static gboolean
@@ -185,15 +166,18 @@ simple_endpoint_link_create (WpEndpointLink * epl, GVariant * src_data,
     GVariant * sink_data, GError ** error)
 {
   WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(epl);
-  g_autoptr (WpCore) core = g_weak_ref_get (&self->core);
+  g_autoptr (WpCore) core = NULL;
   WpRemotePipewire *remote_pipewire;
-  struct pw_properties *props;
   guint32 output_node_id, input_node_id;
   GVariant *src_ports, *sink_ports;
   GVariantIter *out_iter, *in_iter;
   guint64 out_ptr, in_ptr;
-  GHashTable *linked_ports = NULL;
+  g_autoptr (GHashTable) linked_ports = NULL;
-  struct pw_proxy *proxy;
+  g_autoptr (WpProperties) props = NULL;
+  WpProxy *proxy = NULL;
+
+  core = g_weak_ref_get (&self->core);
+  g_return_val_if_fail (core, FALSE);
 
   /* Get the remote pipewire */
   remote_pipewire = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
@@ -230,40 +214,37 @@ simple_endpoint_link_create (WpEndpointLink * epl, GVariant * src_data,
         continue;
 
       /* Skip the ports if they are already linked */
-      if (g_hash_table_contains (linked_ports, GUINT_TO_POINTER(in_id)))
+      if (g_hash_table_contains (linked_ports, GUINT_TO_POINTER(in_id)) ||
-        continue;
+          g_hash_table_contains (linked_ports, GUINT_TO_POINTER(out_id)))
-      if (g_hash_table_contains (linked_ports, GUINT_TO_POINTER(out_id)))
         continue;
 
       /* Create the properties */
-      props = pw_properties_new(NULL, NULL);
+      props = wp_properties_new_empty ();
-      pw_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%d", output_node_id);
+      wp_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%d", output_node_id);
-      pw_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", out_id);
+      wp_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", out_id);
-      pw_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%d", input_node_id);
+      wp_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%d", input_node_id);
-      pw_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", in_id);
+      wp_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", in_id);
 
       /* Create the link */
       proxy = wp_remote_pipewire_create_object(remote_pipewire, "link-factory",
-          PW_TYPE_INTERFACE_Link, &props->dict);
+          PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
       g_return_val_if_fail (proxy, FALSE);
-      wp_proxy_link_new (pw_proxy_get_id(proxy), proxy, on_proxy_link_created,
+      g_ptr_array_add(self->link_proxies, proxy);
-          self);
+
+      /* Wait for the link to be created on the server side
+         by waiting for the info event, which will be signaled anyway */
       self->link_count++;
+      wp_proxy_augment (proxy, WP_PROXY_FEATURE_INFO, NULL,
+          (GAsyncReadyCallback) on_proxy_link_augmented, self);
 
       /* Insert the port ids in the hash tables to know they are linked */
-      g_hash_table_insert (linked_ports, GUINT_TO_POINTER(in_id), NULL);
+      g_hash_table_add (linked_ports, GUINT_TO_POINTER(in_id));
-      g_hash_table_insert (linked_ports, GUINT_TO_POINTER(out_id), NULL);
+      g_hash_table_add (linked_ports, GUINT_TO_POINTER(out_id));
-
-      /* Clean up */
-      pw_properties_free(props);
     }
     g_variant_iter_free (in_iter);
   }
   g_variant_iter_free (out_iter);
 
-  /* Clean up */
-  g_hash_table_unref(linked_ports);
-
   return TRUE;
 }
 
@@ -272,11 +253,7 @@ simple_endpoint_link_destroy (WpEndpointLink * epl)
 {
   WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(epl);
 
-  /* Destroy the proxies port */
+  g_clear_pointer (&self->link_proxies, g_ptr_array_unref);
-  if (self->link_proxies) {
-    g_ptr_array_free(self->link_proxies, TRUE);
-    self->link_proxies = NULL;
-  }
 }
 
 static void

modules/module-pipewire/simple-endpoint.c

@@ -24,9 +24,6 @@ struct _WpPipewireSimpleEndpoint
 {
   WpEndpoint parent;
 
-  /* The global-id this endpoint refers to */
-  guint global_id;
-
   /* properties */
   gchar *role;
   guint64 creation_time;
@@ -34,17 +31,12 @@ struct _WpPipewireSimpleEndpoint
 
   /* The task to signal the endpoint is initialized */
   GTask *init_task;
-  gboolean init_abort;
 
   /* The remote pipewire */
   WpRemotePipewire *remote_pipewire;
 
-  /* Handler */
-  gulong proxy_node_done_handler_id;
-
   /* Proxies */
   WpProxyNode *proxy_node;
-  struct spa_hook node_proxy_listener;
   GPtrArray *proxies_port;
 
   /* controls cache */
@@ -54,7 +46,7 @@ struct _WpPipewireSimpleEndpoint
 
 enum {
   PROP_0,
-  PROP_GLOBAL_ID,
+  PROP_PROXY_NODE,
   PROP_ROLE,
   PROP_CREATION_TIME,
   PROP_TARGET,
@@ -77,43 +69,32 @@ G_DEFINE_TYPE_WITH_CODE (WpPipewireSimpleEndpoint, simple_endpoint,
     G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
                            wp_simple_endpoint_async_initable_init))
 
-typedef GObject* (*WpObjectNewFinishFunc)(GObject *initable, GAsyncResult *res,
+static gboolean
-    GError **error);
+proxy_safe_augment_finish (WpPipewireSimpleEndpoint * self, WpProxy *proxy,
-
+    GAsyncResult *res)
-static GObject *
-object_safe_new_finish(WpPipewireSimpleEndpoint * self, GObject *initable,
-    GAsyncResult *res, WpObjectNewFinishFunc new_finish_func)
 {
-  GObject *object = NULL;
   GError *error = NULL;
 
-  /* Return NULL if we are already aborting */
+  /* Return FALSE if we are already aborting */
-  if (self->init_abort)
+  if (!self->init_task)
-    return NULL;
+    return FALSE;
 
-  /* Get the object */
+  wp_proxy_augment_finish (proxy, res, &error);
-  object = G_OBJECT (new_finish_func (initable, res, &error));
-  g_return_val_if_fail (object, NULL);
-
-  /* Check for error */
   if (error) {
-    g_clear_object (&object);
     g_warning ("WpPipewireSimpleEndpoint:%p Aborting construction", self);
-    self->init_abort = TRUE;
     g_task_return_error (self->init_task, error);
     g_clear_object (&self->init_task);
-    return NULL;
+    return FALSE;
   }
 
-  return object;
+  return TRUE;
 }
 
 static void
-node_proxy_param (void *object, int seq, uint32_t id,
+node_proxy_param (WpProxy *proxy, int seq, uint32_t id,
-    uint32_t index, uint32_t next, const struct spa_pod *param)
+    uint32_t index, uint32_t next, const struct spa_pod *param,
+    WpPipewireSimpleEndpoint *self)
 {
-  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
-
   switch (id) {
     case SPA_PARAM_Props:
     {
@@ -154,90 +135,78 @@ node_proxy_param (void *object, int seq, uint32_t id,
   }
 }
 
-static const struct pw_node_proxy_events node_node_proxy_events = {
-  PW_VERSION_NODE_PROXY_EVENTS,
-  .param = node_proxy_param,
-};
-
 static void
-on_all_ports_done(WpProxy *proxy, gpointer data)
+on_all_ports_done (WpProxy *proxy, GAsyncResult *res,
+    WpPipewireSimpleEndpoint *self)
 {
-  WpPipewireSimpleEndpoint *self = data;
+  GError *error = NULL;
 
-  /* Don't do anything if the endpoint has already been initialized */
+  /* return if already aborted */
   if (!self->init_task)
     return;
 
-  /* Finish the creation of the endpoint */
+  wp_proxy_sync_finish (proxy, res, &error);
-  g_task_return_boolean (self->init_task, TRUE);
+
+  if (error)
+    g_task_return_error (self->init_task, error);
+  else
+    g_task_return_boolean (self->init_task, TRUE);
+
   g_clear_object(&self->init_task);
 }
 
 static void
-on_proxy_port_created(GObject *initable, GAsyncResult *res, gpointer data)
+on_proxy_port_augmented (WpProxy *proxy, GAsyncResult *res,
+    WpPipewireSimpleEndpoint *self)
 {
-  WpPipewireSimpleEndpoint *self = data;
+  if (!proxy_safe_augment_finish (self, proxy, res))
-  WpProxyPort *proxy_port = NULL;
-
-  /* Get the proxy port */
-  proxy_port = WP_PROXY_PORT (object_safe_new_finish (self, initable, res,
-      (WpObjectNewFinishFunc)wp_proxy_port_new_finish));
-  if (!proxy_port)
     return;
 
   /* Add the proxy port to the array */
-  g_return_if_fail (self->proxies_port);
+  g_ptr_array_add(self->proxies_port, g_object_ref (proxy));
-  g_ptr_array_add(self->proxies_port, proxy_port);
 
-  /* Register the done callback */
+  /* Sync with the server and use the task data as a flag to know
-  if (!self->proxy_node_done_handler_id) {
+     whether we already called sync or not */
-    self->proxy_node_done_handler_id = g_signal_connect_object(self->proxy_node,
+  if (!g_task_get_task_data (self->init_task)) {
-        "done", (GCallback)on_all_ports_done, self, 0);
+    wp_proxy_sync (WP_PROXY(self->proxy_node), NULL,
-    wp_proxy_sync (WP_PROXY(self->proxy_node));
+        (GAsyncReadyCallback) on_all_ports_done, self);
+    g_task_set_task_data (self->init_task, GUINT_TO_POINTER (1), NULL);
   }
 }
 
 static void
-on_port_added(WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
+on_port_added(WpRemotePipewire *rp, WpProxy *proxy, gpointer d)
 {
   WpPipewireSimpleEndpoint *self = d;
-  struct pw_port_proxy *port_proxy = NULL;
-  const struct spa_dict *props = p;
   const char *s;
   guint node_id = 0;
+  g_autoptr (WpProperties) props = wp_proxy_get_global_properties (proxy);
 
   /* Don't do anything if we are aborting */
-  if (self->init_abort)
+  if (!self->init_task)
     return;
 
-  if ((s = spa_dict_lookup(props, PW_KEY_NODE_ID)))
+  if ((s = wp_properties_get (props, PW_KEY_NODE_ID)))
     node_id = atoi(s);
 
   /* Only handle ports owned by this endpoint */
-  if (node_id != self->global_id)
+  if (node_id != wp_proxy_get_global_id (WP_PROXY (self->proxy_node)))
     return;
 
-  /* Create the proxy port async */
+  /* Augment */
-  port_proxy = wp_remote_pipewire_proxy_bind (self->remote_pipewire, id,
+  wp_proxy_augment (proxy, WP_PROXY_FEATURE_PW_PROXY | WP_PROXY_FEATURE_INFO,
-    PW_TYPE_INTERFACE_Port);
+      NULL, (GAsyncReadyCallback) on_proxy_port_augmented, self);
-  g_return_if_fail(port_proxy);
-  wp_proxy_port_new(id, port_proxy, on_proxy_port_created, self);
 }
 
 static void
 emit_endpoint_ports(WpPipewireSimpleEndpoint *self)
 {
   enum pw_direction direction = wp_endpoint_get_direction (WP_ENDPOINT (self));
-  struct pw_node_proxy* node_proxy = NULL;
   struct spa_audio_info_raw format = { 0, };
   struct spa_pod *param;
   char buf[1024];
   struct spa_pod_builder pod_builder = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
 
-  /* Get the pipewire node proxy */
-  node_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy_node));
-  g_return_if_fail (node_proxy);
-
   /* The default format for audio clients */
   format.format = SPA_AUDIO_FORMAT_F32P;
   format.flags = 1;
@@ -255,22 +224,16 @@ emit_endpoint_ports(WpPipewireSimpleEndpoint *self)
       SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(param));
 
   /* Set the param profile to emit the ports */
-  pw_node_proxy_set_param(node_proxy, SPA_PARAM_PortConfig, 0, param);
+  wp_proxy_node_set_param (self->proxy_node, SPA_PARAM_PortConfig, 0, param);
 }
 
 static void
-on_proxy_node_created(GObject *initable, GAsyncResult *res, gpointer data)
+on_proxy_node_augmented (WpProxy *proxy, GAsyncResult *res, gpointer data)
 {
   WpPipewireSimpleEndpoint *self = data;
   GVariantDict d;
-  uint32_t ids[1] = { SPA_PARAM_Props };
-  uint32_t n_ids = 1;
-  struct pw_node_proxy *node_proxy = NULL;
 
-  /* Get the proxy node */
+  if (!proxy_safe_augment_finish (self, proxy, res))
-  self->proxy_node = WP_PROXY_NODE (object_safe_new_finish (self, initable,
-      res, (WpObjectNewFinishFunc)wp_proxy_node_new_finish));
-  if (!self->proxy_node)
     return;
 
   /* Set the role and target name */
@@ -282,12 +245,9 @@ on_proxy_node_created(GObject *initable, GAsyncResult *res, gpointer data)
   /* Emit the ports */
   emit_endpoint_ports(self);
 
-  /* Add a custom node proxy event listener */
+  g_signal_connect (self->proxy_node, "param", (GCallback) node_proxy_param,
-  node_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy_node));
+      self);
-  g_return_if_fail (node_proxy);
+  wp_proxy_node_subscribe_params (self->proxy_node, 1, SPA_PARAM_Props);
-  pw_node_proxy_add_listener (node_proxy, &self->node_proxy_listener,
-      &node_node_proxy_events, self);
-  pw_node_proxy_subscribe_params (node_proxy, ids, n_ids);
 
   g_variant_dict_init (&d, NULL);
   g_variant_dict_insert (&d, "id", "u", 0);
@@ -321,25 +281,20 @@ wp_simple_endpoint_init_async (GAsyncInitable *initable, int io_priority,
 {
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (initable);
   g_autoptr (WpCore) core = wp_endpoint_get_core(WP_ENDPOINT(self));
-  struct pw_node_proxy *node_proxy = NULL;
 
   /* Create the async task */
   self->init_task = g_task_new (initable, cancellable, callback, data);
 
-  /* Init the proxies_port array */
-  self->proxies_port = g_ptr_array_new_full(2, (GDestroyNotify)g_object_unref);
-
   /* Register a port_added callback */
   self->remote_pipewire = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
   g_return_if_fail(self->remote_pipewire);
   g_signal_connect_object(self->remote_pipewire, "global-added::port",
     (GCallback)on_port_added, self, 0);
 
-  /* Create the proxy node async */
+  /* Augment to get the info */
-  node_proxy = wp_remote_pipewire_proxy_bind (self->remote_pipewire,
+  wp_proxy_augment (WP_PROXY (self->proxy_node),
-      self->global_id, PW_TYPE_INTERFACE_Node);
+      WP_PROXY_FEATURE_PW_PROXY | WP_PROXY_FEATURE_INFO, cancellable,
-  g_return_if_fail(node_proxy);
+      (GAsyncReadyCallback) on_proxy_node_augmented, self);
-  wp_proxy_node_new(self->global_id, node_proxy, on_proxy_node_created, self);
 
   /* Call the parent interface */
   wp_simple_endpoint_parent_interface->init_async (initable, io_priority,
@@ -361,8 +316,8 @@ wp_simple_endpoint_async_initable_init (gpointer iface, gpointer iface_data)
 static void
 simple_endpoint_init (WpPipewireSimpleEndpoint * self)
 {
-  self->init_abort = FALSE;
   self->creation_time = (guint64) g_get_monotonic_time ();
+  self->proxies_port = g_ptr_array_new_full(2, (GDestroyNotify)g_object_unref);
 }
 
 static void
@@ -371,10 +326,7 @@ simple_endpoint_finalize (GObject * object)
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
 
   /* Destroy the proxies port */
-  if (self->proxies_port) {
+  g_clear_pointer (&self->proxies_port, g_ptr_array_unref);
-    g_ptr_array_free(self->proxies_port, TRUE);
-    self->proxies_port = NULL;
-  }
 
   /* Destroy the proxy node */
   g_clear_object(&self->proxy_node);
@@ -394,8 +346,8 @@ simple_endpoint_set_property (GObject * object, guint property_id,
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
 
   switch (property_id) {
-  case PROP_GLOBAL_ID:
+  case PROP_PROXY_NODE:
-    self->global_id = g_value_get_uint(value);
+    self->proxy_node = g_value_dup_object (value);
     break;
   case PROP_ROLE:
     g_free (self->role);
@@ -418,14 +370,15 @@ simple_endpoint_get_property (GObject * object, guint property_id,
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
 
   switch (property_id) {
-  case PROP_GLOBAL_ID:
+  case PROP_PROXY_NODE:
-    g_value_set_uint (value, self->global_id);
+    g_value_set_object (value, self->proxy_node);
     break;
   case PROP_ROLE:
     g_value_set_string (value, self->role);
     break;
   case PROP_CREATION_TIME:
     g_value_set_uint64 (value, self->creation_time);
+    break;
   case PROP_TARGET:
     g_value_set_string (value, self->target);
     break;
@@ -457,8 +410,8 @@ simple_endpoint_prepare_link (WpEndpoint * ep, guint32 stream_id,
 
   /* Set the properties */
   g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
-  g_variant_builder_add (&b, "{sv}", "node-id",
+  g_variant_builder_add (&b, "{sv}", "node-id", g_variant_new_uint32 (
-      g_variant_new_uint32 (self->global_id));
+          wp_proxy_get_global_id (WP_PROXY (self->proxy_node))));
   g_variant_builder_add (&b, "{sv}", "ports", v_ports);
   *properties = g_variant_builder_end (&b);
 
@@ -490,10 +443,6 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
   struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
   float volume;
   bool mute;
-  struct pw_node_proxy *node_proxy = NULL;
-
-  /* Get the node proxy */
-  node_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy_node));
 
   switch (control_id) {
     case CONTROL_VOLUME:
@@ -502,7 +451,7 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
       g_debug("WpEndpoint:%p set volume control (%u) value, vol:%f", self,
           control_id, volume);
 
-      pw_node_proxy_set_param (node_proxy,
+      wp_proxy_node_set_param (self->proxy_node,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
@@ -516,7 +465,7 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
       g_debug("WpEndpoint:%p set mute control (%u) value, mute:%d", self,
           control_id, mute);
 
-      pw_node_proxy_set_param (node_proxy,
+      wp_proxy_node_set_param (self->proxy_node,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
@@ -546,9 +495,9 @@ simple_endpoint_class_init (WpPipewireSimpleEndpointClass * klass)
   endpoint_class->get_control_value = simple_endpoint_get_control_value;
   endpoint_class->set_control_value = simple_endpoint_set_control_value;
 
-  g_object_class_install_property (object_class, PROP_GLOBAL_ID,
+  g_object_class_install_property (object_class, PROP_PROXY_NODE,
-      g_param_spec_uint ("global-id", "global-id",
+      g_param_spec_object ("proxy-node", "proxy-node",
-          "The global Id this endpoint refers to", 0, G_MAXUINT, 0,
+          "The node this endpoint refers to", WP_TYPE_PROXY_NODE,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
   g_object_class_install_property (object_class, PROP_ROLE,
       g_param_spec_string ("role", "role", "The role of the wrapped node", NULL,
@@ -569,7 +518,7 @@ simple_endpoint_factory (WpFactory * factory, GType type,
   g_autoptr (WpCore) core = NULL;
   const gchar *name, *media_class;
   guint direction;
-  guint global_id;
+  WpProxy *node;
 
   /* Make sure the type is correct */
   g_return_if_fail (type == WP_TYPE_ENDPOINT);
@@ -585,7 +534,7 @@ simple_endpoint_factory (WpFactory * factory, GType type,
       return;
   if (!g_variant_lookup (properties, "direction", "u", &direction))
       return;
-  if (!g_variant_lookup (properties, "global-id", "u", &global_id))
+  if (!g_variant_lookup (properties, "proxy-node", "t", &node))
       return;
 
   g_async_initable_new_async (
@@ -594,6 +543,6 @@ simple_endpoint_factory (WpFactory * factory, GType type,
       "name", name,
       "media-class", media_class,
       "direction", direction,
-      "global-id", global_id,
+      "proxy-node", node,
       NULL);
 }

modules/module-pw-alsa-udev.c

@@ -54,9 +54,7 @@ struct node {
   struct spa_list link;
   uint32_t id;
 
-  struct pw_properties *props;
+  WpProxy *proxy;
-
-  struct pw_proxy *proxy;
   struct spa_node *node;
 };
 
@@ -64,47 +62,45 @@ static void
 on_endpoint_created(GObject *initable, GAsyncResult *res, gpointer d)
 {
   struct impl *impl = d;
-  WpEndpoint *endpoint = NULL;
+  g_autoptr (WpEndpoint) endpoint = NULL;
+  g_autoptr (WpProxy) proxy = NULL;
   guint global_id = 0;
   GError *error = NULL;
 
   /* Get the endpoint */
-  endpoint = wp_endpoint_new_finish(initable, res, NULL);
+  endpoint = wp_endpoint_new_finish(initable, res, &error);
-  g_return_if_fail (endpoint);
-
-  /* Check for error */
   if (error) {
-    g_clear_object (&endpoint);
     g_warning ("Failed to create alsa endpoint: %s", error->message);
     return;
   }
 
   /* Get the endpoint global id */
-  g_object_get (endpoint, "global-id", &global_id, NULL);
+  g_object_get (endpoint, "proxy-node", &proxy, NULL);
+  global_id = wp_proxy_get_global_id (proxy);
+
   g_debug ("Created alsa endpoint for global id %d", global_id);
 
   /* Register the endpoint and add it to the table */
   wp_endpoint_register (endpoint);
   g_hash_table_insert (impl->registered_endpoints, GUINT_TO_POINTER(global_id),
-      endpoint);
+      g_steal_pointer (&endpoint));
 }
 
 static void
-on_node_added(WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
+on_node_added(WpRemotePipewire *rp, WpProxy *proxy, struct impl *impl)
 {
-  struct impl *impl = d;
-  const struct spa_dict *props = p;
   g_autoptr (WpCore) core = wp_module_get_core (impl->module);
   const gchar *media_class, *name;
   enum pw_direction direction;
   GVariantBuilder b;
+  g_autoptr (WpProperties) props = NULL;
   g_autoptr (GVariant) endpoint_props = NULL;
 
-  /* Make sure the node has properties */
+  props = wp_proxy_get_global_properties (proxy);
   g_return_if_fail(props);
 
   /* Get the media_class */
-  media_class = spa_dict_lookup(props, "media.class");
+  media_class = wp_properties_get (props, PW_KEY_MEDIA_CLASS);
 
   /* Make sure the media class is non-convert audio */
   if (!g_str_has_prefix (media_class, "Audio/"))
@@ -113,9 +109,9 @@ on_node_added(WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
     return;
 
   /* Get the name */
-  name = spa_dict_lookup (props, "media.name");
+  name = wp_properties_get (props, PW_KEY_MEDIA_NAME);
   if (!name)
-    name = spa_dict_lookup (props, "node.name");
+    name = wp_properties_get (props, PW_KEY_NODE_NAME);
 
   /* Don't handle bluetooth nodes */
   if (g_str_has_prefix (name, "api.bluez5"))
@@ -140,7 +136,7 @@ on_node_added(WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
   g_variant_builder_add (&b, "{sv}",
       "direction", g_variant_new_uint32 (direction));
   g_variant_builder_add (&b, "{sv}",
-      "global-id", g_variant_new_uint32 (id));
+      "proxy-node", g_variant_new_uint64 ((guint64) proxy));
   g_variant_builder_add (&b, "{sv}",
       "streams", impl->streams);
   endpoint_props = g_variant_builder_end (&b);
@@ -151,10 +147,10 @@ on_node_added(WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
 }
 
 static void
-on_global_removed (WpRemotePipewire *rp, guint id, gpointer d)
+on_global_removed (WpRemotePipewire *rp, WpProxy *proxy, struct impl *impl)
 {
-  struct impl *impl = d;
   WpEndpoint *endpoint = NULL;
+  guint32 id = wp_proxy_get_global_id (proxy);
 
   /* Get the endpoint */
   endpoint = g_hash_table_lookup (impl->registered_endpoints,
@@ -173,6 +169,7 @@ create_node(struct impl *impl, struct device *dev, uint32_t id,
 {
   struct node *node;
   const char *str;
+  g_autoptr (WpProperties) props = NULL;
 
   /* Check if the type is a node */
   if (info->type != SPA_TYPE_INTERFACE_Node)
@@ -182,25 +179,27 @@ create_node(struct impl *impl, struct device *dev, uint32_t id,
   node = g_slice_new0(struct node);
 
   /* Set the node properties */
-  node->props = pw_properties_copy(dev->props);
+  props = wp_properties_new_copy (dev->props);
-  pw_properties_update(node->props, info->props);
+
-  str = pw_properties_get(dev->props, SPA_KEY_DEVICE_NICK);
+  str = wp_properties_get (props, SPA_KEY_DEVICE_NICK);
   if (str == NULL)
-    str = pw_properties_get(dev->props, SPA_KEY_DEVICE_NAME);
+    str = wp_properties_get (props, SPA_KEY_DEVICE_NAME);
   if (str == NULL)
-    str = pw_properties_get(dev->props, SPA_KEY_DEVICE_ALIAS);
+    str = wp_properties_get (props, SPA_KEY_DEVICE_ALIAS);
   if (str == NULL)
     str = "alsa-device";
-  pw_properties_set(node->props, PW_KEY_NODE_NAME, str);
+
-  pw_properties_set(node->props, "factory.name", info->factory_name);
+  wp_properties_update_from_dict (props, info->props);
-  pw_properties_set(node->props, "merger.monitor", "1");
+  wp_properties_set(props, PW_KEY_NODE_NAME, str);
+  wp_properties_set(props, "factory.name", info->factory_name);
+  wp_properties_set(props, "merger.monitor", "1");
 
   /* Set the node info */
   node->impl = impl;
   node->device = dev;
   node->id = id;
-  node->proxy = wp_remote_pipewire_create_object(impl->remote_pipewire,
+  node->proxy = wp_remote_pipewire_create_object (impl->remote_pipewire,
-      "adapter", PW_TYPE_INTERFACE_Node, &node->props->dict);
+      "adapter", PW_TYPE_INTERFACE_Node, PW_VERSION_NODE_PROXY, props);
   if (!node->proxy) {
     g_slice_free (struct node, node);
     return NULL;
@@ -216,8 +215,6 @@ static void
 update_node(struct impl *impl, struct device *dev, struct node *node,
     const struct spa_device_object_info *info)
 {
-  /* Just update the properties */
-  pw_properties_update(node->props, info->props);
 }
 
 static void destroy_node(struct impl *impl, struct device *dev, struct node *node)
@@ -226,7 +223,7 @@ static void destroy_node(struct impl *impl, struct device *dev, struct node *nod
   spa_list_remove(&node->link);
 
   /* Destroy the proxy node */
-  pw_proxy_destroy(node->proxy);
+  g_clear_object (&node->proxy);
 
   /* Destroy the node */
   g_slice_free (struct node, node);

modules/module-pw-audio-client.c

@@ -25,47 +25,44 @@ static void
 on_endpoint_created(GObject *initable, GAsyncResult *res, gpointer d)
 {
   struct module_data *data = d;
-  WpEndpoint *endpoint = NULL;
+  g_autoptr (WpEndpoint) endpoint = NULL;
+  g_autoptr (WpProxy) proxy = NULL;
   guint global_id = 0;
   GError *error = NULL;
 
   /* Get the endpoint */
-  endpoint = wp_endpoint_new_finish(initable, res, NULL);
+  endpoint = wp_endpoint_new_finish(initable, res, &error);
-  g_return_if_fail (endpoint);
-
-  /* Check for error */
   if (error) {
-    g_clear_object (&endpoint);
     g_warning ("Failed to create client endpoint: %s", error->message);
     return;
   }
 
   /* Get the endpoint global id */
-  g_object_get (endpoint, "global-id", &global_id, NULL);
+  g_object_get (endpoint, "proxy-node", &proxy, NULL);
+  global_id = wp_proxy_get_global_id (proxy);
+
   g_debug ("Created client endpoint for global id %d", global_id);
 
   /* Register the endpoint and add it to the table */
   wp_endpoint_register (endpoint);
   g_hash_table_insert (data->registered_endpoints, GUINT_TO_POINTER(global_id),
-      endpoint);
+      g_steal_pointer (&endpoint));
 }
 
 static void
-on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
+on_node_added (WpRemotePipewire *rp, WpProxy *proxy, gpointer d)
 {
   struct module_data *data = d;
-  const struct spa_dict *props = p;
   g_autoptr (WpCore) core = wp_module_get_core (data->module);
   const gchar *name, *media_class;
   enum pw_direction direction;
   GVariantBuilder b;
   g_autoptr (GVariant) endpoint_props = NULL;
-
+  guint32 id = wp_proxy_get_global_id (proxy);
-  /* Make sure the node has properties */
+  g_autoptr (WpProperties) props = wp_proxy_get_global_properties (proxy);
-  g_return_if_fail(props);
 
   /* Get the media_class */
-  media_class = spa_dict_lookup(props, "media.class");
+  media_class = wp_properties_get (props, PW_KEY_MEDIA_CLASS);
 
   /* Only handle client Stream nodes */
   if (!g_str_has_prefix (media_class, "Stream/"))
@@ -82,9 +79,9 @@ on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
   }
 
   /* Get the name */
-  name = spa_dict_lookup (props, "media.name");
+  name = wp_properties_get (props, PW_KEY_MEDIA_NAME);
   if (!name)
-    name = spa_dict_lookup (props, "node.name");
+    name = wp_properties_get (props, PW_KEY_NODE_NAME);
 
   /* Set the properties */
   g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
@@ -97,7 +94,7 @@ on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
   g_variant_builder_add (&b, "{sv}",
       "direction", g_variant_new_uint32 (direction));
   g_variant_builder_add (&b, "{sv}",
-      "global-id", g_variant_new_uint32 (id));
+      "proxy-node", g_variant_new_uint64 ((guint64) proxy));
   endpoint_props = g_variant_builder_end (&b);
 
   /* Create the endpoint async */
@@ -106,10 +103,11 @@ on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
 }
 
 static void
-on_global_removed (WpRemotePipewire *rp, guint id, gpointer d)
+on_global_removed (WpRemotePipewire *rp, WpProxy *proxy, gpointer d)
 {
   struct module_data *data = d;
   WpEndpoint *endpoint = NULL;
+  guint32 id = wp_proxy_get_global_id (proxy);
 
   /* Get the endpoint */
   endpoint = g_hash_table_lookup (data->registered_endpoints,
@@ -132,8 +130,7 @@ module_destroy (gpointer d)
   data->remote_pipewire = NULL;
 
   /* Destroy the registered endpoints table */
-  g_hash_table_unref(data->registered_endpoints);
+  g_clear_pointer (&data->registered_endpoints, g_hash_table_unref);
-  data->registered_endpoints = NULL;
 
   /* Clean up */
   g_slice_free (struct module_data, data);

modules/module-pw-audio-softdsp-endpoint.c

@@ -32,7 +32,7 @@ struct _WpPwAudioSoftdspEndpoint
   WpEndpoint parent;
 
   /* Properties */
-  guint global_id;
+  WpProxyNode *proxy_node;
   GVariant *streams;
 
   guint stream_count;
@@ -40,7 +40,6 @@ struct _WpPwAudioSoftdspEndpoint
 
   /* The task to signal the endpoint is initialized */
   GTask *init_task;
-  gboolean init_abort;
 
   /* Audio Streams */
   WpAudioStream *adapter;
@@ -49,7 +48,7 @@ struct _WpPwAudioSoftdspEndpoint
 
 enum {
   PROP_0,
-  PROP_GLOBAL_ID,
+  PROP_PROXY_NODE,
   PROP_STREAMS,
 };
 
@@ -71,28 +70,23 @@ static GObject *
 object_safe_new_finish(WpPwAudioSoftdspEndpoint * self, GObject *initable,
     GAsyncResult *res, WpObjectNewFinishFunc new_finish_func)
 {
-  GObject *object = NULL;
+  g_autoptr (GObject) object = NULL;
   GError *error = NULL;
 
   /* Return NULL if we are already aborting */
-  if (self->init_abort)
+  if (!self->init_task)
     return NULL;
 
   /* Get the object */
   object = G_OBJECT (new_finish_func (initable, res, &error));
-  g_return_val_if_fail (object, NULL);
-
-  /* Check for error */
   if (error) {
-    g_clear_object (&object);
     g_warning ("WpPwAudioSoftdspEndpoint:%p Aborting construction", self);
-    self->init_abort = TRUE;
     g_task_return_error (self->init_task, error);
     g_clear_object (&self->init_task);
     return NULL;
   }
 
-  return object;
+  return g_steal_pointer (&object);
 }
 
 static gboolean
@@ -161,8 +155,8 @@ on_audio_adapter_created(GObject *initable, GAsyncResult *res,
   WpPwAudioSoftdspEndpoint *self = data;
   enum pw_direction direction = wp_endpoint_get_direction(WP_ENDPOINT(self));
   g_autoptr (WpCore) core = wp_endpoint_get_core(WP_ENDPOINT(self));
+  g_autoptr (WpProperties) props = NULL;
   g_autofree gchar *name = NULL;
-  const struct pw_node_info *adapter_info = NULL;
   GVariantDict d;
   GVariantIter iter;
   const gchar *stream;
@@ -174,25 +168,23 @@ on_audio_adapter_created(GObject *initable, GAsyncResult *res,
   if (!self->adapter)
     return;
 
-  /* Get the adapter info */
+  props = wp_proxy_node_get_properties (self->proxy_node);
-  adapter_info = wp_audio_stream_get_info (self->adapter);
-  g_return_if_fail (adapter_info);
 
   /* Give a proper name to this endpoint based on adapter properties */
-  if (0 == g_strcmp0(spa_dict_lookup (adapter_info->props, "device.api"), "alsa")) {
+  if (0 == g_strcmp0(wp_properties_get (props, "device.api"), "alsa")) {
-    name = g_strdup_printf ("%s on %s (%s / node %d)",
+    name = g_strdup_printf ("%s on %s (%s / node %s)",
-        spa_dict_lookup (adapter_info->props, "api.alsa.pcm.name"),
+        wp_properties_get (props, "api.alsa.pcm.name"),
-        spa_dict_lookup (adapter_info->props, "api.alsa.card.name"),
+        wp_properties_get (props, "api.alsa.card.name"),
-        spa_dict_lookup (adapter_info->props, "api.alsa.path"),
+        wp_properties_get (props, "api.alsa.path"),
-        adapter_info->id);
+        wp_properties_get (props, PW_KEY_NODE_ID));
     g_object_set (self, "name", name, NULL);
   }
 
   /* Create the audio converters */
   g_variant_iter_init (&iter, self->streams);
   for (i = 0; g_variant_iter_next (&iter, "&s", &stream); i++) {
-    wp_audio_convert_new (WP_ENDPOINT(self), i, stream, direction, adapter_info,
+    wp_audio_convert_new (WP_ENDPOINT(self), i, stream, direction,
-        on_audio_convert_created, self);
+        self->proxy_node, on_audio_convert_created, self);
 
     /* Register the stream */
     g_variant_dict_init (&d, NULL);
@@ -219,6 +211,8 @@ endpoint_finalize (GObject * object)
   /* Destroy the done task */
   g_clear_object(&self->init_task);
 
+  g_clear_object(&self->proxy_node);
+
   G_OBJECT_CLASS (endpoint_parent_class)->finalize (object);
 }
 
@@ -229,8 +223,8 @@ endpoint_set_property (GObject * object, guint property_id,
   WpPwAudioSoftdspEndpoint *self = WP_PW_AUDIO_SOFTDSP_ENDPOINT (object);
 
   switch (property_id) {
-  case PROP_GLOBAL_ID:
+  case PROP_PROXY_NODE:
-    self->global_id = g_value_get_uint(value);
+    self->proxy_node = g_value_dup_object (value);
     break;
   case PROP_STREAMS:
     self->streams = g_value_dup_variant(value);
@@ -248,8 +242,8 @@ endpoint_get_property (GObject * object, guint property_id,
   WpPwAudioSoftdspEndpoint *self = WP_PW_AUDIO_SOFTDSP_ENDPOINT (object);
 
   switch (property_id) {
-  case PROP_GLOBAL_ID:
+  case PROP_PROXY_NODE:
-    g_value_set_uint (value, self->global_id);
+    g_value_set_object (value, self->proxy_node);
     break;
   case PROP_STREAMS:
     g_value_set_variant (value, self->streams);
@@ -323,7 +317,7 @@ wp_endpoint_init_async (GAsyncInitable *initable, int io_priority,
 
   /* Create the adapter proxy */
   wp_audio_adapter_new (WP_ENDPOINT(self), WP_STREAM_ID_NONE, "master",
-      direction, self->global_id, FALSE, on_audio_adapter_created, self);
+      direction, self->proxy_node, FALSE, on_audio_adapter_created, self);
 
   /* Register the selected control */
   self->selected = FALSE;
@@ -354,7 +348,6 @@ wp_endpoint_async_initable_init (gpointer iface, gpointer iface_data)
 static void
 endpoint_init (WpPwAudioSoftdspEndpoint * self)
 {
-  self->init_abort = FALSE;
   self->converters = g_ptr_array_new_with_free_func (g_object_unref);
 }
 
@@ -373,9 +366,9 @@ endpoint_class_init (WpPwAudioSoftdspEndpointClass * klass)
   endpoint_class->set_control_value = endpoint_set_control_value;
 
   /* Instal the properties */
-  g_object_class_install_property (object_class, PROP_GLOBAL_ID,
+  g_object_class_install_property (object_class, PROP_PROXY_NODE,
-      g_param_spec_uint ("global-id", "global-id",
+      g_param_spec_object ("proxy-node", "proxy-node",
-          "The global Id this endpoint refers to", 0, G_MAXUINT, 0,
+          "The node this endpoint refers to", WP_TYPE_PROXY_NODE,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (object_class, PROP_STREAMS,
@@ -392,7 +385,7 @@ endpoint_factory (WpFactory * factory, GType type, GVariant * properties,
   g_autoptr (WpCore) core = NULL;
   const gchar *name, *media_class;
   guint direction;
-  guint global_id;
+  WpProxy *node;
   g_autoptr (GVariant) streams = NULL;
 
   /* Make sure the type is correct */
@@ -409,7 +402,7 @@ endpoint_factory (WpFactory * factory, GType type, GVariant * properties,
       return;
   if (!g_variant_lookup (properties, "direction", "u", &direction))
       return;
-  if (!g_variant_lookup (properties, "global-id", "u", &global_id))
+  if (!g_variant_lookup (properties, "proxy-node", "t", &node))
       return;
   if (!(streams = g_variant_lookup_value (properties, "streams",
           G_VARIANT_TYPE ("as"))))
@@ -422,7 +415,7 @@ endpoint_factory (WpFactory * factory, GType type, GVariant * properties,
       "name", name,
       "media-class", media_class,
       "direction", direction,
-      "global-id", global_id,
+      "proxy-node", node,
       "streams", streams,
       NULL);
 }

modules/module-pw-audio-softdsp-endpoint/adapter.c

@@ -15,7 +15,6 @@
 
 enum {
   PROP_0,
-  PROP_ADAPTER_ID,
   PROP_CONVERT,
 };
 
@@ -23,17 +22,8 @@ struct _WpAudioAdapter
 {
   WpAudioStream parent;
 
-  /* The task to signal the proxy is initialized */
-  GTask *init_task;
-  gboolean init_abort;
-  gboolean ports_done;
-
   /* Props */
-  guint adapter_id;
   gboolean convert;
-
-  /* Proxies */
-  WpProxyNode *proxy;
 };
 
 static GAsyncInitableIface *wp_audio_adapter_parent_interface = NULL;
@@ -44,132 +34,59 @@ G_DEFINE_TYPE_WITH_CODE (WpAudioAdapter, wp_audio_adapter, WP_TYPE_AUDIO_STREAM,
     G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
                            wp_audio_adapter_async_initable_init))
 
-typedef GObject* (*WpObjectNewFinishFunc)(GObject *initable, GAsyncResult *res,
+static void
-    GError **error);
+on_proxy_enum_format_done (WpProxyNode *proxy, GAsyncResult *res,
-
+    WpAudioAdapter *self)
-static GObject *
-object_safe_new_finish(WpAudioAdapter * self, GObject *initable,
-    GAsyncResult *res, WpObjectNewFinishFunc new_finish_func)
 {
-  GObject *object = NULL;
+  g_autoptr (GPtrArray) formats = NULL;
-  GError *error = NULL;
+  g_autoptr (GError) error = NULL;
+  enum pw_direction direction =
+      wp_audio_stream_get_direction (WP_AUDIO_STREAM (self));
+  uint8_t buf[1024];
+  struct spa_pod_builder pod_builder = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
+  struct spa_pod *param;
+  uint32_t media_type, media_subtype;
+  struct spa_audio_info_raw fmt_raw;
 
-  /* Return NULL if we are already aborting */
+  formats = wp_proxy_node_enum_params_collect_finish (proxy, res, &error);
-  if (self->init_abort)
-    return NULL;
-
-  /* Get the object */
-  object = G_OBJECT (new_finish_func (initable, res, &error));
-  g_return_val_if_fail (object, NULL);
-
-  /* Check for error */
   if (error) {
-    g_clear_object (&object);
+    g_message("WpAudioAdapter:%p enum format error: %s", self, error->message);
-    g_warning ("WpAudioAdapter:%p Aborting construction", self);
+    wp_audio_stream_init_task_finish (WP_AUDIO_STREAM (self),
-    self->init_abort = TRUE;
+        g_steal_pointer (&error));
-    g_task_return_error (self->init_task, error);
-    g_clear_object (&self->init_task);
-    return NULL;
-  }
-
-  return object;
-}
-
-static void
-on_audio_adapter_done(WpProxy *proxy, gpointer data)
-{
-  WpAudioAdapter *self = data;
-
-  /* Emit the ports if not done and sync again */
-  if (!self->ports_done) {
-    enum pw_direction direction =
-      wp_audio_stream_get_direction ( WP_AUDIO_STREAM (self));
-    struct pw_node_proxy *pw_proxy = NULL;
-    uint8_t buf[1024];
-    struct spa_pod_builder pod_builder = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
-    struct spa_pod *param;
-
-    /* Emit the props param */
-    pw_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy));
-    pw_node_proxy_enum_params (pw_proxy, 0, SPA_PARAM_Props, 0, -1, NULL);
-
-    /* Emit the ports */
-    if (self->convert) {
-      param = spa_pod_builder_add_object(&pod_builder,
-          SPA_TYPE_OBJECT_ParamPortConfig,  SPA_PARAM_PortConfig,
-          SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(direction),
-          SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_convert));
-    } else {
-      struct spa_audio_info_raw format = *wp_proxy_node_get_format (self->proxy);
-      param = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format, &format);
-      param = spa_pod_builder_add_object(&pod_builder,
-          SPA_TYPE_OBJECT_ParamPortConfig,  SPA_PARAM_PortConfig,
-          SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(direction),
-          SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_dsp),
-          SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(param));
-    }
-    pw_node_proxy_set_param(pw_proxy, SPA_PARAM_PortConfig, 0, param);
-
-    /* Sync */
-    self->ports_done = TRUE;
-    wp_proxy_sync (WP_PROXY(self->proxy));
     return;
   }
 
-  /* Don't do anything if the audio adapter has already been initialized */
+  if (formats->len == 0 ||
-  if (!self->init_task)
+      !(param = g_ptr_array_index (formats, 0)) ||
+      spa_format_parse (param, &media_type, &media_subtype) < 0 ||
+      media_type != SPA_MEDIA_TYPE_audio ||
+      media_subtype != SPA_MEDIA_SUBTYPE_raw) {
+    g_message("WpAudioAdapter:%p node does not support audio/raw format", self);
+    wp_audio_stream_init_task_finish (WP_AUDIO_STREAM (self),
+        g_error_new (WP_DOMAIN_LIBRARY, WP_LIBRARY_ERROR_OPERATION_FAILED,
+            "node does not support audio/raw format"));
     return;
+  }
 
-  /* Finish the creation of the audio adapter */
+  /* Parse the raw audio format */
-  g_task_return_boolean (self->init_task, TRUE);
+  spa_pod_fixate (param);
-  g_clear_object(&self->init_task);
+  spa_format_audio_raw_parse (param, &fmt_raw);
-}
 
-static void
+  if (self->convert) {
-on_audio_adapter_proxy_created(GObject *initable, GAsyncResult *res,
+    param = spa_pod_builder_add_object(&pod_builder,
-    gpointer data)
+        SPA_TYPE_OBJECT_ParamPortConfig,  SPA_PARAM_PortConfig,
-{
+        SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(direction),
-  WpAudioAdapter *self = data;
+        SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_convert));
+  } else {
+    param = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format, &fmt_raw);
+    param = spa_pod_builder_add_object(&pod_builder,
+        SPA_TYPE_OBJECT_ParamPortConfig,  SPA_PARAM_PortConfig,
+        SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(direction),
+        SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_dsp),
+        SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(param));
+  }
 
-  /* Get the adapter proxy */
+  wp_audio_stream_set_port_config (WP_AUDIO_STREAM (self), param);
-  self->proxy = WP_PROXY_NODE (object_safe_new_finish (self, initable,
-      res, (WpObjectNewFinishFunc)wp_proxy_node_new_finish));
-  if (!self->proxy)
-    return;
-
-  /* Emit the EnumFormat param */
-  wp_proxy_node_enum_params (self->proxy, 0, SPA_PARAM_EnumFormat, 0, -1, NULL);
-
-  /* Register the done callback */
-  g_signal_connect_object(self->proxy, "done", (GCallback)on_audio_adapter_done,
-      self, 0);
-  wp_proxy_sync (WP_PROXY(self->proxy));
-}
-
-static gpointer
-wp_audio_adapter_create_proxy (WpAudioStream * as, WpRemotePipewire *rp)
-{
-  WpAudioAdapter * self = WP_AUDIO_ADAPTER (as);
-  struct pw_node_proxy *proxy = NULL;
-
-  /* Create the adapter proxy by binding it */
-  proxy = wp_remote_pipewire_proxy_bind (rp, self->adapter_id,
-      PW_TYPE_INTERFACE_Node);
-  g_return_val_if_fail (proxy, NULL);
-  wp_proxy_node_new(self->adapter_id, proxy, on_audio_adapter_proxy_created,
-      self);
-
-  return proxy;
-}
-
-static gconstpointer
-wp_audio_adapter_get_info (WpAudioStream * as)
-{
-  WpAudioAdapter * self = WP_AUDIO_ADAPTER (as);
-  g_return_val_if_fail (self->proxy, NULL);
-
-  /* Return the info */
-  return wp_proxy_node_get_info (self->proxy);
 }
 
 static void
@@ -177,13 +94,15 @@ wp_audio_adapter_init_async (GAsyncInitable *initable, int io_priority,
     GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
 {
   WpAudioAdapter *self = WP_AUDIO_ADAPTER(initable);
-
+  WpProxyNode *proxy = wp_audio_stream_get_proxy_node (WP_AUDIO_STREAM (self));
-  /* Create the async task */
-  self->init_task = g_task_new (initable, cancellable, callback, data);
 
   /* Call the parent interface */
+  /* This will also augment the proxy and therefore bind it */
   wp_audio_adapter_parent_interface->init_async (initable, io_priority,
       cancellable, callback, data);
+
+  wp_proxy_node_enum_params_collect (proxy, SPA_PARAM_EnumFormat, NULL, NULL,
+      (GAsyncReadyCallback) on_proxy_enum_format_done, self);
 }
 
 static void
@@ -204,9 +123,6 @@ wp_audio_adapter_set_property (GObject * object, guint property_id,
   WpAudioAdapter *self = WP_AUDIO_ADAPTER (object);
 
   switch (property_id) {
-  case PROP_ADAPTER_ID:
-    self->adapter_id = g_value_get_uint(value);
-    break;
   case PROP_CONVERT:
     self->convert = g_value_get_boolean(value);
     break;
@@ -223,9 +139,6 @@ wp_audio_adapter_get_property (GObject * object, guint property_id,
   WpAudioAdapter *self = WP_AUDIO_ADAPTER (object);
 
   switch (property_id) {
-  case PROP_ADAPTER_ID:
-    g_value_set_uint (value, self->adapter_id);
-    break;
   case PROP_CONVERT:
     g_value_set_boolean (value, self->convert);
     break;
@@ -235,42 +148,20 @@ wp_audio_adapter_get_property (GObject * object, guint property_id,
   }
 }
 
-static void
-wp_audio_adapter_finalize (GObject * object)
-{
-  WpAudioAdapter *self = WP_AUDIO_ADAPTER(object);
-
-  /* Destroy the proxy */
-  g_clear_object(&self->proxy);
-
-  G_OBJECT_CLASS (wp_audio_adapter_parent_class)->finalize (object);
-}
-
 static void
 wp_audio_adapter_init (WpAudioAdapter * self)
 {
-  self->init_abort = FALSE;
-  self->ports_done = FALSE;
 }
 
 static void
 wp_audio_adapter_class_init (WpAudioAdapterClass * klass)
 {
   GObjectClass *object_class = (GObjectClass *) klass;
-  WpAudioStreamClass *audio_stream_class = (WpAudioStreamClass *) klass;
 
-  object_class->finalize = wp_audio_adapter_finalize;
   object_class->set_property = wp_audio_adapter_set_property;
   object_class->get_property = wp_audio_adapter_get_property;
 
-  audio_stream_class->create_proxy = wp_audio_adapter_create_proxy;
-  audio_stream_class->get_info = wp_audio_adapter_get_info;
-
   /* Install the properties */
-  g_object_class_install_property (object_class, PROP_ADAPTER_ID,
-      g_param_spec_uint ("adapter-id", "adapter-id", "The Id of the adapter", 0,
-          G_MAXUINT, 0,
-          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
   g_object_class_install_property (object_class, PROP_CONVERT,
       g_param_spec_boolean ("convert", "convert", "Do convert only or not",
           FALSE,
@@ -279,7 +170,7 @@ wp_audio_adapter_class_init (WpAudioAdapterClass * klass)
 
 void
 wp_audio_adapter_new (WpEndpoint *endpoint, guint stream_id,
-    const char *stream_name, enum pw_direction direction, guint adapter_id,
+    const char *stream_name, enum pw_direction direction, WpProxyNode *node,
     gboolean convert,  GAsyncReadyCallback callback, gpointer user_data)
 {
   g_async_initable_new_async (
@@ -288,19 +179,7 @@ wp_audio_adapter_new (WpEndpoint *endpoint, guint stream_id,
       "id", stream_id,
       "name", stream_name,
       "direction", direction,
-      "adapter-id", adapter_id,
+      "proxy-node", node,
       "convert", convert,
       NULL);
 }
-
-guint
-wp_audio_adapter_get_adapter_id (WpAudioAdapter *self)
-{
-  return self->adapter_id;
-}
-
-gboolean
-wp_audio_adapter_is_convert (WpAudioAdapter *self)
-{
-  return self->convert;
-}

modules/module-pw-audio-softdsp-endpoint/adapter.h

@@ -21,12 +21,9 @@ G_DECLARE_FINAL_TYPE (WpAudioAdapter, wp_audio_adapter, WP, AUDIO_ADAPTER,
     WpAudioStream)
 
 void wp_audio_adapter_new (WpEndpoint *endpoint, guint stream_id,
-    const char *stream_name, enum pw_direction direction, guint adapter_id,
+    const char *stream_name, enum pw_direction direction, WpProxyNode *node,
     gboolean convert, GAsyncReadyCallback callback, gpointer user_data);
 
-guint wp_audio_adapter_get_adapter_id (WpAudioAdapter *self);
-gboolean wp_audio_adapter_is_convert (WpAudioAdapter *self);
-
 G_END_DECLS
 
 #endif

modules/module-pw-audio-softdsp-endpoint/convert.c

@@ -23,16 +23,11 @@ struct _WpAudioConvert
 {
   WpAudioStream parent;
 
-  /* The task to signal the audio convert is initialized */
-  GTask *init_task;
-  gboolean init_abort;
-
   /* Props */
-  const struct pw_node_info *target;
+  WpProxyNode *target;
 
   /* Proxies */
-  WpProxyNode *proxy;
+  WpProxy *link_proxy;
-  WpProxyLink *link_proxy;
 };
 
 static GAsyncInitableIface *wp_audio_convert_parent_interface = NULL;
@@ -43,140 +38,91 @@ G_DEFINE_TYPE_WITH_CODE (WpAudioConvert, wp_audio_convert, WP_TYPE_AUDIO_STREAM,
     G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
                            wp_audio_convert_async_initable_init))
 
-typedef GObject* (*WpObjectNewFinishFunc)(GObject *initable, GAsyncResult *res,
-    GError **error);
-
-static GObject *
-object_safe_new_finish(WpAudioConvert * self, GObject *initable,
-    GAsyncResult *res, WpObjectNewFinishFunc new_finish_func)
-{
-  GObject *object = NULL;
-  GError *error = NULL;
-
-  /* Return NULL if we are already aborting */
-  if (self->init_abort)
-    return NULL;
-
-  /* Get the object */
-  object = G_OBJECT (new_finish_func (initable, res, &error));
-  g_return_val_if_fail (object, NULL);
-
-  /* Check for error */
-  if (error) {
-    g_clear_object (&object);
-    g_warning ("WpAudioConvert:%p Aborting construction", self);
-    self->init_abort = TRUE;
-    g_task_return_error (self->init_task, error);
-    g_clear_object (&self->init_task);
-    return NULL;
-  }
-
-  return object;
-}
-
 static void
-on_audio_convert_done(WpProxy *proxy, gpointer data)
+on_audio_convert_running(WpAudioConvert *self)
-{
-  WpAudioConvert *self = data;
-
-  /* Don't do anything if the endpoint has already been initialized */
-  if (!self->init_task)
-      return;
-
-  /* Finish the creation of the audio convert */
-  g_task_return_boolean (self->init_task, TRUE);
-  g_clear_object(&self->init_task);
-}
-
-static void
-on_proxy_link_created(GObject *initable, GAsyncResult *res, gpointer data)
-{
-  WpAudioConvert *self = data;
-
-  /* Get the link */
-  self->link_proxy = WP_PROXY_LINK (object_safe_new_finish (self, initable,
-      res, (WpObjectNewFinishFunc)wp_proxy_link_new_finish));
-  g_return_if_fail (self->link_proxy);
-}
-
-static void
-on_audio_convert_running(WpAudioConvert *self, WpRemotePipewire *rp)
 {
+  WpRemotePipewire *rp = wp_audio_stream_get_remote (WP_AUDIO_STREAM (self));
   enum pw_direction direction =
-      wp_audio_stream_get_direction ( WP_AUDIO_STREAM (self));
+      wp_audio_stream_get_direction (WP_AUDIO_STREAM (self));
-  struct pw_properties *props;
+  g_autoptr (WpProperties) props = NULL;
-  const struct pw_node_info *info = NULL;
+  const struct pw_node_info *info = NULL, *target_info = NULL;
-  struct pw_proxy *proxy = NULL;
 
   /* Return if the node has already been linked */
   if (self->link_proxy)
     return;
 
   /* Get the info */
-  info = wp_proxy_node_get_info(self->proxy);
+  info = wp_audio_stream_get_info (WP_AUDIO_STREAM (self));
   g_return_if_fail (info);
+  target_info = wp_proxy_node_get_info (self->target);
+  g_return_if_fail (target_info);
 
   /* Create new properties */
-  props = pw_properties_new(NULL, NULL);
+  props = wp_properties_new_empty ();
 
   /* Set the new properties */
-  pw_properties_set(props, PW_KEY_LINK_PASSIVE, "true");
+  wp_properties_set (props, PW_KEY_LINK_PASSIVE, "true");
   if (direction == PW_DIRECTION_INPUT) {
-    pw_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%d", info->id);
+    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_NODE, "%d", info->id);
-    pw_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
+    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
-    pw_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%d", self->target->id);
+    wp_properties_setf (props, PW_KEY_LINK_INPUT_NODE, "%d", target_info->id);
-    pw_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
+    wp_properties_setf (props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
   } else {
-    pw_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%d", self->target->id);
+    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_NODE, "%d", target_info->id);
-    pw_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
+    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
-    pw_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%d", info->id);
+    wp_properties_setf (props, PW_KEY_LINK_INPUT_NODE, "%d", info->id);
-    pw_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
+    wp_properties_setf (props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
   }
 
   g_debug ("%p linking audio convert to target", self);
 
   /* Create the link */
-  proxy = wp_remote_pipewire_create_object(rp, "link-factory",
+  self->link_proxy = wp_remote_pipewire_create_object (rp, "link-factory",
-      PW_TYPE_INTERFACE_Link, &props->dict);
+      PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
-  wp_proxy_link_new (pw_proxy_get_id(proxy), proxy, on_proxy_link_created,
-      self);
-
-  /* Clean up */
-  pw_properties_free(props);
 }
 
 static void
-on_audio_convert_idle (WpAudioConvert *self, WpRemotePipewire *rp)
+wp_audio_convert_event_info (WpProxyNode * proxy, GParamSpec *spec,
+    WpAudioConvert * self)
 {
-  /* Clear the proxy */
+  const struct pw_node_info *info = wp_proxy_node_get_info (proxy);
-  g_clear_object (&self->link_proxy);
+
+  /* Handle the different states */
+  switch (info->state) {
+  case PW_NODE_STATE_IDLE:
+    g_clear_object (&self->link_proxy);
+    break;
+  case PW_NODE_STATE_RUNNING:
+    on_audio_convert_running (self);
+    break;
+  case PW_NODE_STATE_SUSPENDED:
+    break;
+  default:
+    break;
+  }
 }
 
 static void
-on_audio_convert_proxy_created(GObject *initable, GAsyncResult *res,
+on_audio_convert_proxy_done (WpProxy *proxy, GAsyncResult *res,
-    gpointer data)
+    WpAudioConvert *self)
 {
-  WpAudioConvert *self = data;
+  g_autoptr (GError) error = NULL;
   enum pw_direction direction =
-      wp_audio_stream_get_direction ( WP_AUDIO_STREAM (self));
+      wp_audio_stream_get_direction (WP_AUDIO_STREAM (self));
-  struct pw_node_proxy *pw_proxy = NULL;
   struct spa_audio_info_raw format;
   uint8_t buf[1024];
   struct spa_pod_builder pod_builder = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
   struct spa_pod *param;
 
-  /* Get the convert proxy */
+  wp_proxy_sync_finish (proxy, res, &error);
-  self->proxy = WP_PROXY_NODE (object_safe_new_finish (self, initable,
+  if (error) {
-      res, (WpObjectNewFinishFunc)wp_proxy_node_new_finish));
+    g_message("WpAudioConvert:%p initial sync failed: %s", self, error->message);
-  if (!self->proxy)
+    wp_audio_stream_init_task_finish (WP_AUDIO_STREAM (self),
+        g_steal_pointer (&error));
     return;
+  }
 
-  /* Get the pipewire proxy */
+  g_debug ("%s:%p setting format", G_OBJECT_TYPE_NAME (self), self);
-  pw_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy));
-  g_return_if_fail (pw_proxy);
-
-  /* Emit the props param */
-  pw_node_proxy_enum_params (pw_proxy, 0, SPA_PARAM_Props, 0, -1, NULL);
 
   /* Use the default format */
   format.format = SPA_AUDIO_FORMAT_F32P;
@@ -193,76 +139,8 @@ on_audio_convert_proxy_created(GObject *initable, GAsyncResult *res,
       SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(direction),
       SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_dsp),
       SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(param));
-  pw_node_proxy_set_param(pw_proxy, SPA_PARAM_PortConfig, 0, param);
 
-  /* Register a callback to know when all the convert ports have been emitted */
+  wp_audio_stream_set_port_config (WP_AUDIO_STREAM (self), param);
-  g_signal_connect_object(self->proxy, "done", (GCallback)on_audio_convert_done,
-      self, 0);
-  wp_proxy_sync (WP_PROXY(self->proxy));
-}
-
-static void
-wp_audio_convert_event_info (WpAudioStream * as, gconstpointer i,
-    WpRemotePipewire *rp)
-{
-  WpAudioConvert * self = WP_AUDIO_CONVERT (as);
-  const struct pw_node_info *info = i;
-
-  /* Handle the different states */
-  switch (info->state) {
-  case PW_NODE_STATE_IDLE:
-    on_audio_convert_idle (self, rp);
-    break;
-  case PW_NODE_STATE_RUNNING:
-    on_audio_convert_running (self, rp);
-    break;
-  case PW_NODE_STATE_SUSPENDED:
-    break;
-  default:
-    break;
-  }
-}
-
-static gpointer
-wp_audio_convert_create_proxy (WpAudioStream * as, WpRemotePipewire *rp)
-{
-  WpAudioConvert * self = WP_AUDIO_CONVERT (as);
-  const char *name = wp_audio_stream_get_name (as);
-  struct pw_properties *props = NULL;
-  struct pw_node_proxy *proxy = NULL;
-
-  /* Create the properties */
-  g_return_val_if_fail (self->target, NULL);
-  props = pw_properties_new_dict(self->target->props);
-  g_return_val_if_fail (props, NULL);
-  pw_properties_set(props, PW_KEY_NODE_NAME, name);
-  pw_properties_set(props, PW_KEY_MEDIA_CLASS, "Audio/Convert");
-  pw_properties_set(props, "factory.name", SPA_NAME_AUDIO_CONVERT);
-
-  /* Create the proxy async */
-  proxy = wp_remote_pipewire_create_object(rp, "spa-node-factory",
-      PW_TYPE_INTERFACE_Node, &props->dict);
-  g_return_val_if_fail (proxy, NULL);
-  wp_proxy_node_new(pw_proxy_get_id((struct pw_proxy *)proxy), proxy,
-      on_audio_convert_proxy_created, self);
-
-  /* Clean up */
-  pw_properties_free(props);
-
-  return proxy;
-}
-
-static gconstpointer
-wp_audio_convert_get_info (WpAudioStream * as)
-{
-  WpAudioConvert * self = WP_AUDIO_CONVERT (as);
-
-  /* Make sure proxy is valid */
-  if (!self->proxy)
-    return NULL;
-
-  /* Return the info */
-  return wp_proxy_node_get_info (self->proxy);
 }
 
 static void
@@ -270,13 +148,34 @@ wp_audio_convert_init_async (GAsyncInitable *initable, int io_priority,
     GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
 {
   WpAudioConvert *self = WP_AUDIO_CONVERT (initable);
+  g_autoptr (WpProxy) proxy = NULL;
+  g_autoptr (WpProperties) props = NULL;
+  WpRemotePipewire *remote =
+      wp_audio_stream_get_remote (WP_AUDIO_STREAM (self));
 
-  /* Create the async task */
+  /* Create the properties */
-  self->init_task = g_task_new (initable, cancellable, callback, data);
+  props = wp_properties_copy (wp_proxy_node_get_properties (self->target));
+  wp_properties_set (props, PW_KEY_NODE_NAME,
+      wp_audio_stream_get_name (WP_AUDIO_STREAM (self)));
+  wp_properties_set (props, PW_KEY_MEDIA_CLASS, "Audio/Convert");
+  wp_properties_set (props, "factory.name", SPA_NAME_AUDIO_CONVERT);
+
+  /* Create the proxy */
+  proxy = wp_remote_pipewire_create_object (remote, "spa-node-factory",
+      PW_TYPE_INTERFACE_Node, PW_VERSION_NODE_PROXY, props);
+  g_return_if_fail (proxy);
+
+  g_object_set (self, "proxy-node", proxy, NULL);
+  g_signal_connect_object (proxy, "notify::info",
+      (GCallback) wp_audio_convert_event_info, self, 0);
 
   /* Call the parent interface */
   wp_audio_convert_parent_interface->init_async (initable, io_priority,
       cancellable, callback, data);
+
+  /* Register a callback to be called after all the initialization is done */
+  wp_proxy_sync (proxy, NULL,
+      (GAsyncReadyCallback) on_audio_convert_proxy_done, self);
 }
 
 static void
@@ -298,7 +197,7 @@ wp_audio_convert_set_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_TARGET:
-    self->target = g_value_get_pointer(value);
+    self->target = g_value_dup_object (value);
     break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
@@ -314,7 +213,7 @@ wp_audio_convert_get_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_TARGET:
-    g_value_set_pointer (value, (gpointer)self->target);
+    g_value_set_object (value, self->target);
     break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
@@ -327,14 +226,8 @@ wp_audio_convert_finalize (GObject * object)
 {
   WpAudioConvert *self = WP_AUDIO_CONVERT (object);
 
-  /* Destroy the init task */
-  g_clear_object(&self->init_task);
-
-  /* Destroy the proxy */
-  g_clear_object(&self->proxy);
-
-  /* Destroy the link proxy */
   g_clear_object (&self->link_proxy);
+  g_clear_object (&self->target);
 
   G_OBJECT_CLASS (wp_audio_convert_parent_class)->finalize (object);
 }
@@ -342,34 +235,28 @@ wp_audio_convert_finalize (GObject * object)
 static void
 wp_audio_convert_init (WpAudioConvert * self)
 {
-  self->init_abort = FALSE;
 }
 
 static void
 wp_audio_convert_class_init (WpAudioConvertClass * klass)
 {
   GObjectClass *object_class = (GObjectClass *) klass;
-  WpAudioStreamClass *audio_stream_class = (WpAudioStreamClass *) klass;
 
   object_class->finalize = wp_audio_convert_finalize;
   object_class->set_property = wp_audio_convert_set_property;
   object_class->get_property = wp_audio_convert_get_property;
 
-  audio_stream_class->create_proxy = wp_audio_convert_create_proxy;
-  audio_stream_class->get_info = wp_audio_convert_get_info;
-  audio_stream_class->event_info = wp_audio_convert_event_info;
-
   /* Install the properties */
   g_object_class_install_property (object_class, PROP_TARGET,
-      g_param_spec_pointer ("target", "target",
+      g_param_spec_object ("target", "target", "The target device node",
-          "The target stream info",
+          WP_TYPE_PROXY_NODE,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 }
 
 void
 wp_audio_convert_new (WpEndpoint *endpoint, guint stream_id,
     const char *stream_name, enum pw_direction direction,
-    const struct pw_node_info *target, GAsyncReadyCallback callback,
+    WpProxyNode *target, GAsyncReadyCallback callback,
     gpointer user_data)
 {
   g_async_initable_new_async (
@@ -381,9 +268,3 @@ wp_audio_convert_new (WpEndpoint *endpoint, guint stream_id,
       "target", target,
       NULL);
 }
-
-const struct pw_node_info *
-wp_audio_convert_get_target (WpAudioConvert *self)
-{
-  return self->target;
-}

modules/module-pw-audio-softdsp-endpoint/convert.h

@@ -22,11 +22,9 @@ G_DECLARE_FINAL_TYPE (WpAudioConvert, wp_audio_convert, WP, AUDIO_CONVERT,
 
 void wp_audio_convert_new (WpEndpoint *endpoint, guint stream_id,
     const char *stream_name, enum pw_direction direction,
-    const struct pw_node_info *target, GAsyncReadyCallback callback,
+    WpProxyNode *target, GAsyncReadyCallback callback,
     gpointer user_data);
 
-const struct pw_node_info *wp_audio_convert_get_target (WpAudioConvert *self);
-
 G_END_DECLS
 
 #endif

modules/module-pw-audio-softdsp-endpoint/stream.c

@@ -16,18 +16,16 @@ struct _WpAudioStreamPrivate
 {
   GObject parent;
 
+  GTask *init_task;
+
   /* Props */
   GWeakRef endpoint;
   guint id;
   gchar *name;
   enum pw_direction direction;
 
-  /* Remote Pipewire */
+  /* Stream Proxy */
-  WpRemotePipewire *remote_pipewire;
+  WpProxyNode *proxy;
-
-  /* Stream Proxy and Listener */
-  struct pw_node_proxy *proxy;
-  struct spa_hook listener;
 
   /* Stream Port Proxies */
   GPtrArray *port_proxies;
@@ -43,6 +41,7 @@ enum {
   PROP_ID,
   PROP_NAME,
   PROP_DIRECTION,
+  PROP_PROXY_NODE,
 };
 
 enum {
@@ -90,80 +89,99 @@ wp_audio_stream_id_decode (guint id, guint *stream_id, guint *control_id)
     *control_id = c_id;
 }
 
+/* called once after all the ports are augmented with INFO */
 static void
-on_audio_stream_port_created(GObject *initable, GAsyncResult *res,
+on_all_ports_augmented (WpProxy *proxy, GAsyncResult *res, WpAudioStream *self)
-    gpointer data)
 {
-  WpAudioStream *self = data;
+  g_autoptr (GError) error = NULL;
-  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
-  WpProxyPort *port_proxy = NULL;
-  GError *error = NULL;
 
-  /* Get the proxy port */
+  wp_proxy_sync_finish (proxy, res, &error);
-  port_proxy = WP_PROXY_PORT (wp_proxy_port_new_finish (initable, res, &error));
-  if (!port_proxy)
-    return;
-
-  /* Check for error */
   if (error) {
-    g_warning ("WpAudioStream:%p Stream port failed on creation", self);
+    g_warning ("WpAudioStream:%p second sync failed: %s", self,
-    g_clear_object (&port_proxy);
+        error->message);
+    wp_audio_stream_init_task_finish (self, g_steal_pointer (&error));
+    return;
+  }
+
+  g_debug ("%s:%p second sync done", G_OBJECT_TYPE_NAME (self), self);
+
+  wp_audio_stream_init_task_finish (self, NULL);
+}
+
+/* called multiple times after on_port_config_done */
+static void
+on_audio_stream_port_augmented (WpProxy *port_proxy, GAsyncResult *res,
+    WpAudioStream *self)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+  g_autoptr (GError) error = NULL;
+
+  wp_proxy_augment_finish (port_proxy, res, &error);
+  if (error) {
+    g_warning ("WpAudioStream:%p Stream port failed to augment: %s", self,
+        error->message);
+    wp_audio_stream_init_task_finish (self, g_steal_pointer (&error));
     return;
   }
 
   /* Add the proxy port to the array */
-  g_return_if_fail (priv->port_proxies);
+  g_ptr_array_add(priv->port_proxies, g_object_ref (port_proxy));
-  g_ptr_array_add(priv->port_proxies, port_proxy);
 }
 
+/* called once after we have all the ports added */
 static void
-on_audio_stream_port_added(WpRemotePipewire *rp, guint id, gconstpointer p,
+on_port_config_done (WpProxy *proxy, GAsyncResult *res, WpAudioStream *self)
-    gpointer d)
 {
-  WpAudioStream *self = d;
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+  g_autoptr (GError) error = NULL;
+
+  wp_proxy_sync_finish (proxy, res, &error);
+  if (error) {
+    g_warning ("WpAudioStream:%p port config sync failed: %s", self,
+        error->message);
+    wp_audio_stream_init_task_finish (self, g_steal_pointer (&error));
+    return;
+  }
+
+  g_debug ("%s:%p port config done", G_OBJECT_TYPE_NAME (self), self);
+
+  wp_proxy_sync (WP_PROXY (priv->proxy), NULL,
+      (GAsyncReadyCallback) on_all_ports_augmented, self);
+}
+
+/* called multiple times after we set the PortConfig */
+static void
+on_audio_stream_port_added(WpRemotePipewire *rp, WpProxy *proxy,
+    WpAudioStream *self)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+  g_autoptr (WpProperties) props = wp_proxy_get_global_properties (proxy);
   const struct pw_node_info *info = NULL;
-  struct pw_proxy *proxy = NULL;
-  const struct spa_dict *props = p;
   const char *s;
   guint node_id = 0;
 
   /* Get the node id */
-  g_return_if_fail (WP_AUDIO_STREAM_GET_CLASS (self)->get_info);
+  info = wp_proxy_node_get_info (priv->proxy);
-  info = WP_AUDIO_STREAM_GET_CLASS (self)->get_info (self);
   if (!info)
     return;
 
-  if ((s = spa_dict_lookup(props, PW_KEY_NODE_ID)))
+  if ((s = wp_properties_get (props, PW_KEY_NODE_ID)))
     node_id = atoi(s);
 
   /* Skip ports that are not owned by this stream */
   if (info->id != node_id)
     return;
 
-  /* Create the port proxy async */
+  wp_proxy_augment (proxy, WP_PROXY_FEATURE_PW_PROXY | WP_PROXY_FEATURE_INFO,
-  proxy = wp_remote_pipewire_proxy_bind (rp, id, PW_TYPE_INTERFACE_Port);
+      NULL, (GAsyncReadyCallback) on_audio_stream_port_augmented, self);
-  g_return_if_fail(proxy);
-  wp_proxy_port_new(id, proxy, on_audio_stream_port_created, self);
 }
 
 static void
-audio_stream_event_info (void *object, const struct pw_node_info *info)
+audio_stream_event_param (WpProxy *proxy, int seq, uint32_t id,
+    uint32_t index, uint32_t next, const struct spa_pod *param,
+    WpAudioStream *self)
 {
-  WpAudioStream *self = object;
   WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
-
-  /* Let the derived class handle this it is implemented */
-  if (WP_AUDIO_STREAM_GET_CLASS (self)->event_info)
-    WP_AUDIO_STREAM_GET_CLASS (self)->event_info (self, info,
-        priv->remote_pipewire);
-}
-
-static void
-audio_stream_event_param (void *object, int seq, uint32_t id,
-    uint32_t index, uint32_t next, const struct spa_pod *param)
-{
-  WpAudioStreamPrivate *priv =
-      wp_audio_stream_get_instance_private (WP_AUDIO_STREAM (object));
   g_autoptr (WpEndpoint) ep = g_weak_ref_get (&priv->endpoint);
 
   switch (id) {
@@ -204,11 +222,24 @@ audio_stream_event_param (void *object, int seq, uint32_t id,
   }
 }
 
-static const struct pw_node_proxy_events audio_stream_proxy_events = {
+static void
-  PW_VERSION_NODE_PROXY_EVENTS,
+on_node_proxy_augmented (WpProxy * proxy, GAsyncResult * res,
-  .info = audio_stream_event_info,
+    WpAudioStream * self)
-  .param = audio_stream_event_param,
+{
-};
+  g_autoptr (GError) error = NULL;
+
+  wp_proxy_augment_finish (proxy, res, &error);
+  if (error) {
+    g_warning ("WpAudioStream:%p Node proxy failed to augment: %s", self,
+        error->message);
+    wp_audio_stream_init_task_finish (self, g_steal_pointer (&error));
+    return;
+  }
+
+  g_signal_connect_object (proxy, "param",
+      (GCallback) audio_stream_event_param, self, 0);
+  wp_proxy_node_subscribe_params (WP_PROXY_NODE (proxy), 1, SPA_PARAM_Props);
+}
 
 static void
 wp_audio_stream_finalize (GObject * object)
@@ -220,14 +251,12 @@ wp_audio_stream_finalize (GObject * object)
   g_weak_ref_clear (&priv->endpoint);
 
   /* Clear the name */
-  g_free (priv->name);
+  g_clear_pointer (&priv->name, g_free);
-  priv->name = NULL;
 
   /* Clear the port proxies */
-  if (priv->port_proxies) {
+  g_clear_pointer (&priv->port_proxies, g_ptr_array_unref);
-    g_ptr_array_free(priv->port_proxies, TRUE);
+
-    priv->port_proxies = NULL;
+  g_clear_object (&priv->init_task);
-  }
 
   G_OBJECT_CLASS (wp_audio_stream_parent_class)->finalize (object);
 }
@@ -252,6 +281,9 @@ wp_audio_stream_set_property (GObject * object, guint property_id,
   case PROP_DIRECTION:
     priv->direction = g_value_get_uint(value);
     break;
+  case PROP_PROXY_NODE:
+    priv->proxy = g_value_dup_object (value);
+    break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
     break;
@@ -278,6 +310,9 @@ wp_audio_stream_get_property (GObject * object, guint property_id,
   case PROP_DIRECTION:
     g_value_set_uint (value, priv->direction);
     break;
+  case PROP_PROXY_NODE:
+    g_value_set_object (value, priv->proxy);
+    break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
     break;
@@ -291,14 +326,12 @@ wp_audio_stream_init_async (GAsyncInitable *initable, int io_priority,
   WpAudioStream *self = WP_AUDIO_STREAM(initable);
   WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
   g_autoptr (WpEndpoint) ep = g_weak_ref_get (&priv->endpoint);
-  g_autoptr (WpCore) core = wp_endpoint_get_core (ep);
   GVariantDict d;
 
-  /* Set the remote pipewire */
+  g_debug ("WpEndpoint:%p init stream %s (%s:%p)", ep, priv->name,
-  priv->remote_pipewire = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
+      G_OBJECT_TYPE_NAME (self), self);
 
-  /* Init the list of port proxies */
+  priv->init_task = g_task_new (initable, cancellable, callback, data);
-  priv->port_proxies = g_ptr_array_new_full(4, (GDestroyNotify)g_object_unref);
 
   /* Register the volume control */
   g_variant_dict_init (&d, NULL);
@@ -323,19 +356,14 @@ wp_audio_stream_init_async (GAsyncInitable *initable, int io_priority,
   g_variant_dict_insert (&d, "default-value", "b", priv->mute);
   wp_endpoint_register_control (ep, g_variant_dict_end (&d));
 
-  /* Create and set the proxy */
-  g_return_if_fail (WP_AUDIO_STREAM_GET_CLASS (self)->create_proxy);
-  priv->proxy = WP_AUDIO_STREAM_GET_CLASS (self)->create_proxy (self,
-      priv->remote_pipewire);
   g_return_if_fail (priv->proxy);
-
+  wp_proxy_augment (WP_PROXY (priv->proxy),
-  /* Add a custom listener */
+      WP_PROXY_FEATURE_PW_PROXY | WP_PROXY_FEATURE_INFO, NULL,
-  pw_node_proxy_add_listener(priv->proxy, &priv->listener,
+      (GAsyncReadyCallback) on_node_proxy_augmented, self);
-      &audio_stream_proxy_events, self);
 
   /* Register a port_added callback */
-  g_signal_connect_object(priv->remote_pipewire, "global-added::port",
+  g_signal_connect_object(wp_audio_stream_get_remote (self),
-      (GCallback)on_audio_stream_port_added, self, 0);
+      "global-added::port", (GCallback)on_audio_stream_port_added, self, 0);
 }
 
 static gboolean
@@ -364,6 +392,7 @@ wp_audio_stream_init (WpAudioStream * self)
   /* Controls */
   priv->volume = 1.0;
   priv->mute = FALSE;
+  priv->port_proxies = g_ptr_array_new_full(4, (GDestroyNotify)g_object_unref);
 }
 
 static void
@@ -390,6 +419,10 @@ wp_audio_stream_class_init (WpAudioStreamClass * klass)
       g_param_spec_uint ("direction", "direction",
           "The direction of the audio stream", 0, 1, 0,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, PROP_PROXY_NODE,
+      g_param_spec_object ("proxy-node", "proxy-node",
+          "The node proxy of the stream", WP_TYPE_PROXY_NODE,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 }
 
 WpAudioStream *
@@ -416,15 +449,20 @@ wp_audio_stream_get_direction (WpAudioStream * self)
   return priv->direction;
 }
 
-gconstpointer
+WpProxyNode *
+wp_audio_stream_get_proxy_node (WpAudioStream * self)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+
+  return priv->proxy;
+}
+
+const struct pw_node_info *
 wp_audio_stream_get_info (WpAudioStream * self)
 {
-  const struct pw_node_info *info = NULL;
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
 
-  g_return_val_if_fail (WP_AUDIO_STREAM_GET_CLASS (self)->get_info, NULL);
+  return wp_proxy_node_get_info (priv->proxy);
-  info = WP_AUDIO_STREAM_GET_CLASS (self)->get_info (self);
-
-  return info;
 }
 
 static void
@@ -444,8 +482,7 @@ wp_audio_stream_prepare_link (WpAudioStream * self, GVariant ** properties,
   GVariant *v_ports;
 
   /* Get the proxy node id */
-  g_return_val_if_fail (WP_AUDIO_STREAM_GET_CLASS (self)->get_info, FALSE);
+  info = wp_proxy_node_get_info (priv->proxy);
-  info = WP_AUDIO_STREAM_GET_CLASS (self)->get_info (self);
   g_return_val_if_fail (info, FALSE);
 
   /* Create a variant array with all the ports */
@@ -497,24 +534,22 @@ wp_audio_stream_set_control_value (WpAudioStream * self, guint32 control_id,
   switch (control_id) {
     case CONTROL_VOLUME:
       volume = g_variant_get_double (value);
-      pw_node_proxy_set_param (priv->proxy,
+      wp_proxy_node_set_param (priv->proxy,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
               SPA_PROP_volume, SPA_POD_Float(volume),
               NULL));
-      pw_node_proxy_enum_params (priv->proxy, 0, SPA_PARAM_Props, 0, -1, NULL);
       break;
 
     case CONTROL_MUTE:
       mute = g_variant_get_boolean (value);
-      pw_node_proxy_set_param (priv->proxy,
+      wp_proxy_node_set_param (priv->proxy,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
               SPA_PROP_mute, SPA_POD_Bool(mute),
               NULL));
-      pw_node_proxy_enum_params (priv->proxy, 0, SPA_PARAM_Props, 0, -1, NULL);
       break;
 
     default:
@@ -524,3 +559,45 @@ wp_audio_stream_set_control_value (WpAudioStream * self, guint32 control_id,
 
   return TRUE;
 }
+
+WpRemotePipewire *
+wp_audio_stream_get_remote (WpAudioStream * self)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+  g_autoptr (WpEndpoint) ep = NULL;
+  g_autoptr (WpCore) core = NULL;
+
+  ep = g_weak_ref_get (&priv->endpoint);
+  core = wp_endpoint_get_core (ep);
+
+  /* FIXME this is theoretically not safe */
+  return wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
+}
+
+void
+wp_audio_stream_init_task_finish (WpAudioStream * self, GError * err)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+  g_autoptr (GError) error = err;
+
+  if (!priv->init_task)
+    return;
+
+  if (error)
+    g_task_return_error (priv->init_task, g_steal_pointer (&error));
+  else
+    g_task_return_boolean (priv->init_task, TRUE);
+
+  g_clear_object (&priv->init_task);
+}
+
+void
+wp_audio_stream_set_port_config (WpAudioStream * self,
+    const struct spa_pod * param)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+
+  wp_proxy_node_set_param (priv->proxy, SPA_PARAM_PortConfig, 0, param);
+  wp_proxy_sync (WP_PROXY (priv->proxy), NULL,
+      (GAsyncReadyCallback) on_port_config_done, self);
+}

modules/module-pw-audio-softdsp-endpoint/stream.h

@@ -24,18 +24,14 @@ G_DECLARE_DERIVABLE_TYPE (WpAudioStream, wp_audio_stream, WP, AUDIO_STREAM, GObj
 struct _WpAudioStreamClass
 {
   GObjectClass parent_class;
-
-  /* Methods */
-  gpointer (*create_proxy) (WpAudioStream * self, WpRemotePipewire *rp);
-  gconstpointer (*get_info) (WpAudioStream * self);
-  void (*event_info) (WpAudioStream * self, gconstpointer info, WpRemotePipewire *rp);
 };
 
 WpAudioStream * wp_audio_stream_new_finish (GObject *initable,
     GAsyncResult *res, GError **error);
 const char *wp_audio_stream_get_name (WpAudioStream * self);
 enum pw_direction wp_audio_stream_get_direction (WpAudioStream * self);
-gconstpointer wp_audio_stream_get_info (WpAudioStream * self);
+WpProxyNode * wp_audio_stream_get_proxy_node (WpAudioStream * self);
+const struct pw_node_info * wp_audio_stream_get_info (WpAudioStream * self);
 gboolean wp_audio_stream_prepare_link (WpAudioStream * self,
       GVariant ** properties, GError ** error);
 GVariant * wp_audio_stream_get_control_value (WpAudioStream * self,
@@ -43,6 +39,13 @@ GVariant * wp_audio_stream_get_control_value (WpAudioStream * self,
 gboolean wp_audio_stream_set_control_value (WpAudioStream * self,
     guint32 control_id, GVariant * value);
 
+/* for subclasses */
+
+WpRemotePipewire *wp_audio_stream_get_remote (WpAudioStream * self);
+void wp_audio_stream_init_task_finish (WpAudioStream * self, GError * error);
+void wp_audio_stream_set_port_config (WpAudioStream * self,
+    const struct spa_pod * param);
+
 G_END_DECLS
 
 #endif

modules/module-pw-bluez.c

@@ -88,26 +88,26 @@ on_endpoint_created(GObject *initable, GAsyncResult *res, gpointer d)
 }
 
 static void
-on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
+on_node_added (WpRemotePipewire *rp, WpProxy *proxy, struct impl *data)
 {
-  struct impl *data = d;
-  const struct spa_dict *props = p;
   g_autoptr (WpCore) core = wp_module_get_core (data->module);
   const gchar *name, *media_class;
   enum pw_direction direction;
   GVariantBuilder b;
+  g_autoptr (WpProperties) props = NULL;
   g_autoptr (GVariant) endpoint_props = NULL;
+  guint32 id = wp_proxy_get_global_id (proxy);
 
-  /* Make sure the node has properties */
+  props = wp_proxy_get_global_properties (proxy);
   g_return_if_fail(props);
 
   /* Get the media_class */
-  media_class = spa_dict_lookup(props, "media.class");
+  media_class = wp_properties_get (props, PW_KEY_MEDIA_CLASS);
 
   /* Get the name */
-  name = spa_dict_lookup (props, "media.name");
+  name = wp_properties_get (props, PW_KEY_MEDIA_NAME);
   if (!name)
-    name = spa_dict_lookup (props, "node.name");
+    name = wp_properties_get (props, PW_KEY_NODE_NAME);
 
   /* Only handle bluetooth nodes */
   if (!g_str_has_prefix (name, "api.bluez5"))
@@ -134,7 +134,7 @@ on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
   g_variant_builder_add (&b, "{sv}",
       "direction", g_variant_new_uint32 (direction));
   g_variant_builder_add (&b, "{sv}",
-      "global-id", g_variant_new_uint32 (id));
+      "proxy-node", g_variant_new_uint64 ((guint64) proxy));
   endpoint_props = g_variant_builder_end (&b);
 
   /* Create the endpoint async */
@@ -143,10 +143,10 @@ on_node_added (WpRemotePipewire *rp, guint id, gconstpointer p, gpointer d)
 }
 
 static void
-on_global_removed (WpRemotePipewire *rp, guint id, gpointer d)
+on_global_removed (WpRemotePipewire *rp, WpProxy *proxy, struct impl *data)
 {
-  struct impl *data = d;
   WpEndpoint *endpoint = NULL;
+  guint32 id = wp_proxy_get_global_id (proxy);
 
   /* Get the endpoint */
   endpoint = g_hash_table_lookup (data->registered_endpoints,