media-session: track nodes we create

Add a method to create a node and track the proxy. We can then
use this to track the params on a node.

Don't listen end enumerate the params in the monitors bbut let the
session do that. Use the collected params from the session to create
endpoints and streams.

Always activate the default stream on alsa nodes to create the
ports and have something we can present to apps.

6 files changed, 268 insertions, 174 deletions

diff --git a/src/examples/media-session/alsa-endpoint.c b/src/examples/media-session/alsa-endpoint.c
index c37125af..8f6d6c0b 100644
--- a/src/examples/media-session/alsa-endpoint.c
+++ b/src/examples/media-session/alsa-endpoint.c
@@ -44,6 +44,9 @@
 
 #include "pipewire/pipewire.h"
 
+#undef NAME
+#define NAME "alsa-endpoint"
+
 struct endpoint {
 	struct spa_list link;
 
@@ -56,15 +59,17 @@ struct endpoint {
 	struct spa_hook client_endpoint_listener;
 	struct pw_endpoint_info info;
 
+	struct spa_param_info params[5];
+	uint32_t n_params;
+
 	struct endpoint *monitor;
 
 	unsigned int use_ucm:1;
 	snd_use_case_mgr_t *ucm;
 
-	struct spa_list stream_list;
 	struct spa_audio_info format;
 
-	unsigned int active:1;
+	struct spa_list stream_list;
 };
 
 struct stream {
@@ -73,6 +78,8 @@ struct stream {
 	struct pw_properties *props;
 	struct pw_endpoint_stream_info info;
 
+	struct spa_audio_info format;
+
 	unsigned int active:1;
 };
 
@@ -80,10 +87,6 @@ static int client_endpoint_set_id(void *object, uint32_t id)
 {
 	struct endpoint *endpoint = object;
 	endpoint->info.id = id;
-	pw_client_endpoint_proxy_update(endpoint->client_endpoint,
-			PW_CLIENT_ENDPOINT_UPDATE_INFO,
-			0, NULL,
-			&endpoint->info);
 	return 0;
 }
 
@@ -111,26 +114,20 @@ static int client_endpoint_stream_set_param(void *object, uint32_t stream_id,
 	return -ENOTSUP;
 }
 
-static int client_endpoint_create_link(void *object, const struct spa_dict *props)
+static int stream_set_active(struct endpoint *endpoint, struct stream *stream, bool active)
 {
-	struct endpoint *endpoint = object;
-	struct impl *impl = endpoint->obj->impl;
-	struct pw_properties *p;
 	char buf[1024];
 	struct spa_pod_builder b = { 0, };
 	struct spa_pod *param;
-	int res;
-
-	pw_log_debug(NAME" %p: endpoint %p", impl, endpoint);
 
-	if (props == NULL)
-		return -EINVAL;
+	if (stream->active == active)
+		return 0;
 
-	if (!endpoint->active) {
-		endpoint->format.info.raw.rate = 48000;
+	if (active) {
+		stream->format.info.raw.rate = 48000;
 
 		spa_pod_builder_init(&b, buf, sizeof(buf));
-		param = spa_format_audio_raw_build(&b, SPA_PARAM_Format, &endpoint->format.info.raw);
+		param = spa_format_audio_raw_build(&b, SPA_PARAM_Format, &stream->format.info.raw);
 		param = spa_pod_builder_add_object(&b,
 			SPA_TYPE_OBJECT_ParamPortConfig, SPA_PARAM_PortConfig,
 			SPA_PARAM_PORT_CONFIG_direction, SPA_POD_Id(endpoint->info.direction),
@@ -143,9 +140,22 @@ static int client_endpoint_create_link(void *object, const struct spa_dict *prop
 
 		pw_node_proxy_set_param((struct pw_node_proxy*)endpoint->obj->proxy,
 				SPA_PARAM_PortConfig, 0, param);
-
-		endpoint->active = true;
 	}
+	stream->active = active;
+	return 0;
+}
+
+static int client_endpoint_create_link(void *object, const struct spa_dict *props)
+{
+	struct endpoint *endpoint = object;
+	struct impl *impl = endpoint->obj->impl;
+	struct pw_properties *p;
+	int res;
+
+	pw_log_debug(NAME" %p: endpoint %p", impl, endpoint);
+
+	if (props == NULL)
+		return -EINVAL;
 
 	p = pw_properties_new_dict(props);
 	if (p == NULL)
@@ -195,14 +205,14 @@ static const struct pw_client_endpoint_proxy_events client_endpoint_events = {
 	.create_link = client_endpoint_create_link,
 };
 
-static int endpoint_add_stream(struct endpoint *endpoint)
+static struct stream *endpoint_add_stream(struct endpoint *endpoint)
 {
 	struct stream *s;
 	const char *str;
 
 	s = calloc(1, sizeof(*s));
 	if (s == NULL)
-		return -errno;
+		return NULL;
 
 	s->props = pw_properties_new(NULL, NULL);
 	if ((str = pw_properties_get(endpoint->props, PW_KEY_MEDIA_CLASS)) != NULL)
@@ -224,6 +234,7 @@ static int endpoint_add_stream(struct endpoint *endpoint)
 	s->info.name = (char*)pw_properties_get(s->props, PW_KEY_ENDPOINT_STREAM_NAME);
 	s->info.change_mask = PW_ENDPOINT_STREAM_CHANGE_MASK_PROPS;
 	s->info.props = &s->props->dict;
+	s->format = endpoint->format;
 
 	pw_log_debug("stream %d", s->info.id);
 	pw_client_endpoint_proxy_stream_update(endpoint->client_endpoint,
@@ -234,61 +245,51 @@ static int endpoint_add_stream(struct endpoint *endpoint)
 
 	spa_list_append(&endpoint->stream_list, &s->link);
 	endpoint->info.n_streams++;
-	return 0;
-}
-
-static void node_event_param(void *object, int seq,
-                       uint32_t id, uint32_t index, uint32_t next,
-                       const struct spa_pod *param)
-{
-	struct endpoint *endpoint = object;
-	struct alsa_node *n = endpoint->obj;
-	struct impl *impl = n->impl;
-	struct spa_audio_info info = { 0, };
 
-	pw_log_debug(NAME" %p: param for node %d, %d", impl, n->info->id, id);
+	return s;
+}
 
-	if (id != SPA_PARAM_EnumFormat)
-		goto error;
+static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *monitor);
 
-	if (spa_format_parse(param, &info.media_type, &info.media_subtype) < 0)
-		goto error;
+static void complete_endpoint(void *data)
+{
+	struct endpoint *endpoint = data;
+	struct stream *stream;
+	struct sm_param *p;
 
-	if (info.media_type != SPA_MEDIA_TYPE_audio ||
-	    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
-		return;
+	pw_log_debug("endpoint %p: complete", endpoint);
 
-	spa_pod_object_fixate((struct spa_pod_object*)param);
-	if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
-		spa_debug_pod(2, NULL, param);
+	spa_list_for_each(p, &endpoint->obj->snode->param_list, link) {
+		struct spa_audio_info info = { 0, };
 
-	if (spa_format_audio_raw_parse(param, &info.info.raw) < 0)
-		goto error;
+		if (p->id != SPA_PARAM_EnumFormat)
+			continue;
 
-	if (endpoint->format.info.raw.channels < info.info.raw.channels)
-		endpoint->format = info;
+		if (spa_format_parse(p->param, &info.media_type, &info.media_subtype) < 0)
+			continue;
 
-	return;
+		if (info.media_type != SPA_MEDIA_TYPE_audio ||
+		    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
+			continue;
 
-      error:
-	pw_log_warn("unhandled param:");
-	if (pw_log_level_enabled(SPA_LOG_LEVEL_WARN))
-		spa_debug_pod(2, NULL, param);
-	return;
-}
+		spa_pod_object_fixate((struct spa_pod_object*)p->param);
+		if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
+			spa_debug_pod(2, NULL, p->param);
 
-static const struct pw_node_proxy_events endpoint_node_events = {
-	PW_VERSION_NODE_PROXY_EVENTS,
-	.param = node_event_param,
-};
+		if (spa_format_audio_raw_parse(p->param, &info.info.raw) < 0)
+			continue;
 
-static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *monitor);
+		if (endpoint->format.info.raw.channels < info.info.raw.channels)
+			endpoint->format = info;
+	}
 
-static void complete_endpoint(void *data)
-{
-	struct endpoint *endpoint = data;
+	pw_client_endpoint_proxy_update(endpoint->client_endpoint,
+			PW_CLIENT_ENDPOINT_UPDATE_PARAMS |
+			PW_CLIENT_ENDPOINT_UPDATE_INFO,
+			0, NULL,
+			&endpoint->info);
 
-	endpoint_add_stream(endpoint);
+	stream = endpoint_add_stream(endpoint);
 
 	if (endpoint->info.direction == PW_DIRECTION_INPUT) {
 		struct endpoint *monitor;
@@ -300,6 +301,7 @@ static void complete_endpoint(void *data)
 
 		endpoint_add_stream(monitor);
 	}
+	stream_set_active(endpoint, stream, true);
 }
 
 static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *monitor)
@@ -309,6 +311,7 @@ static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *mo
 	struct endpoint *endpoint;
 	struct pw_proxy *proxy;
 	const char *str, *media_class = NULL, *name = NULL;
+	uint32_t subscribe[4], n_subscribe = 0;
 
 	props = pw_properties_new(NULL, NULL);
 	if (props == NULL)
@@ -362,9 +365,14 @@ static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *mo
 	endpoint->info.change_mask =
 		PW_ENDPOINT_CHANGE_MASK_STREAMS |
 		PW_ENDPOINT_CHANGE_MASK_SESSION |
-		PW_ENDPOINT_CHANGE_MASK_PROPS;
+		PW_ENDPOINT_CHANGE_MASK_PROPS |
+		PW_ENDPOINT_CHANGE_MASK_PARAMS;
 	endpoint->info.n_streams = 0;
 	endpoint->info.props = &endpoint->props->dict;
+	endpoint->params[0] = SPA_PARAM_INFO(SPA_PARAM_PropInfo, SPA_PARAM_INFO_READ);
+	endpoint->params[1] = SPA_PARAM_INFO(SPA_PARAM_Props, SPA_PARAM_INFO_READWRITE);
+	endpoint->info.params = endpoint->params;
+	endpoint->info.n_params = 2;
 	spa_list_init(&endpoint->stream_list);
 
 	pw_log_debug(NAME" %p: new endpoint %p for alsa node %p", impl, endpoint, obj);
@@ -374,11 +382,13 @@ static struct endpoint *make_endpoint(struct alsa_node *obj, struct endpoint *mo
 			&client_endpoint_events,
 			endpoint);
 
-	pw_proxy_add_object_listener(obj->proxy, &endpoint->listener, &endpoint_node_events, endpoint);
-
-	pw_node_proxy_enum_params((struct pw_node_proxy*)obj->proxy,
-				0, SPA_PARAM_EnumFormat,
-				0, -1, NULL);
+	subscribe[n_subscribe++] = SPA_PARAM_EnumFormat;
+	subscribe[n_subscribe++] = SPA_PARAM_Props;
+	subscribe[n_subscribe++] = SPA_PARAM_PropInfo;
+	pw_log_debug(NAME" %p: endpoint %p proxy %p subscribe %d params", impl,
+				endpoint, obj->proxy, n_subscribe);
+	pw_node_proxy_subscribe_params((struct pw_node_proxy*)obj->proxy,
+				subscribe, n_subscribe);
 
 	if (monitor == NULL)
 		sm_media_session_sync(impl->session, complete_endpoint, endpoint);
diff --git a/src/examples/media-session/alsa-monitor.c b/src/examples/media-session/alsa-monitor.c
index 4d94f0f2..3b1c93c6 100644
--- a/src/examples/media-session/alsa-monitor.c
+++ b/src/examples/media-session/alsa-monitor.c
@@ -63,6 +63,8 @@ struct alsa_node {
 
 	struct spa_node *node;
 
+	struct sm_node *snode;
+
 	struct pw_proxy *proxy;
 	struct spa_hook listener;
 	struct pw_node_info *info;
@@ -108,6 +110,9 @@ struct impl {
 
 #include "alsa-endpoint.c"
 
+#undef NAME
+#define NAME "alsa-monitor"
+
 static struct alsa_node *alsa_find_node(struct alsa_object *obj, uint32_t id)
 {
 	struct alsa_node *node;
@@ -133,6 +138,7 @@ static void alsa_update_node(struct alsa_object *obj, struct alsa_node *node,
 static void node_event_info(void *object, const struct pw_node_info *info)
 {
 	struct alsa_node *node = object;
+	pw_log_debug("node info %d", info->id);
 	node->info = pw_node_info_update(node->info, info);
 }
 
@@ -238,12 +244,11 @@ static struct alsa_node *alsa_create_node(struct alsa_object *obj, uint32_t id,
 	node->impl = impl;
 	node->object = obj;
 	node->id = id;
-	node->proxy = sm_media_session_create_object(impl->session,
+	node->snode = sm_media_session_create_node(impl->session,
 				"adapter",
-				PW_TYPE_INTERFACE_Node,
-				PW_VERSION_NODE_PROXY,
 				&node->props->dict,
                                 0);
+	node->proxy = node->snode->obj.proxy;
 	if (node->proxy == NULL) {
 		res = -errno;
 		goto clean_node;
@@ -420,7 +425,7 @@ static void set_jack_profile(struct impl *impl, int index)
 	pw_device_proxy_set_param((struct pw_device_proxy*)impl->jack_device,
 			SPA_PARAM_Profile, 0,
 			spa_pod_builder_add_object(&b,
-				SPA_TYPE_OBJECT_ParamProfile, 0,
+				SPA_TYPE_OBJECT_ParamProfile, SPA_PARAM_Profile,
 				SPA_PARAM_PROFILE_index,   SPA_POD_Int(index)));
 }
 
@@ -431,7 +436,7 @@ static void set_profile(struct alsa_object *obj, int index)
 	spa_device_set_param(obj->device,
 			SPA_PARAM_Profile, 0,
 			spa_pod_builder_add_object(&b,
-				SPA_TYPE_OBJECT_ParamProfile, 0,
+				SPA_TYPE_OBJECT_ParamProfile, SPA_PARAM_Profile,
 				SPA_PARAM_PROFILE_index,   SPA_POD_Int(index)));
 }
 
diff --git a/src/examples/media-session/media-session.c b/src/examples/media-session/media-session.c
index 4674e0df..be7ca43c 100644
--- a/src/examples/media-session/media-session.c
+++ b/src/examples/media-session/media-session.c
@@ -47,9 +47,13 @@
 
 #define NAME "media-session"
 
+#define sm_object_emit(o,m,v,...) spa_hook_list_call(&(o)->hooks, struct sm_object_events, m, v, ##__VA_ARGS__)
+
+#define sm_object_emit_update(s)		sm_object_emit(s, update, 0)
+
 #define sm_media_session_emit(s,m,v,...) spa_hook_list_call(&s->hooks, struct sm_media_session_events, m, v, ##__VA_ARGS__)
 
-#define sm_media_session_emit_update(s,obj)		sm_media_session_emit(s, update, 0, obj)
+#define sm_media_session_emit_create(s,obj)		sm_media_session_emit(s, create, 0, obj)
 #define sm_media_session_emit_remove(s,obj)		sm_media_session_emit(s, remove, 0, obj)
 #define sm_media_session_emit_rescan(s,seq)		sm_media_session_emit(s, rescan, 0, seq)
 
@@ -70,7 +74,6 @@ struct data {
 
 struct param {
 	struct sm_param this;
-	struct spa_list link;
 };
 
 struct sync {
@@ -225,7 +228,7 @@ static void client_event_info(void *object, const struct pw_client_info *info)
 
 	client->avail |= SM_CLIENT_CHANGE_MASK_INFO;
 	client->changed |= SM_CLIENT_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &client->obj);
+	sm_object_emit_update(&client->obj);
 	client->changed = 0;
 }
 
@@ -261,7 +264,7 @@ static struct param *add_param(struct spa_list *param_list,
 	p->this.param = SPA_MEMBER(p, sizeof(struct param), struct spa_pod);
 	memcpy(p->this.param, param, SPA_POD_SIZE(param));
 
-	spa_list_append(param_list, &p->link);
+	spa_list_append(param_list, &p->this.link);
 
 	return p;
 }
@@ -271,9 +274,9 @@ static void clear_params(struct spa_list *param_list, uint32_t id)
 {
 	struct param *p, *t;
 
-	spa_list_for_each_safe(p, t, param_list, link) {
+	spa_list_for_each_safe(p, t, param_list, this.link) {
 		if (id == SPA_ID_INVALID || p->this.id == id) {
-			spa_list_remove(&p->link);
+			spa_list_remove(&p->this.link);
 			free(p);
 		}
 	}
@@ -291,9 +294,15 @@ static void node_event_info(void *object, const struct pw_node_info *info)
 	pw_log_debug(NAME" %p: node %d info", impl, node->obj.id);
 	node->info = pw_node_info_update(node->info, info);
 
+	if (node->obj.id == SPA_ID_INVALID) {
+		node->obj.id = info->id;
+		pw_log_debug(NAME" %p: node %d added", impl, node->obj.id);
+		add_object(impl, &node->obj);
+	}
+
 	node->avail |= SM_NODE_CHANGE_MASK_INFO;
 	node->changed |= SM_NODE_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &node->obj);
+	sm_object_emit_update(&node->obj);
 	node->changed = 0;
 
 	if (info->change_mask & PW_NODE_CHANGE_MASK_PARAMS &&
@@ -313,6 +322,8 @@ static void node_event_info(void *object, const struct pw_node_info *info)
 			}
 		}
 		if (n_subscribe > 0) {
+			pw_log_debug(NAME" %p: node %d subscribe %d params", impl,
+					node->obj.id, n_subscribe);
 			pw_node_proxy_subscribe_params((struct pw_node_proxy*)node->obj.proxy,
 					subscribe, n_subscribe);
 			node->subscribe = true;
@@ -327,6 +338,7 @@ static void node_event_param(void *object, int seq,
 	struct sm_node *node = object;
 	struct impl *impl = SPA_CONTAINER_OF(node->obj.session, struct impl, this);
 
+	pw_log_debug(NAME" %p: node %p param %d index:%d", impl, node, id, index);
 	if (index == 0)
 		clear_params(&node->param_list, id);
 
@@ -334,7 +346,7 @@ static void node_event_param(void *object, int seq,
 
 	node->avail |= SM_NODE_CHANGE_MASK_PARAMS;
 	node->changed |= SM_NODE_CHANGE_MASK_PARAMS;
-	sm_media_session_emit_update(impl, &node->obj);
+	sm_object_emit_update(&node->obj);
 	node->changed = 0;
 }
 
@@ -372,7 +384,7 @@ static void port_event_info(void *object, const struct pw_port_info *info)
 
 	port->avail |= SM_PORT_CHANGE_MASK_INFO;
 	port->changed |= SM_PORT_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &port->obj);
+	sm_object_emit_update(&port->obj);
 	port->changed = 0;
 }
 
@@ -416,7 +428,7 @@ static void session_event_info(void *object, const struct pw_session_info *info)
 
 	sess->avail |= SM_SESSION_CHANGE_MASK_INFO;
 	sess->changed |= SM_SESSION_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &sess->obj);
+	sm_object_emit_update(&sess->obj);
 	sess->changed = 0;
 }
 
@@ -473,7 +485,7 @@ static void endpoint_event_info(void *object, const struct pw_endpoint_info *inf
 
 	endpoint->avail |= SM_ENDPOINT_CHANGE_MASK_INFO;
 	endpoint->changed |= SM_ENDPOINT_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &endpoint->obj);
+	sm_object_emit_update(&endpoint->obj);
 	endpoint->changed = 0;
 }
 
@@ -523,7 +535,7 @@ static void endpoint_stream_event_info(void *object, const struct pw_endpoint_st
 
 	stream->avail |= SM_ENDPOINT_CHANGE_MASK_INFO;
 	stream->changed |= SM_ENDPOINT_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &stream->obj);
+	sm_object_emit_update(&stream->obj);
 	stream->changed = 0;
 }
 
@@ -568,7 +580,7 @@ static void endpoint_link_event_info(void *object, const struct pw_endpoint_link
 
 	link->avail |= SM_ENDPOINT_LINK_CHANGE_MASK_INFO;
 	link->changed |= SM_ENDPOINT_LINK_CHANGE_MASK_INFO;
-	sm_media_session_emit_update(impl, &link->obj);
+	sm_object_emit_update(&link->obj);
 	link->changed = 0;
 }
 
@@ -616,21 +628,21 @@ static const struct pw_proxy_events proxy_events = {
 };
 
 static void
-registry_global(void *data,uint32_t id,
+init_object(struct impl *impl, struct sm_object *obj, uint32_t id,
 		uint32_t permissions, uint32_t type, uint32_t version,
 		const struct spa_dict *props)
 {
-	struct impl *impl = data;
 	int res;
 	const void *events;
         uint32_t client_version;
         pw_destroy_t destroy;
-        struct sm_object *obj = NULL;
-        struct pw_proxy *proxy;
 	size_t user_data_size;
 	const char *str;
+	struct pw_proxy *proxy;
 
-	pw_log_debug(NAME " %p: new global '%d' %d", impl, id, type);
+	proxy = obj ? obj->proxy : NULL;
+
+	pw_log_debug(NAME " %p: init '%d' %d", impl, id, type);
 
 	switch (type) {
 	case PW_TYPE_INTERFACE_Client:
@@ -686,13 +698,14 @@ registry_global(void *data,uint32_t id,
 		return;
 	}
 
-	proxy = pw_registry_proxy_bind(impl->registry_proxy,
-			id, type, client_version, user_data_size);
 	if (proxy == NULL) {
-		res = -errno;
-		goto error;
+		proxy = pw_registry_proxy_bind(impl->registry_proxy,
+				id, type, client_version, user_data_size);
+		if (proxy == NULL) {
+			res = -errno;
+			goto error;
+		}
 	}
-
 	if (obj == NULL)
 		obj = pw_proxy_get_user_data(proxy);
 	obj->session = &impl->this;
@@ -703,9 +716,11 @@ registry_global(void *data,uint32_t id,
 	obj->destroy = destroy;
 	obj->mask = SM_OBJECT_CHANGE_MASK_PROPERTIES | SM_OBJECT_CHANGE_MASK_BIND;
 	obj->avail = obj->mask;
+	spa_hook_list_init(&obj->hooks);
 	spa_list_init(&obj->data);
+	if (id != SPA_ID_INVALID)
+		add_object(impl, obj);
 
-	add_object(impl, obj);
 	pw_proxy_add_listener(proxy, &obj->proxy_listener, &proxy_events, obj);
 
 	switch (type) {
@@ -776,8 +791,8 @@ registry_global(void *data,uint32_t id,
 	default:
 		break;
 	}
-	sm_media_session_emit_update(impl, obj);
 
+	sm_media_session_emit_create(impl, obj);
 	pw_proxy_add_object_listener(proxy, &obj->object_listener, events, obj);
 
 	return;
@@ -786,6 +801,31 @@ error:
 	pw_log_warn(NAME" %p: can't handle global %d: %s", impl, id, spa_strerror(res));
 }
 
+static void
+registry_global(void *data, uint32_t id,
+		uint32_t permissions, uint32_t type, uint32_t version,
+		const struct spa_dict *props)
+{
+	struct impl *impl = data;
+        struct sm_object *obj;
+
+	pw_log_debug(NAME " %p: new global '%d' %d", impl, id, type);
+
+	obj = find_object(impl, id);
+	if (obj == NULL) {
+		init_object(impl, obj, id, permissions, type, version, props);
+	} else {
+		pw_log_debug(NAME " %p: our object %d appeared", impl, id);
+	}
+}
+
+int sm_object_add_listener(struct sm_object *obj, struct spa_hook *listener,
+		const struct sm_object_events *events, void *data)
+{
+	spa_hook_list_append(&obj->hooks, listener, events, data);
+	return 0;
+}
+
 int sm_media_session_add_listener(struct sm_media_session *sess, struct spa_hook *listener,
                 const struct sm_media_session_events *events, void *data)
 {
@@ -796,7 +836,7 @@ int sm_media_session_add_listener(struct sm_media_session *sess, struct spa_hook
 	spa_hook_list_isolate(&impl->hooks, &save, listener, events, data);
 
 	spa_list_for_each(obj, &impl->global_list, link)
-		sm_media_session_emit_update(impl, obj);
+		sm_media_session_emit_create(impl, obj);
 
         spa_hook_list_join(&impl->hooks, &save);
 
@@ -908,6 +948,32 @@ struct pw_proxy *sm_media_session_create_object(struct sm_media_session *sess,
 			factory_name, type, version, props, user_data_size);
 }
 
+struct sm_node *sm_media_session_create_node(struct sm_media_session *sess,
+		const char *factory_name, const struct spa_dict *props,
+		size_t user_data_size)
+{
+	struct impl *impl = SPA_CONTAINER_OF(sess, struct impl, this);
+	struct sm_node *node;
+	struct pw_proxy *proxy;
+
+	pw_log_debug(NAME " %p: node '%s'", impl, factory_name);
+
+	proxy = pw_core_proxy_create_object(impl->core_proxy,
+				factory_name,
+				PW_TYPE_INTERFACE_Node,
+				PW_VERSION_NODE_PROXY,
+				props,
+				sizeof(struct sm_node) + user_data_size);
+
+	node = pw_proxy_get_user_data(proxy);
+	node->obj.proxy = proxy;
+	init_object(impl, &node->obj, SPA_ID_INVALID,
+			PW_PERM_RWX, PW_TYPE_INTERFACE_Node,
+			PW_VERSION_NODE_PROXY, props);
+
+	return node;
+}
+
 static void check_endpoint_link(struct endpoint_link *link)
 {
 	if (!spa_list_is_empty(&link->link_list))
diff --git a/src/examples/media-session/media-session.h b/src/examples/media-session/media-session.h
index 73d5adef..4eeaaf36 100644
--- a/src/examples/media-session/media-session.h
+++ b/src/examples/media-session/media-session.h
@@ -32,6 +32,14 @@ extern "C" {
 
 struct sm_media_session;
 
+struct sm_object_events {
+#define SM_VERSION_OBJECT_EVENTS	0
+	uint32_t version;
+
+	void (*update) (void *data);
+	void (*destroy) (void *data);
+};
+
 struct sm_object {
 	uint32_t id;
 	uint32_t type;
@@ -50,11 +58,17 @@ struct sm_object {
 	struct spa_hook object_listener;
 	pw_destroy_t destroy;
 
+	struct spa_hook_list hooks;
+
 	struct spa_list data;
 };
 
+int sm_object_add_listener(struct sm_object *obj, struct spa_hook *listener,
+		const struct sm_object_events *events, void *data);
+
 struct sm_param {
 	uint32_t id;
+	struct spa_list link;		/**< link in param_list */
 	struct spa_pod *param;
 };
 
@@ -170,7 +184,7 @@ struct sm_media_session_events {
 #define SM_VERSION_MEDIA_SESSION_EVENTS	0
 	uint32_t version;
 
-	void (*update) (void *data, struct sm_object *object);
+	void (*create) (void *data, struct sm_object *object);
 	void (*remove) (void *data, struct sm_object *object);
 
 	void (*rescan) (void *data, int seq);
@@ -205,6 +219,10 @@ struct pw_proxy *sm_media_session_create_object(struct sm_media_session *sess,
 		const char *factory_name, uint32_t type, uint32_t version,
 		const struct spa_dict *props, size_t user_data_size);
 
+struct sm_node *sm_media_session_create_node(struct sm_media_session *sess,
+		const char *factory_name, const struct spa_dict *props,
+		size_t user_data_size);
+
 int sm_media_session_create_links(struct sm_media_session *sess,
 		const struct spa_dict *dict);
 
diff --git a/src/examples/media-session/policy-ep.c b/src/examples/media-session/policy-ep.c
index 60ffb7da..3a5b019a 100644
--- a/src/examples/media-session/policy-ep.c
+++ b/src/examples/media-session/policy-ep.c
@@ -109,9 +109,6 @@ handle_endpoint(struct impl *impl, struct sm_object *object)
 	struct endpoint *ep;
 	uint32_t client_id = SPA_ID_INVALID;
 
-	if (sm_object_get_data(object, SESSION_KEY) != NULL)
-		return 0;
-
 	if (object->props) {
 		if ((str = pw_properties_get(object->props, PW_KEY_CLIENT_ID)) != NULL)
 			client_id = atoi(str);
@@ -184,9 +181,6 @@ handle_stream(struct impl *impl, struct sm_object *object)
 	struct stream *s;
 	struct endpoint *ep;
 
-	if (sm_object_get_data(object, SESSION_KEY) != NULL)
-		return 0;
-
 	if (stream->endpoint == NULL)
 		return 0;
 
@@ -203,7 +197,7 @@ handle_stream(struct impl *impl, struct sm_object *object)
 	return 0;
 }
 
-static void session_update(void *data, struct sm_object *object)
+static void session_create(void *data, struct sm_object *object)
 {
 	struct impl *impl = data;
 	int res;
@@ -534,7 +528,7 @@ static void session_rescan(void *data, int seq)
 
 static const struct sm_media_session_events session_events = {
 	SM_VERSION_MEDIA_SESSION_EVENTS,
-	.update = session_update,
+	.create = session_create,
 	.remove = session_remove,
 	.rescan = session_rescan,
 };
diff --git a/src/examples/media-session/stream-monitor.c b/src/examples/media-session/stream-monitor.c
index ff07944b..3684f4b0 100644
--- a/src/examples/media-session/stream-monitor.c
+++ b/src/examples/media-session/stream-monitor.c
@@ -70,15 +70,15 @@ struct node {
 
 	struct client_endpoint *endpoint;
 
-	uint32_t media_type;
-	uint32_t media_subtype;
-	struct spa_audio_info_raw format;
+	struct spa_audio_info format;
 };
 
 struct stream {
 	struct pw_properties *props;
 	struct pw_endpoint_stream_info info;
 
+	struct spa_audio_info format;
+
 	unsigned int active:1;
 };
 
@@ -137,28 +137,22 @@ static int client_endpoint_stream_set_param(void *object, uint32_t stream_id,
 	return -ENOTSUP;
 }
 
-static int client_endpoint_create_link(void *object, const struct spa_dict *props)
+static int stream_set_active(struct client_endpoint *endpoint, struct stream *stream, bool active)
 {
-	struct client_endpoint *endpoint = object;
-	struct impl *impl = endpoint->impl;
 	struct node *node = endpoint->node;
-	struct pw_properties *p;
-	int res;
-
-	pw_log_debug("create link");
-
-	if (props == NULL)
-		return -EINVAL;
+	char buf[1024];
+	struct spa_pod_builder b = { 0, };
+	struct spa_pod *param;
 
-	if (!endpoint->stream.active) {
-		char buf[1024];
-		struct spa_pod_builder b = { 0, };
-		struct spa_pod *param;
+	if (stream->active == active)
+		return 0;
 
-		node->format.rate = 48000;
+	if (active) {
+		stream->format = node->format;
+		stream->format.info.raw.rate = 48000;
 
 		spa_pod_builder_init(&b, buf, sizeof(buf));
-		param = spa_format_audio_raw_build(&b, SPA_PARAM_Format, &node->format);
+		param = spa_format_audio_raw_build(&b, SPA_PARAM_Format, &stream->format.info.raw);
 		param = spa_pod_builder_add_object(&b,
 			SPA_TYPE_OBJECT_ParamPortConfig, SPA_PARAM_PortConfig,
 			SPA_PARAM_PORT_CONFIG_direction, SPA_POD_Id(endpoint->info.direction),
@@ -171,9 +165,24 @@ static int client_endpoint_create_link(void *object, const struct spa_dict *prop
 
 		pw_node_proxy_set_param((struct pw_node_proxy*)node->obj->obj.proxy,
 				SPA_PARAM_PortConfig, 0, param);
-
-		endpoint->stream.active = true;
 	}
+	stream->active = active;
+	return 0;
+}
+
+static int client_endpoint_create_link(void *object, const struct spa_dict *props)
+{
+	struct client_endpoint *endpoint = object;
+	struct impl *impl = endpoint->impl;
+	struct pw_properties *p;
+	int res;
+
+	pw_log_debug("create link");
+
+	if (props == NULL)
+		return -EINVAL;
+
+	stream_set_active(endpoint, &endpoint->stream, true);
 
 	p = pw_properties_new_dict(props);
 	if (p == NULL)
@@ -315,62 +324,49 @@ static void destroy_endpoint(struct client_endpoint *endpoint)
 	pw_proxy_destroy((struct pw_proxy*)endpoint->client_endpoint);
 }
 
-static void node_event_param(void *object, int seq,
-                       uint32_t id, uint32_t index, uint32_t next,
-                       const struct spa_pod *param)
+static void complete_stream(void *data)
 {
-	struct node *n = object;
-	struct impl *impl = n->impl;
-	struct spa_audio_info_raw info = { 0, };
+	struct node *node = data;
+	struct impl *impl = node->impl;
+	struct sm_param *p;
 
-	pw_log_debug(NAME" %p: param for node %d %d", impl, n->id, id);
+	pw_log_debug(NAME" %p: node %p", impl, node);
 
-	if (id != SPA_PARAM_EnumFormat)
-		return;
+	spa_list_for_each(p, &node->obj->param_list, link) {
+		struct spa_audio_info info = { 0, };
 
-	if (spa_format_parse(param, &n->media_type, &n->media_subtype) < 0)
-		goto error;
+		if (p->id != SPA_PARAM_EnumFormat)
+			continue;
 
-	if (n->media_type != SPA_MEDIA_TYPE_audio ||
-	    n->media_subtype != SPA_MEDIA_SUBTYPE_raw)
-		return;
+		if (spa_format_parse(p->param, &info.media_type, &info.media_subtype) < 0)
+			continue;
 
-	spa_pod_object_fixate((struct spa_pod_object*)param);
-	if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
-		spa_debug_pod(2, NULL, param);
+		if (info.media_type != SPA_MEDIA_TYPE_audio ||
+		    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
+			continue;
 
-	if (spa_format_audio_raw_parse(param, &info) < 0)
-		goto error;
+		spa_pod_object_fixate((struct spa_pod_object*)p->param);
+		if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
+			spa_debug_pod(2, NULL, p->param);
 
-	if (n->format.channels < info.channels)
-		n->format = info;
+		if (spa_format_audio_raw_parse(p->param, &info.info.raw) < 0)
+			continue;
 
-	if (n->endpoint == NULL) {
-		n->endpoint = make_endpoint(n);
+		if (node->format.info.raw.channels < info.info.raw.channels)
+			node->format = info;
 	}
-	return;
 
-      error:
-	pw_log_warn("unhandled param:");
-	if (pw_log_level_enabled(SPA_LOG_LEVEL_WARN))
-		spa_debug_pod(2, NULL, param);
-	return;
+	pw_log_debug("node %p: complete", node);
+	node->endpoint = make_endpoint(node);
 }
 
-static const struct pw_node_proxy_events node_events = {
-	PW_VERSION_NODE_PROXY_EVENTS,
-	.param = node_event_param,
-};
-
 static int
 handle_node(struct impl *impl, struct sm_object *obj)
 {
 	const char *media_class;
 	enum pw_direction direction;
 	struct node *node;
-
-	if (sm_object_get_data(obj, SESSION_KEY) != NULL)
-		return 0;
+	uint32_t subscribe[4], n_subscribe = 0;
 
 	media_class = obj->props ? pw_properties_get(obj->props, PW_KEY_MEDIA_CLASS) : NULL;
 
@@ -403,15 +399,20 @@ handle_node(struct impl *impl, struct sm_object *obj)
 	node->media = strdup(media_class);
 	pw_log_debug(NAME "%p: node %d is stream %s", impl, node->id, node->media);
 
-	pw_proxy_add_object_listener(obj->proxy, &node->listener, &node_events, node);
+	subscribe[n_subscribe++] = SPA_PARAM_EnumFormat;
+	subscribe[n_subscribe++] = SPA_PARAM_Props;
+	subscribe[n_subscribe++] = SPA_PARAM_PropInfo;
+	pw_log_debug(NAME" %p: node %p proxy %p subscribe %d params", impl,
+				node, obj->proxy, n_subscribe);
+	pw_node_proxy_subscribe_params((struct pw_node_proxy*)obj->proxy,
+				subscribe, n_subscribe);
+
+	sm_media_session_sync(impl->session, complete_stream, node);
 
-	pw_node_proxy_enum_params((struct pw_node_proxy*)obj->proxy,
-				0, SPA_PARAM_EnumFormat,
-				0, -1, NULL);
 	return 1;
 }
 
-static void session_update(void *data, struct sm_object *object)
+static void session_create(void *data, struct sm_object *object)
 {
 	struct impl *impl = data;
 	int res;
@@ -451,7 +452,7 @@ static void session_remove(void *data, struct sm_object *object)
 
 static const struct sm_media_session_events session_events = {
 	SM_VERSION_MEDIA_SESSION_EVENTS,
-	.update = session_update,
+	.create = session_create,
 	.remove = session_remove,
 };