utils: don't remove pending after first callback

Let the caller remove the pending result when finished.
Use non _sync verions to enum_params in node/port/device.
Set result.id in alsa and v4l2 correctly.

spa/include/spa/node/utils.h

@@ -64,8 +64,9 @@ static inline int spa_node_enum_params_sync(struct spa_node *node,
 	spa_pending_queue_add(queue, 0, &pending,
 			spa_result_func_node_params, &data);
 	res = spa_node_enum_params(node, 0, id, *index, 1, filter);
-	if (data.data.param == NULL) {
 	spa_pending_remove(&pending);
+
+	if (data.data.param == NULL) {
 		if (res > 0)
 			res = 0;
 	} else {
@@ -92,8 +93,9 @@ static inline int spa_node_port_enum_params_sync(struct spa_node *node,
 			spa_result_func_node_params, &data);
 	res = spa_node_port_enum_params(node, 0, direction, port_id,
 			id, *index, 1, filter);
-	if (data.data.param == NULL) {
 	spa_pending_remove(&pending);
+
+	if (data.data.param == NULL) {
 		if (res > 0)
 			res = 0;
 	} else {

spa/include/spa/utils/result.h

@@ -86,7 +86,6 @@ static inline int spa_pending_queue_complete(struct spa_pending_queue *queue,
 	spa_list_for_each_safe(p, t, &queue->pending, link) {
 		if (p->seq == seq) {
 			p->res = res;
-			spa_pending_remove(p);
 			p->func(p, result);
 		}
 	}

spa/plugins/alsa/alsa-utils.c

@@ -242,6 +242,7 @@ spa_alsa_enum_format(struct state *state, int seq, uint32_t start, uint32_t num,
 	if ((err = spa_alsa_open(state)) < 0)
 		return err;
 
+	result.id = SPA_PARAM_EnumFormat;
 	result.next = start;
 
       next:

spa/plugins/v4l2/v4l2-utils.c

@@ -539,6 +539,7 @@ spa_v4l2_enum_format(struct impl *this, int seq,
 	if ((res = spa_v4l2_open(dev, this->props.device)) < 0)
 		return res;
 
+	result.id = SPA_PARAM_EnumFormat;
 	result.next = start;
 
 	if (result.next == 0) {

src/pipewire/device.c

@@ -130,6 +130,22 @@ static const struct pw_resource_events resource_events = {
 	.destroy = device_unbind_func,
 };
 
+struct result_device_params_data {
+	void *data;
+	int (*callback) (void *data, int seq,
+			uint32_t id, uint32_t index, uint32_t next,
+			struct spa_pod *param);
+};
+
+static int result_device_params(struct spa_pending *pending, const void *result)
+{
+	struct result_device_params_data *d = pending->data;
+	const struct spa_result_device_params *r =
+		(const struct spa_result_device_params *)result;
+	d->callback(d->data, pending->seq, r->id, r->index, r->next, r->param);
+	return 0;
+}
+
 SPA_EXPORT
 int pw_device_for_each_param(struct pw_device *device,
 			     int seq, uint32_t param_id,
@@ -141,28 +157,21 @@ int pw_device_for_each_param(struct pw_device *device,
 			     void *data)
 {
 	struct impl *impl = SPA_CONTAINER_OF(device, struct impl, this);
-	int res = 0;
+	int res;
-	uint32_t idx, count;
+	struct result_device_params_data user_data = { data, callback };
-	uint8_t buf[4096];
+	struct spa_pending pending;
-	struct spa_pod_builder b = { 0 };
-	struct spa_pod *param;
 
 	if (max == 0)
 		max = UINT32_MAX;
 
-	for (count = 0; count < max; count++) {
+	spa_pending_queue_add(&impl->pending, seq, &pending,
-		spa_pod_builder_init(&b, buf, sizeof(buf));
+			result_device_params, &user_data);
 
-		idx = index;
+	res = spa_device_enum_params(device->implementation, seq,
-		if ((res = spa_device_enum_params_sync(device->implementation,
+			param_id, index, max, filter);
-						param_id, &index,
+
-						filter, &param, &b,
+	spa_pending_remove(&pending);
-						&impl->pending)) != 1)
-			break;
 
-		if ((res = callback(data, seq, param_id, idx, index, param)) != 0)
-			break;
-	}
 	return res;
 }
 

src/pipewire/node.c

@@ -249,7 +249,7 @@ static const struct pw_resource_events resource_events = {
 	.destroy = node_unbind_func,
 };
 
-static int reply_param(void *data, uint32_t seq, uint32_t id,
+static int reply_param(void *data, int seq, uint32_t id,
 		uint32_t index, uint32_t next, struct spa_pod *param)
 {
 	struct resource_data *d = data;
@@ -1066,22 +1066,36 @@ int pw_node_for_each_port(struct pw_node *node,
 	return 0;
 }
 
+struct result_node_params_data {
+	void *data;
+	int (*callback) (void *data, int seq,
+			uint32_t id, uint32_t index, uint32_t next,
+			struct spa_pod *param);
+};
+
+static int result_node_params(struct spa_pending *pending, const void *result)
+{
+	struct result_node_params_data *d = pending->data;
+	const struct spa_result_node_params *r =
+		(const struct spa_result_node_params *)result;
+	d->callback(d->data, pending->seq, r->id, r->index, r->next, r->param);
+	return 0;
+}
+
 SPA_EXPORT
 int pw_node_for_each_param(struct pw_node *node,
-			   uint32_t seq, uint32_t param_id,
+			   int seq, uint32_t param_id,
 			   uint32_t index, uint32_t max,
 			   const struct spa_pod *filter,
-			   int (*callback) (void *data, uint32_t seq,
+			   int (*callback) (void *data, int seq,
 					    uint32_t id, uint32_t index, uint32_t next,
 					    struct spa_pod *param),
 			   void *data)
 {
 	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
-	int res = 0;
+	int res;
-	uint32_t idx, count;
+	struct result_node_params_data user_data = { data, callback };
-	uint8_t buf[4096];
+	struct spa_pending pending;
-	struct spa_pod_builder b = { 0 };
-	struct spa_pod *param;
 
 	if (max == 0)
 		max = UINT32_MAX;
@@ -1090,19 +1104,13 @@ int pw_node_for_each_param(struct pw_node *node,
 			spa_debug_type_find_name(spa_type_param, param_id),
 			index, max);
 
-	for (count = 0; count < max; count++) {
+	spa_pending_queue_add(&impl->pending, seq, &pending,
-		spa_pod_builder_init(&b, buf, sizeof(buf));
+			result_node_params, &user_data);
+	res = spa_node_enum_params(node->node, seq,
+					param_id, index, max,
+					filter);
+	spa_pending_remove(&pending);
 
-		idx = index;
-		if ((res = spa_node_enum_params_sync(node->node,
-						param_id, &index,
-						filter, &param, &b,
-						&impl->pending)) != 1)
-			break;
-
-		if ((res = callback(data, seq, param_id, idx, index, param)) != 0)
-			break;
-	}
 	return res;
 }
 

src/pipewire/node.h

@@ -168,10 +168,10 @@ int pw_node_for_each_port(struct pw_node *node,
 			  void *data);
 
 int pw_node_for_each_param(struct pw_node *node,
-			   uint32_t seq, uint32_t param_id,
+			   int seq, uint32_t param_id,
 			   uint32_t index, uint32_t max,
 			   const struct spa_pod *filter,
-			   int (*callback) (void *data, uint32_t seq,
+			   int (*callback) (void *data, int seq,
 					    uint32_t id, uint32_t index, uint32_t next,
 					    struct spa_pod *param),
 			   void *data);

src/pipewire/port.c

@@ -467,7 +467,7 @@ static int reply_param(void *data, int seq, uint32_t id,
 {
 	struct resource_data *d = data;
 	struct pw_resource *resource = d->resource;
-	pw_log_debug("resource %p: reply param %d %d %d", resource, id, index, next);
+	pw_log_debug("resource %p: reply param %u %u %u", resource, id, index, next);
 	pw_port_resource_param(resource, seq, id, index, next, param);
 	return 0;
 }
@@ -743,6 +743,22 @@ void pw_port_destroy(struct pw_port *port)
 	free(port);
 }
 
+struct result_port_params_data {
+	void *data;
+	int (*callback) (void *data, int seq,
+			uint32_t id, uint32_t index, uint32_t next,
+			struct spa_pod *param);
+};
+
+static int result_port_params(struct spa_pending *pending, const void *result)
+{
+	struct result_port_params_data *d = pending->data;
+	const struct spa_result_node_params *r =
+		(const struct spa_result_node_params *)result;
+	d->callback(d->data, pending->seq, r->id, r->index, r->next, r->param);
+	return 0;
+}
+
 int pw_port_for_each_param(struct pw_port *port,
 			   int seq,
 			   uint32_t param_id,
@@ -753,12 +769,10 @@ int pw_port_for_each_param(struct pw_port *port,
 					    struct spa_pod *param),
 			   void *data)
 {
-	int res = 0;
+	int res;
-	uint8_t buf[4096];
-	struct spa_pod_builder b = { 0 };
-	uint32_t idx, count;
 	struct pw_node *node = port->node;
-	struct spa_pod *param;
+	struct result_port_params_data user_data = { data, callback };
+	struct spa_pending pending;
 
 	if (max == 0)
 		max = UINT32_MAX;
@@ -767,20 +781,15 @@ int pw_port_for_each_param(struct pw_port *port,
 			spa_debug_type_find_name(spa_type_param, param_id),
 			index, max);
 
-	for (count = 0; count < max; count++) {
+	spa_pending_queue_add(node->pending, seq, &pending,
-		spa_pod_builder_init(&b, buf, sizeof(buf));
+			result_port_params, &user_data);
-		idx = index;
+
-		if ((res = spa_node_port_enum_params_sync(node->node,
+	res = spa_node_port_enum_params(node->node, seq,
-						port->direction, port->port_id,
+					port->direction, port->port_id,
-						param_id, &index,
+					param_id, index, max,
-						filter, &param, &b,
+					filter);
-						node->pending)) != 1)
+	spa_pending_remove(&pending);
-			break;
 
-		pw_log_debug("port %p: have param %d %u %u", port, param_id, idx, index);
-		if ((res = callback(data, seq, param_id, idx, index, param)) != 0)
-			break;
-	}
 	pw_log_debug("port %p: res %d: (%s)", port, res, spa_strerror(res));
 	return res;
 }