device: implement subscribe_params

Also add subscribe params for the device, like for Node and Port.

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

@@ -905,6 +905,38 @@ static int device_demarshal_param(void *object, const struct pw_protocol_native_
 			seq, id, index, next, param);
 }
 
+static int device_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_DEVICE_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 device_demarshal_subscribe_params(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	uint32_t csize, ctype, n_ids;
+	uint32_t *ids;
+
+	spa_pod_parser_init(&prs, msg->data, msg->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_notify(resource, struct pw_device_methods, subscribe_params, 0,
+			ids, n_ids);
+}
+
 static int device_marshal_enum_params(void *object, int seq,
 		uint32_t id, uint32_t index, uint32_t num, const struct spa_pod *filter)
 {
@@ -2067,6 +2099,7 @@ const struct pw_protocol_marshal pw_protocol_native_factory_marshal = {
 static const struct pw_device_methods pw_protocol_native_device_method_marshal = {
 	PW_VERSION_DEVICE_METHODS,
 	.add_listener = &device_method_marshal_add_listener,
+	.subscribe_params = &device_marshal_subscribe_params,
 	.enum_params = &device_marshal_enum_params,
 	.set_param = &device_marshal_set_param,
 };
@@ -2074,6 +2107,7 @@ static const struct pw_device_methods pw_protocol_native_device_method_marshal =
 static const struct pw_protocol_native_demarshal
 pw_protocol_native_device_method_demarshal[PW_DEVICE_METHOD_NUM] = {
 	[PW_DEVICE_METHOD_ADD_LISTENER] = { NULL, 0, },
+	[PW_DEVICE_METHOD_SUBSCRIBE_PARAMS] = { &device_demarshal_subscribe_params, 0, },
 	[PW_DEVICE_METHOD_ENUM_PARAMS] = { &device_demarshal_enum_params, 0, },
 	[PW_DEVICE_METHOD_SET_PARAM] = { &device_demarshal_set_param, PW_PERM_W, },
 };

src/pipewire/device.h

@@ -91,9 +91,10 @@ struct pw_device_events {
 
 
 #define PW_DEVICE_METHOD_ADD_LISTENER		0
-#define PW_DEVICE_METHOD_ENUM_PARAMS	1
+#define PW_DEVICE_METHOD_SUBSCRIBE_PARAMS	1
-#define PW_DEVICE_METHOD_SET_PARAM	2
+#define PW_DEVICE_METHOD_ENUM_PARAMS		2
-#define PW_DEVICE_METHOD_NUM		3
+#define PW_DEVICE_METHOD_SET_PARAM		3
+#define PW_DEVICE_METHOD_NUM			4
 
 /** Device methods */
 struct pw_device_methods {
@@ -104,6 +105,17 @@ struct pw_device_methods {
 			struct spa_hook *listener,
 			const struct pw_device_events *events,
 			void *data);
+	/**
+	 * 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 device parameters
 	 *
@@ -139,6 +151,7 @@ struct pw_device_methods {
 })
 
 #define pw_device_add_listener(c,...)		pw_device_method(c,add_listener,0,__VA_ARGS__)
+#define pw_device_subscribe_params(c,...)	pw_device_method(c,subscribe_params,0,__VA_ARGS__)
 #define pw_device_enum_params(c,...)		pw_device_method(c,enum_params,0,__VA_ARGS__)
 #define pw_device_set_param(c,...)		pw_device_method(c,set_param,0,__VA_ARGS__)
 

src/pipewire/impl-device.c

@@ -54,6 +54,9 @@ struct resource_data {
 	struct spa_hook resource_listener;
 	struct spa_hook object_listener;
 
+	uint32_t subscribe_ids[MAX_PARAMS];
+	uint32_t n_subscribe_ids;
+
 	/* for async replies */
 	int seq;
 	int end;
@@ -318,6 +321,25 @@ static int device_enum_params(void *object, int seq, uint32_t id, uint32_t start
 	return res;
 }
 
+static int device_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(NAME" %p: resource %p subscribe param %s",
+				data->device, resource,
+				spa_debug_type_find_name(spa_type_param, ids[i]));
+		device_enum_params(resource, 1, ids[i], 0, UINT32_MAX, NULL);
+	}
+	return 0;
+}
+
 static void result_device_done(void *data, int seq, int res, uint32_t type, const void *result)
 {
 	struct resource_data *d = data;
@@ -361,6 +383,7 @@ static int device_set_param(void *object, uint32_t id, uint32_t flags,
 
 static const struct pw_device_methods device_methods = {
 	PW_VERSION_DEVICE_METHODS,
+	.subscribe_params = device_subscribe_params,
 	.enum_params = device_enum_params,
 	.set_param = device_set_param
 };
@@ -532,19 +555,96 @@ static int update_properties(struct pw_impl_device *device, const struct spa_dic
 	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_impl_device *device = data;
+	struct pw_resource *resource;
+
+	spa_list_for_each(resource, &device->global->resource_list, link) {
+		if (!resource_is_subscribed(resource, id))
+			continue;
+
+		pw_log_debug(NAME" %p: resource %p notify param %d", device, resource, id);
+		pw_device_resource_param(resource, seq, id, index, next, param);
+	}
+	return 0;
+}
+
+static void emit_params(struct pw_impl_device *device, uint32_t *changed_ids, uint32_t n_changed_ids)
+{
+	uint32_t i;
+	int res;
+
+	if (device->global == NULL)
+		return;
+
+	pw_log_debug(NAME" %p: emit %d params", device, 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, &device->global->resource_list, link) {
+			if ((subscribed = resource_is_subscribed(resource, changed_ids[i])))
+				break;
+		}
+		if (!subscribed)
+			continue;
+
+		if ((res = pw_impl_device_for_each_param(device, 1, changed_ids[i], 0, UINT32_MAX,
+					NULL, notify_param, device)) < 0) {
+			pw_log_error(NAME" %p: error %d (%s)", device, res, spa_strerror(res));
+		}
+	}
+}
+
 static void device_info(void *data, const struct spa_device_info *info)
 {
 	struct pw_impl_device *device = data;
+	uint32_t changed_ids[MAX_PARAMS], n_changed_ids = 0;
+
 	if (info->change_mask & SPA_DEVICE_CHANGE_MASK_PROPS) {
 		update_properties(device, info->props);
 	}
 	if (info->change_mask & SPA_DEVICE_CHANGE_MASK_PARAMS) {
+		uint32_t i;
+
 		device->info.change_mask |= PW_DEVICE_CHANGE_MASK_PARAMS;
 		device->info.n_params = SPA_MIN(info->n_params, SPA_N_ELEMENTS(device->params));
-		memcpy(device->info.params, info->params,
+
-				device->info.n_params * sizeof(struct spa_param_info));
+		for (i = 0; i < device->info.n_params; i++) {
+			pw_log_debug(NAME" %p: param %d id:%d (%s) %08x:%08x", device, i,
+                                        info->params[i].id,
+                                        spa_debug_type_find_name(spa_type_param, info->params[i].id),
+                                        device->info.params[i].flags, info->params[i].flags);
+
+                        if (device->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;
+
+                        device->info.params[i] = info->params[i];
+                }
 	}
 	emit_info_changed(device);
+
+	if (n_changed_ids > 0)
+		emit_params(device, changed_ids, n_changed_ids);
 }
 
 static void device_add_object(struct pw_impl_device *device, uint32_t id,

src/pipewire/impl-node.c

@@ -1137,7 +1137,7 @@ static void node_info(void *data, const struct spa_node_info *info)
 	}
 	emit_info_changed(node);
 
-	if (info->change_mask & SPA_NODE_CHANGE_MASK_PARAMS)
+	if (n_changed_ids > 0)
 		emit_params(node, changed_ids, n_changed_ids);
 }
 

src/pipewire/impl-port.c

@@ -352,7 +352,7 @@ static void update_info(struct pw_impl_port *port, const struct spa_port_info *i
 		}
 	}
 
-	if (info->change_mask & SPA_NODE_CHANGE_MASK_PARAMS)
+	if (n_changed_ids > 0)
 		emit_params(port, changed_ids, n_changed_ids);
 }
 

src/tests/test-interfaces.c

@@ -164,6 +164,7 @@ static void test_device_abi(void)
 			struct spa_hook *listener,
 			const struct pw_device_events *events,
 			void *data);
+		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);
@@ -180,6 +181,7 @@ static void test_device_abi(void)
 
 	TEST_FUNC(m, methods, version);
 	TEST_FUNC(m, methods, add_listener);
+	TEST_FUNC(m, methods, subscribe_params);
 	TEST_FUNC(m, methods, enum_params);
 	TEST_FUNC(m, methods, set_param);
 	spa_assert(PW_VERSION_DEVICE_METHODS == 0);