gst: deviceprodiver: Use GstPipeWireCore and some cleanups

Quoting the commit introducing `GstPipeWireCore`:
```
Make all sources in the same process with the same fd share the
connection to the server. This makes it possible to set the same
fd on multiple sources/sinks and have them all use the same
connection, like when capturing multiple monitors from screencast
with the portal.
```

Do the same for `GstPipeWireDeviceProvider`, so it can share a
connection with device sinks as well. This will be needed for fd based
connections introduced in the next commit.

Further more it allows some cleanups.

1: 70652d1a37

src/gst/gstpipewiredeviceprovider.c

@@ -180,15 +180,6 @@ enum
   PROP_LAST
 };
 
-struct core_data {
-  int seq;
-  GstPipeWireDeviceProvider *self;
-  struct spa_hook core_listener;
-  struct pw_registry *registry;
-  struct spa_hook registry_listener;
-  struct spa_list nodes;
-};
-
 struct node_data {
   struct spa_list link;
   GstPipeWireDeviceProvider *self;
@@ -211,10 +202,10 @@ struct port_data {
   struct spa_hook port_listener;
 };
 
-static struct node_data *find_node_data(struct core_data *rd, uint32_t id)
+static struct node_data *find_node_data(struct spa_list *nodes, uint32_t id)
 {
   struct node_data *n;
-  spa_list_for_each(n, &rd->nodes, link) {
+  spa_list_for_each(n, nodes, link) {
     if (n->id == id)
       return n;
   }
@@ -269,12 +260,11 @@ new_node (GstPipeWireDeviceProvider *self, struct node_data *data)
   return GST_DEVICE (gstdev);
 }
 
-static void do_add_nodes(struct core_data *rd)
+static void do_add_nodes(GstPipeWireDeviceProvider *self)
 {
-  GstPipeWireDeviceProvider *self = rd->self;
   struct node_data *nd;
 
-  spa_list_for_each(nd, &rd->nodes, link) {
+  spa_list_for_each(nd, &self->nodes, link) {
     if (nd->dev != NULL)
 	    continue;
     pw_log_info("add node %d", nd->id);
@@ -290,22 +280,21 @@ static void do_add_nodes(struct core_data *rd)
 
 static void resync(GstPipeWireDeviceProvider *self)
 {
-  self->seq = pw_core_sync(self->core, PW_ID_CORE, self->seq);
+  self->seq = pw_core_sync(self->core->core, PW_ID_CORE, self->seq);
   pw_log_debug("resync %d", self->seq);
 }
 
 static void
 on_core_done (void *data, uint32_t id, int seq)
 {
-  struct core_data *rd = data;
+  GstPipeWireDeviceProvider *self = data;
-  GstPipeWireDeviceProvider *self = rd->self;
 
   pw_log_debug("check %d %d", seq, self->seq);
   if (id == PW_ID_CORE && seq == self->seq) {
-    do_add_nodes(rd);
+    do_add_nodes(self);
     self->end = true;
-    if (self->loop)
+    if (self->core)
-      pw_thread_loop_signal (self->loop, FALSE);
+      pw_thread_loop_signal (self->core->loop, FALSE);
   }
 }
 
@@ -313,8 +302,7 @@ on_core_done (void *data, uint32_t id, int seq)
 static void
 on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
 {
-  struct core_data *rd = data;
+  GstPipeWireDeviceProvider *self = data;
-  GstPipeWireDeviceProvider *self = rd->self;
 
   pw_log_warn("error id:%u seq:%d res:%d (%s): %s",
           id, seq, res, spa_strerror(res), message);
@@ -322,7 +310,7 @@ on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
   if (id == PW_ID_CORE) {
     self->error = res;
   }
-  pw_thread_loop_signal(self->loop, FALSE);
+  pw_thread_loop_signal(self->core->loop, FALSE);
 }
 
 static const struct pw_core_events core_events = {
@@ -470,8 +458,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
                                 const char *type, uint32_t version,
                                 const struct spa_dict *props)
 {
-  struct core_data *rd = data;
+  GstPipeWireDeviceProvider *self = data;
-  GstPipeWireDeviceProvider *self = rd->self;
   GstDeviceProvider *provider = (GstDeviceProvider*)self;
   struct node_data *nd;
   const char *str;
@@ -479,7 +466,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
   if (spa_streq(type, PW_TYPE_INTERFACE_Node)) {
     struct pw_node *node;
 
-    node = pw_registry_bind(rd->registry,
+    node = pw_registry_bind(self->registry,
                     id, type, PW_VERSION_NODE, sizeof(*nd));
     if (node == NULL)
       goto no_mem;
@@ -502,7 +489,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
     nd->id = id;
     if (!props || !spa_atou64(spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL), &nd->serial, 0))
       nd->serial = SPA_ID_INVALID;
-    spa_list_append(&rd->nodes, &nd->link);
+    spa_list_append(&self->nodes, &nd->link);
     pw_node_add_listener(node, &nd->node_listener, &node_events, nd);
     pw_proxy_add_listener((struct pw_proxy*)node, &nd->proxy_listener, &proxy_node_events, nd);
     resync(self);
@@ -514,10 +501,10 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
     if ((str = spa_dict_lookup(props, PW_KEY_NODE_ID)) == NULL)
       return;
 
-    if ((nd = find_node_data(rd, atoi(str))) == NULL)
+    if ((nd = find_node_data(&self->nodes, atoi(str))) == NULL)
       return;
 
-    port = pw_registry_bind(rd->registry,
+    port = pw_registry_bind(self->registry,
                     id, type, PW_VERSION_PORT, sizeof(*pd));
     if (port == NULL)
       goto no_mem;
@@ -554,36 +541,29 @@ static GList *
 gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
 {
   GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
-  struct pw_loop *l = NULL;
-  struct pw_context *c = NULL;
-  struct core_data *data;
 
   GST_DEBUG_OBJECT (self, "starting probe");
 
-  if (!(l = pw_loop_new (NULL)))
+  self->core = gst_pipewire_core_get(-1);
-    return NULL;
+  if (self->core == NULL) {
-
+    GST_ERROR_OBJECT (self, "Failed to connect");
-  if (!(c = pw_context_new (l, NULL, sizeof(*data))))
-    return NULL;
-
-  data = pw_context_get_user_data(c);
-  data->self = self;
-  spa_list_init(&data->nodes);
-
-  spa_list_init(&self->pending);
-  self->core = pw_context_connect (c, NULL, 0);
-  if (self->core == NULL)
     goto failed;
+  }
 
   GST_DEBUG_OBJECT (self, "connected");
-  pw_core_add_listener(self->core, &data->core_listener, &core_events, data);
 
+  pw_thread_loop_lock (self->core->loop);
+
+  spa_list_init(&self->nodes);
+  spa_list_init(&self->pending);
   self->end = FALSE;
+  self->error = 0;
   self->list_only = TRUE;
   self->devices = NULL;
+  self->registry = pw_core_get_registry(self->core->core, PW_VERSION_REGISTRY, 0);
 
-  data->registry = pw_core_get_registry(self->core, PW_VERSION_REGISTRY, 0);
+  pw_core_add_listener(self->core->core, &self->core_listener, &core_events, self);
-  pw_registry_add_listener(data->registry, &data->registry_listener, &registry_events, data);
+  pw_registry_add_listener(self->registry, &self->registry_listener, &registry_events, self);
 
   resync(self);
 
@@ -592,20 +572,18 @@ gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
       break;
     if (self->end)
       break;
-    pw_loop_iterate (l, -1);
+    pw_thread_loop_wait (self->core->loop);
   }
 
   GST_DEBUG_OBJECT (self, "disconnect");
-  pw_proxy_destroy ((struct pw_proxy*)data->registry);
+
-  pw_core_disconnect (self->core);
+  g_clear_pointer ((struct pw_proxy**)&self->registry, pw_proxy_destroy);
-  self->core = NULL;
+  pw_thread_loop_unlock (self->core->loop);
-  pw_context_destroy (c);
+  g_clear_pointer (&self->core, gst_pipewire_core_release);
-  pw_loop_destroy (l);
 
   return self->devices;
 
 failed:
-  pw_loop_destroy (l);
   return NULL;
 }
 
@@ -613,46 +591,28 @@ static gboolean
 gst_pipewire_device_provider_start (GstDeviceProvider * provider)
 {
   GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
-  struct core_data *data;
 
   GST_DEBUG_OBJECT (self, "starting provider");
 
-  self->list_only = FALSE;
+  self->core = gst_pipewire_core_get(-1);
-  spa_list_init(&self->pending);
+  if (self->core == NULL) {
-
-  if (!(self->loop = pw_thread_loop_new ("pipewire-device-monitor", NULL))) {
-    GST_ERROR_OBJECT (self, "Could not create PipeWire mainloop");
-    goto failed_loop;
-  }
-
-  if (!(self->context = pw_context_new (pw_thread_loop_get_loop(self->loop), NULL, sizeof(*data)))) {
-    GST_ERROR_OBJECT (self, "Could not create PipeWire context");
-    goto failed_context;
-  }
-
-  if (pw_thread_loop_start (self->loop) < 0) {
-    GST_ERROR_OBJECT (self, "Could not start PipeWire mainloop");
-    goto failed_start;
-  }
-
-  pw_thread_loop_lock (self->loop);
-
-  if ((self->core = pw_context_connect (self->context, NULL, 0)) == NULL) {
     GST_ERROR_OBJECT (self, "Failed to connect");
-    goto failed_connect;
+    goto failed;
   }
 
   GST_DEBUG_OBJECT (self, "connected");
 
-  data = pw_context_get_user_data(self->context);
+  pw_thread_loop_lock (self->core->loop);
-  data->self = self;
-  spa_list_init(&data->nodes);
 
-  pw_core_add_listener(self->core, &data->core_listener, &core_events, data);
+  spa_list_init(&self->nodes);
+  spa_list_init(&self->pending);
+  self->end = FALSE;
+  self->error = 0;
+  self->list_only = FALSE;
+  self->registry = pw_core_get_registry(self->core->core, PW_VERSION_REGISTRY, 0);
 
-  self->registry = pw_core_get_registry(self->core, PW_VERSION_REGISTRY, 0);
+  pw_core_add_listener(self->core->core, &self->core_listener, &core_events, self);
-  data->registry = self->registry;
+  pw_registry_add_listener(self->registry, &self->registry_listener, &registry_events, self);
-  pw_registry_add_listener(self->registry, &data->registry_listener, &registry_events, data);
 
   resync(self);
 
@@ -661,24 +621,16 @@ gst_pipewire_device_provider_start (GstDeviceProvider * provider)
       break;
     if (self->end)
       break;
-    pw_thread_loop_wait (self->loop);
+    pw_thread_loop_wait (self->core->loop);
   }
 
   GST_DEBUG_OBJECT (self, "started");
 
-  pw_thread_loop_unlock (self->loop);
+  pw_thread_loop_unlock (self->core->loop);
 
   return TRUE;
 
-failed_connect:
+failed:
-  pw_thread_loop_unlock (self->loop);
-failed_start:
-  pw_context_destroy (self->context);
-  self->context = NULL;
-failed_context:
-  pw_thread_loop_destroy (self->loop);
-  self->loop = NULL;
-failed_loop:
   return TRUE;
 }
 
@@ -688,25 +640,9 @@ gst_pipewire_device_provider_stop (GstDeviceProvider * provider)
   GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
 
   GST_DEBUG_OBJECT (self, "stopping provider");
-  if (self->loop)
-    pw_thread_loop_stop (self->loop);
 
-  if (self->registry) {
+  g_clear_pointer ((struct pw_proxy**)&self->registry, pw_proxy_destroy);
-    pw_proxy_destroy ((struct pw_proxy*)self->registry);
+  g_clear_pointer (&self->core, gst_pipewire_core_release);
-    self->registry = NULL;
-  }
-  if (self->core) {
-    pw_core_disconnect (self->core);
-    self->core = NULL;
-  }
-  if (self->context) {
-    pw_context_destroy (self->context);
-    self->context = NULL;
-  }
-  if (self->loop) {
-    pw_thread_loop_destroy (self->loop);
-    self->loop = NULL;
-  }
 }
 
 static void

src/gst/gstpipewiredeviceprovider.h

@@ -28,10 +28,11 @@
 
 #include "config.h"
 
-#include <pipewire/pipewire.h>
-
 #include <gst/gst.h>
 
+#include <pipewire/pipewire.h>
+#include <gst/gstpipewirecore.h>
+
 G_BEGIN_DECLS
 
 typedef struct _GstPipeWireDevice GstPipeWireDevice;
@@ -82,16 +83,14 @@ struct _GstPipeWireDeviceProvider {
 
   gchar *client_name;
 
-  struct pw_thread_loop *loop;
+  GstPipeWireCore *core;
-
+  struct spa_hook core_listener;
-  struct pw_context *context;
+  struct pw_registry *registry;
-
+  struct spa_hook registry_listener;
-  struct pw_core *core;
+  struct spa_list nodes;
   struct spa_list pending;
   int seq;
 
-  struct pw_registry *registry;
-
   int error;
   gboolean end;
   gboolean list_only;