Merge branch 'proxy-api' into 'master'

Proxy API

See merge request gkiagia/wireplumber!7

lib/wp/core.c

@@ -220,3 +220,4 @@ G_DEFINE_QUARK (endpoint, wp_global_endpoint)
 G_DEFINE_QUARK (factory, wp_global_factory)
 G_DEFINE_QUARK (module, wp_global_module)
 G_DEFINE_QUARK (policy-manager, wp_global_policy_manager)
+G_DEFINE_QUARK (proxy, wp_global_proxy)

lib/wp/core.h

@@ -60,6 +60,9 @@ GQuark wp_global_module_quark (void);
 #define WP_GLOBAL_POLICY_MANAGER (wp_global_policy_manager_quark ())
 GQuark wp_global_policy_manager_quark (void);
 
+#define WP_GLOBAL_PROXY (wp_global_proxy_quark ())
+GQuark wp_global_proxy_quark (void);
+
 G_END_DECLS
 
 #endif

lib/wp/meson.build

@@ -5,6 +5,9 @@ wp_lib_sources = [
   'factory.c',
   'module.c',
   'policy.c',
+  'proxy.c',
+  'proxy-node.c',
+  'proxy-port.c',
 ]
 
 wp_lib_headers = [
@@ -14,6 +17,9 @@ wp_lib_headers = [
   'factory.h',
   'module.h',
   'policy.h',
+  'proxy.h',
+  'proxy-node.h',
+  'proxy-port.h',
   'wp.h',
 ]
 
@@ -33,7 +39,7 @@ wp_lib = library('wireplumber-' + wireplumber_api_version,
   ],
   install: true,
   include_directories: wp_lib_include_dir,
-  dependencies : [gobject_dep, gmodule_dep],
+  dependencies : [gobject_dep, gmodule_dep, gio_dep, pipewire_dep],
   soversion: wireplumber_so_version,
   version: meson.project_version(),
 )

lib/wp/proxy-node.c

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include "proxy-node.h"
+#include <pipewire/pipewire.h>
+
+struct _WpProxyNode
+{
+  WpProxy parent;
+  
+  /* The node proxy listener */
+  struct spa_hook listener;
+
+  /* The node info */
+  struct pw_node_info *info;
+};
+
+static GAsyncInitableIface *proxy_node_parent_interface = NULL;
+static void wp_proxy_node_async_initable_init (gpointer iface,
+    gpointer iface_data);
+
+G_DEFINE_TYPE_WITH_CODE (WpProxyNode, wp_proxy_node, WP_TYPE_PROXY,
+    G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
+                           wp_proxy_node_async_initable_init))
+
+static void
+node_event_info(void *data, const struct pw_node_info *info)
+{
+  WpProxyNode *self = data;
+
+  /* Update the node info */
+  self->info = pw_node_info_update(self->info, info);
+}
+
+static const struct pw_node_proxy_events node_events = {
+  PW_VERSION_NODE_PROXY_EVENTS,
+  .info = node_event_info,
+};
+
+static void
+wp_proxy_node_finalize (GObject * object)
+{
+  WpProxyNode *self = WP_PROXY_NODE(object);
+
+  /* Remove the listener */
+  spa_hook_remove (&self->listener);
+  
+  /* Clear the info */
+  if (self->info) {
+    pw_node_info_free(self->info);
+    self->info = NULL;
+  }
+
+  G_OBJECT_CLASS (wp_proxy_node_parent_class)->finalize (object);
+}
+
+static void
+wp_proxy_node_init_async (GAsyncInitable *initable, int io_priority,
+    GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
+{
+  WpProxyNode *self = WP_PROXY_NODE(initable);
+  WpProxy *wp_proxy = WP_PROXY(initable);
+  struct pw_node_proxy *proxy = NULL;
+
+  /* Get the proxy from the base class */
+  proxy = wp_proxy_get_pw_proxy(wp_proxy);
+
+  /* Add the node proxy listener */
+  pw_node_proxy_add_listener(proxy, &self->listener, &node_events, self);
+
+  /* Call the parent interface */
+  proxy_node_parent_interface->init_async (initable, io_priority, cancellable,
+      callback, data);
+}
+
+static void
+wp_proxy_node_async_initable_init (gpointer iface, gpointer iface_data)
+{
+  GAsyncInitableIface *ai_iface = iface;
+  
+  /* Set the parent interface */
+  proxy_node_parent_interface = g_type_interface_peek_parent (iface);
+
+  /* Only set the init_async */
+  ai_iface->init_async = wp_proxy_node_init_async;
+}
+
+static void
+wp_proxy_node_init (WpProxyNode * self)
+{
+}
+
+static void
+wp_proxy_node_class_init (WpProxyNodeClass * klass)
+{
+  GObjectClass *object_class = (GObjectClass *) klass;
+
+  object_class->finalize = wp_proxy_node_finalize;
+}
+
+void
+wp_proxy_node_new (WpCore *core, gpointer proxy,
+    GAsyncReadyCallback callback, gpointer user_data)
+{
+  g_async_initable_new_async (
+      WP_TYPE_PROXY_NODE, G_PRIORITY_DEFAULT, NULL, callback, user_data,
+      "core", (gpointer) core,
+      "pw-proxy", proxy,
+      NULL);
+}
+
+WpProxyNode *
+wp_proxy_node_new_finish(GObject *initable, GAsyncResult *res, GError **error)
+{
+  GAsyncInitable *ai = G_ASYNC_INITABLE(initable);
+  return WP_PROXY_NODE(g_async_initable_new_finish(ai, res, error));
+}
+
+const struct pw_node_info *
+wp_proxy_node_get_info (WpProxyNode * self)
+{
+  return self->info;
+}

lib/wp/proxy-node.h

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#ifndef __WIREPLUMBER_PROXY_NODE_H__
+#define __WIREPLUMBER_PROXY_NODE_H__
+
+#include "core.h"
+#include "proxy.h"
+
+G_BEGIN_DECLS
+
+#define WP_TYPE_PROXY_NODE (wp_proxy_node_get_type ())
+G_DECLARE_FINAL_TYPE (WpProxyNode, wp_proxy_node, WP, PROXY_NODE, WpProxy)
+
+void wp_proxy_node_new (WpCore *core, gpointer proxy,
+    GAsyncReadyCallback callback, gpointer user_data);
+WpProxyNode *wp_proxy_node_new_finish(GObject *initable, GAsyncResult *res,
+    GError **error);
+
+const struct pw_node_info *wp_proxy_node_get_info (WpProxyNode * self);
+
+G_END_DECLS
+
+#endif

lib/wp/proxy-port.c

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include "proxy-port.h"
+#include <pipewire/pipewire.h>
+#include <spa/param/audio/format-utils.h>
+
+struct _WpProxyPort
+{
+  WpProxy parent;
+  
+  /* The port proxy listener */
+  struct spa_hook listener;
+
+  /* The port info */
+  struct pw_port_info *info;
+
+  /* The port format */
+  uint32_t media_type;
+  uint32_t media_subtype;
+  struct spa_audio_info_raw format;
+};
+
+static GAsyncInitableIface *proxy_port_parent_interface = NULL;
+static void wp_proxy_port_async_initable_init (gpointer iface,
+    gpointer iface_data);
+
+G_DEFINE_TYPE_WITH_CODE (WpProxyPort, wp_proxy_port, WP_TYPE_PROXY,
+    G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
+                           wp_proxy_port_async_initable_init))
+
+static void
+port_event_info(void *data, const struct pw_port_info *info)
+{
+  WpProxyPort *self = data;
+
+  /* Update the port info */
+  self->info = pw_port_info_update(self->info, info);
+}
+
+static void
+port_event_param(void *data, int seq, uint32_t id, uint32_t index,
+    uint32_t next, const struct spa_pod *param)
+{
+  WpProxyPort *self = data;
+
+  /* Only handle EnumFormat */
+  if (id != SPA_PARAM_EnumFormat)
+    return;
+
+  /* Parse the format */
+  spa_format_parse(param, &self->media_type, &self->media_subtype);
+
+  /* Only handle raw audio formats for now */
+  if (self->media_type != SPA_MEDIA_TYPE_audio ||
+      self->media_subtype != SPA_MEDIA_SUBTYPE_raw)
+    return;
+
+  /* Parse the raw audio format */
+  spa_pod_fixate((struct spa_pod*)param);
+  spa_format_audio_raw_parse(param, &self->format);
+}
+
+static const struct pw_port_proxy_events port_events = {
+  PW_VERSION_PORT_PROXY_EVENTS,
+  .info = port_event_info,
+  .param = port_event_param,
+};
+
+static void
+wp_proxy_port_finalize (GObject * object)
+{
+  WpProxyPort *self = WP_PROXY_PORT(object);
+
+  /* Remove the listener */
+  spa_hook_remove (&self->listener);
+
+  /* Clear the indo */
+  if (self->info) {
+    pw_port_info_free(self->info);
+    self->info = NULL;
+  }
+
+  G_OBJECT_CLASS (wp_proxy_port_parent_class)->finalize (object);
+}
+
+static void
+wp_proxy_port_init_async (GAsyncInitable *initable, int io_priority,
+    GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
+{
+  WpProxyPort *self = WP_PROXY_PORT(initable);
+  WpProxy *wp_proxy = WP_PROXY(initable);
+  struct pw_port_proxy *proxy = NULL;
+
+  /* Get the proxy from the base class */
+  proxy = wp_proxy_get_pw_proxy(wp_proxy);
+
+  /* Add the port proxy listener */
+  pw_port_proxy_add_listener(proxy, &self->listener, &port_events, self);
+
+  /* Emit the EnumFormat param */
+  pw_port_proxy_enum_params((struct pw_port_proxy*)proxy, 0,
+          SPA_PARAM_EnumFormat, 0, -1, NULL);
+
+  /* Call the parent interface */
+  proxy_port_parent_interface->init_async (initable, io_priority, cancellable,
+      callback, data);
+}
+
+static void
+wp_proxy_port_async_initable_init (gpointer iface, gpointer iface_data)
+{
+  GAsyncInitableIface *ai_iface = iface;
+  
+  /* Set the parent interface */
+  proxy_port_parent_interface = g_type_interface_peek_parent (iface);
+
+  /* Only set the init_async */
+  ai_iface->init_async = wp_proxy_port_init_async;
+}
+
+static void
+wp_proxy_port_init (WpProxyPort * self)
+{
+}
+
+static void
+wp_proxy_port_class_init (WpProxyPortClass * klass)
+{
+  GObjectClass *object_class = (GObjectClass *) klass;
+
+  object_class->finalize = wp_proxy_port_finalize;
+}
+
+void
+wp_proxy_port_new (WpCore *core, gpointer proxy,
+    GAsyncReadyCallback callback, gpointer user_data)
+{
+  g_async_initable_new_async (
+      WP_TYPE_PROXY_PORT, G_PRIORITY_DEFAULT, NULL, callback, user_data,
+      "core", (gpointer) core,
+      "pw-proxy", proxy,
+      NULL);
+}
+
+WpProxyPort *
+wp_proxy_port_new_finish(GObject *initable, GAsyncResult *res, GError **error)
+{
+  GAsyncInitable *ai = G_ASYNC_INITABLE(initable);
+  return WP_PROXY_PORT(g_async_initable_new_finish(ai, res, error));
+}
+
+const struct pw_port_info *
+wp_proxy_port_get_info (WpProxyPort * self)
+{
+  return self->info;
+}
+
+const struct spa_audio_info_raw *
+wp_proxy_port_get_format (WpProxyPort * self)
+{
+  return &self->format;
+}

lib/wp/proxy-port.h

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#ifndef __WIREPLUMBER_PROXY_PORT_H__
+#define __WIREPLUMBER_PROXY_PORT_H__
+
+#include "core.h"
+#include "proxy.h"
+
+G_BEGIN_DECLS
+
+#define WP_TYPE_PROXY_PORT (wp_proxy_port_get_type ())
+G_DECLARE_FINAL_TYPE (WpProxyPort, wp_proxy_port, WP, PROXY_PORT, WpProxy)
+
+void wp_proxy_port_new (WpCore *core, gpointer proxy,
+    GAsyncReadyCallback callback, gpointer user_data);
+WpProxyPort *wp_proxy_port_new_finish(GObject *initable, GAsyncResult *res,
+    GError **error);
+
+const struct pw_port_info *wp_proxy_port_get_info (WpProxyPort * self);
+const struct spa_audio_info_raw *wp_proxy_port_get_format (WpProxyPort * self);
+
+G_END_DECLS
+
+#endif

lib/wp/proxy.c

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <pipewire/pipewire.h>
+
+#include "proxy.h"
+
+typedef struct _WpProxyPrivate WpProxyPrivate;
+struct _WpProxyPrivate
+{
+  /* The core */
+  WpCore *core;
+
+  /* The proxy  */
+  struct pw_proxy *proxy;
+
+  /* The proxy listener */
+  struct spa_hook listener;
+
+  /* The done info */
+  GTask *done_task;
+};
+
+enum {
+  PROP_0,
+  PROP_CORE,
+  PROP_PROXY,
+};
+
+static void wp_proxy_async_initable_init (gpointer iface, gpointer iface_data);
+
+G_DEFINE_ABSTRACT_TYPE_WITH_CODE (WpProxy, wp_proxy, G_TYPE_OBJECT,
+    G_ADD_PRIVATE (WpProxy)
+    G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE, wp_proxy_async_initable_init))
+
+static void
+proxy_event_destroy (void *data)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(data));
+
+  /* Set the proxy to NULL */
+  self->proxy = NULL;
+
+  /* Remove the proxy from core */
+  wp_core_remove_global (self->core, WP_GLOBAL_PROXY, data);
+}
+
+static void
+proxy_event_done (void *data, int seq)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(data));
+
+  /* Make sure the task is valid */
+  if (!self->done_task)
+    return;
+
+  /* Execute the task */
+  g_task_return_boolean (self->done_task, TRUE);
+
+  /* Clean up */
+  g_object_unref (self->done_task);
+  self->done_task = NULL;
+}
+
+static const struct pw_proxy_events proxy_events = {
+  PW_VERSION_PROXY_EVENTS,
+  .destroy = proxy_event_destroy,
+  .done = proxy_event_done,
+};
+
+static void
+wp_proxy_finalize (GObject * object)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(object));
+  
+  /* Remove the listener */
+  spa_hook_remove (&self->listener);
+  
+  /* Destroy the proxy */
+  if (self->proxy) {
+    pw_proxy_destroy (self->proxy);
+    self->proxy = NULL;
+  }
+
+  G_OBJECT_CLASS (wp_proxy_parent_class)->finalize (object);
+}
+
+static void
+wp_proxy_set_property (GObject * object, guint property_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(object));
+
+  switch (property_id) {
+  case PROP_CORE:
+    self->core = g_value_get_pointer (value);
+    break;
+  case PROP_PROXY:
+    self->proxy = g_value_get_pointer (value);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+wp_proxy_get_property (GObject * object, guint property_id, GValue * value,
+    GParamSpec * pspec)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(object));
+
+  switch (property_id) {
+  case PROP_CORE:
+    g_value_set_pointer (value, self->core);
+    break;
+  case PROP_PROXY:
+    g_value_set_pointer (value, self->proxy);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+wp_proxy_init_async (GAsyncInitable *initable, int io_priority,
+    GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
+{
+  WpProxyPrivate *self = wp_proxy_get_instance_private (WP_PROXY(initable));
+
+  /* Create the async task */
+  self->done_task = g_task_new (initable, cancellable, callback, data);
+
+  /* Add the event listener */
+  pw_proxy_add_listener (self->proxy, &self->listener, &proxy_events, initable);
+
+  /* Trigger the done callback */
+  pw_proxy_sync(self->proxy, 0);
+}
+
+static gboolean
+wp_proxy_init_finish (GAsyncInitable *initable, GAsyncResult *result,
+    GError **error)
+{
+  g_return_val_if_fail (g_task_is_valid (result, initable), FALSE);
+
+  return g_task_propagate_boolean (G_TASK (result), error);
+}
+
+static void
+wp_proxy_async_initable_init (gpointer iface, gpointer iface_data)
+{
+  GAsyncInitableIface *ai_iface = iface;
+
+  ai_iface->init_async = wp_proxy_init_async;
+  ai_iface->init_finish = wp_proxy_init_finish;
+}
+
+static void
+wp_proxy_init (WpProxy * self)
+{
+}
+
+static void
+wp_proxy_class_init (WpProxyClass * klass)
+{
+  GObjectClass *object_class = (GObjectClass *) klass;
+
+  object_class->finalize = wp_proxy_finalize;
+  object_class->get_property = wp_proxy_get_property;
+  object_class->set_property = wp_proxy_set_property;
+
+  /* Install the properties */
+  g_object_class_install_property (object_class, PROP_CORE,
+      g_param_spec_pointer ("core", "core", "The wireplumber core",
+      G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, PROP_PROXY,
+      g_param_spec_pointer ("pw-proxy", "pw-proxy", "The pipewire proxy",
+      G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+}
+
+void
+wp_proxy_register(WpProxy * self)
+{
+  WpProxyPrivate *priv;
+  
+  g_return_if_fail (WP_IS_PROXY (self));
+  
+  priv = wp_proxy_get_instance_private (self);
+  wp_core_register_global (priv->core, WP_GLOBAL_PROXY, g_object_ref (self),
+      g_object_unref);
+}
+
+WpCore *
+wp_proxy_get_core (WpProxy * self)
+{
+  WpProxyPrivate *priv;
+
+  g_return_val_if_fail (WP_IS_PROXY (self), NULL);
+
+  priv = wp_proxy_get_instance_private (self);
+  return priv->core;
+}
+
+gpointer
+wp_proxy_get_pw_proxy (WpProxy * self)
+{
+  WpProxyPrivate *priv;
+
+  g_return_val_if_fail (WP_IS_PROXY (self), NULL);
+
+  priv = wp_proxy_get_instance_private (self);
+  return priv->proxy;
+}

lib/wp/proxy.h

+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#ifndef __WIREPLUMBER_PROXY_H__
+#define __WIREPLUMBER_PROXY_H__
+
+#include <gio/gio.h>
+
+#include "core.h"
+
+G_BEGIN_DECLS
+
+#define WP_TYPE_PROXY (wp_proxy_get_type ())
+G_DECLARE_DERIVABLE_TYPE (WpProxy, wp_proxy, WP, PROXY, GObject)
+
+/* The proxy base class */
+struct _WpProxyClass
+{
+  GObjectClass parent_class;
+};
+
+void wp_proxy_register (WpProxy * self);
+WpCore *wp_proxy_get_core (WpProxy * self);
+gpointer wp_proxy_get_pw_proxy (WpProxy * self);
+
+G_END_DECLS
+
+#endif

lib/wp/wp.h

@@ -12,3 +12,6 @@
 #include "factory.h"
 #include "module.h"
 #include "policy.h"
+#include "proxy.h"
+#include "proxy-node.h"
+#include "proxy-port.h"

modules/module-pipewire/port.h

-/* WirePlumber
- *
- * Copyright © 2019 Collabora Ltd.
- *    @author Julian Bouzas <julian.bouzas@collabora.com>
- *
- * SPDX-License-Identifier: MIT
- */
-
-#include <wp/wp.h>
-#include <pipewire/pipewire.h>
-#include <spa/param/audio/format-utils.h>
-
-struct _WpPort
-{
-  struct spa_list l;
-
-  /* Port proxy and listener */
-  struct pw_proxy *proxy;
-  struct spa_hook listener;
-
-  /* Port info */
-  uint32_t id;
-  uint32_t parent_id;
-  enum pw_direction direction;
-  uint32_t media_type;
-  uint32_t media_subtype;
-  struct pw_port_info *info;
-  struct spa_audio_info_raw format;
-};
-typedef struct _WpPort WpPort;
\ No newline at end of file

modules/module-pipewire/simple-endpoint.c

@@ -18,8 +18,6 @@
 #include <spa/pod/parser.h>
 #include <spa/param/props.h>
 
-#include "port.h"
-
 struct _WpPipewireSimpleEndpoint
 {
   WpEndpoint parent;

modules/module-pw-alsa-udev.c

@@ -15,153 +15,155 @@
 #include <pipewire/pipewire.h>
 #include <spa/param/audio/format-utils.h>
 
-#include "module-pipewire/port.h"
-
-typedef void (*WpDoneCallback)(gpointer);
-
-struct done_data {
-  WpDoneCallback callback;
-  gpointer data;
-  GDestroyNotify data_destroy;
-};
-
-struct impl {
-  WpCore *wp_core;
+struct impl
+{
+  WpCore *core;
 
   /* Remote */
   struct pw_remote *remote;
   struct spa_hook remote_listener;
 
-  /* Core */
-  struct pw_core_proxy *core_proxy;
-  struct spa_hook core_listener;
-  int core_seq;
-  GQueue *done_queue;
-
   /* Registry */
   struct pw_registry_proxy *registry_proxy;
   struct spa_hook registry_listener;
 
-  /* Ports */
-  struct spa_list port_list;
+  /* The alsa node proxies */
+  GHashTable *alsa_nodes_info;
 };
 
-struct endpoint_info {
-  struct impl *impl;
-  uint32_t id;
-  uint32_t parent_id;
+struct endpoint_info
+{
   gchar *name;
   gchar *media_class;
 };
 
-static void endpoint_info_destroy(gpointer p) {
+struct proxy_info
+{
+  const struct impl *impl;
+  uint32_t node_id;
+  WpProxyPort *proxy_port;
+};
+
+static void
+endpoint_info_destroy(gpointer p)
+{
   struct endpoint_info *ei = p;
+
+  /* Free the name */
   if (ei->name) {
     g_free (ei->name);
     ei->name = NULL;
   }
+
+  /* Free the media class */
   if (ei->media_class) {
     g_free (ei->media_class);
     ei->media_class = NULL;
   }
-  g_slice_free (struct endpoint_info, p);
-}
 
-static void done_data_destroy(gpointer p) {
-  struct done_data *dd = p;
-  if (dd->data_destroy) {
-    dd->data_destroy(dd->data);
-    dd->data = NULL;
-  }
-  g_slice_free (struct done_data, dd);
+  /* Clean up */
+  g_slice_free (struct endpoint_info, p);
 }
 
-static void sync_core_with_callabck(struct impl* impl,
-    WpDoneCallback callback, gpointer data, GDestroyNotify data_destroy) {
-  struct done_data *dd = g_new0(struct done_data, 1);
-
-  /* Set the data */
-  dd->callback = callback;
-  dd->data = data;
-  dd->data_destroy = data_destroy;
+static void
+proxy_info_destroy(gpointer p)
+{
+  struct proxy_info *pi = p;
 
-  /* Add the data to the queue */
-  g_queue_push_tail (impl->done_queue, dd);
+  /* Unref the proxy port */
+  if (pi->proxy_port) {
+    g_object_unref (pi->proxy_port);
+    pi->proxy_port = NULL;
+  }
 
-  /* Sync the core */
-  impl->core_seq = pw_core_proxy_sync(impl->core_proxy, 0, impl->core_seq);
+  /* Clean up */
+  g_slice_free (struct proxy_info, p);
 }
 
-static void create_endpoint(gpointer p) {
-  struct endpoint_info *ei = p;
-  struct spa_proxy *proxy = NULL;
+static void
+proxy_node_created(GObject *initable, GAsyncResult *res, gpointer data)
+{
+  struct proxy_info *pi = data;
+  const struct impl *impl = pi->impl;
+  WpProxyNode *proxy_node = NULL;
+  struct endpoint_info *ei = NULL;
   GVariantBuilder b;
   g_autoptr(GVariant) endpoint_props = NULL;
   WpEndpoint *endpoint = NULL;
 
-  /* Make sure the endpoint info is valid */
-  if (!ei)
+  /* Get the proxy */
+  proxy_node = wp_proxy_node_new_finish(initable, res, NULL);
+  if (!proxy_node)
     return;
 
-  /* Register the proxy */
-  proxy = pw_registry_proxy_bind (ei->impl->registry_proxy,
-      ei->id, PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0);
+  /* Register the proxy node */
+  wp_proxy_register(WP_PROXY(proxy_node));
+  
+  /* Get the alsa node info */
+  ei = g_hash_table_lookup(impl->alsa_nodes_info, GINT_TO_POINTER(pi->node_id));
+  if (!data)
+    return;
 
-  /* Build the GVariant properties for the endpoint */
+  /* Build the properties for the endpoint */
   g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
   g_variant_builder_add (&b, "{sv}", "name",
-      g_variant_new_take_string (g_strdup_printf ("Endpoint %u: %s", ei->id,
-      ei->name)));
+      g_variant_new_take_string (g_strdup_printf ("Endpoint %u: %s",
+          pi->node_id, ei->name)));
   g_variant_builder_add (&b, "{sv}",
       "media-class", g_variant_new_string (ei->media_class));
   g_variant_builder_add (&b, "{sv}",
-      "node-proxy", g_variant_new_uint64 ((guint64) proxy));
+      "proxy-node", g_variant_new_uint64 ((guint64) proxy_node));
   g_variant_builder_add (&b, "{sv}",
-      "port-list", g_variant_new_uint64 ((guint64) &ei->impl->port_list));
+      "proxy-port", g_variant_new_uint64 ((guint64)
+          g_object_ref(pi->proxy_port)));
   endpoint_props = g_variant_builder_end (&b);
 
   /* Create and register the endpoint */
-  endpoint = wp_factory_make (ei->impl->wp_core, "pw-audio-softdsp-endpoint",
+  endpoint = wp_factory_make (impl->core, "pw-audio-softdsp-endpoint",
       WP_TYPE_ENDPOINT, endpoint_props);
-  wp_endpoint_register (endpoint, ei->impl->wp_core);
-}
 
-static void enum_format_and_create_endpoint(gpointer p) {
-  struct endpoint_info *ei = p, *ei_copy = NULL;
-  WpPort *port = NULL;
+  /* Register the endpoint */
+  wp_endpoint_register (endpoint, impl->core);
 
-  /* Make sure the endpoint info is valid */
-  if (!ei)
-    return;
+  /* Clean up */
+  proxy_info_destroy (pi);
+}
 
-  /* Find the unique alsa port */
-  spa_list_for_each(port, &ei->impl->port_list, l) {
-    if (port->parent_id == ei->id)
-      break;
-  }
+static void
+proxy_port_created(GObject *initable, GAsyncResult *res, gpointer data)
+{
+  struct proxy_info *pi = data;
+  const struct impl *impl = pi->impl;
+  WpProxyPort *proxy_port = NULL;
+  struct pw_proxy *proxy = NULL;
+
+  /* Get the proxy port */
+  proxy_port = wp_proxy_port_new_finish(initable, res, NULL);
+  if (!proxy_port)
+    return;
 
-  /* Emit the port EnumFormat */
-  pw_port_proxy_enum_params((struct pw_port_proxy*)port->proxy, 0,
-          SPA_PARAM_EnumFormat, 0, -1, NULL);
+  /* Register the proxy port */
+  wp_proxy_register(WP_PROXY(proxy_port));
 
-  /* Copy endpoint info */
-  ei_copy = g_new0(struct endpoint_info, 1);
-  ei_copy->impl = ei->impl;
-  ei_copy->id = ei->id;
-  ei_copy->name = g_strdup(ei->name);
-  ei_copy->media_class = g_strdup(ei->media_class);
+  /* Forward the proxy port */
+  pi->proxy_port = proxy_port;
+  
+  /* Get the node proxy */
+  proxy = pw_registry_proxy_bind (impl->registry_proxy, pi->node_id,
+      PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0);
+  if (!proxy)
+    return;
 
-  /* Forward the endpoint creation until the port EnumFormat is emitted */
-  sync_core_with_callabck(ei->impl, create_endpoint, ei_copy,
-      endpoint_info_destroy);
+  /* Create the proxy node asynchronically */
+  wp_proxy_node_new(impl->core, proxy, proxy_node_created, pi);
 }
 
 static void
 handle_node(struct impl *impl, uint32_t id, uint32_t parent_id,
             const struct spa_dict *props)
 {
-  struct endpoint_info *ei = NULL;
   const gchar *media_class = NULL, *name = NULL;
+  struct endpoint_info *ei = NULL;
 
   /* Make sure the node has properties */
   if (!props) {
@@ -181,93 +183,46 @@ handle_node(struct impl *impl, uint32_t id, uint32_t parent_id,
 
   /* Create the endpoint info */
   ei = g_new0(struct endpoint_info, 1);
-  ei->impl = impl;
-  ei->id = id;
   ei->name = g_strdup(name);
   ei->media_class = g_strdup(media_class);
 
-  /* Delay the creation of the endpoint until all ports have been created */
-  sync_core_with_callabck(impl, enum_format_and_create_endpoint, ei,
-      endpoint_info_destroy);
-}
-
-static void port_event_info(void *data, const struct pw_port_info *info)
-{
-  WpPort *port = data;
-  port->info  = pw_port_info_update(port->info, info);
-}
-
-static void port_event_param(void *data, int seq, uint32_t id, uint32_t index,
-  uint32_t next, const struct spa_pod *param)
-{
-  WpPort *port = data;
-
-  /* Only handle EnumFormat */
-  if (id != SPA_PARAM_EnumFormat)
-    return;
-
-  /* Parse the format */
-  if (spa_format_parse(param, &port->media_type, &port->media_subtype) < 0)
-    return;
-
-  /* Only handle RAW audio types */
-  if (port->media_type != SPA_MEDIA_TYPE_audio ||
-      port->media_subtype != SPA_MEDIA_SUBTYPE_raw)
-    return;
-
-  /* Parse the raw audio format */
-  spa_pod_fixate((struct spa_pod*)param);
-  spa_format_audio_raw_parse(param, &port->format);
+  /* Insert the alsa node info in the hash table */
+  g_hash_table_insert(impl->alsa_nodes_info, GINT_TO_POINTER (id), ei);
 }
 
-static const struct pw_port_proxy_events port_events = {
-  PW_VERSION_PORT_PROXY_EVENTS,
-  .info = port_event_info,
-  .param = port_event_param,
-};
-
 static void
 handle_port(struct impl *impl, uint32_t id, uint32_t parent_id,
             const struct spa_dict *props)
 {
-  struct pw_proxy *proxy;
-  WpPort *port;
-  const char *direction_prop;
+  struct proxy_info *pi = NULL;
+  struct pw_proxy *proxy = NULL;
 
+  /* Only handle ports whose parent is an alsa node */
+  if (!g_hash_table_contains(impl->alsa_nodes_info, GINT_TO_POINTER (parent_id)))
+    return;
+  
   /* Make sure the port has porperties */
   if (!props)
     return;
 
-  /* Get the direction property */
-  direction_prop = spa_dict_lookup(props, "port.direction");
-  if (!direction_prop)
-    return;
-
-  /* Get the proxy */
+  /* Get the port proxy */
   proxy = pw_registry_proxy_bind (impl->registry_proxy, id,
-      PW_TYPE_INTERFACE_Port, PW_VERSION_NODE, sizeof(WpPort));
+      PW_TYPE_INTERFACE_Port, PW_VERSION_PORT, 0);
   if (!proxy)
     return;
 
-  /* Get the port */
-  port = pw_proxy_get_user_data(proxy);
-
-  /* Set the info */
-  port->id = id;
-  port->parent_id = parent_id;
-  port->direction =
-      !strcmp(direction_prop, "out") ? PW_DIRECTION_OUTPUT : PW_DIRECTION_INPUT;
-
-  /* Set the proxy and listener */
-  port->proxy = proxy;
-  pw_proxy_add_proxy_listener(proxy, &port->listener, &port_events, port);
+  /* Create the port info */
+  pi = g_new0(struct proxy_info, 1);
+  pi->impl = impl;
+  pi->node_id = parent_id;
+  pi->proxy_port = NULL;
 
-  /* Add the port to the list */
-  spa_list_append(&impl->port_list, &port->l);
+  /* Create the proxy port asynchronically */
+  wp_proxy_port_new(impl->core, proxy, proxy_port_created, pi);
 }
 
 static void
-registry_global(void *data,uint32_t id, uint32_t parent_id,
+registry_global(void *data, uint32_t id, uint32_t parent_id,
 		uint32_t permissions, uint32_t type, uint32_t version,
 		const struct spa_dict *props)
 {
@@ -292,35 +247,16 @@ static const struct pw_registry_proxy_events registry_events = {
   .global = registry_global,
 };
 
-static void core_done(void *d, uint32_t id, int seq)
-{
-  struct impl * impl = d;
-  struct done_data * dd = NULL;
-
-  /* Process all the done_data queue */
-  while ((dd = g_queue_pop_head(impl->done_queue))) {
-    if (dd->callback)
-      dd->callback(dd->data);
-    done_data_destroy(dd);
-  }
-}
-
-static const struct pw_core_proxy_events core_events = {
-  PW_VERSION_CORE_EVENTS,
-  .done = core_done
-};
-
 static void on_state_changed(void *_data, enum pw_remote_state old,
     enum pw_remote_state state, const char *error)
 {
   struct impl *impl = _data;
+  struct pw_core_proxy *core_proxy = NULL;
 
   switch (state) {
   case PW_REMOTE_STATE_CONNECTED:
-    impl->core_proxy = pw_remote_get_core_proxy (impl->remote);
-    pw_core_proxy_add_listener(impl->core_proxy, &impl->core_listener,
-        &core_events, impl);
-    impl->registry_proxy = pw_core_proxy_get_registry (impl->core_proxy,
+    core_proxy = pw_remote_get_core_proxy (impl->remote);
+    impl->registry_proxy = pw_core_proxy_get_registry (core_proxy,
         PW_TYPE_INTERFACE_Registry, PW_VERSION_REGISTRY, 0);
     pw_registry_proxy_add_listener(impl->registry_proxy,
         &impl->registry_listener, &registry_events, impl);
@@ -340,29 +276,46 @@ static void
 module_destroy (gpointer data)
 {
   struct impl *impl = data;
-  g_queue_free_full(impl->done_queue, done_data_destroy);
+
+  /* Destroy the hash table */
+  if (impl->alsa_nodes_info) {
+    g_hash_table_destroy(impl->alsa_nodes_info);
+    impl->alsa_nodes_info = NULL;
+  }
+
+  /* Clean up */
   g_slice_free (struct impl, impl);
 }
 
+struct impl *
+module_create (WpCore * core) {
+  /* Allocate impl */
+  struct impl *impl = g_new0(struct impl, 1);
+
+  /* Set core */
+  impl->core = core;
+
+  /* Set remote */
+  impl->remote = wp_core_get_global(core, WP_GLOBAL_PW_REMOTE);
+
+  /* Create the hash table */
+  impl->alsa_nodes_info = g_hash_table_new_full (g_direct_hash,
+      g_direct_equal, NULL, endpoint_info_destroy);
+
+  /* Add the remote listener */
+  pw_remote_add_listener(impl->remote, &impl->remote_listener, &remote_events,
+      impl);
+
+  /* Return the module */
+  return impl;
+}
+
 void
 wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
 {
-  /* This needs to create the alsa sink and alsa source nodes, but since
-   * this is already implemented in the alsa-module of pipewire, for now
-   * we just listen for the alsa nodes created by pipewire. We eventually
-   * need to move all the node creation logic here */
-
   /* Create the impl */
-  struct impl *impl = g_new0(struct impl, 1);
-  impl->wp_core = core;
-  impl->remote = wp_core_get_global(core, WP_GLOBAL_PW_REMOTE);
-  impl->done_queue = g_queue_new();
-  spa_list_init(&impl->port_list);
+  struct impl *impl = module_create (core);
 
   /* Set destroy callback for impl */
   wp_module_set_destroy_callback (module, module_destroy, impl);
-
-  /* Add a state changed listener */
-  pw_remote_add_listener(impl->remote, &impl->remote_listener, &remote_events,
-      impl);
 }

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

@@ -19,53 +19,49 @@
 #include <spa/pod/builder.h>
 #include <spa/param/props.h>
 
-#include "module-pipewire/port.h"
-
 #define MIN_QUANTUM_SIZE  64
 #define MAX_QUANTUM_SIZE  1024
 
-struct _WpPwAudioSoftdspEndpoint {
+struct _WpPwAudioSoftdspEndpoint
+{
   WpEndpoint parent;
 
   /* temporary method to select which endpoint
    * is going to be the default input/output */
   gboolean selected;
-
-  /* The core proxy */
+  
+  /* Core */
   struct pw_core_proxy *core_proxy;
 
-  /* Node proxy and listener */
-  struct pw_node_proxy *node_proxy;
-  struct spa_hook listener;
-  struct spa_hook proxy_listener;
+  /* Registry */
+  struct pw_registry_proxy *registry_proxy;
+  struct spa_hook registry_listener;
 
-  /* Node info */
-  struct pw_node_info *node_info;
-  uint32_t media_type;
-  uint32_t media_subtype;
-  struct spa_audio_info_raw format;
+  /* Direction */
   enum pw_direction direction;
 
-  /* DSP proxy and listener */
-  struct pw_node_proxy *dsp_proxy;
-  struct spa_hook dsp_listener;
+  /* Proxy */
+  WpProxyNode *proxy_node;
+  WpProxyPort *proxy_port;
 
-  /* DSP info */
-  struct pw_node_info *dsp_info;
+  /* DSP port id */
+  uint32_t dsp_port_id;
+
+  /* Volume */
   gfloat master_volume;
   gboolean master_mute;
 
-  /* Link proxy and listener */
+  /* TODO: This needs to use the new proxy API */
+  struct pw_node_proxy *dsp_proxy;
+  struct spa_hook dsp_listener;
+  struct pw_node_info *dsp_info;
   struct pw_proxy *link_proxy;
-
-  /* The all port list reference */
-  /* TODO: make it thread safe */
-  struct spa_list *port_list;
 };
 
 enum {
   PROP_0,
   PROP_NODE_PROXY,
+  PROP_PORT_PROXY,
 };
 
 enum {
@@ -79,15 +75,43 @@ G_DECLARE_FINAL_TYPE (WpPwAudioSoftdspEndpoint, endpoint,
 
 G_DEFINE_TYPE (WpPwAudioSoftdspEndpoint, endpoint, WP_TYPE_ENDPOINT)
 
+static gboolean
+endpoint_prepare_link (WpEndpoint * ep, guint32 stream_id,
+    WpEndpointLink * link, GVariant ** properties, GError ** error)
+{
+  WpPwAudioSoftdspEndpoint *self = WP_PW_AUDIO_SOFTDSP_ENDPOINT (ep);
+  GVariantBuilder b;
+  
+  /* Make sure dsp info is valid */
+  if (!self->dsp_info)
+    return FALSE;
+  
+  /* Set the properties */
+  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+  g_variant_builder_add (&b, "{sv}", "node-id",
+      g_variant_new_uint32 (self->dsp_info->id));
+  g_variant_builder_add (&b, "{sv}", "node-port-id",
+      g_variant_new_uint32 (self->dsp_port_id));
+  *properties = g_variant_builder_end (&b);
+
+  return TRUE;
+}
+
 static void
-on_dsp_running (WpPwAudioSoftdspEndpoint *self)
+on_dsp_running(WpPwAudioSoftdspEndpoint *self)
 {
   struct pw_properties *props;
+  const struct pw_node_info *node_info = NULL;
 
   /* Return if the node has already been linked */
   if (self->link_proxy)
     return;
 
+  /* Get the node info */
+  node_info = wp_proxy_node_get_info(self->proxy_node);
+  if (!node_info)
+    return;
+
   /* Create new properties */
   props = pw_properties_new(NULL, NULL);
 
@@ -96,10 +120,10 @@ on_dsp_running (WpPwAudioSoftdspEndpoint *self)
   if (self->direction == PW_DIRECTION_OUTPUT) {
     pw_properties_setf(props, PW_LINK_OUTPUT_NODE_ID, "%d", self->dsp_info->id);
     pw_properties_setf(props, PW_LINK_OUTPUT_PORT_ID, "%d", -1);
-    pw_properties_setf(props, PW_LINK_INPUT_NODE_ID, "%d", self->node_info->id);
+    pw_properties_setf(props, PW_LINK_INPUT_NODE_ID, "%d", node_info->id);
     pw_properties_setf(props, PW_LINK_INPUT_PORT_ID, "%d", -1);
   } else {
-    pw_properties_setf(props, PW_LINK_OUTPUT_NODE_ID, "%d", self->node_info->id);
+    pw_properties_setf(props, PW_LINK_OUTPUT_NODE_ID, "%d", node_info->id);
     pw_properties_setf(props, PW_LINK_OUTPUT_PORT_ID, "%d", -1);
     pw_properties_setf(props, PW_LINK_INPUT_NODE_ID, "%d", self->dsp_info->id);
     pw_properties_setf(props, PW_LINK_INPUT_PORT_ID, "%d", -1);
@@ -209,20 +233,30 @@ emit_audio_dsp_node (WpPwAudioSoftdspEndpoint *self)
   struct spa_pod_builder pod_builder = { 0, };
   struct spa_pod *param;
   uint32_t ids[1] = { SPA_PARAM_Props };
+  const struct pw_node_info *node_info;
+  const struct spa_audio_info_raw *port_format;
+  struct spa_audio_info_raw format;
+  
+  /* Get the node info */
+  node_info = wp_proxy_node_get_info(self->proxy_node);
+  if (!node_info)
+    return;
 
-  /* Return if the node has been already emitted */
-  if (self->dsp_proxy)
+  /* Get the port format */
+  port_format = wp_proxy_port_get_format(self->proxy_port);
+  if (!port_format)
     return;
+  format = *port_format;
 
   /* Get the properties */
-  props = pw_properties_new_dict(self->node_info->props);
+  props = pw_properties_new_dict(node_info->props);
   if (!props)
     return;
 
   /* Get the DSP name */
   dsp_name = pw_properties_get(props, "device.nick");
   if (!dsp_name)
-    dsp_name = self->node_info->name;
+    dsp_name = node_info->name;
 
   /* Set the properties */
   pw_properties_set(props, "audio-dsp.name", dsp_name);
@@ -239,7 +273,7 @@ emit_audio_dsp_node (WpPwAudioSoftdspEndpoint *self)
 
   /* Set DSP proxy params */
   spa_pod_builder_init(&pod_builder, buf, sizeof(buf));
-  param = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format, &self->format);
+  param = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format, &format);
   param = spa_pod_builder_add_object(&pod_builder,
       SPA_TYPE_OBJECT_ParamProfile, SPA_PARAM_Profile,
       SPA_PARAM_PROFILE_direction,  SPA_POD_Id(pw_direction_reverse(self->direction)),
@@ -251,64 +285,6 @@ emit_audio_dsp_node (WpPwAudioSoftdspEndpoint *self)
   pw_properties_free(props);
 }
 
-static void
-node_event_info (void *data, const struct pw_node_info *info)
-{
-  WpPwAudioSoftdspEndpoint *self = data;
-  WpPort *port = NULL;
-
-  /* Set the node info */
-  self->node_info = pw_node_info_update(self->node_info, info);
-
-  /* Find the node port */
-  spa_list_for_each(port, self->port_list, l) {
-    if (port->parent_id == self->node_info->id)
-      break;
-  }
-
-  /* Set the format using the port format */
-  self->format = port->format;
-
-  /* Emit the audio DSP node */
-  emit_audio_dsp_node(self);
-
-  /* TODO: Handle the different states */
-  switch (info->state) {
-  case PW_NODE_STATE_IDLE:
-    break;
-  case PW_NODE_STATE_RUNNING:
-    break;
-  case PW_NODE_STATE_SUSPENDED:
-    break;
-  default:
-    break;
-  }
-}
-
-static const struct pw_node_proxy_events node_events = {
-  PW_VERSION_NODE_PROXY_EVENTS,
-  .info = node_event_info,
-};
-
-static void
-node_proxy_destroy(void *data)
-{
-  WpPwAudioSoftdspEndpoint *self = data;
-
-  self->node_proxy = NULL;
-  wp_endpoint_unregister (WP_ENDPOINT (self));
-}
-
-static const struct pw_proxy_events node_proxy_events = {
-  PW_VERSION_PROXY_EVENTS,
-  .destroy = node_proxy_destroy,
-};
-
-static void
-endpoint_init (WpPwAudioSoftdspEndpoint * self)
-{
-}
-
 static void
 endpoint_constructed (GObject * object)
 {
@@ -325,12 +301,6 @@ endpoint_constructed (GObject * object)
     g_critical ("failed to parse direction");
   }
 
-  /* Set the node and proxy listeners */
-  pw_proxy_add_listener ((struct pw_proxy *) self->node_proxy, &self->listener,
-      &node_proxy_events, self);
-  pw_node_proxy_add_listener(self->node_proxy, &self->proxy_listener,
-      &node_events, self);
-
   g_variant_dict_init (&d, NULL);
   g_variant_dict_insert (&d, "id", "u", 0);
   g_variant_dict_insert (&d, "name", "s", "default");
@@ -369,23 +339,30 @@ endpoint_finalize (GObject * object)
 {
   WpPwAudioSoftdspEndpoint *self = WP_PW_AUDIO_SOFTDSP_ENDPOINT (object);
 
-  /* Clear node_info */
-  if (self->node_info)
-    pw_node_info_free(self->node_info);
+  /* Unref the proxy node */
+  if (self->proxy_node) {
+    g_object_unref(self->proxy_node);
+    self->proxy_node = NULL;
+  }
 
-  /* Clear dsp_info */
-  if (self->dsp_info)
-    pw_node_info_free(self->dsp_info);
+  /* Unref the proxy port */
+  if (self->proxy_port) {
+    g_object_unref(self->proxy_port);
+    self->proxy_port = NULL;
+  }
 
-  /* Remove and destroy the node_proxy */
-  if (self->node_proxy) {
-    spa_hook_remove (&self->listener);
-    pw_proxy_destroy ((struct pw_proxy *) self->node_proxy);
+  /* Clear the dsp info */
+  if (self->dsp_info) {
+    pw_node_info_free(self->dsp_info);
+    self->dsp_info = NULL;
   }
 
-  /* Remove and destroy the dsp_proxy */
-  if (self->dsp_proxy)
+  /* Destroy the dsp_proxy */
+  if (self->dsp_proxy) {
+    spa_hook_remove (&self->dsp_listener);
     pw_proxy_destroy ((struct pw_proxy *) self->dsp_proxy);
+    self->dsp_proxy = NULL;
+  }
 
   G_OBJECT_CLASS (endpoint_parent_class)->finalize (object);
 }
@@ -398,7 +375,12 @@ endpoint_set_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_NODE_PROXY:
-    self->node_proxy = g_value_get_pointer (value);
+    g_clear_object(&self->proxy_node);
+    self->proxy_node = g_value_get_object(value);
+    break;
+  case PROP_PORT_PROXY:
+    g_clear_object(&self->proxy_port);
+    self->proxy_port = g_value_get_object(value);
     break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
@@ -414,7 +396,10 @@ endpoint_get_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_NODE_PROXY:
-    g_value_set_pointer (value, self->node_proxy);
+    g_value_set_object (value, self->proxy_node);
+    break;
+  case PROP_PORT_PROXY:
+    g_value_set_object (value, self->proxy_port);
     break;
   default:
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
@@ -422,46 +407,6 @@ endpoint_get_property (GObject * object, guint property_id,
   }
 }
 
-static gboolean
-endpoint_prepare_link (WpEndpoint * ep, guint32 stream_id,
-    WpEndpointLink * link, GVariant ** properties, GError ** error)
-{
-  WpPwAudioSoftdspEndpoint *self = WP_PW_AUDIO_SOFTDSP_ENDPOINT (ep);
-  WpPort *port = NULL, *node_port = NULL, *dsp_port = NULL;
-  GVariantBuilder b;
-
-  /* Find the node port */
-  spa_list_for_each(port, self->port_list, l) {
-    if (self->node_info->id == port->parent_id) {
-      node_port = port;
-      break;
-    }
-  }
-  if (!node_port)
-    return FALSE;
-
-  /* Find the first dsp port with the same direction as the node port */
-  spa_list_for_each(port, self->port_list, l) {
-    if (self->dsp_info->id == port->parent_id
-        && port->direction == node_port->direction) {
-      dsp_port = port;
-      break;
-    }
-  }
-  if (!dsp_port)
-    return FALSE;
-
-  /* Set the properties */
-  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
-  g_variant_builder_add (&b, "{sv}", "node-id",
-      g_variant_new_uint32 (self->dsp_info->id));
-  g_variant_builder_add (&b, "{sv}", "node-port-id",
-      g_variant_new_uint32 (dsp_port->id));
-  *properties = g_variant_builder_end (&b);
-
-  return TRUE;
-}
-
 static GVariant *
 endpoint_get_control_value (WpEndpoint * ep, guint32 control_id)
 {
@@ -538,6 +483,11 @@ endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
   return TRUE;
 }
 
+static void
+endpoint_init (WpPwAudioSoftdspEndpoint * self)
+{
+}
+
 static void
 endpoint_class_init (WpPwAudioSoftdspEndpointClass * klass)
 {
@@ -554,11 +504,71 @@ endpoint_class_init (WpPwAudioSoftdspEndpointClass * klass)
   endpoint_class->set_control_value = endpoint_set_control_value;
 
   g_object_class_install_property (object_class, PROP_NODE_PROXY,
-      g_param_spec_pointer ("node-proxy", "node-proxy",
-          "Pointer to the source/sink pw_node_proxy* of the device",
+      g_param_spec_object ("node-proxy", "node-proxy",
+          "Pointer to the node proxy of the device", WP_TYPE_PROXY_NODE,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, PROP_PORT_PROXY,
+      g_param_spec_object ("port-proxy", "port-proxy",
+          "Pointer to the port ptoxy of the device", WP_TYPE_PROXY_PORT,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+}
+
+static void
+handle_port(WpPwAudioSoftdspEndpoint *self, uint32_t id, uint32_t parent_id,
+    const struct spa_dict *props)
+{
+  const char *direction_prop = NULL;
+  enum pw_direction direction;
+
+  /* Make sure the dsp port is not already set*/
+  if (self->dsp_port_id != 0)
+    return;
+
+  /* Make sure the port has porperties */
+  if (!props)
+    return;
+
+  /* Only handle ports owned by this endpoint */
+  if (!self->dsp_info || self->dsp_info->id != parent_id)
+    return;
+
+  /* Get the direction property */
+  direction_prop = spa_dict_lookup(props, "port.direction");
+  if (!direction_prop)
+    return;
+  direction =
+      !strcmp(direction_prop, "out") ? PW_DIRECTION_OUTPUT : PW_DIRECTION_INPUT;
+
+  /* Only handle ports with the oposit direction of the endpoint */
+  if (self->direction == direction)
+    return;
+  
+  /* Set the dsp port id */
+  self->dsp_port_id = id;
+}
+
+static void
+registry_global(void *data, uint32_t id, uint32_t parent_id,
+		uint32_t permissions, uint32_t type, uint32_t version,
+		const struct spa_dict *props)
+{
+  WpPwAudioSoftdspEndpoint *self = data;
+
+  switch (type) {
+  case PW_TYPE_INTERFACE_Port:
+    handle_port(self, id, parent_id, props);
+    break;
+
+  default:
+    break;
+  }
 }
 
+static const struct pw_registry_proxy_events registry_events = {
+  PW_VERSION_REGISTRY_PROXY_EVENTS,
+  .global = registry_global,
+};
+
 static gpointer
 endpoint_factory (WpFactory * factory, GType type, GVariant * properties)
 {
@@ -566,7 +576,7 @@ endpoint_factory (WpFactory * factory, GType type, GVariant * properties)
   struct pw_remote *remote;
   const gchar *name = NULL;
   const gchar *media_class = NULL;
-  guint64 proxy, port_list;
+  guint64 proxy_node, proxy_port;
 
   /* Make sure the type is not the base class type */
   if (type != WP_TYPE_ENDPOINT)
@@ -591,29 +601,36 @@ endpoint_factory (WpFactory * factory, GType type, GVariant * properties)
       return NULL;
   if (!g_variant_lookup (properties, "media-class", "&s", &media_class))
       return NULL;
-  if (!g_variant_lookup (properties, "node-proxy", "t", &proxy))
+  if (!g_variant_lookup (properties, "proxy-node", "t", &proxy_node))
       return NULL;
-  if (!g_variant_lookup (properties, "port-list", "t", &port_list))
+  if (!g_variant_lookup (properties, "proxy-port", "t", &proxy_port))
       return NULL;
 
   /* Create the softdsp endpoint object */
   WpPwAudioSoftdspEndpoint *ep = g_object_new (endpoint_get_type (),
       "name", name,
       "media-class", media_class,
-      "node-proxy", (gpointer) proxy,
+      "node-proxy", (gpointer) proxy_node,
+      "port-proxy", (gpointer) proxy_port,
       NULL);
   if (!ep)
     return NULL;
 
-  /* Set the port list reference */
-  ep->port_list = (gpointer) port_list;
-
   /* Set the core proxy */
   ep->core_proxy = pw_remote_get_core_proxy(remote);
   if (!ep->core_proxy) {
     g_warning("failed to get core proxy. Skipping...");
     return NULL;
   }
+  
+  /* Add registry listener */
+  ep->registry_proxy = pw_core_proxy_get_registry (ep->core_proxy,
+      PW_TYPE_INTERFACE_Registry, PW_VERSION_REGISTRY, 0);
+  pw_registry_proxy_add_listener(ep->registry_proxy, &ep->registry_listener,
+      &registry_events, ep);
+
+  /* Emit the audio DSP node */
+  emit_audio_dsp_node(ep);
 
   /* Return the object */
   return ep;