Work on cleanup

Add signal handlers for daemon to shut down
Add destroy listeners for modules and do cleanup
Fix some leaks
Simplify port registration in new nodes
Hide some more data structures
Let the node implementation take care of the reuse_buffer signal because
then we can get to the graph nodes to find the destination port.
Destroy modules in core cleanup. Modules should undo everything they
have done.
Activate the link after we negotiated format and buffers.

spa/plugins/support/loop.c

@@ -563,8 +563,10 @@ static void source_signal_func(struct spa_source *source)
 {
 	struct source_impl *impl = SPA_CONTAINER_OF(source, struct source_impl, source);
 	struct signalfd_siginfo signal_info;
+	int len;
 
-	if (read(source->fd, &signal_info, sizeof(signal_info)) != sizeof(signal_info))
+	len = read(source->fd, &signal_info, sizeof signal_info);
+	if (!(len == -1 && errno == EAGAIN) && len != sizeof signal_info)
 		spa_log_warn(impl->impl->log, NAME " %p: failed to read signal fd %d: %s",
 				source, source->fd, strerror(errno));
 
@@ -586,10 +588,12 @@ static struct spa_source *loop_add_signal(struct spa_loop_utils *utils,
 	source->source.loop = &impl->loop;
 	source->source.func = source_signal_func;
 	source->source.data = data;
+
 	sigemptyset(&mask);
 	sigaddset(&mask, signal_number);
 	source->source.fd = signalfd(-1, &mask, SFD_CLOEXEC | SFD_NONBLOCK);
 	sigprocmask(SIG_BLOCK, &mask, NULL);
+
 	source->source.mask = SPA_IO_IN;
 	source->impl = impl;
 	source->close = true;

src/daemon/main.c

@@ -17,12 +17,20 @@
  * Boston, MA 02110-1301, USA.
  */
 
+#include <signal.h>
+
 #include <pipewire/pipewire.h>
 #include <pipewire/core.h>
 #include <pipewire/module.h>
 
 #include "daemon-config.h"
 
+static void do_quit(void *data, int signal_number)
+{
+	struct pw_main_loop *loop = data;
+	pw_main_loop_quit(loop);
+}
+
 int main(int argc, char *argv[])
 {
 	struct pw_core *core;
@@ -45,16 +53,20 @@ int main(int argc, char *argv[])
 				  "pipewire.daemon", "1", NULL);
 
 	loop = pw_main_loop_new(props);
+	pw_loop_add_signal(pw_main_loop_get_loop(loop), SIGINT, do_quit, loop);
+	pw_loop_add_signal(pw_main_loop_get_loop(loop), SIGTERM, do_quit, loop);
 
 	core = pw_core_new(pw_main_loop_get_loop(loop), props);
 
 	pw_daemon_config_run_commands(config, core);
 
+	pw_log_info("start main loop");
 	pw_main_loop_run(loop);
+	pw_log_info("leave main loop");
 
-	pw_main_loop_destroy(loop);
+	pw_daemon_config_free(config);
-
 	pw_core_destroy(core);
+	pw_main_loop_destroy(loop);
 
 	return 0;
 }

src/modules/module-autolink.c

@@ -36,6 +36,7 @@ struct impl {
 	struct pw_properties *properties;
 
 	struct spa_hook core_listener;
+	struct spa_hook module_listener;
 
 	struct spa_list node_list;
 };
@@ -71,8 +72,11 @@ static struct node_info *find_node_info(struct impl *impl, struct pw_node *node)
 
 static void link_data_remove(struct link_data *data)
 {
+	if (data->node_info) {
 		spa_list_remove(&data->l);
 		spa_hook_remove(&data->link_listener);
+		data->node_info = NULL;
+	}
 }
 
 static void node_info_free(struct node_info *info)
@@ -290,6 +294,27 @@ core_global_removed(void *data, struct pw_global *global)
 	}
 }
 
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct node_info *info, *t;
+
+	spa_list_for_each_safe(info, t, &impl->node_list, l)
+		node_info_free(info);
+
+	spa_hook_remove(&impl->core_listener);
+	spa_hook_remove(&impl->module_listener);
+
+	if (impl->properties)
+		pw_properties_free(impl->properties);
+
+	free(impl);
+}
+
+const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+        .destroy = module_destroy,
+};
 
 const struct pw_core_events core_events = {
 	PW_VERSION_CORE_EVENTS,
@@ -322,20 +347,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 	spa_list_init(&impl->node_list);
 
 	pw_core_add_listener(core, &impl->core_listener, &core_events, impl);
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
 
 	return impl;
 }
 
-#if 0
-static void module_destroy(struct impl *impl)
-{
-	pw_log_debug("module %p: destroy", impl);
-
-	spa_hook_remove(&impl->core_listener);
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

src/modules/module-client-node.c

@@ -36,6 +36,8 @@ struct pw_protocol *pw_protocol_native_ext_client_node_init(struct pw_core *core
 struct factory_data {
 	struct pw_node_factory *this;
 	struct pw_properties *properties;
+
+	struct spa_hook module_listener;
 };
 
 static struct pw_node *create_node(void *_data,
@@ -64,6 +66,23 @@ static const struct pw_node_factory_implementation impl_factory = {
 	.create_node = create_node,
 };
 
+static void module_destroy(void *data)
+{
+	struct factory_data *d = data;
+
+	spa_hook_remove(&d->module_listener);
+
+	if (d->properties)
+		pw_properties_free(d->properties);
+
+	pw_node_factory_destroy(d->this);
+}
+
+const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 static bool module_init(struct pw_module *module, struct pw_properties *properties)
 {
 	struct pw_core *core = pw_module_get_core(module);
@@ -88,18 +107,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 
 	pw_node_factory_export(factory, NULL, pw_module_get_global(module));
 
+	pw_module_add_listener(module, &data->module_listener, &module_events, data);
+
 	return true;
 }
 
-#if 0
-static void module_destroy(struct impl *impl)
-{
-	pw_log_debug("module %p: destroy", impl);
-
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

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

@@ -1165,7 +1165,7 @@ struct pw_client_node *pw_client_node_new(struct pw_resource *resource,
 				     true,
 				     &impl->proxy.node,
 				     NULL,
-				     properties);
+				     properties, 0);
 	if (this->node == NULL)
 		goto error_no_node;
 

src/modules/module-flatpak.c

@@ -44,6 +44,7 @@ struct impl {
 	DBusConnection *bus;
 
 	struct spa_hook core_listener;
+	struct spa_hook module_listener;
 
 	struct spa_list client_list;
 
@@ -711,6 +712,33 @@ static void wakeup_main(void *userdata)
 	pw_loop_enable_idle(pw_core_get_main_loop(impl->core), impl->dispatch_event, true);
 }
 
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct client_info *info, *t;
+
+	spa_hook_remove(&impl->core_listener);
+	spa_hook_remove(&impl->module_listener);
+
+	dbus_connection_close(impl->bus);
+	dbus_connection_unref(impl->bus);
+
+	spa_list_for_each_safe(info, t, &impl->client_list, link)
+		client_info_free(info);
+
+	pw_loop_destroy_source(pw_core_get_main_loop(impl->core), impl->dispatch_event);
+
+	if (impl->properties)
+		pw_properties_free(impl->properties);
+
+	free(impl);
+}
+
+const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 static bool module_init(struct pw_module *module, struct pw_properties *properties)
 {
 	struct pw_core *core = pw_module_get_core(module);
@@ -743,6 +771,7 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 	spa_list_init(&impl->client_list);
 
 	pw_core_add_listener(core, &impl->core_listener, &core_events, impl);
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
 
 	pw_core_set_permission_callback(core, do_permission, impl);
 

src/modules/module-jack.c

@@ -73,7 +73,7 @@ struct impl {
 	struct pw_core *core;
 	struct pw_type *t;
 	struct pw_module *module;
-	struct spa_list link;
+	struct spa_hook module_listener;
 
         struct spa_source *timer;
 
@@ -1434,9 +1434,6 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous)
 	for (i = 0; i < CLIENT_NUM; i++)
 		server->synchro_table[i] = JACK_SYNCHRO_INIT;
 
-	if (!init_nodes(impl))
-		return -1;
-
 	s = create_socket();
 
 	if (!init_socket_name(&s->addr, name, promiscuous, 0))
@@ -1445,6 +1442,9 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous)
 	if (!add_socket(impl, s))
 		goto error;
 
+	if (!init_nodes(impl))
+		goto error;
+
 	return 0;
 
       error:
@@ -1452,6 +1452,26 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous)
 	return -1;
 }
 
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct link *ld, *t;
+
+	spa_hook_remove(&impl->module_listener);
+
+	spa_list_for_each_safe(ld, t, &impl->link_list, link_link)
+		pw_link_destroy(ld->link);
+
+	if (impl->properties)
+		pw_properties_free(impl->properties);
+
+	free(impl);
+}
+
+static const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
 
 static bool module_init(struct pw_module *module, struct pw_properties *properties)
 {
@@ -1492,6 +1512,8 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 	if (init_server(impl, name, promiscuous) < 0)
 		goto error;
 
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
 	return true;
 
       error:
@@ -1499,17 +1521,6 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 	return false;
 }
 
-#if 0
-static void module_destroy(struct impl *impl)
-{
-	struct impl *object, *tmp;
-
-	pw_log_debug("module %p: destroy", impl);
-
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

src/modules/module-mixer.c

@@ -37,6 +37,7 @@ struct impl {
 	struct pw_core *core;
 	struct pw_type *t;
 	struct pw_module *module;
+	struct spa_hook module_listener;
 	struct pw_properties *properties;
 
 	void *hnd;
@@ -119,7 +120,7 @@ static struct pw_node *make_node(struct impl *impl)
 	spa_clock = iface;
 
 	node = pw_spa_node_new(impl->core, NULL, pw_module_get_global(impl->module),
-			       "audiomixer", false, spa_node, spa_clock, NULL);
+			       "audiomixer", false, spa_node, spa_clock, NULL, 0);
 
 	return node;
 
@@ -166,6 +167,23 @@ static bool on_global(void *data, struct pw_global *global)
 	return true;
 }
 
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+
+	spa_hook_remove(&impl->module_listener);
+
+	if (impl->properties)
+		pw_properties_free(impl->properties);
+
+	free(impl);
+}
+
+static const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 static bool module_init(struct pw_module *module, struct pw_properties *properties)
 {
 	struct pw_core *core = pw_module_get_core(module);
@@ -183,18 +201,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 
 	pw_core_for_each_global(core, on_global, impl);
 
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
 	return true;
 }
 
-#if 0
-static void module_destroy(struct impl *impl)
-{
-	pw_log_debug("module %p: destroy", impl);
-
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

src/modules/module-protocol-native.c

@@ -57,10 +57,13 @@ void pw_protocol_native_init(struct pw_protocol *protocol);
 
 struct protocol_data {
 	struct pw_module *module;
+	struct spa_hook module_listener;
+	struct pw_protocol *protocol;
+	struct pw_properties *properties;
 };
 
-struct connection {
+struct client {
-	struct pw_protocol_connection this;
+	struct pw_protocol_client this;
 
 	int fd;
 
@@ -74,8 +77,8 @@ struct connection {
         struct spa_source *flush_event;
 };
 
-struct listener {
+struct server {
-	struct pw_protocol_listener this;
+	struct pw_protocol_server this;
 
 	int fd;
 	int fd_lock;
@@ -223,11 +226,11 @@ static const struct pw_client_events client_events = {
 	.busy_changed = client_busy_changed,
 };
 
-static struct pw_client *client_new(struct listener *l, int fd)
+static struct pw_client *client_new(struct server *s, int fd)
 {
 	struct client_data *this;
 	struct pw_client *client;
-	struct pw_protocol *protocol = l->this.protocol;
+	struct pw_protocol *protocol = s->this.protocol;
 	struct protocol_data *pd = protocol->user_data;
 	socklen_t len;
 	struct ucred ucred, *ucredp;
@@ -263,7 +266,7 @@ static struct pw_client *client_new(struct listener *l, int fd)
 		goto no_connection;
 
 	client->protocol = protocol;
-	spa_list_insert(l->this.client_list.prev, &client->protocol_link);
+	spa_list_append(&s->this.client_list, &client->protocol_link);
 
 	pw_client_add_listener(client, &this->client_listener, &client_events, this);
 
@@ -279,7 +282,7 @@ static struct pw_client *client_new(struct listener *l, int fd)
 	return NULL;
 }
 
-static bool init_socket_name(struct listener *l, const char *name)
+static bool init_socket_name(struct server *s, const char *name)
 {
 	int name_size;
 	const char *runtime_dir;
@@ -289,62 +292,62 @@ static bool init_socket_name(struct listener *l, const char *name)
 		return false;
 	}
 
-	l->addr.sun_family = AF_LOCAL;
+	s->addr.sun_family = AF_LOCAL;
-	name_size = snprintf(l->addr.sun_path, sizeof(l->addr.sun_path),
+	name_size = snprintf(s->addr.sun_path, sizeof(s->addr.sun_path),
 			     "%s/%s", runtime_dir, name) + 1;
 
-	if (name_size > (int) sizeof(l->addr.sun_path)) {
+	if (name_size > (int) sizeof(s->addr.sun_path)) {
 		pw_log_error("socket path \"%s/%s\" plus null terminator exceeds 108 bytes",
 			     runtime_dir, name);
-		*l->addr.sun_path = 0;
+		*s->addr.sun_path = 0;
 		return false;
 	}
 	return true;
 }
 
-static bool lock_socket(struct listener *l)
+static bool lock_socket(struct server *s)
 {
 	struct stat socket_stat;
 
-	snprintf(l->lock_addr, sizeof(l->lock_addr), "%s%s", l->addr.sun_path, LOCK_SUFFIX);
+	snprintf(s->lock_addr, sizeof(s->lock_addr), "%s%s", s->addr.sun_path, LOCK_SUFFIX);
 
-	l->fd_lock = open(l->lock_addr, O_CREAT | O_CLOEXEC,
+	s->fd_lock = open(s->lock_addr, O_CREAT | O_CLOEXEC,
 			  (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP));
 
-	if (l->fd_lock < 0) {
+	if (s->fd_lock < 0) {
-		pw_log_error("unable to open lockfile %s check permissions", l->lock_addr);
+		pw_log_error("unable to open lockfile %s check permissions", s->lock_addr);
 		goto err;
 	}
 
-	if (flock(l->fd_lock, LOCK_EX | LOCK_NB) < 0) {
+	if (flock(s->fd_lock, LOCK_EX | LOCK_NB) < 0) {
 		pw_log_error("unable to lock lockfile %s, maybe another daemon is running",
-			     l->lock_addr);
+			     s->lock_addr);
 		goto err_fd;
 	}
 
-	if (stat(l->addr.sun_path, &socket_stat) < 0) {
+	if (stat(s->addr.sun_path, &socket_stat) < 0) {
 		if (errno != ENOENT) {
-			pw_log_error("did not manage to stat file %s\n", l->addr.sun_path);
+			pw_log_error("did not manage to stat file %s\n", s->addr.sun_path);
 			goto err_fd;
 		}
 	} else if (socket_stat.st_mode & S_IWUSR || socket_stat.st_mode & S_IWGRP) {
-		unlink(l->addr.sun_path);
+		unlink(s->addr.sun_path);
 	}
 	return true;
 
       err_fd:
-	close(l->fd_lock);
+	close(s->fd_lock);
-	l->fd_lock = -1;
+	s->fd_lock = -1;
       err:
-	*l->lock_addr = 0;
+	*s->lock_addr = 0;
-	*l->addr.sun_path = 0;
+	*s->addr.sun_path = 0;
 	return false;
 }
 
 static void
 socket_data(void *data, int fd, enum spa_io mask)
 {
-	struct listener *l = data;
+	struct server *s = data;
 	struct pw_client *client;
 	struct client_data *c;
 	struct sockaddr_un name;
@@ -358,7 +361,7 @@ socket_data(void *data, int fd, enum spa_io mask)
 		return;
 	}
 
-	client = client_new(l, client_fd);
+	client = client_new(s, client_fd);
 	if (client == NULL) {
 		pw_log_error("failed to create client");
 		close(client_fd);
@@ -370,27 +373,27 @@ socket_data(void *data, int fd, enum spa_io mask)
 			  c->source, SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP);
 }
 
-static bool add_socket(struct pw_protocol *protocol, struct listener *l)
+static bool add_socket(struct pw_protocol *protocol, struct server *s)
 {
 	socklen_t size;
 
-	if ((l->fd = socket(PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0)
+	if ((s->fd = socket(PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0)
 		return false;
 
-	size = offsetof(struct sockaddr_un, sun_path) + strlen(l->addr.sun_path);
+	size = offsetof(struct sockaddr_un, sun_path) + strlen(s->addr.sun_path);
-	if (bind(l->fd, (struct sockaddr *) &l->addr, size) < 0) {
+	if (bind(s->fd, (struct sockaddr *) &s->addr, size) < 0) {
 		pw_log_error("bind() failed with error: %m");
 		return false;
 	}
 
-	if (listen(l->fd, 128) < 0) {
+	if (listen(s->fd, 128) < 0) {
 		pw_log_error("listen() failed with error: %m");
 		return false;
 	}
 
-	l->loop = pw_core_get_main_loop(protocol->core);
+	s->loop = pw_core_get_main_loop(protocol->core);
-	l->source = pw_loop_add_io(l->loop, l->fd, SPA_IO_IN, false, socket_data, l);
+	s->source = pw_loop_add_io(s->loop, s->fd, SPA_IO_IN, false, socket_data, s);
-	if (l->source == NULL)
+	if (s->source == NULL)
 		return false;
 
 	return true;
@@ -410,7 +413,7 @@ get_name(const struct pw_properties *properties)
 	return name;
 }
 
-static int impl_connect(struct pw_protocol_connection *conn)
+static int impl_connect(struct pw_protocol_client *conn)
 {
         struct sockaddr_un addr;
         socklen_t size;
@@ -454,7 +457,7 @@ static int impl_connect(struct pw_protocol_connection *conn)
 static void
 on_remote_data(void *data, int fd, enum spa_io mask)
 {
-        struct connection *impl = data;
+        struct client *impl = data;
         struct pw_remote *this = impl->this.remote;
         struct pw_protocol_native_connection *conn = impl->connection;
 	struct pw_core *core = pw_remote_get_core(this);
@@ -522,7 +525,7 @@ on_remote_data(void *data, int fd, enum spa_io mask)
 
 static void do_flush_event(void *data, uint64_t count)
 {
-        struct connection *impl = data;
+        struct client *impl = data;
 	impl->flush_signaled = false;
         if (impl->connection)
                 if (!pw_protocol_native_connection_flush(impl->connection))
@@ -531,7 +534,7 @@ static void do_flush_event(void *data, uint64_t count)
 
 static void on_need_flush(void *data)
 {
-        struct connection *impl = data;
+        struct client *impl = data;
         struct pw_remote *remote = impl->this.remote;
 
 	if (!impl->flush_signaled) {
@@ -545,9 +548,9 @@ static const struct pw_protocol_native_connection_events conn_events = {
 	.need_flush = on_need_flush,
 };
 
-static int impl_connect_fd(struct pw_protocol_connection *conn, int fd)
+static int impl_connect_fd(struct pw_protocol_client *conn, int fd)
 {
-        struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this);
+        struct client *impl = SPA_CONTAINER_OF(conn, struct client, this);
 	struct pw_remote *remote = conn->remote;
 
         impl->connection = pw_protocol_native_connection_new(fd);
@@ -572,9 +575,9 @@ static int impl_connect_fd(struct pw_protocol_connection *conn, int fd)
         return -1;
 }
 
-static void impl_disconnect(struct pw_protocol_connection *conn)
+static void impl_disconnect(struct pw_protocol_client *conn)
 {
-        struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this);
+        struct client *impl = SPA_CONTAINER_OF(conn, struct client, this);
 	struct pw_remote *remote = conn->remote;
 
         impl->disconnecting = true;
@@ -592,9 +595,9 @@ static void impl_disconnect(struct pw_protocol_connection *conn)
         impl->fd = -1;
 }
 
-static void impl_destroy(struct pw_protocol_connection *conn)
+static void impl_destroy(struct pw_protocol_client *conn)
 {
-        struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this);
+        struct client *impl = SPA_CONTAINER_OF(conn, struct client, this);
 	struct pw_remote *remote = conn->remote;
 
         pw_loop_destroy_source(remote->core->main_loop, impl->flush_event);
@@ -603,15 +606,15 @@ static void impl_destroy(struct pw_protocol_connection *conn)
 	free(impl);
 }
 
-static struct pw_protocol_connection *
+static struct pw_protocol_client *
-impl_new_connection(struct pw_protocol *protocol,
+impl_new_client(struct pw_protocol *protocol,
 		struct pw_remote *remote,
 		struct pw_properties *properties)
 {
-	struct connection *impl;
+	struct client *impl;
-	struct pw_protocol_connection *this;
+	struct pw_protocol_client *this;
 
-	if ((impl = calloc(1, sizeof(struct connection))) == NULL)
+	if ((impl = calloc(1, sizeof(struct client))) == NULL)
 		return NULL;
 
 	this = &impl->this;
@@ -625,32 +628,34 @@ impl_new_connection(struct pw_protocol *protocol,
 
         impl->flush_event = pw_loop_add_event(remote->core->main_loop, do_flush_event, impl);
 
-	spa_list_insert(protocol->connection_list.prev, &this->link);
+	spa_list_append(&protocol->client_list, &this->link);
 
 	return this;
 }
 
-static void destroy_listener(struct pw_protocol_listener *listener)
+static void destroy_server(struct pw_protocol_server *server)
 {
-	struct listener *l = SPA_CONTAINER_OF(listener, struct listener, this);
+	struct server *s = SPA_CONTAINER_OF(server, struct server, this);
 
-	if (l->source)
+	spa_list_remove(&server->link);
-		pw_loop_destroy_source(l->loop, l->source);
+
-	if (l->addr.sun_path[0])
+	if (s->source)
-		unlink(l->addr.sun_path);
+		pw_loop_destroy_source(s->loop, s->source);
-	if (l->fd >= 0)
+	if (s->addr.sun_path[0])
-		close(l->fd);
+		unlink(s->addr.sun_path);
-	if (l->lock_addr[0])
+	if (s->fd >= 0)
-		unlink(l->lock_addr);
+		close(s->fd);
-	if (l->fd_lock >= 0)
+	if (s->lock_addr[0])
-		close(l->fd_lock);
+		unlink(s->lock_addr);
-	free(l);
+	if (s->fd_lock >= 0)
+		close(s->fd_lock);
+	free(s);
 }
 
 static void on_before_hook(void *_data)
 {
-	struct listener *listener = _data;
+	struct server *server = _data;
-	struct pw_protocol_listener *this = &listener->this;
+	struct pw_protocol_server *this = &server->this;
 	struct pw_client *client, *tmp;
 	struct client_data *data;
 
@@ -665,79 +670,79 @@ static const struct spa_loop_control_hooks impl_hooks = {
 	.before = on_before_hook,
 };
 
-static struct pw_protocol_listener *
+static struct pw_protocol_server *
-impl_add_listener(struct pw_protocol *protocol,
+impl_add_server(struct pw_protocol *protocol,
                 struct pw_core *core,
                 struct pw_properties *properties)
 {
-	struct pw_protocol_listener *this;
+	struct pw_protocol_server *this;
-	struct listener *l;
+	struct server *s;
 	const char *name;
 
-	if ((l = calloc(1, sizeof(struct listener))) == NULL)
+	if ((s = calloc(1, sizeof(struct server))) == NULL)
 		return NULL;
 
-	l->fd = -1;
+	s->fd = -1;
-	l->fd_lock = -1;
+	s->fd_lock = -1;
 
-	this = &l->this;
+	this = &s->this;
 	this->protocol = protocol;
 	spa_list_init(&this->client_list);
-	this->destroy = destroy_listener;
+	this->destroy = destroy_server;
 
 	name = get_name(pw_core_get_properties(core));
 
-	if (!init_socket_name(l, name))
+	if (!init_socket_name(s, name))
 		goto error;
 
-	if (!lock_socket(l))
+	if (!lock_socket(s))
 		goto error;
 
-	if (!add_socket(protocol, l))
+	if (!add_socket(protocol, s))
 		goto error;
 
-	spa_list_insert(protocol->listener_list.prev, &this->link);
+	spa_list_append(&protocol->server_list, &this->link);
 
-	pw_loop_add_hook(pw_core_get_main_loop(core), &l->hook, &impl_hooks, l);
+	pw_loop_add_hook(pw_core_get_main_loop(core), &s->hook, &impl_hooks, s);
 
-	pw_log_info("protocol-native %p: Added listener %p", protocol, this);
+	pw_log_info("protocol-native %p: Added server %p", protocol, this);
 
 	return this;
 
       error:
-	destroy_listener(this);
+	destroy_server(this);
 	return NULL;
 }
 
 const static struct pw_protocol_implementaton protocol_impl = {
 	PW_VERSION_PROTOCOL_IMPLEMENTATION,
-	impl_new_connection,
+	.new_client = impl_new_client,
-	impl_add_listener,
+	.add_server = impl_add_server,
 };
 
 static struct spa_pod_builder *
 impl_ext_begin_proxy(struct pw_proxy *proxy, uint8_t opcode)
 {
-	struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this);
+	struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this);
 	return pw_protocol_native_connection_begin_proxy(impl->connection, proxy, opcode);
 }
 
 static uint32_t impl_ext_add_proxy_fd(struct pw_proxy *proxy, int fd)
 {
-	struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this);
+	struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this);
 	return pw_protocol_native_connection_add_fd(impl->connection, fd);
 }
 
 static int impl_ext_get_proxy_fd(struct pw_proxy *proxy, uint32_t index)
 {
-	struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this);
+	struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this);
 	return pw_protocol_native_connection_get_fd(impl->connection, index);
 }
 
 static void impl_ext_end_proxy(struct pw_proxy *proxy,
 			       struct spa_pod_builder *builder)
 {
-	struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this);
+	struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this);
 	pw_protocol_native_connection_end(impl->connection, builder);
 }
 
@@ -778,6 +783,23 @@ const static struct pw_protocol_native_ext protocol_ext_impl = {
 	impl_ext_end_resource,
 };
 
+static void module_destroy(void *data)
+{
+	struct protocol_data *d = data;
+
+	spa_hook_remove(&d->module_listener);
+
+	if (d->properties)
+		pw_properties_free(d->properties);
+
+	pw_protocol_destroy(d->protocol);
+}
+
+static const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 static bool module_init(struct pw_module *module, struct pw_properties *properties)
 {
 	struct pw_core *core = pw_module_get_core(module);
@@ -800,29 +822,20 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 	pw_log_debug("protocol-native %p: new", this);
 
 	d = pw_protocol_get_user_data(this);
+	d->protocol = this;
 	d->module = module;
+	d->properties = properties;
 
 	if ((val = pw_properties_get(pw_core_get_properties(core), "pipewire.daemon"))) {
 		if (atoi(val) == 1)
-			impl_add_listener(this, core, properties);
+			impl_add_server(this, core, properties);
 	}
+
+	pw_module_add_listener(module, &d->module_listener, &module_events, d);
+
 	return true;
 }
 
-#if 0
-static void pw_protocol_native_destroy(struct impl *impl)
-{
-	struct impl *object, *tmp;
-
-	pw_log_debug("protocol-native %p: destroy", impl);
-
-	spa_list_for_each_safe(object, tmp, &impl->object_list, link)
-	    object_destroy(object);
-
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

src/modules/module-suspend-on-idle.c

@@ -33,6 +33,7 @@ struct impl {
 	struct pw_type *t;
 	struct pw_properties *properties;
 
+	struct spa_hook module_listener;
 	struct spa_hook core_listener;
 
 	struct spa_list node_list;
@@ -127,7 +128,7 @@ core_global_added(void *data, struct pw_global *global)
 		info = calloc(1, sizeof(struct node_info));
 		info->impl = impl;
 		info->node = node;
-		spa_list_insert(impl->node_list.prev, &info->link);
+		spa_list_append(&impl->node_list, &info->link);
 
 		pw_node_add_listener(node, &info->node_listener, &node_events, info);
 
@@ -151,6 +152,28 @@ core_global_removed(void *data, struct pw_global *global)
 	}
 }
 
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct node_info *info, *t;
+
+	spa_list_for_each_safe(info, t, &impl->node_list, link)
+		node_info_free(info);
+
+	spa_hook_remove(&impl->core_listener);
+	spa_hook_remove(&impl->module_listener);
+
+	if (impl->properties)
+		pw_properties_free(impl->properties);
+
+	free(impl);
+}
+
+const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 const struct pw_core_events core_events = {
 	PW_VERSION_CORE_EVENTS,
 	.global_added = core_global_added,
@@ -179,22 +202,12 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti
 
 	spa_list_init(&impl->node_list);
 
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
 	pw_core_add_listener(impl->core, &impl->core_listener, &core_events, impl);
 
 	return impl;
 }
 
-#if 0
-static void module_destroy(struct impl *impl)
-{
-	pw_log_debug("module %p: destroy", impl);
-
-	pw_global_destroy(impl->global);
-
-	free(impl);
-}
-#endif
-
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	return module_init(module, NULL);

src/modules/spa/module-monitor.c

@@ -34,11 +34,32 @@
 
 #include "spa-monitor.h"
 
+struct data {
+	struct pw_spa_monitor *monitor;
+	struct spa_hook module_listener;
+};
+
+static void module_destroy(void *data)
+{
+	struct data *d = data;
+
+	spa_hook_remove(&d->module_listener);
+
+	pw_spa_monitor_destroy(d->monitor);
+}
+
+const struct pw_module_events module_events = {
+	PW_VERSION_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
 bool pipewire__module_init(struct pw_module *module, const char *args)
 {
 	const char *dir;
 	char **argv;
 	int n_tokens;
+	struct pw_spa_monitor *monitor;
+	struct data *data;
 
 	if (args == NULL)
 		goto wrong_arguments;
@@ -50,12 +71,18 @@ bool pipewire__module_init(struct pw_module *module, const char *args)
 	if ((dir = getenv("SPA_PLUGIN_DIR")) == NULL)
 		dir = PLUGINDIR;
 
-	pw_spa_monitor_load(pw_module_get_core(module),
+	monitor = pw_spa_monitor_load(pw_module_get_core(module),
 				      pw_module_get_global(module),
-			    dir, argv[0], argv[1], argv[2]);
+				      dir, argv[0], argv[1], argv[2],
+				      sizeof(struct data));
+
+	data = monitor->user_data;
+	data->monitor = monitor;
 
 	pw_free_strv(argv);
 
+	pw_module_add_listener(module, &data->module_listener, &module_events, data);
+
 	return true;
 
       not_enough_arguments:

src/modules/spa/module-node-factory.c

@@ -61,7 +61,7 @@ static struct pw_node *create_node(void *_data,
 				lib,
 				factory_name,
 				name,
-				properties);
+				properties, 0);
 	if (node == NULL)
 		goto no_mem;
 

src/modules/spa/module-node.c

@@ -61,7 +61,11 @@ bool pipewire__module_init(struct pw_module *module, const char *args)
 		pw_free_strv(prop);
 	}
 
-	pw_spa_node_load(pw_module_get_core(module), NULL, pw_module_get_global(module), argv[0], argv[1], argv[2], props);
+	pw_spa_node_load(pw_module_get_core(module),
+			 NULL,
+			 pw_module_get_global(module),
+			 argv[0], argv[1], argv[2],
+			 props, 0);
 
 	pw_free_strv(argv);
 

src/modules/spa/spa-monitor.c

@@ -42,6 +42,7 @@ struct monitor_item {
 	char *id;
 	struct spa_list link;
 	struct pw_node *node;
+	struct spa_handle *handle;
 };
 
 struct impl {
@@ -120,10 +121,11 @@ static void add_item(struct pw_spa_monitor *this, struct spa_monitor_item *item)
 
 	mitem = calloc(1, sizeof(struct monitor_item));
 	mitem->id = strdup(id);
+	mitem->handle = handle;
 	mitem->node = pw_spa_node_new(impl->core, NULL, impl->parent, name,
-				      false, node_iface, clock_iface, props);
+				      false, node_iface, clock_iface, props, 0);
 
-	spa_list_insert(impl->item_list.prev, &mitem->link);
+	spa_list_append(&impl->item_list, &mitem->link);
 }
 
 static struct monitor_item *find_item(struct pw_spa_monitor *this, const char *id)
@@ -143,6 +145,8 @@ void destroy_item(struct monitor_item *mitem)
 {
 	pw_node_destroy(mitem->node);
 	spa_list_remove(&mitem->link);
+	spa_handle_clear(mitem->handle);
+	free(mitem->handle);
 	free(mitem->id);
 	free(mitem);
 }
@@ -222,7 +226,9 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core,
 					   struct pw_global *parent,
 					   const char *dir,
 					   const char *lib,
-					   const char *factory_name, const char *system_name)
+					   const char *factory_name,
+					   const char *system_name,
+					   size_t user_data_size)
 {
 	struct impl *impl;
 	struct pw_spa_monitor *this;
@@ -271,7 +277,7 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core,
 		goto interface_failed;
 	}
 
-	impl = calloc(1, sizeof(struct impl));
+	impl = calloc(1, sizeof(struct impl) + user_data_size);
 	impl->core = core;
 	impl->t = t;
 	impl->parent = parent;
@@ -284,6 +290,9 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core,
 	this->system_name = strdup(system_name);
 	this->handle = handle;
 
+        if (user_data_size > 0)
+		this->user_data = SPA_MEMBER(impl, sizeof(struct impl), void);
+
 	update_monitor(core, this->system_name);
 
 	spa_list_init(&impl->item_list);
@@ -321,7 +330,7 @@ void pw_spa_monitor_destroy(struct pw_spa_monitor *monitor)
 	struct impl *impl = SPA_CONTAINER_OF(monitor, struct impl, this);
 	struct monitor_item *mitem, *tmp;
 
-	pw_log_debug("spa-monitor %p: dispose", impl);
+	pw_log_debug("spa-monitor %p: destroy", impl);
 
 	spa_list_for_each_safe(mitem, tmp, &impl->item_list, link)
 		destroy_item(mitem);

src/modules/spa/spa-monitor.h

@@ -35,6 +35,8 @@ struct pw_spa_monitor {
 	char *factory_name;
 	char *system_name;
 	struct spa_handle *handle;
+
+	void *user_data;
 };
 
 struct pw_spa_monitor *
@@ -42,7 +44,9 @@ pw_spa_monitor_load(struct pw_core *core,
 		    struct pw_global *parent,
 		    const char *dir,
 		    const char *lib,
-		    const char *factory_name, const char *system_name);
+		    const char *factory_name,
+		    const char *system_name,
+		    size_t user_data_size);
 void
 pw_spa_monitor_destroy(struct pw_spa_monitor *monitor);
 

src/modules/spa/spa-node.c

@@ -26,8 +26,12 @@
 #include <dlfcn.h>
 
 #include <spa/node.h>
+#include <spa/graph.h>
 
 #include "spa-node.h"
+#include "pipewire/node.h"
+#include "pipewire/port.h"
+#include "pipewire/log.h"
 #include "pipewire/private.h"
 
 struct impl {
@@ -170,13 +174,11 @@ static void update_port_ids(struct impl *impl)
         uint32_t *input_port_ids, *output_port_ids;
         uint32_t n_input_ports, n_output_ports, max_input_ports, max_output_ports;
         uint32_t i;
-        struct spa_list *ports;
 
         spa_node_get_n_ports(impl->node,
                              &n_input_ports, &max_input_ports, &n_output_ports, &max_output_ports);
 
-        this->info.max_input_ports = max_input_ports;
+	pw_node_set_max_ports(this, max_input_ports, max_output_ports);
-        this->info.max_output_ports = max_output_ports;
 
         input_port_ids = alloca(sizeof(uint32_t) * n_input_ports);
         output_port_ids = alloca(sizeof(uint32_t) * n_output_ports);
@@ -187,59 +189,13 @@ static void update_port_ids(struct impl *impl)
         pw_log_debug("node %p: update_port ids %u/%u, %u/%u", this,
                      n_input_ports, max_input_ports, n_output_ports, max_output_ports);
 
-	i = 0;
+	for (i = 0; i < n_input_ports; i++) {
-        ports = &this->input_ports;
-        while (true) {
-                struct pw_port *p = (ports == &this->input_ports) ? NULL :
-                    SPA_CONTAINER_OF(ports, struct pw_port, link);
-
-                if (p && i < n_input_ports && p->port_id == input_port_ids[i]) {
-                        pw_log_debug("node %p: exiting input port %d", this, input_port_ids[i]);
-                        i++;
-                        ports = ports->next;
-                } else if ((p && i < n_input_ports && input_port_ids[i] < p->port_id)
-                           || i < n_input_ports) {
-                        struct pw_port *np;
 		pw_log_debug("node %p: input port added %d", this, input_port_ids[i]);
-                        np = make_port(impl, PW_DIRECTION_INPUT, input_port_ids[i]);
+		make_port(impl, PW_DIRECTION_INPUT, input_port_ids[i]);
-
-                        ports = np->link.next;
-                        i++;
-                } else if (p) {
-                        ports = ports->next;
-                        pw_log_debug("node %p: input port removed %d", this, p->port_id);
-                        pw_port_destroy(p);
-                } else {
-                        pw_log_debug("node %p: no more input ports", this);
-                        break;
 	}
-        }
+	for (i = 0; i < n_output_ports; i++) {
-
-        i = 0;
-        ports = &this->output_ports;
-        while (true) {
-                struct pw_port *p = (ports == &this->output_ports) ? NULL :
-                    SPA_CONTAINER_OF(ports, struct pw_port, link);
-
-                if (p && i < n_output_ports && p->port_id == output_port_ids[i]) {
-                        pw_log_debug("node %p: exiting output port %d", this, output_port_ids[i]);
-                        i++;
-                        ports = ports->next;
-                } else if ((p && i < n_output_ports && output_port_ids[i] < p->port_id)
-                           || i < n_output_ports) {
-                        struct pw_port *np;
 		pw_log_debug("node %p: output port added %d", this, output_port_ids[i]);
-                        np = make_port(impl, PW_DIRECTION_OUTPUT, output_port_ids[i]);
+		make_port(impl, PW_DIRECTION_OUTPUT, output_port_ids[i]);
-                        ports = np->link.next;
-                        i++;
-                } else if (p) {
-                        ports = ports->next;
-                        pw_log_debug("node %p: output port removed %d", this, p->port_id);
-                        pw_port_destroy(p);
-                } else {
-                        pw_log_debug("node %p: no more output ports", this);
-                        break;
-                }
 	}
 }
 
@@ -335,7 +291,7 @@ static void on_node_done(void *data, int seq, int res)
 	}
 
         pw_log_debug("spa-node %p: async complete event %d %d", this, seq, res);
-	spa_hook_list_call(&this->listener_list, struct pw_node_events, async_complete, seq, res);
+	spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, async_complete, seq, res);
 }
 
 static void on_node_event(void *data, struct spa_event *event)
@@ -343,21 +299,21 @@ static void on_node_event(void *data, struct spa_event *event)
         struct impl *impl = data;
         struct pw_node *this = impl->this;
 
-	spa_hook_list_call(&this->listener_list, struct pw_node_events, event, event);
+	spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, event, event);
 }
 
 static void on_node_need_input(void *data)
 {
         struct impl *impl = data;
         struct pw_node *this = impl->this;
-	spa_hook_list_call(&this->listener_list, struct pw_node_events, need_input);
+	spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, need_input);
 }
 
 static void on_node_have_output(void *data)
 {
         struct impl *impl = data;
         struct pw_node *this = impl->this;
-	spa_hook_list_call(&this->listener_list, struct pw_node_events, have_output);
+	spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, have_output);
 }
 
 static void
@@ -365,18 +321,8 @@ on_node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
 {
         struct impl *impl = data;
         struct pw_node *this = impl->this;
-        struct spa_graph_port *p, *pp;
+	spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events,
-
+			   reuse_buffer, port_id, buffer_id);
-        spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link) {
-		if (p->port_id != port_id)
-			continue;
-
-		pp = p->peer;
-		if (pp && pp->callbacks->reuse_buffer)
-			pp->callbacks->reuse_buffer(pp->callbacks_data, buffer_id);
-
-		break;
-	}
 }
 
 static const struct spa_node_callbacks spa_node_callbacks = {
@@ -401,7 +347,8 @@ pw_spa_node_new(struct pw_core *core,
 		bool async,
 		struct spa_node *node,
 		struct spa_clock *clock,
-		struct pw_properties *properties)
+		struct pw_properties *properties,
+		size_t user_data_size)
 {
 	struct pw_node *this;
 	struct impl *impl;
@@ -411,8 +358,7 @@ pw_spa_node_new(struct pw_core *core,
 
 		if (properties == NULL)
 			properties = pw_properties_new(NULL, NULL);
-
+		if (properties == NULL)
-		if (properties)
 			return NULL;
 
 		for (i = 0; i < node->info->n_items; i++)
@@ -421,7 +367,7 @@ pw_spa_node_new(struct pw_core *core,
 					  node->info->items[i].value);
 	}
 
-	this = pw_node_new(core, owner, parent, name, properties, sizeof(struct impl));
+	this = pw_node_new(core, owner, parent, name, properties, sizeof(struct impl) + user_data_size);
 	if (this == NULL)
 		return NULL;
 
@@ -439,9 +385,8 @@ pw_spa_node_new(struct pw_core *core,
 	if (spa_node_set_callbacks(impl->node, &spa_node_callbacks, impl) < 0)
 		pw_log_warn("spa-node %p: error setting callback", this);
 
-	if (!async) {
+	if (!async)
 		complete_init(impl);
-	}
 
 	return this;
 }
@@ -453,6 +398,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie
 	struct spa_props *props;
 	void *state = NULL;
 	const char *key;
+	struct pw_type *t = pw_core_get_type(core);
 
 	if ((res = spa_node_get_props(spa_node, &props)) != SPA_RESULT_OK) {
 		pw_log_debug("spa_node_get_props failed: %d", res);
@@ -466,7 +412,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie
 		if (!spa_type_is_a(key, SPA_TYPE_PROPS_BASE))
 			continue;
 
-		id = spa_type_map_get_id(core->type.map, key);
+		id = spa_type_map_get_id(t->map, key);
 		if (id == SPA_ID_INVALID)
 			continue;
 
@@ -478,7 +424,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie
 			switch(prop->body.value.type) {
 			case SPA_POD_TYPE_ID:
 				SPA_POD_VALUE(struct spa_pod_id, &prop->body.value) =
-					spa_type_map_get_id(core->type.map, value);
+					spa_type_map_get_id(t->map, value);
 				break;
 			case SPA_POD_TYPE_INT:
 				SPA_POD_VALUE(struct spa_pod_int, &prop->body.value) = atoi(value);
@@ -514,7 +460,8 @@ struct pw_node *pw_spa_node_load(struct pw_core *core,
 				 const char *lib,
 				 const char *factory_name,
 				 const char *name,
-				 struct pw_properties *properties)
+				 struct pw_properties *properties,
+				 size_t user_data_size)
 {
 	struct pw_node *this;
 	struct impl *impl;
@@ -530,6 +477,9 @@ struct pw_node *pw_spa_node_load(struct pw_core *core,
 	char *filename;
 	const char *dir;
 	bool async;
+	const struct spa_support *support;
+	uint32_t n_support;
+	struct pw_type *t = pw_core_get_type(core);
 
 	if ((dir = getenv("SPA_PLUGIN_DIR")) == NULL)
 		dir = PLUGINDIR;
@@ -555,21 +505,23 @@ struct pw_node *pw_spa_node_load(struct pw_core *core,
 			break;
 	}
 
+	support = pw_core_get_support(core, &n_support);
+
 	handle = calloc(1, factory->size);
 	if ((res = spa_handle_factory_init(factory,
-					   handle, NULL, core->support, core->n_support)) < 0) {
+					   handle, NULL, support, n_support)) < 0) {
 		pw_log_error("can't make factory instance: %d", res);
 		goto init_failed;
 	}
 	async = SPA_RESULT_IS_ASYNC(res);
 
-	if ((res = spa_handle_get_interface(handle, core->type.spa_node, &iface)) < 0) {
+	if ((res = spa_handle_get_interface(handle, t->spa_node, &iface)) < 0) {
 		pw_log_error("can't get node interface %d", res);
 		goto interface_failed;
 	}
 	spa_node = iface;
 
-	if ((res = spa_handle_get_interface(handle, core->type.spa_clock, &iface)) < 0) {
+	if ((res = spa_handle_get_interface(handle, t->spa_clock, &iface)) < 0) {
 		iface = NULL;
 	}
 	spa_clock = iface;
@@ -580,7 +532,10 @@ struct pw_node *pw_spa_node_load(struct pw_core *core,
 		}
 	}
 
-	this = pw_spa_node_new(core, owner, parent, name, async, spa_node, spa_clock, properties);
+	this = pw_spa_node_new(core, owner, parent, name, async,
+			       spa_node, spa_clock, properties, user_data_size);
+
+	impl = this->user_data;
 	impl->hnd = hnd;
 	impl->handle = handle;
 	impl->lib = filename;

src/modules/spa/spa-node.h

@@ -38,7 +38,8 @@ pw_spa_node_new(struct pw_core *core,
 		bool async,
 		struct spa_node *node,
 		struct spa_clock *clock,
-		struct pw_properties *properties);
+		struct pw_properties *properties,
+		size_t user_data_size);
 
 struct pw_node *
 pw_spa_node_load(struct pw_core *core,
@@ -47,7 +48,8 @@ pw_spa_node_load(struct pw_core *core,
 		 const char *lib,
 		 const char *factory_name,
 		 const char *name,
-		 struct pw_properties *properties);
+		 struct pw_properties *properties,
+		 size_t user_data_size);
 
 #ifdef __cplusplus
 }

src/pipewire/core.c

@@ -403,10 +403,14 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop, struct pw_properties *pro
 void pw_core_destroy(struct pw_core *core)
 {
 	struct pw_global *global, *t;
+	struct pw_module *module, *tm;
 
 	pw_log_debug("core %p: destroy", core);
 	spa_hook_list_call(&core->listener_list, struct pw_core_events, destroy);
 
+	spa_list_for_each_safe(module, tm, &core->module_list, link)
+		pw_module_destroy(module);
+
 	spa_list_for_each_safe(global, t, &core->global_list, link)
 		pw_global_destroy(global);
 

src/pipewire/global.h

@@ -89,8 +89,7 @@ pw_global_bind(struct pw_global *global,
 	       uint32_t version,
 	       uint32_t id);
 
-void
+void pw_global_destroy(struct pw_global *global);
-pw_global_destroy(struct pw_global *global);
 
 #ifdef __cplusplus
 }

src/pipewire/link.c

@@ -705,6 +705,15 @@ static int do_start(struct pw_link *this, uint32_t in_state, uint32_t out_state)
 	return res;
 }
 
+static int
+do_activate_link(struct spa_loop *loop,
+		 bool async, uint32_t seq, size_t size, const void *data, void *user_data)
+{
+        struct pw_link *this = user_data;
+	spa_graph_port_link(&this->rt.out_port, &this->rt.in_port);
+	return SPA_RESULT_OK;
+}
+
 static int check_states(struct pw_link *this, void *user_data, int res)
 {
 	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
@@ -726,6 +735,8 @@ static int check_states(struct pw_link *this, void *user_data, int res)
 	pw_log_debug("link %p: input state %d, output state %d", this, in_state, out_state);
 
 	if (in_state == PW_PORT_STATE_STREAMING && out_state == PW_PORT_STATE_STREAMING) {
+		pw_loop_invoke(this->output->node->data_loop,
+			       do_activate_link, SPA_ID_INVALID, 0, NULL, false, this);
 		pw_link_update_state(this, PW_LINK_STATE_RUNNING, NULL);
 		return SPA_RESULT_OK;
 	}
@@ -865,15 +876,6 @@ static void output_port_destroy(void *data)
 	on_port_destroy(&impl->this, impl->this.output);
 }
 
-static int
-do_activate_link(struct spa_loop *loop,
-		 bool async, uint32_t seq, size_t size, const void *data, void *user_data)
-{
-        struct pw_link *this = user_data;
-	spa_graph_port_link(&this->rt.out_port, &this->rt.in_port);
-	return SPA_RESULT_OK;
-}
-
 bool pw_link_activate(struct pw_link *this)
 {
 	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
@@ -884,9 +886,6 @@ bool pw_link_activate(struct pw_link *this)
 	impl->active = true;
 
 	pw_log_debug("link %p: activate", this);
-	pw_loop_invoke(this->output->node->data_loop,
-		       do_activate_link, SPA_ID_INVALID, 0, NULL, false, this);
-
 	this->output->node->n_used_output_links++;
 	this->input->node->n_used_input_links++;
 
@@ -1170,6 +1169,9 @@ void pw_link_destroy(struct pw_link *link)
 
 	pw_work_queue_destroy(impl->work);
 
+	if (link->properties)
+		pw_properties_free(link->properties);
+
 	if (link->info.format)
 		free(link->info.format);
 

src/pipewire/link.h

@@ -103,6 +103,7 @@ struct pw_port *pw_link_get_input(struct pw_link *link);
 /** Find the link between 2 ports \memberof pw_link */
 struct pw_link *pw_link_find(struct pw_port *output, struct pw_port *input);
 
+/** Indicate that linked nodes can be idle even when linked with this link */
 void pw_link_inc_idle(struct pw_link *link);
 
 /** Activate a link \memberof pw_link

src/pipewire/module.c

@@ -257,6 +257,7 @@ void pw_module_destroy(struct pw_module *module)
 	struct impl *impl = SPA_CONTAINER_OF(module, struct impl, this);
 	struct pw_resource *resource, *tmp;
 
+	pw_log_debug("module %p: destroy", module);
 	spa_hook_list_call(&module->listener_list, struct pw_module_events, destroy);
 
 	spa_list_for_each_safe(resource, tmp, &module->resource_list, link)

src/pipewire/node-factory.c

@@ -34,15 +34,31 @@ struct pw_node_factory *pw_node_factory_new(struct pw_core *core,
 	if (user_data_size > 0)
 		this->user_data = SPA_MEMBER(this, sizeof(*this), void);
 
+	pw_log_debug("node-factory %p: new", this);
+
 	return this;
 }
 
+void pw_node_factory_destroy(struct pw_node_factory *factory)
+{
+	pw_log_debug("node-factory %p: destroy", factory);
+
+	if (factory->global) {
+		spa_list_remove(&factory->link);
+		pw_global_destroy(factory->global);
+	}
+	if (factory->name)
+		free((char *)factory->name);
+
+	free(factory);
+}
+
 void pw_node_factory_export(struct pw_node_factory *factory,
 			    struct pw_client *owner,
 			    struct pw_global *parent)
 {
 	struct pw_core *core = factory->core;
-	spa_list_insert(core->node_factory_list.prev, &factory->link);
+	spa_list_append(&core->node_factory_list, &factory->link);
         factory->global = pw_core_add_global(core, owner, parent, core->type.node_factory, 0, NULL, factory);
 }
 

src/pipewire/node-factory.h

@@ -60,6 +60,8 @@ void pw_node_factory_export(struct pw_node_factory *factory,
 			    struct pw_client *owner,
 			    struct pw_global *parent);
 
+void pw_node_factory_destroy(struct pw_node_factory *factory);
+
 void *pw_node_factory_get_user_data(struct pw_node_factory *factory);
 
 void pw_node_factory_set_implementation(struct pw_node_factory *factory,

src/pipewire/node.c

@@ -180,6 +180,23 @@ static void node_have_output(void *data)
 	spa_graph_have_output(this->rt.graph, &this->rt.node);
 }
 
+static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
+{
+        struct impl *impl = data;
+	struct pw_node *this = &impl->this;
+        struct spa_graph_port *p, *pp;
+
+	spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link) {
+		if (p->port_id != port_id)
+			continue;
+
+		pp = p->peer;
+		if (pp && pp->callbacks->reuse_buffer)
+			pp->callbacks->reuse_buffer(pp->callbacks_data, buffer_id);
+		break;
+	}
+}
+
 static void node_unbind_func(void *data)
 {
 	struct pw_resource *resource = data;
@@ -190,7 +207,6 @@ static void
 update_info(struct pw_node *this)
 {
 	this->info.input_formats = NULL;
-
 	if (!spa_list_is_empty(&this->input_ports)) {
 		struct pw_port *port = spa_list_first(&this->input_ports, struct pw_port, link);
 
@@ -356,6 +372,7 @@ static const struct pw_node_events node_events = {
 	.event = node_event,
 	.need_input = node_need_input,
 	.have_output = node_have_output,
+	.reuse_buffer = node_reuse_buffer,
 };
 
 struct pw_node *pw_node_new(struct pw_core *core,
@@ -485,6 +502,11 @@ void pw_node_add_listener(struct pw_node *node,
 	spa_hook_list_append(&node->listener_list, listener, events, data);
 }
 
+struct spa_hook_list *pw_node_get_listeners(struct pw_node *node)
+{
+	return &node->listener_list;
+}
+
 static int
 do_node_remove(struct spa_loop *loop,
 	       bool async, uint32_t seq, size_t size, const void *data, void *user_data)
@@ -553,6 +575,20 @@ void pw_node_destroy(struct pw_node *node)
 	free(impl);
 }
 
+void pw_node_set_max_ports(struct pw_node *node,
+			   uint32_t max_input_ports,
+			   uint32_t max_output_ports)
+{
+	if (node->info.max_input_ports != max_input_ports) {
+		node->info.max_input_ports = max_input_ports;
+		node->info.change_mask |= PW_NODE_CHANGE_MASK_INPUT_PORTS;
+	}
+	if (node->info.max_output_ports != max_output_ports) {
+		node->info.max_output_ports = max_output_ports;
+		node->info.change_mask |= PW_NODE_CHANGE_MASK_OUTPUT_PORTS;
+	}
+}
+
 bool pw_node_for_each_port(struct pw_node *node,
 			   enum pw_direction direction,
 			   bool (*callback) (void *data, struct pw_port *port),

src/pipewire/node.h

@@ -102,6 +102,8 @@ struct pw_node_events {
 	void (*need_input) (void *data);
         /** the node has output */
 	void (*have_output) (void *data);
+        /** the node has a buffer to reuse */
+	void (*reuse_buffer) (void *data, uint32_t port_id, uint32_t buffer_id);
 };
 
 /** Create a new node \memberof pw_node */
@@ -119,6 +121,10 @@ void pw_node_register(struct pw_node *node);
 /** Destroy a node */
 void pw_node_destroy(struct pw_node *node);
 
+void pw_node_set_max_ports(struct pw_node *node,
+			   uint32_t max_input_ports,
+			   uint32_t max_output_ports);
+
 const struct pw_node_info *pw_node_get_info(struct pw_node *node);
 
 void * pw_node_get_user_data(struct pw_node *node);
@@ -142,6 +148,8 @@ void pw_node_add_listener(struct pw_node *node,
 			  const struct pw_node_events *events,
 			  void *data);
 
+struct spa_hook_list *pw_node_get_listeners(struct pw_node *node);
+
 /** iterate the ports in the given direction */
 bool pw_node_for_each_port(struct pw_node *node,
 			   enum pw_direction direction,

src/pipewire/port.c

@@ -340,6 +340,15 @@ void pw_port_destroy(struct pw_port *port)
 
 	pw_log_debug("port %p: free", port);
 	spa_hook_list_call(&port->listener_list, struct pw_port_events, free);
+
+	if (port->buffers)
+		free(port->buffers);
+	if (port->allocated)
+		pw_memblock_free(&port->buffer_mem);
+
+	if (port->properties)
+		pw_properties_free(port->properties);
+
 	free(port);
 }
 

src/pipewire/private.h

@@ -37,6 +37,24 @@ struct pw_command {
 	const char *name;	/**< command name */
 };
 
+struct pw_protocol {
+	struct spa_list link;                   /**< link in core protocol_list */
+	struct pw_core *core;                   /**< core for this protocol */
+
+	char *name;                             /**< type name of the protocol */
+
+	struct spa_list marshal_list;           /**< list of marshallers for supported interfaces */
+	struct spa_list client_list;            /**< list of current clients */
+	struct spa_list server_list;            /**< list of current servers */
+	struct spa_hook_list listener_list;	/**< event listeners */
+
+	const struct pw_protocol_implementaton *implementation; /**< implementation of the protocol */
+
+	const void *extension;  /**< extension API */
+
+	void *user_data;        /**< user data for the implementation */
+};
+
 struct pw_client {
 	struct pw_core *core;		/**< core object */
 	struct spa_list link;		/**< link in core object client list */
@@ -316,7 +334,7 @@ struct pw_remote {
 	struct spa_list stream_list;		/**< list of \ref pw_stream objects */
 	struct spa_list remote_node_list;	/**< list of \ref pw_remote_node objects */
 
-	struct pw_protocol_connection *conn;	/**< the protocol connection */
+	struct pw_protocol_client *conn;	/**< the protocol client connection */
 
 	enum pw_remote_state state;
 	char *error;

src/pipewire/protocol.c

@@ -41,8 +41,9 @@ struct pw_protocol *pw_protocol_new(struct pw_core *core,
 	protocol->name = strdup(name);
 
 	spa_list_init(&protocol->marshal_list);
-	spa_list_init(&protocol->connection_list);
+	spa_list_init(&protocol->server_list);
-	spa_list_init(&protocol->listener_list);
+	spa_list_init(&protocol->client_list);
+	spa_hook_list_init(&protocol->listener_list);
 
 	if (user_data_size > 0)
 		protocol->user_data = SPA_MEMBER(protocol, sizeof(struct impl), void);
@@ -59,34 +60,52 @@ void *pw_protocol_get_user_data(struct pw_protocol *protocol)
 	return protocol->user_data;
 }
 
+const struct pw_protocol_implementaton *
+pw_protocol_get_implementation(struct pw_protocol *protocol)
+{
+	return protocol->implementation;
+}
+
+const void *
+pw_protocol_get_extension(struct pw_protocol *protocol)
+{
+	return protocol->extension;
+}
+
 void pw_protocol_destroy(struct pw_protocol *protocol)
 {
 	struct impl *impl = SPA_CONTAINER_OF(protocol, struct impl, this);
 	struct marshal *marshal, *t1;
-	struct pw_protocol_listener *listener, *t2;
+	struct pw_protocol_server *server, *t2;
-	struct pw_protocol_connection *connection, *t3;
+	struct pw_protocol_client *client, *t3;
 
 	pw_log_info("protocol %p: destroy", protocol);
+	spa_hook_list_call(&protocol->listener_list, struct pw_protocol_events, destroy);
 
 	spa_list_remove(&protocol->link);
 
 	spa_list_for_each_safe(marshal, t1, &protocol->marshal_list, link)
 		free(marshal);
 
-	spa_list_for_each_safe(listener, t2, &protocol->listener_list, link)
+	spa_list_for_each_safe(server, t2, &protocol->server_list, link)
-		pw_protocol_listener_destroy(listener);
+		pw_protocol_server_destroy(server);
 
-	spa_list_for_each_safe(connection, t3, &protocol->connection_list, link)
+	spa_list_for_each_safe(client, t3, &protocol->client_list, link)
-		pw_protocol_connection_destroy(connection);
+		pw_protocol_client_destroy(client);
 
 	free(protocol->name);
 
-	if (protocol->destroy)
-		protocol->destroy(protocol);
-
 	free(impl);
 }
 
+void pw_protocol_add_listener(struct pw_protocol *protocol,
+                              struct spa_hook *listener,
+                              const struct pw_protocol_events *events,
+                              void *data)
+{
+	spa_hook_list_append(&protocol->listener_list, listener, events, data);
+}
+
 void
 pw_protocol_add_marshal(struct pw_protocol *protocol,
 			const struct pw_protocol_marshal *marshal)

src/pipewire/protocol.h

@@ -26,6 +26,8 @@ extern "C" {
 
 #include <spa/list.h>
 
+struct pw_protocol;
+
 #include <pipewire/core.h>
 #include <pipewire/properties.h>
 #include <pipewire/utils.h>
@@ -33,33 +35,33 @@ extern "C" {
 #define PW_TYPE__Protocol               "PipeWire:Protocol"
 #define PW_TYPE_PROTOCOL_BASE           PW_TYPE__Protocol ":"
 
-struct pw_protocol_connection {
+struct pw_protocol_client {
-	struct spa_list link;		/**< link in protocol connection_list */
+	struct spa_list link;		/**< link in protocol client_list */
 	struct pw_protocol *protocol;	/**< the owner protocol */
 
 	struct pw_remote *remote;	/**< the associated remote */
 
-	int (*connect) (struct pw_protocol_connection *conn);
+	int (*connect) (struct pw_protocol_client *client);
-	int (*connect_fd) (struct pw_protocol_connection *conn, int fd);
+	int (*connect_fd) (struct pw_protocol_client *client, int fd);
-	void (*disconnect) (struct pw_protocol_connection *conn);
+	void (*disconnect) (struct pw_protocol_client *client);
-	void (*destroy) (struct pw_protocol_connection *conn);
+	void (*destroy) (struct pw_protocol_client *client);
 };
 
-#define pw_protocol_connection_connect(c)	((c)->connect(c))
+#define pw_protocol_client_connect(c)		((c)->connect(c))
-#define pw_protocol_connection_connect_fd(c,fd)	((c)->connect_fd(c,fd))
+#define pw_protocol_client_connect_fd(c,fd)	((c)->connect_fd(c,fd))
-#define pw_protocol_connection_disconnect(c)	((c)->disconnect(c))
+#define pw_protocol_client_disconnect(c)	((c)->disconnect(c))
-#define pw_protocol_connection_destroy(c)	((c)->destroy(c))
+#define pw_protocol_client_destroy(c)		((c)->destroy(c))
 
-struct pw_protocol_listener {
+struct pw_protocol_server {
-	struct spa_list link;		/**< link in protocol listener_list */
+	struct spa_list link;		/**< link in protocol server_list */
 	struct pw_protocol *protocol;	/**< the owner protocol */
 
-	struct spa_list client_list;	/**< list of client of this protocol */
+	struct spa_list client_list;	/**< list of clients of this protocol */
 
-	void (*destroy) (struct pw_protocol_listener *listen);
+	void (*destroy) (struct pw_protocol_server *listen);
 };
 
-#define pw_protocol_listener_destroy(l)	((l)->destroy(l))
+#define pw_protocol_server_destroy(l)	((l)->destroy(l))
 
 struct pw_protocol_marshal {
         const char *type;               /**< interface type */
@@ -75,41 +77,44 @@ struct pw_protocol_marshal {
 struct pw_protocol_implementaton {
 #define PW_VERSION_PROTOCOL_IMPLEMENTATION	0
 	uint32_t version;
-	struct pw_protocol_connection * (*new_connection) (struct pw_protocol *protocol,
+
+	struct pw_protocol_client * (*new_client) (struct pw_protocol *protocol,
 						   struct pw_remote *remote,
 						   struct pw_properties *properties);
-	struct pw_protocol_listener * (*add_listener) (struct pw_protocol *protocol,
+	struct pw_protocol_server * (*add_server) (struct pw_protocol *protocol,
 						   struct pw_core *core,
 						   struct pw_properties *properties);
 };
 
-struct pw_protocol {
+struct pw_protocol_events {
-	struct spa_list link;			/**< link in core protocol_list */
+#define PW_VERSION_PROTOCOL_EVENTS		0
-	struct pw_core *core;			/**< core for this protocol */
+	uint32_t version;
 
-	char *name;				/**< type name of the protocol */
+	void (*destroy) (void *data);
-
-	struct spa_list marshal_list;		/**< list of marshallers for supported interfaces */
-	struct spa_list connection_list;	/**< list of current connections */
-	struct spa_list listener_list;		/**< list of current listeners */
-
-	const struct pw_protocol_implementaton *implementation;	/**< implementation of the protocol */
-
-	const void *extension;	/**< extension API */
-
-	void *user_data;	/**< user data for the implementation */
-        pw_destroy_t destroy;	/**< function to clean up the object */
 };
 
-#define pw_protocol_new_connection(p,...)	((p)->implementation->new_connection(p,__VA_ARGS__))
+#define pw_protocol_new_client(p,...)	(pw_protocol_get_implementation(p)->new_client(p,__VA_ARGS__))
-#define pw_protocol_add_listener(p,...)		((p)->implementation->add_listener(p,__VA_ARGS__))
+#define pw_protocol_add_server(p,...)	(pw_protocol_get_implementation(p)->add_server(p,__VA_ARGS__))
-
+#define pw_protocol_ext(p,type,method,...)	(((type*)pw_protocol_get_extension(p))->method( __VA_ARGS__))
-#define pw_protocol_ext(p,type,method,...)	(((type*)(p)->extension)->method( __VA_ARGS__))
 
 struct pw_protocol *pw_protocol_new(struct pw_core *core, const char *name, size_t user_data_size);
 
+void pw_protocol_destroy(struct pw_protocol *protocol);
+
 void *pw_protocol_get_user_data(struct pw_protocol *protocol);
 
+const struct pw_protocol_implementaton *
+pw_protocol_get_implementation(struct pw_protocol *protocol);
+
+const void *
+pw_protocol_get_extension(struct pw_protocol *protocol);
+
+
+void pw_protocol_add_listener(struct pw_protocol *protocol,
+                              struct spa_hook *listener,
+                              const struct pw_protocol_events *events,
+                              void *data);
+
 /** \class pw_protocol
  *
  * \brief Manages protocols and their implementation

src/pipewire/remote.c

@@ -241,7 +241,7 @@ struct pw_remote *pw_remote_new(struct pw_core *core,
 	if (protocol == NULL)
 		goto no_protocol;
 
-	this->conn = pw_protocol_new_connection(protocol, this, properties);
+	this->conn = pw_protocol_new_client(protocol, this, properties);
 	if (this->conn == NULL)
 		goto no_connection;
 
@@ -273,7 +273,7 @@ void pw_remote_destroy(struct pw_remote *remote)
 	spa_list_for_each_safe(stream, s2, &remote->stream_list, link)
 		pw_stream_destroy(stream);
 
-	pw_protocol_connection_destroy (remote->conn);
+	pw_protocol_client_destroy (remote->conn);
 
 	spa_list_remove(&remote->link);
 
@@ -322,7 +322,7 @@ static int do_connect(struct pw_remote *remote)
 	return 0;
 
       no_proxy:
-	pw_protocol_connection_disconnect (remote->conn);
+	pw_protocol_client_disconnect (remote->conn);
 	pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "can't connect: no memory");
 	return -1;
 }
@@ -348,7 +348,7 @@ int pw_remote_connect(struct pw_remote *remote)
 
 	pw_remote_update_state(remote, PW_REMOTE_STATE_CONNECTING, NULL);
 
-	if ((res = pw_protocol_connection_connect (remote->conn)) < 0) {
+	if ((res = pw_protocol_client_connect (remote->conn)) < 0) {
 		pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "connect failed");
 		return res;
 	}
@@ -362,7 +362,7 @@ int pw_remote_connect_fd(struct pw_remote *remote, int fd)
 
 	pw_remote_update_state(remote, PW_REMOTE_STATE_CONNECTING, NULL);
 
-	if ((res = pw_protocol_connection_connect_fd (remote->conn, fd)) < 0) {
+	if ((res = pw_protocol_client_connect_fd (remote->conn, fd)) < 0) {
 		pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "connect_fd failed");
 		return res;
 	}
@@ -379,7 +379,7 @@ void pw_remote_disconnect(struct pw_remote *remote)
 	spa_list_for_each_safe(stream, s2, &remote->stream_list, link)
 	    pw_stream_disconnect(stream);
 
-	pw_protocol_connection_disconnect (remote->conn);
+	pw_protocol_client_disconnect (remote->conn);
 
 	spa_list_for_each_safe(proxy, t2, &remote->proxy_list, link)
 	    pw_proxy_destroy(proxy);