modules: implement si-standard-link, the default WpSiLink implementation

lib/wp/si-interfaces.c

@@ -100,6 +100,24 @@ wp_si_endpoint_get_stream (WpSiEndpoint * self, guint index)
   return WP_SI_ENDPOINT_GET_IFACE (self)->get_stream (self, index);
 }
 
+/**
+ * wp_si_endpoint_get_stream_acquisition: (virtual get_stream_acquisition)
+ * @self: the session item
+ *
+ * Returns: (transfer none): the stream acquisition interface associated with
+ *   this endpoint, or %NULL if this endpoint does not require acquiring
+ *   streams before linking them
+ */
+WpSiStreamAcquisition *
+wp_si_endpoint_get_stream_acquisition (WpSiEndpoint * self)
+{
+  g_return_val_if_fail (WP_IS_SI_ENDPOINT (self), NULL);
+  g_return_val_if_fail (WP_SI_ENDPOINT_GET_IFACE (self)->get_stream_acquisition,
+      NULL);
+
+  return WP_SI_ENDPOINT_GET_IFACE (self)->get_stream_acquisition (self);
+}
+
 /**
  * WpSiMultiEndpoint:
  *
@@ -296,3 +314,161 @@ wp_si_link_get_in_stream (WpSiLink * self)
 
   return WP_SI_LINK_GET_IFACE (self)->get_in_stream (self);
 }
+
+/**
+ * WpSiPortInfo:
+ *
+ * An interface for retrieving PipeWire port information from a session item.
+ * This information is used to create links in the nodes graph.
+ *
+ * This is normally implemented by the same session items that implement
+ * #WpSiStream. The standard link implementation expects to be able to cast
+ * a #WpSiStream into a #WpSiPortInfo.
+ */
+G_DEFINE_INTERFACE (WpSiPortInfo, wp_si_port_info, WP_TYPE_SESSION_ITEM)
+
+static void
+wp_si_port_info_default_init (WpSiPortInfoInterface * iface)
+{
+}
+
+/**
+ * wp_si_port_info_get_ports: (virtual get_ports)
+ * @self: the session item
+ * @context: (nullable): an optional context for the ports
+ *
+ * This method returns a variant of type "a(uuu)", where each tuple in the
+ * array contains the following information:
+ *   - u: (guint32) node id
+ *   - u: (guint32) port id (the port must belong on the node specified above)
+ *   - u: (guint32) the audio channel (enum spa_audio_channel) that this port
+ *        makes available, or 0 for non-audio content
+ *
+ * The order in which ports appear in this array is important when no channel
+ * information is available. The link implementation should link the ports
+ * in the order they appear. This is normally a good enough substitute for
+ * channel matching.
+ *
+ * The @context argument can be used to get different sets of ports from
+ * the item. The following well-known contexts are defined:
+ *   - %NULL: get the standard ports to be linked
+ *   - "monitor": get the monitor ports
+ *   - "control": get the control port
+ *   - "reverse": get the reverse direction ports, if this item controls a
+ *                filter node, which would have ports on both directions
+ *
+ * Contexts other than %NULL may only be used internally to ease the
+ * implementation of more complex endpoint relationships. For example, a
+ * #WpSessionItem that is in control of an input (sink) adapter node may
+ * implement #WpSiStream and #WpSiPortInfo where the %NULL context will return
+ * the standard input ports and the "monitor" context will return the adapter's
+ * monitor ports. When linking this stream to another stream, the %NULL context
+ * will always be used, but the item may internally spawn a secondary
+ * #WpSessionItem that implements the "monitor" endpoint & stream. That
+ * secondary stream may implement #WpSiPortInfo, chaining calls to the
+ * #WpSiPortInfo of the original item using the "monitor" context. This way,
+ * the monitor #WpSessionItem does not need to share control of the underlying
+ * node; it only proxies calls to satisfy the API.
+ *
+ * Returns: (transfer full): a #GVariant containing information about the
+ *   ports of this item
+ */
+GVariant *
+wp_si_port_info_get_ports (WpSiPortInfo * self, const gchar * context)
+{
+  g_return_val_if_fail (WP_IS_SI_PORT_INFO (self), NULL);
+  g_return_val_if_fail (WP_SI_PORT_INFO_GET_IFACE (self)->get_ports, NULL);
+
+  return WP_SI_PORT_INFO_GET_IFACE (self)->get_ports (self, context);
+}
+
+/**
+ * WpSiStreamAcquisition:
+ *
+ * This interface provides a way to request a stream for linking before doing
+ * so. This allows endpoint implementations to apply internal policy rules
+ * (such as, streams that can only be linked once or mutually exclusive streams).
+ *
+ * A #WpSiStreamAcquisition is associated directly with a #WpSiEndpoint via
+ * wp_si_endpoint_get_stream_acquisition(). In order to allow switching policies,
+ * it is recommended that endpoint implementations use a separate session item
+ * to implement this interface and allow replacing it.
+ */
+G_DEFINE_INTERFACE (WpSiStreamAcquisition, wp_si_stream_acquisition,
+                    WP_TYPE_SESSION_ITEM)
+
+static void
+wp_si_stream_acquisition_default_init (WpSiStreamAcquisitionInterface * iface)
+{
+}
+
+/**
+ * wp_si_stream_acquisition_acquire: (virtual acquire)
+ * @self: the session item
+ * @acquisitor: the link that is trying to acquire a stream
+ * @stream: the stream that is being acquired
+ * @callback: (scope async): the callback to call when the operation is done
+ * @data: (closure): user data for @callback
+ *
+ * Acquires the @stream for linking by @acquisitor.
+ *
+ * When a link is not allowed by policy, this operation should return
+ * an error.
+ *
+ * When a link needs to be delayed for a short amount of time (ex. to apply
+ * a fade out effect on another stream), this operation should finish with a
+ * delay. It is safe to assume that after this operation completes,
+ * the stream will be linked immediately.
+ */
+void
+wp_si_stream_acquisition_acquire (WpSiStreamAcquisition * self,
+    WpSiLink * acquisitor, WpSiStream * stream,
+    GAsyncReadyCallback callback, gpointer data)
+{
+  g_return_if_fail (WP_IS_SI_STREAM_ACQUISITION (self));
+  g_return_if_fail (WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->acquire);
+
+  WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->acquire (self, acquisitor, stream,
+      callback, data);
+}
+
+/**
+ * wp_si_stream_acquisition_acquire_finish: (virtual acquire_finish)
+ * @self: the session item
+ * @res: the async result
+ * @error: (out) (optional): the operation's error, if it occurred
+ *
+ * Finishes the operation started by wp_si_stream_acquisition_acquire().
+ * This is meant to be called in the callback that was passed to that method.
+ *
+ * Returns: %TRUE on success, %FALSE if there was an error
+ */
+gboolean
+wp_si_stream_acquisition_acquire_finish (WpSiStreamAcquisition * self,
+    GAsyncResult * res, GError ** error)
+{
+  g_return_val_if_fail (WP_IS_SI_STREAM_ACQUISITION (self), FALSE);
+  g_return_val_if_fail (
+      WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->acquire_finish, FALSE);
+
+  return WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->acquire_finish (self, res,
+      error);
+}
+
+/**
+ * wp_si_stream_acquisition_release: (virtual release)
+ * @self: the session item
+ * @acquisitor: the link that had previously acquired the stream
+ * @stream: the stream that is being released
+ *
+ * Releases the @stream, which means that it is being unlinked.
+ */
+void
+wp_si_stream_acquisition_release (WpSiStreamAcquisition * self,
+    WpSiLink * acquisitor, WpSiStream * stream)
+{
+  g_return_if_fail (WP_IS_SI_STREAM_ACQUISITION (self));
+  g_return_if_fail (WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->release);
+
+  WP_SI_STREAM_ACQUISITION_GET_IFACE (self)->release (self, acquisitor, stream);
+}

lib/wp/si-interfaces.h

@@ -17,6 +17,7 @@
 G_BEGIN_DECLS
 
 typedef struct _WpSiStream WpSiStream;
+typedef struct _WpSiStreamAcquisition WpSiStreamAcquisition;
 
 /**
  * WP_TYPE_SI_ENDPOINT:
@@ -37,6 +38,8 @@ struct _WpSiEndpointInterface
 
   guint (*get_n_streams) (WpSiEndpoint * self);
   WpSiStream * (*get_stream) (WpSiEndpoint * self, guint index);
+
+  WpSiStreamAcquisition * (*get_stream_acquisition) (WpSiEndpoint * self);
 };
 
 WP_API
@@ -51,6 +54,9 @@ guint wp_si_endpoint_get_n_streams (WpSiEndpoint * self);
 WP_API
 WpSiStream * wp_si_endpoint_get_stream (WpSiEndpoint * self, guint index);
 
+WP_API
+WpSiStreamAcquisition * wp_si_endpoint_get_stream_acquisition (WpSiEndpoint * self);
+
 /**
  * WP_TYPE_SI_MULTI_ENDPOINT:
  *
@@ -138,6 +144,62 @@ WpSiStream * wp_si_link_get_out_stream (WpSiLink * self);
 WP_API
 WpSiStream * wp_si_link_get_in_stream (WpSiLink * self);
 
+/**
+ * WP_TYPE_SI_PORT_INFO:
+ *
+ * The #WpSiPortInfo #GType
+ */
+#define WP_TYPE_SI_PORT_INFO (wp_si_port_info_get_type ())
+WP_API
+G_DECLARE_INTERFACE (WpSiPortInfo, wp_si_port_info,
+                     WP, SI_PORT_INFO, WpSessionItem)
+
+struct _WpSiPortInfoInterface
+{
+  GTypeInterface interface;
+
+  GVariant * (*get_ports) (WpSiPortInfo * self, const gchar * context);
+};
+
+WP_API
+GVariant * wp_si_port_info_get_ports (WpSiPortInfo * self,
+    const gchar * context);
+
+/**
+ * WP_TYPE_SI_STREAM_ACQUISITION:
+ *
+ * The #WpSiStreamAcquisition #GType
+ */
+#define WP_TYPE_SI_STREAM_ACQUISITION (wp_si_stream_acquisition_get_type ())
+WP_API
+G_DECLARE_INTERFACE (WpSiStreamAcquisition, wp_si_stream_acquisition,
+                     WP, SI_STREAM_ACQUISITION, WpSessionItem)
+
+struct _WpSiStreamAcquisitionInterface
+{
+  GTypeInterface interface;
+
+  void (*acquire) (WpSiStreamAcquisition * self, WpSiLink * acquisitor,
+      WpSiStream * stream, GAsyncReadyCallback callback, gpointer data);
+  gboolean (*acquire_finish) (WpSiStreamAcquisition * self,
+      GAsyncResult * res, GError ** error);
+
+  void (*release) (WpSiStreamAcquisition * self, WpSiLink * acquisitor,
+      WpSiStream * stream);
+};
+
+WP_API
+void wp_si_stream_acquisition_acquire (WpSiStreamAcquisition * self,
+    WpSiLink * acquisitor, WpSiStream * stream,
+    GAsyncReadyCallback callback, gpointer data);
+
+WP_API
+gboolean wp_si_stream_acquisition_acquire_finish (WpSiStreamAcquisition * self,
+    GAsyncResult * res, GError ** error);
+
+WP_API
+void wp_si_stream_acquisition_release (WpSiStreamAcquisition * self,
+    WpSiLink * acquisitor, WpSiStream * stream);
 
 G_END_DECLS
 

modules/meson.build

@@ -114,3 +114,14 @@ shared_library(
   install_dir : wireplumber_module_dir,
   dependencies : [wp_dep, pipewire_dep],
 )
+
+shared_library(
+  'wireplumber-module-si-standard-link',
+  [
+    'module-si-standard-link.c',
+  ],
+  c_args : [common_c_args, '-DG_LOG_DOMAIN="m-si-standard-link"'],
+  install : true,
+  install_dir : wireplumber_module_dir,
+  dependencies : [wp_dep, pipewire_dep],
+)

modules/module-si-standard-link.c

+/* WirePlumber
+ *
+ * Copyright © 2020 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>
+
+enum {
+  STEP_ACQUIRE = WP_TRANSITION_STEP_CUSTOM_START,
+  STEP_LINK,
+};
+
+struct _WpSiStandardLink
+{
+  WpSessionItem parent;
+
+  WpSiStream *out_stream;
+  WpSiStream *in_stream;
+
+  GPtrArray *node_links;
+  guint n_async_ops_wait;
+};
+
+static void si_standard_link_link_init (WpSiLinkInterface * iface);
+
+G_DECLARE_FINAL_TYPE (WpSiStandardLink, si_standard_link, WP, SI_STANDARD_LINK, WpSessionItem)
+G_DEFINE_TYPE_WITH_CODE (WpSiStandardLink, si_standard_link, WP_TYPE_SESSION_ITEM,
+    G_IMPLEMENT_INTERFACE (WP_TYPE_SI_LINK, si_standard_link_link_init))
+
+static void
+on_stream_destroyed (gpointer data, GObject * stream)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (data);
+
+  if ((gpointer) self->out_stream == (gpointer) stream)
+    self->out_stream = NULL;
+  else if ((gpointer) self->in_stream == (gpointer) stream)
+    self->in_stream = NULL;
+
+  wp_session_item_reset (WP_SESSION_ITEM (self));
+}
+
+static void
+on_stream_flags_changed (WpSessionItem * stream, WpSiFlags flags,
+    WpSiStandardLink *self)
+{
+  /* stream was deactivated; treat it as destroyed and reset */
+  if (!(flags & WP_SI_FLAG_ACTIVE))
+    wp_session_item_reset (WP_SESSION_ITEM (self));
+}
+
+static inline void
+disconnect_stream (WpSiStandardLink *self, WpSiStream * stream)
+{
+  if (stream) {
+    g_signal_handlers_disconnect_by_data (stream, self);
+    g_object_weak_unref (G_OBJECT (stream), on_stream_destroyed, self);
+  }
+}
+
+static void
+si_standard_link_init (WpSiStandardLink * self)
+{
+}
+
+static void
+si_standard_link_reset (WpSessionItem * item)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+
+  WP_SESSION_ITEM_CLASS (si_standard_link_parent_class)->reset (item);
+
+  disconnect_stream (self, self->out_stream);
+  disconnect_stream (self, self->in_stream);
+  self->out_stream = NULL;
+  self->in_stream = NULL;
+
+  wp_session_item_clear_flag (item, WP_SI_FLAG_CONFIGURED);
+}
+
+static GVariant *
+si_standard_link_get_configuration (WpSessionItem * item)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+  GVariantBuilder b;
+
+  /* Set the properties */
+  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+  g_variant_builder_add (&b, "{sv}",
+      "out-stream", g_variant_new_uint64 ((guint64) self->out_stream));
+  g_variant_builder_add (&b, "{sv}",
+      "in-stream", g_variant_new_uint64 ((guint64) self->in_stream));
+  return g_variant_builder_end (&b);
+}
+
+static gboolean
+si_standard_link_configure (WpSessionItem * item, GVariant * args)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+  guint64 out_stream_i, in_stream_i;
+  WpSessionItem *out_stream, *in_stream;
+
+  if (wp_session_item_get_flags (item) &
+          (WP_SI_FLAG_ACTIVATING | WP_SI_FLAG_ACTIVE |
+           WP_SI_FLAG_EXPORTING | WP_SI_FLAG_EXPORTED))
+    return FALSE;
+
+  if (!g_variant_lookup (args, "out-stream", "t", &out_stream_i) ||
+      !g_variant_lookup (args, "in-stream", "t", &in_stream_i))
+    return FALSE;
+
+  out_stream = GUINT_TO_POINTER (out_stream_i);
+  in_stream = GUINT_TO_POINTER (in_stream_i);
+
+  if (!WP_IS_SI_STREAM (out_stream) || !WP_IS_SI_STREAM (in_stream) ||
+      !WP_IS_SI_PORT_INFO (out_stream) || !WP_IS_SI_PORT_INFO (in_stream) ||
+      !(wp_session_item_get_flags (out_stream) & WP_SI_FLAG_ACTIVE) ||
+      !(wp_session_item_get_flags (in_stream) & WP_SI_FLAG_ACTIVE))
+    return FALSE;
+
+  disconnect_stream (self, self->out_stream);
+  disconnect_stream (self, self->in_stream);
+
+  self->out_stream = WP_SI_STREAM (out_stream);
+  self->in_stream = WP_SI_STREAM (in_stream);
+
+  g_signal_connect_object (self->out_stream, "flags-changed",
+      G_CALLBACK (on_stream_flags_changed), self, 0);
+  g_signal_connect_object (self->in_stream, "flags-changed",
+      G_CALLBACK (on_stream_flags_changed), self, 0);
+  g_object_weak_ref (G_OBJECT (self->out_stream), on_stream_destroyed, self);
+  g_object_weak_ref (G_OBJECT (self->in_stream), on_stream_destroyed, self);
+
+  wp_session_item_set_flag (item, WP_SI_FLAG_CONFIGURED);
+
+  return TRUE;
+}
+
+static guint
+si_standard_link_get_next_step (WpSessionItem * item,
+     WpTransition * transition, guint step)
+{
+  WpSiStandardLink *self = wp_transition_get_source_object (transition);
+
+  switch (step) {
+    case WP_TRANSITION_STEP_NONE:
+      return STEP_ACQUIRE;
+
+    case STEP_ACQUIRE:
+      if (self->n_async_ops_wait == 0)
+        return STEP_LINK;
+      else
+        return step;
+
+    case STEP_LINK:
+      if (self->n_async_ops_wait == 0)
+        return WP_TRANSITION_STEP_NONE;
+      else
+        return step;
+
+    default:
+      return WP_TRANSITION_STEP_ERROR;
+  }
+}
+
+static void
+on_stream_acquired (WpSiStreamAcquisition * acq, GAsyncResult * res,
+    WpTransition * transition)
+{
+  WpSiStandardLink *self = wp_transition_get_source_object (transition);
+  g_autoptr (GError) error = NULL;
+
+  if (!wp_si_stream_acquisition_acquire_finish (acq, res, &error)) {
+    wp_transition_return_error (transition, g_steal_pointer (&error));
+    return;
+  }
+
+  self->n_async_ops_wait--;
+  wp_transition_advance (transition);
+}
+
+static void
+on_link_augmented (WpProxy * proxy, GAsyncResult * res,
+    WpTransition * transition)
+{
+  WpSiStandardLink *self = wp_transition_get_source_object (transition);
+  g_autoptr (GError) error = NULL;
+
+  if (!wp_proxy_augment_finish (proxy, res, &error)) {
+    wp_transition_return_error (transition, g_steal_pointer (&error));
+    return;
+  }
+
+  self->n_async_ops_wait--;
+  wp_transition_advance (transition);
+}
+
+static WpCore *
+find_core (WpSiStandardLink * self)
+{
+  /* session items are not associated with a core, but surely when linking
+    we should be able to find a WpImplEndpointLink associated, or at the very
+    least a WpEndpoint associated with one of the streams... */
+  g_autoptr (WpProxy) proxy = wp_session_item_get_associated_proxy (
+      WP_SESSION_ITEM (self), WP_TYPE_ENDPOINT_LINK);
+  if (!proxy) {
+      proxy = wp_session_item_get_associated_proxy (
+          WP_SESSION_ITEM (self->out_stream), WP_TYPE_ENDPOINT);
+  }
+  return proxy ? wp_proxy_get_core (proxy) : NULL;
+}
+
+static gboolean
+create_links (WpSiStandardLink * self, GVariant * out_ports, GVariant * in_ports)
+{
+  g_autoptr (GPtrArray) in_ports_arr = NULL;
+  g_autoptr (WpCore) core = NULL;
+  WpLink *link;
+  GVariantIter *iter;
+  GVariant *child;
+  guint32 out_node_id, in_node_id;
+  guint32 out_port_id, in_port_id;
+  guint32 out_channel, in_channel;
+  gboolean link_all = FALSE;
+  guint i;
+
+  /* tuple format:
+      uint32 node_id;
+      uint32 port_id;
+      uint32 channel;  // enum spa_audio_channel
+   */
+  if (!g_variant_is_of_type (out_ports, G_VARIANT_TYPE("a(uuu)")))
+    return FALSE;
+  if (!g_variant_is_of_type (in_ports, G_VARIANT_TYPE("a(uuu)")))
+    return FALSE;
+
+  core = find_core (self);
+  g_return_val_if_fail (core, FALSE);
+
+  self->n_async_ops_wait = 0;
+  self->node_links = g_ptr_array_new_with_free_func (g_object_unref);
+
+  /* transfer the in ports to an array so that we can
+     delete them when they are linked */
+  i = g_variant_n_children (in_ports);
+  in_ports_arr = g_ptr_array_new_full (i, (GDestroyNotify) g_variant_unref);
+  g_ptr_array_set_size (in_ports_arr, i);
+
+  g_variant_get (in_ports, "a(uuu)", &iter);
+  while ((child = g_variant_iter_next_value (iter)))
+    g_ptr_array_insert (in_ports_arr, --i, child);
+  g_variant_iter_free (iter);
+
+  /* now loop over the out ports and figure out where they should be linked */
+  g_variant_get (out_ports, "a(uuu)", &iter);
+
+  /* special case for mono inputs: link to all outputs,
+     since we don't support proper channel mapping yet */
+  if (g_variant_iter_n_children (iter) == 1)
+    link_all = TRUE;
+
+  while (g_variant_iter_loop (iter, "(uuu)", &out_node_id, &out_port_id,
+              &out_channel))
+  {
+    for (i = in_ports_arr->len; i > 0; i--) {
+      child = g_ptr_array_index (in_ports_arr, i - 1);
+      g_variant_get (child, "(uuu)", &in_node_id, &in_port_id, &in_channel);
+
+      /* the channel has to match, unless we don't have any information
+         on channel ordering on either side */
+      if (link_all ||
+          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));
+
+        /* create the link */
+        link = wp_link_new_from_factory (core, "link-factory",
+            g_steal_pointer (&props));
+        g_ptr_array_add (self->node_links, link);
+
+        /* augment to ensure it is created without errors */
+        self->n_async_ops_wait++;
+        wp_proxy_augment (WP_PROXY (link), WP_PROXY_FEATURES_STANDARD, NULL,
+            (GAsyncReadyCallback) on_link_augmented, self);
+
+        /* continue to link all input ports, if requested */
+        if (link_all)
+          continue;
+
+        /* remove the linked input port from the array */
+        g_ptr_array_remove_index (in_ports_arr, i - 1);
+
+        /* break out of the for loop; go for the next out port */
+        break;
+      }
+    }
+  }
+  return TRUE;
+}
+
+static void
+si_standard_link_execute_step (WpSessionItem * item, WpTransition * transition,
+    guint step)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+
+  switch (step) {
+  case STEP_ACQUIRE: {
+    WpSiEndpoint *out_endpoint, *in_endpoint;
+    WpSiStreamAcquisition *out_acquisition, *in_acquisition;
+
+    out_endpoint = wp_si_stream_get_parent_endpoint (self->out_stream);
+    in_endpoint = wp_si_stream_get_parent_endpoint (self->in_stream);
+    out_acquisition = wp_si_endpoint_get_stream_acquisition (out_endpoint);
+    in_acquisition = wp_si_endpoint_get_stream_acquisition (in_endpoint);
+
+    if (out_acquisition && in_acquisition)
+      self->n_async_ops_wait = 2;
+    else if (out_acquisition || in_acquisition)
+      self->n_async_ops_wait = 1;
+    else {
+      self->n_async_ops_wait = 0;
+      wp_transition_advance (transition);
+      return;
+    }
+
+    if (out_acquisition) {
+      wp_si_stream_acquisition_acquire (out_acquisition, WP_SI_LINK (self),
+          self->out_stream, (GAsyncReadyCallback) on_stream_acquired,
+          transition);
+    }
+    if (in_acquisition) {
+      wp_si_stream_acquisition_acquire (in_acquisition, WP_SI_LINK (self),
+          self->in_stream, (GAsyncReadyCallback) on_stream_acquired,
+          transition);
+    }
+    break;
+  }
+  case STEP_LINK: {
+    g_autoptr (GVariant) out_ports = NULL;
+    g_autoptr (GVariant) in_ports = NULL;
+
+    out_ports = wp_si_port_info_get_ports (WP_SI_PORT_INFO (self->out_stream),
+        NULL);
+    in_ports = wp_si_port_info_get_ports (WP_SI_PORT_INFO (self->in_stream),
+        NULL);
+
+    if (!create_links (self, out_ports, in_ports)) {
+      wp_transition_return_error (transition, g_error_new (WP_DOMAIN_LIBRARY,
+              WP_LIBRARY_ERROR_INVARIANT,
+              "Bad port info returned from one of the streams"));
+    }
+    break;
+  }
+  default:
+    WP_SESSION_ITEM_GET_CLASS (si_standard_link_parent_class)->execute_step (
+          item, transition, step);
+      break;
+  }
+}
+
+static void
+si_standard_link_deactivate (WpSessionItem * item)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+  WpSiEndpoint *out_endpoint, *in_endpoint;
+  WpSiStreamAcquisition *out_acquisition, *in_acquisition;
+
+  out_endpoint = wp_si_stream_get_parent_endpoint (self->out_stream);
+  in_endpoint = wp_si_stream_get_parent_endpoint (self->in_stream);
+  out_acquisition = wp_si_endpoint_get_stream_acquisition (out_endpoint);
+  in_acquisition = wp_si_endpoint_get_stream_acquisition (in_endpoint);
+
+  if (out_acquisition) {
+    wp_si_stream_acquisition_release (out_acquisition, WP_SI_LINK (self),
+        self->out_stream);
+  }
+  if (in_acquisition) {
+    wp_si_stream_acquisition_release (in_acquisition, WP_SI_LINK (self),
+        self->in_stream);
+  }
+
+  g_clear_pointer (&self->node_links, g_ptr_array_unref);
+
+  WP_SESSION_ITEM_CLASS (si_standard_link_parent_class)->deactivate (item);
+}
+
+static void
+si_standard_link_class_init (WpSiStandardLinkClass * klass)
+{
+  WpSessionItemClass *si_class = (WpSessionItemClass *) klass;
+
+  si_class->reset = si_standard_link_reset;
+  si_class->configure = si_standard_link_configure;
+  si_class->get_configuration = si_standard_link_get_configuration;
+  si_class->get_next_step = si_standard_link_get_next_step;
+  si_class->execute_step = si_standard_link_execute_step;
+  si_class->deactivate = si_standard_link_deactivate;
+}
+
+static GVariant *
+si_standard_link_get_registration_info (WpSiLink * item)
+{
+  GVariantBuilder b;
+  g_variant_builder_init (&b, G_VARIANT_TYPE ("a{ss}"));
+  return g_variant_builder_end (&b);
+}
+
+static WpProperties *
+si_standard_link_get_properties (WpSiLink * item)
+{
+  return wp_properties_new_empty ();
+}
+
+static WpSiStream *
+si_standard_link_get_out_stream (WpSiLink * item)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+  return self->out_stream;
+}
+
+static WpSiStream *
+si_standard_link_get_in_stream (WpSiLink * item)
+{
+  WpSiStandardLink *self = WP_SI_STANDARD_LINK (item);
+  return self->in_stream;
+}
+
+static void
+si_standard_link_link_init (WpSiLinkInterface * iface)
+{
+  iface->get_registration_info = si_standard_link_get_registration_info;
+  iface->get_properties = si_standard_link_get_properties;
+  iface->get_out_stream = si_standard_link_get_out_stream;
+  iface->get_in_stream = si_standard_link_get_in_stream;
+}
+
+WP_PLUGIN_EXPORT void
+wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
+{
+  GVariantBuilder b;
+
+  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+  g_variant_builder_add (&b, "(ssymv)", "out-stream", "t",
+      WP_SI_CONFIG_OPTION_WRITEABLE | WP_SI_CONFIG_OPTION_REQUIRED, NULL);
+  g_variant_builder_add (&b, "(ssymv)", "in-stream", "t",
+      WP_SI_CONFIG_OPTION_WRITEABLE | WP_SI_CONFIG_OPTION_REQUIRED, NULL);
+
+  wp_si_factory_register (core, wp_si_factory_new_simple (
+          "si-standard-link",
+          si_standard_link_get_type (),
+          g_variant_builder_end (&b)));
+}