Add enabled state as wellenable-2

The idea is to have the client use the active/inactive flag to
start negotiation and the enabled+active flag to start or stop the
node. The session manager would then enable the node only after
it was configured to make sure the server doesn't start the node
before it's configured.

12 files changed, 123 insertions, 33 deletions

diff --git a/src/examples/media-session/policy-node.c b/src/examples/media-session/policy-node.c
index dcf96970..fa7fcb0c 100644
--- a/src/examples/media-session/policy-node.c
+++ b/src/examples/media-session/policy-node.c
@@ -139,7 +139,16 @@ static bool find_format(struct node *node)
 	return have_format;
 }
 
-static int configure_node(struct node *node, struct spa_audio_info *info)
+static int enable_node(struct node *node, bool enable)
+{
+	node->active = node->configured = enable;
+	pw_node_send_command((struct pw_node*)node->obj->obj.proxy,
+			&SPA_NODE_COMMAND_INIT(enable ?
+				SPA_NODE_COMMAND_Enable : SPA_NODE_COMMAND_Disable));
+	return 0;
+}
+
+static int configure_node(struct node *node, struct spa_audio_info *info, bool enable)
 {
 	struct impl *impl = node->impl;
 	char buf[1024];
@@ -171,7 +180,7 @@ static int configure_node(struct node *node, struct spa_audio_info *info)
 	pw_node_set_param((struct pw_node*)node->obj->obj.proxy,
 			SPA_PARAM_PortConfig, 0, param);
 
-	node->configured = true;
+	enable_node(node, true);
 
 	return 0;
 }
@@ -229,7 +238,7 @@ handle_node(struct impl *impl, struct sm_object *object)
 	spa_list_append(&impl->node_list, &node->link);
 
 	if (role && !strcmp(role, "DSP"))
-		node->active = node->configured = true;
+		enable_node(node, true);
 
 	if (strstr(media_class, "Stream/") == media_class) {
 		media_class += strlen("Stream/");
@@ -252,10 +261,10 @@ handle_node(struct impl *impl, struct sm_object *object)
 				else
 					node->plugged = SPA_TIMESPEC_TO_NSEC(&impl->now);
 			}
-			node->active = node->configured = true;
+			enable_node(node, true);
 		}
 		else if (strstr(media_class, "Unknown") == media_class) {
-			node->active = node->configured = true;
+			enable_node(node, true);
 		}
 
 		node->direction = direction;
@@ -272,7 +281,7 @@ handle_node(struct impl *impl, struct sm_object *object)
 		else if (strstr(media_class, "Video/") == media_class) {
 			media_class += strlen("Video/");
 			media = "Video";
-			node->active = node->configured = true;
+			enable_node(node, true);
 		}
 		else
 			return 0;
@@ -447,7 +456,7 @@ static int link_nodes(struct node *node, struct node *peer)
 	pw_log_debug(NAME " %p: link nodes %d %d", impl, node->id, peer->id);
 
 	if (node->dont_remix)
-		configure_node(node, NULL);
+		configure_node(node, NULL, false);
 	else {
 #if 0
 		bool configured = node->configured;
@@ -457,7 +466,7 @@ static int link_nodes(struct node *node, struct node *peer)
 					&SPA_NODE_COMMAND_INIT(SPA_NODE_COMMAND_Suspend));
 		}
 #endif
-		configure_node(node, &peer->format);
+		configure_node(node, &peer->format, false);
 #if 0
 		if (configured) {
 			pw_node_send_command((struct pw_node*)node->obj->obj.proxy,
@@ -483,6 +492,8 @@ static int link_nodes(struct node *node, struct node *peer)
 
 	pw_properties_free(props);
 
+	enable_node(node, true);
+
 	return 0;
 }
 
@@ -530,7 +541,7 @@ static int rescan_node(struct impl *impl, struct node *n)
 	}
 
 	if (n->type == NODE_TYPE_DEVICE) {
-		configure_node(n, NULL);
+		configure_node(n, NULL, true);
 		return 0;
 	}
 
@@ -556,7 +567,7 @@ static int rescan_node(struct impl *impl, struct node *n)
         str = spa_dict_lookup(props, PW_KEY_NODE_AUTOCONNECT);
         if (str == NULL || !pw_properties_parse_bool(str)) {
 		pw_log_debug(NAME" %p: node %d does not need autoconnect", impl, n->id);
-		configure_node(n, NULL);
+		configure_node(n, NULL, true);
 		return 0;
 	}
 
diff --git a/src/gst/gstpipewiresink.c b/src/gst/gstpipewiresink.c
index 748992dd..3512dc8f 100644
--- a/src/gst/gstpipewiresink.c
+++ b/src/gst/gstpipewiresink.c
@@ -522,6 +522,7 @@ gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
   if (state == PW_STREAM_STATE_UNCONNECTED) {
     enum pw_stream_flags flags = 0;
 
+    flags = PW_STREAM_FLAG_INACTIVE;
     if (pwsink->mode != GST_PIPEWIRE_SINK_MODE_PROVIDE)
       flags |= PW_STREAM_FLAG_AUTOCONNECT;
     else
diff --git a/src/gst/gstpipewiresrc.c b/src/gst/gstpipewiresrc.c
index fdeb9c40..733dce7c 100644
--- a/src/gst/gstpipewiresrc.c
+++ b/src/gst/gstpipewiresrc.c
@@ -615,6 +615,7 @@ wait_started (GstPipeWireSrc *this)
     pw_thread_loop_wait (this->core->loop);
   }
   GST_DEBUG_OBJECT (this, "got started signal");
+  pw_stream_set_active(this->stream, true);
   pw_thread_loop_unlock (this->core->loop);
 
   return state;
@@ -686,6 +687,7 @@ gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
   pw_stream_connect (pwsrc->stream,
                      PW_DIRECTION_INPUT,
                      pwsrc->path ? (uint32_t)atoi(pwsrc->path) : PW_ID_ANY,
+                     PW_STREAM_FLAG_INACTIVE |
                      PW_STREAM_FLAG_AUTOCONNECT,
                      (const struct spa_pod **)possible->pdata,
                      possible->len);
diff --git a/src/modules/module-adapter.c b/src/modules/module-adapter.c
index 34d6a899..6f4ad09d 100644
--- a/src/modules/module-adapter.c
+++ b/src/modules/module-adapter.c
@@ -155,6 +155,7 @@ static void *create_object(void *_data,
 	const char *str, *factory_name;
 	int res;
 	struct node_data *nd;
+	bool activate;
 
 	if (properties == NULL)
 		goto error_properties;
@@ -169,6 +170,9 @@ static void *create_object(void *_data,
 				pw_impl_client_get_info(client)->id);
 	}
 
+	str = pw_properties_get(properties, "node.active");
+	activate = str ? pw_properties_parse_bool(str) : true;
+
 	follower = NULL;
 	str = pw_properties_get(properties, "adapt.follower.node");
 	if (str != NULL) {
@@ -184,7 +188,7 @@ static void *create_object(void *_data,
 
 		follower = pw_spa_node_load(d->context,
 					factory_name,
-					PW_SPA_NODE_FLAG_ACTIVATE |
+					(activate ? PW_SPA_NODE_FLAG_ACTIVATE : 0) |
 					PW_SPA_NODE_FLAG_NO_REGISTER,
 					pw_properties_copy(properties), 0);
 		if (follower == NULL)
@@ -216,7 +220,7 @@ static void *create_object(void *_data,
 
 	pw_impl_node_register(adapter, NULL);
 
-	pw_impl_node_set_active(adapter, true);
+	pw_impl_node_set_active(adapter, activate);
 
 	return adapter;
 
diff --git a/src/modules/module-adapter/adapter.c b/src/modules/module-adapter/adapter.c
index 5fe99684..5d46b6b7 100644
--- a/src/modules/module-adapter/adapter.c
+++ b/src/modules/module-adapter/adapter.c
@@ -258,7 +258,7 @@ struct pw_impl_node *pw_adapter_new(struct pw_context *context,
 
 	node = pw_spa_node_load(context,
 				factory_name,
-				PW_SPA_NODE_FLAG_ACTIVATE | PW_SPA_NODE_FLAG_NO_REGISTER,
+				PW_SPA_NODE_FLAG_NO_REGISTER,
 				pw_properties_copy(props),
 				sizeof(struct node) + user_data_size);
         if (node == NULL) {
diff --git a/src/modules/module-client-node/remote-node.c b/src/modules/module-client-node/remote-node.c
index 1ce71c53..fdb8c1a9 100644
--- a/src/modules/module-client-node/remote-node.c
+++ b/src/modules/module-client-node/remote-node.c
@@ -504,6 +504,16 @@ static int client_node_command(void *object, const struct spa_command *command)
 		}
 		break;
 
+	case SPA_NODE_COMMAND_Enable:
+		pw_log_debug("node %p: enable", proxy);
+		res = pw_impl_node_set_enabled(data->node, true);
+		break;
+
+	case SPA_NODE_COMMAND_Disable:
+		pw_log_debug("node %p: disable", proxy);
+		res = pw_impl_node_set_enabled(data->node, false);
+		break;
+
 	default:
 		pw_log_warn("unhandled node command %d", SPA_NODE_COMMAND_ID(command));
 		res = -ENOTSUP;
diff --git a/src/pipewire/context.c b/src/pipewire/context.c
index a1c6412e..ece4a7f5 100644
--- a/src/pipewire/context.c
+++ b/src/pipewire/context.c
@@ -790,7 +790,7 @@ error:
 static int ensure_state(struct pw_impl_node *node, bool running)
 {
 	enum pw_node_state state = node->info.state;
-	if (node->active && running)
+	if (node->active && node->enabled && running)
 		state = PW_NODE_STATE_RUNNING;
 	else if (state > PW_NODE_STATE_IDLE)
 		state = PW_NODE_STATE_IDLE;
diff --git a/src/pipewire/impl-link.c b/src/pipewire/impl-link.c
index 7bdfa382..77a31cb2 100644
--- a/src/pipewire/impl-link.c
+++ b/src/pipewire/impl-link.c
@@ -856,26 +856,31 @@ static void output_node_result(void *data, int seq, int res, uint32_t type, cons
 	node_result(impl, port, seq, res, type, result);
 }
 
-static void node_active_changed(void *data, bool active)
+static void check_prepare(struct pw_impl_link *this)
 {
-	struct impl *impl = data;
-	struct pw_impl_link *this = &impl->this;
-	pw_log_debug(NAME" %p: input active:%d output active:%d", impl,
-			impl->inode->active, impl->onode->active);
-	if (impl->inode->active && impl->onode->active)
+	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
+	pw_log_debug(NAME" %p: input enabled:%d output enabled:%d", impl,
+			impl->inode->enabled, impl->onode->enabled);
+	if (impl->inode->enabled && impl->onode->enabled)
 		pw_impl_link_prepare(this);
 }
 
+static void node_enabled_changed(void *data, bool active)
+{
+	struct impl *impl = data;
+	check_prepare(&impl->this);
+}
+
 static const struct pw_impl_node_events input_node_events = {
 	PW_VERSION_IMPL_NODE_EVENTS,
 	.result = input_node_result,
-	.active_changed = node_active_changed,
+	.enabled_changed = node_enabled_changed,
 };
 
 static const struct pw_impl_node_events output_node_events = {
 	PW_VERSION_IMPL_NODE_EVENTS,
 	.result = output_node_result,
-	.active_changed = node_active_changed,
+	.enabled_changed = node_enabled_changed,
 };
 
 static bool pw_impl_node_can_reach(struct pw_impl_node *output, struct pw_impl_node *input)
@@ -1218,10 +1223,7 @@ int pw_impl_link_register(struct pw_impl_link *link,
 	pw_global_add_listener(link->global, &link->global_listener, &global_events, link);
 	pw_global_register(link->global);
 
-	pw_log_debug(NAME" %p: output_active:%d input_active:%d", link,
-			input_node->active, output_node->active);
-	if (input_node->active && output_node->active)
-		pw_impl_link_prepare(link);
+	check_prepare(link);
 
 	return 0;
 
diff --git a/src/pipewire/impl-node.c b/src/pipewire/impl-node.c
index d11da9c7..e743a39b 100644
--- a/src/pipewire/impl-node.c
+++ b/src/pipewire/impl-node.c
@@ -479,10 +479,20 @@ static int node_send_command(void *object, const struct spa_command *command)
 	struct resource_data *data = object;
 	struct pw_impl_node *node = data->node;
 
+	pw_log_debug(NAME" %p: resource command %d", node, SPA_NODE_COMMAND_ID(command));
+
 	switch (SPA_NODE_COMMAND_ID(command)) {
 	case SPA_NODE_COMMAND_Suspend:
 		suspend_node(node);
 		break;
+	case SPA_NODE_COMMAND_Enable:
+		spa_node_send_command(node->node, command);
+		pw_impl_node_set_enabled(node, true);
+		break;
+	case SPA_NODE_COMMAND_Disable:
+		pw_impl_node_set_enabled(node, false);
+		spa_node_send_command(node->node, command);
+		break;
 	default:
 		spa_node_send_command(node->node, command);
 		break;
@@ -1294,7 +1304,9 @@ static void node_result(void *data, int seq, int res, uint32_t type, const void
 	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
 
 	pw_log_trace(NAME" %p: result seq:%d res:%d type:%u", node, seq, res, type);
-	impl->last_error = res;
+
+	if (res < 0)
+		impl->last_error = res;
 
 	if (SPA_RESULT_IS_ASYNC(seq))
 	        pw_work_queue_complete(impl->work, &impl->this, SPA_RESULT_ASYNC_SEQ(seq), res);
@@ -1840,13 +1852,22 @@ error:
 static void on_state_complete(void *obj, void *data, int res, uint32_t seq)
 {
 	struct pw_impl_node *node = obj;
+	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
 	enum pw_node_state state = SPA_PTR_TO_INT(data);
 	char *error = NULL;
 
-	pw_log_debug(NAME" %p: state complete %d", node, res);
+	pw_log_debug(NAME" %p: state complete %d seq:%d", node, res, seq);
+	if (impl->last_error < 0) {
+		res = impl->last_error;
+		impl->last_error = 0;
+	}
 	if (SPA_RESULT_IS_ERROR(res)) {
-		error = spa_aprintf("error changing node state: %s", spa_strerror(res));
-		state = PW_NODE_STATE_ERROR;
+		if (node->info.state == PW_NODE_STATE_SUSPENDED) {
+			state = PW_NODE_STATE_SUSPENDED;
+		} else {
+			error = spa_aprintf("error changing node state: %s", spa_strerror(res));
+			state = PW_NODE_STATE_ERROR;
+		}
 	}
 	node_update_state(node, state, error);
 }
@@ -1953,3 +1974,27 @@ bool pw_impl_node_is_active(struct pw_impl_node *node)
 {
 	return node->active;
 }
+
+SPA_EXPORT
+int pw_impl_node_set_enabled(struct pw_impl_node *node, bool enabled)
+{
+	bool old = node->enabled;
+
+	if (old != enabled) {
+		pw_log_debug(NAME" %p: %s", node, enabled ? "enable" : "disable");
+
+		node->enabled = enabled;
+		pw_impl_node_emit_enabled_changed(node, enabled);
+
+		if (node->registered)
+			pw_context_recalc_graph(node->context,
+					enabled ? "node enable" : "node disable");
+	}
+	return 0;
+}
+
+SPA_EXPORT
+bool pw_impl_node_is_enabled(struct pw_impl_node *node)
+{
+	return node->enabled;
+}
diff --git a/src/pipewire/impl-node.h b/src/pipewire/impl-node.h
index 21fce957..5e2f89e1 100644
--- a/src/pipewire/impl-node.h
+++ b/src/pipewire/impl-node.h
@@ -51,7 +51,7 @@ struct pw_impl_port;
 
 /** Node events, listen to them with \ref pw_impl_node_add_listener */
 struct pw_impl_node_events {
-#define PW_VERSION_IMPL_NODE_EVENTS	0
+#define PW_VERSION_IMPL_NODE_EVENTS	1
 	uint32_t version;
 
 	/** the node is destroyed */
@@ -95,6 +95,9 @@ struct pw_impl_node_events {
 	void (*peer_added) (void *data, struct pw_impl_node *peer);
 	/** a peer was removed */
 	void (*peer_removed) (void *data, struct pw_impl_node *peer);
+
+	/** the node enabled state changed  Since v1*/
+	void (*enabled_changed) (void *data, bool enabled);
 };
 
 /** Create a new node \memberof pw_impl_node */
@@ -175,6 +178,12 @@ int pw_impl_node_set_active(struct pw_impl_node *node, bool active);
 /** Check if a node is active */
 bool pw_impl_node_is_active(struct pw_impl_node *node);
 
+/** Set a node enabled. This will start negotiation with all linked active
+  * nodes and start data transport */
+int pw_impl_node_set_enabled(struct pw_impl_node *node, bool enabled);
+/** Check if a node is enabled */
+bool pw_impl_node_is_enabled(struct pw_impl_node *node);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/src/pipewire/private.h b/src/pipewire/private.h
index 366a2202..5f83dce0 100644
--- a/src/pipewire/private.h
+++ b/src/pipewire/private.h
@@ -521,6 +521,7 @@ struct pw_node_activation {
 #define pw_impl_node_emit_driver_changed(n,o,d)		pw_impl_node_emit(n, driver_changed, 0, o, d)
 #define pw_impl_node_emit_peer_added(n,p)		pw_impl_node_emit(n, peer_added, 0, p)
 #define pw_impl_node_emit_peer_removed(n,p)		pw_impl_node_emit(n, peer_removed, 0, p)
+#define pw_impl_node_emit_enabled_changed(n,e)		pw_impl_node_emit(n, enabled_changed, 1, e)
 
 struct pw_impl_node {
 	struct pw_context *context;		/**< context object */
@@ -550,6 +551,7 @@ struct pw_impl_node {
 	unsigned int visited:1;		/**< for sorting */
 	unsigned int want_driver:1;	/**< this node wants to be assigned to a driver */
 	unsigned int passive:1;		/**< driver graph only has passive links */
+	unsigned int enabled:1;		/**< if the node is enabled */
 
 	uint32_t port_user_data_size;	/**< extra size for port user data */
 
diff --git a/src/pipewire/stream.c b/src/pipewire/stream.c
index 4b21e505..02bbd45e 100644
--- a/src/pipewire/stream.c
+++ b/src/pipewire/stream.c
@@ -1415,6 +1415,7 @@ pw_stream_connect(struct pw_stream *stream,
 	struct pw_impl_node *follower;
 	const char *str;
 	uint32_t i;
+	bool activate;
 	int res;
 
 	pw_log_debug(NAME" %p: connect target:%d", stream, target_id);
@@ -1464,6 +1465,8 @@ pw_stream_connect(struct pw_stream *stream,
 	impl->disconnecting = false;
 	stream_set_state(stream, PW_STREAM_STATE_CONNECTING, NULL);
 
+	activate = !SPA_FLAG_IS_SET(impl->flags, PW_STREAM_FLAG_INACTIVE);
+
 	if (target_id != PW_ID_ANY)
 		pw_properties_setf(stream->properties, PW_KEY_NODE_TARGET, "%d", target_id);
 	else if ((str = getenv("PIPEWIRE_NODE")) != NULL)
@@ -1536,6 +1539,7 @@ pw_stream_connect(struct pw_stream *stream,
 			goto error_node;
 		}
 		pw_properties_setf(props, "adapt.follower.node", "pointer:%p", follower);
+		pw_properties_setf(props, "node.active", "%s", activate ? "true" : "false");
 		impl->node = pw_impl_factory_create_object(factory,
 				NULL,
 				PW_TYPE_INTERFACE_Node,
@@ -1552,8 +1556,6 @@ pw_stream_connect(struct pw_stream *stream,
 		pw_properties_free(props);
 		props = NULL;
 	}
-	if (!SPA_FLAG_IS_SET(impl->flags, PW_STREAM_FLAG_INACTIVE))
-		pw_impl_node_set_active(impl->node, true);
 
 	pw_log_debug(NAME" %p: export node %p", stream, impl->node);
 	stream->proxy = pw_core_export(stream->core,
@@ -1563,6 +1565,8 @@ pw_stream_connect(struct pw_stream *stream,
 		goto error_proxy;
 	}
 
+	pw_impl_node_set_active(impl->node, activate);
+
 	pw_proxy_add_listener(stream->proxy, &stream->proxy_listener, &proxy_events, stream);
 
 	pw_impl_node_add_listener(impl->node, &stream->node_listener, &node_events, stream);