module-pipewire: implement client node detection and simple-endpoint creation

modules/module-pipewire.c

@@ -22,6 +22,106 @@ gpointer simple_endpoint_factory (WpFactory * factory, GType type,
 gpointer simple_endpoint_link_factory (WpFactory * factory, GType type,
     GVariant * properties);
 
+struct module_data
+{
+  WpModule *module;
+
+  struct pw_core *core;
+
+  struct pw_remote *remote;
+  struct spa_hook remote_listener;
+
+  struct pw_registry_proxy *registry_proxy;
+  struct spa_hook registry_listener;
+};
+
+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;
+  const gchar *name;
+  const gchar *media_class;
+  struct pw_proxy *proxy;
+  GVariantBuilder b;
+  g_autoptr (GVariant) endpoint_props = NULL;
+  g_autoptr (WpCore) core = NULL;
+
+  /* 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);
+
+    g_variant_builder_init (&b, G_VARIANT_TYPE_VARDICT);
+    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);
+    wp_factory_make (core, "pipewire-simple-endpoint", WP_TYPE_ENDPOINT,
+        endpoint_props);
+  }
+}
+
+static const struct pw_registry_proxy_events registry_events = {
+  PW_VERSION_REGISTRY_PROXY_EVENTS,
+  .global = registry_global,
+};
+
+static void
+on_remote_state_changed (void *d, enum pw_remote_state old_state,
+    enum pw_remote_state new_state, const char *error)
+{
+  struct module_data *data = d;
+  struct pw_core_proxy *core_proxy;
+
+  g_debug ("remote state changed, old:%s new:%s",
+      pw_remote_state_as_string (old_state),
+      pw_remote_state_as_string (new_state));
+
+  switch (new_state) {
+  case PW_REMOTE_STATE_CONNECTED:
+    core_proxy = pw_remote_get_core_proxy (data->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,
+        &data->registry_listener, &registry_events, data);
+    break;
+
+  case PW_REMOTE_STATE_UNCONNECTED:
+    // TODO quit wireplumber
+    break;
+
+  case PW_REMOTE_STATE_ERROR:
+    // TODO quit wireplumber
+    break;
+
+  default:
+    break;
+  }
+}
+
+static const struct pw_remote_events remote_events = {
+  PW_VERSION_REMOTE_EVENTS,
+  .state_changed = on_remote_state_changed,
+};
+
 static gboolean
 connect_in_idle (struct pw_remote *remote)
 {
@@ -30,38 +130,41 @@ connect_in_idle (struct pw_remote *remote)
 }
 
 static void
-module_destroy (gpointer r)
+module_destroy (gpointer d)
 {
-  struct pw_remote *remote = r;
-  struct pw_core *core = pw_remote_get_core (remote);
+  struct module_data *data = d;
 
-  pw_remote_destroy (remote);
-  pw_core_destroy (core);
+  pw_remote_destroy (data->remote);
+  pw_core_destroy (data->core);
+  g_slice_free (struct module_data, data);
 }
 
 void
 wireplumber__module_init (WpModule * module, WpCore * core, GVariant * args)
 {
   GSource *source;
-  struct pw_core *pw_core;
-  struct pw_remote *pw_remote;
+  struct module_data *data;
 
   pw_init (NULL, NULL);
 
+  data = g_slice_new0 (struct module_data);
+  data->module = module;
+  wp_module_set_destroy_callback (module, module_destroy, data);
+
   source = wp_loop_source_new ();
   g_source_attach (source, NULL);
 
-  pw_core = pw_core_new (WP_LOOP_SOURCE (source)->loop, NULL, 0);
-  wp_core_register_global (core, WP_GLOBAL_PW_CORE, pw_core, NULL);
-
-  pw_remote = pw_remote_new (pw_core, NULL, 0);
-  wp_core_register_global (core, WP_GLOBAL_PW_REMOTE, pw_remote, NULL);
+  data->core = pw_core_new (WP_LOOP_SOURCE (source)->loop, NULL, 0);
+  wp_core_register_global (core, WP_GLOBAL_PW_CORE, data->core, NULL);
 
-  wp_module_set_destroy_callback (module, module_destroy, pw_remote);
+  data->remote = pw_remote_new (data->core, NULL, 0);
+  pw_remote_add_listener (data->remote, &data->remote_listener, &remote_events,
+      data->remote);
+  wp_core_register_global (core, WP_GLOBAL_PW_REMOTE, data->remote, NULL);
 
   wp_factory_new (core, "pipewire-simple-endpoint", simple_endpoint_factory);
   wp_factory_new (core, "pipewire-simple-endpoint-link",
       simple_endpoint_link_factory);
 
-  g_idle_add ((GSourceFunc) connect_in_idle, pw_remote);
+  g_idle_add ((GSourceFunc) connect_in_idle, data->remote);
 }

modules/module-pipewire/simple-endpoint.c

@@ -19,6 +19,13 @@
 struct _WpPipewireSimpleEndpoint
 {
   WpEndpoint parent;
+  struct pw_node_proxy *node;
+  struct spa_hook proxy_listener;
+};
+
+enum {
+  PROP_0,
+  PROP_NODE_PROXY,
 };
 
 G_DECLARE_FINAL_TYPE (WpPipewireSimpleEndpoint,
@@ -37,6 +44,51 @@ simple_endpoint_init (WpPipewireSimpleEndpoint * self)
   wp_endpoint_register_stream (WP_ENDPOINT (self), g_variant_builder_end (&b));
 }
 
+static void
+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);
+  }
+
+  G_OBJECT_CLASS (simple_endpoint_parent_class)->finalize (object);
+}
+
+static void
+simple_endpoint_set_property (GObject * object, guint property_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
+
+  switch (property_id) {
+  case PROP_NODE_PROXY:
+    self->node = g_value_get_pointer (value);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
+static void
+simple_endpoint_get_property (GObject * object, guint property_id,
+    GValue * value, GParamSpec * pspec)
+{
+  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (object);
+
+  switch (property_id) {
+  case PROP_NODE_PROXY:
+    g_value_set_pointer (value, self->node);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
+    break;
+  }
+}
+
 static gboolean
 simple_endpoint_prepare_link (WpEndpoint * self, guint32 stream_id,
     WpEndpointLink * link, GVariant ** properties, GError ** error)
@@ -48,22 +100,65 @@ simple_endpoint_prepare_link (WpEndpoint * self, guint32 stream_id,
 static void
 simple_endpoint_class_init (WpPipewireSimpleEndpointClass * klass)
 {
+  GObjectClass *object_class = (GObjectClass *) klass;
   WpEndpointClass *endpoint_class = (WpEndpointClass *) klass;
 
+  object_class->finalize = simple_endpoint_finalize;
+  object_class->set_property = simple_endpoint_set_property;
+  object_class->get_property = simple_endpoint_get_property;
+
   endpoint_class->prepare_link = simple_endpoint_prepare_link;
+
+  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_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 }
 
+static void
+node_proxy_destroy (void *data)
+{
+  WpPipewireSimpleEndpoint *self = WP_PIPEWIRE_SIMPLE_ENDPOINT (data);
+  self->node = NULL;
+  //TODO unregister the endpoint from the session manager
+}
+
+static const struct pw_proxy_events node_proxy_events = {
+  PW_VERSION_PROXY_EVENTS,
+  .destroy = node_proxy_destroy,
+};
+
 gpointer
 simple_endpoint_factory (WpFactory * factory, GType type,
     GVariant * properties)
 {
-  if (type != WP_TYPE_ENDPOINT)
+  WpPipewireSimpleEndpoint *ep;
+  guint64 proxy;
+  const gchar *name;
+  const gchar *media_class;
+
+  g_return_val_if_fail (type == WP_TYPE_ENDPOINT, NULL);
+  g_return_val_if_fail (properties != NULL, NULL);
+  g_return_val_if_fail (g_variant_is_of_type (properties,
+          G_VARIANT_TYPE_VARDICT), NULL);
+
+  if (!g_variant_lookup (properties, "name", "&s", &name))
+      return NULL;
+  if (!g_variant_lookup (properties, "media-class", "&s", &media_class))
+      return NULL;
+  if (!g_variant_lookup (properties, "node-proxy", "t", &proxy))
       return NULL;
 
-  /* TODO: retrieve pw_node* from @properties and keep it
-   * TODO: populate media_class and name on the endpoint
-   * TODO: potentially choose between subclasses of SimpleEndpoint
-   *  in order to add interfaces (volume, color balance, etc)
-   */
-  return g_object_new (simple_endpoint_get_type (), NULL);
+  ep = g_object_new (simple_endpoint_get_type (),
+      "name", name,
+      "media-class", media_class,
+      "node-proxy", (gpointer) proxy,
+      NULL);
+
+  pw_proxy_add_listener ((gpointer) proxy, &ep->proxy_listener,
+      &node_proxy_events, ep);
+
+  //TODO register the endpoint with the session manager
+
+  return ep;
 }