client-node: add get_node method

Make a get node method that binds to the server side node of the
client-node immediately. use this in the remote_export and always
return a node proxy.

Use the node proxy to get property updates and signal those in the
stream.

src/extensions/client-node.h

@@ -48,17 +48,21 @@ struct pw_client_node_buffer {
 	struct spa_buffer *buffer;	/**< buffer describing metadata and buffer memory */
 };
 
-#define PW_CLIENT_NODE_PROXY_METHOD_UPDATE		0
+#define PW_CLIENT_NODE_PROXY_METHOD_GET_NODE		0
-#define PW_CLIENT_NODE_PROXY_METHOD_PORT_UPDATE		1
+#define PW_CLIENT_NODE_PROXY_METHOD_UPDATE		1
-#define PW_CLIENT_NODE_PROXY_METHOD_SET_ACTIVE		2
+#define PW_CLIENT_NODE_PROXY_METHOD_PORT_UPDATE		2
-#define PW_CLIENT_NODE_PROXY_METHOD_EVENT		3
+#define PW_CLIENT_NODE_PROXY_METHOD_SET_ACTIVE		3
-#define PW_CLIENT_NODE_PROXY_METHOD_NUM			4
+#define PW_CLIENT_NODE_PROXY_METHOD_EVENT		4
+#define PW_CLIENT_NODE_PROXY_METHOD_NUM			5
 
 /** \ref pw_client_node methods */
 struct pw_client_node_proxy_methods {
 #define PW_VERSION_CLIENT_NODE_PROXY_METHODS		0
 	uint32_t version;
 
+	/** get the node object
+	 */
+	int (*get_node) (void *object, uint32_t version, uint32_t new_id);
 	/**
 	 * Update the node ports and properties
 	 *
@@ -108,6 +112,15 @@ struct pw_client_node_proxy_methods {
 	int (*event) (void *object, struct spa_event *event);
 };
 
+static inline struct pw_node_proxy *
+pw_client_node_proxy_get_node(struct pw_client_node_proxy *p, uint32_t version, size_t user_data_size)
+{
+	struct pw_proxy *np = pw_proxy_new((struct pw_proxy*)p, PW_TYPE_INTERFACE_Node, user_data_size);
+	pw_proxy_do((struct pw_proxy*)p, struct pw_client_node_proxy_methods,
+			get_node, version, pw_proxy_get_id(np));
+	return (struct pw_node_proxy *) np;
+}
+
 static inline int
 pw_client_node_proxy_update(struct pw_client_node_proxy *p,
 			    uint32_t change_mask,

src/modules/module-client-node/client-node.c

@@ -165,7 +165,9 @@ struct impl {
 	struct spa_hook resource_listener;
 
 	struct pw_array mems;
-	uint32_t init_seq;
+
+	uint32_t bind_node_version;
+	uint32_t bind_node_id;
 
 	int fds[2];
 	int other_fds[2];
@@ -987,6 +989,20 @@ static int impl_node_process(struct spa_node *node)
 	return SPA_STATUS_OK;
 }
 
+static int
+client_node_get_node(void *data,
+		   uint32_t version,
+		   uint32_t new_id)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	pw_log_debug("node %p: bind %u/%u", this, new_id, version);
+	impl->bind_node_version = version;
+	impl->bind_node_id = new_id;
+	pw_map_insert_at(&this->resource->client->objects, new_id, NULL);
+	return 0;
+}
+
 static int
 client_node_update(void *data,
 		   uint32_t change_mask,
@@ -1076,6 +1092,7 @@ static int client_node_event(void *data, struct spa_event *event)
 
 static struct pw_client_node_proxy_methods client_node_methods = {
 	PW_VERSION_CLIENT_NODE_PROXY_METHODS,
+	.get_node = client_node_get_node,
 	.update = client_node_update,
 	.port_update = client_node_port_update,
 	.set_active = client_node_set_active,
@@ -1232,10 +1249,11 @@ static void client_node_resource_pong(void *data, int seq)
 	spa_node_emit_result(&this->hooks, seq, 0, NULL);
 }
 
-void pw_client_node_registered(struct pw_client_node *this, uint32_t node_id)
+void pw_client_node_registered(struct pw_client_node *this, struct pw_global *global)
 {
 	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
 	struct pw_node *node = this->node;
+	uint32_t node_id = global->id;
 	struct mem *m;
 
 	pw_log_debug("client-node %p: %d", this, node_id);
@@ -1252,6 +1270,11 @@ void pw_client_node_registered(struct pw_client_node *this, uint32_t node_id)
 					  m->id,
 					  0,
 					  sizeof(struct pw_node_activation));
+
+	if (impl->bind_node_id) {
+		pw_global_bind(global, this->resource->client, PW_PERM_RWX,
+				impl->bind_node_version, impl->bind_node_id);
+	}
 }
 
 static void node_initialized(void *data)
@@ -1287,7 +1310,7 @@ static void node_initialized(void *data)
 	pw_log_debug("client-node %p: io areas %p", node, impl->io_areas->ptr);
 
 	if ((global = pw_node_get_global(node)) != NULL)
-		pw_client_node_registered(this, pw_global_get_id(global));
+		pw_client_node_registered(this, global);
 }
 
 static void node_free(void *data)

src/modules/module-client-node/client-node.h

@@ -53,7 +53,7 @@ pw_client_node_new(struct pw_resource *resource,
 void
 pw_client_node_destroy(struct pw_client_node *node);
 
-void pw_client_node_registered(struct pw_client_node *node, uint32_t node_id);
+void pw_client_node_registered(struct pw_client_node *node, struct pw_global *global);
 
 #ifdef __cplusplus
 }

src/modules/module-client-node/client-stream.c

@@ -1210,7 +1210,7 @@ static void node_free(void *data)
 static void node_initialized(void *data)
 {
 	struct impl *impl = data;
-	pw_client_node_registered(impl->client_node, impl->this.node->global->id);
+	pw_client_node_registered(impl->client_node, impl->this.node->global);
 }
 
 static const struct pw_node_events node_events = {

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

@@ -47,6 +47,21 @@ static void push_dict(struct spa_pod_builder *b, const struct spa_dict *dict)
 	spa_pod_builder_pop(b, &f);
 }
 
+static int
+client_node_marshal_get_node(void *object, uint32_t version, uint32_t new_id)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_PROXY_METHOD_GET_NODE, NULL);
+
+	spa_pod_builder_add_struct(b,
+		       SPA_POD_Int(version),
+		       SPA_POD_Int(new_id));
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
 static int
 client_node_marshal_update(void *object,
 			   uint32_t change_mask,
@@ -770,6 +785,22 @@ client_node_marshal_set_io(void *object,
 	return pw_protocol_native_end_resource(resource, b);
 }
 
+static int client_node_demarshal_get_node(void *object, void *data, size_t size)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	int32_t version, new_id;
+
+	spa_pod_parser_init(&prs, data, size);
+	if (spa_pod_parser_get_struct(&prs,
+				SPA_POD_Int(&version),
+				SPA_POD_Int(&new_id)) < 0)
+		return -EINVAL;
+
+	return pw_resource_do(resource, struct pw_client_node_proxy_methods, get_node, 0,
+			version, new_id);
+}
+
 static int client_node_demarshal_update(void *object, void *data, size_t size)
 {
 	struct pw_resource *resource = object;
@@ -966,6 +997,7 @@ static int client_node_demarshal_event_method(void *object, void *data, size_t s
 
 static const struct pw_client_node_proxy_methods pw_protocol_native_client_node_method_marshal = {
 	PW_VERSION_CLIENT_NODE_PROXY_METHODS,
+	&client_node_marshal_get_node,
 	&client_node_marshal_update,
 	&client_node_marshal_port_update,
 	&client_node_marshal_set_active,
@@ -973,6 +1005,7 @@ static const struct pw_client_node_proxy_methods pw_protocol_native_client_node_
 };
 
 static const struct pw_protocol_native_demarshal pw_protocol_native_client_node_method_demarshal[] = {
+	{ &client_node_demarshal_get_node, 0 },
 	{ &client_node_demarshal_update, 0 },
 	{ &client_node_demarshal_port_update, 0 },
 	{ &client_node_demarshal_set_active, 0 },

src/modules/module-client-node/remote-node.c

@@ -107,6 +107,7 @@ struct node_data {
         struct pw_client_node_proxy *node_proxy;
 	struct spa_hook node_proxy_listener;
 	struct spa_hook proxy_listener;
+        struct pw_proxy *proxy;
 
 	struct spa_io_position *position;
 
@@ -419,7 +420,7 @@ static int client_node_transport(void *object, uint32_t node_id,
 	if (data->node->active)
 		pw_client_node_proxy_set_active(data->node_proxy, true);
 
-	pw_remote_emit_exported(remote, proxy->id, node_id);
+	pw_remote_emit_exported(remote, data->proxy->id, node_id);
 	return 0;
 }
 
@@ -1282,7 +1283,10 @@ static struct pw_proxy *node_export(struct pw_remote *remote, void *object, bool
 					  proxy);
         do_node_init(proxy);
 
-	return proxy;
+	data->proxy = (struct pw_proxy*) pw_client_node_proxy_get_node(data->node_proxy,
+			PW_VERSION_NODE, 0);
+
+	return data->proxy;
 }
 
 struct pw_proxy *pw_remote_node_export(struct pw_remote *remote,

src/pipewire/private.h

@@ -645,7 +645,8 @@ struct pw_remote {
 #define pw_stream_emit_add_buffer(s,b)		pw_stream_emit(s, add_buffer, 0, b)
 #define pw_stream_emit_remove_buffer(s,b)	pw_stream_emit(s, remove_buffer, 0, b)
 #define pw_stream_emit_process(s)		pw_stream_emit(s, process, 0)
-#define pw_stream_emit_drained(s)		pw_stream_emit(s, drained, 1)
+#define pw_stream_emit_drained(s)		pw_stream_emit(s, drained,0)
+#define pw_stream_emit_control_changed(s,i,v)	pw_stream_emit(s, control_changed, 0, i, v)
 
 
 struct pw_stream {
@@ -664,6 +665,9 @@ struct pw_stream {
 
 	struct pw_proxy *proxy;
 	struct spa_hook proxy_listener;
+
+	struct pw_node_proxy *node;
+	struct spa_hook node_listener;
 };
 
 #define pw_factory_emit(s,m,v,...) spa_hook_list_call(&s->listener_list, struct pw_factory_events, m, v, ##__VA_ARGS__)

src/pipewire/stream.c

@@ -133,6 +133,8 @@ struct stream {
 	uintptr_t seq;
 	struct pw_time time;
 
+	uint32_t param_propinfo;
+
 	int async_connect:1;
 	int disconnecting:1;
 	int free_data:1;
@@ -849,6 +851,69 @@ static const struct pw_proxy_events proxy_events = {
 	.error = proxy_error,
 };
 
+static void node_event_info(void *object, const struct pw_node_info *info)
+{
+	struct pw_stream *stream = object;
+	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
+	uint32_t i;
+
+	if (info->change_mask & PW_NODE_CHANGE_MASK_PARAMS) {
+		for (i = 0; i < info->n_params; i++) {
+			if (!(info->params[i].flags & SPA_PARAM_INFO_READ))
+				continue;
+
+			switch (info->params[i].id) {
+			case SPA_PARAM_PropInfo:
+				if (info->params[i].flags == impl->param_propinfo)
+					break;
+				impl->param_propinfo = info->params[i].flags;
+				/* fallthrough */
+			case SPA_PARAM_Props:
+				pw_node_proxy_enum_params((struct pw_node_proxy*)stream->proxy,
+					0, info->params[i].id, 0, -1, NULL);
+				break;
+			default:
+				break;
+			}
+		}
+	}
+}
+
+static void node_event_param(void *object, int seq,
+		uint32_t id, uint32_t index, uint32_t next,
+		const struct spa_pod *param)
+{
+	struct pw_stream *stream = object;
+
+	switch (id) {
+	case SPA_PARAM_PropInfo:
+		pw_log_debug("info");
+		break;
+	case SPA_PARAM_Props:
+	{
+		struct spa_pod_prop *prop;
+		struct spa_pod_object *obj = (struct spa_pod_object *) param;
+		float value;
+
+		SPA_POD_OBJECT_FOREACH(obj, prop) {
+			if (spa_pod_get_float(&prop->value, &value) < 0)
+				continue;
+			pw_log_debug("stream %p: control %d changed %f", stream, prop->key, value);
+			pw_stream_emit_control_changed(stream, prop->key, value);
+		}
+		break;
+	}
+	default:
+		break;
+	}
+}
+
+static const struct pw_node_proxy_events node_events = {
+	PW_VERSION_NODE_PROXY_EVENTS,
+	.info = node_event_info,
+	.param = node_event_param,
+};
+
 static int handle_connect(struct pw_stream *stream)
 {
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
@@ -879,6 +944,8 @@ static int handle_connect(struct pw_stream *stream)
 		goto no_proxy;
 
 	pw_proxy_add_listener(stream->proxy, &stream->proxy_listener, &proxy_events, stream);
+	pw_node_proxy_add_listener((struct pw_node_proxy*)stream->proxy,
+			&stream->node_listener, &node_events, stream);
 
 	return 0;
 
@@ -1266,35 +1333,26 @@ void pw_stream_finish_format(struct pw_stream *stream,
 }
 
 SPA_EXPORT
-int pw_stream_set_control(struct pw_stream *stream,
+int pw_stream_set_control(struct pw_stream *stream, uint32_t id, float value)
-			  const char *name, float value)
 {
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
 
-	if (strcmp(name, PW_STREAM_CONTROL_VOLUME) == 0) {
+	switch (id) {
+	case SPA_PROP_volume:
 		impl->props.volume = value;
-	}
+		break;
-	else
+	default:
 		return -ENOTSUP;
-
+	}
 	impl->props.changed = true;
 
 	return 0;
 }
 
 SPA_EXPORT
-int pw_stream_get_control(struct pw_stream *stream,
+const struct pw_stream_control *pw_stream_get_control(struct pw_stream *stream, uint32_t id)
-			  const char *name, float *value)
 {
-	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
+	return NULL;
-
-	if (strcmp(name, PW_STREAM_CONTROL_VOLUME) == 0) {
-		*value = impl->props.volume;
-	}
-	else
-		return -ENOTSUP;
-
-	return 0;
 }
 
 SPA_EXPORT

src/pipewire/stream.h

@@ -178,6 +178,15 @@ struct pw_buffer {
 					  *  returned in the time info. */
 };
 
+struct pw_stream_control {
+	const char *name;
+	uint32_t flags;
+	float value;
+	float def;
+	float min;
+	float max;
+};
+
 /** Events for a stream. These events are always called from the mainloop
  * unless explicitly documented otherwise. */
 struct pw_stream_events {
@@ -188,6 +197,10 @@ struct pw_stream_events {
 	/** when the stream state changes */
 	void (*state_changed) (void *data, enum pw_stream_state old,
 				enum pw_stream_state state, const char *error);
+
+	/** Notify information about a control.  */
+	void (*control_changed) (void *data, uint32_t id, float value);
+
 	/** when the format changed. The listener should call
 	 * pw_stream_finish_format() from within this callback or later to complete
 	 * the format negotiation and start the buffer negotiation. */
@@ -206,6 +219,7 @@ struct pw_stream_events {
 
 	/** The stream is drained */
         void (*drained) (void *data);
+
 };
 
 /** Convert a stream state to a readable string \memberof pw_stream */
@@ -309,19 +323,11 @@ pw_stream_finish_format(struct pw_stream *stream,	/**< a \ref pw_stream */
 			uint32_t n_params		/**< number of elements in \a params */);
 
 
-/** Audio controls */
-#define PW_STREAM_CONTROL_VOLUME	"volume"
-
-/** Video controls */
-#define PW_STREAM_CONTROL_CONTRAST	"contrast"
-#define PW_STREAM_CONTROL_BRIGHTNESS	"brightness"
-#define PW_STREAM_CONTROL_HUE		"hue"
-#define PW_STREAM_CONTROL_SATURATION	"saturation"
-
 /** Set a control value */
-int pw_stream_set_control(struct pw_stream *stream, const char *name, float value);
+int pw_stream_set_control(struct pw_stream *stream, uint32_t id, float value);
-/** Get a control value */
+
-int pw_stream_get_control(struct pw_stream *stream, const char *name, float *value);
+/** Get control information */
+const struct pw_stream_control * pw_stream_get_control(struct pw_stream *stream, uint32_t id);
 
 /** A time structure \memberof pw_stream */
 struct pw_time {