diff --git a/spa/plugins/v4l2/v4l2-source.c b/spa/plugins/v4l2/v4l2-source.c
index bd1018fe2..c375acb90 100644
--- a/spa/plugins/v4l2/v4l2-source.c
+++ b/spa/plugins/v4l2/v4l2-source.c
@@ -32,6 +32,8 @@
 #include <spa/param/video/format-utils.h>
 #include <spa/param/param.h>
 #include <spa/pod/filter.h>
+#include <spa/control/control.h>
+#include <spa/debug/pod.h>
 
 #define NAME "v4l2-source"
 
@@ -69,7 +71,6 @@ struct control {
 	uint32_t id;
 	uint32_t ctrl_id;
 	double value;
-	double *io;
 };
 
 struct port {
@@ -110,6 +111,7 @@ struct port {
 	struct spa_port_info info;
 	struct spa_io_buffers *io;
 	struct spa_io_clock *clock;
+	struct spa_io_sequence *control;
 };
 
 struct impl {
@@ -509,11 +511,6 @@ static int impl_node_port_enum_params(struct spa_node *node,
 			return 0;
 		}
 		break;
-#if 0
-	else if (id == t->param_io.idPropsIn) {
-		return spa_v4l2_enum_controls(this, index, filter, result, builder);
-	}
-#endif
 	case SPA_PARAM_IO:
 		switch (*index) {
 		case 0:
@@ -528,6 +525,12 @@ static int impl_node_port_enum_params(struct spa_node *node,
 				":", SPA_PARAM_IO_id,   "I", SPA_IO_Clock,
 				":", SPA_PARAM_IO_size, "i", sizeof(struct spa_io_clock));
 			break;
+		case 2:
+			param = spa_pod_builder_object(&b,
+				SPA_TYPE_OBJECT_ParamIO, id,
+				":", SPA_PARAM_IO_id,   "I", SPA_IO_Control,
+				":", SPA_PARAM_IO_size, "i", sizeof(struct spa_io_sequence));
+			break;
 		default:
 			return 0;
 		}
@@ -703,6 +706,7 @@ impl_node_port_alloc_buffers(struct spa_node *node,
 	return res;
 }
 
+#if 0
 static struct control *find_control(struct port *port, uint32_t id)
 {
 	int i;
@@ -713,6 +717,7 @@ static struct control *find_control(struct port *port, uint32_t id)
 	}
 	return NULL;
 }
+#endif
 
 static int impl_node_port_set_io(struct spa_node *node,
 				 enum spa_direction direction,
@@ -722,7 +727,6 @@ static int impl_node_port_set_io(struct spa_node *node,
 {
 	struct impl *this;
 	struct port *port;
-	struct control *control;
 
 	spa_return_val_if_fail(node != NULL, -EINVAL);
 
@@ -732,21 +736,19 @@ static int impl_node_port_set_io(struct spa_node *node,
 
 	port = GET_PORT(this, direction, port_id);
 
-	if (id == SPA_IO_Buffers) {
+	switch (id) {
+	case SPA_IO_Buffers:
 		port->io = data;
-	}
-	else if (id == SPA_IO_Clock) {
+		break;
+	case SPA_IO_Clock:
 		port->clock = data;
-	}
-	else if ((control = find_control(port, id))) {
-		if (data && size >= sizeof(struct spa_pod_double))
-			control->io = &SPA_POD_VALUE(struct spa_pod_double, data);
-		else
-			control->io = &control->value;
-	}
-	else
+		break;
+	case SPA_IO_Control:
+		port->control = data;
+		break;
+	default:
 		return -ENOENT;
-
+	}
 	return 0;
 }
 
@@ -779,10 +781,69 @@ static int impl_node_port_send_command(struct spa_node *node,
 	return -ENOTSUP;
 }
 
+static uint32_t prop_to_control_id(uint32_t prop)
+{
+	switch (prop) {
+	case SPA_PROP_brightness:
+		return V4L2_CID_BRIGHTNESS;
+	case SPA_PROP_contrast:
+		return V4L2_CID_CONTRAST;
+	case SPA_PROP_saturation:
+		return V4L2_CID_SATURATION;
+	case SPA_PROP_hue:
+		return V4L2_CID_HUE;
+	case SPA_PROP_gamma:
+		return V4L2_CID_GAMMA;
+	case SPA_PROP_exposure:
+		return V4L2_CID_EXPOSURE;
+	case SPA_PROP_gain:
+		return V4L2_CID_GAIN;
+	case SPA_PROP_sharpness:
+		return V4L2_CID_SHARPNESS;
+	default:
+		return 0;
+	}
+}
+
+static int process_control(struct impl *this, struct port *port, struct spa_pod_sequence *control)
+{
+	struct spa_pod_control *c;
+
+	SPA_POD_SEQUENCE_FOREACH(control, c) {
+		switch (c->type) {
+		case SPA_CONTROL_Properties:
+		{
+			struct spa_pod *pod;
+			struct spa_pod_object *obj = (struct spa_pod_object *) &c->value;
+
+			SPA_POD_OBJECT_FOREACH(obj, pod) {
+				struct spa_pod_prop *prop = (struct spa_pod_prop *)pod;
+				struct v4l2_control c;
+				uint32_t control_id;
+
+				if ((control_id = prop_to_control_id(prop->body.key)) == 0)
+					continue;
+
+				memset (&c, 0, sizeof (c));
+				c.id = control_id;
+				c.value = SPA_POD_VALUE(struct spa_pod_float, &prop->body.value);
+
+				if (ioctl(port->fd, VIDIOC_S_CTRL, &c) < 0)
+					spa_log_error(port->log, "VIDIOC_S_CTRL %m");
+			}
+			break;
+		}
+		default:
+			break;
+		}
+	}
+	return 0;
+}
+
 static int impl_node_process(struct spa_node *node)
 {
 	struct impl *this;
-	int i, res;
+	int res;
 	struct spa_io_buffers *io;
 	struct port *port;
 	struct buffer *b;
@@ -797,6 +858,9 @@ static int impl_node_process(struct spa_node *node)
 
 	spa_log_trace(port->log, NAME " %p; status %d", node, io->status);
 
+	if (port->control)
+		process_control(this, port, &port->control->sequence);
+
 	if (io->status == SPA_STATUS_HAVE_BUFFER)
 		return SPA_STATUS_HAVE_BUFFER;
 
@@ -806,25 +870,7 @@ static int impl_node_process(struct spa_node *node)
 
 		io->buffer_id = SPA_ID_INVALID;
 	}
-	for (i = 0; i < port->n_controls; i++) {
-		struct control *control = &port->controls[i];
 
-		if (control->io == NULL)
-			continue;
-
-		if (control->value != *control->io) {
-			struct v4l2_control c;
-
-			memset (&c, 0, sizeof (c));
-			c.id = control->ctrl_id;
-			c.value = *control->io;
-
-			if (ioctl(port->fd, VIDIOC_S_CTRL, &c) < 0)
-				spa_log_error(port->log, "VIDIOC_S_CTRL %m");
-
-			control->value = *control->io = c.value;
-		}
-	}
 	if (spa_list_is_empty(&port->queue))
 		return -EPIPE;
 
diff --git a/src/examples/export-sink.c b/src/examples/export-sink.c
index d7a186e89..ed56b9e10 100644
--- a/src/examples/export-sink.c
+++ b/src/examples/export-sink.c
@@ -24,7 +24,9 @@
 #include <spa/param/video/format-utils.h>
 #include <spa/param/props.h>
 #include <spa/node/io.h>
+#include <spa/control/control.h>
 #include <spa/debug/format.h>
+#include <spa/debug/pod.h>
 
 #include <pipewire/pipewire.h>
 #include <pipewire/module.h>
@@ -74,7 +76,8 @@ struct data {
 	const struct spa_node_callbacks *callbacks;
 	void *callbacks_data;
 	struct spa_io_buffers *io;
-	struct spa_pod_double *ctrl_param;
+	struct spa_io_sequence *io_notify;
+	uint32_t io_notify_size;
 	double param_accum;
 
 	uint8_t buffer[1024];
@@ -103,10 +106,17 @@ static void handle_events(struct data *data)
 
 static void update_param(struct data *data)
 {
-	if (data->ctrl_param == NULL)
+	struct spa_pod_builder b = { 0, };
+
+	if (data->io_notify == NULL)
 		return;
 
-        data->ctrl_param->value = (sin(data->param_accum) * 127.0) + 127.0;
+	spa_pod_builder_init(&b, data->io_notify, data->io_notify_size);
+	spa_pod_builder_sequence(&b, 0,
+	".", 0, SPA_CONTROL_Properties,
+		SPA_POD_OBJECT(SPA_TYPE_OBJECT_Props, 0,
+		":", SPA_PROP_contrast, "f", (sin(data->param_accum) * 127.0) + 127.0));
+
         data->param_accum += M_PI_M2 / 30.0;
         if (data->param_accum >= M_PI_M2)
                 data->param_accum -= M_PI_M2;
@@ -154,19 +164,17 @@ static int impl_port_set_io(struct spa_node *node,
 {
 	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
 
-	if (id == SPA_IO_Buffers)
+	switch (id) {
+	case SPA_IO_Buffers:
 		d->io = data;
-#if 0
-	else if (id == d->type.io_prop_param) {
-		if (data && size >= sizeof(struct spa_pod_double))
-			d->ctrl_param = data;
-		else
-			d->ctrl_param = NULL;
-	}
-#endif
-	else
+		break;
+	case SPA_IO_Notify:
+		d->io_notify = data;
+		d->io_notify_size = size;
+		break;
+	default:
 		return -ENOENT;
-
+	}
 	return 0;
 }
 
@@ -273,25 +281,25 @@ static int impl_port_enum_params(struct spa_node *node,
 			return 0;
 		}
 		break;
-#if 0
-	else if (id == t->param_io.idPropsOut) {
-		struct props *p = &d->props;
 
+	case SPA_PARAM_IO:
 		switch (*index) {
 		case 0:
 			param = spa_pod_builder_object(builder,
-				id, t->param_io.Prop,
-				":", t->param_io.id, "I", d->type.io_prop_param,
-				":", t->param_io.size, "i", sizeof(struct spa_pod_double),
-				":", t->param.propId, "I", d->type.prop_param,
-				":", t->param.propType, "dru", p->param,
-					SPA_POD_PROP_MIN_MAX(0.0, 10.0));
+				SPA_TYPE_OBJECT_ParamIO, id,
+				":", SPA_PARAM_IO_id,   "I", SPA_IO_Buffers,
+				":", SPA_PARAM_IO_size, "i", sizeof(struct spa_io_buffers));
+			break;
+		case 1:
+			param = spa_pod_builder_object(builder,
+				SPA_TYPE_OBJECT_ParamIO, id,
+				":", SPA_PARAM_IO_id,   "I", SPA_IO_Notify,
+				":", SPA_PARAM_IO_size, "i", sizeof(struct spa_io_sequence) + 1024);
 			break;
 		default:
 			return 0;
 		}
-	}
-#endif
+		break;
 	default:
 		return -ENOENT;
 	}
diff --git a/src/modules/module-client-node/client-stream.c b/src/modules/module-client-node/client-stream.c
index 7f5e9d9fb..8a7edc068 100644
--- a/src/modules/module-client-node/client-stream.c
+++ b/src/modules/module-client-node/client-stream.c
@@ -559,7 +559,7 @@ static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw
 	pw_log_debug("module %p: trying controls", impl);
 	spa_list_for_each(cout, &port->control_list[SPA_DIRECTION_OUTPUT], port_link) {
 		spa_list_for_each(cin, &target->control_list[SPA_DIRECTION_INPUT], port_link) {
-			if (cin->prop_id == cout->prop_id) {
+			if (cin->id != cout->id) {
 				if ((res = pw_control_link(cout, cin)) < 0)
 					pw_log_error("failed to link controls: %s", spa_strerror(res));
 			}
@@ -567,7 +567,7 @@ static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw
 	}
 	spa_list_for_each(cin, &port->control_list[SPA_DIRECTION_INPUT], port_link) {
 		spa_list_for_each(cout, &target->control_list[SPA_DIRECTION_OUTPUT], port_link) {
-			if (cin->prop_id == cout->prop_id) {
+			if (cin->id != cout->id) {
 				if ((res = pw_control_link(cout, cin)) < 0)
 					pw_log_error("failed to link controls: %s", spa_strerror(res));
 			}
diff --git a/src/modules/module-media-session.c b/src/modules/module-media-session.c
index 4fe921d26..2666cc03f 100644
--- a/src/modules/module-media-session.c
+++ b/src/modules/module-media-session.c
@@ -212,7 +212,7 @@ static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw
 	pw_log_debug("module %p: trying controls", impl);
 	spa_list_for_each(cout, &port->control_list[SPA_DIRECTION_OUTPUT], port_link) {
 		spa_list_for_each(cin, &target->control_list[SPA_DIRECTION_INPUT], port_link) {
-			if (cin->prop_id == cout->prop_id) {
+			if (cin->id != cout->id) {
 				if ((res = pw_control_link(cout, cin)) < 0)
 					pw_log_error("failed to link controls: %s", spa_strerror(res));
 			}
@@ -220,7 +220,7 @@ static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw
 	}
 	spa_list_for_each(cin, &port->control_list[SPA_DIRECTION_INPUT], port_link) {
 		spa_list_for_each(cout, &target->control_list[SPA_DIRECTION_OUTPUT], port_link) {
-			if (cin->prop_id == cout->prop_id) {
+			if (cin->id != cout->id) {
 				if ((res = pw_control_link(cout, cin)) < 0)
 					pw_log_error("failed to link controls: %s", spa_strerror(res));
 			}
diff --git a/src/pipewire/control.c b/src/pipewire/control.c
index 6fd4a81df..c94d31407 100644
--- a/src/pipewire/control.c
+++ b/src/pipewire/control.c
@@ -38,27 +38,34 @@ pw_control_new(struct pw_core *core,
 	struct impl *impl;
 	struct pw_control *this;
 	enum spa_direction direction;
+	uint32_t id, size;
+
+	if (spa_pod_object_parse(param,
+		":", SPA_PARAM_IO_id,  "I", &id,
+		":", SPA_PARAM_IO_size, "i", &size) < 0)
+		goto exit;
+
+	switch (id) {
+	case SPA_IO_Control:
+		direction = SPA_DIRECTION_INPUT;
+		break;
+	case SPA_IO_Notify:
+		direction = SPA_DIRECTION_OUTPUT;
+		break;
+	default:
+		goto exit;
+	}
 
 	impl = calloc(1, sizeof(struct impl) + user_data_size);
 	if (impl == NULL)
 		goto exit;
 
 	this = &impl->this;
-
-	direction = SPA_DIRECTION_OUTPUT;
-#if 0
-	direction = spa_pod_is_object_id(param, SPA_PARAM_PropsOut) ?
-		SPA_DIRECTION_OUTPUT : SPA_DIRECTION_INPUT;
-
-	if (spa_pod_object_parse(param,
-				":", t->param_io.id, "I", &this->id,
-				":", t->param_io.size, "i", &this->size,
-				":", t->param.propId, "I", &this->prop_id) < 0)
-		goto exit_free;
-#endif
+	this->id = id;
+	this->size = size;
 
 	pw_log_debug("control %p: new %s %d", this,
-			spa_debug_type_find_name(spa_debug_types, this->prop_id), direction);
+			spa_debug_type_find_name(spa_types, this->id), direction);
 
 	this->core = core;
 	this->port = port;
@@ -80,8 +87,6 @@ pw_control_new(struct pw_core *core,
 
 	return this;
 
-//    exit_free:
-//	free(impl);
     exit:
 	return NULL;
 }
@@ -141,6 +146,7 @@ int pw_control_link(struct pw_control *control, struct pw_control *other)
 {
 	int res = 0;
 	struct impl *impl;
+	uint32_t size;
 
 	if (control->direction == SPA_DIRECTION_INPUT) {
 		struct pw_control *tmp = control;
@@ -158,13 +164,15 @@ int pw_control_link(struct pw_control *control, struct pw_control *other)
 	impl = SPA_CONTAINER_OF(control, struct impl, this);
 
 	pw_log_debug("control %p: link to %p %s", control, other,
-			spa_debug_type_find_name(spa_debug_types, control->prop_id));
+			spa_debug_type_find_name(spa_debug_types, control->id));
+
+	size = SPA_MAX(control->size, other->size);
 
 	if (impl->mem == NULL) {
 		if ((res = pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
 					     PW_MEMBLOCK_FLAG_SEAL |
 					     PW_MEMBLOCK_FLAG_MAP_READWRITE,
-					     control->size,
+					     size,
 					     &impl->mem)) < 0)
 			goto exit;
 
@@ -175,7 +183,7 @@ int pw_control_link(struct pw_control *control, struct pw_control *other)
 		if ((res = spa_node_port_set_io(port->node->node,
 				     port->direction, port->port_id,
 				     other->id,
-				     impl->mem->ptr, control->size)) < 0) {
+				     impl->mem->ptr, size)) < 0) {
 			pw_log_warn("control %p: set io failed %d %s", control,
 					res, spa_strerror(res));
 			goto exit;
@@ -188,7 +196,7 @@ int pw_control_link(struct pw_control *control, struct pw_control *other)
 			if ((res = spa_node_port_set_io(port->node->node,
 					     port->direction, port->port_id,
 					     control->id,
-					     impl->mem->ptr, control->size)) < 0) {
+					     impl->mem->ptr, size)) < 0) {
 				pw_log_warn("control %p: set io failed %d %s", control,
 					res, spa_strerror(res));
 				/* undo */
diff --git a/src/pipewire/port.c b/src/pipewire/port.c
index 2b0cf8713..3da4c66f3 100644
--- a/src/pipewire/port.c
+++ b/src/pipewire/port.c
@@ -341,7 +341,6 @@ static int do_add_port(struct spa_loop *loop,
 	return 0;
 }
 
-#if 0
 static int make_control(void *data, uint32_t id, uint32_t index, uint32_t next, struct spa_pod *param)
 {
 	struct pw_port *port = data;
@@ -349,7 +348,6 @@ static int make_control(void *data, uint32_t id, uint32_t index, uint32_t next,
 	pw_control_new(node->core, port, param, 0);
 	return 0;
 }
-#endif
 
 static void port_unbind_func(void *data)
 {
@@ -526,10 +524,7 @@ int pw_port_add(struct pw_port *port, struct pw_node *node)
 		node->info.change_mask |= PW_NODE_CHANGE_MASK_OUTPUT_PORTS;
 	}
 
-#if 0
-	pw_port_for_each_param(port, SPA_PARAM_PropsOut, 0, 0, NULL, make_control, port);
-	pw_port_for_each_param(port, SPA_PARAM_PropsIn, 0, 0, NULL, make_control, port);
-#endif
+	pw_port_for_each_param(port, SPA_PARAM_IO, 0, 0, NULL, make_control, port);
 
 	pw_log_debug("port %p: setting node io", port);
 	spa_node_port_set_io(node->node,
diff --git a/src/pipewire/private.h b/src/pipewire/private.h
index 3ddc7fbd4..0bf166636 100644
--- a/src/pipewire/private.h
+++ b/src/pipewire/private.h
@@ -627,7 +627,6 @@ struct pw_control {
 	struct spa_list inputs_link;	/**< link in linked input control */
 
 	uint32_t id;
-	uint32_t prop_id;
 	int32_t size;
 
 	struct spa_hook_list listener_list;