impl: emit error on resource when error

When the state of the object changes to ERROR, emit an error message
on the resources as well. Also pass the error code around for this
purpose.

4 files changed, 61 insertions, 36 deletions

diff --git a/src/pipewire/impl-link.c b/src/pipewire/impl-link.c
index 57a20e58..29dc0c91 100644
--- a/src/pipewire/impl-link.c
+++ b/src/pipewire/impl-link.c
@@ -82,7 +82,7 @@ static void info_changed(struct pw_impl_link *link)
 	link->info.change_mask = 0;
 }
 
-static void pw_impl_link_update_state(struct pw_impl_link *link, enum pw_link_state state, char *error)
+static void link_update_state(struct pw_impl_link *link, enum pw_link_state state, int res, char *error)
 {
 	enum pw_link_state old = link->info.state;
 
@@ -111,6 +111,12 @@ static void pw_impl_link_update_state(struct pw_impl_link *link, enum pw_link_st
 	link->info.change_mask |= PW_LINK_CHANGE_MASK_STATE;
 	info_changed(link);
 
+	if (state == PW_LINK_STATE_ERROR && link->global) {
+		struct pw_resource *resource;
+		spa_list_for_each(resource, &link->global->resource_list, link)
+			pw_resource_error(resource, res, error);
+	}
+
 	if (old != PW_LINK_STATE_PAUSED && state == PW_LINK_STATE_PAUSED) {
 		link->prepared = true;
 		link->preparing = false;
@@ -130,13 +136,15 @@ static void complete_ready(void *obj, void *data, int res, uint32_t id)
 	pw_log_debug(NAME" %p: obj:%p port %p complete READY: %s", this, obj, port, spa_strerror(res));
 
 	if (SPA_RESULT_IS_OK(res)) {
-		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY, NULL);
+		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY,
+				0, NULL);
 	} else {
-		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR, NULL);
+		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR,
+				res, spa_aprintf("port error going to READY: %s", spa_strerror(res)));
 	}
 	if (this->input->state >= PW_IMPL_PORT_STATE_READY &&
 	    this->output->state >= PW_IMPL_PORT_STATE_READY)
-		pw_impl_link_update_state(this, PW_LINK_STATE_ALLOCATING, NULL);
+		link_update_state(this, PW_LINK_STATE_ALLOCATING, 0, NULL);
 }
 
 static void complete_paused(void *obj, void *data, int res, uint32_t id)
@@ -148,14 +156,16 @@ static void complete_paused(void *obj, void *data, int res, uint32_t id)
 	pw_log_debug(NAME" %p: obj:%p port %p complete PAUSED: %s", this, obj, port, spa_strerror(res));
 
 	if (SPA_RESULT_IS_OK(res)) {
-		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_PAUSED, NULL);
+		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_PAUSED,
+				0, NULL);
 		mix->have_buffers = true;
 	} else {
-		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR, NULL);
+		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR,
+				res, spa_aprintf("port error going to PAUSED: %s", spa_strerror(res)));
 		mix->have_buffers = false;
 	}
 	if (this->rt.in_mix.have_buffers && this->rt.out_mix.have_buffers)
-		pw_impl_link_update_state(this, PW_LINK_STATE_PAUSED, NULL);
+		link_update_state(this, PW_LINK_STATE_PAUSED, 0, NULL);
 }
 
 static int do_negotiate(struct pw_impl_link *this)
@@ -186,7 +196,7 @@ static int do_negotiate(struct pw_impl_link *this)
 	if (in_state != PW_IMPL_PORT_STATE_CONFIGURE && out_state != PW_IMPL_PORT_STATE_CONFIGURE)
 		return 0;
 
-	pw_impl_link_update_state(this, PW_LINK_STATE_NEGOTIATING, NULL);
+	link_update_state(this, PW_LINK_STATE_NEGOTIATING, 0, NULL);
 
 	input = this->input;
 	output = this->output;
@@ -332,7 +342,7 @@ error:
 	pw_context_debug_port_params(context, output->node->node, output->direction,
 			output->port_id, SPA_PARAM_EnumFormat, res,
 			"output format (%s)", error);
-	pw_impl_link_update_state(this, PW_LINK_STATE_ERROR, error);
+	link_update_state(this, PW_LINK_STATE_ERROR, res, error);
 	free(format);
 	return res;
 }
@@ -397,7 +407,7 @@ static int do_allocation(struct pw_impl_link *this)
 
 	pw_log_debug(NAME" %p: out-state:%d in-state:%d", this, output->state, input->state);
 
-	pw_impl_link_update_state(this, PW_LINK_STATE_ALLOCATING, NULL);
+	link_update_state(this, PW_LINK_STATE_ALLOCATING, 0, NULL);
 
 	out_flags = output->spa_flags;
 	in_flags = input->spa_flags;
@@ -481,7 +491,7 @@ static int do_allocation(struct pw_impl_link *this)
 
 error:
 	pw_buffers_clear(&output->buffers);
-	pw_impl_link_update_state(this, PW_LINK_STATE_ERROR, error);
+	link_update_state(this, PW_LINK_STATE_ERROR, res, error);
 	return res;
 }
 
@@ -558,7 +568,7 @@ static void check_states(void *obj, void *user_data, int res, uint32_t id)
 	input = this->input;
 
 	if (output == NULL || input == NULL) {
-		pw_impl_link_update_state(this, PW_LINK_STATE_ERROR,
+		link_update_state(this, PW_LINK_STATE_ERROR, -EIO,
 				strdup(NAME" without input or output port"));
 		return;
 	}
@@ -577,14 +587,14 @@ static void check_states(void *obj, void *user_data, int res, uint32_t id)
 	pw_log_debug(NAME" %p: output state %d, input state %d", this, out_state, in_state);
 
 	if (out_state == PW_IMPL_PORT_STATE_ERROR || in_state == PW_IMPL_PORT_STATE_ERROR) {
-		pw_impl_link_update_state(this, PW_LINK_STATE_ERROR, strdup("ports are in error"));
+		link_update_state(this, PW_LINK_STATE_ERROR, -EIO, strdup("ports are in error"));
 		return;
 	}
 
 	if (PW_IMPL_PORT_IS_CONTROL(output) && PW_IMPL_PORT_IS_CONTROL(input)) {
-		pw_impl_port_update_state(output, PW_IMPL_PORT_STATE_PAUSED, NULL);
-		pw_impl_port_update_state(input, PW_IMPL_PORT_STATE_PAUSED, NULL);
-		pw_impl_link_update_state(this, PW_LINK_STATE_PAUSED, NULL);
+		pw_impl_port_update_state(output, PW_IMPL_PORT_STATE_PAUSED, 0, NULL);
+		pw_impl_port_update_state(input, PW_IMPL_PORT_STATE_PAUSED, 0, NULL);
+		link_update_state(this, PW_LINK_STATE_PAUSED, 0, NULL);
 	}
 
 	if ((res = do_negotiate(this)) != 0)
@@ -743,19 +753,19 @@ static void port_state_changed(struct pw_impl_link *this, struct pw_impl_port *p
 
 	switch (state) {
 	case PW_IMPL_PORT_STATE_ERROR:
-		pw_impl_link_update_state(this, PW_LINK_STATE_ERROR, error ? strdup(error) : NULL);
+		link_update_state(this, PW_LINK_STATE_ERROR, -EIO, error ? strdup(error) : NULL);
 		break;
 	case PW_IMPL_PORT_STATE_INIT:
 	case PW_IMPL_PORT_STATE_CONFIGURE:
 		if (this->prepared) {
 			this->prepared = false;
-			pw_impl_link_update_state(this, PW_LINK_STATE_INIT, NULL);
+			link_update_state(this, PW_LINK_STATE_INIT, 0, NULL);
 		}
 		break;
 	case PW_IMPL_PORT_STATE_READY:
 		if (this->prepared) {
 			this->prepared = false;
-			pw_impl_link_update_state(this, PW_LINK_STATE_NEGOTIATING, NULL);
+			link_update_state(this, PW_LINK_STATE_NEGOTIATING, 0, NULL);
 		}
 		break;
 	case PW_IMPL_PORT_STATE_PAUSED:
@@ -782,9 +792,9 @@ static void port_param_changed(struct pw_impl_link *this, uint32_t id,
 		return;
 	}
 	if (outport)
-		pw_impl_port_update_state(outport, target, NULL);
+		pw_impl_port_update_state(outport, target, 0, NULL);
 	if (inport)
-		pw_impl_port_update_state(inport, target, NULL);
+		pw_impl_port_update_state(inport, target, 0, NULL);
 
 	pw_impl_link_prepare(this);
 }
diff --git a/src/pipewire/impl-node.c b/src/pipewire/impl-node.c
index 5e789776..bf3eeafd 100644
--- a/src/pipewire/impl-node.c
+++ b/src/pipewire/impl-node.c
@@ -201,16 +201,17 @@ static int start_node(struct pw_impl_node *this)
 
 static void emit_info_changed(struct pw_impl_node *node)
 {
-	struct pw_resource *resource;
 
 	if (node->info.change_mask == 0)
 		return;
 
 	pw_impl_node_emit_info_changed(node, &node->info);
 
-	if (node->global)
+	if (node->global) {
+		struct pw_resource *resource;
 		spa_list_for_each(resource, &node->global->resource_list, link)
 			pw_node_resource_info(resource, &node->info);
+	}
 
 	node->info.change_mask = 0;
 }
@@ -286,7 +287,7 @@ do_node_add(struct spa_loop *loop,
 	return 0;
 }
 
-static void node_update_state(struct pw_impl_node *node, enum pw_node_state state, char *error)
+static void node_update_state(struct pw_impl_node *node, enum pw_node_state state, int res, char *error)
 {
 	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
 	enum pw_node_state old = node->info.state;
@@ -322,6 +323,12 @@ static void node_update_state(struct pw_impl_node *node, enum pw_node_state stat
 
 	node->info.change_mask |= PW_NODE_CHANGE_MASK_STATE;
 	emit_info_changed(node);
+
+	if (state == PW_NODE_STATE_ERROR && node->global) {
+		struct pw_resource *resource;
+		spa_list_for_each(resource, &node->global->resource_list, link)
+			pw_resource_error(resource, res, error);
+	}
 }
 
 static int suspend_node(struct pw_impl_node *this)
@@ -361,7 +368,7 @@ static int suspend_node(struct pw_impl_node *this)
 	if (res < 0 && res != -EIO)
 		pw_log_warn(NAME" %p: suspend node error %s", this, spa_strerror(res));
 
-	node_update_state(this, PW_NODE_STATE_SUSPENDED, NULL);
+	node_update_state(this, PW_NODE_STATE_SUSPENDED, 0, NULL);
 
 	return res;
 }
@@ -685,7 +692,7 @@ int pw_impl_node_initialized(struct pw_impl_node *this)
 {
 	pw_log_debug(NAME" %p initialized", this);
 	pw_impl_node_emit_initialized(this);
-	node_update_state(this, PW_NODE_STATE_SUSPENDED, NULL);
+	node_update_state(this, PW_NODE_STATE_SUSPENDED, 0, NULL);
 	return 0;
 }
 
@@ -1321,7 +1328,8 @@ static void node_event(void *data, const struct spa_event *event)
 	switch (SPA_NODE_EVENT_ID(event)) {
 	case SPA_NODE_EVENT_Error:
 		impl->last_error = -EFAULT;
-		node_update_state(node, PW_NODE_STATE_ERROR, strdup("error"));
+		node_update_state(node, PW_NODE_STATE_ERROR,
+				-EFAULT, strdup("Received error event"));
 		break;
 	default:
 		pw_log_debug("unhandled event");
@@ -1861,12 +1869,13 @@ static void on_state_complete(void *obj, void *data, int res, uint32_t seq)
 	if (SPA_RESULT_IS_ERROR(res)) {
 		if (node->info.state == PW_NODE_STATE_SUSPENDED) {
 			state = PW_NODE_STATE_SUSPENDED;
+			res = 0;
 		} else {
 			error = spa_aprintf("error changing node state: %s", spa_strerror(res));
 			state = PW_NODE_STATE_ERROR;
 		}
 	}
-	node_update_state(node, state, error);
+	node_update_state(node, state, res, error);
 }
 
 static void node_activate(struct pw_impl_node *this)
diff --git a/src/pipewire/impl-port.c b/src/pipewire/impl-port.c
index a2e27b25..bc8e71ca 100644
--- a/src/pipewire/impl-port.c
+++ b/src/pipewire/impl-port.c
@@ -94,7 +94,7 @@ static const char *port_state_as_string(enum pw_impl_port_state state)
 	return "invalid-state";
 }
 
-void pw_impl_port_update_state(struct pw_impl_port *port, enum pw_impl_port_state state, char *error)
+void pw_impl_port_update_state(struct pw_impl_port *port, enum pw_impl_port_state state, int res, char *error)
 {
 	enum pw_impl_port_state old = port->state;
 
@@ -111,6 +111,12 @@ void pw_impl_port_update_state(struct pw_impl_port *port, enum pw_impl_port_stat
 		port_state_as_string(old), port_state_as_string(state), error);
 
 	pw_impl_port_emit_state_changed(port, old, state, error);
+
+	if (state == PW_IMPL_PORT_STATE_ERROR && port->global) {
+		struct pw_resource *resource;
+		spa_list_for_each(resource, &port->global->resource_list, link)
+			pw_resource_error(resource, res, error);
+	}
 }
 
 static int tee_process(void *object)
@@ -908,7 +914,7 @@ int pw_impl_port_add(struct pw_impl_port *port, struct pw_impl_node *node)
 	pw_loop_invoke(node->data_loop, do_add_port, SPA_ID_INVALID, NULL, 0, false, port);
 
 	if (port->state <= PW_IMPL_PORT_STATE_INIT)
-		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_CONFIGURE, NULL);
+		pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_CONFIGURE, 0, NULL);
 
 	pw_impl_node_emit_port_added(node, port);
 	emit_info_changed(port);
@@ -1183,10 +1189,10 @@ int pw_impl_port_set_param(struct pw_impl_port *port, uint32_t id, uint32_t flag
 		pw_buffers_clear(&port->mix_buffers);
 
 		if (param == NULL || res < 0) {
-			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_CONFIGURE, NULL);
+			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_CONFIGURE, 0, NULL);
 		}
 		else if (!SPA_RESULT_IS_ASYNC(res)) {
-			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY, NULL);
+			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY, 0, NULL);
 		}
 	}
 	return res;
@@ -1261,7 +1267,7 @@ int pw_impl_port_use_buffers(struct pw_impl_port *port, struct pw_impl_port_mix
 
 	if (n_buffers == 0) {
 		if (port->n_mix == 1)
-			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY, NULL);
+			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_READY, 0, NULL);
 	}
 
 	/* first negotiate with the node, this makes it possible to let the
@@ -1272,10 +1278,10 @@ int pw_impl_port_use_buffers(struct pw_impl_port *port, struct pw_impl_port_mix
 		if (res < 0) {
 			pw_log_error(NAME" %p: negotiate buffers on node: %d (%s)",
 				port, res, spa_strerror(res));
-			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR,
+			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_ERROR, res,
 					strdup("can't negotiate buffers on port"));
 		} else if (n_buffers > 0 && !SPA_RESULT_IS_ASYNC(res)) {
-			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_PAUSED, NULL);
+			pw_impl_port_update_state(port, PW_IMPL_PORT_STATE_PAUSED, 0, NULL);
 		}
 	}
 
diff --git a/src/pipewire/private.h b/src/pipewire/private.h
index 366a2202..aeb570cf 100644
--- a/src/pipewire/private.h
+++ b/src/pipewire/private.h
@@ -1013,7 +1013,7 @@ int pw_impl_port_set_mix(struct pw_impl_port *port, struct spa_node *node, uint3
 int pw_impl_port_init_mix(struct pw_impl_port *port, struct pw_impl_port_mix *mix);
 int pw_impl_port_release_mix(struct pw_impl_port *port, struct pw_impl_port_mix *mix);
 
-void pw_impl_port_update_state(struct pw_impl_port *port, enum pw_impl_port_state state, char *error);
+void pw_impl_port_update_state(struct pw_impl_port *port, enum pw_impl_port_state state, int res, char *error);
 
 /** Unlink a port \memberof pw_impl_port */
 void pw_impl_port_unlink(struct pw_impl_port *port);