Merge branch 'async-endpoint-links' into 'master'

Async endpoint links See merge request gkiagia/wireplumber!15
2019-07-01 12:54:29 +00:00
parent 4ed5e30d9d ab55116653
commit ccc88b39c3
10 changed files with 404 additions and 204 deletions
--- a/lib/wp/endpoint.c
+++ b/lib/wp/endpoint.c
@@ -737,39 +737,199 @@ wp_endpoint_get_links (WpEndpoint * self)
 typedef struct _WpEndpointLinkPrivate WpEndpointLinkPrivate;
 struct _WpEndpointLinkPrivate
 {
-  WpEndpoint *src;
+  /* The task to signal the endpoint link is initialized */
+  GTask *init_task;
+
+  GWeakRef src;
  guint32 src_stream;
-  WpEndpoint *sink;
+  GWeakRef sink;
  guint32 sink_stream;
 };

-G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (WpEndpointLink, wp_endpoint_link, G_TYPE_OBJECT)
+enum {
+  LINKPROP_0,
+  LINKPROP_SRC,
+  LINKPROP_SRC_STREAM,
+  LINKPROP_SINK,
+  LINKPROP_SINK_STREAM,
+};
+
+static void wp_endpoint_link_async_initable_init (gpointer iface,
+    gpointer iface_data);
+
+G_DEFINE_ABSTRACT_TYPE_WITH_CODE (WpEndpointLink, wp_endpoint_link, G_TYPE_OBJECT,
+    G_ADD_PRIVATE (WpEndpointLink)
+    G_IMPLEMENT_INTERFACE (G_TYPE_ASYNC_INITABLE,
+                           wp_endpoint_link_async_initable_init))
+
+static void
+endpoint_link_finalize (GObject * object)
+{
+  WpEndpointLinkPrivate *priv =
+      wp_endpoint_link_get_instance_private (WP_ENDPOINT_LINK (object));
+
+  /* Destroy the init task */
+  g_clear_object(&priv->init_task);
+
+  /* Clear the endpoint weak reaferences */
+  g_weak_ref_clear(&priv->src);
+  g_weak_ref_clear(&priv->sink);
+}
+
+static void
+endpoint_link_set_property (GObject * object, guint property_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  WpEndpointLinkPrivate *priv =
+      wp_endpoint_link_get_instance_private (WP_ENDPOINT_LINK (object));
+
+  switch (property_id) {
+  case LINKPROP_SRC:
+    g_weak_ref_set (&priv->src, g_value_get_object (value));
+    break;
+  case LINKPROP_SRC_STREAM:
+    priv->src_stream = g_value_get_uint(value);
+    break;
+  case LINKPROP_SINK:
+    g_weak_ref_set (&priv->sink, g_value_get_object (value));
+    break;
+  case LINKPROP_SINK_STREAM:
+    priv->sink_stream = g_value_get_uint(value);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+endpoint_link_get_property (GObject * object, guint property_id,
+    GValue * value, GParamSpec * pspec)
+{
+  WpEndpointLinkPrivate *priv =
+      wp_endpoint_link_get_instance_private (WP_ENDPOINT_LINK (object));
+
+  switch (property_id) {
+  case LINKPROP_SRC:
+    g_value_take_object (value, g_weak_ref_get (&priv->src));
+    break;
+  case LINKPROP_SRC_STREAM:
+    g_value_set_uint (value, priv->src_stream);
+    break;
+  case LINKPROP_SINK:
+    g_value_take_object (value, g_weak_ref_get (&priv->sink));
+    break;
+  case LINKPROP_SINK_STREAM:
+    g_value_set_uint (value, priv->sink_stream);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+wp_endpoint_link_init_async (GAsyncInitable *initable, int io_priority,
+    GCancellable *cancellable, GAsyncReadyCallback callback, gpointer data)
+{
+  WpEndpointLink *link = WP_ENDPOINT_LINK(initable);
+  WpEndpointLinkPrivate *priv =
+      wp_endpoint_link_get_instance_private (WP_ENDPOINT_LINK (initable));
+  g_autoptr (WpEndpoint) src = g_weak_ref_get (&priv->src);
+  g_autoptr (WpEndpoint) sink = g_weak_ref_get (&priv->sink);
+  g_autoptr (GError) error = NULL;
+  g_autoptr (GVariant) src_props = NULL;
+  g_autoptr (GVariant) sink_props = NULL;
+  WpEndpointPrivate *endpoint_priv;
+
+  /* Create the async task */
+  priv->init_task = g_task_new (initable, cancellable, callback, data);
+
+  /* Prepare the endpoints */
+  if (!WP_ENDPOINT_GET_CLASS (src)->prepare_link (src, priv->src_stream, link,
+      &src_props, &error)) {
+    g_task_return_error (priv->init_task, error);
+    g_clear_object(&priv->init_task);
+    return;
+  }
+  if (!WP_ENDPOINT_GET_CLASS (sink)->prepare_link (sink, priv->sink_stream,
+      link, &sink_props, &error)) {
+    g_task_return_error (priv->init_task, error);
+    g_clear_object(&priv->init_task);
+    return;
+  }
+
+  /* Create the link */
+  g_return_if_fail (WP_ENDPOINT_LINK_GET_CLASS (link)->create);
+  if (!WP_ENDPOINT_LINK_GET_CLASS (link)->create (link, src_props,
+      sink_props, &error)) {
+    g_task_return_error (priv->init_task, error);
+    g_clear_object(&priv->init_task);
+    return;
+  }
+
+  /* Register the link on the endpoints */
+  endpoint_priv = wp_endpoint_get_instance_private (src);
+  g_ptr_array_add (endpoint_priv->links, g_object_ref (link));
+  endpoint_priv = wp_endpoint_get_instance_private (sink);
+  g_ptr_array_add (endpoint_priv->links, g_object_ref (link));
+
+  /* Finish the creation of the endpoint */
+  g_task_return_boolean (priv->init_task, TRUE);
+  g_clear_object(&priv->init_task);
+}
+
+static gboolean
+wp_endpoint_link_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_endpoint_link_async_initable_init (gpointer iface, gpointer iface_data)
+{
+  GAsyncInitableIface *ai_iface = iface;
+
+  ai_iface->init_async = wp_endpoint_link_init_async;
+  ai_iface->init_finish = wp_endpoint_link_init_finish;
+}

 static void
 wp_endpoint_link_init (WpEndpointLink * self)
 {
+  WpEndpointLinkPrivate *priv = wp_endpoint_link_get_instance_private (self);
+
+  /* Init the endpoint weak references */
+  g_weak_ref_init (&priv->src, NULL);
+  g_weak_ref_init (&priv->sink, NULL);
 }

 static void
 wp_endpoint_link_class_init (WpEndpointLinkClass * klass)
 {
-}
+  GObjectClass *object_class = (GObjectClass *) klass;

-void
-wp_endpoint_link_set_endpoints (WpEndpointLink * self, WpEndpoint * src,
-    guint32 src_stream, WpEndpoint * sink, guint32 sink_stream)
-{
-  WpEndpointLinkPrivate *priv;
+  object_class->finalize = endpoint_link_finalize;
+  object_class->set_property = endpoint_link_set_property;
+  object_class->get_property = endpoint_link_get_property;

-  g_return_if_fail (WP_IS_ENDPOINT_LINK (self));
-  g_return_if_fail (WP_IS_ENDPOINT (src));
-  g_return_if_fail (WP_IS_ENDPOINT (sink));
-
-  priv = wp_endpoint_link_get_instance_private (self);
-  priv->src = src;
-  priv->src_stream = src_stream;
-  priv->sink = sink;
-  priv->sink_stream = sink_stream;
+  g_object_class_install_property (object_class, LINKPROP_SRC,
+      g_param_spec_object ("src", "src", "The src endpoint", WP_TYPE_ENDPOINT,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, LINKPROP_SRC_STREAM,
+      g_param_spec_uint ("src-stream", "src-stream", "The src stream",
+          0, G_MAXUINT, 0,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, LINKPROP_SINK,
+      g_param_spec_object ("sink", "sink", "The sink endpoint", WP_TYPE_ENDPOINT,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (object_class, LINKPROP_SINK_STREAM,
+      g_param_spec_uint ("sink-stream", "sink-stream", "The sink stream",
+          0, G_MAXUINT, 0,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 }

 WpEndpoint *
@@ -780,7 +940,7 @@ wp_endpoint_link_get_source_endpoint (WpEndpointLink * self)
  g_return_val_if_fail (WP_IS_ENDPOINT_LINK (self), NULL);

  priv = wp_endpoint_link_get_instance_private (self);
-  return priv->src;
+  return g_weak_ref_get (&priv->src);
 }

 guint32
@@ -802,7 +962,7 @@ wp_endpoint_link_get_sink_endpoint (WpEndpointLink * self)
  g_return_val_if_fail (WP_IS_ENDPOINT_LINK (self), NULL);

  priv = wp_endpoint_link_get_instance_private (self);
-  return priv->sink;
+  return g_weak_ref_get (&priv->sink);
 }

 guint32
@@ -816,19 +976,19 @@ wp_endpoint_link_get_sink_stream (WpEndpointLink * self)
  return priv->sink_stream;
 }

-WpEndpointLink * wp_endpoint_link_new (WpCore * core, WpEndpoint * src,
-    guint32 src_stream, WpEndpoint * sink, guint32 sink_stream, GError ** error)
+void
+wp_endpoint_link_new (WpCore * core, WpEndpoint * src, guint32 src_stream,
+    WpEndpoint * sink, guint32 sink_stream, GAsyncReadyCallback ready,
+    gpointer data)
 {
-  g_autoptr (WpEndpointLink) link = NULL;
-  g_autoptr (GVariant) src_props = NULL;
-  g_autoptr (GVariant) sink_props = NULL;
  const gchar *src_factory = NULL, *sink_factory = NULL;
-  WpEndpointPrivate *endpoint_priv;
+  GVariantBuilder b;
+  g_autoptr (GVariant) link_props = NULL;

-  g_return_val_if_fail (WP_IS_ENDPOINT (src), NULL);
-  g_return_val_if_fail (WP_IS_ENDPOINT (sink), NULL);
-  g_return_val_if_fail (WP_ENDPOINT_GET_CLASS (src)->prepare_link, NULL);
-  g_return_val_if_fail (WP_ENDPOINT_GET_CLASS (sink)->prepare_link, NULL);
+  g_return_if_fail (WP_IS_ENDPOINT (src));
+  g_return_if_fail (WP_IS_ENDPOINT (sink));
+  g_return_if_fail (WP_ENDPOINT_GET_CLASS (src)->prepare_link);
+  g_return_if_fail (WP_ENDPOINT_GET_CLASS (sink)->prepare_link);

  /* find the factory */

@@ -839,52 +999,38 @@ WpEndpointLink * wp_endpoint_link_new (WpCore * core, WpEndpoint * src,

  if (src_factory || sink_factory) {
    if (src_factory && sink_factory && strcmp (src_factory, sink_factory) != 0) {
-      g_set_error (error, WP_DOMAIN_LIBRARY, WP_LIBRARY_ERROR_INVALID_ARGUMENT,
-          "It is not possible to link endpoints that both specify different "
-          "custom link factories");
-      return NULL;
+      g_critical ("It is not possible to link endpoints that both specify"
+          "different custom link factories");
+      return;
    } else if (sink_factory)
      src_factory = sink_factory;
  } else {
    src_factory = "pipewire-simple-endpoint-link";
  }

-  /* create link object */
+  /* Build the properties */
+  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+  g_variant_builder_add (&b, "{sv}", "src",
+      g_variant_new_uint64 ((guint64)src));
+  g_variant_builder_add (&b, "{sv}", "src-stream",
+      g_variant_new_uint32 (src_stream));
+  g_variant_builder_add (&b, "{sv}", "sink",
+      g_variant_new_uint64 ((guint64)sink));
+  g_variant_builder_add (&b, "{sv}", "sink-stream",
+      g_variant_new_uint32 (sink_stream));
+  link_props = g_variant_builder_end (&b);

-  link = wp_factory_make (core, src_factory, WP_TYPE_ENDPOINT_LINK, NULL);
-  if (!link) {
-    g_set_error (error, WP_DOMAIN_LIBRARY, WP_LIBRARY_ERROR_OPERATION_FAILED,
-        "Failed to create link object from factory %s", src_factory);
-    return NULL;
+  /* Create the link object async */
+  wp_factory_make (core, src_factory, WP_TYPE_ENDPOINT_LINK, link_props, ready,
+      data);
 }
-  g_return_val_if_fail (WP_ENDPOINT_LINK_GET_CLASS (link)->create, NULL);

-  /* prepare the link */
-
-  wp_endpoint_link_set_endpoints (link, src, src_stream, sink, sink_stream);
-
-  if (!WP_ENDPOINT_GET_CLASS (src)->prepare_link (src, src_stream, link,
-          &src_props, error))
-    return NULL;
-  if (!WP_ENDPOINT_GET_CLASS (sink)->prepare_link (sink, sink_stream, link,
-          &sink_props, error))
-    return NULL;
-
-  /* create the link */
-
-  if (!WP_ENDPOINT_LINK_GET_CLASS (link)->create (link, src_props, sink_props,
-          error))
-    return NULL;
-
-  /* register the link on the endpoints */
-
-  endpoint_priv = wp_endpoint_get_instance_private (src);
-  g_ptr_array_add (endpoint_priv->links, g_object_ref (link));
-
-  endpoint_priv = wp_endpoint_get_instance_private (sink);
-  g_ptr_array_add (endpoint_priv->links, g_object_ref (link));
-
-  return link;
+WpEndpointLink *
+wp_endpoint_link_new_finish (GObject *initable, GAsyncResult *res,
+    GError **error)
+{
+  GAsyncInitable *ai = G_ASYNC_INITABLE(initable);
+  return WP_ENDPOINT_LINK(g_async_initable_new_finish(ai, res, error));
 }

 void
@@ -892,21 +1038,29 @@ wp_endpoint_link_destroy (WpEndpointLink * self)
 {
  WpEndpointLinkPrivate *priv;
  WpEndpointPrivate *endpoint_priv;
+  g_autoptr (WpEndpoint) src = NULL;
+  g_autoptr (WpEndpoint) sink = NULL;

  g_return_if_fail (WP_IS_ENDPOINT_LINK (self));
  g_return_if_fail (WP_ENDPOINT_LINK_GET_CLASS (self)->destroy);

  priv = wp_endpoint_link_get_instance_private (self);
+  src = g_weak_ref_get (&priv->src);
+  sink = g_weak_ref_get (&priv->sink);

-  WP_ENDPOINT_LINK_GET_CLASS (self)->destroy (self);
-  if (WP_ENDPOINT_GET_CLASS (priv->src)->release_link)
-    WP_ENDPOINT_GET_CLASS (priv->src)->release_link (priv->src, self);
-  if (WP_ENDPOINT_GET_CLASS (priv->sink)->release_link)
-    WP_ENDPOINT_GET_CLASS (priv->sink)->release_link (priv->sink, self);
+  if (src && WP_ENDPOINT_GET_CLASS (src)->release_link)
+    WP_ENDPOINT_GET_CLASS (src)->release_link (src, self);
+  if (sink && WP_ENDPOINT_GET_CLASS (sink)->release_link)
+    WP_ENDPOINT_GET_CLASS (sink)->release_link (sink, self);

-  endpoint_priv = wp_endpoint_get_instance_private (priv->src);
-  g_ptr_array_remove_fast (endpoint_priv->links, self);
-
-  endpoint_priv = wp_endpoint_get_instance_private (priv->sink);
+  if (src) {
+    endpoint_priv = wp_endpoint_get_instance_private (src);
    g_ptr_array_remove_fast (endpoint_priv->links, self);
  }
+  if (sink) {
+    endpoint_priv = wp_endpoint_get_instance_private (sink);
+    g_ptr_array_remove_fast (endpoint_priv->links, self);
+  }
+
+  WP_ENDPOINT_LINK_GET_CLASS (self)->destroy (self);
+}
--- a/lib/wp/endpoint.h
+++ b/lib/wp/endpoint.h
@@ -78,17 +78,16 @@ struct _WpEndpointLinkClass
  void (*destroy) (WpEndpointLink * self);
 };

-void wp_endpoint_link_set_endpoints (WpEndpointLink * self, WpEndpoint * src,
-    guint32 src_stream, WpEndpoint * sink, guint32 sink_stream);
-
 WpEndpoint * wp_endpoint_link_get_source_endpoint (WpEndpointLink * self);
 guint32 wp_endpoint_link_get_source_stream (WpEndpointLink * self);
 WpEndpoint * wp_endpoint_link_get_sink_endpoint (WpEndpointLink * self);
 guint32 wp_endpoint_link_get_sink_stream (WpEndpointLink * self);

-WpEndpointLink * wp_endpoint_link_new (WpCore * core, WpEndpoint * src,
+void wp_endpoint_link_new (WpCore * core, WpEndpoint * src,
    guint32 src_stream, WpEndpoint * sink, guint32 sink_stream,
-    GError ** error);
+    GAsyncReadyCallback ready, gpointer data);
+WpEndpointLink * wp_endpoint_link_new_finish (GObject *initable,
+    GAsyncResult *res, GError **error);
 void wp_endpoint_link_destroy (WpEndpointLink * self);

 G_END_DECLS
--- a/lib/wp/factory.c
+++ b/lib/wp/factory.c
@@ -15,10 +15,7 @@ struct _WpFactory
  GWeakRef core;
  gchar *name;
  GQuark name_quark;
-  union {
-    WpFactoryFunc sync;
-    WpFactoryAsyncFunc async;
-  } create_object;
+  WpFactoryFunc create_object;
 };

 G_DEFINE_TYPE (WpFactory, wp_factory, G_TYPE_OBJECT)
@@ -48,29 +45,20 @@ wp_factory_class_init (WpFactoryClass * klass)
  object_class->finalize = wp_factory_finalize;
 }

-static
-WpFactory * create_factory (WpCore * core, const gchar * name)
+WpFactory *
+wp_factory_new (WpCore * core, const gchar * name,
+    WpFactoryFunc func)
 {
  WpFactory *f = NULL;

+  g_return_val_if_fail (func, NULL);
  g_return_val_if_fail (name != NULL && *name != '\0', NULL);

  f = g_object_new (WP_TYPE_FACTORY, NULL);
  g_weak_ref_init (&f->core, core);
  f->name = g_strdup (name);
  f->name_quark = g_quark_from_string (f->name);
-
-  return f;
-}
-
-WpFactory *
-wp_factory_new (WpCore * core, const gchar * name, WpFactoryFunc func)
-{
-  WpFactory *f = NULL;
-  g_return_val_if_fail (func, NULL);
-
-  f = create_factory(core, name);
-  f->create_object.sync = func;
+  f->create_object = func;

  g_info ("WpFactory:%p new factory: %s", f, name);

@@ -79,23 +67,6 @@ wp_factory_new (WpCore * core, const gchar * name, WpFactoryFunc func)
  return f;
 }

-WpFactory *
-wp_factory_new_async (WpCore * core, const gchar * name,
-    WpFactoryAsyncFunc func)
-{
-  WpFactory *f = NULL;
-  g_return_val_if_fail (func, NULL);
-
-  f = create_factory(core, name);
-  f->create_object.async = func;
-
-  g_info ("WpFactory:%p new async factory: %s", f, name);
-
-  wp_core_register_global (core, WP_GLOBAL_FACTORY, f, g_object_unref);
-
-  return f;
-}
-
 const gchar *
 wp_factory_get_name (WpFactory * self)
 {
@@ -114,23 +85,14 @@ wp_factory_get_core (WpFactory * self)
  return g_weak_ref_get (&self->core);
 }

-gpointer
-wp_factory_create_object (WpFactory * self, GType type, GVariant * properties)
+void
+wp_factory_create_object (WpFactory * self, GType type,
+    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data)
 {
  g_debug ("WpFactory:%p (%s) create object of type %s", self, self->name,
      g_type_name (type));

-  return self->create_object.sync (self, type, properties);
-}
-
-void
-wp_factory_create_object_async (WpFactory * self, GType type,
-    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data)
-{
-  g_debug ("WpFactory:%p (%s) create object async of type %s", self, self->name,
-      g_type_name (type));
-
-  self->create_object.async (self, type, properties, ready, user_data);
+  self->create_object (self, type, properties, ready, user_data);
 }

 struct find_factory_data
@@ -160,20 +122,11 @@ wp_factory_find (WpCore * core, const gchar * name)
  return d.ret;
 }

-gpointer
-wp_factory_make (WpCore * core, const gchar * name, GType type,
-    GVariant * properties)
-{
-  WpFactory *f = wp_factory_find (core, name);
-  if (!f) return NULL;
-  return wp_factory_create_object (f, type, properties);
-}
-
 void
-wp_factory_make_async (WpCore * core, const gchar * name, GType type,
+wp_factory_make (WpCore * core, const gchar * name, GType type,
    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data)
 {
  WpFactory *f = wp_factory_find (core, name);
  if (!f) return;
-  wp_factory_create_object_async (f, type, properties, ready, user_data);
+  wp_factory_create_object (f, type, properties, ready, user_data);
 }
--- a/lib/wp/factory.h
+++ b/lib/wp/factory.h
@@ -18,27 +18,19 @@ G_BEGIN_DECLS
 #define WP_TYPE_FACTORY (wp_factory_get_type ())
 G_DECLARE_FINAL_TYPE (WpFactory, wp_factory, WP, FACTORY, GObject)

-typedef gpointer (*WpFactoryFunc) (WpFactory * self, GType type,
-    GVariant * properties);
-typedef void (*WpFactoryAsyncFunc) (WpFactory * self, GType type,
+typedef void (*WpFactoryFunc) (WpFactory * self, GType type,
    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data);

 WpFactory * wp_factory_new (WpCore * core, const gchar * name,
    WpFactoryFunc func);
-WpFactory * wp_factory_new_async (WpCore * core, const gchar * name,
-    WpFactoryAsyncFunc func);

 const gchar * wp_factory_get_name (WpFactory * self);
 WpCore * wp_factory_get_core (WpFactory * self);
-gpointer wp_factory_create_object (WpFactory * self, GType type,
-    GVariant * properties);
-void wp_factory_create_object_async (WpFactory * self, GType type,
+void wp_factory_create_object (WpFactory * self, GType type,
    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data);

 WpFactory * wp_factory_find (WpCore * core, const gchar * name);
-gpointer wp_factory_make (WpCore * core, const gchar * name, GType type,
-    GVariant * properties);
-void wp_factory_make_async (WpCore * core, const gchar * name, GType type,
+void wp_factory_make (WpCore * core, const gchar * name, GType type,
    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data);

 G_END_DECLS
--- a/modules/module-pipewire.c
+++ b/modules/module-pipewire.c
@@ -19,8 +19,8 @@ void remote_endpoint_init (WpCore * core, struct pw_core * pw_core,
    struct pw_remote * remote);
 void simple_endpoint_factory (WpFactory * factory, GType type,
    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data);
-gpointer simple_endpoint_link_factory (WpFactory * factory, GType type,
-    GVariant * properties);
+void simple_endpoint_link_factory (WpFactory * factory, GType type,
+    GVariant * properties, GAsyncReadyCallback ready, gpointer user_data);

 struct module_data
 {
@@ -89,7 +89,7 @@ on_node_added (WpRemotePipewire *rp, guint id, guint parent_id, gconstpointer p,
  endpoint_props = g_variant_builder_end (&b);

  /* Create the endpoint async */
-  wp_factory_make_async (core, "pipewire-simple-endpoint", WP_TYPE_ENDPOINT,
+  wp_factory_make (core, "pipewire-simple-endpoint", WP_TYPE_ENDPOINT,
      endpoint_props, on_endpoint_created, data);
 }

@@ -162,7 +162,7 @@ wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
  remote_endpoint_init (core, pw_core, pw_remote);

  /* Register simple-endpoint and simple-endpoint-link */
-  wp_factory_new_async (core, "pipewire-simple-endpoint",
+  wp_factory_new (core, "pipewire-simple-endpoint",
      simple_endpoint_factory);
  wp_factory_new (core, "pipewire-simple-endpoint-link",
      simple_endpoint_link_factory);
--- a/modules/module-pipewire/simple-endpoint-link.c
+++ b/modules/module-pipewire/simple-endpoint-link.c
@@ -27,8 +27,13 @@ struct _WpPipewireSimpleEndpointLink
 {
  WpEndpointLink parent;

-  /* The core proxy */
-  struct pw_core_proxy *core_proxy;
+  /* The wireplumber core */
+  GWeakRef core;
+};
+
+enum {
+  PROP_0,
+  PROP_CORE,
 };

 G_DECLARE_FINAL_TYPE (WpPipewireSimpleEndpointLink,
@@ -40,6 +45,51 @@ G_DEFINE_TYPE (WpPipewireSimpleEndpointLink,
 static void
 simple_endpoint_link_init (WpPipewireSimpleEndpointLink * self)
 {
+  /* Init the core weak reference */
+  g_weak_ref_init (&self->core, NULL);
+}
+
+static void
+simple_endpoint_link_finalize (GObject * object)
+{
+  WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(object);
+
+  /* Clear the core weak reference */
+  g_weak_ref_clear (&self->core);
+}
+
+static void
+simple_endpoint_link_set_property (GObject * object, guint property_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  WpPipewireSimpleEndpointLink *self =
+      WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK (object);
+
+  switch (property_id) {
+  case PROP_CORE:
+    g_weak_ref_set (&self->core, g_value_get_object (value));
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+simple_endpoint_link_get_property (GObject * object, guint property_id,
+    GValue * value, GParamSpec * pspec)
+{
+  WpPipewireSimpleEndpointLink *self =
+      WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK (object);
+
+  switch (property_id) {
+  case PROP_CORE:
+    g_value_take_object (value, g_weak_ref_get (&self->core));
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
 }

 static gboolean
@@ -47,12 +97,18 @@ simple_endpoint_link_create (WpEndpointLink * epl, GVariant * src_data,
    GVariant * sink_data, GError ** error)
 {
  WpPipewireSimpleEndpointLink *self = WP_PIPEWIRE_SIMPLE_ENDPOINT_LINK(epl);
+  g_autoptr (WpCore) core = g_weak_ref_get (&self->core);
+  WpRemotePipewire *remote_pipewire;
  struct pw_properties *props;
  guint32 output_node_id, input_node_id;
  GVariant *src_ports, *sink_ports;
  GVariantIter *out_iter, *in_iter;
  guint64 out_ptr, in_ptr;

+  /* Get the remote pipewire */
+  remote_pipewire = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
+  g_return_val_if_fail (remote_pipewire, FALSE);
+
  /* Get the node ids and port ids */
  if (!g_variant_lookup (src_data, "node-id", "u", &output_node_id))
      return FALSE;
@@ -90,8 +146,8 @@ simple_endpoint_link_create (WpEndpointLink * epl, GVariant * src_data,
      pw_properties_setf(props, PW_LINK_INPUT_PORT_ID, "%d", in_id);

      /* Create the link */
-      pw_core_proxy_create_object(self->core_proxy, "link-factory",
-          PW_TYPE_INTERFACE_Link, PW_VERSION_LINK, &props->dict, 0);
+      wp_remote_pipewire_create_object(remote_pipewire, "link-factory",
+          PW_TYPE_INTERFACE_Link, &props->dict);

      /* Clean up */
      pw_properties_free(props);
@@ -112,49 +168,54 @@ simple_endpoint_link_destroy (WpEndpointLink * self)
 static void
 simple_endpoint_link_class_init (WpPipewireSimpleEndpointLinkClass * klass)
 {
+  GObjectClass *object_class = (GObjectClass *) klass;
  WpEndpointLinkClass *link_class = (WpEndpointLinkClass *) klass;

+  object_class->finalize = simple_endpoint_link_finalize;
+  object_class->set_property = simple_endpoint_link_set_property;
+  object_class->get_property = simple_endpoint_link_get_property;
+
  link_class->create = simple_endpoint_link_create;
  link_class->destroy = simple_endpoint_link_destroy;
+
+  g_object_class_install_property (object_class, PROP_CORE,
+      g_param_spec_object ("core", "core",
+          "The wireplumber core object this links belongs to", WP_TYPE_CORE,
+          G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 }

-gpointer
+void
 simple_endpoint_link_factory (WpFactory * factory, GType type,
-    GVariant * properties)
+    GVariant * properties, GAsyncReadyCallback ready, gpointer data)
 {
-  WpCore *wp_core = NULL;
-  WpRemote *remote;
-  struct pw_remote *pw_remote;
+  g_autoptr(WpCore) core = NULL;
+  guint64 src, sink;
+  guint src_stream, sink_stream;

  /* Make sure the type is an endpoint link */
-  if (type != WP_TYPE_ENDPOINT_LINK)
-    return NULL;
+  g_return_if_fail (type == WP_TYPE_ENDPOINT_LINK);

-  /* Get the WirePlumber core */
-  wp_core = wp_factory_get_core(factory);
-  if (!wp_core) {
-    g_warning("failed to get wireplumbe core. Skipping...");
-    return NULL;
-  }
+  /* Get the Core */
+  core = wp_factory_get_core (factory);
+  g_return_if_fail (core);

-  /* Get the remote */
-  remote = wp_core_get_global(wp_core, WP_GLOBAL_REMOTE_PIPEWIRE);
-  if (!remote) {
-    g_warning("failed to get core remote. Skipping...");
-    return NULL;
-  }
+  /* Get the properties */
+  if (!g_variant_lookup (properties, "src", "t", &src))
+      return;
+  if (!g_variant_lookup (properties, "src-stream", "u", &src_stream))
+      return;
+  if (!g_variant_lookup (properties, "sink", "t", &sink))
+      return;
+  if (!g_variant_lookup (properties, "sink-stream", "u", &sink_stream))
+      return;

  /* Create the endpoint link */
-  WpPipewireSimpleEndpointLink *epl = g_object_new (
-      simple_endpoint_link_get_type (), NULL);
-
-  /* Set the core proxy */
-  g_object_get (remote, "pw-remote", &pw_remote, NULL);
-  epl->core_proxy = pw_remote_get_core_proxy(pw_remote);
-  if (!epl->core_proxy) {
-    g_warning("failed to get core proxy. Skipping...");
-    return NULL;
-  }
-
-  return epl;
+  g_async_initable_new_async (
+      simple_endpoint_link_get_type (), G_PRIORITY_DEFAULT, NULL, ready, data,
+      "src", (gpointer)src,
+      "src-stream", src_stream,
+      "sink", (gpointer)sink,
+      "sink-stream", sink_stream,
+      "core", core,
+      NULL);
 }
--- a/modules/module-pipewire/simple-endpoint.c
+++ b/modules/module-pipewire/simple-endpoint.c
@@ -36,6 +36,9 @@ struct _WpPipewireSimpleEndpoint
  /* Handler */
  gulong proxy_node_done_handler_id;

+  /* Direction */
+  enum pw_direction direction;
+
  /* Proxies */
  WpProxyNode *proxy_node;
  struct spa_hook node_proxy_listener;
@@ -199,7 +202,7 @@ emit_endpoint_ports(WpPipewireSimpleEndpoint *self)
  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_OUTPUT),
+      SPA_PARAM_PROFILE_direction,  SPA_POD_Id(self->direction),
      SPA_PARAM_PROFILE_format,     SPA_POD_Pod(param));

  /* Set the param profile to emit the ports */
@@ -261,6 +264,7 @@ wp_simple_endpoint_init_async (GAsyncInitable *initable, int io_priority,
 {
  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (initable);
  g_autoptr (WpCore) core = wp_endpoint_get_core(WP_ENDPOINT(self));
+  const gchar *media_class = wp_endpoint_get_media_class (WP_ENDPOINT (self));
  struct pw_node_proxy *node_proxy = NULL;

  /* Create the async task */
@@ -269,6 +273,14 @@ wp_simple_endpoint_init_async (GAsyncInitable *initable, int io_priority,
  /* Init the proxies_port array */
  self->proxies_port = g_ptr_array_new_full(2, (GDestroyNotify)g_object_unref);

+  /* Set the direction */
+  if (g_str_has_prefix (media_class, "Stream/Input"))
+    self->direction = PW_DIRECTION_INPUT;
+  else if (g_str_has_prefix (media_class, "Stream/Output"))
+    self->direction = PW_DIRECTION_OUTPUT;
+  else
+    g_critical ("failed to parse direction");
+
  /* Register a port_added callback */
  self->remote_pipewire = wp_core_get_global (core, WP_GLOBAL_REMOTE_PIPEWIRE);
  g_return_if_fail(self->remote_pipewire);
--- a/modules/module-pw-alsa-udev.c
+++ b/modules/module-pw-alsa-udev.c
@@ -76,7 +76,7 @@ on_node_added(WpRemotePipewire *rp, guint id, guint parent_id, gconstpointer p,
  endpoint_props = g_variant_builder_end (&b);

  /* Create the endpoint async */
-  wp_factory_make_async (core, "pw-audio-softdsp-endpoint", WP_TYPE_ENDPOINT,
+  wp_factory_make (core, "pw-audio-softdsp-endpoint", WP_TYPE_ENDPOINT,
      endpoint_props, on_endpoint_created, impl);
 }

--- a/modules/module-pw-audio-softdsp-endpoint.c
+++ b/modules/module-pw-audio-softdsp-endpoint.c
@@ -727,5 +727,5 @@ void
 wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
 {
  /* Register the softdsp endpoint */
-  wp_factory_new_async (core, "pw-audio-softdsp-endpoint", endpoint_factory);
+  wp_factory_new (core, "pw-audio-softdsp-endpoint", endpoint_factory);
 }
--- a/modules/module-simple-policy.c
+++ b/modules/module-simple-policy.c
@@ -225,6 +225,29 @@ simple_policy_endpoint_removed (WpPolicy *policy, WpEndpoint *ep)
      g_object_ref (self), g_object_unref);
 }

+static void
+on_endpoint_link_created(GObject *initable, GAsyncResult *res, gpointer d)
+{
+  g_autoptr (WpEndpointLink) link = NULL;
+  g_autoptr (GError) error = NULL;
+  g_autoptr (WpEndpoint) src_ep = NULL;
+  g_autoptr (WpEndpoint) sink_ep = NULL;
+
+  /* Get the link */
+  link = wp_endpoint_link_new_finish(initable, res, &error);
+  g_return_if_fail (link);
+
+  /* Log linking info */
+  if (error) {
+    g_warning ("Could not link endpoints: %s\n", error->message);
+  } else {
+    src_ep = wp_endpoint_link_get_source_endpoint (link);
+    sink_ep = wp_endpoint_link_get_sink_endpoint (link);
+    g_info ("Sucessfully linked '%s' to '%s'\n", wp_endpoint_get_name (src_ep),
+        wp_endpoint_get_name (sink_ep));
+  }
+}
+
 static gboolean
 simple_policy_handle_endpoint (WpPolicy *policy, WpEndpoint *ep)
 {
@@ -232,34 +255,40 @@ simple_policy_handle_endpoint (WpPolicy *policy, WpEndpoint *ep)
  GVariantDict d;
  g_autoptr (WpCore) core = NULL;
  g_autoptr (WpEndpoint) target = NULL;
-  g_autoptr (GError) error = NULL;
  guint32 stream_id;
+  gboolean is_sink = FALSE;

-  /* TODO: For now we only accept audio output clients */
+  /* TODO: For now we only accept audio stream clients */
  media_class = wp_endpoint_get_media_class(ep);
-  if (!g_str_equal (media_class, "Stream/Output/Audio"))
+  if (!g_str_has_prefix (media_class, "Stream") ||
+      !g_str_has_suffix (media_class, "Audio"))
    return FALSE;

+  /* Detect if the client is a sink or a source */
+  is_sink = g_str_has_prefix (media_class, "Stream/Input");
+
  /* Locate the target endpoint */
  g_variant_dict_init (&d, NULL);
  g_variant_dict_insert (&d, "action", "s", "link");
-  g_variant_dict_insert (&d, "media.class", "s", "Audio/Sink");
+  g_variant_dict_insert (&d, "media.class", "s",
+      is_sink ? "Audio/Source" : "Audio/Sink");
  /* TODO: more properties are needed here */

  core = wp_policy_get_core (policy);
  target = wp_policy_find_endpoint (core, g_variant_dict_end (&d), &stream_id);
  if (!target) {
-    g_warning ("Could not find an Audio/Sink target endpoint\n");
+    g_warning ("Could not find a target endpoint\n");
    /* TODO: we should kill the client, otherwise it's going to hang waiting */
    return FALSE;
  }

  /* Link the client with the target */
-  if (!wp_endpoint_link_new (core, ep, 0, target, stream_id, &error)) {
-    g_warning ("Could not link endpoints: %s\n", error->message);
+  if (is_sink) {
+    wp_endpoint_link_new (core, target, 0, ep, stream_id,
+        on_endpoint_link_created, NULL);
  } else {
-    g_info ("Sucessfully linked '%s' to '%s'\n", wp_endpoint_get_name (ep),
-        wp_endpoint_get_name (target));
+    wp_endpoint_link_new (core, ep, 0, target, stream_id,
+        on_endpoint_link_created, NULL);
  }

  return TRUE;