pulse-server: implement stream-restore EVENT

Add method to update the client routes. Add an event when the client
routes changed.

Listen for route_changed events on clients that do a subscribe on the
stream-restore extension. Emit an EVENT when the routes change.

This keeps the system notifications volumes in sync between
gnome-control center and pavucontrol and probably in other tools
as well.

Fixes #3805

src/modules/module-protocol-pulse/client.c

@@ -32,6 +32,7 @@
 PW_LOG_TOPIC_EXTERN(pulse_conn);
 
 #define client_emit_disconnect(c) spa_hook_list_call(&(c)->listener_list, struct client_events, disconnect, 0)
+#define client_emit_routes_changed(c) spa_hook_list_call(&(c)->listener_list, struct client_events, routes_changed, 0)
 
 struct client *client_new(struct server *server)
 {
@@ -169,6 +170,16 @@ void client_free(struct client *client)
 	free(client);
 }
 
+void client_update_routes(struct client *client, const char *key, const char *value)
+{
+	if (key == NULL)
+		pw_properties_clear(client->routes);
+	else
+		pw_properties_set(client->routes, key, value);
+
+	client_emit_routes_changed(client);
+}
+
 int client_queue_message(struct client *client, struct message *msg)
 {
 	struct impl *impl = client->impl;

src/modules/module-protocol-pulse/client.h

@@ -85,10 +85,12 @@ struct client {
 };
 
 struct client_events {
-#define VERSION_CLIENT_EVENTS	0
+#define VERSION_CLIENT_EVENTS	1
 	uint32_t version;
 
 	void (*disconnect) (void *data);
+
+	void (*routes_changed) (void *data);
 };
 
 struct client *client_new(struct server *server);
@@ -99,6 +101,8 @@ int client_queue_message(struct client *client, struct message *msg);
 int client_flush_messages(struct client *client);
 int client_queue_subscribe_event(struct client *client, uint32_t mask, uint32_t event, uint32_t id);
 
+void client_update_routes(struct client *client, const char *key, const char *value);
+
 static inline void client_unref(struct client *client)
 {
 	if (--client->ref == 0)

src/modules/module-protocol-pulse/modules/module-stream-restore.c

@@ -5,6 +5,7 @@
 #include <pipewire/pipewire.h>
 
 #include "../module.h"
+#include "../commands.h"
 
 /** \page page_pulse_module_stream_restore Stream restore extension
  *
@@ -32,6 +33,8 @@ PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
 
 struct module_stream_restore_data {
 	struct module *module;
+
+	struct spa_list subscribed;
 };
 
 static const struct spa_dict_item module_stream_restore_info[] = {
@@ -43,6 +46,15 @@ static const struct spa_dict_item module_stream_restore_info[] = {
 
 #define EXT_STREAM_RESTORE_VERSION	1
 
+enum {
+    SUBCOMMAND_TEST,
+    SUBCOMMAND_READ,
+    SUBCOMMAND_WRITE,
+    SUBCOMMAND_DELETE,
+    SUBCOMMAND_SUBSCRIBE,
+    SUBCOMMAND_EVENT
+};
+
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -312,18 +324,108 @@ static int do_extension_stream_restore_delete(struct module *module, struct clie
 	return reply_simple_ack(client, tag);
 }
 
+struct subscribe {
+	struct spa_list link;
+	struct module_stream_restore_data *data;
+
+	struct client *client;
+	struct spa_hook listener;
+};
+
+static void remove_subscribe(struct subscribe *s)
+{
+	spa_list_remove(&s->link);
+	spa_hook_remove(&s->listener);
+	free(s);
+}
+
+static void module_client_disconnect(void *data)
+{
+	struct subscribe *s = data;
+	remove_subscribe(s);
+}
+
+static void module_client_routes_changed(void *data)
+{
+	struct subscribe *s = data;
+	struct client *client = s->client;
+	struct message *msg = message_alloc(client->impl, -1, 0);
+
+	pw_log_info("[%s] EVENT index:%u name:%s", client->name,
+			s->data->module->index, s->data->module->info->name);
+
+	message_put(msg,
+		TAG_U32, COMMAND_EXTENSION,
+		TAG_U32, 0,
+		TAG_U32, s->data->module->index,
+		TAG_STRING, s->data->module->info->name,
+		TAG_U32, SUBCOMMAND_EVENT,
+		TAG_INVALID);
+
+	client_queue_message(client, msg);
+}
+
+static const struct client_events module_client_events = {
+	VERSION_CLIENT_EVENTS,
+	.disconnect = module_client_disconnect,
+	.routes_changed = module_client_routes_changed,
+};
+
+static struct subscribe *add_subscribe(struct module_stream_restore_data *data, struct client *c)
+{
+	struct subscribe *s;
+	s = calloc(1, sizeof(*s));
+	if (s == NULL)
+		return NULL;
+	s->data = data;
+	s->client = c;
+	client_add_listener(c, &s->listener, &module_client_events, s);
+	spa_list_append(&data->subscribed, &s->link);
+	return s;
+}
+
+static struct subscribe *find_subscribe(struct module_stream_restore_data *data, struct client *c)
+{
+	struct subscribe *s;
+	spa_list_for_each(s, &data->subscribed, link) {
+		if (s->client == c)
+			return s;
+	}
+	return NULL;
+}
+
 static int do_extension_stream_restore_subscribe(struct module *module, struct client *client, uint32_t command, uint32_t tag, struct message *m)
 {
+	struct module_stream_restore_data * const d = module->user_data;
+	int res;
+	bool enabled;
+	struct subscribe *s;
+
+	if ((res = message_get(m,
+			TAG_BOOLEAN, &enabled,
+			TAG_INVALID)) < 0)
+		return -EPROTO;
+
+	s = find_subscribe(d, client);
+	if (enabled) {
+		if (s == NULL)
+			s = add_subscribe(d, client);
+		if (s == NULL)
+			return -errno;
+	} else {
+		if (s != NULL)
+			remove_subscribe(s);
+	}
 	return reply_simple_ack(client, tag);
 }
 
 static const struct extension module_stream_restore_extension[] = {
-	{ "TEST", 0, do_extension_stream_restore_test, },
+	{ "TEST", SUBCOMMAND_TEST, do_extension_stream_restore_test, },
-	{ "READ", 1, do_extension_stream_restore_read, },
+	{ "READ", SUBCOMMAND_READ, do_extension_stream_restore_read, },
-	{ "WRITE", 2, do_extension_stream_restore_write, },
+	{ "WRITE", SUBCOMMAND_WRITE, do_extension_stream_restore_write, },
-	{ "DELETE", 3, do_extension_stream_restore_delete, },
+	{ "DELETE", SUBCOMMAND_DELETE, do_extension_stream_restore_delete, },
-	{ "SUBSCRIBE", 4, do_extension_stream_restore_subscribe, },
+	{ "SUBSCRIBE", SUBCOMMAND_SUBSCRIBE, do_extension_stream_restore_subscribe, },
-	{ "EVENT", 5, },
+	{ "EVENT", SUBCOMMAND_EVENT, },
 	{ NULL, },
 };
 
@@ -339,6 +441,17 @@ static int module_stream_restore_prepare(struct module * const module)
 
 static int module_stream_restore_load(struct module *module)
 {
+	struct module_stream_restore_data * const data = module->user_data;
+	spa_list_init(&data->subscribed);
+	return 0;
+}
+static int module_stream_restore_unload(struct module *module)
+{
+	struct module_stream_restore_data * const data = module->user_data;
+	struct subscribe *s;
+
+	spa_list_consume(s, &data->subscribed, link)
+		remove_subscribe(s);
 	return 0;
 }
 
@@ -347,6 +460,7 @@ DEFINE_MODULE_INFO(module_stream_restore) = {
 	.load_once = true,
 	.prepare = module_stream_restore_prepare,
 	.load = module_stream_restore_load,
+	.unload = module_stream_restore_unload,
 	.extension = module_stream_restore_extension,
 	.properties = &SPA_DICT_INIT_ARRAY(module_stream_restore_info),
 	.data_size = sizeof(struct module_stream_restore_data),

src/modules/module-protocol-pulse/pulse-server.c

@@ -989,12 +989,8 @@ static void manager_metadata(void *data, struct pw_manager_object *o,
 		if (changed)
 			send_default_change_subscribe_event(client, true, true);
 	}
-	if (subject == PW_ID_CORE && o == client->metadata_routes) {
+	if (subject == PW_ID_CORE && o == client->metadata_routes)
-		if (key == NULL)
+		client_update_routes(client, key, value);
-			pw_properties_clear(client->routes);
-		else
-			pw_properties_set(client->routes, key, value);
-	}
 }