diff --git a/spa/include/spa/param/io.h b/spa/include/spa/param/io.h
index 79ef56130..f33ecd6b9 100644
--- a/spa/include/spa/param/io.h
+++ b/spa/include/spa/param/io.h
@@ -32,6 +32,8 @@ extern "C" {
 #define SPA_TYPE_PARAM_IO__id		SPA_TYPE_PARAM_IO_BASE "id"
 /** size of the io area for a port */
 #define SPA_TYPE_PARAM_IO__size		SPA_TYPE_PARAM_IO_BASE "size"
+/** type of the io area for a port, this is an enum spa_pod_type */
+#define SPA_TYPE_PARAM_IO__type		SPA_TYPE_PARAM_IO_BASE "type"
 
 /** enumerate buffer io areas */
 #define SPA_TYPE_PARAM_ID_IO__Buffers	SPA_TYPE_PARAM_ID_IO_BASE "Buffers"
@@ -64,6 +66,7 @@ extern "C" {
 struct spa_type_param_io {
 	uint32_t id;		/**< id to configure the io area */
 	uint32_t size;		/**< size of io area */
+	uint32_t type;		/**< type of io area */
 	uint32_t idBuffers;	/**< id to enumerate buffer io */
 	uint32_t Buffers;	/**< object type of buffer io area */
 	uint32_t idControl;	/**< id to enumerate control io */
@@ -82,6 +85,7 @@ spa_type_param_io_map(struct spa_type_map *map,
 	if (type->id == 0) {
 		type->id = spa_type_map_get_id(map, SPA_TYPE_PARAM_IO__id);
 		type->size = spa_type_map_get_id(map, SPA_TYPE_PARAM_IO__size);
+		type->type = spa_type_map_get_id(map, SPA_TYPE_PARAM_IO__type);
 		type->idBuffers = spa_type_map_get_id(map, SPA_TYPE_PARAM_ID_IO__Buffers);
 		type->Buffers = spa_type_map_get_id(map, SPA_TYPE_PARAM_IO__Buffers);
 		type->idControl = spa_type_map_get_id(map, SPA_TYPE_PARAM_ID_IO__Control);
diff --git a/spa/plugins/alsa/alsa-sink.c b/spa/plugins/alsa/alsa-sink.c
index 4ada603b1..44fb8a485 100644
--- a/spa/plugins/alsa/alsa-sink.c
+++ b/spa/plugins/alsa/alsa-sink.c
@@ -703,6 +703,8 @@ impl_init(const struct spa_handle_factory *factory,
 	this->stream = SND_PCM_STREAM_PLAYBACK;
 	reset_props(&this->props);
 
+	this->info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
+
 	spa_list_init(&this->ready);
 
 	for (i = 0; info && i < info->n_items; i++) {
diff --git a/spa/plugins/alsa/alsa-source.c b/spa/plugins/alsa/alsa-source.c
index 9d683b171..a5395f021 100644
--- a/spa/plugins/alsa/alsa-source.c
+++ b/spa/plugins/alsa/alsa-source.c
@@ -793,6 +793,8 @@ impl_init(const struct spa_handle_factory *factory,
 	this->stream = SND_PCM_STREAM_CAPTURE;
 	reset_props(&this->props);
 
+	this->info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
+
 	spa_list_init(&this->free);
 	spa_list_init(&this->ready);
 
diff --git a/spa/plugins/audiomixer/audiomixer.c b/spa/plugins/audiomixer/audiomixer.c
index e93ae6614..9523180df 100644
--- a/spa/plugins/audiomixer/audiomixer.c
+++ b/spa/plugins/audiomixer/audiomixer.c
@@ -535,7 +535,7 @@ impl_node_port_enum_params(struct spa_node *node,
 				":", t->param_io.id, "I", t->io_prop_volume,
 				":", t->param_io.size, "i", sizeof(struct spa_pod_double),
 				":", t->param_io.propId, "I", t->prop_volume,
-				":", t->param_io.propType, "dr", p->volume, 2, 0.0, 10.0);
+				":", t->param_io.propType, "dru", p->volume, 2, 0.0, 10.0);
 			break;
 		case 1:
 			param = spa_pod_builder_object(&b,
@@ -747,15 +747,9 @@ impl_node_port_set_io(struct spa_node *node,
 	else if (id == t->io.ControlRange)
 		port->io_range = data;
 	else if (id == t->io_prop_volume && direction == SPA_DIRECTION_INPUT)
-		if (SPA_POD_TYPE(data) == SPA_POD_TYPE_DOUBLE)
-			port->io_volume = &SPA_POD_VALUE(struct spa_pod_double, data);
-		else
-			return -EINVAL;
+		port->io_volume = &SPA_POD_VALUE(struct spa_pod_double, data);
 	else if (id == t->io_prop_mute && direction == SPA_DIRECTION_INPUT)
-		if (SPA_POD_TYPE(data) == SPA_POD_TYPE_BOOL)
-			port->io_mute = &SPA_POD_VALUE(struct spa_pod_bool, data);
-		else
-			return -EINVAL;
+		port->io_mute = &SPA_POD_VALUE(struct spa_pod_bool, data);
 	else
 		return -ENOENT;
 
diff --git a/src/examples/export-source.c b/src/examples/export-source.c
index 0d580525d..8eebcbd2f 100644
--- a/src/examples/export-source.c
+++ b/src/examples/export-source.c
@@ -101,7 +101,7 @@ struct data {
 	void *callbacks_data;
 	struct spa_io_buffers *io;
 
-	double *io_volume;
+	struct spa_pod_double *ctrl_volume;
 
 	uint8_t buffer[1024];
 
@@ -112,8 +112,20 @@ struct data {
 	struct spa_list empty;
 
 	double accumulator;
+	double volume_accum;
 };
 
+static void update_volume(struct data *data)
+{
+	if (data->ctrl_volume == NULL)
+		return;
+
+        data->ctrl_volume->value = (sin(data->volume_accum) / 2.0) + 0.5;
+        data->volume_accum += M_PI_M2 / 1000.0;
+        if (data->volume_accum >= M_PI_M2)
+                data->volume_accum -= M_PI_M2;
+}
+
 static int impl_send_command(struct spa_node *node, const struct spa_command *command)
 {
 	return 0;
@@ -158,10 +170,8 @@ static int impl_port_set_io(struct spa_node *node, enum spa_direction direction,
 	if (id == d->t->io.Buffers)
 		d->io = data;
 	else if (id == d->type.io_prop_volume) {
-		if (SPA_POD_TYPE(data) == SPA_POD_TYPE_ID)
-			d->io_volume = &SPA_POD_VALUE(struct spa_pod_double, data);
-		else
-			return -EINVAL;
+		d->ctrl_volume = data;
+		*d->ctrl_volume = SPA_POD_DOUBLE_INIT(1.0);
 	}
 	else
 		return -ENOENT;
@@ -315,7 +325,7 @@ static int impl_port_enum_params(struct spa_node *node,
 				":", t->param_io.id, "I", d->type.io_prop_volume,
 				":", t->param_io.size, "i", sizeof(struct spa_pod_double),
 				":", t->param_io.propId, "I", d->type.prop_volume,
-				":", t->param_io.propType, "dr", p->volume, 2, 0.0, 10.0);
+				":", t->param_io.propType, "dru", p->volume, 2, 0.0, 10.0);
 			break;
 		default:
 			return 0;
@@ -406,7 +416,7 @@ static int impl_port_use_buffers(struct spa_node *node, enum spa_direction direc
 
 static inline void reuse_buffer(struct data *d, uint32_t id)
 {
-	pw_log_trace("sine-source %p: recycle buffer %d", d, id);
+	pw_log_trace("export-source %p: recycle buffer %d", d, id);
         spa_list_append(&d->empty, &d->buffers[id].link);
 }
 
@@ -433,7 +443,7 @@ static int impl_node_process_output(struct spa_node *node)
 		io->buffer_id = SPA_ID_INVALID;
 	}
 	if (spa_list_is_empty(&d->empty)) {
-                pw_log_error("sine-source %p: out of buffers", d);
+                pw_log_error("export-source %p: out of buffers", d);
                 return -EPIPE;
         }
         b = spa_list_first(&d->empty, struct buffer, link);
@@ -474,6 +484,8 @@ static int impl_node_process_output(struct spa_node *node)
 	io->buffer_id = b->buffer->id;
 	io->status = SPA_STATUS_HAVE_BUFFER;
 
+	update_volume(d);
+
 	return SPA_STATUS_HAVE_BUFFER;
 }
 
@@ -500,7 +512,7 @@ static void make_node(struct data *data)
 	if (data->path)
 		pw_properties_set(props, PW_NODE_PROP_TARGET_NODE, data->path);
 
-	data->node = pw_node_new(data->core, "sine-source", props, 0);
+	data->node = pw_node_new(data->core, "export-source", props, 0);
 	data->impl_node = impl_node;
 	pw_node_set_implementation(data->node, &data->impl_node);
 
@@ -551,7 +563,6 @@ int main(int argc, char *argv[])
 	spa_list_init(&data.empty);
 	init_type(&data.type, data.t->map);
 	reset_props(&data.props);
-	data.io_volume = &data.props.volume;
 	spa_debug_set_type_map(data.t->map);
 
 	pw_remote_add_listener(data.remote, &data.remote_listener, &remote_events, &data);
diff --git a/src/extensions/client-node.h b/src/extensions/client-node.h
index 9e35241c6..7b136ea6c 100644
--- a/src/extensions/client-node.h
+++ b/src/extensions/client-node.h
@@ -298,7 +298,8 @@ pw_client_node_proxy_destroy(struct pw_client_node_proxy *p)
 #define PW_CLIENT_NODE_PROXY_EVENT_PORT_ADD_MEM		7
 #define PW_CLIENT_NODE_PROXY_EVENT_PORT_USE_BUFFERS	8
 #define PW_CLIENT_NODE_PROXY_EVENT_PORT_COMMAND		9
-#define PW_CLIENT_NODE_PROXY_EVENT_NUM			10
+#define PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_IO		10
+#define PW_CLIENT_NODE_PROXY_EVENT_NUM			11
 
 /** \ref pw_client_node events */
 struct pw_client_node_proxy_events {
@@ -433,6 +434,26 @@ struct pw_client_node_proxy_events {
 			      enum spa_direction direction,
 			      uint32_t port_id,
 			      const struct spa_command *command);
+
+	/**
+	 * Configure the io area with \a id of \a port_id.
+	 *
+	 * \param seq a sequence number
+	 * \param direction the direction of the port
+	 * \param port_id the port id
+	 * \param id the id of the io area to set
+	 * \param mem_id the id of the memory to use
+	 * \param offset offset of io area in memory
+	 * \param size size of the io area
+	 */
+	void (*port_set_io) (void *object,
+			     uint32_t seq,
+			     enum spa_direction direction,
+			     uint32_t port_id,
+			     uint32_t id,
+			     uint32_t mem_id,
+			     uint32_t offset,
+			     uint32_t size);
 };
 
 static inline void
@@ -464,6 +485,8 @@ pw_client_node_proxy_add_listener(struct pw_client_node_proxy *p,
 	pw_resource_notify(r,struct pw_client_node_proxy_events,port_use_buffers,__VA_ARGS__)
 #define pw_client_node_resource_port_command(r,...)	\
 	pw_resource_notify(r,struct pw_client_node_proxy_events,port_command,__VA_ARGS__)
+#define pw_client_node_resource_port_set_io(r,...)	\
+	pw_resource_notify(r,struct pw_client_node_proxy_events,port_set_io,__VA_ARGS__)
 
 #ifdef __cplusplus
 }  /* extern "C" */
diff --git a/src/modules/module-autolink.c b/src/modules/module-autolink.c
index 38ba2959a..e5eb800f8 100644
--- a/src/modules/module-autolink.c
+++ b/src/modules/module-autolink.c
@@ -28,6 +28,7 @@
 #include "pipewire/link.h"
 #include "pipewire/log.h"
 #include "pipewire/module.h"
+#include "pipewire/control.h"
 #include "pipewire/private.h"
 
 struct impl {
@@ -142,6 +143,23 @@ link_state_changed(void *data, enum pw_link_state old, enum pw_link_state state,
 	}
 }
 
+static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw_port *target)
+{
+	struct pw_control *cin, *cout;
+	int res;
+
+	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) {
+			pw_log_debug("controls %d <-> %d", cin->id, cout->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));
+			}
+		}
+	}
+
+}
+
 static void
 link_destroy(void *data)
 {
@@ -210,6 +228,8 @@ static void try_link_port(struct pw_node *node, struct pw_port *port, struct nod
 	spa_list_append(&info->links, &ld->l);
 	pw_link_register(link, NULL, pw_module_get_global(impl->module));
 
+	try_link_controls(impl, port, target);
+
 	return;
 
       error:
diff --git a/src/modules/module-client-node/client-node.c b/src/modules/module-client-node/client-node.c
index 41208dd30..4384b332c 100644
--- a/src/modules/module-client-node/client-node.c
+++ b/src/modules/module-client-node/client-node.c
@@ -113,7 +113,7 @@ struct proxy {
 	uint32_t n_params;
 	struct spa_pod **params;
 
-	uint8_t format_buffer[1024];
+	uint32_t membase;
 	uint32_t seq;
 };
 
@@ -337,7 +337,6 @@ do_update_port(struct proxy *this,
 			if (spa_pod_is_object_id(port->params[i], t->param.idFormat))
 				port->have_format = true;
 		}
-
 	}
 
 	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_INFO && info)
@@ -534,9 +533,27 @@ spa_proxy_node_port_set_io(struct spa_node *node,
 
 	if (id == t->io.Buffers)
 		port->io = data;
-	else
-		return -ENOENT;
+	else {
+		struct pw_memblock *mem;
+		uint32_t memid = this->membase++;
 
+		if ((mem = pw_memblock_find(data)) == NULL)
+			return -EINVAL;
+
+		pw_client_node_resource_port_add_mem(this->resource,
+						     direction, port_id,
+						     memid,
+						     t->data.MemFd,
+						     mem->fd, mem->flags,
+						     0, mem->offset + mem->size);
+
+		pw_client_node_resource_port_set_io(this->resource,
+						    this->seq,
+						    direction, port_id,
+						    id,
+						    memid,
+						    mem->offset, mem->size);
+	}
 	return 0;
 }
 
@@ -582,7 +599,7 @@ spa_proxy_node_port_use_buffers(struct spa_node *node,
 	if (this->resource == NULL)
 		return 0;
 
-	n_mem = 0;
+	n_mem = this->membase;
 	for (i = 0; i < n_buffers; i++) {
 		struct buffer *b = &port->buffers[i];
 		struct pw_memblock *m;
@@ -1109,10 +1126,13 @@ static void node_initialized(void *data)
 	impl->other_fds[1] = impl->fds[0];
 
 	spa_loop_add_source(impl->proxy.data_loop, &impl->proxy.data_source);
-	pw_log_debug("client-node %p: add data fd %d", node, impl->proxy.data_source.fd);
+	pw_log_debug("client-node %p: transport fd %d %d", node, impl->fds[0], impl->fds[1]);
 
-	pw_client_node_resource_transport(this->resource, pw_global_get_id(pw_node_get_global(node)),
-					  impl->other_fds[0], impl->other_fds[1], impl->transport);
+	pw_client_node_resource_transport(this->resource,
+					  pw_global_get_id(pw_node_get_global(node)),
+					  impl->other_fds[0],
+					  impl->other_fds[1],
+					  impl->transport);
 }
 
 static void node_free(void *data)
diff --git a/src/modules/module-client-node/protocol-native.c b/src/modules/module-client-node/protocol-native.c
index 5013b68be..b83553f3e 100644
--- a/src/modules/module-client-node/protocol-native.c
+++ b/src/modules/module-client-node/protocol-native.c
@@ -408,6 +408,32 @@ static bool client_node_demarshal_port_command(void *object, void *data, size_t
 	return true;
 }
 
+static bool client_node_demarshal_port_set_io(void *object, void *data, size_t size)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_parser prs;
+	uint32_t seq, direction, port_id, id, memid, off, sz;
+
+	spa_pod_parser_init(&prs, data, size, 0);
+	if (spa_pod_parser_get(&prs,
+			"["
+			"i", &seq,
+			"i", &direction,
+			"i", &port_id,
+			"I", &id,
+			"i", &memid,
+			"i", &off,
+			"i", &sz, NULL) < 0)
+		return false;
+
+	pw_proxy_notify(proxy, struct pw_client_node_proxy_events, port_set_io,
+							seq,
+							direction, port_id,
+							id, memid,
+							off, sz);
+	return true;
+}
+
 static void client_node_marshal_transport(void *object, uint32_t node_id, int readfd, int writefd,
 					  struct pw_client_node_transport *transport)
 {
@@ -629,6 +655,33 @@ client_node_marshal_port_command(void *object,
 	pw_protocol_native_end_resource(resource, b);
 }
 
+static void
+client_node_marshal_port_set_io(void *object,
+				uint32_t seq,
+				uint32_t direction,
+				uint32_t port_id,
+				uint32_t id,
+				uint32_t memid,
+				uint32_t offset,
+				uint32_t size)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_IO);
+
+	spa_pod_builder_struct(b,
+			       "i", seq,
+			       "i", direction,
+			       "i", port_id,
+			       "I", id,
+			       "i", memid,
+			       "i", offset,
+			       "i", size);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
 
 static bool client_node_demarshal_done(void *object, void *data, size_t size)
 {
@@ -683,7 +736,7 @@ static bool client_node_demarshal_port_update(void *object, void *data, size_t s
 	struct spa_pod_parser prs;
 	uint32_t i, direction, port_id, change_mask, n_params;
 	const struct spa_pod **params = NULL;
-	struct spa_port_info info, *infop = NULL;
+	struct spa_port_info info = { 0 }, *infop = NULL;
 	struct spa_pod *ipod;
 
 	spa_pod_parser_init(&prs, data, size, 0);
@@ -799,6 +852,7 @@ static const struct pw_client_node_proxy_events pw_protocol_native_client_node_e
 	&client_node_marshal_port_add_mem,
 	&client_node_marshal_port_use_buffers,
 	&client_node_marshal_port_command,
+	&client_node_marshal_port_set_io,
 };
 
 static const struct pw_protocol_native_demarshal pw_protocol_native_client_node_event_demarshal[] = {
@@ -812,6 +866,7 @@ static const struct pw_protocol_native_demarshal pw_protocol_native_client_node_
 	{ &client_node_demarshal_port_add_mem, PW_PROTOCOL_NATIVE_REMAP },
 	{ &client_node_demarshal_port_use_buffers, PW_PROTOCOL_NATIVE_REMAP },
 	{ &client_node_demarshal_port_command, PW_PROTOCOL_NATIVE_REMAP },
+	{ &client_node_demarshal_port_set_io, PW_PROTOCOL_NATIVE_REMAP },
 };
 
 const struct pw_protocol_marshal pw_protocol_native_client_node_marshal = {
diff --git a/src/modules/module-client-node/transport.c b/src/modules/module-client-node/transport.c
index 21add9bdc..32cce5dbe 100644
--- a/src/modules/module-client-node/transport.c
+++ b/src/modules/module-client-node/transport.c
@@ -195,12 +195,15 @@ pw_client_node_transport_new(uint32_t max_input_ports, uint32_t max_output_ports
 	if (impl == NULL)
 		return NULL;
 
+	pw_log_debug("transport %p: new %d %d", impl, max_input_ports, max_output_ports);
+
 	trans = &impl->trans;
 	impl->offset = 0;
 
 	if (pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
 			  PW_MEMBLOCK_FLAG_MAP_READWRITE |
-			  PW_MEMBLOCK_FLAG_SEAL, area_get_size(&area),
+			  PW_MEMBLOCK_FLAG_SEAL,
+			  area_get_size(&area),
 			  &impl->mem) < 0)
 		return NULL;
 
@@ -229,6 +232,7 @@ pw_client_node_transport_new_from_info(struct pw_client_node_transport_info *inf
 		return NULL;
 
 	trans = &impl->trans;
+	pw_log_debug("transport %p: new from info", impl);
 
 	if ((res = pw_memblock_import(PW_MEMBLOCK_FLAG_MAP_READWRITE |
 				      PW_MEMBLOCK_FLAG_WITH_FD,
diff --git a/src/pipewire/control.c b/src/pipewire/control.c
new file mode 100644
index 000000000..45e7def4f
--- /dev/null
+++ b/src/pipewire/control.c
@@ -0,0 +1,178 @@
+/* PipeWire
+ * Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <spa/pod/parser.h>
+
+#include <pipewire/control.h>
+#include <pipewire/private.h>
+
+struct impl {
+	struct pw_control this;
+
+	struct pw_memblock *mem;
+};
+
+struct pw_control *
+pw_control_new(struct pw_core *core,
+	       struct pw_port *port,
+	       const struct spa_pod *param,
+	       size_t user_data_size)
+{
+	struct impl *impl;
+	struct pw_control *this;
+	enum spa_direction direction;
+	struct pw_type *t = &core->type;
+
+	impl = calloc(1, sizeof(struct impl) + user_data_size);
+	if (impl == NULL)
+		goto exit;
+
+	this = &impl->this;
+
+	direction = spa_pod_is_object_id(param, t->param_io.idPropsOut) ?
+		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) < 0)
+		goto exit_free;
+
+	pw_log_debug("control %p: new %s %d", this, spa_type_map_get_type(t->map, this->id), direction);
+
+	this->core = core;
+	this->port = port;
+	this->param = pw_spa_pod_copy(param);
+	this->direction = direction;
+
+        if (user_data_size > 0)
+		this->user_data = SPA_MEMBER(impl, sizeof(struct impl), void);
+
+	spa_hook_list_init(&this->listener_list);
+
+	spa_list_append(&core->control_list[direction], &this->link);
+	if (port) {
+		spa_list_append(&port->control_list[direction], &this->port_link);
+		spa_hook_list_call(&port->listener_list, struct pw_port_events, control_added, this);
+	}
+
+	return this;
+
+    exit_free:
+	free(impl);
+    exit:
+	pw_log_error("control failed");
+	return NULL;
+}
+
+void pw_control_destroy(struct pw_control *control)
+{
+	struct impl *impl = SPA_CONTAINER_OF(control, struct impl, this);
+
+	pw_log_debug("control %p: destroy", control);
+
+	spa_hook_list_call(&control->listener_list, struct pw_control_events, destroy);
+
+	spa_list_remove(&control->link);
+
+	if (control->port) {
+		spa_list_remove(&control->port_link);
+		spa_hook_list_call(&control->port->listener_list,
+				struct pw_port_events, control_removed, control);
+	}
+
+	pw_log_debug("control %p: free", control);
+	spa_hook_list_call(&control->listener_list, struct pw_control_events, free);
+
+	if (control->direction == SPA_DIRECTION_OUTPUT)
+		if (impl->mem)
+			pw_memblock_free(impl->mem);
+
+	free(control->param);
+
+	free(control);
+}
+
+struct pw_port *pw_control_get_port(struct pw_control *control)
+{
+	return control->port;
+}
+
+void pw_control_add_listener(struct pw_control *control,
+			     struct spa_hook *listener,
+			     const struct pw_control_events *events,
+			     void *data)
+{
+	spa_hook_list_append(&control->listener_list, listener, events, data);
+}
+
+int pw_control_link(struct pw_control *control, struct pw_control *other)
+{
+	int res = 0;
+	struct impl *impl;
+
+	if (control->direction == SPA_DIRECTION_INPUT) {
+		struct pw_control *tmp = control;
+		control = other;
+		other = tmp;
+	}
+	if (control->direction != SPA_DIRECTION_OUTPUT ||
+	    other->direction != SPA_DIRECTION_INPUT)
+		return -EINVAL;
+
+	impl = SPA_CONTAINER_OF(control, struct impl, this);
+
+	pw_log_debug("control %p: link to %p", control, other);
+
+	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,
+					     &impl->mem)) < 0)
+			goto exit;
+
+		if (control->port) {
+			struct pw_port *port = control->port;
+			if ((res = spa_node_port_set_io(port->node->node,
+					     port->direction, port->port_id,
+					     control->id,
+					     impl->mem->ptr, control->size)) < 0) {
+				goto exit;
+			}
+		}
+	}
+
+	if (other->port) {
+		struct pw_port *port = other->port;
+		if ((res = spa_node_port_set_io(port->node->node,
+				     port->direction, port->port_id,
+				     other->id,
+				     impl->mem->ptr, control->size)) < 0) {
+			goto exit;
+		}
+	}
+
+     exit:
+	return res;
+}
+
+int pw_control_unlink(struct pw_control *control, struct pw_control *other)
+{
+	return -ENOTSUP;
+}
diff --git a/src/pipewire/control.h b/src/pipewire/control.h
new file mode 100644
index 000000000..f3b5b13ba
--- /dev/null
+++ b/src/pipewire/control.h
@@ -0,0 +1,76 @@
+/* PipeWire
+ * Copyright (C) 2015 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __PIPEWIRE_CONTROL_H__
+#define __PIPEWIRE_CONTROL_H__
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define PW_TYPE__Control			"PipeWire:Object:Control"
+#define PW_TYPE_CONTROL_BASE			PW_TYPE__Control ":"
+
+#include <spa/utils/hook.h>
+
+/** \page page_control Control
+ *
+ * \section page_control_overview Overview
+ *
+ * A control can be used to control a port property.
+ */
+/** \class pw_control
+ *
+ * The control object
+ */
+struct pw_control;
+
+#include <pipewire/core.h>
+#include <pipewire/introspect.h>
+#include <pipewire/node.h>
+
+/** Port events, use \ref pw_control_add_listener */
+struct pw_control_events {
+#define PW_VERSION_PORT_EVENTS 0
+	uint32_t version;
+
+	/** The control is destroyed */
+	void (*destroy) (void *data);
+
+	/** The control is freed */
+	void (*free) (void *data);
+};
+
+/** Get the control parent port or NULL when not set */
+struct pw_port *pw_control_get_port(struct pw_control *control);
+
+/** Add an event listener on the control */
+void pw_control_add_listener(struct pw_control *control,
+			     struct spa_hook *listener,
+			     const struct pw_control_events *events,
+			     void *data);
+
+int pw_control_link(struct pw_control *control, struct pw_control *other);
+int pw_control_unlink(struct pw_control *control, struct pw_control *other);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __PIPEWIRE_CONTROL_H__ */
diff --git a/src/pipewire/core.c b/src/pipewire/core.c
index af3058f5d..4954e3322 100644
--- a/src/pipewire/core.c
+++ b/src/pipewire/core.c
@@ -440,6 +440,8 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop, struct pw_properties *pro
 	spa_list_init(&this->node_list);
 	spa_list_init(&this->factory_list);
 	spa_list_init(&this->link_list);
+	spa_list_init(&this->control_list[0]);
+	spa_list_init(&this->control_list[1]);
 	spa_hook_list_init(&this->listener_list);
 
 	if ((name = pw_properties_get(properties, PW_CORE_PROP_NAME)) == NULL) {
diff --git a/src/pipewire/meson.build b/src/pipewire/meson.build
index 30d4ff152..9c480d1a3 100644
--- a/src/pipewire/meson.build
+++ b/src/pipewire/meson.build
@@ -2,6 +2,7 @@ pipewire_headers = [
   'array.h',
   'client.h',
   'command.h',
+  'control.h',
   'core.h',
   'data-loop.h',
   'global.h',
@@ -34,6 +35,7 @@ pipewire_headers = [
 pipewire_sources = [
   'client.c',
   'command.c',
+  'control.c',
   'core.c',
   'data-loop.c',
   'global.c',
diff --git a/src/pipewire/node.c b/src/pipewire/node.c
index 99fa75f30..051e26c8e 100644
--- a/src/pipewire/node.c
+++ b/src/pipewire/node.c
@@ -23,6 +23,7 @@
 #include <errno.h>
 
 #include <spa/clock/clock.h>
+#include <spa/lib/debug.h>
 
 #include "pipewire/pipewire.h"
 #include "pipewire/interfaces.h"
@@ -328,7 +329,6 @@ do_node_add(struct spa_loop *loop,
 	return 0;
 }
 
-
 void pw_node_register(struct pw_node *this,
 		      struct pw_client *owner,
 		      struct pw_global *parent)
diff --git a/src/pipewire/port.c b/src/pipewire/port.c
index d7b9d1d8c..212fe0db8 100644
--- a/src/pipewire/port.c
+++ b/src/pipewire/port.c
@@ -183,6 +183,8 @@ struct pw_port *pw_port_new(enum pw_direction direction,
 		this->user_data = SPA_MEMBER(impl, sizeof(struct impl), void);
 
 	spa_list_init(&this->links);
+	spa_list_init(&this->control_list[0]);
+	spa_list_init(&this->control_list[1]);
 
 	spa_hook_list_init(&this->listener_list);
 
@@ -267,13 +269,29 @@ static int do_add_port(struct spa_loop *loop,
 	return 0;
 }
 
+static int make_control(void *data, struct spa_pod *param)
+{
+	struct pw_port *port = data;
+	struct pw_node *node = port->node;
+	pw_control_new(node->core, port, param, 0);
+	return 0;
+}
+
 bool pw_port_add(struct pw_port *port, struct pw_node *node)
 {
 	uint32_t port_id = port->port_id;
+	struct pw_type *t = &node->core->type;
 
 	port->node = node;
 
-	pw_log_debug("port %p: add to node %p", port, node);
+	spa_node_port_get_info(node->node,
+			       port->direction, port_id,
+			       &port->info);
+
+	if (port->info->props)
+		pw_port_update_properties(port, port->info->props);
+
+	pw_log_debug("port %p: add to node %p %08x", port, node, port->info->flags);
 	if (port->direction == PW_DIRECTION_INPUT) {
 		spa_list_insert(&node->input_ports, &port->link);
 		pw_map_insert_at(&node->input_port_map, port_id, port);
@@ -287,6 +305,10 @@ bool pw_port_add(struct pw_port *port, struct pw_node *node)
 		node->info.change_mask |= PW_NODE_CHANGE_MASK_OUTPUT_PORTS;
 	}
 
+	pw_port_for_each_param(port, t->param_io.idPropsOut, NULL, make_control, port);
+	pw_port_for_each_param(port, t->param_io.idPropsIn, NULL, make_control, port);
+
+	pw_log_debug("port %p: setting node io", port);
 	spa_node_port_set_io(node->node,
 			     port->direction, port_id,
 			     node->core->type.io.Buffers,
@@ -329,7 +351,9 @@ void pw_port_destroy(struct pw_port *port)
 	spa_hook_list_call(&port->listener_list, struct pw_port_events, destroy);
 
 	if (node) {
-		pw_loop_invoke(port->node->data_loop, do_remove_port, SPA_ID_INVALID, NULL, 0, true, port);
+		if (port->rt.graph)
+			pw_loop_invoke(port->node->data_loop, do_remove_port,
+				       SPA_ID_INVALID, NULL, 0, true, port);
 
 		if (port->direction == PW_DIRECTION_INPUT) {
 			pw_map_remove(&node->input_port_map, port->port_id);
@@ -409,6 +433,42 @@ int pw_port_for_each_param(struct pw_port *port,
 	return res;
 }
 
+struct param_filter {
+	struct pw_port *in_port;
+	struct pw_port *out_port;
+	uint32_t in_param_id;
+	uint32_t out_param_id;
+	int (*callback) (void *data, struct spa_pod *param);
+	void *data;
+	uint32_t n_params;
+};
+
+static int do_filter(void *data, struct spa_pod *param)
+{
+	struct param_filter *f = data;
+	f->n_params++;
+	return pw_port_for_each_param(f->out_port, f->out_param_id, param, f->callback, f->data);
+}
+
+int pw_port_for_each_filtered_param(struct pw_port *in_port,
+				    struct pw_port *out_port,
+				    uint32_t in_param_id,
+				    uint32_t out_param_id,
+				    int (*callback) (void *data, struct spa_pod *param),
+				    void *data)
+{
+	int res;
+	struct param_filter filter = { in_port, out_port, in_param_id, out_param_id, callback, data, 0 };
+
+	if ((res = pw_port_for_each_param(in_port, in_param_id, NULL, do_filter, &filter)) < 0)
+		return res;
+
+	if (filter.n_params == 0)
+		res = do_filter(&filter, NULL);
+
+	return res;
+}
+
 int pw_port_set_param(struct pw_port *port, uint32_t id, uint32_t flags,
 		      const struct spa_pod *param)
 {
diff --git a/src/pipewire/port.h b/src/pipewire/port.h
index 5ca70c250..6386d4b94 100644
--- a/src/pipewire/port.h
+++ b/src/pipewire/port.h
@@ -41,6 +41,7 @@ extern "C" {
  */
 struct pw_port;
 struct pw_link;
+struct pw_control;
 
 #include <pipewire/core.h>
 #include <pipewire/introspect.h>
@@ -77,6 +78,12 @@ struct pw_port_events {
 
 	/** the properties of the port changed */
 	void (*properties_changed) (void *data, const struct pw_properties *properties);
+
+	/** a control was added to the port */
+	void (*control_added) (void *data, struct pw_control *control);
+
+	/** a control was removed from the port */
+	void (*control_removed) (void *data, struct pw_control *control);
 };
 
 /** Get the port direction */
diff --git a/src/pipewire/private.h b/src/pipewire/private.h
index 6159491ba..3e534abcd 100644
--- a/src/pipewire/private.h
+++ b/src/pipewire/private.h
@@ -137,6 +137,7 @@ struct pw_core {
 	struct spa_list node_list;		/**< list of nodes */
 	struct spa_list factory_list;		/**< list of factories */
 	struct spa_list link_list;		/**< list of links */
+	struct spa_list control_list[2];	/**< list of controls, indexed by direction */
 
 	struct spa_hook_list listener_list;
 
@@ -267,6 +268,7 @@ struct pw_port {
 	enum pw_direction direction;	/**< port direction */
 	uint32_t port_id;		/**< port id */
 	struct pw_properties *properties;
+	const struct spa_port_info *info;
 
 	enum pw_port_state state;	/**< state of the port */
 
@@ -279,6 +281,8 @@ struct pw_port {
 
 	struct spa_list links;		/**< list of \ref pw_link */
 
+	struct spa_list control_list[2];	/**< list of \ref pw_control indexed by direction */
+
 	struct spa_hook_list listener_list;
 
 	struct spa_node *mix;		/**< optional port buffer mix/split */
@@ -391,6 +395,24 @@ struct pw_factory {
 	void *user_data;
 };
 
+struct pw_control {
+	struct spa_list link;		/**< link in core control_list */
+	struct pw_core *core;		/**< the core */
+
+	struct pw_port *port;		/**< owner port or NULL */
+	struct spa_list port_link;	/**< link in port control_list */
+
+	enum spa_direction direction;	/**< the direction */
+	struct spa_pod *param;		/**< control params */
+
+	uint32_t id;
+	int32_t size;
+
+	struct spa_hook_list listener_list;
+
+	void *user_data;
+};
+
 /** Find a good format between 2 ports */
 int pw_core_find_format(struct pw_core *core,
 			struct pw_port *output,
@@ -438,6 +460,13 @@ int pw_port_for_each_param(struct pw_port *port,
 			   int (*callback) (void *data, struct spa_pod *param),
 			   void *data);
 
+int pw_port_for_each_filtered_param(struct pw_port *in_port,
+				    struct pw_port *out_port,
+				    uint32_t in_param_id,
+				    uint32_t out_param_id,
+				    int (*callback) (void *data, struct spa_pod *param),
+				    void *data);
+
 /** Set a param on a port \memberof pw_port */
 int pw_port_set_param(struct pw_port *port, uint32_t id, uint32_t flags,
 		      const struct spa_pod *param);
@@ -470,6 +499,14 @@ bool pw_link_activate(struct pw_link *link);
 /** Deactivate a link \memberof pw_link */
 bool pw_link_deactivate(struct pw_link *link);
 
+struct pw_control *
+pw_control_new(struct pw_core *core,
+	       struct pw_port *owner,		/**< can be NULL */
+	       const struct spa_pod *param,	/**< copy is taken */
+	       size_t user_data_size		/**< extra user data */);
+
+void pw_control_destroy(struct pw_control *control);
+
 /** \endcond */
 
 #ifdef __cplusplus
diff --git a/src/pipewire/remote.c b/src/pipewire/remote.c
index 4ad690539..35330f3ca 100644
--- a/src/pipewire/remote.c
+++ b/src/pipewire/remote.c
@@ -1053,7 +1053,7 @@ client_node_port_command(void *object,
 {
 	struct pw_proxy *proxy = object;
 	struct node_data *data = proxy->user_data;
-	static struct port *port;
+	struct port *port;
 
 	port = find_port(data, direction, port_id);
 	if (port == NULL)
@@ -1062,6 +1062,51 @@ client_node_port_command(void *object,
 	pw_port_send_command(port->port, true, command);
 }
 
+static void
+client_node_port_set_io(void *object,
+                        uint32_t seq,
+                        uint32_t direction,
+                        uint32_t port_id,
+                        uint32_t id,
+                        uint32_t memid,
+                        uint32_t offset,
+                        uint32_t size)
+{
+	struct pw_proxy *proxy = object;
+	struct node_data *data = proxy->user_data;
+	struct port *port;
+	struct mem_id *mid;
+
+	port = find_port(data, direction, port_id);
+	if (port == NULL)
+		return;
+
+	mid = find_mem(port, memid);
+	if (mid == NULL) {
+		pw_log_warn("unknown memory id %u", memid);
+		return;
+	}
+
+	if (mid->ptr == NULL) {
+		mid->ptr =
+		    mmap(NULL, mid->size + mid->offset, PROT_READ|PROT_WRITE, MAP_SHARED, mid->fd, 0);
+		if (mid->ptr == MAP_FAILED) {
+			mid->ptr = NULL;
+			pw_log_warn("Failed to mmap memory %d %p: %s", mid->size, mid,
+				    strerror(errno));
+			return;
+		}
+	}
+
+	spa_node_port_set_io(port->port->node->node,
+			     direction, port_id,
+			     id,
+			     SPA_MEMBER(mid->ptr, offset, void),
+			     size);
+
+}
+
+
 static const struct pw_client_node_proxy_events client_node_events = {
 	PW_VERSION_CLIENT_NODE_PROXY_EVENTS,
 	.transport = client_node_transport,
@@ -1074,6 +1119,7 @@ static const struct pw_client_node_proxy_events client_node_events = {
 	.port_add_mem = client_node_port_add_mem,
 	.port_use_buffers = client_node_port_use_buffers,
 	.port_command = client_node_port_command,
+	.port_set_io = client_node_port_set_io,
 };
 
 static void do_node_init(struct pw_proxy *proxy)