port: add port subscribe_params

src/modules/module-protocol-native/protocol-native.c

@@ -1206,6 +1206,38 @@ static int port_demarshal_param(void *object, void *data, size_t size)
 			seq, id, index, next, param);
 }
 
+static int port_marshal_subscribe_params(void *object, uint32_t *ids, uint32_t n_ids)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_PORT_PROXY_METHOD_SUBSCRIBE_PARAMS, NULL);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Array(sizeof(uint32_t), SPA_TYPE_Id, n_ids, ids));
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int port_demarshal_subscribe_params(void *object, void *data, size_t size)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	uint32_t csize, ctype, n_ids;
+	uint32_t *ids;
+
+	spa_pod_parser_init(&prs, data, size);
+	if (spa_pod_parser_get_struct(&prs,
+				SPA_POD_Array(&csize, &ctype, &n_ids, &ids)) < 0)
+		return -EINVAL;
+
+	if (ctype != SPA_TYPE_Id)
+		return -EINVAL;
+
+	return pw_resource_do(resource, struct pw_port_proxy_methods, subscribe_params, 0,
+			ids, n_ids);
+}
+
 static int port_marshal_enum_params(void *object, int seq, uint32_t id,
 		uint32_t index, uint32_t num, const struct spa_pod *filter)
 {
@@ -1839,10 +1871,12 @@ static const struct pw_protocol_marshal pw_protocol_native_node_marshal = {
 
 static const struct pw_port_proxy_methods pw_protocol_native_port_method_marshal = {
 	PW_VERSION_PORT_PROXY_METHODS,
+	&port_marshal_subscribe_params,
 	&port_marshal_enum_params,
 };
 
 static const struct pw_protocol_native_demarshal pw_protocol_native_port_method_demarshal[] = {
+	{ &port_demarshal_subscribe_params, 0, },
 	{ &port_demarshal_enum_params, 0, },
 };
 

src/pipewire/interfaces.h

@@ -752,13 +752,25 @@ pw_node_proxy_add_listener(struct pw_node_proxy *node,
 
 #define PW_VERSION_PORT			0
 
-#define PW_PORT_PROXY_METHOD_ENUM_PARAMS	0
+#define PW_PORT_PROXY_METHOD_SUBSCRIBE_PARAMS	0
-#define PW_PORT_PROXY_METHOD_NUM		1
+#define PW_PORT_PROXY_METHOD_ENUM_PARAMS	1
+#define PW_PORT_PROXY_METHOD_NUM		2
 
 /** Port methods */
 struct pw_port_proxy_methods {
 #define PW_VERSION_PORT_PROXY_METHODS	0
 	uint32_t version;
+	/**
+	 * Subscribe to parameter changes
+	 *
+	 * Automatically emit param events for the given ids when
+	 * they are changed.
+	 *
+	 * \param ids an array of param ids
+	 * \param n_ids the number of ids in \a ids
+	 */
+	int (*subscribe_params) (void *object, uint32_t *ids, uint32_t n_ids);
+
 	/**
 	 * Enumerate port parameters
 	 *
@@ -777,6 +789,13 @@ struct pw_port_proxy_methods {
 };
 
 /** Port params */
+static inline int
+pw_port_proxy_subscribe_params(struct pw_port_proxy *port, uint32_t *ids, uint32_t n_ids)
+{
+	return pw_proxy_do((struct pw_proxy*)port, struct pw_port_proxy_methods, subscribe_params,
+			ids, n_ids);
+}
+
 static inline int
 pw_port_proxy_enum_params(struct pw_port_proxy *port, int seq, uint32_t id, uint32_t index,
 		uint32_t num, const struct spa_pod *filter)

src/pipewire/node.c

@@ -62,7 +62,7 @@ struct resource_data {
 	struct spa_hook resource_listener;
 	struct pw_node *node;
 	struct pw_resource *resource;
-	uint32_t subscribe_ids[64];
+	uint32_t subscribe_ids[MAX_PARAMS];
 	uint32_t n_subscribe_ids;
 };
 

src/pipewire/port.c

@@ -46,6 +46,8 @@ struct resource_data {
 	struct spa_hook resource_listener;
 	struct pw_port *port;
 	struct pw_resource *resource;
+	uint32_t subscribe_ids[MAX_PARAMS];
+	uint32_t n_subscribe_ids;
 };
 
 /** \endcond */
@@ -239,8 +241,67 @@ static int update_properties(struct pw_port *port, const struct spa_dict *dict)
 	return changed;
 }
 
+static int resource_is_subscribed(struct pw_resource *resource, uint32_t id)
+{
+	struct resource_data *data = pw_resource_get_user_data(resource);
+	uint32_t i;
+
+	for (i = 0; i < data->n_subscribe_ids; i++) {
+		if (data->subscribe_ids[i] == id)
+			return 1;
+	}
+	return 0;
+}
+
+static int notify_param(void *data, int seq, uint32_t id,
+		uint32_t index, uint32_t next, struct spa_pod *param)
+{
+	struct pw_port *port = data;
+	struct pw_resource *resource;
+
+	spa_list_for_each(resource, &port->global->resource_list, link) {
+		if (!resource_is_subscribed(resource, id))
+			continue;
+
+		pw_log_debug("resource %p: notify param %d", resource, id);
+		pw_port_resource_param(resource, seq, id, index, next, param);
+	}
+	return 0;
+}
+
+static void emit_params(struct pw_port *port, uint32_t *changed_ids, uint32_t n_changed_ids)
+{
+	uint32_t i;
+	int res;
+
+	if (port->global == NULL)
+		return;
+
+	pw_log_debug("port %p: emit %d params", port, n_changed_ids);
+
+	for (i = 0; i < n_changed_ids; i++) {
+		struct pw_resource *resource;
+		int subscribed = 0;
+
+		/* first check if anyone is subscribed */
+		spa_list_for_each(resource, &port->global->resource_list, link) {
+			if ((subscribed = resource_is_subscribed(resource, changed_ids[i])))
+				break;
+		}
+		if (!subscribed)
+			continue;
+
+		if ((res = pw_port_for_each_param(port, 1, changed_ids[i], 0, UINT32_MAX,
+					NULL, notify_param, port)) < 0) {
+			pw_log_error("port %p: error %d (%s)", port, res, spa_strerror(res));
+		}
+	}
+}
+
 static void update_info(struct pw_port *port, const struct spa_port_info *info)
 {
+	uint32_t changed_ids[MAX_PARAMS], n_changed_ids = 0;
+
 	if (info->change_mask & SPA_PORT_CHANGE_MASK_FLAGS) {
 		port->spa_flags = info->flags;
 	}
@@ -248,11 +309,24 @@ static void update_info(struct pw_port *port, const struct spa_port_info *info)
 		update_properties(port, info->props);
 	}
 	if (info->change_mask & SPA_PORT_CHANGE_MASK_PARAMS) {
+		uint32_t i;
+
 		port->info.change_mask |= PW_PORT_CHANGE_MASK_PARAMS;
 		port->info.n_params = SPA_MIN(info->n_params, SPA_N_ELEMENTS(port->params));
-		memcpy(port->info.params, info->params,
+
-				port->info.n_params * sizeof(struct spa_param_info));
+		for (i = 0; i < port->info.n_params; i++) {
+			if (port->info.params[i].flags == info->params[i].flags)
+				continue;
+
+			if (info->params[i].flags & SPA_PARAM_INFO_READ)
+				changed_ids[n_changed_ids++] = info->params[i].id;
+
+			port->info.params[i] = info->params[i];
 		}
+	}
+
+	if (info->change_mask & SPA_NODE_CHANGE_MASK_PARAMS)
+		emit_params(port, changed_ids, n_changed_ids);
 }
 
 SPA_EXPORT
@@ -481,8 +555,27 @@ static int port_enum_params(void *object, int seq, uint32_t id, uint32_t index,
 	return res;
 }
 
+static int port_subscribe_params(void *object, uint32_t *ids, uint32_t n_ids)
+{
+	struct pw_resource *resource = object;
+	struct resource_data *data = pw_resource_get_user_data(resource);
+	uint32_t i;
+
+	n_ids = SPA_MIN(n_ids, SPA_N_ELEMENTS(data->subscribe_ids));
+	data->n_subscribe_ids = n_ids;
+
+	for (i = 0; i < n_ids; i++) {
+		data->subscribe_ids[i] = ids[i];
+		pw_log_debug("resource %p: subscribe param %s", resource,
+			spa_debug_type_find_name(spa_type_param, ids[i]));
+		port_enum_params(resource, 1, ids[i], 0, UINT32_MAX, NULL);
+	}
+	return 0;
+}
+
 static const struct pw_port_proxy_methods port_methods = {
 	PW_VERSION_PORT_PROXY_METHODS,
+	.subscribe_params = port_subscribe_params,
 	.enum_params = port_enum_params
 };
 

src/tests/test-interfaces.c

@@ -171,6 +171,7 @@ static void test_node_abi(void)
 	struct pw_node_proxy_events e;
 	struct {
 		uint32_t version;
+		int (*subscribe_params) (void *object, uint32_t *ids, uint32_t n_ids);
 		int (*enum_params) (void *object, int seq, uint32_t id,
 			uint32_t start, uint32_t num, const struct spa_pod *filter);
 		int (*set_param) (void *object, uint32_t id, uint32_t flags,
@@ -186,6 +187,7 @@ static void test_node_abi(void)
 	} events = { PW_VERSION_NODE_PROXY_EVENTS, };
 
 	TEST_FUNC(m, methods, version);
+	TEST_FUNC(m, methods, subscribe_params);
 	TEST_FUNC(m, methods, enum_params);
 	TEST_FUNC(m, methods, set_param);
 	TEST_FUNC(m, methods, send_command);
@@ -205,6 +207,7 @@ static void test_port_abi(void)
 	struct pw_port_proxy_events e;
 	struct {
 		uint32_t version;
+		int (*subscribe_params) (void *object, uint32_t *ids, uint32_t n_ids);
 		int (*enum_params) (void *object, int seq, uint32_t id,
 			uint32_t start, uint32_t num, const struct spa_pod *filter);
 	} methods = { PW_VERSION_PORT_PROXY_METHODS, };