Adding upstream version 0.3.65.upstream/0.3.65upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

10 files changed, 7252 insertions, 0 deletions

diff --git a/src/modules/module-client-node/client-node.c b/src/modules/module-client-node/client-node.c
new file mode 100644
index 0000000..d5c133a
--- /dev/null
+++ b/src/modules/module-client-node/client-node.c
@@ -0,0 +1,1777 @@
+/* PipeWire
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <string.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <errno.h>
+#include <unistd.h>
+#include <time.h>
+
+#include <spa/support/system.h>
+#include <spa/node/node.h>
+#include <spa/node/utils.h>
+#include <spa/pod/filter.h>
+#include <spa/pod/dynamic.h>
+#include <spa/pod/parser.h>
+#include <spa/debug/types.h>
+
+#include <pipewire/pipewire.h>
+#include "pipewire/private.h"
+
+#include "modules/spa/spa-node.h"
+#include "client-node.h"
+
+PW_LOG_TOPIC_EXTERN(mod_topic);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+/** \cond */
+
+#define MAX_BUFFERS	64
+#define MAX_METAS	16u
+#define MAX_DATAS	64u
+#define MAX_AREAS	2048
+
+#define CHECK_FREE_PORT(this,d,p)	(p <= pw_map_get_size(&this->ports[d]) && !CHECK_PORT(this,d,p))
+#define CHECK_PORT(this,d,p)		(pw_map_lookup(&this->ports[d], p) != NULL)
+#define GET_PORT(this,d,p)		(pw_map_lookup(&this->ports[d], p))
+
+#define CHECK_PORT_BUFFER(this,b,p)      (b < p->n_buffers)
+
+struct buffer {
+	struct spa_buffer *outbuf;
+	struct spa_buffer buffer;
+	struct spa_meta metas[MAX_METAS];
+	struct spa_data datas[MAX_DATAS];
+	struct pw_memblock *mem;
+};
+
+struct mix {
+	unsigned int valid:1;
+	uint32_t id;
+	struct port *port;
+	uint32_t peer_id;
+	uint32_t n_buffers;
+	struct buffer buffers[MAX_BUFFERS];
+};
+
+struct params {
+	uint32_t n_params;
+	struct spa_pod **params;
+};
+
+struct port {
+	struct pw_impl_port *port;
+	struct node *node;
+	struct impl *impl;
+
+	enum spa_direction direction;
+	uint32_t id;
+
+	struct spa_node mix_node;
+
+	struct spa_port_info info;
+	struct pw_properties *properties;
+
+	struct params params;
+
+	unsigned int removed:1;
+	unsigned int destroyed:1;
+
+	struct pw_array mix;
+};
+
+struct node {
+	struct spa_node node;
+
+	struct impl *impl;
+
+	struct spa_log *log;
+	struct spa_loop *data_loop;
+	struct spa_system *data_system;
+
+	struct spa_hook_list hooks;
+	struct spa_callbacks callbacks;
+
+	struct pw_resource *resource;
+	struct pw_impl_client *client;
+
+	struct spa_source data_source;
+	int writefd;
+
+	struct pw_map ports[2];
+
+	struct port dummy;
+
+	struct params params;
+};
+
+struct impl {
+	struct pw_impl_client_node this;
+
+	struct pw_context *context;
+
+	struct node node;
+
+	struct pw_map io_map;
+	struct pw_memblock *io_areas;
+
+	struct pw_memblock *activation;
+
+	struct spa_hook node_listener;
+	struct spa_hook resource_listener;
+	struct spa_hook object_listener;
+
+	uint32_t node_id;
+
+	uint32_t bind_node_version;
+	uint32_t bind_node_id;
+
+	int fds[2];
+	int other_fds[2];
+};
+
+#define pw_client_node_resource(r,m,v,...)	\
+	pw_resource_call_res(r,struct pw_client_node_events,m,v,__VA_ARGS__)
+
+#define pw_client_node_resource_transport(r,...)	\
+	pw_client_node_resource(r,transport,0,__VA_ARGS__)
+#define pw_client_node_resource_set_param(r,...)	\
+	pw_client_node_resource(r,set_param,0,__VA_ARGS__)
+#define pw_client_node_resource_set_io(r,...)	\
+	pw_client_node_resource(r,set_io,0,__VA_ARGS__)
+#define pw_client_node_resource_event(r,...)	\
+	pw_client_node_resource(r,event,0,__VA_ARGS__)
+#define pw_client_node_resource_command(r,...)	\
+	pw_client_node_resource(r,command,0,__VA_ARGS__)
+#define pw_client_node_resource_add_port(r,...)	\
+	pw_client_node_resource(r,add_port,0,__VA_ARGS__)
+#define pw_client_node_resource_remove_port(r,...)	\
+	pw_client_node_resource(r,remove_port,0,__VA_ARGS__)
+#define pw_client_node_resource_port_set_param(r,...)	\
+	pw_client_node_resource(r,port_set_param,0,__VA_ARGS__)
+#define pw_client_node_resource_port_use_buffers(r,...)	\
+	pw_client_node_resource(r,port_use_buffers,0,__VA_ARGS__)
+#define pw_client_node_resource_port_set_io(r,...)	\
+	pw_client_node_resource(r,port_set_io,0,__VA_ARGS__)
+#define pw_client_node_resource_set_activation(r,...)	\
+	pw_client_node_resource(r,set_activation,0,__VA_ARGS__)
+#define pw_client_node_resource_port_set_mix_info(r,...)	\
+	pw_client_node_resource(r,port_set_mix_info,1,__VA_ARGS__)
+
+static int update_params(struct params *p, uint32_t n_params, const struct spa_pod **params)
+{
+	uint32_t i;
+	for (i = 0; i < p->n_params; i++)
+		free(p->params[i]);
+	p->n_params = n_params;
+	if (p->n_params == 0) {
+		free(p->params);
+		p->params = NULL;
+	} else {
+		struct spa_pod **np;
+		np = pw_reallocarray(p->params, p->n_params, sizeof(struct spa_pod *));
+		if (np == NULL) {
+			pw_log_error("%p: can't realloc: %m", p);
+			free(p->params);
+			p->params = NULL;
+			p->n_params = 0;
+			return -errno;
+		}
+		p->params = np;
+	}
+	for (i = 0; i < p->n_params; i++)
+		p->params[i] = params[i] ? spa_pod_copy(params[i]) : NULL;
+	return 0;
+}
+
+static int
+do_port_use_buffers(struct impl *impl,
+		    enum spa_direction direction,
+		    uint32_t port_id,
+		    uint32_t mix_id,
+		    uint32_t flags,
+		    struct spa_buffer **buffers,
+		    uint32_t n_buffers);
+
+/** \endcond */
+
+static struct mix *find_mix(struct port *p, uint32_t mix_id)
+{
+	struct mix *mix;
+	size_t len;
+
+	if (mix_id == SPA_ID_INVALID)
+		mix_id = 0;
+	else
+		mix_id++;
+
+	len = pw_array_get_len(&p->mix, struct mix);
+	if (mix_id >= len) {
+		size_t need = sizeof(struct mix) * (mix_id + 1 - len);
+		void *ptr = pw_array_add(&p->mix, need);
+		if (ptr == NULL)
+			return NULL;
+		memset(ptr, 0, need);
+	}
+	mix = pw_array_get_unchecked(&p->mix, mix_id, struct mix);
+	return mix;
+}
+
+static void mix_init(struct mix *mix, struct port *p, uint32_t id)
+{
+	mix->valid = true;
+	mix->id = id;
+	mix->port = p;
+	mix->n_buffers = 0;
+}
+
+static struct mix *ensure_mix(struct impl *impl, struct port *p, uint32_t mix_id)
+{
+	struct mix *mix;
+
+	if ((mix = find_mix(p, mix_id)) == NULL)
+		return NULL;
+	if (mix->valid)
+		return mix;
+	mix_init(mix, p, mix_id);
+	return mix;
+}
+
+static void clear_data(struct node *this, struct spa_data *d)
+{
+	struct impl *impl = this->impl;
+
+	switch (d->type) {
+	case SPA_DATA_MemId:
+	{
+		uint32_t id;
+		struct pw_memblock *m;
+
+		id = SPA_PTR_TO_UINT32(d->data);
+		m = pw_mempool_find_id(this->client->pool, id);
+		if (m) {
+			pw_log_debug("%p: mem %d", impl, m->id);
+			pw_memblock_unref(m);
+		}
+		break;
+	}
+	case SPA_DATA_MemFd:
+	case SPA_DATA_DmaBuf:
+		pw_log_debug("%p: close fd:%d", impl, (int)d->fd);
+		close(d->fd);
+		break;
+	default:
+		break;
+	}
+}
+
+static int clear_buffers(struct node *this, struct mix *mix)
+{
+	uint32_t i, j;
+
+	for (i = 0; i < mix->n_buffers; i++) {
+		struct buffer *b = &mix->buffers[i];
+
+		spa_log_debug(this->log, "%p: clear buffer %d", this, i);
+
+		for (j = 0; j < b->buffer.n_datas; j++) {
+			struct spa_data *d = &b->datas[j];
+			clear_data(this, d);
+		}
+		pw_memblock_unref(b->mem);
+	}
+	mix->n_buffers = 0;
+	return 0;
+}
+
+static void mix_clear(struct node *this, struct mix *mix)
+{
+	struct port *port = mix->port;
+
+	if (!mix->valid)
+		return;
+	do_port_use_buffers(this->impl, port->direction, port->id,
+			mix->id, 0, NULL, 0);
+	mix->valid = false;
+}
+
+static int impl_node_enum_params(void *object, int seq,
+				 uint32_t id, uint32_t start, uint32_t num,
+				 const struct spa_pod *filter)
+{
+	struct node *this = object;
+	uint8_t buffer[1024];
+	struct spa_pod_dynamic_builder b;
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+	bool found = false;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	result.id = id;
+	result.next = 0;
+
+	while (true) {
+		struct spa_pod *param;
+
+		result.index = result.next++;
+		if (result.index >= this->params.n_params)
+			break;
+
+		param = this->params.params[result.index];
+
+		if (param == NULL || !spa_pod_is_object_id(param, id))
+			continue;
+
+		found = true;
+
+		if (result.index < start)
+			continue;
+
+		spa_pod_dynamic_builder_init(&b, buffer, sizeof(buffer), 4096);
+		if (spa_pod_filter(&b.b, &result.param, param, filter) == 0) {
+			pw_log_debug("%p: %d param %u", this, seq, result.index);
+			spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+			count++;
+		}
+		spa_pod_dynamic_builder_clean(&b);
+
+		if (count == num)
+			break;
+	}
+	return found ? 0 : -ENOENT;
+}
+
+static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
+			       const struct spa_pod *param)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	if (this->resource == NULL)
+		return param == NULL ? 0 : -EIO;
+
+	return pw_client_node_resource_set_param(this->resource, id, flags, param);
+}
+
+static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
+{
+	struct node *this = object;
+	struct impl *impl = this->impl;
+	struct pw_memmap *mm, *old;
+	uint32_t memid, mem_offset, mem_size;
+	uint32_t tag[5] = { impl->node_id, id, };
+
+	if (impl->this.flags & 1)
+		return 0;
+
+	old = pw_mempool_find_tag(this->client->pool, tag, sizeof(tag));
+
+	if (data) {
+		mm = pw_mempool_import_map(this->client->pool,
+				impl->context->pool, data, size, tag);
+		if (mm == NULL)
+			return -errno;
+
+		mem_offset = mm->offset;
+		memid = mm->block->id;
+		mem_size = size;
+	}
+	else {
+		memid = SPA_ID_INVALID;
+		mem_offset = mem_size = 0;
+	}
+	pw_memmap_free(old);
+
+	if (this->resource == NULL)
+		return data == NULL ? 0 : -EIO;
+
+	return pw_client_node_resource_set_io(this->resource,
+				       id,
+				       memid,
+				       mem_offset, mem_size);
+}
+
+static int impl_node_send_command(void *object, const struct spa_command *command)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(command != NULL, -EINVAL);
+
+	pw_log_debug("%p: send command %d", this, SPA_COMMAND_TYPE(command));
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	return pw_client_node_resource_command(this->resource, command);
+}
+
+
+static void emit_port_info(struct node *this, struct port *port)
+{
+	spa_node_emit_port_info(&this->hooks,
+				port->direction, port->id, &port->info);
+}
+
+static int impl_node_add_listener(void *object,
+		struct spa_hook *listener,
+		const struct spa_node_events *events,
+		void *data)
+{
+	struct node *this = object;
+	struct spa_hook_list save;
+	union pw_map_item *item;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_hook_list_isolate(&this->hooks, &save, listener, events, data);
+
+	pw_array_for_each(item, &this->ports[SPA_DIRECTION_INPUT].items) {
+		if (item->data)
+			emit_port_info(this, item->data);
+	}
+	pw_array_for_each(item, &this->ports[SPA_DIRECTION_OUTPUT].items) {
+		if (item->data)
+			emit_port_info(this, item->data);
+	}
+	spa_hook_list_join(&this->hooks, &save);
+
+	return 0;
+}
+
+static int
+impl_node_set_callbacks(void *object,
+			const struct spa_node_callbacks *callbacks,
+			void *data)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	this->callbacks = SPA_CALLBACKS_INIT(callbacks, data);
+
+	return 0;
+}
+
+static int
+impl_node_sync(void *object, int seq)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	pw_log_debug("%p: sync", this);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	return pw_resource_ping(this->resource, seq);
+}
+
+static void
+do_update_port(struct node *this,
+	       struct port *port,
+	       uint32_t change_mask,
+	       uint32_t n_params,
+	       const struct spa_pod **params,
+	       const struct spa_port_info *info)
+{
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_PARAMS) {
+		spa_log_debug(this->log, "%p: port %u update %d params", this, port->id, n_params);
+		update_params(&port->params, n_params, params);
+	}
+
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_INFO) {
+		pw_properties_free(port->properties);
+		port->properties = NULL;
+		port->info.props = NULL;
+		port->info.n_params = 0;
+		port->info.params = NULL;
+
+		if (info) {
+			port->info = *info;
+			if (info->props) {
+				port->properties = pw_properties_new_dict(info->props);
+				port->info.props = &port->properties->dict;
+			}
+			port->info.n_params = 0;
+			port->info.params = NULL;
+			spa_node_emit_port_info(&this->hooks, port->direction, port->id, info);
+		}
+	}
+}
+
+static void
+clear_port(struct node *this, struct port *port)
+{
+	struct mix *mix;
+
+	spa_log_debug(this->log, "%p: clear port %p", this, port);
+
+	do_update_port(this, port,
+		       PW_CLIENT_NODE_PORT_UPDATE_PARAMS |
+		       PW_CLIENT_NODE_PORT_UPDATE_INFO, 0, NULL, NULL);
+
+	pw_array_for_each(mix, &port->mix)
+		mix_clear(this, mix);
+	pw_array_clear(&port->mix);
+	pw_array_init(&port->mix, sizeof(struct mix) * 2);
+
+	pw_map_insert_at(&this->ports[port->direction], port->id, NULL);
+
+	if (!port->removed)
+		spa_node_emit_port_info(&this->hooks, port->direction, port->id, NULL);
+}
+
+static int
+impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_FREE_PORT(this, direction, port_id), -EINVAL);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	return pw_client_node_resource_add_port(this->resource, direction, port_id, props);
+}
+
+static int
+impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	return pw_client_node_resource_remove_port(this->resource, direction, port_id);
+}
+
+static int
+impl_node_port_enum_params(void *object, int seq,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t id, uint32_t start, uint32_t num,
+			   const struct spa_pod *filter)
+{
+	struct node *this = object;
+	struct port *port;
+	uint8_t buffer[1024];
+	struct spa_pod_dynamic_builder b;
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+	bool found = false;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	port = GET_PORT(this, direction, port_id);
+	spa_return_val_if_fail(port != NULL, -EINVAL);
+
+	pw_log_debug("%p: seq:%d port %d.%d id:%u start:%u num:%u n_params:%d",
+			this, seq, direction, port_id, id, start, num, port->params.n_params);
+
+	result.id = id;
+	result.next = 0;
+
+	while (true) {
+		struct spa_pod *param;
+
+		result.index = result.next++;
+		if (result.index >= port->params.n_params)
+			break;
+
+		param = port->params.params[result.index];
+
+		if (param == NULL || !spa_pod_is_object_id(param, id))
+			continue;
+
+		found = true;
+
+		if (result.index < start)
+			continue;
+
+		spa_pod_dynamic_builder_init(&b, buffer, sizeof(buffer), 4096);
+		if (spa_pod_filter(&b.b, &result.param, param, filter) == 0) {
+			pw_log_debug("%p: %d param %u", this, seq, result.index);
+			spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+			count++;
+		}
+		spa_pod_dynamic_builder_clean(&b);
+
+		if (count == num)
+			break;
+	}
+	return found ? 0 : -ENOENT;
+}
+
+static int
+impl_node_port_set_param(void *object,
+			 enum spa_direction direction, uint32_t port_id,
+			 uint32_t id, uint32_t flags,
+			 const struct spa_pod *param)
+{
+	struct node *this = object;
+	struct port *port;
+	struct mix *mix;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	port = GET_PORT(this, direction, port_id);
+	if(port == NULL)
+		return param == NULL ? 0 : -EINVAL;
+
+	pw_log_debug("%p: port %d.%d set param %s %d", this,
+			direction, port_id,
+			spa_debug_type_find_name(spa_type_param, id), id);
+
+	if (id == SPA_PARAM_Format) {
+		pw_array_for_each(mix, &port->mix)
+			clear_buffers(this, mix);
+	}
+	if (this->resource == NULL)
+		return param == NULL ? 0 : -EIO;
+
+	return pw_client_node_resource_port_set_param(this->resource,
+					       direction, port_id,
+					       id, flags,
+					       param);
+}
+
+static int do_port_set_io(struct impl *impl,
+			  enum spa_direction direction, uint32_t port_id,
+			  uint32_t mix_id,
+			  uint32_t id, void *data, size_t size)
+{
+	struct node *this = &impl->node;
+	uint32_t memid, mem_offset, mem_size;
+	struct port *port;
+	struct mix *mix;
+	uint32_t tag[5] = { impl->node_id, direction, port_id, mix_id, id };
+	struct pw_memmap *mm, *old;
+
+	pw_log_debug("%p: %s port %d.%d set io %p %zd", this,
+			direction == SPA_DIRECTION_INPUT ? "input" : "output",
+			port_id, mix_id, data, size);
+
+	port = GET_PORT(this, direction, port_id);
+	if (port == NULL)
+		return data == NULL ? 0 : -EINVAL;
+
+	if ((mix = find_mix(port, mix_id)) == NULL || !mix->valid)
+		return -EINVAL;
+
+	old = pw_mempool_find_tag(this->client->pool, tag, sizeof(tag));
+
+	if (data) {
+		mm = pw_mempool_import_map(this->client->pool,
+				impl->context->pool, data, size, tag);
+		if (mm == NULL)
+			return -errno;
+
+		mem_offset = mm->offset;
+		memid = mm->block->id;
+		mem_size = size;
+	}
+	else {
+		memid = SPA_ID_INVALID;
+		mem_offset = mem_size = 0;
+	}
+	pw_memmap_free(old);
+
+	if (this->resource == NULL)
+		return data == NULL ? 0 : -EIO;
+
+	return pw_client_node_resource_port_set_io(this->resource,
+					    direction, port_id,
+					    mix_id,
+					    id,
+					    memid,
+					    mem_offset, mem_size);
+}
+
+static int
+impl_node_port_set_io(void *object,
+		      enum spa_direction direction,
+		      uint32_t port_id,
+		      uint32_t id,
+		      void *data, size_t size)
+{
+	/* ignore io on the node itself, we only care about the io on the
+	 * port mixers, the io on the node ports itself is handled on the
+	 * client side */
+	return -EINVAL;
+}
+
+static int
+do_port_use_buffers(struct impl *impl,
+		    enum spa_direction direction,
+		    uint32_t port_id,
+		    uint32_t mix_id,
+		    uint32_t flags,
+		    struct spa_buffer **buffers,
+		    uint32_t n_buffers)
+{
+	struct node *this = &impl->node;
+	struct port *p;
+	struct mix *mix;
+	uint32_t i, j;
+	struct pw_client_node_buffer *mb;
+
+	p = GET_PORT(this, direction, port_id);
+	if (p == NULL)
+		return n_buffers == 0 ? 0 : -EINVAL;
+
+	if (n_buffers > MAX_BUFFERS)
+		return -ENOSPC;
+
+	spa_log_debug(this->log, "%p: %s port %d.%d use buffers %p %u flags:%08x", this,
+			direction == SPA_DIRECTION_INPUT ? "input" : "output",
+			port_id, mix_id, buffers, n_buffers, flags);
+
+	if ((mix = find_mix(p, mix_id)) == NULL || !mix->valid)
+		return -EINVAL;
+
+	if (direction == SPA_DIRECTION_OUTPUT) {
+		mix_id = SPA_ID_INVALID;
+		if ((mix = find_mix(p, mix_id)) == NULL || !mix->valid)
+			return -EINVAL;
+	}
+
+	clear_buffers(this, mix);
+
+	if (n_buffers > 0) {
+		mb = alloca(n_buffers * sizeof(struct pw_client_node_buffer));
+	} else {
+		mb = NULL;
+	}
+
+	if (this->resource == NULL)
+		return n_buffers == 0 ? 0 : -EIO;
+
+	if (p->destroyed)
+		return 0;
+
+	for (i = 0; i < n_buffers; i++) {
+		struct buffer *b = &mix->buffers[i];
+		struct pw_memblock *mem, *m;
+		void *baseptr, *endptr;
+
+		b->outbuf = buffers[i];
+		memcpy(&b->buffer, buffers[i], sizeof(struct spa_buffer));
+		b->buffer.datas = b->datas;
+		b->buffer.metas = b->metas;
+
+		if (buffers[i]->n_metas > 0)
+			baseptr = buffers[i]->metas[0].data;
+		else if (buffers[i]->n_datas > 0)
+			baseptr = buffers[i]->datas[0].chunk;
+		else
+			return -EINVAL;
+
+		if ((mem = pw_mempool_find_ptr(impl->context->pool, baseptr)) == NULL)
+			return -EINVAL;
+
+		endptr = SPA_PTROFF(baseptr, buffers[i]->n_datas * sizeof(struct spa_chunk), void);
+		for (j = 0; j < buffers[i]->n_metas; j++) {
+			endptr = SPA_PTROFF(endptr, SPA_ROUND_UP_N(buffers[i]->metas[j].size, 8), void);
+		}
+		for (j = 0; j < buffers[i]->n_datas; j++) {
+			struct spa_data *d = &buffers[i]->datas[j];
+			if (d->type == SPA_DATA_MemPtr) {
+				if ((m = pw_mempool_find_ptr(impl->context->pool, d->data)) == NULL ||
+				    m != mem)
+					return -EINVAL;
+				endptr = SPA_MAX(endptr, SPA_PTROFF(d->data, d->maxsize, void));
+			}
+		}
+		if (endptr > SPA_PTROFF(baseptr, mem->size, void))
+			return -EINVAL;
+
+		m = pw_mempool_import_block(this->client->pool, mem);
+		if (m == NULL)
+			return -errno;
+
+		b->mem = m;
+
+		mb[i].buffer = &b->buffer;
+		mb[i].mem_id = m->id;
+		mb[i].offset = SPA_PTRDIFF(baseptr, mem->map->ptr);
+		mb[i].size = SPA_PTRDIFF(endptr, baseptr);
+		spa_log_debug(this->log, "%p: buffer %d %d %d %d", this, i, mb[i].mem_id,
+				mb[i].offset, mb[i].size);
+
+		b->buffer.n_metas = SPA_MIN(buffers[i]->n_metas, MAX_METAS);
+		for (j = 0; j < b->buffer.n_metas; j++)
+			memcpy(&b->buffer.metas[j], &buffers[i]->metas[j], sizeof(struct spa_meta));
+
+		b->buffer.n_datas = SPA_MIN(buffers[i]->n_datas, MAX_DATAS);
+		for (j = 0; j < b->buffer.n_datas; j++) {
+			struct spa_data *d = &buffers[i]->datas[j];
+
+			memcpy(&b->datas[j], d, sizeof(struct spa_data));
+
+			if (flags & SPA_NODE_BUFFERS_FLAG_ALLOC)
+				continue;
+
+			switch (d->type) {
+			case SPA_DATA_DmaBuf:
+			case SPA_DATA_MemFd:
+			{
+				uint32_t flags = PW_MEMBLOCK_FLAG_DONT_CLOSE;
+
+				if (d->flags & SPA_DATA_FLAG_READABLE)
+					flags |= PW_MEMBLOCK_FLAG_READABLE;
+				if (d->flags & SPA_DATA_FLAG_WRITABLE)
+					flags |= PW_MEMBLOCK_FLAG_WRITABLE;
+
+				spa_log_debug(this->log, "mem %d type:%d fd:%d", j, d->type, (int)d->fd);
+				m = pw_mempool_import(this->client->pool,
+					flags, d->type, d->fd);
+				if (m == NULL)
+					return -errno;
+
+				b->datas[j].type = SPA_DATA_MemId;
+				b->datas[j].data = SPA_UINT32_TO_PTR(m->id);
+				break;
+			}
+			case SPA_DATA_MemPtr:
+				spa_log_debug(this->log, "mem %d %zd", j, SPA_PTRDIFF(d->data, baseptr));
+				b->datas[j].data = SPA_INT_TO_PTR(SPA_PTRDIFF(d->data, baseptr));
+				break;
+			default:
+				b->datas[j].type = SPA_ID_INVALID;
+				b->datas[j].data = NULL;
+				spa_log_error(this->log, "invalid memory type %d", d->type);
+				break;
+			}
+		}
+	}
+	mix->n_buffers = n_buffers;
+
+	return pw_client_node_resource_port_use_buffers(this->resource,
+						 direction, port_id, mix_id, flags,
+						 n_buffers, mb);
+}
+
+static int
+impl_node_port_use_buffers(void *object,
+			   enum spa_direction direction,
+			   uint32_t port_id,
+			   uint32_t flags,
+			   struct spa_buffer **buffers,
+			   uint32_t n_buffers)
+{
+	struct node *this = object;
+	struct impl *impl;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	impl = this->impl;
+
+	return do_port_use_buffers(impl, direction, port_id,
+			SPA_ID_INVALID, flags, buffers, n_buffers);
+}
+
+static int
+impl_node_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, SPA_DIRECTION_OUTPUT, port_id), -EINVAL);
+
+	spa_log_trace_fp(this->log, "reuse buffer %d", buffer_id);
+
+	return -ENOTSUP;
+}
+
+static int impl_node_process(void *object)
+{
+	struct node *this = object;
+	struct impl *impl = this->impl;
+	struct pw_impl_node *n = impl->this.node;
+	struct timespec ts;
+
+	spa_log_trace_fp(this->log, "%p: send process driver:%p", this, impl->this.node->driver_node);
+
+	if (SPA_UNLIKELY(spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &ts) < 0))
+		spa_zero(ts);
+
+	n->rt.activation->status = PW_NODE_ACTIVATION_TRIGGERED;
+	n->rt.activation->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
+
+	if (SPA_UNLIKELY(spa_system_eventfd_write(this->data_system, this->writefd, 1) < 0))
+		spa_log_warn(this->log, "%p: error %m", this);
+
+	return SPA_STATUS_OK;
+}
+
+static struct pw_node *
+client_node_get_node(void *data,
+		   uint32_t version,
+		   size_t user_data_size)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	uint32_t new_id = user_data_size;
+
+	pw_log_debug("%p: bind %u/%u", this, new_id, version);
+
+	impl->bind_node_version = version;
+	impl->bind_node_id = new_id;
+	pw_map_insert_at(&this->client->objects, new_id, NULL);
+
+	return NULL;
+}
+
+static int
+client_node_update(void *data,
+		   uint32_t change_mask,
+		   uint32_t n_params,
+		   const struct spa_pod **params,
+		   const struct spa_node_info *info)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	if (change_mask & PW_CLIENT_NODE_UPDATE_PARAMS) {
+		pw_log_debug("%p: update %d params", this, n_params);
+		update_params(&this->params, n_params, params);
+	}
+	if (change_mask & PW_CLIENT_NODE_UPDATE_INFO) {
+		spa_node_emit_info(&this->hooks, info);
+	}
+	pw_log_debug("%p: got node update", this);
+	return 0;
+}
+
+static int
+client_node_port_update(void *data,
+			enum spa_direction direction,
+			uint32_t port_id,
+			uint32_t change_mask,
+			uint32_t n_params,
+			const struct spa_pod **params,
+			const struct spa_port_info *info)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct port *port;
+	bool remove;
+
+	spa_log_debug(this->log, "%p: got port update change:%08x params:%d",
+			this, change_mask, n_params);
+
+	remove = (change_mask == 0);
+
+	port = GET_PORT(this, direction, port_id);
+
+	if (remove) {
+		if (port == NULL)
+			return 0;
+		port->destroyed = true;
+		clear_port(this, port);
+	} else {
+		struct port *target;
+
+		if (port == NULL) {
+			if (!CHECK_FREE_PORT(this, direction, port_id))
+				return -EINVAL;
+
+			target = &this->dummy;
+			spa_zero(this->dummy);
+			target->direction = direction;
+			target->id = port_id;
+		} else
+			target = port;
+
+		do_update_port(this,
+			       target,
+			       change_mask,
+			       n_params, params,
+			       info);
+	}
+	return 0;
+}
+
+static int client_node_set_active(void *data, bool active)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	spa_log_debug(this->log, "%p: active:%d", this, active);
+	return pw_impl_node_set_active(impl->this.node, active);
+}
+
+static int client_node_event(void *data, const struct spa_event *event)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	spa_node_emit_event(&this->hooks, event);
+	return 0;
+}
+
+static int client_node_port_buffers(void *data,
+			enum spa_direction direction,
+			uint32_t port_id,
+			uint32_t mix_id,
+			uint32_t n_buffers,
+			struct spa_buffer **buffers)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct port *p;
+	struct mix *mix;
+	uint32_t i, j;
+
+	spa_log_debug(this->log, "%p: %s port %d.%d buffers %p %u", this,
+			direction == SPA_DIRECTION_INPUT ? "input" : "output",
+			port_id, mix_id, buffers, n_buffers);
+
+	p = GET_PORT(this, direction, port_id);
+	spa_return_val_if_fail(p != NULL, -EINVAL);
+
+	if (direction == SPA_DIRECTION_OUTPUT)
+		mix_id = SPA_ID_INVALID;
+
+	if ((mix = find_mix(p, mix_id)) == NULL || !mix->valid)
+		return -EINVAL;
+
+	if (mix->n_buffers != n_buffers)
+		return -EINVAL;
+
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *oldbuf, *newbuf;
+		struct buffer *b = &mix->buffers[i];
+
+		oldbuf = b->outbuf;
+		newbuf = buffers[i];
+
+		spa_log_debug(this->log, "buffer %d n_datas:%d", i, newbuf->n_datas);
+
+		if (oldbuf->n_datas != newbuf->n_datas)
+			return -EINVAL;
+
+		for (j = 0; j < b->buffer.n_datas; j++) {
+			struct spa_chunk *oldchunk = oldbuf->datas[j].chunk;
+			struct spa_data *d = &newbuf->datas[j];
+
+			/* overwrite everything except the chunk */
+			oldbuf->datas[j] = *d;
+			oldbuf->datas[j].chunk = oldchunk;
+
+			b->datas[j].type = d->type;
+			b->datas[j].fd = d->fd;
+
+			spa_log_debug(this->log, " data %d type:%d fl:%08x fd:%d, offs:%d max:%d",
+					j, d->type, d->flags, (int) d->fd, d->mapoffset,
+					d->maxsize);
+		}
+	}
+	mix->n_buffers = n_buffers;
+
+	return 0;
+}
+
+static const struct pw_client_node_methods client_node_methods = {
+	PW_VERSION_CLIENT_NODE_METHODS,
+	.get_node = client_node_get_node,
+	.update = client_node_update,
+	.port_update = client_node_port_update,
+	.set_active = client_node_set_active,
+	.event = client_node_event,
+	.port_buffers = client_node_port_buffers,
+};
+
+static void node_on_data_fd_events(struct spa_source *source)
+{
+	struct node *this = source->data;
+
+	if (source->rmask & (SPA_IO_ERR | SPA_IO_HUP)) {
+		spa_log_warn(this->log, "%p: got error", this);
+		return;
+	}
+
+	if (source->rmask & SPA_IO_IN) {
+		uint64_t cmd;
+		struct pw_impl_node *node = this->impl->this.node;
+
+		if (SPA_UNLIKELY(spa_system_eventfd_read(this->data_system,
+					this->data_source.fd, &cmd) < 0))
+			pw_log_warn("%p: read failed %m", this);
+		else if (SPA_UNLIKELY(cmd > 1))
+			pw_log_info("(%s-%u) client missed %"PRIu64" wakeups",
+				node->name, node->info.id, cmd - 1);
+
+		spa_log_trace_fp(this->log, "%p: got ready", this);
+		spa_node_call_ready(&this->callbacks, SPA_STATUS_HAVE_DATA);
+	}
+}
+
+static const struct spa_node_methods impl_node = {
+	SPA_VERSION_NODE_METHODS,
+	.add_listener = impl_node_add_listener,
+	.set_callbacks = impl_node_set_callbacks,
+	.sync = impl_node_sync,
+	.enum_params = impl_node_enum_params,
+	.set_param = impl_node_set_param,
+	.set_io = impl_node_set_io,
+	.send_command = impl_node_send_command,
+	.add_port = impl_node_add_port,
+	.remove_port = impl_node_remove_port,
+	.port_enum_params = impl_node_port_enum_params,
+	.port_set_param = impl_node_port_set_param,
+	.port_use_buffers = impl_node_port_use_buffers,
+	.port_set_io = impl_node_port_set_io,
+	.port_reuse_buffer = impl_node_port_reuse_buffer,
+	.process = impl_node_process,
+};
+
+static int
+node_init(struct node *this,
+	  struct spa_dict *info,
+	  const struct spa_support *support,
+	  uint32_t n_support)
+{
+	this->log = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_Log);
+	this->data_loop = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataLoop);
+	this->data_system = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataSystem);
+
+	if (this->data_loop == NULL) {
+		spa_log_error(this->log, "a data-loop is needed");
+		return -EINVAL;
+	}
+	if (this->data_system == NULL) {
+		spa_log_error(this->log, "a data-system is needed");
+		return -EINVAL;
+	}
+
+	this->node.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_Node,
+			SPA_VERSION_NODE,
+			&impl_node, this);
+	spa_hook_list_init(&this->hooks);
+
+	this->data_source.func = node_on_data_fd_events;
+	this->data_source.data = this;
+	this->data_source.fd = -1;
+	this->data_source.mask = SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP;
+	this->data_source.rmask = 0;
+
+	return 0;
+}
+
+static int node_clear(struct node *this)
+{
+	update_params(&this->params, 0, NULL);
+	return 0;
+}
+
+static int do_remove_source(struct spa_loop *loop,
+			    bool async,
+			    uint32_t seq,
+			    const void *data,
+			    size_t size,
+			    void *user_data)
+{
+	struct spa_source *source = user_data;
+	spa_loop_remove_source(loop, source);
+	return 0;
+}
+
+static void client_node_resource_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node *this = &impl->this;
+	struct node *node = &impl->node;
+
+	pw_log_debug("%p: destroy", node);
+
+	impl->node.resource = this->resource = NULL;
+	spa_hook_remove(&impl->resource_listener);
+	spa_hook_remove(&impl->object_listener);
+
+	if (node->data_source.fd != -1) {
+		spa_loop_invoke(node->data_loop,
+				do_remove_source,
+				SPA_ID_INVALID,
+				NULL,
+				0,
+				true,
+				&node->data_source);
+	}
+	if (this->node)
+		pw_impl_node_destroy(this->node);
+}
+
+static void client_node_resource_error(void *data, int seq, int res, const char *message)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct spa_result_node_error result;
+
+	pw_log_error("%p: error seq:%d %d (%s)", this, seq, res, message);
+	result.message = message;
+	spa_node_emit_result(&this->hooks, seq, res, SPA_RESULT_TYPE_NODE_ERROR, &result);
+}
+
+static void client_node_resource_pong(void *data, int seq)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	pw_log_debug("%p: got pong, emit result %d", this, seq);
+	spa_node_emit_result(&this->hooks, seq, 0, 0, NULL);
+}
+
+void pw_impl_client_node_registered(struct pw_impl_client_node *this, struct pw_global *global)
+{
+	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
+	struct pw_impl_node *node = this->node;
+	struct pw_impl_client *client = impl->node.client;
+	uint32_t node_id = global->id;
+
+	pw_log_debug("%p: %d", &impl->node, node_id);
+
+	impl->activation = pw_mempool_import_block(client->pool, node->activation);
+	if (impl->activation == NULL) {
+		pw_log_debug("%p: can't import block: %m", &impl->node);
+		return;
+	}
+	impl->node_id = node_id;
+
+	if (this->resource == NULL)
+		return;
+
+	pw_resource_set_bound_id(this->resource, node_id);
+
+	pw_client_node_resource_transport(this->resource,
+					  impl->other_fds[0],
+					  impl->other_fds[1],
+					  impl->activation->id,
+					  0,
+					  sizeof(struct pw_node_activation));
+
+	if (impl->bind_node_id) {
+		pw_global_bind(global, client, PW_PERM_ALL,
+				impl->bind_node_version, impl->bind_node_id);
+	}
+}
+
+static void node_initialized(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node *this = &impl->this;
+	struct node *node = &impl->node;
+	struct pw_global *global;
+	struct spa_system *data_system = impl->node.data_system;
+	size_t size;
+
+	impl->fds[0] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+	impl->fds[1] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+	impl->other_fds[0] = impl->fds[1];
+	impl->other_fds[1] = impl->fds[0];
+	node->data_source.fd = impl->fds[0];
+	node->writefd = impl->fds[1];
+
+	spa_loop_add_source(node->data_loop, &node->data_source);
+	pw_log_debug("%p: transport read-fd:%d write-fd:%d", node, impl->fds[0], impl->fds[1]);
+
+	size = sizeof(struct spa_io_buffers) * MAX_AREAS;
+
+	impl->io_areas = pw_mempool_alloc(impl->context->pool,
+			PW_MEMBLOCK_FLAG_READWRITE |
+			PW_MEMBLOCK_FLAG_MAP |
+			PW_MEMBLOCK_FLAG_SEAL,
+			SPA_DATA_MemFd, size);
+	if (impl->io_areas == NULL)
+                return;
+
+	pw_log_debug("%p: io areas %p", node, impl->io_areas->map->ptr);
+
+	if ((global = pw_impl_node_get_global(this->node)) != NULL)
+		pw_impl_client_node_registered(this, global);
+}
+
+static void node_free(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node *this = &impl->this;
+	struct node *node = &impl->node;
+	struct spa_system *data_system = node->data_system;
+	uint32_t tag[5] = { impl->node_id, };
+	struct pw_memmap *mm;
+
+	this->node = NULL;
+
+	pw_log_debug("%p: free", node);
+	node_clear(node);
+
+	spa_hook_remove(&impl->node_listener);
+
+	while ((mm = pw_mempool_find_tag(node->client->pool, tag, sizeof(uint32_t))) != NULL)
+		pw_memmap_free(mm);
+
+	if (this->resource)
+		pw_resource_destroy(this->resource);
+
+	if (impl->activation)
+		pw_memblock_unref(impl->activation);
+	if (impl->io_areas)
+		pw_memblock_unref(impl->io_areas);
+
+	pw_map_clear(&impl->node.ports[0]);
+	pw_map_clear(&impl->node.ports[1]);
+	pw_map_clear(&impl->io_map);
+
+	if (impl->fds[0] != -1)
+		spa_system_close(data_system, impl->fds[0]);
+	if (impl->fds[1] != -1)
+		spa_system_close(data_system, impl->fds[1]);
+	free(impl);
+}
+
+static int port_init_mix(void *data, struct pw_impl_port_mix *mix)
+{
+	struct port *port = data;
+	struct impl *impl = port->impl;
+	struct mix *m;
+
+	if ((m = ensure_mix(impl, port, mix->port.port_id)) == NULL)
+		return -ENOMEM;
+
+	mix->id = pw_map_insert_new(&impl->io_map, NULL);
+	if (mix->id == SPA_ID_INVALID) {
+		m->valid = false;
+		return -errno;
+	}
+	if (mix->id > MAX_AREAS) {
+		pw_map_remove(&impl->io_map, mix->id);
+		m->valid = false;
+		return -ENOMEM;
+	}
+
+	mix->io = SPA_PTROFF(impl->io_areas->map->ptr,
+			mix->id * sizeof(struct spa_io_buffers), void);
+	*mix->io = SPA_IO_BUFFERS_INIT;
+
+	m->peer_id = mix->peer_id;
+
+	pw_log_debug("%p: init mix id:%d io:%p base:%p", impl,
+			mix->id, mix->io, impl->io_areas->map->ptr);
+
+	return 0;
+}
+
+static int port_release_mix(void *data, struct pw_impl_port_mix *mix)
+{
+	struct port *port = data;
+	struct impl *impl = port->impl;
+	struct node *this = &impl->node;
+	struct mix *m;
+
+	pw_log_debug("%p: remove mix id:%d io:%p base:%p",
+			this, mix->id, mix->io, impl->io_areas->map->ptr);
+
+	if ((m = find_mix(port, mix->port.port_id)) == NULL || !m->valid)
+		return -EINVAL;
+
+	pw_map_remove(&impl->io_map, mix->id);
+	m->valid = false;
+
+	return 0;
+}
+
+static const struct pw_impl_port_implementation port_impl = {
+	PW_VERSION_PORT_IMPLEMENTATION,
+	.init_mix = port_init_mix,
+	.release_mix = port_release_mix,
+};
+
+static int
+impl_mix_port_enum_params(void *object, int seq,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t id, uint32_t start, uint32_t num,
+			   const struct spa_pod *filter)
+{
+	struct port *port = object;
+
+	if (port->direction != direction)
+		return -ENOTSUP;
+
+	return impl_node_port_enum_params(&port->node->node, seq, direction, port->id,
+			id, start, num, filter);
+}
+
+static int
+impl_mix_port_set_param(void *object,
+			enum spa_direction direction, uint32_t port_id,
+			uint32_t id, uint32_t flags,
+			const struct spa_pod *param)
+{
+	return -ENOTSUP;
+}
+
+static int
+impl_mix_add_port(void *object, enum spa_direction direction, uint32_t mix_id,
+		const struct spa_dict *props)
+{
+	struct port *port = object;
+	pw_log_debug("%p: add port %d:%d.%d", object, direction, port->id, mix_id);
+	return 0;
+}
+
+static int
+impl_mix_remove_port(void *object, enum spa_direction direction, uint32_t mix_id)
+{
+	struct port *port = object;
+	pw_log_debug("%p: remove port %d:%d.%d", object, direction, port->id, mix_id);
+	return 0;
+}
+
+static int
+impl_mix_port_use_buffers(void *object,
+			   enum spa_direction direction,
+			   uint32_t mix_id,
+			   uint32_t flags,
+			   struct spa_buffer **buffers,
+			   uint32_t n_buffers)
+{
+	struct port *port = object;
+	struct impl *impl = port->impl;
+
+	return do_port_use_buffers(impl, direction, port->id, mix_id, flags, buffers, n_buffers);
+}
+
+static int impl_mix_port_set_io(void *object,
+			   enum spa_direction direction, uint32_t mix_id,
+			   uint32_t id, void *data, size_t size)
+{
+	struct port *p = object;
+	struct pw_impl_port *port = p->port;
+	struct impl *impl = port->owner_data;
+	struct node *this = &impl->node;
+	struct pw_impl_port_mix *mix;
+
+	mix = pw_map_lookup(&port->mix_port_map, mix_id);
+	if (mix == NULL)
+		return -EINVAL;
+
+	if (id == SPA_IO_Buffers) {
+		if (data && size >= sizeof(struct spa_io_buffers))
+			mix->io = data;
+		else
+			mix->io = NULL;
+
+		if (mix->io != NULL && this->resource && this->resource->version >= 4)
+			pw_client_node_resource_port_set_mix_info(this->resource,
+						 direction, port->port_id,
+						 mix->port.port_id, mix->peer_id, NULL);
+	}
+
+	return do_port_set_io(impl,
+			      direction, port->port_id, mix->port.port_id,
+			      id, data, size);
+}
+
+static int
+impl_mix_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
+{
+	struct port *p = object;
+	return impl_node_port_reuse_buffer(&p->node->node, p->id, buffer_id);
+}
+
+static int impl_mix_process(void *object)
+{
+	return SPA_STATUS_HAVE_DATA;
+}
+
+static const struct spa_node_methods impl_port_mix = {
+	SPA_VERSION_NODE_METHODS,
+	.port_enum_params = impl_mix_port_enum_params,
+	.port_set_param = impl_mix_port_set_param,
+	.add_port = impl_mix_add_port,
+	.remove_port = impl_mix_remove_port,
+	.port_use_buffers = impl_mix_port_use_buffers,
+	.port_set_io = impl_mix_port_set_io,
+	.port_reuse_buffer = impl_mix_port_reuse_buffer,
+	.process = impl_mix_process,
+};
+
+static void node_port_init(void *data, struct pw_impl_port *port)
+{
+	struct impl *impl = data;
+	struct port *p = pw_impl_port_get_user_data(port);
+	struct node *this = &impl->node;
+
+	pw_log_debug("%p: port %p init", this, port);
+
+	*p = this->dummy;
+	p->port = port;
+	p->node = this;
+	p->direction = port->direction;
+	p->id = port->port_id;
+	p->impl = impl;
+	pw_array_init(&p->mix, sizeof(struct mix) * 2);
+	p->mix_node.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_Node,
+			SPA_VERSION_NODE,
+			&impl_port_mix, p);
+	ensure_mix(impl, p, SPA_ID_INVALID);
+
+	pw_map_insert_at(&this->ports[p->direction], p->id, p);
+	return;
+}
+
+static void node_port_added(void *data, struct pw_impl_port *port)
+{
+	struct impl *impl = data;
+	struct port *p = pw_impl_port_get_user_data(port);
+
+	port->flags |= PW_IMPL_PORT_FLAG_NO_MIXER;
+
+	port->impl = SPA_CALLBACKS_INIT(&port_impl, p);
+	port->owner_data = impl;
+
+	pw_impl_port_set_mix(port, &p->mix_node,
+			PW_IMPL_PORT_MIX_FLAG_MULTI |
+			PW_IMPL_PORT_MIX_FLAG_MIX_ONLY);
+}
+
+static void node_port_removed(void *data, struct pw_impl_port *port)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct port *p = pw_impl_port_get_user_data(port);
+
+	pw_log_debug("%p: port %p remove", this, port);
+
+	p->removed = true;
+	clear_port(this, p);
+}
+
+static void node_peer_added(void *data, struct pw_impl_node *peer)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct pw_memblock *m;
+
+	if (peer == impl->this.node)
+		return;
+
+	m = pw_mempool_import_block(this->client->pool, peer->activation);
+	if (m == NULL) {
+		pw_log_debug("%p: can't ensure mem: %m", this);
+		return;
+	}
+	pw_log_debug("%p: peer %p id:%u added mem_id:%u", &impl->this, peer,
+			peer->info.id, m->id);
+
+	if (this->resource == NULL)
+		return;
+
+	pw_client_node_resource_set_activation(this->resource,
+					  peer->info.id,
+					  peer->source.fd,
+					  m->id,
+					  0,
+					  sizeof(struct pw_node_activation));
+}
+
+static void node_peer_removed(void *data, struct pw_impl_node *peer)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	struct pw_memblock *m;
+
+	if (peer == impl->this.node)
+		return;
+
+	m = pw_mempool_find_fd(this->client->pool, peer->activation->fd);
+	if (m == NULL) {
+		pw_log_warn("%p: unknown peer %p fd:%d", this, peer,
+			peer->source.fd);
+		return;
+	}
+	pw_log_debug("%p: peer %p %u removed", this, peer,
+			peer->info.id);
+
+	if (this->resource != NULL) {
+		pw_client_node_resource_set_activation(this->resource,
+					  peer->info.id,
+					  -1,
+					  SPA_ID_INVALID,
+					  0,
+					  0);
+	}
+
+	pw_memblock_unref(m);
+}
+
+static void node_driver_changed(void *data, struct pw_impl_node *old, struct pw_impl_node *driver)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	pw_log_debug("%p: driver changed %p -> %p", this, old, driver);
+
+	node_peer_removed(data, old);
+	node_peer_added(data, driver);
+}
+
+static const struct pw_impl_node_events node_events = {
+	PW_VERSION_IMPL_NODE_EVENTS,
+	.free = node_free,
+	.initialized = node_initialized,
+	.port_init = node_port_init,
+	.port_added = node_port_added,
+	.port_removed = node_port_removed,
+	.peer_added = node_peer_added,
+	.peer_removed = node_peer_removed,
+	.driver_changed = node_driver_changed,
+};
+
+static const struct pw_resource_events resource_events = {
+	PW_VERSION_RESOURCE_EVENTS,
+	.destroy = client_node_resource_destroy,
+	.error = client_node_resource_error,
+	.pong = client_node_resource_pong,
+};
+
+/** Create a new client node
+ * \param client an owner \ref pw_client
+ * \param id an id
+ * \param name a name
+ * \param properties extra properties
+ * \return a newly allocated client node
+ *
+ * Create a new \ref pw_impl_node.
+ *
+ * \memberof pw_impl_client_node
+ */
+struct pw_impl_client_node *pw_impl_client_node_new(struct pw_resource *resource,
+					  struct pw_properties *properties,
+					  bool do_register)
+{
+	struct impl *impl;
+	struct pw_impl_client_node *this;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct pw_context *context = pw_impl_client_get_context(client);
+	const struct spa_support *support;
+	uint32_t n_support;
+	int res;
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL) {
+		res = -errno;
+		goto error_exit_cleanup;
+	}
+
+	if (properties == NULL)
+		properties = pw_properties_new(NULL, NULL);
+	if (properties == NULL) {
+		res = -errno;
+		goto error_exit_free;
+	}
+
+	pw_properties_setf(properties, PW_KEY_CLIENT_ID, "%d", client->global->id);
+
+	this = &impl->this;
+
+	impl->context = context;
+	impl->fds[0] = impl->fds[1] = -1;
+	pw_log_debug("%p: new", &impl->node);
+
+	support = pw_context_get_support(impl->context, &n_support);
+	node_init(&impl->node, NULL, support, n_support);
+	impl->node.impl = impl;
+	impl->node.resource = resource;
+	impl->node.client = client;
+	this->flags = do_register ? 0 : 1;
+
+	pw_map_init(&impl->node.ports[0], 64, 64);
+	pw_map_init(&impl->node.ports[1], 64, 64);
+	pw_map_init(&impl->io_map, 64, 64);
+
+	this->resource = resource;
+	this->node = pw_spa_node_new(context,
+				     PW_SPA_NODE_FLAG_ASYNC |
+				     (do_register ? 0 : PW_SPA_NODE_FLAG_NO_REGISTER),
+				     (struct spa_node *)&impl->node.node,
+				     NULL,
+				     properties, 0);
+
+	if (this->node == NULL)
+		goto error_no_node;
+
+	this->node->remote = true;
+	this->flags = 0;
+
+	pw_resource_add_listener(this->resource,
+				&impl->resource_listener,
+				&resource_events,
+				impl);
+	pw_resource_add_object_listener(this->resource,
+				&impl->object_listener,
+				&client_node_methods,
+				impl);
+
+	this->node->port_user_data_size = sizeof(struct port);
+
+	pw_impl_node_add_listener(this->node, &impl->node_listener, &node_events, impl);
+
+	return this;
+
+error_no_node:
+	res = -errno;
+	node_clear(&impl->node);
+	properties = NULL;
+	goto error_exit_free;
+
+error_exit_free:
+	free(impl);
+error_exit_cleanup:
+	if (resource)
+		pw_resource_destroy(resource);
+	pw_properties_free(properties);
+	errno = -res;
+	return NULL;
+}
+
+/** Destroy a client node
+ * \param node the client node to destroy
+ * \memberof pw_impl_client_node
+ */
+void pw_impl_client_node_destroy(struct pw_impl_client_node *node)
+{
+	pw_resource_destroy(node->resource);
+}
diff --git a/src/modules/module-client-node/client-node.h b/src/modules/module-client-node/client-node.h
new file mode 100644
index 0000000..f7e060a
--- /dev/null
+++ b/src/modules/module-client-node/client-node.h
@@ -0,0 +1,60 @@
+/* PipeWire
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef PIPEWIRE_CLIENT_NODE_H
+#define PIPEWIRE_CLIENT_NODE_H
+
+#include <pipewire/impl.h>
+#include <pipewire/extensions/client-node.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** \class pw_impl_client_node
+ *
+ * PipeWire client node interface
+ */
+struct pw_impl_client_node {
+	struct pw_impl_node *node;
+
+	struct pw_resource *resource;
+	uint32_t flags;
+};
+
+struct pw_impl_client_node *
+pw_impl_client_node_new(struct pw_resource *resource,
+		   struct pw_properties *properties,
+		   bool do_register);
+
+void
+pw_impl_client_node_destroy(struct pw_impl_client_node *node);
+
+void pw_impl_client_node_registered(struct pw_impl_client_node *node, struct pw_global *global);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PIPEWIRE_CLIENT_NODE_H */
diff --git a/src/modules/module-client-node/protocol-native.c b/src/modules/module-client-node/protocol-native.c
new file mode 100644
index 0000000..dd51692
--- /dev/null
+++ b/src/modules/module-client-node/protocol-native.c
@@ -0,0 +1,1259 @@
+/* PipeWire
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <errno.h>
+
+#include <spa/pod/builder.h>
+#include <spa/pod/parser.h>
+#include <spa/utils/result.h>
+
+#include <pipewire/impl.h>
+
+#include <pipewire/extensions/protocol-native.h>
+#include <pipewire/extensions/client-node.h>
+
+#define MAX_DICT	1024
+#define MAX_PARAMS	4096
+#define MAX_PARAM_INFO	128
+#define MAX_BUFFERS	64
+#define MAX_METAS	16u
+#define MAX_DATAS	64u
+
+PW_LOG_TOPIC_EXTERN(mod_topic);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+static inline void push_item(struct spa_pod_builder *b, const struct spa_dict_item *item)
+{
+	const char *str;
+	spa_pod_builder_string(b, item->key);
+	str = item->value;
+	if (spa_strstartswith(str, "pointer:"))
+		str = "";
+	spa_pod_builder_string(b, str);
+}
+
+static void push_dict(struct spa_pod_builder *b, const struct spa_dict *dict)
+{
+	uint32_t i, n_items;
+	struct spa_pod_frame f;
+
+	n_items = dict ? dict->n_items : 0;
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_int(b, n_items);
+	for (i = 0; i < n_items; i++)
+		push_item(b, &dict->items[i]);
+	spa_pod_builder_pop(b, &f);
+}
+
+static inline int parse_item(struct spa_pod_parser *prs, struct spa_dict_item *item)
+{
+	int res;
+	if ((res = spa_pod_parser_get(prs,
+		       SPA_POD_String(&item->key),
+		       SPA_POD_String(&item->value),
+		       NULL)) < 0)
+		return res;
+	if (spa_strstartswith(item->value, "pointer:"))
+		item->value = "";
+	return 0;
+}
+
+#define parse_dict(prs,d)								\
+do {											\
+	uint32_t i;									\
+	if (spa_pod_parser_get(prs,							\
+			 SPA_POD_Int(&(d)->n_items), NULL) < 0)				\
+		return -EINVAL;								\
+	(d)->items = NULL;								\
+	if ((d)->n_items > 0) {								\
+		if ((d)->n_items > MAX_DICT) 						\
+			return -ENOSPC;							\
+		(d)->items = alloca((d)->n_items * sizeof(struct spa_dict_item));	\
+		for (i = 0; i < (d)->n_items; i++) {					\
+			if (parse_item(prs, (struct spa_dict_item *) &(d)->items[i]) < 0)	\
+				return -EINVAL;						\
+		}									\
+	}										\
+} while(0)
+
+#define parse_dict_struct(prs,f,dict)						\
+do {										\
+	if (spa_pod_parser_push_struct(prs, f) < 0)				\
+		return -EINVAL;							\
+	parse_dict(prs, dict);							\
+	spa_pod_parser_pop(prs, f);						\
+} while(0)
+
+#define parse_params(prs,n_params,params)						\
+do {											\
+	uint32_t i;									\
+	if (spa_pod_parser_get(prs,							\
+			 SPA_POD_Int(&n_params), NULL) < 0)				\
+		return -EINVAL;								\
+	params = NULL;									\
+	if (n_params > 0) {								\
+		if (n_params > MAX_PARAMS)						\
+			return -ENOSPC;							\
+		params = alloca(n_params * sizeof(struct spa_pod *));			\
+		for (i = 0; i < n_params; i++) {					\
+			if (spa_pod_parser_get(prs,					\
+					SPA_POD_PodObject(&params[i]), NULL) < 0)	\
+				return -EINVAL;						\
+		}									\
+	}										\
+} while(0)
+
+#define parse_param_info(prs,n_params,params)						\
+do {											\
+	uint32_t i;									\
+	if (spa_pod_parser_get(prs,							\
+			SPA_POD_Int(&(n_params)), NULL) < 0)				\
+		return -EINVAL;								\
+	params = NULL;									\
+	if (n_params > 0) {								\
+		if (n_params > MAX_PARAM_INFO)						\
+			return -ENOSPC;							\
+		params = alloca(n_params * sizeof(struct spa_param_info));		\
+		for (i = 0; i < n_params; i++) {					\
+			if (spa_pod_parser_get(prs,					\
+					SPA_POD_Id(&(params[i]).id),			\
+					SPA_POD_Int(&(params[i]).flags), NULL) < 0)	\
+				return -EINVAL;						\
+		}									\
+	}										\
+} while(0)
+
+static int client_node_marshal_add_listener(void *object,
+			struct spa_hook *listener,
+			const struct pw_client_node_events *events,
+			void *data)
+{
+	struct pw_proxy *proxy = object;
+	pw_proxy_add_object_listener(proxy, listener, events, data);
+	return 0;
+}
+
+static struct pw_node *
+client_node_marshal_get_node(void *object, uint32_t version, size_t user_data_size)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+	struct pw_proxy *res;
+	uint32_t new_id;
+
+	res = pw_proxy_new(object, PW_TYPE_INTERFACE_Node, version, user_data_size);
+	if (res == NULL)
+		return NULL;
+
+	new_id = pw_proxy_get_id(res);
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_GET_NODE, NULL);
+
+	spa_pod_builder_add_struct(b,
+		       SPA_POD_Int(version),
+		       SPA_POD_Int(new_id));
+
+	pw_protocol_native_end_proxy(proxy, b);
+
+	return (struct pw_node *) res;
+}
+
+static int
+client_node_marshal_update(void *object,
+			   uint32_t change_mask,
+			   uint32_t n_params,
+			   const struct spa_pod **params,
+			   const struct spa_node_info *info)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f[2];
+	uint32_t i, n_items, n_info_params;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_UPDATE, NULL);
+
+	spa_pod_builder_push_struct(b, &f[0]);
+	spa_pod_builder_add(b,
+			SPA_POD_Int(change_mask),
+			SPA_POD_Int(n_params), NULL);
+
+	for (i = 0; i < n_params; i++)
+		spa_pod_builder_add(b, SPA_POD_Pod(params[i]), NULL);
+
+	if (info) {
+		uint64_t change_mask = info->change_mask;
+
+		change_mask &= SPA_NODE_CHANGE_MASK_FLAGS |
+				SPA_NODE_CHANGE_MASK_PROPS |
+				SPA_NODE_CHANGE_MASK_PARAMS;
+
+		n_items = info->props && (change_mask & SPA_NODE_CHANGE_MASK_PROPS) ?
+			info->props->n_items : 0;
+		n_info_params = (change_mask & SPA_NODE_CHANGE_MASK_PARAMS) ?
+			info->n_params : 0;
+
+		spa_pod_builder_push_struct(b, &f[1]);
+		spa_pod_builder_add(b,
+				    SPA_POD_Int(info->max_input_ports),
+				    SPA_POD_Int(info->max_output_ports),
+				    SPA_POD_Long(change_mask),
+				    SPA_POD_Long(info->flags),
+				    SPA_POD_Int(n_items), NULL);
+		for (i = 0; i < n_items; i++)
+			push_item(b, &info->props->items[i]);
+		spa_pod_builder_add(b,
+				    SPA_POD_Int(n_info_params), NULL);
+		for (i = 0; i < n_info_params; i++) {
+			spa_pod_builder_add(b,
+					    SPA_POD_Id(info->params[i].id),
+					    SPA_POD_Int(info->params[i].flags), NULL);
+		}
+		spa_pod_builder_pop(b, &f[1]);
+
+	} else {
+		spa_pod_builder_add(b,
+				SPA_POD_Pod(NULL), NULL);
+	}
+	spa_pod_builder_pop(b, &f[0]);
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int
+client_node_marshal_port_update(void *object,
+				enum spa_direction direction,
+				uint32_t port_id,
+				uint32_t change_mask,
+				uint32_t n_params,
+				const struct spa_pod **params,
+				const struct spa_port_info *info)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f[2];
+	uint32_t i, n_items;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_PORT_UPDATE, NULL);
+
+	spa_pod_builder_push_struct(b, &f[0]);
+	spa_pod_builder_add(b,
+			    SPA_POD_Int(direction),
+			    SPA_POD_Int(port_id),
+			    SPA_POD_Int(change_mask),
+			    SPA_POD_Int(n_params), NULL);
+
+	for (i = 0; i < n_params; i++)
+		spa_pod_builder_add(b,
+				SPA_POD_Pod(params[i]), NULL);
+
+	if (info) {
+		uint64_t change_mask = info->change_mask;
+
+		n_items = info->props ? info->props->n_items : 0;
+
+		change_mask &= SPA_PORT_CHANGE_MASK_FLAGS |
+				SPA_PORT_CHANGE_MASK_RATE |
+				SPA_PORT_CHANGE_MASK_PROPS |
+				SPA_PORT_CHANGE_MASK_PARAMS;
+
+		spa_pod_builder_push_struct(b, &f[1]);
+		spa_pod_builder_add(b,
+				    SPA_POD_Long(change_mask),
+				    SPA_POD_Long(info->flags),
+				    SPA_POD_Int(info->rate.num),
+				    SPA_POD_Int(info->rate.denom),
+				    SPA_POD_Int(n_items), NULL);
+		for (i = 0; i < n_items; i++)
+			push_item(b, &info->props->items[i]);
+		spa_pod_builder_add(b,
+				    SPA_POD_Int(info->n_params), NULL);
+		for (i = 0; i < info->n_params; i++) {
+			spa_pod_builder_add(b,
+					    SPA_POD_Id(info->params[i].id),
+					    SPA_POD_Int(info->params[i].flags), NULL);
+		}
+		spa_pod_builder_pop(b, &f[1]);
+
+	} else {
+		spa_pod_builder_add(b,
+				SPA_POD_Pod(NULL), NULL);
+	}
+	spa_pod_builder_pop(b, &f[0]);
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int client_node_marshal_set_active(void *object, bool active)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_SET_ACTIVE, NULL);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Bool(active));
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int client_node_marshal_event_method(void *object, const struct spa_event *event)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_EVENT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Pod(event));
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int
+client_node_marshal_port_buffers(void *object,
+				enum spa_direction direction,
+				uint32_t port_id,
+				uint32_t mix_id,
+				uint32_t n_buffers,
+				struct spa_buffer **buffers)
+{
+	struct pw_proxy *proxy = object;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f[2];
+	uint32_t i, j;
+
+	b = pw_protocol_native_begin_proxy(proxy, PW_CLIENT_NODE_METHOD_PORT_BUFFERS, NULL);
+
+	spa_pod_builder_push_struct(b, &f[0]);
+	spa_pod_builder_add(b,
+			    SPA_POD_Int(direction),
+			    SPA_POD_Int(port_id),
+			    SPA_POD_Int(mix_id),
+			    SPA_POD_Int(n_buffers), NULL);
+
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *buf = buffers[i];
+
+		spa_pod_builder_add(b,
+				SPA_POD_Int(buf->n_datas), NULL);
+
+		for (j = 0; j < buf->n_datas; j++) {
+			struct spa_data *d = &buf->datas[j];
+			spa_pod_builder_add(b,
+					SPA_POD_Id(d->type),
+					SPA_POD_Fd(pw_protocol_native_add_proxy_fd(proxy, d->fd)),
+					SPA_POD_Int(d->flags),
+					SPA_POD_Int(d->mapoffset),
+					SPA_POD_Int(d->maxsize), NULL);
+		}
+	}
+	spa_pod_builder_pop(b, &f[0]);
+
+	return pw_protocol_native_end_proxy(proxy, b);
+}
+
+static int client_node_demarshal_transport(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t mem_id, offset, sz;
+	int64_t ridx, widx;
+	int readfd, writefd;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Fd(&ridx),
+			SPA_POD_Fd(&widx),
+			SPA_POD_Int(&mem_id),
+			SPA_POD_Int(&offset),
+			SPA_POD_Int(&sz)) < 0)
+		return -EINVAL;
+
+	readfd = pw_protocol_native_get_proxy_fd(proxy, ridx);
+	writefd = pw_protocol_native_get_proxy_fd(proxy, widx);
+
+	if (readfd < 0 || writefd < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, transport, 0,
+								   readfd, writefd, mem_id,
+								   offset, sz);
+	return 0;
+}
+
+static int client_node_demarshal_set_param(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t id, flags;
+	const struct spa_pod *param = NULL;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Id(&id),
+			SPA_POD_Int(&flags),
+			SPA_POD_PodObject(&param)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, set_param, 0, id, flags, param);
+	return 0;
+}
+
+static int client_node_demarshal_event_event(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	const struct spa_event *event;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+				SPA_POD_PodObject(&event)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, event, 0, event);
+	return 0;
+}
+
+static int client_node_demarshal_command(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	const struct spa_command *command;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_PodObject(&command)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, command, 0, command);
+	return 0;
+}
+
+static int client_node_demarshal_add_port(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f[2];
+	int32_t direction, port_id;
+	struct spa_dict props = SPA_DICT_INIT(NULL, 0);
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f[0]) < 0)
+		return -EINVAL;
+
+	if (spa_pod_parser_get(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id), NULL) < 0)
+		return -EINVAL;
+
+	parse_dict_struct(&prs, &f[1], &props);
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, add_port, 0, direction, port_id,
+			props.n_items ? &props : NULL);
+	return 0;
+}
+
+static int client_node_demarshal_remove_port(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	int32_t direction, port_id;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, remove_port, 0, direction, port_id);
+	return 0;
+}
+
+static int client_node_demarshal_port_set_param(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t direction, port_id, id, flags;
+	const struct spa_pod *param = NULL;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Id(&id),
+			SPA_POD_Int(&flags),
+			SPA_POD_PodObject(&param)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, port_set_param, 0,
+			direction, port_id, id, flags, param);
+	return 0;
+}
+
+static int client_node_demarshal_port_use_buffers(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f;
+	uint32_t direction, port_id, mix_id, flags, n_buffers, data_id;
+	struct pw_client_node_buffer *buffers;
+	uint32_t i, j;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f) < 0 ||
+	    spa_pod_parser_get(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Int(&mix_id),
+			SPA_POD_Int(&flags),
+			SPA_POD_Int(&n_buffers), NULL) < 0)
+		return -EINVAL;
+
+	if (n_buffers > MAX_BUFFERS)
+		return -ENOSPC;
+
+	buffers = alloca(sizeof(struct pw_client_node_buffer) * n_buffers);
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *buf = buffers[i].buffer = alloca(sizeof(struct spa_buffer));
+
+		if (spa_pod_parser_get(&prs,
+				      SPA_POD_Int(&buffers[i].mem_id),
+				      SPA_POD_Int(&buffers[i].offset),
+				      SPA_POD_Int(&buffers[i].size),
+				      SPA_POD_Int(&buf->n_metas), NULL) < 0)
+			return -EINVAL;
+
+		if (buf->n_metas > MAX_METAS)
+			return -ENOSPC;
+
+		buf->metas = alloca(sizeof(struct spa_meta) * buf->n_metas);
+		for (j = 0; j < buf->n_metas; j++) {
+			struct spa_meta *m = &buf->metas[j];
+
+			if (spa_pod_parser_get(&prs,
+					      SPA_POD_Id(&m->type),
+					      SPA_POD_Int(&m->size), NULL) < 0)
+				return -EINVAL;
+
+			m->data = NULL;
+		}
+		if (spa_pod_parser_get(&prs,
+					SPA_POD_Int(&buf->n_datas), NULL) < 0)
+			return -EINVAL;
+
+		if (buf->n_datas > MAX_DATAS)
+			return -ENOSPC;
+
+		buf->datas = alloca(sizeof(struct spa_data) * buf->n_datas);
+		for (j = 0; j < buf->n_datas; j++) {
+			struct spa_data *d = &buf->datas[j];
+
+			if (spa_pod_parser_get(&prs,
+					      SPA_POD_Id(&d->type),
+					      SPA_POD_Int(&data_id),
+					      SPA_POD_Int(&d->flags),
+					      SPA_POD_Int(&d->mapoffset),
+					      SPA_POD_Int(&d->maxsize), NULL) < 0)
+				return -EINVAL;
+
+			d->fd = -1;
+			d->data = SPA_UINT32_TO_PTR(data_id);
+			d->chunk = NULL;
+		}
+	}
+	pw_proxy_notify(proxy, struct pw_client_node_events, port_use_buffers, 0,
+									  direction,
+									  port_id,
+									  mix_id,
+									  flags,
+									  n_buffers, buffers);
+	return 0;
+}
+
+static int client_node_demarshal_port_set_io(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t direction, port_id, mix_id, id, memid, off, sz;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Int(&mix_id),
+			SPA_POD_Id(&id),
+			SPA_POD_Int(&memid),
+			SPA_POD_Int(&off),
+			SPA_POD_Int(&sz)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, port_set_io, 0,
+							direction, port_id, mix_id,
+							id, memid,
+							off, sz);
+	return 0;
+}
+
+static int client_node_demarshal_set_activation(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t node_id, memid, off, sz;
+	int64_t sigidx;
+	int signalfd;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Int(&node_id),
+			SPA_POD_Fd(&sigidx),
+			SPA_POD_Int(&memid),
+			SPA_POD_Int(&off),
+			SPA_POD_Int(&sz)) < 0)
+		return -EINVAL;
+
+	signalfd = pw_protocol_native_get_proxy_fd(proxy, sigidx);
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, set_activation, 0,
+							node_id,
+							signalfd,
+							memid,
+							off, sz);
+	return 0;
+}
+
+static int client_node_demarshal_port_set_mix_info(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t direction, port_id, mix_id, peer_id;
+	struct spa_pod_frame f[2];
+	struct spa_dict props = SPA_DICT_INIT(NULL, 0);
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f[0]) < 0 ||
+	    spa_pod_parser_get(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Int(&mix_id),
+			SPA_POD_Int(&peer_id), NULL) < 0)
+		return -EINVAL;
+
+	parse_dict_struct(&prs, &f[1], &props);
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, port_set_mix_info, 1,
+							direction, port_id, mix_id,
+							peer_id, &props);
+	return 0;
+}
+
+static int client_node_demarshal_set_io(void *data, const struct pw_protocol_native_message *msg)
+{
+	struct pw_proxy *proxy = data;
+	struct spa_pod_parser prs;
+	uint32_t id, memid, off, sz;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_Id(&id),
+			SPA_POD_Int(&memid),
+			SPA_POD_Int(&off),
+			SPA_POD_Int(&sz)) < 0)
+		return -EINVAL;
+
+	pw_proxy_notify(proxy, struct pw_client_node_events, set_io, 0,
+			id, memid, off, sz);
+	return 0;
+}
+
+static int client_node_marshal_transport(void *data, int readfd, int writefd,
+		uint32_t mem_id, uint32_t offset, uint32_t size)
+{
+	struct pw_protocol_native_message *msg;
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_TRANSPORT, &msg);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Fd(pw_protocol_native_add_resource_fd(resource, readfd)),
+			       SPA_POD_Fd(pw_protocol_native_add_resource_fd(resource, writefd)),
+			       SPA_POD_Int(mem_id),
+			       SPA_POD_Int(offset),
+			       SPA_POD_Int(size));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_set_param(void *data, uint32_t id, uint32_t flags,
+			      const struct spa_pod *param)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_SET_PARAM, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Id(id),
+			       SPA_POD_Int(flags),
+			       SPA_POD_Pod(param));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int client_node_marshal_event_event(void *data, const struct spa_event *event)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_EVENT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Pod(event));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_command(void *data, const struct spa_command *command)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_COMMAND, NULL);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Pod(command));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_add_port(void *data,
+			     enum spa_direction direction, uint32_t port_id,
+			     const struct spa_dict *props)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_ADD_PORT, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			SPA_POD_Int(direction),
+			SPA_POD_Int(port_id), NULL);
+	push_dict(b, props);
+	spa_pod_builder_pop(b, &f);
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_remove_port(void *data,
+				enum spa_direction direction, uint32_t port_id)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_REMOVE_PORT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Int(direction),
+			       SPA_POD_Int(port_id));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_port_set_param(void *data,
+				   enum spa_direction direction,
+				   uint32_t port_id,
+				   uint32_t id,
+				   uint32_t flags,
+				   const struct spa_pod *param)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_PORT_SET_PARAM, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Int(direction),
+			       SPA_POD_Int(port_id),
+			       SPA_POD_Id(id),
+			       SPA_POD_Int(flags),
+			       SPA_POD_Pod(param));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_port_use_buffers(void *data,
+				     enum spa_direction direction,
+				     uint32_t port_id,
+				     uint32_t mix_id,
+				     uint32_t flags,
+				     uint32_t n_buffers, struct pw_client_node_buffer *buffers)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+	uint32_t i, j;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_PORT_USE_BUFFERS, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			SPA_POD_Int(direction),
+			SPA_POD_Int(port_id),
+			SPA_POD_Int(mix_id),
+			SPA_POD_Int(flags),
+			SPA_POD_Int(n_buffers), NULL);
+
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *buf = buffers[i].buffer;
+
+		spa_pod_builder_add(b,
+				    SPA_POD_Int(buffers[i].mem_id),
+				    SPA_POD_Int(buffers[i].offset),
+				    SPA_POD_Int(buffers[i].size),
+				    SPA_POD_Int(buf->n_metas), NULL);
+
+		for (j = 0; j < buf->n_metas; j++) {
+			struct spa_meta *m = &buf->metas[j];
+			spa_pod_builder_add(b,
+					    SPA_POD_Id(m->type),
+					    SPA_POD_Int(m->size), NULL);
+		}
+		spa_pod_builder_add(b,
+				SPA_POD_Int(buf->n_datas), NULL);
+		for (j = 0; j < buf->n_datas; j++) {
+			struct spa_data *d = &buf->datas[j];
+			spa_pod_builder_add(b,
+					    SPA_POD_Id(d->type),
+					    SPA_POD_Int(SPA_PTR_TO_UINT32(d->data)),
+					    SPA_POD_Int(d->flags),
+					    SPA_POD_Int(d->mapoffset),
+					    SPA_POD_Int(d->maxsize), NULL);
+		}
+	}
+	spa_pod_builder_pop(b, &f);
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_port_set_io(void *data,
+				uint32_t direction,
+				uint32_t port_id,
+				uint32_t mix_id,
+				uint32_t id,
+				uint32_t memid,
+				uint32_t offset,
+				uint32_t size)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_PORT_SET_IO, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Int(direction),
+			       SPA_POD_Int(port_id),
+			       SPA_POD_Int(mix_id),
+			       SPA_POD_Id(id),
+			       SPA_POD_Int(memid),
+			       SPA_POD_Int(offset),
+			       SPA_POD_Int(size));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_set_activation(void *data,
+				uint32_t node_id,
+				int signalfd,
+				uint32_t memid,
+				uint32_t offset,
+				uint32_t size)
+{
+	struct pw_protocol_native_message *msg;
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_SET_ACTIVATION, &msg);
+
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Int(node_id),
+			       SPA_POD_Fd(pw_protocol_native_add_resource_fd(resource, signalfd)),
+			       SPA_POD_Int(memid),
+			       SPA_POD_Int(offset),
+			       SPA_POD_Int(size));
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_set_io(void *data,
+			   uint32_t id,
+			   uint32_t memid,
+			   uint32_t offset,
+			   uint32_t size)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_SET_IO, NULL);
+	spa_pod_builder_add_struct(b,
+			       SPA_POD_Id(id),
+			       SPA_POD_Int(memid),
+			       SPA_POD_Int(offset),
+			       SPA_POD_Int(size));
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+static int
+client_node_marshal_port_set_mix_info(void *data,
+				uint32_t direction,
+				uint32_t port_id,
+				uint32_t mix_id,
+				uint32_t peer_id,
+				const struct spa_dict *props)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_EVENT_PORT_SET_MIX_INFO, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			SPA_POD_Int(direction),
+			SPA_POD_Int(port_id),
+			SPA_POD_Int(mix_id),
+			SPA_POD_Int(peer_id), NULL);
+	push_dict(b, props);
+	spa_pod_builder_pop(b, &f);
+
+	return pw_protocol_native_end_resource(resource, b);
+}
+
+
+static int client_node_demarshal_get_node(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	int32_t version, new_id;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+				SPA_POD_Int(&version),
+				SPA_POD_Int(&new_id)) < 0)
+		return -EINVAL;
+
+	return pw_resource_notify(resource, struct pw_client_node_methods, get_node, 0,
+			version, new_id);
+}
+
+static int client_node_demarshal_update(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f[2];
+	uint32_t change_mask, n_params;
+	const struct spa_pod **params = NULL;
+	struct spa_node_info info = SPA_NODE_INFO_INIT(), *infop = NULL;
+	struct spa_pod *ipod;
+	struct spa_dict props = SPA_DICT_INIT(NULL, 0);
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f[0]) < 0 ||
+	    spa_pod_parser_get(&prs,
+			SPA_POD_Int(&change_mask), NULL) < 0)
+		return -EINVAL;
+
+	parse_params(&prs, n_params, params);
+
+	if (spa_pod_parser_get(&prs,
+				SPA_POD_PodStruct(&ipod), NULL) < 0)
+		return -EINVAL;
+
+	if (ipod) {
+		struct spa_pod_parser p2;
+		struct spa_pod_frame f2;
+		infop = &info;
+
+		spa_pod_parser_pod(&p2, ipod);
+		if (spa_pod_parser_push_struct(&p2, &f2) < 0 ||
+		    spa_pod_parser_get(&p2,
+				SPA_POD_Int(&info.max_input_ports),
+				SPA_POD_Int(&info.max_output_ports),
+				SPA_POD_Long(&info.change_mask),
+				SPA_POD_Long(&info.flags), NULL) < 0)
+			return -EINVAL;
+
+		info.change_mask &= SPA_NODE_CHANGE_MASK_FLAGS |
+				SPA_NODE_CHANGE_MASK_PROPS |
+				SPA_NODE_CHANGE_MASK_PARAMS;
+
+		parse_dict(&p2, &props);
+		if (props.n_items > 0)
+			info.props = &props;
+
+		parse_param_info(&p2, info.n_params, info.params);
+	}
+
+	pw_resource_notify(resource, struct pw_client_node_methods, update, 0, change_mask,
+									n_params,
+									params, infop);
+	return 0;
+}
+
+static int client_node_demarshal_port_update(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f;
+	uint32_t direction, port_id, change_mask, n_params;
+	const struct spa_pod **params = NULL;
+	struct spa_port_info info = SPA_PORT_INFO_INIT(), *infop = NULL;
+	struct spa_pod *ipod;
+	struct spa_dict props = SPA_DICT_INIT(NULL, 0);
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f) < 0 ||
+	    spa_pod_parser_get(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Int(&change_mask), NULL) < 0)
+		return -EINVAL;
+
+	parse_params(&prs, n_params, params);
+
+	if (spa_pod_parser_get(&prs,
+				SPA_POD_PodStruct(&ipod), NULL) < 0)
+		return -EINVAL;
+
+	if (ipod) {
+		struct spa_pod_parser p2;
+		struct spa_pod_frame f2;
+		infop = &info;
+
+		spa_pod_parser_pod(&p2, ipod);
+		if (spa_pod_parser_push_struct(&p2, &f2) < 0 ||
+		    spa_pod_parser_get(&p2,
+				SPA_POD_Long(&info.change_mask),
+				SPA_POD_Long(&info.flags),
+				SPA_POD_Int(&info.rate.num),
+				SPA_POD_Int(&info.rate.denom), NULL) < 0)
+			return -EINVAL;
+
+		info.change_mask &= SPA_PORT_CHANGE_MASK_FLAGS |
+				SPA_PORT_CHANGE_MASK_RATE |
+				SPA_PORT_CHANGE_MASK_PROPS |
+				SPA_PORT_CHANGE_MASK_PARAMS;
+
+		parse_dict(&p2, &props);
+		if (props.n_items > 0)
+			info.props = &props;
+
+		parse_param_info(&p2, info.n_params, info.params);
+	}
+
+	pw_resource_notify(resource, struct pw_client_node_methods, port_update, 0, direction,
+									     port_id,
+									     change_mask,
+									     n_params,
+									     params, infop);
+	return 0;
+}
+
+static int client_node_demarshal_set_active(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	bool active;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+				SPA_POD_Bool(&active)) < 0)
+		return -EINVAL;
+
+	pw_resource_notify(resource, struct pw_client_node_methods, set_active, 0, active);
+	return 0;
+}
+
+static int client_node_demarshal_event_method(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	const struct spa_event *event;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			SPA_POD_PodObject(&event)) < 0)
+		return -EINVAL;
+
+	pw_resource_notify(resource, struct pw_client_node_methods, event, 0, event);
+	return 0;
+}
+
+static int client_node_demarshal_port_buffers(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f;
+	uint32_t i, j, direction, port_id, mix_id, n_buffers;
+	int64_t data_fd;
+	struct spa_buffer **buffers = NULL;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f) < 0 ||
+	    spa_pod_parser_get(&prs,
+			SPA_POD_Int(&direction),
+			SPA_POD_Int(&port_id),
+			SPA_POD_Int(&mix_id),
+			SPA_POD_Int(&n_buffers), NULL) < 0)
+		return -EINVAL;
+
+	if (n_buffers > MAX_BUFFERS)
+		return -ENOSPC;
+
+	buffers = alloca(sizeof(struct spa_buffer*) * n_buffers);
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *buf = buffers[i] = alloca(sizeof(struct spa_buffer));
+
+		spa_zero(*buf);
+		if (spa_pod_parser_get(&prs,
+					SPA_POD_Int(&buf->n_datas), NULL) < 0)
+			return -EINVAL;
+
+		if (buf->n_datas > MAX_DATAS)
+			return -ENOSPC;
+
+		buf->datas = alloca(sizeof(struct spa_data) * buf->n_datas);
+		for (j = 0; j < buf->n_datas; j++) {
+			struct spa_data *d = &buf->datas[j];
+
+			if (spa_pod_parser_get(&prs,
+					      SPA_POD_Id(&d->type),
+					      SPA_POD_Fd(&data_fd),
+					      SPA_POD_Int(&d->flags),
+					      SPA_POD_Int(&d->mapoffset),
+					      SPA_POD_Int(&d->maxsize), NULL) < 0)
+				return -EINVAL;
+
+			d->fd = pw_protocol_native_get_resource_fd(resource, data_fd);
+		}
+	}
+
+	pw_resource_notify(resource, struct pw_client_node_methods, port_buffers, 0,
+			direction, port_id, mix_id, n_buffers, buffers);
+
+	return 0;
+}
+
+static const struct pw_client_node_methods pw_protocol_native_client_node_method_marshal = {
+	PW_VERSION_CLIENT_NODE_METHODS,
+	.add_listener = &client_node_marshal_add_listener,
+	.get_node = &client_node_marshal_get_node,
+	.update = &client_node_marshal_update,
+	.port_update = &client_node_marshal_port_update,
+	.set_active = &client_node_marshal_set_active,
+	.event = &client_node_marshal_event_method,
+	.port_buffers = &client_node_marshal_port_buffers
+};
+
+static const struct pw_protocol_native_demarshal
+pw_protocol_native_client_node_method_demarshal[PW_CLIENT_NODE_METHOD_NUM] =
+{
+	[PW_CLIENT_NODE_METHOD_ADD_LISTENER] = { NULL, 0 },
+	[PW_CLIENT_NODE_METHOD_GET_NODE] = { &client_node_demarshal_get_node, 0 },
+	[PW_CLIENT_NODE_METHOD_UPDATE] = { &client_node_demarshal_update, 0 },
+	[PW_CLIENT_NODE_METHOD_PORT_UPDATE] = { &client_node_demarshal_port_update, 0 },
+	[PW_CLIENT_NODE_METHOD_SET_ACTIVE] = { &client_node_demarshal_set_active, 0 },
+	[PW_CLIENT_NODE_METHOD_EVENT] = { &client_node_demarshal_event_method, 0 },
+	[PW_CLIENT_NODE_METHOD_PORT_BUFFERS] = { &client_node_demarshal_port_buffers, 0 }
+};
+
+static const struct pw_client_node_events pw_protocol_native_client_node_event_marshal = {
+	PW_VERSION_CLIENT_NODE_EVENTS,
+	.transport = &client_node_marshal_transport,
+	.set_param = &client_node_marshal_set_param,
+	.set_io = &client_node_marshal_set_io,
+	.event = &client_node_marshal_event_event,
+	.command = &client_node_marshal_command,
+	.add_port = &client_node_marshal_add_port,
+	.remove_port = &client_node_marshal_remove_port,
+	.port_set_param = &client_node_marshal_port_set_param,
+	.port_use_buffers = &client_node_marshal_port_use_buffers,
+	.port_set_io = &client_node_marshal_port_set_io,
+	.set_activation = &client_node_marshal_set_activation,
+	.port_set_mix_info = &client_node_marshal_port_set_mix_info,
+};
+
+static const struct pw_protocol_native_demarshal
+pw_protocol_native_client_node_event_demarshal[PW_CLIENT_NODE_EVENT_NUM] =
+{
+	[PW_CLIENT_NODE_EVENT_TRANSPORT] = { &client_node_demarshal_transport, 0 },
+	[PW_CLIENT_NODE_EVENT_SET_PARAM] = { &client_node_demarshal_set_param, 0 },
+	[PW_CLIENT_NODE_EVENT_SET_IO] = { &client_node_demarshal_set_io, 0 },
+	[PW_CLIENT_NODE_EVENT_EVENT] = { &client_node_demarshal_event_event, 0 },
+	[PW_CLIENT_NODE_EVENT_COMMAND] = { &client_node_demarshal_command, 0 },
+	[PW_CLIENT_NODE_EVENT_ADD_PORT] = { &client_node_demarshal_add_port, 0 },
+	[PW_CLIENT_NODE_EVENT_REMOVE_PORT] = { &client_node_demarshal_remove_port, 0 },
+	[PW_CLIENT_NODE_EVENT_PORT_SET_PARAM] = { &client_node_demarshal_port_set_param, 0 },
+	[PW_CLIENT_NODE_EVENT_PORT_USE_BUFFERS] = { &client_node_demarshal_port_use_buffers, 0 },
+	[PW_CLIENT_NODE_EVENT_PORT_SET_IO] = { &client_node_demarshal_port_set_io, 0 },
+	[PW_CLIENT_NODE_EVENT_SET_ACTIVATION] = { &client_node_demarshal_set_activation, 0 },
+	[PW_CLIENT_NODE_EVENT_PORT_SET_MIX_INFO] = { &client_node_demarshal_port_set_mix_info, 0 }
+};
+
+static const struct pw_protocol_marshal pw_protocol_native_client_node_marshal = {
+	PW_TYPE_INTERFACE_ClientNode,
+	PW_VERSION_CLIENT_NODE,
+	0,
+	PW_CLIENT_NODE_METHOD_NUM,
+	PW_CLIENT_NODE_EVENT_NUM,
+	.client_marshal = &pw_protocol_native_client_node_method_marshal,
+	.server_demarshal = &pw_protocol_native_client_node_method_demarshal,
+	.server_marshal = &pw_protocol_native_client_node_event_marshal,
+	.client_demarshal = pw_protocol_native_client_node_event_demarshal,
+};
+
+struct pw_protocol *pw_protocol_native_ext_client_node_init(struct pw_context *context)
+{
+	struct pw_protocol *protocol;
+
+	protocol = pw_context_find_protocol(context, PW_TYPE_INFO_PROTOCOL_Native);
+
+	if (protocol == NULL)
+		return NULL;
+
+	pw_protocol_add_marshal(protocol, &pw_protocol_native_client_node_marshal);
+
+	return protocol;
+}
diff --git a/src/modules/module-client-node/remote-node.c b/src/modules/module-client-node/remote-node.c
new file mode 100644
index 0000000..051ab07
--- /dev/null
+++ b/src/modules/module-client-node/remote-node.c
@@ -0,0 +1,1339 @@
+/* PipeWire
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <errno.h>
+#include <time.h>
+#include <sys/mman.h>
+
+#include <spa/pod/parser.h>
+#include <spa/pod/dynamic.h>
+#include <spa/node/utils.h>
+#include <spa/utils/result.h>
+#include <spa/debug/types.h>
+
+#include "pipewire/pipewire.h"
+#include "pipewire/private.h"
+
+#include "pipewire/extensions/protocol-native.h"
+#include "pipewire/extensions/client-node.h"
+
+#define MAX_BUFFERS	64
+
+PW_LOG_TOPIC_EXTERN(mod_topic);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+/** \cond */
+static bool mlock_warned = false;
+
+struct buffer {
+	uint32_t id;
+	struct spa_buffer *buf;
+	struct pw_memmap *mem;
+};
+
+struct mix {
+	struct spa_list link;
+	struct pw_impl_port *port;
+	uint32_t mix_id;
+	struct pw_impl_port_mix mix;
+	struct pw_array buffers;
+	bool active;
+};
+
+struct node_data {
+	struct pw_context *context;
+
+	struct pw_mempool *pool;
+
+	uint32_t remote_id;
+	int rtwritefd;
+	struct pw_memmap *activation;
+
+	struct spa_list mix[2];
+	struct spa_list free_mix;
+
+	struct pw_impl_node *node;
+	struct spa_hook node_listener;
+	unsigned int do_free:1;
+	unsigned int have_transport:1;
+	unsigned int allow_mlock:1;
+	unsigned int warn_mlock:1;
+
+	struct pw_client_node *client_node;
+	struct spa_hook client_node_listener;
+	struct spa_hook proxy_client_node_listener;
+
+	struct spa_list links;
+};
+
+struct link {
+	struct spa_list link;
+	struct node_data *data;
+	struct pw_memmap *map;
+	struct pw_node_target target;
+	uint32_t node_id;
+	int signalfd;
+};
+
+/** \endcond */
+
+static struct link *find_activation(struct spa_list *links, uint32_t node_id)
+{
+	struct link *l;
+
+	spa_list_for_each(l, links, link) {
+		if (l->node_id == node_id)
+			return l;
+	}
+	return NULL;
+}
+
+static int
+do_deactivate_link(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct link *link = user_data;
+	pw_log_trace("link %p deactivate", link);
+	spa_list_remove(&link->target.link);
+	return 0;
+}
+
+static void clear_link(struct node_data *data, struct link *link)
+{
+	struct pw_context *context = data->context;
+	pw_log_debug("link %p", link);
+	pw_loop_invoke(context->data_loop,
+		do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, link);
+	pw_memmap_free(link->map);
+	spa_system_close(context->data_system, link->signalfd);
+	spa_list_remove(&link->link);
+	free(link);
+}
+
+static void clean_transport(struct node_data *data)
+{
+	struct link *l;
+	uint32_t tag[5] = { data->remote_id, };
+	struct pw_memmap *mm;
+
+	if (!data->have_transport)
+		return;
+
+	spa_list_consume(l, &data->links, link)
+		clear_link(data, l);
+
+	while ((mm = pw_mempool_find_tag(data->pool, tag, sizeof(uint32_t))) != NULL) {
+		if (mm->tag[1] == SPA_ID_INVALID)
+			spa_node_set_io(data->node->node, mm->tag[2], NULL, 0);
+
+		pw_memmap_free(mm);
+	}
+
+	pw_memmap_free(data->activation);
+	data->node->rt.activation = data->node->activation->map->ptr;
+
+	spa_system_close(data->context->data_system, data->rtwritefd);
+	data->have_transport = false;
+}
+
+static void mix_init(struct mix *mix, struct pw_impl_port *port, uint32_t mix_id)
+{
+	pw_log_debug("port %p: mix init %d.%d", port, port->port_id, mix_id);
+	mix->port = port;
+	mix->mix_id = mix_id;
+	pw_impl_port_init_mix(port, &mix->mix);
+	mix->active = false;
+	pw_array_init(&mix->buffers, 32);
+	pw_array_ensure_size(&mix->buffers, sizeof(struct buffer) * 64);
+}
+
+static int
+do_deactivate_mix(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct mix *mix = user_data;
+	spa_list_remove(&mix->mix.rt_link);
+        return 0;
+}
+
+static int
+deactivate_mix(struct node_data *data, struct mix *mix)
+{
+	if (mix->active) {
+		pw_log_debug("node %p: mix %p deactivate", data, mix);
+		pw_loop_invoke(data->context->data_loop,
+                       do_deactivate_mix, SPA_ID_INVALID, NULL, 0, true, mix);
+		mix->active = false;
+	}
+	return 0;
+}
+
+static int
+do_activate_mix(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct mix *mix = user_data;
+
+	spa_list_append(&mix->port->rt.mix_list, &mix->mix.rt_link);
+        return 0;
+}
+
+static int
+activate_mix(struct node_data *data, struct mix *mix)
+{
+	if (!mix->active) {
+		pw_log_debug("node %p: mix %p activate", data, mix);
+		pw_loop_invoke(data->context->data_loop,
+                       do_activate_mix, SPA_ID_INVALID, NULL, 0, false, mix);
+		mix->active = true;
+	}
+	return 0;
+}
+
+static struct mix *find_mix(struct node_data *data,
+		enum spa_direction direction, uint32_t port_id, uint32_t mix_id)
+{
+	struct mix *mix;
+
+	spa_list_for_each(mix, &data->mix[direction], link) {
+		if (mix->port->port_id == port_id &&
+		    mix->mix_id == mix_id) {
+			pw_log_debug("port %p: found mix %d:%d.%d", mix->port,
+					direction, port_id, mix_id);
+			return mix;
+		}
+	}
+	return NULL;
+}
+
+static struct mix *ensure_mix(struct node_data *data,
+		enum spa_direction direction, uint32_t port_id, uint32_t mix_id)
+{
+	struct mix *mix;
+	struct pw_impl_port *port;
+
+	if ((mix = find_mix(data, direction, port_id, mix_id)))
+		return mix;
+
+	port = pw_impl_node_find_port(data->node, direction, port_id);
+	if (port == NULL)
+		return NULL;
+
+	if (spa_list_is_empty(&data->free_mix)) {
+		if ((mix = calloc(1, sizeof(*mix))) == NULL)
+			return NULL;
+	} else {
+		mix = spa_list_first(&data->free_mix, struct mix, link);
+		spa_list_remove(&mix->link);
+	}
+
+	mix_init(mix, port, mix_id);
+	spa_list_append(&data->mix[direction], &mix->link);
+
+	return mix;
+}
+
+
+static int client_node_transport(void *_data,
+			int readfd, int writefd, uint32_t mem_id, uint32_t offset, uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+
+	clean_transport(data);
+
+	data->activation = pw_mempool_map_id(data->pool, mem_id,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
+	if (data->activation == NULL) {
+		pw_log_warn("remote-node %p: can't map activation: %m", proxy);
+		return -errno;
+	}
+
+	data->node->rt.activation = data->activation->ptr;
+
+	pw_log_debug("remote-node %p: fds:%d %d node:%u activation:%p",
+		proxy, readfd, writefd, data->remote_id, data->activation->ptr);
+
+	data->rtwritefd = writefd;
+	spa_system_close(data->context->data_system, data->node->source.fd);
+	data->node->source.fd = readfd;
+
+	data->have_transport = true;
+
+	if (data->node->active)
+		pw_client_node_set_active(data->client_node, true);
+
+	return 0;
+}
+
+static int add_node_update(struct node_data *data, uint32_t change_mask, uint32_t info_mask)
+{
+	struct pw_impl_node *node = data->node;
+	struct spa_node_info ni = SPA_NODE_INFO_INIT();
+	uint32_t n_params = 0;
+	struct spa_pod **params = NULL;
+	int res;
+
+	if (change_mask & PW_CLIENT_NODE_UPDATE_PARAMS) {
+		uint32_t i, idx, id;
+		uint8_t buf[4096];
+		struct spa_pod_dynamic_builder b;
+
+		for (i = 0; i < node->info.n_params; i++) {
+			struct spa_pod *param;
+
+			id = node->info.params[i].id;
+			if (id == SPA_PARAM_Invalid)
+				continue;
+
+			for (idx = 0;;) {
+				spa_pod_dynamic_builder_init(&b, buf, sizeof(buf), 4096);
+
+	                        res = spa_node_enum_params_sync(node->node,
+							id, &idx, NULL, &param, &b.b);
+				if (res == 1) {
+					void *p;
+					p = pw_reallocarray(params, n_params + 1, sizeof(struct spa_pod *));
+					if (p == NULL) {
+						res = -errno;
+						pw_log_error("realloc failed: %m");
+					} else {
+						params = p;
+						params[n_params++] = spa_pod_copy(param);
+					}
+				}
+				spa_pod_dynamic_builder_clean(&b);
+				if (res != 1)
+	                                break;
+			}
+                }
+	}
+	if (change_mask & PW_CLIENT_NODE_UPDATE_INFO) {
+		ni.max_input_ports = node->info.max_input_ports;
+		ni.max_output_ports = node->info.max_output_ports;
+		ni.change_mask = info_mask;
+		ni.flags = node->spa_flags;
+		ni.props = node->info.props;
+		ni.params = node->info.params;
+		ni.n_params = node->info.n_params;
+	}
+
+        res = pw_client_node_update(data->client_node,
+				change_mask,
+				n_params,
+				(const struct spa_pod **)params,
+				&ni);
+
+	if (params) {
+		while (n_params > 0)
+			free(params[--n_params]);
+		free(params);
+	}
+	return res;
+}
+
+static int add_port_update(struct node_data *data, struct pw_impl_port *port, uint32_t change_mask)
+{
+	struct spa_port_info pi = SPA_PORT_INFO_INIT();
+	uint32_t n_params = 0;
+	struct spa_pod **params = NULL;
+	int res;
+
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_PARAMS) {
+		uint32_t i, idx, id;
+		uint8_t buf[4096];
+		struct spa_pod_dynamic_builder b;
+
+		for (i = 0; i < port->info.n_params; i++) {
+			struct spa_pod *param;
+
+			id = port->info.params[i].id;
+			if (id == SPA_PARAM_Invalid)
+				continue;
+
+			for (idx = 0;;) {
+				spa_pod_dynamic_builder_init(&b, buf, sizeof(buf), 4096);
+
+	                        res = spa_node_port_enum_params_sync(port->node->node,
+							port->direction, port->port_id,
+							id, &idx, NULL, &param, &b.b);
+				if (res == 1) {
+					void *p;
+					p = pw_reallocarray(params, n_params + 1, sizeof(struct spa_pod*));
+					if (p == NULL) {
+						res = -errno;
+						pw_log_error("realloc failed: %m");
+					} else {
+						params = p;
+						params[n_params++] = spa_pod_copy(param);
+					}
+				}
+				spa_pod_dynamic_builder_clean(&b);
+
+				if (res != 1)
+	                                break;
+
+			}
+                }
+	}
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_INFO) {
+		pi.change_mask = SPA_PORT_CHANGE_MASK_FLAGS |
+			SPA_PORT_CHANGE_MASK_RATE |
+			SPA_PORT_CHANGE_MASK_PROPS |
+			SPA_PORT_CHANGE_MASK_PARAMS;
+		pi.flags = port->spa_flags;
+		pi.rate = SPA_FRACTION(0, 1);
+		pi.props = &port->properties->dict;
+		SPA_FLAG_CLEAR(pi.flags, SPA_PORT_FLAG_DYNAMIC_DATA);
+		pi.n_params = port->info.n_params;
+		pi.params = port->info.params;
+	}
+
+	res = pw_client_node_port_update(data->client_node,
+                                         port->direction,
+                                         port->port_id,
+                                         change_mask,
+                                         n_params,
+                                         (const struct spa_pod **)params,
+					 &pi);
+	if (params) {
+		while (n_params > 0)
+			free(params[--n_params]);
+		free(params);
+	}
+	return res;
+}
+
+static int
+client_node_set_param(void *_data, uint32_t id, uint32_t flags,
+		      const struct spa_pod *param)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	int res;
+
+	pw_log_debug("node %p: set_param %s:", proxy,
+			spa_debug_type_find_name(spa_type_param, id));
+
+	res = spa_node_set_param(data->node->node, id, flags, param);
+
+	if (res < 0) {
+		pw_log_error("node %p: set_param %s (%d) %p: %s", proxy,
+				spa_debug_type_find_name(spa_type_param, id),
+				id, param, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "node_set_param(%s) failed: %s",
+				spa_debug_type_find_name(spa_type_param, id),
+				spa_strerror(res));
+	}
+	return res;
+}
+
+static int
+client_node_set_io(void *_data,
+		   uint32_t id,
+		   uint32_t memid,
+		   uint32_t offset,
+		   uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_memmap *old, *mm;
+	void *ptr;
+	uint32_t tag[5] = { data->remote_id, SPA_ID_INVALID, id, };
+	int res;
+
+	old = pw_mempool_find_tag(data->pool, tag, sizeof(tag));
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	} else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
+		if (mm == NULL) {
+			pw_log_warn("can't map memory id %u: %m", memid);
+			res = -errno;
+			goto exit;
+		}
+		ptr = mm->ptr;
+	}
+
+	pw_log_debug("node %p: set io %s %p", proxy,
+			spa_debug_type_find_name(spa_type_io, id), ptr);
+
+	res =  spa_node_set_io(data->node->node, id, ptr, size);
+
+	pw_memmap_free(old);
+exit:
+	if (res < 0) {
+		pw_log_error("node %p: set_io: %s", proxy, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "node_set_io failed: %s", spa_strerror(res));
+	}
+	return res;
+}
+
+static int client_node_event(void *data, const struct spa_event *event)
+{
+	pw_log_warn("unhandled node event %d", SPA_EVENT_TYPE(event));
+	return -ENOTSUP;
+}
+
+static int client_node_command(void *_data, const struct spa_command *command)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	int res;
+
+	switch (SPA_NODE_COMMAND_ID(command)) {
+	case SPA_NODE_COMMAND_Pause:
+		pw_log_debug("node %p: pause", proxy);
+
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_IDLE)) < 0) {
+			pw_log_warn("node %p: pause failed", proxy);
+			pw_proxy_error(proxy, res, "pause failed");
+		}
+
+		break;
+	case SPA_NODE_COMMAND_Start:
+		pw_log_debug("node %p: start", proxy);
+
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_RUNNING)) < 0) {
+			pw_log_warn("node %p: start failed", proxy);
+			pw_proxy_error(proxy, res, "start failed");
+		}
+		break;
+
+	case SPA_NODE_COMMAND_Suspend:
+		pw_log_debug("node %p: suspend", proxy);
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_SUSPENDED)) < 0) {
+			pw_log_warn("node %p: suspend failed", proxy);
+			pw_proxy_error(proxy, res, "suspend failed");
+		}
+		break;
+	case SPA_NODE_COMMAND_RequestProcess:
+		res = pw_impl_node_send_command(data->node, command);
+		break;
+	default:
+		pw_log_warn("unhandled node command %d", SPA_NODE_COMMAND_ID(command));
+		res = -ENOTSUP;
+		pw_proxy_errorf(proxy, res, "command %d not supported", SPA_NODE_COMMAND_ID(command));
+	}
+	return res;
+}
+
+static int
+client_node_add_port(void *_data, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	pw_log_warn("add port not supported");
+	pw_proxy_error(proxy, -ENOTSUP, "add port not supported");
+	return -ENOTSUP;
+}
+
+static int
+client_node_remove_port(void *_data, enum spa_direction direction, uint32_t port_id)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	pw_log_warn("remove port not supported");
+	pw_proxy_error(proxy, -ENOTSUP, "remove port not supported");
+	return -ENOTSUP;
+}
+
+static int clear_buffers(struct node_data *data, struct mix *mix)
+{
+	struct pw_impl_port *port = mix->port;
+        struct buffer *b;
+	int res;
+
+        pw_log_debug("port %p: clear %zd buffers mix:%d", port,
+			pw_array_get_len(&mix->buffers, struct buffer *),
+			mix->mix_id);
+
+	if ((res = pw_impl_port_use_buffers(port, &mix->mix, 0, NULL, 0)) < 0) {
+		pw_log_error("port %p: error clear buffers %s", port, spa_strerror(res));
+		return res;
+	}
+
+        pw_array_for_each(b, &mix->buffers) {
+		pw_log_debug("port %p: clear buffer %d map %p %p",
+			port, b->id, b->mem, b->buf);
+		pw_memmap_free(b->mem);
+		free(b->buf);
+        }
+	mix->buffers.size = 0;
+	return 0;
+}
+
+static int
+client_node_port_set_param(void *_data,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t id, uint32_t flags,
+			   const struct spa_pod *param)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_impl_port *port;
+	int res;
+
+	port = pw_impl_node_find_port(data->node, direction, port_id);
+	if (port == NULL) {
+		res = -EINVAL;
+		goto error_exit;
+	}
+
+	pw_log_debug("port %p: set_param %s %p", port,
+			spa_debug_type_find_name(spa_type_param, id), param);
+
+	res = pw_impl_port_set_param(port, id, flags, param);
+	if (res < 0)
+		goto error_exit;
+
+	if (id == SPA_PARAM_Format) {
+		struct mix *mix;
+		spa_list_for_each(mix, &data->mix[direction], link) {
+			if (mix->port->port_id == port_id)
+				clear_buffers(data, mix);
+		}
+	}
+	return res;
+
+error_exit:
+	pw_log_error("port %p: set_param %d %p: %s", port, id, param, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "port_set_param(%s) failed: %s",
+				spa_debug_type_find_name(spa_type_param, id),
+				spa_strerror(res));
+	return res;
+}
+
+static int
+client_node_port_use_buffers(void *_data,
+			     enum spa_direction direction, uint32_t port_id, uint32_t mix_id,
+			     uint32_t flags,
+			     uint32_t n_buffers, struct pw_client_node_buffer *buffers)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct buffer *bid;
+	uint32_t i, j;
+	struct spa_buffer *b, **bufs;
+	struct mix *mix;
+	int res, prot;
+
+	mix = ensure_mix(data, direction, port_id, mix_id);
+	if (mix == NULL) {
+		res = -ENOENT;
+		goto error_exit;
+	}
+
+	if (n_buffers > MAX_BUFFERS)
+		return -ENOSPC;
+
+	prot = PW_MEMMAP_FLAG_READWRITE;
+
+	/* clear previous buffers */
+	clear_buffers(data, mix);
+
+	bufs = alloca(n_buffers * sizeof(struct spa_buffer *));
+
+	for (i = 0; i < n_buffers; i++) {
+		size_t size;
+		off_t offset;
+		struct pw_memmap *mm;
+
+		mm = pw_mempool_map_id(data->pool, buffers[i].mem_id,
+				prot, buffers[i].offset, buffers[i].size, NULL);
+		if (mm == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+
+		bid = pw_array_add(&mix->buffers, sizeof(struct buffer));
+		if (bid == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+		bid->id = i;
+		bid->mem = mm;
+
+		if (data->allow_mlock && mlock(mm->ptr, mm->size) < 0)
+			if (errno != ENOMEM || !mlock_warned) {
+				pw_log(data->warn_mlock ? SPA_LOG_LEVEL_WARN : SPA_LOG_LEVEL_DEBUG,
+						"Failed to mlock memory %p %u: %s",
+						mm->ptr, mm->size,
+						errno == ENOMEM ?
+						"This is not a problem but for best performance, "
+						"consider increasing RLIMIT_MEMLOCK" : strerror(errno));
+				mlock_warned |= errno == ENOMEM;
+			}
+
+		size = sizeof(struct spa_buffer);
+		for (j = 0; j < buffers[i].buffer->n_metas; j++)
+			size += sizeof(struct spa_meta);
+		for (j = 0; j < buffers[i].buffer->n_datas; j++)
+			size += sizeof(struct spa_data);
+
+		b = bid->buf = malloc(size);
+		if (b == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+		memcpy(b, buffers[i].buffer, sizeof(struct spa_buffer));
+
+		b->metas = SPA_PTROFF(b, sizeof(struct spa_buffer), struct spa_meta);
+		b->datas = SPA_PTROFF(b->metas, sizeof(struct spa_meta) * b->n_metas,
+				       struct spa_data);
+
+		pw_log_debug("add buffer mem:%d id:%d offset:%u size:%u %p", mm->block->id,
+				bid->id, buffers[i].offset, buffers[i].size, bid->buf);
+
+		offset = 0;
+		for (j = 0; j < b->n_metas; j++) {
+			struct spa_meta *m = &b->metas[j];
+			memcpy(m, &buffers[i].buffer->metas[j], sizeof(struct spa_meta));
+			m->data = SPA_PTROFF(mm->ptr, offset, void);
+			offset += SPA_ROUND_UP_N(m->size, 8);
+		}
+
+		for (j = 0; j < b->n_datas; j++) {
+			struct spa_data *d = &b->datas[j];
+
+			memcpy(d, &buffers[i].buffer->datas[j], sizeof(struct spa_data));
+			d->chunk =
+			    SPA_PTROFF(mm->ptr, offset + sizeof(struct spa_chunk) * j,
+				       struct spa_chunk);
+
+			if (flags & SPA_NODE_BUFFERS_FLAG_ALLOC)
+				continue;
+
+			if (d->type == SPA_DATA_MemId) {
+				uint32_t mem_id = SPA_PTR_TO_UINT32(d->data);
+				struct pw_memblock *bm;
+
+				bm = pw_mempool_find_id(data->pool, mem_id);
+				if (bm == NULL) {
+					pw_log_error("unknown buffer mem %u", mem_id);
+					res = -ENODEV;
+					goto error_exit_cleanup;
+				}
+
+				d->fd = bm->fd;
+				d->type = bm->type;
+				d->data = NULL;
+
+				pw_log_debug(" data %d %u -> fd %d maxsize %d",
+						j, bm->id, bm->fd, d->maxsize);
+			} else if (d->type == SPA_DATA_MemPtr) {
+				int offs = SPA_PTR_TO_INT(d->data);
+				d->data = SPA_PTROFF(mm->ptr, offs, void);
+				d->fd = -1;
+				pw_log_debug(" data %d id:%u -> mem:%p offs:%d maxsize:%d",
+						j, bid->id, d->data, offs, d->maxsize);
+			} else {
+				pw_log_warn("unknown buffer data type %d", d->type);
+			}
+		}
+		bufs[i] = b;
+	}
+
+	if ((res = pw_impl_port_use_buffers(mix->port, &mix->mix, flags, bufs, n_buffers)) < 0)
+		goto error_exit_cleanup;
+
+	if (flags & SPA_NODE_BUFFERS_FLAG_ALLOC) {
+		pw_client_node_port_buffers(data->client_node,
+				direction, port_id, mix_id,
+				n_buffers,
+				bufs);
+	}
+	return res;
+
+error_exit_cleanup:
+	clear_buffers(data, mix);
+error_exit:
+        pw_log_error("port %p: use_buffers: %d %s", mix, res, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "port_use_buffers error: %s", spa_strerror(res));
+	return res;
+}
+
+static int
+client_node_port_set_io(void *_data,
+                        uint32_t direction,
+                        uint32_t port_id,
+                        uint32_t mix_id,
+                        uint32_t id,
+                        uint32_t memid,
+                        uint32_t offset,
+                        uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct mix *mix;
+	struct pw_memmap *mm, *old;
+	void *ptr;
+	int res = 0;
+	uint32_t tag[5] = { data->remote_id, direction, port_id, mix_id, id };
+
+	mix = ensure_mix(data, direction, port_id, mix_id);
+	if (mix == NULL) {
+		res = -ENOENT;
+		goto exit;
+	}
+
+	old = pw_mempool_find_tag(data->pool, tag, sizeof(tag));
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	}
+	else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
+		if (mm == NULL) {
+			pw_log_warn("can't map memory id %u: %m", memid);
+			res = -errno;
+			goto exit;
+		}
+		ptr = mm->ptr;
+	}
+
+	pw_log_debug("port %p: set io:%s new:%p old:%p", mix->port,
+			spa_debug_type_find_name(spa_type_io, id), ptr, mix->mix.io);
+
+	if (id == SPA_IO_Buffers) {
+		if (ptr == NULL && mix->mix.io)
+			deactivate_mix(data, mix);
+	}
+
+	if ((res = spa_node_port_set_io(mix->port->mix,
+			     direction, mix->mix.port.port_id, id, ptr, size)) < 0) {
+		if (res == -ENOTSUP)
+			res = 0;
+		else
+			goto exit_free;
+	}
+	if (id == SPA_IO_Buffers) {
+		mix->mix.io = ptr;
+		if (ptr)
+			activate_mix(data, mix);
+	}
+exit_free:
+	pw_memmap_free(old);
+exit:
+	if (res < 0) {
+		pw_log_error("port %p: set_io: %s", mix, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "port_set_io failed: %s", spa_strerror(res));
+	}
+	return res;
+}
+
+static int link_signal_func(void *user_data)
+{
+	struct link *link = user_data;
+	struct spa_system *data_system = link->data->context->data_system;
+
+	pw_log_trace_fp("link %p: signal %p", link, link->target.activation);
+	if (SPA_UNLIKELY(spa_system_eventfd_write(data_system, link->signalfd, 1) < 0))
+		pw_log_warn("link %p: write failed %m", link);
+
+	return 0;
+}
+
+static int
+do_activate_link(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct link *link = user_data;
+	struct node_data *d = link->data;
+	pw_log_trace("link %p activate", link);
+	spa_list_append(&d->node->rt.target_list, &link->target.link);
+	return 0;
+}
+
+static int
+client_node_set_activation(void *_data,
+                        uint32_t node_id,
+                        int signalfd,
+                        uint32_t memid,
+                        uint32_t offset,
+                        uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_impl_node *node = data->node;
+	struct pw_memmap *mm;
+	void *ptr;
+	struct link *link;
+	int res = 0;
+
+	if (data->remote_id == node_id) {
+		pw_log_debug("node %p: our activation %u: %u %u %u", node, node_id,
+				memid, offset, size);
+		spa_system_close(data->context->data_system, signalfd);
+		return 0;
+	}
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	} else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
+		if (mm == NULL) {
+			res = -errno;
+			goto error_exit;
+		}
+		ptr = mm->ptr;
+	}
+	pw_log_debug("node %p: set activation %d %p %u %u", node, node_id, ptr, offset, size);
+
+	if (ptr) {
+		link = calloc(1, sizeof(struct link));
+		if (link == NULL) {
+			res = -errno;
+			goto error_exit;
+		}
+		link->data = data;
+		link->node_id = node_id;
+		link->map = mm;
+		link->target.activation = ptr;
+		link->signalfd = signalfd;
+		link->target.signal_func = link_signal_func;
+		link->target.data = link;
+		link->target.node = NULL;
+		spa_list_append(&data->links, &link->link);
+
+		pw_loop_invoke(data->context->data_loop,
+                       do_activate_link, SPA_ID_INVALID, NULL, 0, false, link);
+
+		pw_log_debug("node %p: link %p: fd:%d id:%u state %p required %d, pending %d",
+				node, link, signalfd,
+				link->target.activation->position.clock.id,
+				&link->target.activation->state[0],
+				link->target.activation->state[0].required,
+				link->target.activation->state[0].pending);
+	} else {
+		link = find_activation(&data->links, node_id);
+		if (link == NULL) {
+			res = -ENOENT;
+			goto error_exit;
+		}
+		clear_link(data, link);
+	}
+	return res;
+
+error_exit:
+	pw_log_error("node %p: set activation %d: %s", node, node_id, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "set_activation: %s", spa_strerror(res));
+	return res;
+}
+
+static const struct pw_client_node_events client_node_events = {
+	PW_VERSION_CLIENT_NODE_EVENTS,
+	.transport = client_node_transport,
+	.set_param = client_node_set_param,
+	.set_io = client_node_set_io,
+	.event = client_node_event,
+	.command = client_node_command,
+	.add_port = client_node_add_port,
+	.remove_port = client_node_remove_port,
+	.port_set_param = client_node_port_set_param,
+	.port_use_buffers = client_node_port_use_buffers,
+	.port_set_io = client_node_port_set_io,
+	.set_activation = client_node_set_activation,
+};
+
+static void do_node_init(struct node_data *data)
+{
+	struct pw_impl_port *port;
+
+	pw_log_debug("%p: node %p init", data, data->node);
+	add_node_update(data, PW_CLIENT_NODE_UPDATE_PARAMS |
+				PW_CLIENT_NODE_UPDATE_INFO,
+				SPA_NODE_CHANGE_MASK_FLAGS |
+				SPA_NODE_CHANGE_MASK_PROPS |
+				SPA_NODE_CHANGE_MASK_PARAMS);
+
+	spa_list_for_each(port, &data->node->input_ports, link) {
+		add_port_update(data, port,
+				PW_CLIENT_NODE_PORT_UPDATE_PARAMS |
+				PW_CLIENT_NODE_PORT_UPDATE_INFO);
+	}
+	spa_list_for_each(port, &data->node->output_ports, link) {
+		add_port_update(data, port,
+				PW_CLIENT_NODE_PORT_UPDATE_PARAMS |
+				PW_CLIENT_NODE_PORT_UPDATE_INFO);
+	}
+}
+
+static void clear_mix(struct node_data *data, struct mix *mix)
+{
+	pw_log_debug("port %p: mix clear %d.%d", mix->port, mix->port->port_id, mix->mix_id);
+
+	deactivate_mix(data, mix);
+
+	spa_list_remove(&mix->link);
+
+	clear_buffers(data, mix);
+	pw_array_clear(&mix->buffers);
+
+	spa_list_append(&data->free_mix, &mix->link);
+	pw_impl_port_release_mix(mix->port, &mix->mix);
+}
+
+static void clean_node(struct node_data *d)
+{
+	struct mix *mix;
+
+	if (d->have_transport) {
+		spa_list_consume(mix, &d->mix[SPA_DIRECTION_INPUT], link)
+			clear_mix(d, mix);
+		spa_list_consume(mix, &d->mix[SPA_DIRECTION_OUTPUT], link)
+			clear_mix(d, mix);
+	}
+	spa_list_consume(mix, &d->free_mix, link) {
+		spa_list_remove(&mix->link);
+		free(mix);
+	}
+	clean_transport(d);
+}
+
+static void node_destroy(void *data)
+{
+	struct node_data *d = data;
+
+	pw_log_debug("%p: destroy", d);
+
+	clean_node(d);
+}
+
+static void node_free(void *data)
+{
+	struct node_data *d = data;
+	pw_log_debug("%p: free", d);
+	d->node = NULL;
+}
+
+static void node_info_changed(void *data, const struct pw_node_info *info)
+{
+	struct node_data *d = data;
+	uint32_t change_mask, info_mask;
+
+	pw_log_debug("info changed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	change_mask = PW_CLIENT_NODE_UPDATE_INFO;
+	info_mask = SPA_NODE_CHANGE_MASK_FLAGS;
+	if (info->change_mask & PW_NODE_CHANGE_MASK_PROPS) {
+		info_mask |= SPA_NODE_CHANGE_MASK_PROPS;
+	}
+	if (info->change_mask & PW_NODE_CHANGE_MASK_PARAMS) {
+		change_mask |= PW_CLIENT_NODE_UPDATE_PARAMS;
+		info_mask |= SPA_NODE_CHANGE_MASK_PARAMS;
+	}
+	add_node_update(d, change_mask, info_mask);
+}
+
+static void node_port_info_changed(void *data, struct pw_impl_port *port,
+		const struct pw_port_info *info)
+{
+	struct node_data *d = data;
+	uint32_t change_mask = 0;
+
+	pw_log_debug("info changed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	if (info->change_mask & PW_PORT_CHANGE_MASK_PROPS)
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_INFO;
+	if (info->change_mask & PW_PORT_CHANGE_MASK_PARAMS) {
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_PARAMS;
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_INFO;
+	}
+	add_port_update(d, port, change_mask);
+}
+
+static void node_port_removed(void *data, struct pw_impl_port *port)
+{
+	struct node_data *d = data;
+	struct mix *mix, *tmp;
+
+	pw_log_debug("removed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	pw_client_node_port_update(d->client_node,
+			port->direction,
+			port->port_id,
+			0, 0, NULL, NULL);
+
+	spa_list_for_each_safe(mix, tmp, &d->mix[port->direction], link) {
+		if (mix->port == port)
+			clear_mix(d, mix);
+	}
+}
+
+static void node_active_changed(void *data, bool active)
+{
+	struct node_data *d = data;
+	pw_log_debug("active %d", active);
+
+	if (d->client_node == NULL)
+		return;
+
+	pw_client_node_set_active(d->client_node, active);
+}
+
+static void node_event(void *data, const struct spa_event *event)
+{
+	struct node_data *d = data;
+	pw_log_debug("%p", d);
+
+	if (d->client_node == NULL)
+		return;
+	pw_client_node_event(d->client_node, event);
+}
+
+static const struct pw_impl_node_events node_events = {
+	PW_VERSION_IMPL_NODE_EVENTS,
+	.destroy = node_destroy,
+	.free = node_free,
+	.info_changed = node_info_changed,
+	.port_info_changed = node_port_info_changed,
+	.port_removed = node_port_removed,
+	.active_changed = node_active_changed,
+	.event = node_event,
+};
+
+static void client_node_removed(void *_data)
+{
+	struct node_data *data = _data;
+	pw_log_debug("%p: removed", data);
+
+	spa_hook_remove(&data->proxy_client_node_listener);
+	spa_hook_remove(&data->client_node_listener);
+
+	if (data->node) {
+		spa_hook_remove(&data->node_listener);
+		pw_impl_node_set_state(data->node, PW_NODE_STATE_SUSPENDED);
+
+		clean_node(data);
+
+		if (data->do_free)
+			pw_impl_node_destroy(data->node);
+	}
+	data->client_node = NULL;
+}
+
+static void client_node_destroy(void *_data)
+{
+	struct node_data *data = _data;
+
+	pw_log_debug("%p: destroy", data);
+	client_node_removed(_data);
+}
+
+static void client_node_bound(void *_data, uint32_t global_id)
+{
+	struct node_data *data = _data;
+	pw_log_debug("%p: bound %u", data, global_id);
+	data->remote_id = global_id;
+}
+
+static const struct pw_proxy_events proxy_client_node_events = {
+	PW_VERSION_PROXY_EVENTS,
+	.removed = client_node_removed,
+	.destroy = client_node_destroy,
+	.bound = client_node_bound,
+};
+
+static int node_ready(void *d, int status)
+{
+	struct node_data *data = d;
+	struct pw_impl_node *node = data->node;
+	struct pw_node_activation *a = node->rt.activation;
+	struct spa_system *data_system = data->context->data_system;
+	struct timespec ts;
+	struct pw_impl_port *p;
+
+	pw_log_trace_fp("node %p: ready driver:%d exported:%d status:%d", node,
+			node->driver, node->exported, status);
+
+	if (status & SPA_STATUS_HAVE_DATA) {
+		spa_list_for_each(p, &node->rt.output_mix, rt.node_link)
+			spa_node_process(p->mix);
+	}
+
+	spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
+	a->status = PW_NODE_ACTIVATION_TRIGGERED;
+	a->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
+
+	if (SPA_UNLIKELY(spa_system_eventfd_write(data_system, data->rtwritefd, 1) < 0))
+		pw_log_warn("node %p: write failed %m", node);
+
+	return 0;
+}
+
+static int node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
+{
+	return 0;
+}
+
+static int node_xrun(void *d, uint64_t trigger, uint64_t delay, struct spa_pod *info)
+{
+	struct node_data *data = d;
+	struct pw_impl_node *node = data->node;
+	struct pw_node_activation *a = node->rt.activation;
+
+	a->xrun_count++;
+	a->xrun_time = trigger;
+	a->xrun_delay = delay;
+	a->max_delay = SPA_MAX(a->max_delay, delay);
+
+	pw_log_debug("node %p: XRun! count:%u time:%"PRIu64" delay:%"PRIu64" max:%"PRIu64,
+			node, a->xrun_count, trigger, delay, a->max_delay);
+
+	pw_context_driver_emit_xrun(data->context, node);
+
+	return 0;
+}
+
+static const struct spa_node_callbacks node_callbacks = {
+	SPA_VERSION_NODE_CALLBACKS,
+	.ready = node_ready,
+	.reuse_buffer = node_reuse_buffer,
+	.xrun = node_xrun
+};
+
+static struct pw_proxy *node_export(struct pw_core *core, void *object, bool do_free,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node = object;
+	struct pw_proxy *client_node;
+	struct node_data *data;
+
+	user_data_size = SPA_ROUND_UP_N(user_data_size, __alignof__(struct node_data));
+
+	client_node = pw_core_create_object(core,
+			"client-node",
+			PW_TYPE_INTERFACE_ClientNode,
+			PW_VERSION_CLIENT_NODE,
+			&node->properties->dict,
+			user_data_size + sizeof(struct node_data));
+	if (client_node == NULL)
+		goto error;
+
+	data = pw_proxy_get_user_data(client_node);
+	data = SPA_PTROFF(data, user_data_size, struct node_data);
+	data->pool = pw_core_get_mempool(core);
+	data->node = node;
+	data->do_free = do_free;
+	data->context = pw_impl_node_get_context(node);
+	data->client_node = (struct pw_client_node *)client_node;
+	data->remote_id = SPA_ID_INVALID;
+
+
+	data->allow_mlock = pw_properties_get_bool(node->properties, "mem.allow-mlock",
+						   data->context->settings.mem_allow_mlock);
+
+	data->warn_mlock = pw_properties_get_bool(node->properties, "mem.warn-mlock",
+						  data->context->settings.mem_warn_mlock);
+
+	node->exported = true;
+
+	spa_list_init(&data->free_mix);
+	spa_list_init(&data->mix[0]);
+	spa_list_init(&data->mix[1]);
+
+	spa_list_init(&data->links);
+
+	pw_proxy_add_listener(client_node,
+			&data->proxy_client_node_listener,
+			&proxy_client_node_events, data);
+
+	spa_node_set_callbacks(node->node, &node_callbacks, data);
+	pw_impl_node_add_listener(node, &data->node_listener, &node_events, data);
+
+	pw_client_node_add_listener(data->client_node,
+					  &data->client_node_listener,
+					  &client_node_events,
+					  data);
+	do_node_init(data);
+
+	return client_node;
+error:
+	if (do_free)
+		pw_impl_node_destroy(node);
+	return NULL;
+
+}
+
+struct pw_proxy *pw_core_node_export(struct pw_core *core,
+		const char *type, const struct spa_dict *props, void *object,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node = object;
+
+	if (props)
+		pw_impl_node_update_properties(node, props);
+	return node_export(core, object, false, user_data_size);
+}
+
+struct pw_proxy *pw_core_spa_node_export(struct pw_core *core,
+		const char *type, const struct spa_dict *props, void *object,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node;
+	struct pw_proxy *proxy;
+	const char *str;
+	bool do_register;
+
+	str = props ? spa_dict_lookup(props, PW_KEY_OBJECT_REGISTER) : NULL;
+	do_register = str ? pw_properties_parse_bool(str) : true;
+
+	node = pw_context_create_node(pw_core_get_context(core),
+			props ? pw_properties_new_dict(props) : NULL, 0);
+	if (node == NULL)
+		return NULL;
+
+	pw_impl_node_set_implementation(node, (struct spa_node*)object);
+
+	if (do_register)
+		pw_impl_node_register(node, NULL);
+
+	proxy = node_export(core, node, true, user_data_size);
+	if (proxy)
+		pw_impl_node_set_active(node, true);
+
+	return proxy;
+}
diff --git a/src/modules/module-client-node/v0/client-node.c b/src/modules/module-client-node/v0/client-node.c
new file mode 100644
index 0000000..e70a7c2
--- /dev/null
+++ b/src/modules/module-client-node/v0/client-node.c
@@ -0,0 +1,1447 @@
+/* PipeWire
+ *
+ * Copyright © 2015 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <string.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <errno.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <dlfcn.h>
+#include <sys/socket.h>
+#include <sys/mman.h>
+
+#include <spa/node/node.h>
+#include <spa/node/utils.h>
+#include <spa/node/io.h>
+#include <spa/pod/filter.h>
+#include <spa/utils/keys.h>
+
+#define PW_ENABLE_DEPRECATED
+
+#include "pipewire/pipewire.h"
+#include "pipewire/private.h"
+
+#include "pipewire/context.h"
+#include "modules/spa/spa-node.h"
+#include "client-node.h"
+#include "transport.h"
+
+PW_LOG_TOPIC_EXTERN(mod_topic);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+/** \cond */
+
+#define MAX_INPUTS       64
+#define MAX_OUTPUTS      64
+
+#define MAX_BUFFERS      64
+
+#define CHECK_IN_PORT_ID(this,d,p)       ((d) == SPA_DIRECTION_INPUT && (p) < MAX_INPUTS)
+#define CHECK_OUT_PORT_ID(this,d,p)      ((d) == SPA_DIRECTION_OUTPUT && (p) < MAX_OUTPUTS)
+#define CHECK_PORT_ID(this,d,p)          (CHECK_IN_PORT_ID(this,d,p) || CHECK_OUT_PORT_ID(this,d,p))
+#define CHECK_FREE_IN_PORT(this,d,p)     (CHECK_IN_PORT_ID(this,d,p) && !(this)->in_ports[p].valid)
+#define CHECK_FREE_OUT_PORT(this,d,p)    (CHECK_OUT_PORT_ID(this,d,p) && !(this)->out_ports[p].valid)
+#define CHECK_FREE_PORT(this,d,p)        (CHECK_FREE_IN_PORT (this,d,p) || CHECK_FREE_OUT_PORT (this,d,p))
+#define CHECK_IN_PORT(this,d,p)          (CHECK_IN_PORT_ID(this,d,p) && (this)->in_ports[p].valid)
+#define CHECK_OUT_PORT(this,d,p)         (CHECK_OUT_PORT_ID(this,d,p) && (this)->out_ports[p].valid)
+#define CHECK_PORT(this,d,p)             (CHECK_IN_PORT (this,d,p) || CHECK_OUT_PORT (this,d,p))
+
+#define GET_IN_PORT(this,p)	(&this->in_ports[p])
+#define GET_OUT_PORT(this,p)	(&this->out_ports[p])
+#define GET_PORT(this,d,p)	(d == SPA_DIRECTION_INPUT ? GET_IN_PORT(this,p) : GET_OUT_PORT(this,p))
+
+#define CHECK_PORT_BUFFER(this,b,p)      (b < p->n_buffers)
+
+extern uint32_t pw_protocol_native0_type_from_v2(struct pw_impl_client *client, uint32_t type);
+extern uint32_t pw_protocol_native0_name_to_v2(struct pw_impl_client *client, const char *name);
+
+struct mem {
+	uint32_t id;
+	int ref;
+	int fd;
+	uint32_t type;
+	uint32_t flags;
+};
+
+struct buffer {
+	struct spa_buffer *outbuf;
+	struct spa_buffer buffer;
+	struct spa_meta metas[4];
+	struct spa_data datas[4];
+	bool outstanding;
+	uint32_t memid;
+};
+
+struct port {
+	uint32_t id;
+	enum spa_direction direction;
+
+	bool valid;
+	struct spa_port_info info;
+	struct pw_properties *properties;
+
+	bool have_format;
+	uint32_t n_params;
+	struct spa_pod **params;
+	struct spa_io_buffers *io;
+
+	uint32_t n_buffers;
+	struct buffer buffers[MAX_BUFFERS];
+};
+
+struct node {
+	struct spa_node node;
+
+	struct impl *impl;
+
+	struct spa_log *log;
+	struct spa_loop *data_loop;
+	struct spa_system *data_system;
+
+	struct spa_hook_list hooks;
+	struct spa_callbacks callbacks;
+
+	struct spa_io_position *position;
+
+	struct pw_resource *resource;
+
+	struct spa_source data_source;
+	int writefd;
+
+	struct spa_node_info info;
+
+	uint32_t n_inputs;
+	uint32_t n_outputs;
+	struct port in_ports[MAX_INPUTS];
+	struct port out_ports[MAX_OUTPUTS];
+
+	uint32_t n_params;
+	struct spa_pod **params;
+
+	uint32_t seq;
+	uint32_t init_pending;
+};
+
+struct impl {
+	struct pw_impl_client_node0 this;
+
+	bool client_reuse;
+
+	struct pw_context *context;
+
+	struct node node;
+
+	struct pw_client_node0_transport *transport;
+
+	struct spa_hook node_listener;
+	struct spa_hook resource_listener;
+	struct spa_hook object_listener;
+
+	struct pw_array mems;
+
+	int fds[2];
+	int other_fds[2];
+
+	uint32_t input_ready;
+	bool out_pending;
+};
+
+/** \endcond */
+
+static struct mem *ensure_mem(struct impl *impl, int fd, uint32_t type, uint32_t flags)
+{
+	struct mem *m, *f = NULL;
+
+	pw_array_for_each(m, &impl->mems) {
+		if (m->ref <= 0)
+			f = m;
+		else if (m->fd == fd)
+			goto found;
+	}
+
+	if (f == NULL) {
+		m = pw_array_add(&impl->mems, sizeof(struct mem));
+		m->id = pw_array_get_len(&impl->mems, struct mem) - 1;
+		m->ref = 0;
+	}
+	else {
+		m = f;
+	}
+	m->fd = fd;
+	m->type = type;
+	m->flags = flags;
+
+	pw_client_node0_resource_add_mem(impl->node.resource,
+					m->id,
+					type,
+					m->fd,
+					m->flags);
+      found:
+	m->ref++;
+	return m;
+}
+
+
+static int clear_buffers(struct node *this, struct port *port)
+{
+	uint32_t i, j;
+	struct impl *impl = this->impl;
+
+	for (i = 0; i < port->n_buffers; i++) {
+		struct buffer *b = &port->buffers[i];
+		struct mem *m;
+
+		spa_log_debug(this->log, "node %p: clear buffer %d", this, i);
+
+		for (j = 0; j < b->buffer.n_datas; j++) {
+			struct spa_data *d = &b->datas[j];
+
+			if (d->type == SPA_DATA_DmaBuf ||
+			    d->type == SPA_DATA_MemFd) {
+				uint32_t id;
+
+				id = SPA_PTR_TO_UINT32(b->buffer.datas[j].data);
+				m = pw_array_get_unchecked(&impl->mems, id, struct mem);
+				m->ref--;
+			}
+		}
+		m = pw_array_get_unchecked(&impl->mems, b->memid, struct mem);
+		m->ref--;
+	}
+	port->n_buffers = 0;
+	return 0;
+}
+
+static void emit_port_info(struct node *this, struct port *port)
+{
+	spa_node_emit_port_info(&this->hooks,
+				port->direction, port->id, &port->info);
+}
+
+static int impl_node_add_listener(void *object,
+		struct spa_hook *listener,
+		const struct spa_node_events *events,
+		void *data)
+{
+	struct node *this = object;
+	struct spa_hook_list save;
+	uint32_t i;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_hook_list_isolate(&this->hooks, &save, listener, events, data);
+
+	for (i = 0; i < MAX_INPUTS; i++) {
+		if (this->in_ports[i].valid)
+			emit_port_info(this, &this->in_ports[i]);
+	}
+	for (i = 0; i < MAX_OUTPUTS; i++) {
+		if (this->out_ports[i].valid)
+			emit_port_info(this, &this->out_ports[i]);
+	}
+	spa_hook_list_join(&this->hooks, &save);
+
+	return 0;
+}
+
+static int impl_node_enum_params(void *object, int seq,
+				 uint32_t id, uint32_t start, uint32_t num,
+				 const struct spa_pod *filter)
+{
+	struct node *this = object;
+	uint8_t buffer[1024];
+	struct spa_pod_builder b = { 0 };
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+	bool found = false;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	result.id = id;
+	result.next = 0;
+
+	while (true) {
+		struct spa_pod *param;
+
+		result.index = result.next++;
+		if (result.index >= this->n_params)
+			break;
+
+		param = this->params[result.index];
+
+		if (param == NULL || !spa_pod_is_object_id(param, id))
+			continue;
+
+		found = true;
+
+		if (result.index < start)
+			continue;
+
+		spa_pod_builder_init(&b, buffer, sizeof(buffer));
+		if (spa_pod_filter(&b, &result.param, param, filter) != 0)
+			continue;
+
+		pw_log_debug("%p: %d param %u", this, seq, result.index);
+		spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+		if (++count == num)
+			break;
+	}
+	return found ? 0 : -ENOENT;
+}
+
+static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
+			       const struct spa_pod *param)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	pw_client_node0_resource_set_param(this->resource, this->seq, id, flags, param);
+
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
+{
+	struct node *this = object;
+	int res = 0;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	switch(id) {
+	case SPA_IO_Position:
+		this->position = data;
+		break;
+	default:
+		res = -ENOTSUP;
+		break;
+	}
+	return res;
+}
+
+static inline void do_flush(struct node *this)
+{
+	if (spa_system_eventfd_write(this->data_system, this->writefd, 1) < 0)
+		spa_log_warn(this->log, "node %p: error flushing : %s", this, strerror(errno));
+
+}
+
+static int send_clock_update(struct node *this)
+{
+	struct pw_impl_client *client = this->resource->client;
+	uint32_t type = pw_protocol_native0_name_to_v2(client, SPA_TYPE_INFO_NODE_COMMAND_BASE "ClockUpdate");
+	struct timespec ts;
+	int64_t now;
+
+	clock_gettime(CLOCK_MONOTONIC, &ts);
+	now = SPA_TIMESPEC_TO_NSEC(&ts);
+	pw_log_trace("%p: now %"PRIi64, this, now);
+
+	struct spa_command_node0_clock_update cu =
+		SPA_COMMAND_NODE0_CLOCK_UPDATE_INIT(type,
+						SPA_COMMAND_NODE0_CLOCK_UPDATE_TIME |
+						SPA_COMMAND_NODE0_CLOCK_UPDATE_SCALE |
+						SPA_COMMAND_NODE0_CLOCK_UPDATE_STATE |
+						SPA_COMMAND_NODE0_CLOCK_UPDATE_LATENCY,   /* change_mask */
+						SPA_USEC_PER_SEC,       /* rate */
+						now / SPA_NSEC_PER_USEC,       /* ticks */
+						now,       /* monotonic_time */
+						0,       /* offset */
+						(1 << 16) | 1,   /* scale */
+						SPA_CLOCK0_STATE_RUNNING, /* state */
+						SPA_COMMAND_NODE0_CLOCK_UPDATE_FLAG_LIVE,       /* flags */
+						0);      /* latency */
+
+	pw_client_node0_resource_command(this->resource, this->seq, (const struct spa_command*)&cu);
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int impl_node_send_command(void *object, const struct spa_command *command)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(command != NULL, -EINVAL);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	if (SPA_NODE_COMMAND_ID(command) == SPA_NODE_COMMAND_Start) {
+		send_clock_update(this);
+	}
+
+	pw_client_node0_resource_command(this->resource, this->seq, command);
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int
+impl_node_set_callbacks(void *object,
+			const struct spa_node_callbacks *callbacks,
+			void *data)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	this->callbacks = SPA_CALLBACKS_INIT(callbacks, data);
+
+	return 0;
+}
+
+static int
+impl_node_sync(void *object, int seq)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	pw_log_debug("%p: sync %p", this, this->resource);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	this->init_pending = SPA_RESULT_RETURN_ASYNC(this->seq++);
+
+	return this->init_pending;
+}
+
+
+extern struct spa_pod *pw_protocol_native0_pod_from_v2(struct pw_impl_client *client, const struct spa_pod *pod);
+extern int pw_protocol_native0_pod_to_v2(struct pw_impl_client *client, const struct spa_pod *pod,
+		struct spa_pod_builder *b);
+
+static void
+do_update_port(struct node *this,
+	       enum spa_direction direction,
+	       uint32_t port_id,
+	       uint32_t change_mask,
+	       uint32_t n_params,
+	       const struct spa_pod **params,
+	       const struct spa_port_info *info)
+{
+	struct port *port;
+
+	port = GET_PORT(this, direction, port_id);
+
+	if (!port->valid) {
+		spa_log_debug(this->log, "node %p: adding port %d, direction %d",
+				this, port_id, direction);
+		port->id = port_id;
+		port->direction = direction;
+		port->have_format = false;
+		port->valid = true;
+
+		if (direction == SPA_DIRECTION_INPUT)
+			this->n_inputs++;
+		else
+			this->n_outputs++;
+	}
+
+	if (change_mask & PW_CLIENT_NODE0_PORT_UPDATE_PARAMS) {
+		uint32_t i;
+
+		port->have_format = false;
+
+		spa_log_debug(this->log, "node %p: port %u update %d params", this, port_id, n_params);
+		for (i = 0; i < port->n_params; i++)
+			free(port->params[i]);
+		port->n_params = n_params;
+		if (port->n_params == 0) {
+			free(port->params);
+			port->params = NULL;
+		} else {
+			void *p;
+			p = pw_reallocarray(port->params, port->n_params, sizeof(struct spa_pod *));
+			if (p == NULL) {
+				pw_log_error("%p: port %u can't realloc: %m", this, port_id);
+				free(port->params);
+				port->n_params = 0;
+			}
+			port->params = p;
+		}
+		for (i = 0; i < port->n_params; i++) {
+			port->params[i] = params[i] ?
+				pw_protocol_native0_pod_from_v2(this->resource->client, params[i]) : NULL;
+
+			if (port->params[i] && spa_pod_is_object_id(port->params[i], SPA_PARAM_Format))
+				port->have_format = true;
+		}
+	}
+
+	if (change_mask & PW_CLIENT_NODE0_PORT_UPDATE_INFO) {
+		pw_properties_free(port->properties);
+		port->properties = NULL;
+		port->info.props = NULL;
+		port->info.n_params = 0;
+		port->info.params = NULL;
+
+		if (info) {
+			port->info = *info;
+			if (info->props) {
+				port->properties = pw_properties_new_dict(info->props);
+				port->info.props = &port->properties->dict;
+			}
+		}
+		spa_node_emit_port_info(&this->hooks, direction, port_id, info);
+	}
+}
+
+static void
+clear_port(struct node *this,
+	   struct port *port, enum spa_direction direction, uint32_t port_id)
+{
+	do_update_port(this,
+		       direction,
+		       port_id,
+		       PW_CLIENT_NODE0_PORT_UPDATE_PARAMS |
+		       PW_CLIENT_NODE0_PORT_UPDATE_INFO, 0, NULL, NULL);
+	clear_buffers(this, port);
+}
+
+static void do_uninit_port(struct node *this, enum spa_direction direction, uint32_t port_id)
+{
+	struct port *port;
+
+	spa_log_debug(this->log, "node %p: removing port %d", this, port_id);
+
+	if (direction == SPA_DIRECTION_INPUT) {
+		port = GET_IN_PORT(this, port_id);
+		this->n_inputs--;
+	} else {
+		port = GET_OUT_PORT(this, port_id);
+		this->n_outputs--;
+	}
+	clear_port(this, port, direction, port_id);
+	port->valid = false;
+	spa_node_emit_port_info(&this->hooks, direction, port_id, NULL);
+}
+
+static int
+impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	struct node *this = object;
+	struct port *port;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_FREE_PORT(this, direction, port_id), -EINVAL);
+
+	port = GET_PORT(this, direction, port_id);
+	clear_port(this, port, direction, port_id);
+
+	return 0;
+}
+
+static int
+impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	do_uninit_port(this, direction, port_id);
+
+	return 0;
+}
+
+static int
+impl_node_port_enum_params(void *object, int seq,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t id, uint32_t start, uint32_t num,
+			   const struct spa_pod *filter)
+{
+	struct node *this = object;
+	struct port *port;
+	uint8_t buffer[1024];
+	struct spa_pod_builder b = { 0 };
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+	bool found = false;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	port = GET_PORT(this, direction, port_id);
+
+	pw_log_debug("%p: %d port %d.%d %u %u %u", this, seq,
+			direction, port_id, id, start, num);
+
+	result.id = id;
+	result.next = 0;
+
+	while (true) {
+		struct spa_pod *param;
+
+		result.index = result.next++;
+		if (result.index >= port->n_params)
+			break;
+
+		param = port->params[result.index];
+
+		if (param == NULL || !spa_pod_is_object_id(param, id))
+			continue;
+
+		found = true;
+
+		if (result.index < start)
+			continue;
+
+		spa_pod_builder_init(&b, buffer, sizeof(buffer));
+		if (spa_pod_filter(&b, &result.param, param, filter) < 0)
+			continue;
+
+		pw_log_debug("%p: %d param %u", this, seq, result.index);
+		spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+		if (++count == num)
+			break;
+	}
+	return found ? 0 : -ENOENT;
+}
+
+static int
+impl_node_port_set_param(void *object,
+			 enum spa_direction direction, uint32_t port_id,
+			 uint32_t id, uint32_t flags,
+			 const struct spa_pod *param)
+{
+	struct node *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	pw_client_node0_resource_port_set_param(this->resource,
+					       this->seq,
+					       direction, port_id,
+					       id, flags,
+					       param);
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int
+impl_node_port_set_io(void *object,
+		      enum spa_direction direction,
+		      uint32_t port_id,
+		      uint32_t id,
+		      void *data, size_t size)
+{
+	struct node *this = object;
+	struct impl *impl;
+	struct pw_memblock *mem;
+	struct mem *m;
+	uint32_t memid, mem_offset, mem_size;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	impl = this->impl;
+
+	spa_log_debug(this->log, "node %p: port %d.%d set io %d %p", this,
+			direction, port_id, id, data);
+
+	if (id == SPA_IO_Buffers) {
+		struct port *port = GET_PORT(this, direction, port_id);
+		port->io = data;
+	}
+
+	if (this->resource == NULL)
+		return -EIO;
+
+
+	if (data) {
+		if ((mem = pw_mempool_find_ptr(impl->context->pool, data)) == NULL)
+			return -EINVAL;
+
+		mem_offset = SPA_PTRDIFF(data, mem->map->ptr);
+		mem_size = mem->size;
+		if (mem_size - mem_offset < size)
+			return -EINVAL;
+
+		mem_offset += mem->map->offset;
+		m = ensure_mem(impl, mem->fd, SPA_DATA_MemFd, mem->flags);
+		memid = m->id;
+	}
+	else {
+		memid = SPA_ID_INVALID;
+		mem_offset = mem_size = 0;
+	}
+
+	pw_client_node0_resource_port_set_io(this->resource,
+					    this->seq,
+					    direction, port_id,
+					    id,
+					    memid,
+					    mem_offset, mem_size);
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int
+impl_node_port_use_buffers(void *object,
+			   enum spa_direction direction,
+			   uint32_t port_id,
+			   uint32_t flags,
+			   struct spa_buffer **buffers,
+			   uint32_t n_buffers)
+{
+	struct node *this = object;
+	struct impl *impl;
+	struct port *port;
+	uint32_t i, j;
+	struct pw_client_node0_buffer *mb;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+
+	impl = this->impl;
+	spa_log_debug(this->log, "node %p: use buffers %p %u", this, buffers, n_buffers);
+
+	port = GET_PORT(this, direction, port_id);
+
+	if (!port->have_format)
+		return -EIO;
+
+	clear_buffers(this, port);
+
+	if (n_buffers > 0) {
+		mb = alloca(n_buffers * sizeof(struct pw_client_node0_buffer));
+	} else {
+		mb = NULL;
+	}
+
+	port->n_buffers = n_buffers;
+
+	if (this->resource == NULL)
+		return -EIO;
+
+	for (i = 0; i < n_buffers; i++) {
+		struct buffer *b = &port->buffers[i];
+		struct pw_memblock *mem;
+		struct mem *m;
+		size_t data_size;
+		void *baseptr;
+
+		b->outbuf = buffers[i];
+		memcpy(&b->buffer, buffers[i], sizeof(struct spa_buffer));
+		b->buffer.datas = b->datas;
+		b->buffer.metas = b->metas;
+
+		if (buffers[i]->n_metas > 0)
+			baseptr = buffers[i]->metas[0].data;
+		else if (buffers[i]->n_datas > 0)
+			baseptr = buffers[i]->datas[0].chunk;
+		else
+			return -EINVAL;
+
+		if ((mem = pw_mempool_find_ptr(impl->context->pool, baseptr)) == NULL)
+			return -EINVAL;
+
+		data_size = 0;
+		for (j = 0; j < buffers[i]->n_metas; j++) {
+			data_size += buffers[i]->metas[j].size;
+		}
+		for (j = 0; j < buffers[i]->n_datas; j++) {
+			struct spa_data *d = buffers[i]->datas;
+			data_size += sizeof(struct spa_chunk);
+			if (d->type == SPA_DATA_MemPtr)
+				data_size += d->maxsize;
+		}
+
+		m = ensure_mem(impl, mem->fd, SPA_DATA_MemFd, mem->flags);
+		b->memid = m->id;
+
+		mb[i].buffer = &b->buffer;
+		mb[i].mem_id = b->memid;
+		mb[i].offset = SPA_PTRDIFF(baseptr, SPA_PTROFF(mem->map->ptr, mem->map->offset, void));
+		mb[i].size = data_size;
+
+		for (j = 0; j < buffers[i]->n_metas; j++)
+			memcpy(&b->buffer.metas[j], &buffers[i]->metas[j], sizeof(struct spa_meta));
+		b->buffer.n_metas = j;
+
+		for (j = 0; j < buffers[i]->n_datas; j++) {
+			struct spa_data *d = &buffers[i]->datas[j];
+
+			memcpy(&b->buffer.datas[j], d, sizeof(struct spa_data));
+
+			if (d->type == SPA_DATA_DmaBuf ||
+			    d->type == SPA_DATA_MemFd) {
+				m = ensure_mem(impl, d->fd, d->type, d->flags);
+				b->buffer.datas[j].data = SPA_UINT32_TO_PTR(m->id);
+			} else if (d->type == SPA_DATA_MemPtr) {
+				b->buffer.datas[j].data = SPA_INT_TO_PTR(SPA_PTRDIFF(d->data, baseptr));
+			} else {
+				b->buffer.datas[j].type = SPA_ID_INVALID;
+				b->buffer.datas[j].data = 0;
+				spa_log_error(this->log, "invalid memory type %d", d->type);
+			}
+		}
+	}
+
+	pw_client_node0_resource_port_use_buffers(this->resource,
+						 this->seq,
+						 direction, port_id,
+						 n_buffers, mb);
+
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int
+impl_node_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
+{
+	struct node *this = object;
+	struct impl *impl;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_OUT_PORT(this, SPA_DIRECTION_OUTPUT, port_id), -EINVAL);
+
+	impl = this->impl;
+
+	spa_log_trace(this->log, "reuse buffer %d", buffer_id);
+
+	pw_client_node0_transport_add_message(impl->transport, (struct pw_client_node0_message *)
+			&PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER_INIT(port_id, buffer_id));
+	do_flush(this);
+
+	return 0;
+}
+
+static int impl_node_process_input(struct spa_node *node)
+{
+	struct node *this = SPA_CONTAINER_OF(node, struct node, node);
+	struct impl *impl = this->impl;
+//	bool client_reuse = impl->client_reuse;
+	uint32_t i;
+	int res;
+
+	if (impl->input_ready == 0) {
+		/* the client is not ready to receive our buffers, recycle them */
+		pw_log_trace("node not ready, recycle buffers");
+		for (i = 0; i < MAX_INPUTS; i++) {
+			struct port *p = &this->in_ports[i];
+			struct spa_io_buffers *io = p->io;
+
+			if (!p->valid || io == NULL)
+				continue;
+
+			io->status = SPA_STATUS_NEED_DATA;
+		}
+		res = SPA_STATUS_NEED_DATA;
+	}
+	else {
+		for (i = 0; i < MAX_INPUTS; i++) {
+			struct port *p = &this->in_ports[i];
+			struct spa_io_buffers *io = p->io;
+
+			if (!p->valid || io == NULL)
+				continue;
+
+			pw_log_trace("set io status to %d %d", io->status, io->buffer_id);
+			impl->transport->inputs[p->id] = *io;
+
+			/* explicitly recycle buffers when the client is not going to do it */
+//			if (!client_reuse && (pp = p->peer))
+//		                spa_node_port_reuse_buffer(pp->node->implementation,
+//						pp->port_id, io->buffer_id);
+		}
+		pw_client_node0_transport_add_message(impl->transport,
+			       &PW_CLIENT_NODE0_MESSAGE_INIT(PW_CLIENT_NODE0_MESSAGE_PROCESS_INPUT));
+		do_flush(this);
+
+		impl->input_ready--;
+		res = SPA_STATUS_OK;
+	}
+	return res;
+}
+
+#if 0
+/** this is used for clients providing data to pipewire and currently
+ * not supported in the compat layer */
+static int impl_node_process_output(struct spa_node *node)
+{
+	struct node *this;
+	struct impl *impl;
+	uint32_t i;
+
+	this = SPA_CONTAINER_OF(node, struct node, node);
+	impl = this->impl;
+
+	if (impl->out_pending)
+		goto done;
+
+	impl->out_pending = true;
+
+	for (i = 0; i < MAX_OUTPUTS; i++) {
+		struct port *p = &this->out_ports[i];
+		struct spa_io_buffers *io = p->io;
+
+		if (!p->valid || io == NULL)
+			continue;
+
+		impl->transport->outputs[p->id] = *io;
+
+		pw_log_trace("%d %d -> %d %d", io->status, io->buffer_id,
+				impl->transport->outputs[p->id].status,
+				impl->transport->outputs[p->id].buffer_id);
+	}
+
+      done:
+	pw_client_node0_transport_add_message(impl->transport,
+			       &PW_CLIENT_NODE0_MESSAGE_INIT(PW_CLIENT_NODE0_MESSAGE_PROCESS_OUTPUT));
+	do_flush(this);
+
+	return SPA_STATUS_OK;
+}
+#endif
+
+static int impl_node_process(void *object)
+{
+	struct node *this = object;
+	struct impl *impl = this->impl;
+	struct pw_impl_node *n = impl->this.node;
+
+	return impl_node_process_input(n->node);
+}
+
+static int handle_node_message(struct node *this, struct pw_client_node0_message *message)
+{
+	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, node);
+	uint32_t i;
+
+	switch (PW_CLIENT_NODE0_MESSAGE_TYPE(message)) {
+	case PW_CLIENT_NODE0_MESSAGE_HAVE_OUTPUT:
+		for (i = 0; i < MAX_OUTPUTS; i++) {
+			struct port *p = &this->out_ports[i];
+			struct spa_io_buffers *io = p->io;
+			if (!p->valid || io == NULL)
+				continue;
+			*io = impl->transport->outputs[p->id];
+			pw_log_trace("have output %d %d", io->status, io->buffer_id);
+		}
+		impl->out_pending = false;
+		spa_node_call_ready(&this->callbacks, SPA_STATUS_HAVE_DATA);
+		break;
+
+	case PW_CLIENT_NODE0_MESSAGE_NEED_INPUT:
+		for (i = 0; i < MAX_INPUTS; i++) {
+			struct port *p = &this->in_ports[i];
+			struct spa_io_buffers *io = p->io;
+			if (!p->valid || io == NULL)
+				continue;
+			pw_log_trace("need input %d %d", i, p->id);
+			*io = impl->transport->inputs[p->id];
+			pw_log_trace("need input %d %d", io->status, io->buffer_id);
+		}
+		impl->input_ready++;
+		spa_node_call_ready(&this->callbacks, SPA_STATUS_NEED_DATA);
+		break;
+
+	case PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER:
+		if (impl->client_reuse) {
+			struct pw_client_node0_message_port_reuse_buffer *p =
+			    (struct pw_client_node0_message_port_reuse_buffer *) message;
+			spa_node_call_reuse_buffer(&this->callbacks, p->body.port_id.value,
+					p->body.buffer_id.value);
+		}
+		break;
+
+	default:
+		pw_log_warn("unhandled message %d", PW_CLIENT_NODE0_MESSAGE_TYPE(message));
+		return -ENOTSUP;
+	}
+	return 0;
+}
+
+static void setup_transport(struct impl *impl)
+{
+	struct node *this = &impl->node;
+	uint32_t max_inputs = 0, max_outputs = 0, n_inputs = 0, n_outputs = 0;
+	struct spa_dict_item items[1];
+
+	n_inputs = this->n_inputs;
+	max_inputs = this->info.max_input_ports == 0 ? this->n_inputs : this->info.max_input_ports;
+	n_outputs = this->n_outputs;
+	max_outputs = this->info.max_output_ports == 0 ? this->n_outputs : this->info.max_output_ports;
+
+	impl->transport = pw_client_node0_transport_new(impl->context, max_inputs, max_outputs);
+	impl->transport->area->n_input_ports = n_inputs;
+	impl->transport->area->n_output_ports = n_outputs;
+
+	if (n_inputs > 0) {
+		items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_MEDIA_CLASS, "Stream/Input/Video");
+	} else {
+		items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_MEDIA_CLASS, "Stream/Output/Video");
+	}
+	pw_impl_node_update_properties(impl->this.node, &SPA_DICT_INIT(items, 1));
+}
+
+static void
+client_node0_done(void *data, int seq, int res)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	if (seq == 0 && res == 0 && impl->transport == NULL)
+		setup_transport(impl);
+
+	pw_log_debug("seq:%d res:%d pending:%d", seq, res, this->init_pending);
+	spa_node_emit_result(&this->hooks, seq, res, 0, NULL);
+
+	if (this->init_pending != SPA_ID_INVALID) {
+		spa_node_emit_result(&this->hooks, this->init_pending, res, 0, NULL);
+		this->init_pending = SPA_ID_INVALID;
+	}
+}
+
+static void
+client_node0_update(void *data,
+		   uint32_t change_mask,
+		   uint32_t max_input_ports,
+		   uint32_t max_output_ports,
+		   uint32_t n_params,
+		   const struct spa_pod **params)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	if (change_mask & PW_CLIENT_NODE0_UPDATE_MAX_INPUTS)
+		this->info.max_input_ports = max_input_ports;
+	if (change_mask & PW_CLIENT_NODE0_UPDATE_MAX_OUTPUTS)
+		this->info.max_output_ports = max_output_ports;
+	if (change_mask & PW_CLIENT_NODE0_UPDATE_PARAMS) {
+		uint32_t i;
+		spa_log_debug(this->log, "node %p: update %d params", this, n_params);
+
+		for (i = 0; i < this->n_params; i++)
+			free(this->params[i]);
+		this->n_params = n_params;
+		if (this->n_params == 0) {
+			free(this->params);
+			this->params = NULL;
+		} else {
+			void *p;
+			p = pw_reallocarray(this->params, this->n_params, sizeof(struct spa_pod *));
+			if (p == NULL) {
+				pw_log_error("%p: can't realloc: %m", this);
+				free(this->params);
+				this->n_params = 0;
+			}
+			this->params = p;
+		}
+		for (i = 0; i < this->n_params; i++)
+			this->params[i] = params[i] ? spa_pod_copy(params[i]) : NULL;
+	}
+	if (change_mask & (PW_CLIENT_NODE0_UPDATE_MAX_INPUTS | PW_CLIENT_NODE0_UPDATE_MAX_OUTPUTS)) {
+		spa_node_emit_info(&this->hooks, &this->info);
+	}
+
+	spa_log_debug(this->log, "node %p: got node update max_in %u, max_out %u", this,
+		     this->info.max_input_ports, this->info.max_output_ports);
+}
+
+static void
+client_node0_port_update(void *data,
+			enum spa_direction direction,
+			uint32_t port_id,
+			uint32_t change_mask,
+			uint32_t n_params,
+			const struct spa_pod **params,
+			const struct spa_port_info *info)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+	bool remove;
+
+	spa_log_debug(this->log, "node %p: got port update", this);
+	if (!CHECK_PORT_ID(this, direction, port_id))
+		return;
+
+	remove = (change_mask == 0);
+
+	if (remove) {
+		do_uninit_port(this, direction, port_id);
+	} else {
+		do_update_port(this,
+			       direction,
+			       port_id,
+			       change_mask,
+			       n_params, params, info);
+	}
+}
+
+static void client_node0_set_active(void *data, bool active)
+{
+	struct impl *impl = data;
+	pw_impl_node_set_active(impl->this.node, active);
+}
+
+static void client_node0_event(void *data, struct spa_event *event)
+{
+	struct impl *impl = data;
+	struct node *this = &impl->node;
+
+	switch (SPA_EVENT_TYPE(event)) {
+	case SPA_NODE0_EVENT_RequestClockUpdate:
+		send_clock_update(this);
+		break;
+	default:
+		spa_node_emit_event(&this->hooks, event);
+	}
+}
+
+static const struct pw_client_node0_methods client_node0_methods = {
+	PW_VERSION_CLIENT_NODE0_METHODS,
+	.done = client_node0_done,
+	.update = client_node0_update,
+	.port_update = client_node0_port_update,
+	.set_active = client_node0_set_active,
+	.event = client_node0_event,
+};
+
+static void node_on_data_fd_events(struct spa_source *source)
+{
+	struct node *this = source->data;
+	struct impl *impl = this->impl;
+
+	if (source->rmask & (SPA_IO_ERR | SPA_IO_HUP)) {
+		spa_log_warn(this->log, "node %p: got error", this);
+		return;
+	}
+
+	if (source->rmask & SPA_IO_IN) {
+		struct pw_client_node0_message message;
+		uint64_t cmd;
+
+		if (spa_system_eventfd_read(this->data_system, this->data_source.fd, &cmd) < 0)
+			spa_log_warn(this->log, "node %p: error reading message: %s",
+					this, strerror(errno));
+
+		while (pw_client_node0_transport_next_message(impl->transport, &message) == 1) {
+			struct pw_client_node0_message *msg = alloca(SPA_POD_SIZE(&message));
+			pw_client_node0_transport_parse_message(impl->transport, msg);
+			handle_node_message(this, msg);
+		}
+	}
+}
+
+static const struct spa_node_methods impl_node = {
+	SPA_VERSION_NODE_METHODS,
+	.add_listener = impl_node_add_listener,
+	.set_callbacks = impl_node_set_callbacks,
+	.sync = impl_node_sync,
+	.enum_params = impl_node_enum_params,
+	.set_param = impl_node_set_param,
+	.set_io = impl_node_set_io,
+	.send_command = impl_node_send_command,
+	.add_port = impl_node_add_port,
+	.remove_port = impl_node_remove_port,
+	.port_enum_params = impl_node_port_enum_params,
+	.port_set_param = impl_node_port_set_param,
+	.port_use_buffers = impl_node_port_use_buffers,
+	.port_set_io = impl_node_port_set_io,
+	.port_reuse_buffer = impl_node_port_reuse_buffer,
+	.process = impl_node_process,
+};
+
+static int
+node_init(struct node *this,
+	  struct spa_dict *info,
+	  const struct spa_support *support,
+	  uint32_t n_support)
+{
+	this->log = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_Log);
+	this->data_loop = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataLoop);
+	this->data_system = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataSystem);
+
+	if (this->data_loop == NULL) {
+		spa_log_error(this->log, "a data-loop is needed");
+		return -EINVAL;
+	}
+
+	this->node.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_Node,
+			SPA_VERSION_NODE,
+			&impl_node, this);
+	spa_hook_list_init(&this->hooks);
+
+	this->data_source.func = node_on_data_fd_events;
+	this->data_source.data = this;
+	this->data_source.fd = -1;
+	this->data_source.mask = SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP;
+	this->data_source.rmask = 0;
+
+	this->seq = 1;
+	this->init_pending = SPA_ID_INVALID;
+
+	return SPA_RESULT_RETURN_ASYNC(this->seq++);
+}
+
+static int node_clear(struct node *this)
+{
+	uint32_t i;
+
+	for (i = 0; i < MAX_INPUTS; i++) {
+		if (this->in_ports[i].valid)
+			clear_port(this, &this->in_ports[i], SPA_DIRECTION_INPUT, i);
+	}
+	for (i = 0; i < MAX_OUTPUTS; i++) {
+		if (this->out_ports[i].valid)
+			clear_port(this, &this->out_ports[i], SPA_DIRECTION_OUTPUT, i);
+	}
+
+	return 0;
+}
+
+static int do_remove_source(struct spa_loop *loop,
+			    bool async,
+			    uint32_t seq,
+			    const void *data,
+			    size_t size,
+			    void *user_data)
+{
+	struct spa_source *source = user_data;
+	spa_loop_remove_source(loop, source);
+	return 0;
+}
+
+static void client_node0_resource_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node0 *this = &impl->this;
+	struct node *node = &impl->node;
+
+	pw_log_debug("client-node %p: destroy", impl);
+
+	impl->node.resource = this->resource = NULL;
+	spa_hook_remove(&impl->resource_listener);
+	spa_hook_remove(&impl->object_listener);
+
+	if (node->data_source.fd != -1) {
+		spa_loop_invoke(node->data_loop,
+				do_remove_source,
+				SPA_ID_INVALID,
+				NULL,
+				0,
+				true,
+				&node->data_source);
+	}
+	if (this->node)
+		pw_impl_node_destroy(this->node);
+}
+
+static void node_initialized(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node0 *this = &impl->this;
+	struct pw_impl_node *node = this->node;
+	struct spa_system *data_system = impl->node.data_system;
+
+	if (this->resource == NULL)
+		return;
+
+	impl->fds[0] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+	impl->fds[1] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+	impl->node.data_source.fd = impl->fds[0];
+	impl->node.writefd = impl->fds[1];
+	impl->other_fds[0] = impl->fds[1];
+	impl->other_fds[1] = impl->fds[0];
+
+	spa_loop_add_source(impl->node.data_loop, &impl->node.data_source);
+	pw_log_debug("client-node %p: transport fd %d %d", node, impl->fds[0], impl->fds[1]);
+
+	pw_client_node0_resource_transport(this->resource,
+					  pw_global_get_id(pw_impl_node_get_global(node)),
+					  impl->other_fds[0],
+					  impl->other_fds[1],
+					  impl->transport);
+}
+
+static void node_free(void *data)
+{
+	struct impl *impl = data;
+	struct pw_impl_client_node0 *this = &impl->this;
+	struct spa_system *data_system = impl->node.data_system;
+
+	this->node = NULL;
+
+	pw_log_debug("client-node %p: free", &impl->this);
+	node_clear(&impl->node);
+
+	if (impl->transport)
+		pw_client_node0_transport_destroy(impl->transport);
+
+	spa_hook_remove(&impl->node_listener);
+
+	if (this->resource)
+		pw_resource_destroy(this->resource);
+
+	pw_array_clear(&impl->mems);
+
+	if (impl->fds[0] != -1)
+		spa_system_close(data_system, impl->fds[0]);
+	if (impl->fds[1] != -1)
+		spa_system_close(data_system, impl->fds[1]);
+	free(impl);
+}
+
+static const struct pw_impl_node_events node_events = {
+	PW_VERSION_IMPL_NODE_EVENTS,
+	.free = node_free,
+	.initialized = node_initialized,
+};
+
+static const struct pw_resource_events resource_events = {
+	PW_VERSION_RESOURCE_EVENTS,
+	.destroy = client_node0_resource_destroy,
+};
+
+static void convert_properties(struct pw_properties *properties)
+{
+	static const struct {
+		const char *from, *to;
+	} props[] = {
+		{ "pipewire.autoconnect", PW_KEY_NODE_AUTOCONNECT, },
+		/* XXX deprecated */
+		{ "pipewire.target.node", PW_KEY_NODE_TARGET, }
+	};
+
+	const char *str;
+
+	SPA_FOR_EACH_ELEMENT_VAR(props, p) {
+		if ((str = pw_properties_get(properties, p->from)) != NULL) {
+			pw_properties_set(properties, p->to, str);
+			pw_properties_set(properties, p->from, NULL);
+		}
+	}
+}
+
+/** Create a new client node
+ * \param client an owner \ref pw_client
+ * \param id an id
+ * \param name a name
+ * \param properties extra properties
+ * \return a newly allocated client node
+ *
+ * Create a new \ref pw_impl_node.
+ *
+ * \memberof pw_impl_client_node
+ */
+struct pw_impl_client_node0 *pw_impl_client_node0_new(struct pw_resource *resource,
+					  struct pw_properties *properties)
+{
+	struct impl *impl;
+	struct pw_impl_client_node0 *this;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct pw_context *context = pw_impl_client_get_context(client);
+	const struct spa_support *support;
+	uint32_t n_support;
+	const char *name;
+	int res;
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL)
+		return NULL;
+
+	this = &impl->this;
+
+	if (properties == NULL)
+		properties = pw_properties_new(NULL, NULL);
+	if (properties == NULL) {
+		res = -errno;
+		goto error_exit_free;
+	}
+	convert_properties(properties);
+
+	pw_properties_setf(properties, PW_KEY_CLIENT_ID, "%d", client->global->id);
+
+	impl->context = context;
+	impl->fds[0] = impl->fds[1] = -1;
+	pw_log_debug("client-node %p: new", impl);
+
+	support = pw_context_get_support(impl->context, &n_support);
+
+	node_init(&impl->node, NULL, support, n_support);
+	impl->node.impl = impl;
+
+	pw_array_init(&impl->mems, 64);
+
+	if ((name = pw_properties_get(properties, "node.name")) == NULL)
+		name = "client-node";
+	pw_properties_set(properties, PW_KEY_MEDIA_TYPE, "Video");
+
+	impl->node.resource = resource;
+	this->resource = resource;
+	this->node = pw_spa_node_new(context,
+				     PW_SPA_NODE_FLAG_ASYNC,
+				     &impl->node.node,
+				     NULL,
+				     properties, 0);
+	if (this->node == NULL) {
+		res = -errno;
+		goto error_no_node;
+	}
+
+	impl->client_reuse = pw_properties_get_bool(properties, "pipewire.client.reuse", false);
+
+	pw_resource_add_listener(this->resource,
+				 &impl->resource_listener,
+				 &resource_events,
+				 impl);
+	pw_resource_add_object_listener(this->resource,
+				&impl->object_listener,
+				&client_node0_methods,
+				impl);
+
+
+	pw_impl_node_add_listener(this->node, &impl->node_listener, &node_events, impl);
+
+	return this;
+
+error_no_node:
+	pw_resource_destroy(this->resource);
+	node_clear(&impl->node);
+error_exit_free:
+	free(impl);
+	errno = -res;
+	return NULL;
+}
+
+/** Destroy a client node
+ * \param node the client node to destroy
+ * \memberof pw_impl_client_node
+ */
+void pw_impl_client_node0_destroy(struct pw_impl_client_node0 *node)
+{
+	pw_resource_destroy(node->resource);
+}
diff --git a/src/modules/module-client-node/v0/client-node.h b/src/modules/module-client-node/v0/client-node.h
new file mode 100644
index 0000000..04b77d4
--- /dev/null
+++ b/src/modules/module-client-node/v0/client-node.h
@@ -0,0 +1,101 @@
+/* PipeWire
+ *
+ * Copyright © 2015 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef PIPEWIRE_CLIENT_NODE0_H
+#define PIPEWIRE_CLIENT_NODE0_H
+
+#include <pipewire/impl.h>
+
+#include "ext-client-node.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** The state of the clock */
+enum spa_clock0_state {
+	SPA_CLOCK0_STATE_STOPPED,	/*< the clock is stopped */
+	SPA_CLOCK0_STATE_PAUSED,	/*< the clock is paused */
+	SPA_CLOCK0_STATE_RUNNING,	/*< the clock is running */
+};
+
+struct spa_command_node0_clock_update_body {
+	struct spa_pod_object_body body;
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_TIME      (1 << 0)
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_SCALE     (1 << 1)
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_STATE     (1 << 2)
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_LATENCY   (1 << 3)
+	struct spa_pod_int change_mask          SPA_ALIGNED(8);
+	struct spa_pod_int rate                 SPA_ALIGNED(8);
+	struct spa_pod_long ticks               SPA_ALIGNED(8);
+	struct spa_pod_long monotonic_time      SPA_ALIGNED(8);
+	struct spa_pod_long offset              SPA_ALIGNED(8);
+	struct spa_pod_int scale                SPA_ALIGNED(8);
+	struct spa_pod_int state                SPA_ALIGNED(8);
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_FLAG_LIVE (1 << 0)
+	struct spa_pod_int flags                SPA_ALIGNED(8);
+	struct spa_pod_long latency             SPA_ALIGNED(8);
+};
+
+struct spa_command_node0_clock_update {
+	struct spa_pod pod;
+	struct spa_command_node0_clock_update_body body;
+};
+
+#define SPA_COMMAND_NODE0_CLOCK_UPDATE_INIT(type,change_mask,rate,ticks,monotonic_time,offset,scale,state,flags,latency)  \
+	SPA_COMMAND_INIT_FULL(struct spa_command_node0_clock_update,			\
+                        sizeof(struct spa_command_node0_clock_update_body), 0, type,	\
+			SPA_POD_INIT_Int(change_mask),					\
+			SPA_POD_INIT_Int(rate),						\
+			SPA_POD_INIT_Long(ticks),					\
+			SPA_POD_INIT_Long(monotonic_time),				\
+			SPA_POD_INIT_Long(offset),					\
+			SPA_POD_INIT_Int(scale),					\
+			SPA_POD_INIT_Int(state),					\
+			SPA_POD_INIT_Int(flags),					\
+			SPA_POD_INIT_Long(latency))
+
+
+/** \class pw_impl_client_node0
+ *
+ * PipeWire client node interface
+ */
+struct pw_impl_client_node0 {
+	struct pw_impl_node *node;
+
+	struct pw_resource *resource;
+};
+
+struct pw_impl_client_node0 *
+pw_impl_client_node0_new(struct pw_resource *resource,
+		   struct pw_properties *properties);
+
+void
+pw_impl_client_node0_destroy(struct pw_impl_client_node0 *node);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PIPEWIRE_CLIENT_NODE0_H */
diff --git a/src/modules/module-client-node/v0/ext-client-node.h b/src/modules/module-client-node/v0/ext-client-node.h
new file mode 100644
index 0000000..21ba597
--- /dev/null
+++ b/src/modules/module-client-node/v0/ext-client-node.h
@@ -0,0 +1,414 @@
+/* PipeWire
+ *
+ * Copyright © 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __PIPEWIRE_EXT_CLIENT_NODE0_H__
+#define __PIPEWIRE_EXT_CLIENT_NODE0_H__
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <spa/utils/defs.h>
+#include <spa/param/param.h>
+#include <spa/node/node.h>
+
+#include <pipewire/proxy.h>
+
+#define PW_TYPE_INTERFACE_ClientNode            PW_TYPE_INFO_INTERFACE_BASE "ClientNode"
+
+#define PW_VERSION_CLIENT_NODE0		0
+
+struct pw_client_node0_message;
+
+/** Shared structure between client and server \memberof pw_client_node */
+struct pw_client_node0_area {
+	uint32_t max_input_ports;	/**< max input ports of the node */
+	uint32_t n_input_ports;		/**< number of input ports of the node */
+	uint32_t max_output_ports;	/**< max output ports of the node */
+	uint32_t n_output_ports;	/**< number of output ports of the node */
+};
+
+/** \class pw_client_node0_transport
+ *
+ * \brief Transport object
+ *
+ * The transport object contains shared data and ringbuffers to exchange
+ * events and data between the server and the client in a low-latency and
+ * lockfree way.
+ */
+struct pw_client_node0_transport {
+	struct pw_client_node0_area *area;	/**< the transport area */
+	struct spa_io_buffers *inputs;		/**< array of buffer input io */
+	struct spa_io_buffers *outputs;		/**< array of buffer output io */
+	void *input_data;			/**< input memory for ringbuffer */
+	struct spa_ringbuffer *input_buffer;	/**< ringbuffer for input memory */
+	void *output_data;			/**< output memory for ringbuffer */
+	struct spa_ringbuffer *output_buffer;	/**< ringbuffer for output memory */
+
+	/** Destroy a transport
+	 * \param trans a transport to destroy
+	 * \memberof pw_client_node0_transport
+	 */
+	void (*destroy) (struct pw_client_node0_transport *trans);
+
+	/** Add a message to the transport
+	 * \param trans the transport to send the message on
+	 * \param message the message to add
+	 * \return 0 on success, < 0 on error
+	 *
+	 * Write \a message to the shared ringbuffer.
+	 */
+	int (*add_message) (struct pw_client_node0_transport *trans, struct pw_client_node0_message *message);
+
+	/** Get next message from a transport
+	 * \param trans the transport to get the message of
+	 * \param[out] message the message to read
+	 * \return < 0 on error, 1 when a message is available,
+	 *           0 when no more messages are available.
+	 *
+	 * Get the skeleton next message from \a trans into \a message. This function will
+	 * only read the head and object body of the message.
+	 *
+	 * After the complete size of the message has been calculated, you should call
+	 * \ref parse_message() to read the complete message contents.
+	 */
+	int (*next_message) (struct pw_client_node0_transport *trans, struct pw_client_node0_message *message);
+
+	/** Parse the complete message on transport
+	 * \param trans the transport to read from
+	 * \param[out] message memory that can hold the complete message
+	 * \return 0 on success, < 0 on error
+	 *
+	 * Use this function after \ref next_message().
+	 */
+	int (*parse_message) (struct pw_client_node0_transport *trans, void *message);
+};
+
+#define pw_client_node0_transport_destroy(t)		((t)->destroy((t)))
+#define pw_client_node0_transport_add_message(t,m)	((t)->add_message((t), (m)))
+#define pw_client_node0_transport_next_message(t,m)	((t)->next_message((t), (m)))
+#define pw_client_node0_transport_parse_message(t,m)	((t)->parse_message((t), (m)))
+
+enum pw_client_node0_message_type {
+	PW_CLIENT_NODE0_MESSAGE_HAVE_OUTPUT,		/*< signal that the node has output */
+	PW_CLIENT_NODE0_MESSAGE_NEED_INPUT,		/*< signal that the node needs input */
+	PW_CLIENT_NODE0_MESSAGE_PROCESS_INPUT,		/*< instruct the node to process input */
+	PW_CLIENT_NODE0_MESSAGE_PROCESS_OUTPUT,		/*< instruct the node output is processed */
+	PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER,	/*< reuse a buffer */
+};
+
+struct pw_client_node0_message_body {
+	struct spa_pod_int type		SPA_ALIGNED(8);	/*< one of enum pw_client_node0_message_type */
+};
+
+struct pw_client_node0_message {
+	struct spa_pod_struct pod;
+	struct pw_client_node0_message_body body;
+};
+
+struct pw_client_node0_message_port_reuse_buffer_body {
+	struct spa_pod_int type		SPA_ALIGNED(8);	/*< PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER */
+	struct spa_pod_int port_id	SPA_ALIGNED(8);	/*< port id */
+	struct spa_pod_int buffer_id	SPA_ALIGNED(8); /*< buffer id to reuse */
+};
+
+struct pw_client_node0_message_port_reuse_buffer {
+	struct spa_pod_struct pod;
+	struct pw_client_node0_message_port_reuse_buffer_body body;
+};
+
+#define PW_CLIENT_NODE0_MESSAGE_TYPE(message)	(((struct pw_client_node0_message*)(message))->body.type.value)
+
+#define PW_CLIENT_NODE0_MESSAGE_INIT(message) ((struct pw_client_node0_message)			\
+	{ { { sizeof(struct pw_client_node0_message_body), SPA_TYPE_Struct } },			\
+	  { SPA_POD_INIT_Int(message) } })
+
+#define PW_CLIENT_NODE0_MESSAGE_INIT_FULL(type,size,message,...) (type)				\
+	{ { { size, SPA_TYPE_Struct } },							\
+	  { SPA_POD_INIT_Int(message), ##__VA_ARGS__ } }					\
+
+#define PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER_INIT(port_id,buffer_id)			\
+	PW_CLIENT_NODE0_MESSAGE_INIT_FULL(struct pw_client_node0_message_port_reuse_buffer,	\
+		sizeof(struct pw_client_node0_message_port_reuse_buffer_body),			\
+		PW_CLIENT_NODE0_MESSAGE_PORT_REUSE_BUFFER,					\
+		SPA_POD_INIT_Int(port_id),							\
+		SPA_POD_INIT_Int(buffer_id))
+
+/** information about a buffer */
+struct pw_client_node0_buffer {
+	uint32_t mem_id;		/**< the memory id for the metadata */
+	uint32_t offset;		/**< offset in memory */
+	uint32_t size;			/**< size in memory */
+	struct spa_buffer *buffer;	/**< buffer describing metadata and buffer memory */
+};
+
+#define PW_CLIENT_NODE0_METHOD_DONE		0
+#define PW_CLIENT_NODE0_METHOD_UPDATE		1
+#define PW_CLIENT_NODE0_METHOD_PORT_UPDATE	2
+#define PW_CLIENT_NODE0_METHOD_SET_ACTIVE		3
+#define PW_CLIENT_NODE0_METHOD_EVENT		4
+#define PW_CLIENT_NODE0_METHOD_DESTROY		5
+#define PW_CLIENT_NODE0_METHOD_NUM		6
+
+/** \ref pw_client_node methods */
+struct pw_client_node0_methods {
+#define PW_VERSION_CLIENT_NODE0_METHODS		0
+	uint32_t version;
+
+	/** Complete an async operation */
+	void (*done) (void *object, int seq, int res);
+
+	/**
+	 * Update the node ports and properties
+	 *
+	 * Update the maximum number of ports and the params of the
+	 * client node.
+	 * \param change_mask bitfield with changed parameters
+	 * \param max_input_ports new max input ports
+	 * \param max_output_ports new max output ports
+	 * \param params new params
+	 */
+	void (*update) (void *object,
+#define PW_CLIENT_NODE0_UPDATE_MAX_INPUTS   (1 << 0)
+#define PW_CLIENT_NODE0_UPDATE_MAX_OUTPUTS  (1 << 1)
+#define PW_CLIENT_NODE0_UPDATE_PARAMS       (1 << 2)
+			uint32_t change_mask,
+			uint32_t max_input_ports,
+			uint32_t max_output_ports,
+			uint32_t n_params,
+			const struct spa_pod **params);
+
+	/**
+	 * Update a node port
+	 *
+	 * Update the information of one port of a node.
+	 * \param direction the direction of the port
+	 * \param port_id the port id to update
+	 * \param change_mask a bitfield of changed items
+	 * \param n_params number of port parameters
+	 * \param params array of port parameters
+	 * \param info port information
+	 */
+	void (*port_update) (void *object,
+			     enum spa_direction direction,
+			     uint32_t port_id,
+#define PW_CLIENT_NODE0_PORT_UPDATE_PARAMS            (1 << 0)
+#define PW_CLIENT_NODE0_PORT_UPDATE_INFO              (1 << 1)
+			     uint32_t change_mask,
+			     uint32_t n_params,
+			     const struct spa_pod **params,
+			     const struct spa_port_info *info);
+	/**
+	 * Activate or deactivate the node
+	 */
+	void (*set_active) (void *object, bool active);
+	/**
+	 * Send an event to the node
+	 * \param event the event to send
+	 */
+	void (*event) (void *object, struct spa_event *event);
+	/**
+	 * Destroy the client_node
+	 */
+	void (*destroy) (void *object);
+};
+
+#define PW_CLIENT_NODE0_EVENT_ADD_MEM		0
+#define PW_CLIENT_NODE0_EVENT_TRANSPORT		1
+#define PW_CLIENT_NODE0_EVENT_SET_PARAM		2
+#define PW_CLIENT_NODE0_EVENT_EVENT		3
+#define PW_CLIENT_NODE0_EVENT_COMMAND		4
+#define PW_CLIENT_NODE0_EVENT_ADD_PORT		5
+#define PW_CLIENT_NODE0_EVENT_REMOVE_PORT	6
+#define PW_CLIENT_NODE0_EVENT_PORT_SET_PARAM	7
+#define PW_CLIENT_NODE0_EVENT_PORT_USE_BUFFERS	8
+#define PW_CLIENT_NODE0_EVENT_PORT_COMMAND	9
+#define PW_CLIENT_NODE0_EVENT_PORT_SET_IO	10
+#define PW_CLIENT_NODE0_EVENT_NUM		11
+
+/** \ref pw_client_node events */
+struct pw_client_node0_events {
+#define PW_VERSION_CLIENT_NODE0_EVENTS		0
+	uint32_t version;
+	/**
+	 * Memory was added to a node
+	 *
+	 * \param mem_id the id of the memory
+	 * \param type the memory type
+	 * \param memfd the fd of the memory
+	 * \param flags flags for the \a memfd
+	 */
+	void (*add_mem) (void *data,
+			 uint32_t mem_id,
+			 uint32_t type,
+			 int memfd,
+			 uint32_t flags);
+	/**
+	 * Notify of a new transport area
+	 *
+	 * The transport area is used to exchange real-time commands between
+	 * the client and the server.
+	 *
+	 * \param node_id the node id created for this client node
+	 * \param readfd fd for signal data can be read
+	 * \param writefd fd for signal data can be written
+	 * \param transport the shared transport area
+	 */
+	void (*transport) (void *data,
+			   uint32_t node_id,
+			   int readfd,
+			   int writefd,
+			   struct pw_client_node0_transport *transport);
+	/**
+	 * Notify of a property change
+	 *
+	 * When the server configures the properties on the node
+	 * this event is sent
+	 *
+	 * \param seq a sequence number
+	 * \param id the id of the parameter
+	 * \param flags parameter flags
+	 * \param param the param to set
+	 */
+	void (*set_param) (void *data, uint32_t seq,
+			   uint32_t id, uint32_t flags,
+			   const struct spa_pod *param);
+	/**
+	 * Receive an event from the client node
+	 * \param event the received event */
+	void (*event) (void *data, const struct spa_event *event);
+	/**
+	 * Notify of a new node command
+	 *
+	 * \param seq a sequence number
+	 * \param command the command
+	 */
+	void (*command) (void *data, uint32_t seq, const struct spa_command *command);
+	/**
+	 * A new port was added to the node
+	 *
+	 * The server can at any time add a port to the node when there
+	 * are free ports available.
+	 *
+	 * \param seq a sequence number
+	 * \param direction the direction of the port
+	 * \param port_id the new port id
+	 */
+	void (*add_port) (void *data,
+			  uint32_t seq,
+			  enum spa_direction direction,
+			  uint32_t port_id);
+	/**
+	 * A port was removed from the node
+	 *
+	 * \param seq a sequence number
+	 * \param direction a port direction
+	 * \param port_id the remove port id
+	 */
+	void (*remove_port) (void *data,
+			     uint32_t seq,
+			     enum spa_direction direction,
+			     uint32_t port_id);
+	/**
+	 * A parameter was configured on the port
+	 *
+	 * \param seq a sequence number
+	 * \param direction a port direction
+	 * \param port_id the port id
+	 * \param id the id of the parameter
+	 * \param flags flags used when setting the param
+	 * \param param the new param
+	 */
+	void (*port_set_param) (void *data,
+				uint32_t seq,
+				enum spa_direction direction,
+				uint32_t port_id,
+				uint32_t id, uint32_t flags,
+				const struct spa_pod *param);
+	/**
+	 * Notify the port of buffers
+	 *
+	 * \param seq a sequence number
+	 * \param direction a port direction
+	 * \param port_id the port id
+	 * \param n_buffer the number of buffers
+	 * \param buffers and array of buffer descriptions
+	 */
+	void (*port_use_buffers) (void *data,
+				  uint32_t seq,
+				  enum spa_direction direction,
+				  uint32_t port_id,
+				  uint32_t n_buffers,
+				  struct pw_client_node0_buffer *buffers);
+	/**
+	 * Notify of a new port command
+	 *
+	 * \param direction a port direction
+	 * \param port_id the port id
+	 * \param command the command
+	 */
+	void (*port_command) (void *data,
+			      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 *data,
+			     uint32_t seq,
+			     enum spa_direction direction,
+			     uint32_t port_id,
+			     uint32_t id,
+			     uint32_t mem_id,
+			     uint32_t offset,
+			     uint32_t size);
+};
+#define pw_client_node0_resource(r,m,v,...) pw_resource_call(r, struct pw_client_node0_events, m, v, ##__VA_ARGS__)
+
+#define pw_client_node0_resource_add_mem(r,...)		 pw_client_node0_resource(r,add_mem,0,__VA_ARGS__)
+#define pw_client_node0_resource_transport(r,...)	 pw_client_node0_resource(r,transport,0,__VA_ARGS__)
+#define pw_client_node0_resource_set_param(r,...)	 pw_client_node0_resource(r,set_param,0,__VA_ARGS__)
+#define pw_client_node0_resource_event(r,...)		 pw_client_node0_resource(r,event,0,__VA_ARGS__)
+#define pw_client_node0_resource_command(r,...)		 pw_client_node0_resource(r,command,0,__VA_ARGS__)
+#define pw_client_node0_resource_add_port(r,...)	 pw_client_node0_resource(r,add_port,0,__VA_ARGS__)
+#define pw_client_node0_resource_remove_port(r,...)	 pw_client_node0_resource(r,remove_port,0,__VA_ARGS__)
+#define pw_client_node0_resource_port_set_param(r,...)	 pw_client_node0_resource(r,port_set_param,0,__VA_ARGS__)
+#define pw_client_node0_resource_port_use_buffers(r,...) pw_client_node0_resource(r,port_use_buffers,0,__VA_ARGS__)
+#define pw_client_node0_resource_port_command(r,...)	 pw_client_node0_resource(r,port_command,0,__VA_ARGS__)
+#define pw_client_node0_resource_port_set_io(r,...)	 pw_client_node0_resource(r,port_set_io,0,__VA_ARGS__)
+
+#ifdef __cplusplus
+}  /* extern "C" */
+#endif
+
+#endif /* __PIPEWIRE_EXT_CLIENT_NODE0_H__ */
diff --git a/src/modules/module-client-node/v0/protocol-native.c b/src/modules/module-client-node/v0/protocol-native.c
new file mode 100644
index 0000000..86d8fe9
--- /dev/null
+++ b/src/modules/module-client-node/v0/protocol-native.c
@@ -0,0 +1,534 @@
+/* PipeWire
+ *
+ * Copyright © 2017 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <errno.h>
+
+#include <spa/pod/parser.h>
+#include <spa/pod/builder.h>
+#include <spa/utils/type-info.h>
+
+#include "pipewire/impl.h"
+
+#include "pipewire/extensions/protocol-native.h"
+
+#include "ext-client-node.h"
+
+#include "transport.h"
+
+#define PW_PROTOCOL_NATIVE_FLAG_REMAP        (1<<0)
+
+extern uint32_t pw_protocol_native0_find_type(struct pw_impl_client *client, const char *type);
+extern int pw_protocol_native0_pod_to_v2(struct pw_impl_client *client, const struct spa_pod *pod,
+		struct spa_pod_builder *b);
+extern struct spa_pod * pw_protocol_native0_pod_from_v2(struct pw_impl_client *client,
+		const struct spa_pod *pod);
+extern uint32_t pw_protocol_native0_type_to_v2(struct pw_impl_client *client,
+		const struct spa_type_info *info, uint32_t type);
+
+static void
+client_node_marshal_add_mem(void *data,
+			    uint32_t mem_id,
+			    uint32_t type,
+			    int memfd, uint32_t flags)
+{
+	struct pw_resource *resource = data;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_builder *b;
+	const char *typename;
+
+	switch (type) {
+	case SPA_DATA_MemFd:
+		typename = "Spa:Enum:DataType:Fd:MemFd";
+		break;
+	case SPA_DATA_DmaBuf:
+		typename = "Spa:Enum:DataType:Fd:DmaBuf";
+		break;
+	default:
+	case SPA_DATA_MemPtr:
+		return;
+
+	}
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_ADD_MEM, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       "i", mem_id,
+			       "I", pw_protocol_native0_find_type(client, typename),
+			       "i", pw_protocol_native_add_resource_fd(resource, memfd),
+			       "i", flags);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void client_node_marshal_transport(void *data, uint32_t node_id, int readfd, int writefd,
+					  struct pw_client_node0_transport *transport)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+	struct pw_client_node0_transport_info info;
+
+	pw_client_node0_transport_get_info(transport, &info);
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_TRANSPORT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       "i", node_id,
+			       "i", pw_protocol_native_add_resource_fd(resource, readfd),
+			       "i", pw_protocol_native_add_resource_fd(resource, writefd),
+			       "i", pw_protocol_native_add_resource_fd(resource, info.memfd),
+			       "i", info.offset,
+			       "i", info.size);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_set_param(void *data, uint32_t seq, uint32_t id, uint32_t flags,
+			      const struct spa_pod *param)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_SET_PARAM, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       "i", seq,
+			       "I", id,
+			       "i", flags,
+			       "P", param);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void client_node_marshal_event_event(void *data, const struct spa_event *event)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_EVENT, NULL);
+
+	spa_pod_builder_add_struct(b, "P", event);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_command(void *data, uint32_t seq, const struct spa_command *command)
+{
+	struct pw_resource *resource = data;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_COMMAND, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b, "i", seq, NULL);
+	if (SPA_COMMAND_TYPE(command) == 0)
+		spa_pod_builder_add(b, "P", command, NULL);
+	else
+		pw_protocol_native0_pod_to_v2(client, (struct spa_pod *)command, b);
+	spa_pod_builder_pop(b, &f);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_add_port(void *data,
+			     uint32_t seq, enum spa_direction direction, uint32_t port_id)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_ADD_PORT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       "i", seq,
+			       "i", direction,
+			       "i", port_id);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_remove_port(void *data,
+				uint32_t seq, enum spa_direction direction, uint32_t port_id)
+{
+	struct pw_resource *resource = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_REMOVE_PORT, NULL);
+
+	spa_pod_builder_add_struct(b,
+			       "i", seq,
+			       "i", direction,
+			       "i", port_id);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_port_set_param(void *data,
+				   uint32_t seq,
+				   enum spa_direction direction,
+				   uint32_t port_id,
+				   uint32_t id,
+				   uint32_t flags,
+				   const struct spa_pod *param)
+{
+	struct pw_resource *resource = data;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+	const char *typename;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_PORT_SET_PARAM, NULL);
+
+	switch (id) {
+	case SPA_PARAM_Props:
+		typename = "Spa:Enum:ParamId:Props";
+		break;
+	case SPA_PARAM_Format:
+		typename = "Spa:Enum:ParamId:Format";
+		break;
+	default:
+		return;
+	}
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			"i", seq,
+			"i", direction,
+			"i", port_id,
+			"I", pw_protocol_native0_find_type(client, typename),
+			"i", flags, NULL);
+	pw_protocol_native0_pod_to_v2(client, param, b);
+	spa_pod_builder_pop(b, &f);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_port_use_buffers(void *data,
+				     uint32_t seq,
+				     enum spa_direction direction,
+				     uint32_t port_id,
+				     uint32_t n_buffers, struct pw_client_node0_buffer *buffers)
+{
+	struct pw_resource *resource = data;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+	uint32_t i, j;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_PORT_USE_BUFFERS, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			    "i", seq,
+			    "i", direction,
+			    "i", port_id,
+			    "i", n_buffers, NULL);
+
+	for (i = 0; i < n_buffers; i++) {
+		struct spa_buffer *buf = buffers[i].buffer;
+
+		spa_pod_builder_add(b,
+				    "i", buffers[i].mem_id,
+				    "i", buffers[i].offset,
+				    "i", buffers[i].size,
+				    "i", i,
+				    "i", buf->n_metas, NULL);
+
+		for (j = 0; j < buf->n_metas; j++) {
+			struct spa_meta *m = &buf->metas[j];
+			spa_pod_builder_add(b,
+					    "I", pw_protocol_native0_type_to_v2(client, spa_type_meta_type, m->type),
+					    "i", m->size, NULL);
+		}
+		spa_pod_builder_add(b, "i", buf->n_datas, NULL);
+		for (j = 0; j < buf->n_datas; j++) {
+			struct spa_data *d = &buf->datas[j];
+			spa_pod_builder_add(b,
+					    "I", pw_protocol_native0_type_to_v2(client, spa_type_data_type, d->type),
+					    "i", SPA_PTR_TO_UINT32(d->data),
+					    "i", d->flags,
+					    "i", d->mapoffset,
+					    "i", d->maxsize, NULL);
+		}
+	}
+	spa_pod_builder_pop(b, &f);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_port_command(void *data,
+				 uint32_t direction,
+				 uint32_t port_id,
+				 const struct spa_command *command)
+{
+	struct pw_resource *resource = data;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_builder *b;
+	struct spa_pod_frame f;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_PORT_COMMAND, NULL);
+
+	spa_pod_builder_push_struct(b, &f);
+	spa_pod_builder_add(b,
+			"i", direction,
+			"i", port_id, NULL);
+	pw_protocol_native0_pod_to_v2(client, (struct spa_pod *)command, b);
+	spa_pod_builder_pop(b, &f);
+
+	pw_protocol_native_end_resource(resource, b);
+}
+
+static void
+client_node_marshal_port_set_io(void *data,
+				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 = data;
+	struct spa_pod_builder *b;
+
+	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE0_EVENT_PORT_SET_IO, NULL);
+
+	spa_pod_builder_add_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 int client_node_demarshal_done(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	uint32_t seq, res;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			"i", &seq,
+			"i", &res) < 0)
+		return -EINVAL;
+
+	return pw_resource_notify(resource, struct pw_client_node0_methods, done, 0, seq, res);
+}
+
+static int client_node_demarshal_update(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f;
+	uint32_t change_mask, max_input_ports, max_output_ports, n_params;
+	const struct spa_pod **params;
+	uint32_t i;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f) < 0 ||
+	    spa_pod_parser_get(&prs,
+			"i", &change_mask,
+			"i", &max_input_ports,
+			"i", &max_output_ports,
+			"i", &n_params, NULL) < 0)
+		return -EINVAL;
+
+	params = alloca(n_params * sizeof(struct spa_pod *));
+	for (i = 0; i < n_params; i++)
+		if (spa_pod_parser_get(&prs, "O", &params[i], NULL) < 0)
+			return -EINVAL;
+
+	return pw_resource_notify(resource, struct pw_client_node0_methods, update, 0, change_mask,
+									max_input_ports,
+									max_output_ports,
+									n_params,
+									params);
+}
+
+static int client_node_demarshal_port_update(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	struct spa_pod_frame f[2];
+	uint32_t i, direction, port_id, change_mask, n_params;
+	const struct spa_pod **params = NULL;
+	struct spa_port_info info = { 0 }, *infop = NULL;
+	struct spa_dict props;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_push_struct(&prs, &f[0]) < 0 ||
+	    spa_pod_parser_get(&prs,
+			"i", &direction,
+			"i", &port_id,
+			"i", &change_mask,
+			"i", &n_params, NULL) < 0)
+		return -EINVAL;
+
+	params = alloca(n_params * sizeof(struct spa_pod *));
+	for (i = 0; i < n_params; i++)
+		if (spa_pod_parser_get(&prs, "O", &params[i], NULL) < 0)
+			return -EINVAL;
+
+
+	if (spa_pod_parser_push_struct(&prs, &f[1]) >= 0) {
+		infop = &info;
+
+		if (spa_pod_parser_get(&prs,
+				"i", &info.flags,
+				"i", &info.rate,
+				"i", &props.n_items, NULL) < 0)
+			return -EINVAL;
+
+		if (props.n_items > 0) {
+			info.props = &props;
+
+			props.items = alloca(props.n_items * sizeof(struct spa_dict_item));
+			for (i = 0; i < props.n_items; i++) {
+				if (spa_pod_parser_get(&prs,
+						"s", &props.items[i].key,
+						"s", &props.items[i].value,
+						NULL) < 0)
+					return -EINVAL;
+			}
+		}
+	}
+
+	return pw_resource_notify(resource, struct pw_client_node0_methods, port_update, 0, direction,
+									     port_id,
+									     change_mask,
+									     n_params,
+									     params, infop);
+}
+
+static int client_node_demarshal_set_active(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	int active;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			"b", &active) < 0)
+		return -EINVAL;
+
+        return pw_resource_notify(resource, struct pw_client_node0_methods, set_active, 0, active);
+}
+
+static int client_node_demarshal_event_method(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct pw_impl_client *client = pw_resource_get_client(resource);
+	struct spa_pod_parser prs;
+	struct spa_event *event;
+	int res;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs,
+			"O", &event) < 0)
+		return -EINVAL;
+
+	event = (struct spa_event*)pw_protocol_native0_pod_from_v2(client, (struct spa_pod *)event);
+
+	res = pw_resource_notify(resource, struct pw_client_node0_methods, event, 0, event);
+	free(event);
+
+	return res;
+}
+
+static int client_node_demarshal_destroy(void *object, const struct pw_protocol_native_message *msg)
+{
+	struct pw_resource *resource = object;
+	struct spa_pod_parser prs;
+	int res;
+
+	spa_pod_parser_init(&prs, msg->data, msg->size);
+	if (spa_pod_parser_get_struct(&prs, NULL) < 0)
+		return -EINVAL;
+
+	res = pw_resource_notify(resource, struct pw_client_node0_methods, destroy, 0);
+	pw_resource_destroy(resource);
+	return res;
+}
+
+static const struct pw_protocol_native_demarshal pw_protocol_native_client_node_method_demarshal[] = {
+	{ &client_node_demarshal_done, 0, 0 },
+	{ &client_node_demarshal_update, 0, PW_PROTOCOL_NATIVE_FLAG_REMAP },
+	{ &client_node_demarshal_port_update, 0, PW_PROTOCOL_NATIVE_FLAG_REMAP },
+	{ &client_node_demarshal_set_active, 0, 0 },
+	{ &client_node_demarshal_event_method, 0, PW_PROTOCOL_NATIVE_FLAG_REMAP },
+	{ &client_node_demarshal_destroy, 0, 0 },
+};
+
+static const struct pw_client_node0_events pw_protocol_native_client_node_event_marshal = {
+	PW_VERSION_CLIENT_NODE0_EVENTS,
+	&client_node_marshal_add_mem,
+	&client_node_marshal_transport,
+	&client_node_marshal_set_param,
+	&client_node_marshal_event_event,
+	&client_node_marshal_command,
+	&client_node_marshal_add_port,
+	&client_node_marshal_remove_port,
+	&client_node_marshal_port_set_param,
+	&client_node_marshal_port_use_buffers,
+	&client_node_marshal_port_command,
+	&client_node_marshal_port_set_io,
+};
+
+static const struct pw_protocol_marshal pw_protocol_native_client_node_marshal = {
+	PW_TYPE_INTERFACE_ClientNode,
+	PW_VERSION_CLIENT_NODE0,
+	PW_CLIENT_NODE0_METHOD_NUM,
+	PW_CLIENT_NODE0_EVENT_NUM,
+	0,
+	NULL,
+	.server_demarshal = &pw_protocol_native_client_node_method_demarshal,
+	.server_marshal = &pw_protocol_native_client_node_event_marshal,
+	NULL,
+};
+
+struct pw_protocol *pw_protocol_native_ext_client_node0_init(struct pw_context *context)
+{
+	struct pw_protocol *protocol;
+
+	protocol = pw_context_find_protocol(context, PW_TYPE_INFO_PROTOCOL_Native);
+
+	if (protocol == NULL)
+		return NULL;
+
+	pw_protocol_add_marshal(protocol, &pw_protocol_native_client_node_marshal);
+
+	return protocol;
+}
diff --git a/src/modules/module-client-node/v0/transport.c b/src/modules/module-client-node/v0/transport.c
new file mode 100644
index 0000000..fa84795
--- /dev/null
+++ b/src/modules/module-client-node/v0/transport.c
@@ -0,0 +1,262 @@
+/* PipeWire
+ *
+ * Copyright © 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <unistd.h>
+#include <errno.h>
+#include <sys/mman.h>
+
+#include <spa/utils/ringbuffer.h>
+#include <spa/node/io.h>
+
+#include <pipewire/impl.h>
+#include <pipewire/private.h>
+
+#include "ext-client-node.h"
+
+#include "transport.h"
+
+/** \cond */
+
+#define INPUT_BUFFER_SIZE       (1<<12)
+#define OUTPUT_BUFFER_SIZE      (1<<12)
+
+struct transport {
+	struct pw_client_node0_transport trans;
+
+	struct pw_memblock *mem;
+	size_t offset;
+
+	struct pw_client_node0_message current;
+	uint32_t current_index;
+};
+/** \endcond */
+
+static size_t area_get_size(struct pw_client_node0_area *area)
+{
+	size_t size;
+	size = sizeof(struct pw_client_node0_area);
+	size += area->max_input_ports * sizeof(struct spa_io_buffers);
+	size += area->max_output_ports * sizeof(struct spa_io_buffers);
+	size += sizeof(struct spa_ringbuffer);
+	size += INPUT_BUFFER_SIZE;
+	size += sizeof(struct spa_ringbuffer);
+	size += OUTPUT_BUFFER_SIZE;
+	return size;
+}
+
+static void transport_setup_area(void *p, struct pw_client_node0_transport *trans)
+{
+	struct pw_client_node0_area *a;
+
+	trans->area = a = p;
+	p = SPA_PTROFF(p, sizeof(struct pw_client_node0_area), struct spa_io_buffers);
+
+	trans->inputs = p;
+	p = SPA_PTROFF(p, a->max_input_ports * sizeof(struct spa_io_buffers), void);
+
+	trans->outputs = p;
+	p = SPA_PTROFF(p, a->max_output_ports * sizeof(struct spa_io_buffers), void);
+
+	trans->input_buffer = p;
+	p = SPA_PTROFF(p, sizeof(struct spa_ringbuffer), void);
+
+	trans->input_data = p;
+	p = SPA_PTROFF(p, INPUT_BUFFER_SIZE, void);
+
+	trans->output_buffer = p;
+	p = SPA_PTROFF(p, sizeof(struct spa_ringbuffer), void);
+
+	trans->output_data = p;
+	p = SPA_PTROFF(p, OUTPUT_BUFFER_SIZE, void);
+}
+
+static void transport_reset_area(struct pw_client_node0_transport *trans)
+{
+	uint32_t i;
+	struct pw_client_node0_area *a = trans->area;
+
+	for (i = 0; i < a->max_input_ports; i++) {
+		trans->inputs[i] = SPA_IO_BUFFERS_INIT;
+	}
+	for (i = 0; i < a->max_output_ports; i++) {
+		trans->outputs[i] = SPA_IO_BUFFERS_INIT;
+	}
+	spa_ringbuffer_init(trans->input_buffer);
+	spa_ringbuffer_init(trans->output_buffer);
+}
+
+static void destroy(struct pw_client_node0_transport *trans)
+{
+	struct transport *impl = (struct transport *) trans;
+
+	pw_log_debug("transport %p: destroy", trans);
+
+	pw_memblock_free(impl->mem);
+	free(impl);
+}
+
+static int add_message(struct pw_client_node0_transport *trans, struct pw_client_node0_message *message)
+{
+	struct transport *impl = (struct transport *) trans;
+	int32_t filled, avail;
+	uint32_t size, index;
+
+	if (impl == NULL || message == NULL)
+		return -EINVAL;
+
+	filled = spa_ringbuffer_get_write_index(trans->output_buffer, &index);
+	avail = OUTPUT_BUFFER_SIZE - filled;
+	size = SPA_POD_SIZE(message);
+	if (avail < (int)size)
+		return -ENOSPC;
+
+	spa_ringbuffer_write_data(trans->output_buffer,
+				  trans->output_data, OUTPUT_BUFFER_SIZE,
+				  index & (OUTPUT_BUFFER_SIZE - 1), message, size);
+	spa_ringbuffer_write_update(trans->output_buffer, index + size);
+
+	return 0;
+}
+
+static int next_message(struct pw_client_node0_transport *trans, struct pw_client_node0_message *message)
+{
+	struct transport *impl = (struct transport *) trans;
+	int32_t avail;
+
+	if (impl == NULL || message == NULL)
+		return -EINVAL;
+
+	avail = spa_ringbuffer_get_read_index(trans->input_buffer, &impl->current_index);
+	if (avail < (int) sizeof(struct pw_client_node0_message))
+		return 0;
+
+	spa_ringbuffer_read_data(trans->input_buffer,
+				 trans->input_data, INPUT_BUFFER_SIZE,
+				 impl->current_index & (INPUT_BUFFER_SIZE - 1),
+				 &impl->current, sizeof(struct pw_client_node0_message));
+
+	if (avail < (int) SPA_POD_SIZE(&impl->current))
+		return 0;
+
+	*message = impl->current;
+
+	return 1;
+}
+
+static int parse_message(struct pw_client_node0_transport *trans, void *message)
+{
+	struct transport *impl = (struct transport *) trans;
+	uint32_t size;
+
+	if (impl == NULL || message == NULL)
+		return -EINVAL;
+
+	size = SPA_POD_SIZE(&impl->current);
+
+	spa_ringbuffer_read_data(trans->input_buffer,
+				 trans->input_data, INPUT_BUFFER_SIZE,
+				 impl->current_index & (INPUT_BUFFER_SIZE - 1), message, size);
+	spa_ringbuffer_read_update(trans->input_buffer, impl->current_index + size);
+
+	return 0;
+}
+
+/** Create a new transport
+ * \param max_input_ports maximum number of input_ports
+ * \param max_output_ports maximum number of output_ports
+ * \return a newly allocated \ref pw_client_node0_transport
+ * \memberof pw_client_node0_transport
+ */
+struct pw_client_node0_transport *
+pw_client_node0_transport_new(struct pw_context *context,
+		uint32_t max_input_ports, uint32_t max_output_ports)
+{
+	struct transport *impl;
+	struct pw_client_node0_transport *trans;
+	struct pw_client_node0_area area = { 0 };
+
+	area.max_input_ports = max_input_ports;
+	area.n_input_ports = 0;
+	area.max_output_ports = max_output_ports;
+	area.n_output_ports = 0;
+
+	impl = calloc(1, sizeof(struct transport));
+	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;
+
+	impl->mem = pw_mempool_alloc(context->pool,
+			PW_MEMBLOCK_FLAG_READWRITE |
+			PW_MEMBLOCK_FLAG_MAP |
+			PW_MEMBLOCK_FLAG_SEAL,
+			SPA_DATA_MemFd, area_get_size(&area));
+	if (impl->mem == NULL) {
+		free(impl);
+		return NULL;
+	}
+
+	memcpy(impl->mem->map->ptr, &area, sizeof(struct pw_client_node0_area));
+	transport_setup_area(impl->mem->map->ptr, trans);
+	transport_reset_area(trans);
+
+	trans->destroy = destroy;
+	trans->add_message = add_message;
+	trans->next_message = next_message;
+	trans->parse_message = parse_message;
+
+	return trans;
+}
+
+struct pw_client_node0_transport *
+pw_client_node0_transport_new_from_info(struct pw_client_node0_transport_info *info)
+{
+	errno = ENOTSUP;
+	return NULL;
+}
+
+/** Get transport info
+ * \param trans the transport to get info of
+ * \param[out] info transport info
+ * \return 0 on success
+ *
+ * Fill \a info with the transport info of \a trans. This information can be
+ * passed to the client to set up the shared transport.
+ *
+ * \memberof pw_client_node0_transport
+ */
+int pw_client_node0_transport_get_info(struct pw_client_node0_transport *trans,
+				      struct pw_client_node0_transport_info *info)
+{
+	struct transport *impl = (struct transport *) trans;
+
+	info->memfd = impl->mem->fd;
+	info->offset = impl->offset;
+	info->size = impl->mem->size;
+
+	return 0;
+}
diff --git a/src/modules/module-client-node/v0/transport.h b/src/modules/module-client-node/v0/transport.h
new file mode 100644
index 0000000..3abcd6d
--- /dev/null
+++ b/src/modules/module-client-node/v0/transport.h
@@ -0,0 +1,59 @@
+/* PipeWire
+ *
+ * Copyright © 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __PIPEWIRE_CLIENT_NODE0_TRANSPORT_H__
+#define __PIPEWIRE_CLIENT_NODE0_TRANSPORT_H__
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <string.h>
+
+#include <spa/utils/defs.h>
+
+#include <pipewire/mem.h>
+
+/** information about the transport region \memberof pw_client_node */
+struct pw_client_node0_transport_info {
+	int memfd;		/**< the memfd of the transport area */
+	uint32_t offset;	/**< offset to map \a memfd at */
+	uint32_t size;		/**< size of memfd mapping */
+};
+
+struct pw_client_node0_transport *
+pw_client_node0_transport_new(struct pw_context *context, uint32_t max_input_ports, uint32_t max_output_ports);
+
+struct pw_client_node0_transport *
+pw_client_node0_transport_new_from_info(struct pw_client_node0_transport_info *info);
+
+int
+pw_client_node0_transport_get_info(struct pw_client_node0_transport *trans,
+				  struct pw_client_node0_transport_info *info);
+
+#ifdef __cplusplus
+} /* extern "C" */
+#endif
+
+#endif /* __PIPEWIRE_CLIENT_NODE0_TRANSPORT_H__ */