module-pipewire and simple-endpoint: use the new Proxy API

modules/module-pipewire.c

@@ -27,150 +27,258 @@ struct module_data
 {
   WpModule *module;
 
+  /* Registry */
   struct pw_registry_proxy *registry_proxy;
   struct spa_hook registry_listener;
 
-  struct pw_core_proxy *core_proxy;
-  struct spa_hook core_listener;
-  GQueue *done_queue;
+  /* Client nodes info */
+  GHashTable *client_nodes_info;
 };
 
+struct endpoint_info
+{
+  gchar *name;
+  gchar *media_class;
+  const struct pw_proxy *proxy;
+};
 
-typedef void (*WpDoneCallback)(gpointer, gpointer);
-
-struct done_data
+struct proxy_info
 {
-  WpDoneCallback callback;
-  gpointer data;
-  GDestroyNotify data_destroy;
+  const struct module_data *data;
+  uint32_t node_id;
+  WpProxyPort *proxy_port;
 };
 
 static void
-done_data_destroy(gpointer p)
+endpoint_info_destroy(gpointer p)
 {
-  struct done_data *dd = p;
-  if (dd->data_destroy) {
-    dd->data_destroy(dd->data);
-    dd->data = NULL;
+  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 done_data, dd);
+
+  /* Clean up */
+  g_slice_free (struct endpoint_info, p);
 }
 
 static void
-sync_core_with_callback(struct module_data* impl, WpDoneCallback callback,
-    gpointer data, GDestroyNotify data_destroy)
+proxy_info_destroy(gpointer p)
 {
-  struct done_data *dd = g_new0(struct done_data, 1);
+  struct proxy_info *pi = p;
 
-  /* Set the data */
-  dd->callback = callback;
-  dd->data = data;
-  dd->data_destroy = data_destroy;
-
-  /* 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 */
-  pw_core_proxy_sync(impl->core_proxy, 0, 0);
+  /* Clean up */
+  g_slice_free (struct proxy_info, p);
 }
 
 static void
-core_done(void *d, uint32_t id, int seq)
+proxy_node_created(GObject *initable, GAsyncResult *res, gpointer d)
 {
-  struct module_data * 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(impl, dd->data);
-    done_data_destroy(dd);
-  }
-}
+  struct proxy_info *pi = d;
+  const struct module_data *data = pi->data;
+  g_autoptr (WpCore) core = wp_module_get_core (data->module);
+  WpProxyNode *proxy_node = NULL;
+  struct endpoint_info *ei = NULL;
+  WpEndpoint *endpoint = NULL;
+  g_autoptr (GVariant) endpoint_props = NULL;
+  GVariantBuilder b;
 
-static const struct pw_core_proxy_events core_events = {
-  PW_VERSION_CORE_EVENTS,
-  .done = core_done
-};
+  /* Get the proxy */
+  proxy_node = wp_proxy_node_new_finish(initable, res, NULL);
+  if (!proxy_node)
+    return;
+
+  /* Register the proxy node */
+  wp_proxy_register(WP_PROXY(proxy_node));
+
+  /* Get the client node info */
+  ei = g_hash_table_lookup(data->client_nodes_info,
+      GINT_TO_POINTER(pi->node_id));
+  if (!ei)
+    return;
+
+  /* Set the properties */
+  g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+  g_variant_builder_add (&b, "{sv}",
+      "name", ei->name ? g_variant_new_string (ei->name) :
+          g_variant_new_take_string (
+              g_strdup_printf ("Stream %u", pi->node_id)));
+  g_variant_builder_add (&b, "{sv}",
+      "media-class", g_variant_new_string (ei->media_class));
+  g_variant_builder_add (&b, "{sv}",
+      "proxy-node", g_variant_new_uint64 ((guint64) proxy_node));
+  g_variant_builder_add (&b, "{sv}",
+      "proxy-port", g_variant_new_uint64 ((guint64)
+          g_object_ref(pi->proxy_port)));
+  endpoint_props = g_variant_builder_end (&b);
+
+  /* Create the endpoint */
+  endpoint = wp_factory_make (core, "pipewire-simple-endpoint",
+      WP_TYPE_ENDPOINT, endpoint_props);
+
+  /* Register the endpoint */
+  wp_endpoint_register (endpoint, core);
+
+  /* Clean up */
+  proxy_info_destroy (pi);
+}
 
 static void
-register_endpoint (struct module_data* data, WpEndpoint *ep)
+proxy_port_created(GObject *initable, GAsyncResult *res, gpointer d)
 {
-  g_autoptr (WpCore) core = NULL;
-  core = wp_module_get_core (data->module);
-  g_return_if_fail (core != NULL);
-  wp_endpoint_register (ep, core);
+  struct proxy_info *pi = d;
+  const struct module_data *data = pi->data;
+  g_autoptr (WpCore) core = wp_module_get_core (data->module);
+  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;
+
+  /* Register the proxy port */
+  wp_proxy_register(WP_PROXY(proxy_port));
+
+  /* Forward the proxy port */
+  pi->proxy_port = proxy_port;
+
+  /* Get the node proxy */
+  proxy = pw_registry_proxy_bind (data->registry_proxy, pi->node_id,
+      PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0);
+  if (!proxy)
+    return;
+
+  /* Create the proxy node asynchronically */
+  wp_proxy_node_new(core, proxy, proxy_node_created, pi);
 }
 
 static void
-registry_global (void * d, uint32_t id, uint32_t parent_id,
-    uint32_t permissions, uint32_t type, uint32_t version,
+handle_node (struct module_data *data, uint32_t id, uint32_t parent_id,
     const struct spa_dict * props)
 {
-  struct module_data *data = d;
+  struct endpoint_info *ei = NULL;
   const gchar *name;
   const gchar *media_class;
   struct pw_proxy *proxy;
-  GVariantBuilder b;
-  g_autoptr (GVariant) endpoint_props = NULL;
-  g_autoptr (WpCore) core = NULL;
-  g_autoptr (WpEndpoint) endpoint = NULL;
   struct spa_audio_info_raw format = { 0, };
   struct spa_pod *param;
   struct spa_pod_builder pod_builder = { 0, };
   char buf[1024];
 
-  /* listen for client "Stream" nodes and create endpoints for them */
-  if (type == PW_TYPE_INTERFACE_Node &&
-      props && (media_class = spa_dict_lookup(props, "media.class")) &&
-      g_str_has_prefix (media_class, "Stream/"))
-  {
-    name = spa_dict_lookup (props, "media.name");
-    if (!name)
-      name = spa_dict_lookup (props, "node.name");
-
-    g_debug ("found stream node: id:%u ; name:%s ; media_class:%s", id, name,
-        media_class);
-
-    proxy = pw_registry_proxy_bind (data->registry_proxy,
-        id, type, PW_VERSION_NODE, 0);
-
-    /* TODO: we need to get this from the EnumFormat event */
-    format.format = SPA_AUDIO_FORMAT_F32P;
-    format.flags = 1;
-    format.rate = 48000;
-    format.channels = 2;
-    format.position[0] = 0;
-    format.position[1] = 0;
-
-    /* Set the profile */
-    spa_pod_builder_init(&pod_builder, buf, sizeof(buf));
-    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_format,     SPA_POD_Pod(param));
-    pw_node_proxy_set_param((struct pw_node_proxy*)proxy,
-        SPA_PARAM_Profile, 0, param);
-
-    g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
-    g_variant_builder_add (&b, "{sv}", "node-id", g_variant_new_uint32 (id));
-    g_variant_builder_add (&b, "{sv}",
-        "name", name ? g_variant_new_string (name) :
-            g_variant_new_take_string (g_strdup_printf ("Stream %u", id)));
-    g_variant_builder_add (&b, "{sv}",
-        "media-class", g_variant_new_string (media_class));
-    g_variant_builder_add (&b, "{sv}",
-        "node-proxy", g_variant_new_uint64 ((guint64) proxy));
-    endpoint_props = g_variant_builder_end (&b);
-
-    core = wp_module_get_core (data->module);
-    g_return_if_fail (core != NULL);
-
-    endpoint = wp_factory_make (core, "pipewire-simple-endpoint",
-        WP_TYPE_ENDPOINT, endpoint_props);
-    sync_core_with_callback (data, (WpDoneCallback) register_endpoint,
-        g_steal_pointer (&endpoint), g_object_unref);
+  /* Make sure the node has properties */
+  if (!props) {
+    g_warning("node has no properties, skipping...");
+    return;
+  }
+
+  /* Get the media_class */
+  media_class = spa_dict_lookup(props, "media.class");
+
+  /* Only handle client Stream nodes */
+  if (!g_str_has_prefix (media_class, "Stream/"))
+    return;
+
+  /* Get the name */
+  name = spa_dict_lookup (props, "media.name");
+  if (!name)
+    name = spa_dict_lookup (props, "node.name");
+
+  g_debug ("found stream node: id:%u ; name:%s ; media_class:%s", id, name,
+      media_class);
+
+  /* Get the proxy */
+  proxy = pw_registry_proxy_bind (data->registry_proxy, id,
+      PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0);
+
+  /* TODO: Assume all clients have this format for now */
+  format.format = SPA_AUDIO_FORMAT_F32P;
+  format.flags = 1;
+  format.rate = 48000;
+  format.channels = 1;
+  format.position[0] = 0;
+
+  /* Set the profile */
+  spa_pod_builder_init(&pod_builder, buf, sizeof(buf));
+  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_format,     SPA_POD_Pod(param));
+  pw_node_proxy_set_param((struct pw_node_proxy*)proxy,
+      SPA_PARAM_Profile, 0, param);
+
+  /* Create the endpoint info */
+  ei = g_new0(struct endpoint_info, 1);
+  ei->name = g_strdup(name);
+  ei->media_class = g_strdup(media_class);
+  ei->proxy = proxy;
+
+  /* Insert the client node info in the hash table */
+  g_hash_table_insert(data->client_nodes_info, GINT_TO_POINTER (id), ei);
+}
+
+static void
+handle_port(struct module_data *data, uint32_t id, uint32_t parent_id,
+            const struct spa_dict *props)
+{
+  g_autoptr (WpCore) core = wp_module_get_core (data->module);
+  struct proxy_info *pi = NULL;
+  struct pw_proxy *proxy = NULL;
+
+  /* Only handle ports whose parent is an alsa node */
+  if (!g_hash_table_contains(data->client_nodes_info,
+      GINT_TO_POINTER (parent_id)))
+    return;
+
+  /* Get the port proxy */
+  proxy = pw_registry_proxy_bind (data->registry_proxy, id,
+      PW_TYPE_INTERFACE_Port, PW_VERSION_PORT, 0);
+  if (!proxy)
+    return;
+
+  /* Create the port info */
+  pi = g_new0(struct proxy_info, 1);
+  pi->data = data;
+  pi->node_id = parent_id;
+  pi->proxy_port = NULL;
+
+  /* Create the proxy port asynchronically */
+  wp_proxy_port_new(core, proxy, proxy_port_created, pi);
+}
+
+static void
+registry_global(void *d, uint32_t id, uint32_t parent_id,
+		uint32_t permissions, uint32_t type, uint32_t version,
+		const struct spa_dict *props)
+{
+  struct module_data *data = d;
+
+  switch (type) {
+  case PW_TYPE_INTERFACE_Node:
+    handle_node(data, id, parent_id, props);
+    break;
+
+  case PW_TYPE_INTERFACE_Port:
+    handle_port(data, id, parent_id, props);
+    break;
+
+  default:
+    break;
   }
 }
 
@@ -188,9 +296,7 @@ on_remote_connected (WpRemote *remote, WpRemoteState state,
 
   g_object_get (remote, "pw-remote", &pw_remote, NULL);
 
-  core_proxy = data->core_proxy = pw_remote_get_core_proxy (pw_remote);
-  pw_core_proxy_add_listener(data->core_proxy, &data->core_listener,
-      &core_events, data);
+  core_proxy = pw_remote_get_core_proxy (pw_remote);
   data->registry_proxy = pw_core_proxy_get_registry (core_proxy,
       PW_TYPE_INTERFACE_Registry, PW_VERSION_REGISTRY, 0);
   pw_registry_proxy_add_listener(data->registry_proxy,
@@ -202,7 +308,13 @@ module_destroy (gpointer d)
 {
   struct module_data *data = d;
 
-  g_queue_free_full(data->done_queue, done_data_destroy);
+  /* Destroy the hash table */
+  if (data->client_nodes_info) {
+    g_hash_table_destroy(data->client_nodes_info);
+    data->client_nodes_info = NULL;
+  }
+
+  /* Clean up */
   g_slice_free (struct module_data, data);
 }
 
@@ -221,9 +333,12 @@ wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
     return;
   }
 
+  /* Create the module data */
   data = g_slice_new0 (struct module_data);
   data->module = module;
-  data->done_queue = g_queue_new();
+  data->client_nodes_info = g_hash_table_new_full (g_direct_hash,
+      g_direct_equal, NULL, endpoint_info_destroy);
+
   wp_module_set_destroy_callback (module, module_destroy, data);
 
   g_signal_connect (remote, "state-changed::connected",

modules/module-pipewire/simple-endpoint.c

@@ -22,14 +22,11 @@ struct _WpPipewireSimpleEndpoint
 {
   WpEndpoint parent;
 
-  /* Node */
-  struct pw_node_proxy *node;
-  struct spa_hook proxy_listener;
+  /* Proxy */
+  WpProxyNode *proxy_node;
+  WpProxyPort *proxy_port;
   struct spa_hook node_proxy_listener;
 
-  /* Info */
-  struct pw_node_info *info;
-
   /* controls cache */
   gfloat volume;
   gboolean mute;
@@ -38,6 +35,7 @@ struct _WpPipewireSimpleEndpoint
 enum {
   PROP_0,
   PROP_NODE_PROXY,
+  PROP_PORT_PROXY
 };
 
 enum {
@@ -96,15 +94,8 @@ node_proxy_param (void *object, int seq, uint32_t id,
   }
 }
 
-static void node_proxy_info(void *object, const struct pw_node_info *info)
-{
-  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
-  self->info = pw_node_info_update(self->info, info);
-}
-
 static const struct pw_node_proxy_events node_node_proxy_events = {
   PW_VERSION_NODE_PROXY_EVENTS,
-  .info = node_proxy_info,
   .param = node_proxy_param,
 };
 
@@ -113,21 +104,6 @@ simple_endpoint_init (WpPipewireSimpleEndpoint * self)
 {
 }
 
-static void
-node_proxy_destroy (void *data)
-{
-  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (data);
-  self->node = 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
 simple_endpoint_constructed (GObject * object)
 {
@@ -135,13 +111,13 @@ simple_endpoint_constructed (GObject * object)
   GVariantDict d;
   uint32_t ids[1] = { SPA_PARAM_Props };
   uint32_t n_ids = 1;
+  struct pw_node_proxy *node_proxy = NULL;
 
-  pw_proxy_add_listener ((struct pw_proxy *) self->node, &self->proxy_listener,
-      &node_proxy_events, self);
-
-  pw_node_proxy_add_listener (self->node, &self->node_proxy_listener,
+  /* Add a custom node proxy event listener */
+  node_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy_node));
+  pw_node_proxy_add_listener (node_proxy, &self->node_proxy_listener,
       &node_node_proxy_events, self);
-  pw_node_proxy_subscribe_params (self->node, ids, n_ids);
+  pw_node_proxy_subscribe_params (node_proxy, ids, n_ids);
 
   g_variant_dict_init (&d, NULL);
   g_variant_dict_insert (&d, "id", "u", 0);
@@ -176,9 +152,10 @@ simple_endpoint_finalize (GObject * object)
 {
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
 
-  if (self->node) {
-    spa_hook_remove (&self->proxy_listener);
-    pw_proxy_destroy ((struct pw_proxy *) self->node);
+  /* Unref the proxy node */
+  if (self->proxy_node) {
+    g_object_unref(self->proxy_node);
+    self->proxy_node = NULL;
   }
 
   G_OBJECT_CLASS (simple_endpoint_parent_class)->finalize (object);
@@ -192,7 +169,12 @@ simple_endpoint_set_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_NODE_PROXY:
-    self->node = 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);
@@ -208,7 +190,10 @@ simple_endpoint_get_property (GObject * object, guint property_id,
 
   switch (property_id) {
   case PROP_NODE_PROXY:
-    g_value_set_pointer (value, self->node);
+    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);
@@ -221,6 +206,7 @@ simple_endpoint_prepare_link (WpEndpoint * ep, guint32 stream_id,
     WpEndpointLink * link, GVariant ** properties, GError ** error)
 {
   WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (ep);
+  const struct pw_node_info *node_info = NULL;
   GVariantBuilder b;
 
   /* TODO: Since the linking with a 1 port client works when passing -1 as
@@ -228,12 +214,15 @@ simple_endpoint_prepare_link (WpEndpoint * ep, guint32 stream_id,
    * properties. However, we need to add logic here and select the correct
    * port in case the client has more than 1 port */
 
-  /* Set the port format here */
+  /* Get the node info */
+  node_info = wp_proxy_node_get_info(self->proxy_node);
+  if (!node_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->info->id));
+      g_variant_new_uint32 (node_info->id));
   g_variant_builder_add (&b, "{sv}", "node-port-id",
       g_variant_new_uint32 (-1));
   *properties = g_variant_builder_end (&b);
@@ -266,6 +255,10 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
   struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buf, sizeof(buf));
   float volume;
   bool mute;
+  struct pw_node_proxy *node_proxy = NULL;
+
+  /* Get the node proxy */
+  node_proxy = wp_proxy_get_pw_proxy(WP_PROXY(self->proxy_node));
 
   switch (control_id) {
     case CONTROL_VOLUME:
@@ -274,7 +267,7 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
       g_debug("WpEndpoint:%p set volume control (%u) value, vol:%f", self,
           control_id, volume);
 
-      pw_node_proxy_set_param (self->node,
+      pw_node_proxy_set_param (node_proxy,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
@@ -288,7 +281,7 @@ simple_endpoint_set_control_value (WpEndpoint * ep, guint32 control_id,
       g_debug("WpEndpoint:%p set mute control (%u) value, mute:%d", self,
           control_id, mute);
 
-      pw_node_proxy_set_param (self->node,
+      pw_node_proxy_set_param (node_proxy,
           SPA_PARAM_Props, 0,
           spa_pod_builder_add_object (&b,
               SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
@@ -320,8 +313,12 @@ simple_endpoint_class_init (WpPipewireSimpleEndpointClass * klass)
   endpoint_class->set_control_value = simple_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 pw_node_proxy* to wrap",
+      g_param_spec_object ("node-proxy", "node-proxy",
+          "Pointer to the node proxy of the client", 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 proxy of the client", WP_TYPE_PROXY_PORT,
           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 }
 
@@ -329,7 +326,8 @@ gpointer
 simple_endpoint_factory (WpFactory * factory, GType type,
     GVariant * properties)
 {
-  guint64 proxy;
+  guint64 proxy_node;
+  guint64 proxy_port;
   const gchar *name;
   const gchar *media_class;
 
@@ -342,12 +340,15 @@ simple_endpoint_factory (WpFactory * factory, GType type,
       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, "proxy-port", "t", &proxy_port))
       return NULL;
 
   return g_object_new (simple_endpoint_get_type (),
       "name", name,
       "media-class", media_class,
-      "node-proxy", (gpointer) proxy,
+      "node-proxy", (gpointer) proxy_node,
+      "port-proxy", (gpointer) proxy_port,
       NULL);
 }