sofdsp-endpoint: run audioconvert in merge+split mode and use a new linking algorithm

Running audioconvert in merge+split mode is the only way to make this work with
the adapter, since the adapter does not support passing multiple channels on
a single port right now, and if it does at some point, it will be without a
mixing node on the port, which means we will not be able to mix multiple
audioconvert nodes on the same adapter. In the future we need to consider
writing a lighter volume node with multiple channels support to replace
audioconvert.

The new linking algorithm now takes into account the channel positions and makes
sure to link the correct channels together. Also, it avoids passing the port
proxies inside the GVariants, thus making the algorithm a bit more generic
and easier to unit test.

modules/meson.build

@@ -40,6 +40,7 @@ shared_library(
   'wireplumber-module-pipewire',
   [
     'module-pipewire.c',
+    'module-pipewire/multiport-link-algorithm.c',
     'module-pipewire/simple-endpoint-link.c',
     'module-pipewire/audio-softdsp-endpoint.c',
     'module-pipewire/audio-softdsp-endpoint/stream.c',

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

@@ -204,7 +204,7 @@ on_audio_adapter_created(GObject *initable, GAsyncResult *res,
   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,
-        self->proxy_node, format, on_audio_convert_created, self);
+        self->adapter, format, on_audio_convert_created, self);
 
     /* Register the stream */
     g_variant_dict_init (&d, NULL);

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

@@ -93,6 +93,7 @@ on_proxy_enum_format_done (WpProxyNode *proxy, GAsyncResult *res,
   }
 
   wp_audio_stream_set_port_config (WP_AUDIO_STREAM (self), param);
+  wp_audio_stream_finish_port_config (WP_AUDIO_STREAM (self));
 }
 
 static void

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

@@ -13,6 +13,7 @@
 #include <spa/param/props.h>
 
 #include "convert.h"
+#include "../multiport-link-algorithm.h"
 
 enum {
   PROP_0,
@@ -25,11 +26,11 @@ struct _WpAudioConvert
   WpAudioStream parent;
 
   /* Props */
-  WpProxyNode *target;
+  WpAudioStream *target;
   struct spa_audio_info_raw format;
 
   /* Proxies */
-  WpProxy *link_proxy;
+  GPtrArray *link_proxies;
 };
 
 static GAsyncInitableIface *wp_audio_convert_parent_interface = NULL;
@@ -40,47 +41,48 @@ 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))
 
+static void
+create_link_cb (WpProperties *props, gpointer user_data)
+{
+  WpAudioConvert *self = WP_AUDIO_CONVERT (user_data);
+  g_autoptr (WpCore) core = NULL;
+  WpProxy *proxy;
+
+  core = wp_audio_stream_get_core (WP_AUDIO_STREAM (self));
+  g_return_if_fail (core);
+
+  /* make the link passive, which means it will not keep
+     the audioconvert node in the running state if the number of non-passive
+     links (i.e. the ones linking another endpoint to this one) drops to 0 */
+  wp_properties_set (props, PW_KEY_LINK_PASSIVE, "1");
+
+  /* Create the link */
+  proxy = wp_core_create_remote_object(core, "link-factory",
+      PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
+  g_return_if_fail (proxy);
+  g_ptr_array_add(self->link_proxies, proxy);
+}
+
 static void
 on_audio_convert_running(WpAudioConvert *self)
 {
-  g_autoptr (WpCore) core = wp_audio_stream_get_core (WP_AUDIO_STREAM (self));
+  g_autoptr (GVariant) src_props = NULL;
+  g_autoptr (GVariant) sink_props = NULL;
+  g_autoptr (GError) error = NULL;
   enum pw_direction direction =
       wp_audio_stream_get_direction (WP_AUDIO_STREAM (self));
-  g_autoptr (WpProperties) props = NULL;
-  const struct pw_node_info *info = NULL, *target_info = NULL;
-
-  /* Return if the node has already been linked */
-  if (self->link_proxy)
-    return;
-
-  /* Get the info */
-  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 = wp_properties_new_empty ();
-
-  /* Set the new properties */
-  wp_properties_set (props, PW_KEY_LINK_PASSIVE, "true");
-  if (direction == PW_DIRECTION_INPUT) {
-    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_NODE, "%d", info->id);
-    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
-    wp_properties_setf (props, PW_KEY_LINK_INPUT_NODE, "%d", target_info->id);
-    wp_properties_setf (props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
-  } else {
-    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_NODE, "%d", target_info->id);
-    wp_properties_setf (props, PW_KEY_LINK_OUTPUT_PORT, "%d", -1);
-    wp_properties_setf (props, PW_KEY_LINK_INPUT_NODE, "%d", info->id);
-    wp_properties_setf (props, PW_KEY_LINK_INPUT_PORT, "%d", -1);
-  }
 
   g_debug ("%p linking audio convert to target", self);
 
-  /* Create the link */
-  self->link_proxy = wp_core_create_remote_object (core, "link-factory",
-      PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
+  if (direction == PW_DIRECTION_INPUT) {
+    wp_audio_stream_prepare_link (WP_AUDIO_STREAM (self), &src_props, &error);
+    wp_audio_stream_prepare_link (self->target, &sink_props, &error);
+  } else {
+    wp_audio_stream_prepare_link (self->target, &src_props, &error);
+    wp_audio_stream_prepare_link (WP_AUDIO_STREAM (self), &sink_props, &error);
+  }
+
+  multiport_link_create (src_props, sink_props, create_link_cb, self, &error);
 }
 
 static void
@@ -92,7 +94,7 @@ wp_audio_convert_event_info (WpProxyNode * proxy, GParamSpec *spec,
   /* Handle the different states */
   switch (info->state) {
   case PW_NODE_STATE_IDLE:
-    g_clear_object (&self->link_proxy);
+    g_ptr_array_set_size (self->link_proxies, 0);
     break;
   case PW_NODE_STATE_RUNNING:
     on_audio_convert_running (self);
@@ -113,6 +115,7 @@ on_audio_convert_proxy_done (WpProxy *proxy, GAsyncResult *res,
       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 *format;
   struct spa_pod *param;
 
   wp_proxy_sync_finish (proxy, res, &error);
@@ -125,15 +128,28 @@ on_audio_convert_proxy_done (WpProxy *proxy, GAsyncResult *res,
 
   g_debug ("%s:%p setting format", G_OBJECT_TYPE_NAME (self), self);
 
-  /* Emit the ports */
-  param = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format, &self->format);
+  format = spa_format_audio_raw_build(&pod_builder, SPA_PARAM_Format,
+      &self->format);
+
+  /* Configure audioconvert to be both merger and splitter; this means it will
+     have an equal number of input and output ports and just passthrough the
+     same format, but with altered volume.
+     In the future we need to consider writing a simpler volume node for this,
+     as doing merge + split is heavy for our needs */
+  param = spa_pod_builder_add_object(&pod_builder,
+      SPA_TYPE_OBJECT_ParamPortConfig,  SPA_PARAM_PortConfig,
+      SPA_PARAM_PORT_CONFIG_direction,  SPA_POD_Id(pw_direction_reverse(direction)),
+      SPA_PARAM_PORT_CONFIG_mode,       SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_dsp),
+      SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(format));
+  wp_audio_stream_set_port_config (WP_AUDIO_STREAM (self), param);
+
   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));
-
+      SPA_PARAM_PORT_CONFIG_format,     SPA_POD_Pod(format));
   wp_audio_stream_set_port_config (WP_AUDIO_STREAM (self), param);
+  wp_audio_stream_finish_port_config (WP_AUDIO_STREAM (self));
 }
 
 static void
@@ -144,10 +160,15 @@ wp_audio_convert_init_async (GAsyncInitable *initable, int io_priority,
   g_autoptr (WpProxy) proxy = NULL;
   g_autoptr (WpProperties) props = NULL;
   g_autoptr (WpCore) core = wp_audio_stream_get_core (WP_AUDIO_STREAM (self));
+  WpProxyNode *node;
 
   /* Create the properties */
-  props = wp_properties_copy (wp_proxy_node_get_properties (self->target));
-  wp_properties_set (props, PW_KEY_NODE_NAME,
+  node = wp_audio_stream_get_proxy_node (self->target);
+  props = wp_properties_copy (wp_proxy_node_get_properties (node));
+
+  wp_properties_setf (props, PW_KEY_NODE_NAME, "%s/%s/%s",
+      SPA_NAME_AUDIO_CONVERT,
+      wp_properties_get(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);
@@ -229,7 +250,7 @@ wp_audio_convert_finalize (GObject * object)
 {
   WpAudioConvert *self = WP_AUDIO_CONVERT (object);
 
-  g_clear_object (&self->link_proxy);
+  g_clear_pointer (&self->link_proxies, g_ptr_array_unref);
   g_clear_object (&self->target);
 
   G_OBJECT_CLASS (wp_audio_convert_parent_class)->finalize (object);
@@ -238,6 +259,7 @@ wp_audio_convert_finalize (GObject * object)
 static void
 wp_audio_convert_init (WpAudioConvert * self)
 {
+  self->link_proxies = g_ptr_array_new_with_free_func (g_object_unref);
 }
 
 static void
@@ -251,8 +273,8 @@ wp_audio_convert_class_init (WpAudioConvertClass * klass)
 
   /* Install the properties */
   g_object_class_install_property (object_class, PROP_TARGET,
-      g_param_spec_object ("target", "target", "The target device node",
-          WP_TYPE_PROXY_NODE,
+      g_param_spec_object ("target", "target", "The target stream",
+          WP_TYPE_AUDIO_STREAM,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
   g_object_class_install_property (object_class, PROP_FORMAT,
       g_param_spec_pointer ("format", "format", "The accepted format",
@@ -262,7 +284,7 @@ wp_audio_convert_class_init (WpAudioConvertClass * klass)
 void
 wp_audio_convert_new (WpEndpoint *endpoint, guint stream_id,
     const char *stream_name, enum pw_direction direction,
-    WpProxyNode *target, const struct spa_audio_info_raw *format,
+    WpAudioStream *target, const struct spa_audio_info_raw *format,
     GAsyncReadyCallback callback, gpointer user_data)
 {
   g_async_initable_new_async (

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

@@ -24,7 +24,7 @@ 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,
-    WpProxyNode *target, const struct spa_audio_info_raw *format,
+    WpAudioStream *target, const struct spa_audio_info_raw *format,
     GAsyncReadyCallback callback, gpointer user_data);
 
 G_END_DECLS

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

@@ -8,6 +8,8 @@
 
 #include <spa/param/props.h>
 #include <pipewire/pipewire.h>
+#include <spa/debug/types.h>
+#include <spa/param/audio/type-info.h>
 
 #include "stream.h"
 
@@ -468,8 +470,40 @@ wp_audio_stream_get_info (WpAudioStream * self)
 static void
 port_proxies_foreach_func(gpointer data, gpointer user_data)
 {
+  WpProxyPort *port = data;
   GVariantBuilder *b = user_data;
-  g_variant_builder_add (b, "t", data);
+  const struct pw_port_info *info;
+  g_autoptr (WpProperties) props = NULL;
+  const gchar *node_id, *channel;
+  uint32_t node_id_n, channel_n = SPA_AUDIO_CHANNEL_UNKNOWN;
+
+  info = wp_proxy_port_get_info (port);
+
+  props = wp_proxy_port_get_properties (port);
+  node_id = wp_properties_get (props, PW_KEY_NODE_ID);
+  g_return_if_fail (node_id);
+  node_id_n = atoi(node_id);
+
+  channel = wp_properties_get (props, PW_KEY_AUDIO_CHANNEL);
+  if (channel) {
+    const struct spa_type_info *t = spa_type_audio_channel;
+    for (; t && t->name; t++) {
+      const char *name = t->name + strlen(SPA_TYPE_INFO_AUDIO_CHANNEL_BASE);
+      if (!g_strcmp0 (channel, name)) {
+        channel_n = t->type;
+        break;
+      }
+    }
+  }
+
+  /* tuple format:
+      uint32 node_id;
+      uint32 port_id;
+      uint32 channel;  // enum spa_audio_channel
+      uint8 direction; // enum spa_direction
+   */
+  g_variant_builder_add (b, "(uuuy)", node_id_n, info->id, channel_n,
+      (guint8) info->direction);
 }
 
 gboolean
@@ -477,24 +511,11 @@ wp_audio_stream_prepare_link (WpAudioStream * self, GVariant ** properties,
     GError ** error)
 {
   WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
-  const struct pw_node_info *info = NULL;
-  GVariantBuilder b, *b_ports;
-  GVariant *v_ports;
-
-  /* Get the proxy node id */
-  info = wp_proxy_node_get_info (priv->proxy);
-  g_return_val_if_fail (info, FALSE);
+  GVariantBuilder b;
 
   /* Create a variant array with all the ports */
-  b_ports = g_variant_builder_new (G_VARIANT_TYPE ("at"));
-  g_ptr_array_foreach(priv->port_proxies, port_proxies_foreach_func, b_ports);
-  v_ports = g_variant_builder_end (b_ports);
-
-  /* Set the properties */
-  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
-  g_variant_builder_add (&b, "{sv}", "node-id",
-      g_variant_new_uint32 (info->id));
-  g_variant_builder_add (&b, "{sv}", "ports", v_ports);
+  g_variant_builder_init (&b, G_VARIANT_TYPE ("a(uuuy)"));
+  g_ptr_array_foreach(priv->port_proxies, port_proxies_foreach_func, &b);
   *properties = g_variant_builder_end (&b);
 
   return TRUE;
@@ -596,6 +617,13 @@ wp_audio_stream_set_port_config (WpAudioStream * self,
   WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
 
   wp_proxy_node_set_param (priv->proxy, SPA_PARAM_PortConfig, 0, param);
+}
+
+void
+wp_audio_stream_finish_port_config (WpAudioStream * self)
+{
+  WpAudioStreamPrivate *priv = wp_audio_stream_get_instance_private (self);
+
   wp_proxy_sync (WP_PROXY (priv->proxy), NULL,
       (GAsyncReadyCallback) on_port_config_done, self);
 }

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

@@ -45,6 +45,7 @@ WpCore *wp_audio_stream_get_core (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);
+void wp_audio_stream_finish_port_config (WpAudioStream * self);
 
 G_END_DECLS
 

modules/module-pipewire/multiport-link-algorithm.c

@@ -0,0 +1,112 @@
+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author George Kiagiadakis <george.kiagiadakis@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <wp/wp.h>
+#include <pipewire/pipewire.h>
+#include <spa/debug/types.h>
+#include <spa/param/audio/type-info.h>
+
+#include "multiport-link-algorithm.h"
+
+gboolean
+multiport_link_create (
+    GVariant * src_data,
+    GVariant * sink_data,
+    CreateLinkCb create_link_cb,
+    gpointer user_data,
+    GError ** error)
+{
+  g_autoptr (GPtrArray) in_ports = NULL;
+  GVariantIter *iter;
+  GVariant *child;
+  guint32 out_node_id, in_node_id;
+  guint32 out_port_id, in_port_id;
+  guint32 out_channel, in_channel;
+  guint8 direction;
+  guint i;
+
+  /* tuple format:
+      uint32 node_id;
+      uint32 port_id;
+      uint32 channel;  // enum spa_audio_channel
+      uint8 direction; // enum spa_direction
+   */
+  if (!g_variant_is_of_type (src_data, G_VARIANT_TYPE("a(uuuy)")))
+    goto invalid_argument;
+  if (!g_variant_is_of_type (sink_data, G_VARIANT_TYPE("a(uuuy)")))
+    goto invalid_argument;
+
+  /* transfer the in ports to an array so that we can
+     delete them when they are linked */
+  in_ports = g_ptr_array_new_full (g_variant_n_children (sink_data),
+      (GDestroyNotify) g_variant_unref);
+
+  g_variant_get (sink_data, "a(uuuy)", &iter);
+  while ((child = g_variant_iter_next_value (iter))) {
+    g_variant_get (child, "(uuuy)", NULL, NULL, NULL, &direction);
+    /* remove non-input direction ports right away */
+    if (direction == PW_DIRECTION_INPUT)
+      g_ptr_array_add (in_ports, child);
+    else
+      g_variant_unref (child);
+  }
+  g_variant_iter_free (iter);
+
+  /* now loop over the out ports and figure out where they should be linked */
+  g_variant_get (src_data, "a(uuuy)", &iter);
+  while (g_variant_iter_loop (iter, "(uuuy)", &out_node_id, &out_port_id,
+              &out_channel, &direction))
+  {
+    /* skip non-output ports right away */
+    if (direction != PW_DIRECTION_OUTPUT)
+      continue;
+
+    for (i = 0; i < in_ports->len; i++) {
+      child = g_ptr_array_index (in_ports, i);
+      g_variant_get (child, "(uuuy)", &in_node_id, &in_port_id, &in_channel, NULL);
+
+      /* the channel has to match, unless we don't have any information
+         on channel ordering on either side */
+      if (out_channel == in_channel ||
+          out_channel == SPA_AUDIO_CHANNEL_UNKNOWN ||
+          in_channel == SPA_AUDIO_CHANNEL_UNKNOWN)
+      {
+        g_autoptr (WpProperties) props = NULL;
+
+        /* Create the properties */
+        props = wp_properties_new_empty ();
+        wp_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%u", out_node_id);
+        wp_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%u", out_port_id);
+        wp_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%u", in_node_id);
+        wp_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%u", in_port_id);
+
+        g_debug ("Create pw link: %u:%u (%s) -> %u:%u (%s)",
+            out_node_id, out_port_id,
+            spa_debug_type_find_name (spa_type_audio_channel, out_channel),
+            in_node_id, in_port_id,
+            spa_debug_type_find_name (spa_type_audio_channel, in_channel));
+
+        /* and the link */
+        create_link_cb (props, user_data);
+
+        /* and remove the linked input port from the array */
+        g_ptr_array_remove_index (in_ports, i);
+
+        /* break out of the for loop; go for the next out port */
+        break;
+      }
+    }
+  }
+
+  return TRUE;
+
+invalid_argument:
+  g_set_error (error, WP_DOMAIN_LIBRARY, WP_LIBRARY_ERROR_INVALID_ARGUMENT,
+      "Endpoint node/port descriptions don't have the required fields");
+  return FALSE;
+}

modules/module-pipewire/multiport-link-algorithm.h

@@ -0,0 +1,11 @@
+/* WirePlumber
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *    @author George Kiagiadakis <george.kiagiadakis@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+typedef void (*CreateLinkCb) (WpProperties *, gpointer);
+gboolean multiport_link_create (GVariant * src_data, GVariant * sink_data,
+    CreateLinkCb create_link_cb, gpointer user_data, GError ** error);

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

@@ -23,6 +23,8 @@
 #include <wp/wp.h>
 #include <pipewire/pipewire.h>
 
+#include "multiport-link-algorithm.h"
+
 struct _WpPipewireSimpleEndpointLink
 {
   WpEndpointLink parent;
@@ -161,86 +163,36 @@ on_proxy_link_augmented (WpProxy *proxy, GAsyncResult *res, gpointer data)
   }
 }
 
+static void
+create_link_cb (WpProperties *props, gpointer user_data)
+{
+  WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(user_data);
+  g_autoptr (WpCore) core = NULL;
+  WpProxy *proxy;
+
+  core = g_weak_ref_get (&self->core);
+  g_return_if_fail (core);
+
+  /* Create the link */
+  proxy = wp_core_create_remote_object(core, "link-factory",
+      PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
+  g_return_if_fail (proxy);
+  g_ptr_array_add(self->link_proxies, proxy);
+
+  /* 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);
+}
+
 static gboolean
 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 = NULL;
-  guint32 output_node_id, input_node_id;
-  GVariant *src_ports, *sink_ports;
-  GVariantIter *out_iter, *in_iter;
-  guint64 out_ptr, in_ptr;
-  g_autoptr (GHashTable) linked_ports = NULL;
-  g_autoptr (WpProperties) props = NULL;
-  WpProxy *proxy = NULL;
 
-  core = g_weak_ref_get (&self->core);
-  g_return_val_if_fail (core, FALSE);
-
-  /* Get the node ids and port ids */
-  if (!g_variant_lookup (src_data, "node-id", "u", &output_node_id))
-      return FALSE;
-  src_ports = g_variant_lookup_value (src_data, "ports", G_VARIANT_TYPE_ARRAY);
-  if (!src_ports)
-      return FALSE;
-  if (!g_variant_lookup (sink_data, "node-id", "u", &input_node_id))
-      return FALSE;
-  sink_ports = g_variant_lookup_value (sink_data, "ports", G_VARIANT_TYPE_ARRAY);
-  if (!sink_ports)
-      return FALSE;
-
-  /* Link all the output ports with the input ports */
-  linked_ports = g_hash_table_new (g_direct_hash, g_direct_equal);
-  g_variant_get (src_ports, "at", &out_iter);
-  while (g_variant_iter_loop (out_iter, "t", &out_ptr)) {
-    WpProxyPort *out_p = (gpointer)out_ptr;
-    enum pw_direction out_direction = wp_proxy_port_get_info(out_p)->direction;
-    guint out_id = wp_proxy_get_global_id(WP_PROXY(out_p));
-    if (out_direction == PW_DIRECTION_INPUT)
-      continue;
-
-    g_variant_get (sink_ports, "at", &in_iter);
-    while (g_variant_iter_loop (in_iter, "t", &in_ptr)) {
-      WpProxyPort *in_p = (gpointer)in_ptr;
-      enum pw_direction in_direction = wp_proxy_port_get_info(in_p)->direction;
-      guint in_id = wp_proxy_get_global_id(WP_PROXY(in_p));
-      if (in_direction == PW_DIRECTION_OUTPUT)
-        continue;
-
-      /* Skip the ports if they are already linked */
-      if (g_hash_table_contains (linked_ports, GUINT_TO_POINTER(in_id)) ||
-          g_hash_table_contains (linked_ports, GUINT_TO_POINTER(out_id)))
-        continue;
-
-      /* Create the properties */
-      props = wp_properties_new_empty ();
-      wp_properties_setf(props, PW_KEY_LINK_OUTPUT_NODE, "%d", output_node_id);
-      wp_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", out_id);
-      wp_properties_setf(props, PW_KEY_LINK_INPUT_NODE, "%d", input_node_id);
-      wp_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", in_id);
-
-      /* Create the link */
-      proxy = wp_core_create_remote_object(core, "link-factory",
-          PW_TYPE_INTERFACE_Link, PW_VERSION_LINK_PROXY, props);
-      g_return_val_if_fail (proxy, FALSE);
-      g_ptr_array_add(self->link_proxies, proxy);
-
-      /* 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_add (linked_ports, GUINT_TO_POINTER(in_id));
-      g_hash_table_add (linked_ports, GUINT_TO_POINTER(out_id));
-    }
-    g_variant_iter_free (in_iter);
-  }
-  g_variant_iter_free (out_iter);
-
-  return TRUE;
+  return multiport_link_create (src_data, sink_data, create_link_cb, self, error);
 }
 
 static void