move audio-dsp to audio-session

Make a new module audio-session that manages clients and what audio
sessions they use.
Move the audio-dsp to the audio-session

src/daemon/pipewire.conf.in

@@ -6,9 +6,9 @@ load-module libpipewire-module-spa-monitor alsa/libspa-alsa alsa-monitor alsa
 load-module libpipewire-module-spa-monitor v4l2/libspa-v4l2 v4l2-monitor v4l2
 #load-module libpipewire-module-spa-monitor bluez5/libspa-bluez5 bluez5-monitor bluez5
 #load-module libpipewire-module-spa-node videotestsrc/libspa-videotestsrc videotestsrc videotestsrc Spa:POD:Object:Props:patternType=Spa:POD:Object:Props:patternType:snow
-load-module libpipewire-module-autolink
+#load-module libpipewire-module-autolink
 #load-module libpipewire-module-mixer
 load-module libpipewire-module-client-node
 load-module libpipewire-module-flatpak
-load-module libpipewire-module-audio-dsp
+load-module libpipewire-module-audio-session
 load-module libpipewire-module-link-factory

src/modules/meson.build

@@ -91,8 +91,10 @@ pipewire_module_protocol_native = shared_library('pipewire-module-protocol-nativ
   dependencies : [mathlib, dl_lib, pipewire_dep],
 )
 
-pipewire_module_audio_dsp = shared_library('pipewire-module-audio-dsp',
+pipewire_module_audio_session = shared_library('pipewire-module-audio-session',
-  [ 'module-audio-dsp.c', 'spa/spa-node.c' ],
+  [ 'module-audio-session.c',
+    'module-audio-session/audio-dsp.c',
+    'spa/spa-node.c' ],
   c_args : pipewire_module_c_args,
   include_directories : [configinc, spa_inc],
   link_with : spalib,

src/modules/module-audio-session.c

@@ -0,0 +1,723 @@
+/* PipeWire
+ * Copyright (C) 2018 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <string.h>
+#include <stdio.h>
+#include <errno.h>
+#include <math.h>
+#include <time.h>
+
+#include "config.h"
+
+#include <spa/node/node.h>
+#include <spa/utils/hook.h>
+#include <spa/param/audio/format-utils.h>
+
+#include <spa/lib/pod.h>
+#include <spa/lib/debug.h>
+
+#include "pipewire/core.h"
+#include "pipewire/control.h"
+#include "pipewire/link.h"
+#include "pipewire/log.h"
+#include "pipewire/module.h"
+#include "pipewire/type.h"
+#include "pipewire/private.h"
+
+#include "module-audio-session/audio-dsp.h"
+
+#define DEFAULT_CHANNELS	2
+#define DEFAULT_SAMPLE_RATE	44100
+#define DEFAULT_BUFFER_SIZE	64
+#define MAX_BUFFER_SIZE		2048
+
+struct type {
+	struct spa_type_media_type media_type;
+        struct spa_type_media_subtype media_subtype;
+        struct spa_type_format_audio format_audio;
+        struct spa_type_audio_format audio_format;
+        struct spa_type_media_subtype_audio media_subtype_audio;
+};
+
+static inline void init_type(struct type *type, struct spa_type_map *map)
+{
+        spa_type_media_type_map(map, &type->media_type);
+        spa_type_media_subtype_map(map, &type->media_subtype);
+        spa_type_format_audio_map(map, &type->format_audio);
+        spa_type_audio_format_map(map, &type->audio_format);
+        spa_type_media_subtype_audio_map(map, &type->media_subtype_audio);
+}
+
+struct impl {
+	struct type type;
+
+	struct pw_core *core;
+	struct pw_type *t;
+	struct pw_module *module;
+	struct spa_hook core_listener;
+	struct spa_hook module_listener;
+	struct pw_properties *properties;
+
+	struct spa_list session_list;
+};
+
+struct session {
+	struct spa_list l;
+
+	uint32_t id;
+
+	struct impl *impl;
+
+	enum pw_direction direction;
+
+	struct pw_node *node;
+	struct spa_hook node_listener;
+	struct pw_port *node_port;
+
+	struct pw_node *dsp;
+	struct spa_hook dsp_listener;
+	struct pw_port *dsp_port;
+
+	struct pw_link *link;
+
+	int sample_rate;
+	int buffer_size;
+
+	struct spa_list node_list;
+};
+
+struct node_info {
+	struct spa_list l;
+
+	struct impl *impl;
+	struct session *session;
+	struct pw_node *node;
+	struct spa_hook node_listener;
+
+	uint32_t sample_rate;
+	uint32_t buffer_size;
+
+	struct spa_list links;
+};
+
+struct link_data {
+	struct spa_list l;
+
+	struct node_info *node_info;
+	struct pw_link *link;
+	struct spa_hook link_listener;
+};
+
+
+/** \endcond */
+
+static void link_data_remove(struct link_data *data)
+{
+	spa_list_remove(&data->l);
+	spa_hook_remove(&data->link_listener);
+}
+
+static void node_info_free(struct node_info *info)
+{
+	struct link_data *ld, *t;
+
+	spa_list_remove(&info->l);
+	spa_hook_remove(&info->node_listener);
+	spa_list_for_each_safe(ld, t, &info->links, l)
+		link_data_remove(ld);
+	free(info);
+}
+
+static void session_destroy(struct session *sess)
+{
+	struct node_info *ni, *t;
+
+	spa_list_remove(&sess->l);
+	spa_hook_remove(&sess->node_listener);
+	spa_list_for_each_safe(ni, t, &sess->node_list, l)
+		node_info_free(ni);
+	pw_node_destroy(sess->dsp);
+}
+
+static void
+link_port_unlinked(void *data, struct pw_port *port)
+{
+	struct link_data *ld = data;
+	struct node_info *info = ld->node_info;
+	struct pw_link *link = ld->link;
+	struct impl *impl = info->impl;
+
+	pw_log_debug("module %p: link %p: port %p unlinked", impl, link, port);
+}
+
+static void
+link_state_changed(void *data, enum pw_link_state old, enum pw_link_state state, const char *error)
+{
+	struct link_data *ld = data;
+	struct node_info *info = ld->node_info;
+	struct pw_link *link = ld->link;
+	struct impl *impl = info->impl;
+
+	switch (state) {
+	case PW_LINK_STATE_ERROR:
+	{
+		struct pw_global *global = pw_node_get_global(info->node);
+		struct pw_client *owner = pw_global_get_owner(global);
+
+		pw_log_debug("module %p: link %p: state error: %s", impl, link, error);
+		if (owner)
+			pw_resource_error(pw_client_get_core_resource(owner), -ENODEV, error);
+
+		break;
+	}
+
+	case PW_LINK_STATE_UNLINKED:
+		pw_log_debug("module %p: link %p: unlinked", impl, link);
+		break;
+
+	case PW_LINK_STATE_INIT:
+	case PW_LINK_STATE_NEGOTIATING:
+	case PW_LINK_STATE_ALLOCATING:
+	case PW_LINK_STATE_PAUSED:
+	case PW_LINK_STATE_RUNNING:
+		break;
+	}
+}
+
+static void try_link_controls(struct impl *impl, struct pw_port *port, struct pw_port *target)
+{
+	struct pw_control *cin, *cout;
+	int res;
+
+	pw_log_debug("module %p: trying controls", impl);
+	spa_list_for_each(cout, &port->control_list[SPA_DIRECTION_OUTPUT], port_link) {
+		spa_list_for_each(cin, &target->control_list[SPA_DIRECTION_INPUT], port_link) {
+			if (cin->prop_id == cout->prop_id) {
+				if ((res = pw_control_link(cout, cin)) < 0)
+					pw_log_error("failed to link controls: %s", spa_strerror(res));
+			}
+		}
+	}
+	spa_list_for_each(cin, &port->control_list[SPA_DIRECTION_INPUT], port_link) {
+		spa_list_for_each(cout, &target->control_list[SPA_DIRECTION_OUTPUT], port_link) {
+			if (cin->prop_id == cout->prop_id) {
+				if ((res = pw_control_link(cout, cin)) < 0)
+					pw_log_error("failed to link controls: %s", spa_strerror(res));
+			}
+		}
+	}
+
+
+}
+
+static void
+link_destroy(void *data)
+{
+	struct link_data *ld = data;
+	pw_log_debug("module %p: link %p destroyed", ld->node_info->impl, ld->link);
+	link_data_remove(ld);
+}
+
+static const struct pw_link_events link_events = {
+	PW_VERSION_LINK_EVENTS,
+	.destroy = link_destroy,
+	.port_unlinked = link_port_unlinked,
+	.state_changed = link_state_changed,
+};
+
+static int link_ports(struct node_info *info, struct pw_port *port, struct pw_port *target)
+{
+	struct impl *impl = info->impl;
+	struct pw_link *link;
+	struct link_data *ld;
+	char *error = NULL;
+
+	if (pw_port_get_direction(port) == PW_DIRECTION_INPUT) {
+	        struct pw_port *tmp = target;
+		target = port;
+		port = tmp;
+	}
+
+	link = pw_link_new(impl->core,
+			   port, target,
+			   NULL, NULL,
+			   &error,
+			   sizeof(struct link_data));
+	if (link == NULL)
+		return -ENOMEM;
+
+	ld = pw_link_get_user_data(link);
+	ld->link = link;
+	ld->node_info = info;
+	pw_link_add_listener(link, &ld->link_listener, &link_events, ld);
+
+	spa_list_append(&info->links, &ld->l);
+	pw_link_register(link, NULL, pw_module_get_global(impl->module), NULL);
+
+	try_link_controls(impl, port, target);
+	return 0;
+}
+
+static int on_peer_port(void *data, struct pw_port *port)
+{
+	struct node_info *info = data;
+	struct pw_port *p;
+
+	p = pw_node_get_free_port(info->node, pw_direction_reverse(port->direction));
+	if (p == NULL)
+		return 0;
+
+	return link_ports(info, p, port);
+}
+
+static void reconfigure_session(struct session *sess)
+{
+	struct node_info *ni;
+	uint32_t buffer_size = 1024;
+
+	spa_list_for_each(ni, &sess->node_list, l)
+		buffer_size = SPA_MIN(buffer_size, ni->buffer_size);
+
+	sess->buffer_size = buffer_size;
+}
+
+static void node_info_destroy(void *data)
+{
+	struct node_info *info = data;
+	struct session *session = info->session;
+
+	node_info_free(info);
+
+	reconfigure_session(session);
+}
+
+static const struct pw_node_events node_info_events = {
+	PW_VERSION_NODE_EVENTS,
+	.destroy = node_info_destroy,
+};
+
+static int link_session_dsp(struct session *session)
+{
+	struct impl *impl = session->impl;
+	struct pw_port *op, *ip;
+	char *error = NULL;
+
+	if (session->direction == PW_DIRECTION_OUTPUT) {
+		op = session->dsp_port;
+		ip = session->node_port;
+	}
+	else {
+		op = session->node_port;
+		ip = session->dsp_port;
+	}
+
+	session->link = pw_link_new(impl->core,
+			   op,
+			   ip,
+			   NULL,
+			   pw_properties_new(PW_LINK_PROP_PASSIVE, "true", NULL),
+			   &error, 0);
+
+	if (session->link == NULL) {
+		pw_log_error("can't create link: %s", error);
+		free(error);
+		return -ENOMEM;
+	}
+	pw_link_register(session->link, NULL, pw_module_get_global(impl->module), NULL);
+
+	return 0;
+}
+
+struct find_data {
+	struct impl *impl;
+	uint32_t path_id;
+	const char *media_class;
+	struct session *sess;
+	uint64_t plugged;
+};
+
+static int find_session(void *data, struct session *sess)
+{
+	struct find_data *find = data;
+	struct impl *impl = find->impl;
+	const struct pw_properties *props;
+	const char *str;
+	uint64_t plugged = 0;
+
+	pw_log_debug("module %p: looking at session '%d'", impl, sess->id);
+
+	if (find->path_id != SPA_ID_INVALID && sess->id != find->path_id)
+		return 0;
+
+	if (find->path_id == SPA_ID_INVALID) {
+		if ((props = pw_node_get_properties(sess->node)) == NULL)
+			return 0;
+
+		if ((str = pw_properties_get(props, "media.class")) == NULL)
+			return 0;
+
+		if (strcmp(str, find->media_class) != 0)
+			return 0;
+
+		if ((str = pw_properties_get(props, "node.plugged")) != NULL)
+			plugged = pw_properties_parse_uint64(str);
+	}
+
+	pw_log_debug("module %p: found session '%d' %" PRIu64, impl,
+			sess->id, plugged);
+
+	if (find->sess == NULL || plugged > find->plugged) {
+		pw_log_debug("module %p: new best %" PRIu64, impl, plugged);
+		find->sess = sess;
+		find->plugged = plugged;
+	}
+	return 0;
+}
+
+static void handle_autoconnect(struct impl *impl, struct pw_node *node,
+		const struct pw_properties *props)
+{
+        struct pw_node *peer;
+        const char *media, *category, *role, *str;
+        bool exclusive;
+        struct find_data find;
+        enum pw_direction direction;
+	struct session *session;
+	struct node_info *info;
+	uint32_t sample_rate, buffer_size;
+
+	sample_rate = 44100;
+	buffer_size = 1024;
+
+	if ((media = pw_properties_get(props, PW_NODE_PROP_MEDIA)) == NULL)
+		media = "Audio";
+	if (strcmp(media, "Audio"))
+		return;
+
+	if ((category = pw_properties_get(props, PW_NODE_PROP_CATEGORY)) == NULL)
+		category = "Playback";
+
+	if ((role = pw_properties_get(props, PW_NODE_PROP_ROLE)) == NULL)
+		role = "Music";
+
+	if ((str = pw_properties_get(props, PW_NODE_PROP_EXCLUSIVE)) != NULL)
+		exclusive = pw_properties_parse_bool(str);
+	else
+		exclusive = false;
+
+	pw_log_debug("module %p: '%s' '%s' '%s' %d", impl, media, category, role, exclusive);
+
+	if (strcmp(category, "Playback") == 0)
+		find.media_class = "Audio/Sink";
+	else if (strcmp(category, "Capture") == 0)
+		find.media_class = "Audio/Source";
+	else
+		return;
+
+	str = pw_properties_get(props, PW_NODE_PROP_TARGET_NODE);
+	if (str != NULL)
+		find.path_id = atoi(str);
+	else
+		find.path_id = SPA_ID_INVALID;
+
+	pw_log_debug("module %p: try to find and link to node '%d'", impl, find.path_id);
+
+	find.impl = impl;
+	find.sess = NULL;
+	spa_list_for_each(session, &impl->session_list, l)
+		find_session(&find, session);
+	if (find.sess == NULL)
+		return;
+
+	session = find.sess;
+
+	if (strcmp(category, "Capture") == 0)
+		direction = PW_DIRECTION_OUTPUT;
+	else if (strcmp(category, "Playback") == 0)
+		direction = PW_DIRECTION_INPUT;
+	else
+		return;
+
+	if (exclusive) {
+		if (!spa_list_is_empty(&session->node_list)) {
+			pw_log_warn("session busy, can't get exclusive access");
+			return;
+		}
+		if (session->link != NULL) {
+			pw_log_warn("session busy with DSP");
+			return;
+		}
+		peer = session->node;
+	}
+	else {
+		if (session->link == NULL) {
+			if (link_session_dsp(session) < 0)
+				return;
+		}
+		peer = session->dsp;
+		sample_rate = session->sample_rate;
+		buffer_size = session->buffer_size;
+	}
+
+	pw_log_debug("module %p: linking to session '%d'", impl, session->id);
+
+	info = calloc(1, sizeof(struct node_info));
+	info->impl = impl;
+	info->session = session;
+	info->node = node;
+	info->sample_rate = sample_rate;
+	info->buffer_size = buffer_size;
+	spa_list_init(&info->links);
+
+	spa_list_append(&info->session->node_list, &info->l);
+
+	pw_node_add_listener(node, &info->node_listener, &node_info_events, info);
+
+	pw_node_for_each_port(peer, direction, on_peer_port, info);
+}
+
+static void node_destroy(void *data)
+{
+	struct session *sess = data;
+	session_destroy(sess);
+}
+
+static const struct pw_node_events node_events = {
+	PW_VERSION_NODE_EVENTS,
+	.destroy = node_destroy,
+};
+
+static void
+dsp_state_changed(void *data, enum pw_node_state old,
+		enum pw_node_state state, const char *error)
+{
+	struct session *sess = data;
+
+	switch(state) {
+	case PW_NODE_STATE_RUNNING:
+		if (sess->link == NULL) {
+			if (link_session_dsp(sess) < 0)
+				return;
+			pw_link_activate(sess->link);
+		}
+		break;
+	case PW_NODE_STATE_SUSPENDED:
+		if (sess->link != NULL) {
+			pw_link_destroy(sess->link);
+			sess->link = NULL;
+		}
+		break;
+	default:
+		break;
+	}
+}
+
+static const struct pw_node_events dsp_events = {
+	PW_VERSION_NODE_EVENTS,
+	.state_changed = dsp_state_changed,
+};
+
+struct channel_data {
+	struct impl *impl;
+	uint32_t channels;
+};
+
+static int collect_channel(void *data, uint32_t id, uint32_t index, uint32_t next, struct spa_pod *param)
+{
+	struct channel_data *d = data;
+	struct impl *impl = d->impl;
+	uint32_t media_type, media_subtype;
+	struct spa_audio_info_raw info;
+
+	spa_pod_object_parse(param,
+			"I", &media_type,
+			"I", &media_subtype);
+
+	if (media_type != impl->type.media_type.audio ||
+	    media_subtype != impl->type.media_subtype.raw)
+		return 0;
+
+        spa_pod_fixate(param);
+        spa_debug_pod(param, SPA_DEBUG_FLAG_FORMAT);
+
+	if (spa_format_audio_raw_parse(param, &info, &impl->type.format_audio) < 0)
+		return 0;
+
+	if (info.channels > d->channels)
+		d->channels = info.channels;
+
+	return 0;
+}
+
+
+static uint32_t find_port_channels(struct impl *impl, struct pw_port *port)
+{
+	struct pw_type *t = impl->t;
+	struct channel_data data = { impl, 0, };
+
+	pw_port_for_each_param(port,
+			t->param.idEnumFormat,
+			0, 0, NULL,
+			collect_channel, &data);
+
+	pw_log_debug("port channels %d", data.channels);
+	return data.channels;
+}
+
+static int on_global(void *data, struct pw_global *global)
+{
+	struct impl *impl = data;
+	struct pw_node *node, *dsp;
+	struct session *sess;
+	const struct pw_properties *properties;
+	const char *str;
+	enum pw_direction direction;
+	struct pw_port *node_port, *dsp_port;
+	uint32_t id, channels;
+
+	if (pw_global_get_type(global) != impl->t->node)
+		return 0;
+
+	node = pw_global_get_object(global);
+	id = pw_global_get_id(global);
+
+	properties = pw_node_get_properties(node);
+
+	str = pw_properties_get(properties, PW_NODE_PROP_AUTOCONNECT);
+        if (str != NULL && pw_properties_parse_bool(str)) {
+		handle_autoconnect(impl, node, properties);
+		return 0;
+	}
+	else if ((str = pw_properties_get(properties, "media.class")) == NULL)
+		return 0;
+
+	pw_log_debug("global added %s (%d)", str, id);
+
+	if (strcmp(str, "Audio/Sink") == 0)
+		direction = PW_DIRECTION_OUTPUT;
+	else if (strcmp(str, "Audio/Source") == 0)
+		direction = PW_DIRECTION_INPUT;
+	else
+		return 0;
+
+	if ((node_port = pw_node_get_free_port(node, pw_direction_reverse(direction))) == NULL)
+		return 0;
+
+	channels = find_port_channels(impl, node_port);
+	if (channels == 0)
+		return 0;
+
+	dsp = pw_audio_dsp_new(impl->core,
+			properties,
+			direction,
+			channels,
+			MAX_BUFFER_SIZE,
+			sizeof(struct session));
+	if (dsp == NULL)
+		return 0;
+
+	if ((dsp_port = pw_node_get_free_port(dsp, direction)) == NULL)
+		return 0;
+
+	sess = pw_audio_dsp_get_user_data(dsp);
+	sess->impl = impl;
+	sess->direction = direction;
+	sess->id = id;
+	sess->node = node;
+	sess->node_port = node_port;
+	sess->dsp = dsp;
+	sess->dsp_port = dsp_port;
+	spa_list_init(&sess->node_list);
+
+	spa_list_append(&impl->session_list, &sess->l);
+
+	pw_node_add_listener(dsp, &sess->dsp_listener, &dsp_events, sess);
+	pw_node_add_listener(node, &sess->node_listener, &node_events, sess);
+
+	pw_node_register(dsp, NULL, pw_module_get_global(impl->module), NULL);
+	pw_node_set_active(dsp, true);
+
+	return 0;
+}
+
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	struct session *s, *t;
+
+	spa_hook_remove(&impl->module_listener);
+	spa_hook_remove(&impl->core_listener);
+
+	spa_list_for_each_safe(s, t, &impl->session_list, l)
+		session_destroy(s);
+
+	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 void
+core_global_added(void *data, struct pw_global *global)
+{
+	on_global(data, global);
+}
+
+static const struct pw_core_events core_events = {
+	PW_VERSION_CORE_EVENTS,
+        .global_added = core_global_added,
+};
+
+static int module_init(struct pw_module *module, struct pw_properties *properties)
+{
+	struct pw_core *core = pw_module_get_core(module);
+	struct impl *impl;
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL)
+		return -ENOMEM;
+
+	pw_log_debug("module %p: new", impl);
+
+	impl->core = core;
+	impl->t = pw_core_get_type(core);
+	impl->module = module;
+	impl->properties = properties;
+
+	init_type(&impl->type, core->type.map);
+
+	spa_list_init(&impl->session_list);
+
+	pw_core_for_each_global(core, on_global, impl);
+
+	pw_core_add_listener(core, &impl->core_listener, &core_events, impl);
+	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
+	return 0;
+}
+
+int pipewire__module_init(struct pw_module *module, const char *args)
+{
+	return module_init(module, NULL);
+}

src/modules/module-audio-session/audio-dsp.c

@@ -44,9 +44,7 @@
 #define MAX_PORTS	256
 #define MAX_BUFFERS	8
 
-#define DEFAULT_CHANNELS	2
 #define DEFAULT_SAMPLE_RATE	44100
-#define DEFAULT_BUFFER_SIZE	1024
 
 struct type {
 	struct spa_type_media_type media_type;
@@ -65,21 +63,6 @@ static inline void init_type(struct type *type, struct spa_type_map *map)
         spa_type_media_subtype_audio_map(map, &type->media_subtype_audio);
 }
 
-struct impl {
-	struct type type;
-
-	struct pw_core *core;
-	struct pw_type *t;
-	struct pw_module *module;
-	struct spa_hook core_listener;
-	struct spa_hook module_listener;
-	struct pw_properties *properties;
-
-	int node_count;
-
-	struct spa_list node_list;
-};
-
 struct buffer {
 #define BUFFER_FLAG_OUT		(1<<0)
 	uint32_t flags;
@@ -117,16 +100,17 @@ typedef void (*conv_func_t)(void *dst, void *src, int index, int n_samples, int
 typedef void (*fill_func_t)(void *dst, int index, int n_samples, int stride);
 
 struct node {
-	struct spa_list link;
+	struct type type;
 
+	struct pw_core *core;
+	struct pw_type *t;
 	struct pw_node *node;
-	struct spa_hook node_listener;
 
-	struct impl *impl;
+	void *user_data;
 
 	int channels;
 	int sample_rate;
-	int buffer_size;
+	int max_buffer_size;
 
 	conv_func_t conv_func;
 	fill_func_t fill_func;
@@ -374,10 +358,6 @@ static int node_process_mix(struct spa_node *node)
 	outio->buffer_id = out->buf->id;
 	outio->status = SPA_STATUS_HAVE_BUFFER;
 
-	out->buf->datas[0].chunk->offset = 0;
-	out->buf->datas[0].chunk->size = n->buffer_size * outp->stride;
-	out->buf->datas[0].chunk->stride = outp->stride;
-
 	pw_log_trace(NAME " %p: output buffer %d %d %d", this,
 			out->buf->id, out->flags, out->buf->datas[0].chunk->size);
 
@@ -392,7 +372,7 @@ static int node_process_split(struct spa_node *node)
 	struct spa_io_buffers *inio = inp->io;
 	struct buffer *in;
 	int i, res, channels = n->channels;
-	size_t buffer_size = n->buffer_size;
+	size_t buffer_size;
 
         if (inio->status != SPA_STATUS_HAVE_BUFFER)
 		return SPA_STATUS_NEED_BUFFER;
@@ -401,6 +381,7 @@ static int node_process_split(struct spa_node *node)
 		return inio->status = -EINVAL;
 
 	in = &inp->buffers[inio->buffer_id];
+	buffer_size = in->buf->datas[0].chunk->size / inp->stride;
 	res = SPA_STATUS_NEED_BUFFER;
 
 	for (i = 0; i < channels; i++) {
@@ -447,7 +428,7 @@ static int port_set_io(struct spa_node *node,
 		       uint32_t id, void *data, size_t size)
 {
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
-	struct pw_type *t = n->impl->t;
+	struct pw_type *t = n->t;
 	struct port *p = GET_PORT(n, direction, port_id);
 
 	if (id == t->io.Buffers)
@@ -485,8 +466,8 @@ static int port_enum_formats(struct spa_node *node,
                              struct spa_pod_builder *builder)
 {
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
-	struct pw_type *type = n->impl->t;
+	struct pw_type *type = n->t;
-	struct type *t = &n->impl->type;
+	struct type *t = &n->type;
 
 	if (*index > 0)
 		return 0;
@@ -522,8 +503,7 @@ static int port_enum_formats(struct spa_node *node,
                         ":", t->format_audio.layout,   "i", SPA_AUDIO_LAYOUT_INTERLEAVED,
                         ":", t->format_audio.rate,     "iru", n->sample_rate,
 				SPA_POD_PROP_MIN_MAX(1, INT32_MAX),
-                        ":", t->format_audio.channels, "iru", n->channels,
+                        ":", t->format_audio.channels, "i", n->channels);
-				SPA_POD_PROP_MIN_MAX(1, MAX_PORTS));
 	}
 
 	return 1;
@@ -538,7 +518,7 @@ static int port_enum_params(struct spa_node *node,
 {
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
 	struct port *p = GET_PORT(n, direction, port_id);
-	struct pw_type *t = n->impl->t;
+	struct pw_type *t = n->t;
 	struct spa_pod *param;
 	struct spa_pod_builder b = { 0 };
 	uint8_t buffer[1024];
@@ -558,11 +538,12 @@ static int port_enum_params(struct spa_node *node,
 	else if (id == t->param.idBuffers) {
 		if (*index > 0)
 			return 0;
+		if (p->stride == 0)
+			return -EIO;
 
 		param = spa_pod_builder_object(&b,
 			id, t->param_buffers.Buffers,
-			":", t->param_buffers.size,    "i", n->buffer_size * p->stride,
+			":", t->param_buffers.size,    "i", n->max_buffer_size * p->stride,
-//				SPA_POD_PROP_MIN_MAX(24, 4096),
 			":", t->param_buffers.blocks,  "i", 1,
 			":", t->param_buffers.stride,  "i", p->stride,
 			":", t->param_buffers.buffers, "ir", 2,
@@ -586,10 +567,11 @@ static int port_set_format(struct spa_node *node, struct port *p,
 {
 	struct spa_audio_info info = { 0 };
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
-	struct type *t = &n->impl->type;
+	struct type *t = &n->type;
 
 	if (format == NULL) {
 		clear_buffers(n, p);
+		p->stride = 0;
 		return 0;
 	}
 
@@ -644,7 +626,7 @@ static int port_set_param(struct spa_node *node,
 {
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
 	struct port *p = GET_PORT(n, direction, port_id);
-	struct pw_type *t = n->impl->t;
+	struct pw_type *t = n->t;
 
 	if (id == t->param.idFormat) {
 		return port_set_format(node, p, direction, flags, param);
@@ -660,7 +642,7 @@ static int port_use_buffers(struct spa_node *node, enum spa_direction direction,
 {
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
 	struct port *p = GET_PORT(n, direction, port_id);
-	struct pw_type *t = n->impl->t;
+	struct pw_type *t = n->t;
 	int i;
 
 	pw_log_debug("use_buffers %d", n_buffers);
@@ -697,7 +679,7 @@ static int port_alloc_buffers(struct spa_node *node, enum spa_direction directio
 #if 0
 	struct node *n = SPA_CONTAINER_OF(node, struct node, node_impl);
 	struct port *p = GET_PORT(n, direction, port_id);
-	struct pw_type *t = n->impl->t;
+	struct pw_type *t = n->t;
 	int i;
 
 	pw_log_debug("alloc %d", *n_buffers);
@@ -765,7 +747,7 @@ static int schedule_mix(struct spa_node *_node)
 	struct spa_graph_node *node = &port->rt.mix_node;
 	struct spa_graph_port *gp;
 	struct spa_io_buffers *io = port->rt.mix_port.io;
-	size_t buffer_size = n->buffer_size;
+	size_t buffer_size = 0;
 	struct buffer *outb;
 	float *out = NULL;
 	int layer = 0;
@@ -775,60 +757,54 @@ static int schedule_mix(struct spa_node *_node)
 
 	spa_list_for_each(gp, &node->ports[SPA_DIRECTION_INPUT], link) {
 		struct pw_port_mix *mix = SPA_CONTAINER_OF(gp, struct pw_port_mix, port);
+		struct spa_io_buffers *inio;
+		struct spa_buffer *inb;
 
-		pw_log_trace("mix %p: input %d %d/%d", node,
+		if ((inio = gp->io) == NULL ||
-				gp->io->status, gp->io->buffer_id, mix->n_buffers);
+		    inio->buffer_id >= mix->n_buffers ||
-
+		    inio->status != SPA_STATUS_HAVE_BUFFER)
-		if (!(gp->io->buffer_id < mix->n_buffers && gp->io->status == SPA_STATUS_HAVE_BUFFER))
 			continue;
 
+		pw_log_trace("mix %p: input %d %d/%d", node,
+				inio->status, inio->buffer_id, mix->n_buffers);
+
+		inb = mix->buffers[inio->buffer_id];
+		buffer_size = inb->datas[0].chunk->size / sizeof(float);
+
 		if (layer++ == 0) {
-			out = mix->buffers[gp->io->buffer_id]->datas[0].data;
+			out = inb->datas[0].data;
 		}
 		else {
-			add_f32(out, mix->buffers[gp->io->buffer_id]->datas[0].data, buffer_size);
+			add_f32(out, inb->datas[0].data, buffer_size);
 		}
 
-		pw_log_trace("mix %p: input %p %p->%p %d %d %zd", node,
+		pw_log_trace("mix %p: input %p %p %zd", node, inio, io, buffer_size);
-				gp, gp->io, io, gp->io->status, gp->io->buffer_id, buffer_size);
 	}
 
 	outb = peek_buffer(n, outp);
 	if (outb == NULL)
 		return -EPIPE;
 
-	if (layer > 0)
+	if (layer > 0) {
 		n->conv_func(outb->ptr, out, port->port_id, buffer_size, stride);
-	else
+
+		outb->buf->datas[0].chunk->offset = 0;
+		outb->buf->datas[0].chunk->size = buffer_size * outp->stride;
+		outb->buf->datas[0].chunk->stride = outp->stride;
+	}
+	else {
+		buffer_size = outb->buf->datas[0].maxsize / outp->stride;
 		n->fill_func(outb->ptr, port->port_id, buffer_size, stride);
+	}
+
+	pw_log_trace("mix %p: layer %d %zd %d", node, layer, buffer_size, outp->stride);
 
 	return SPA_STATUS_HAVE_BUFFER;
 }
 
-static int schedule_mix_use_buffers(struct spa_node *_node,
-				    enum spa_direction direction,
-				    uint32_t port_id,
-				    struct spa_buffer **buffers,
-				    uint32_t n_buffers)
-{
-	struct pw_port *port = SPA_CONTAINER_OF(_node, struct pw_port, mix_node);
-	struct port *p = port->owner_data;
-	struct node *n = p->node;
-
-	pw_log_debug("port %p: %d use buffers %d %p", port, port_id, n_buffers, buffers);
-
-	if (n_buffers > 0) {
-		n->buffer_size = buffers[0]->datas[0].maxsize / p->stride;
-	}
-
-	return 0;
-}
-
-
 static const struct spa_node schedule_mix_node = {
 	SPA_VERSION_NODE,
 	NULL,
-	.port_use_buffers = schedule_mix_use_buffers,
 	.process = schedule_mix,
 };
 
@@ -886,8 +862,12 @@ static struct port *make_port(struct node *n, enum pw_direction direction,
 	return p;
 }
 
-static struct node *make_node(struct impl *impl, const struct pw_properties *props,
+struct pw_node *pw_audio_dsp_new(struct pw_core *core,
-		enum pw_direction direction)
+		const struct pw_properties *props,
+		enum pw_direction direction,
+		uint32_t channels,
+		uint32_t max_buffer_size,
+		size_t user_data_size)
 {
 	struct pw_node *node;
 	struct node *n;
@@ -903,7 +883,6 @@ static struct node *make_node(struct impl *impl, const struct pw_properties *pro
 	if ((alias = pw_properties_get(props, "device.name")) == NULL)
 		goto error;
 
-
 	snprintf(node_name, sizeof(node_name), "system_%s", alias);
 	for (i = 0; node_name[i]; i++) {
 		if (node_name[i] == ':' || node_name[i] == ',')
@@ -921,23 +900,30 @@ static struct node *make_node(struct impl *impl, const struct pw_properties *pro
 	if ((plugged = pw_properties_get(props, "node.plugged")) != NULL)
 		pw_properties_set(pr, "node.plugged", plugged);
 
-	node = pw_node_new(impl->core, node_name, pr, sizeof(struct node));
+	node = pw_node_new(core, node_name, pr, sizeof(struct node) + user_data_size);
         if (node == NULL)
 		goto error;
 
 	n = pw_node_get_user_data(node);
+	n->core = core;
+	n->t = pw_core_get_type(core);
+	init_type(&n->type, n->t->map);
 	n->node = node;
-	n->impl = impl;
 	n->node_impl = node_impl;
 	if (direction == PW_DIRECTION_OUTPUT)
 		n->node_impl.process = node_process_mix;
 	else
 		n->node_impl.process = node_process_split;
-	n->channels = DEFAULT_CHANNELS;
+
+	n->channels = channels;
 	n->sample_rate = DEFAULT_SAMPLE_RATE;
-	n->buffer_size = DEFAULT_BUFFER_SIZE;
+	n->max_buffer_size = max_buffer_size;
+
 	pw_node_set_implementation(node, &n->node_impl);
 
+	if (user_data_size > 0)
+		n->user_data = SPA_MEMBER(n, sizeof(struct node), void);
+
 	p = make_port(n, direction, 0, 0, NULL);
 	if (p == NULL)
 		goto error_free_node;
@@ -967,13 +953,7 @@ static struct node *make_node(struct impl *impl, const struct pw_properties *pro
 	        if (p == NULL)
 			goto error_free_node;
 	}
-
+	return node;
-	spa_list_append(&impl->node_list, &n->link);
-
-	pw_node_register(node, NULL, pw_module_get_global(impl->module), NULL);
-	pw_node_set_active(node, true);
-
-	return n;
 
      error_free_node:
 	pw_node_destroy(node);
@@ -981,141 +961,8 @@ static struct node *make_node(struct impl *impl, const struct pw_properties *pro
 	return NULL;
 }
 
-static void node_destroy(void *data)
+void *pw_audio_dsp_get_user_data(struct pw_node *node)
 {
-	struct node *node = data;
+	struct node *n = pw_node_get_user_data(node);
-
+	return n->user_data;
-	spa_list_remove(&node->link);
-	pw_node_destroy(node->node);
-}
-
-static const struct pw_node_events node_events = {
-	PW_VERSION_NODE_EVENTS,
-	.destroy = node_destroy,
-};
-
-static int on_global(void *data, struct pw_global *global)
-{
-	struct impl *impl = data;
-	struct pw_node *n;
-	struct node *node;
-	const struct pw_properties *properties;
-	const char *str;
-	char *error;
-	struct pw_port *ip, *op;
-	struct pw_link *link;
-
-
-	if (pw_global_get_type(global) != impl->t->node)
-		return 0;
-
-	n = pw_global_get_object(global);
-
-	properties = pw_node_get_properties(n);
-	if ((str = pw_properties_get(properties, "media.class")) == NULL)
-		return 0;
-
-	pw_log_debug("global added %s", str);
-
-	if (strcmp(str, "Audio/Sink") == 0) {
-		if ((ip = pw_node_get_free_port(n, PW_DIRECTION_INPUT)) == NULL)
-			return 0;
-		if ((node = make_node(impl, properties, PW_DIRECTION_OUTPUT)) == NULL)
-			return 0;
-		if ((op = pw_node_get_free_port(node->node, PW_DIRECTION_OUTPUT)) == NULL)
-			return 0;
-	}
-	else if (strcmp(str, "Audio/Source") == 0) {
-		if ((op = pw_node_get_free_port(n, PW_DIRECTION_OUTPUT)) == NULL)
-			return 0;
-		if ((node = make_node(impl, properties, PW_DIRECTION_INPUT)) == NULL)
-			return 0;
-		if ((ip = pw_node_get_free_port(node->node, PW_DIRECTION_INPUT)) == NULL)
-			return 0;
-	}
-	else
-		return 0;
-
-	link = pw_link_new(impl->core,
-			   op,
-			   ip,
-			   NULL,
-			   pw_properties_new(PW_LINK_PROP_PASSIVE, "true", NULL),
-			   &error, 0);
-	if (link == NULL) {
-		pw_log_error("can't create link: %s", error);
-		free(error);
-		return 0;
-	}
-	pw_link_register(link, NULL, pw_module_get_global(impl->module), NULL);
-
-	pw_node_add_listener(n, &node->node_listener, &node_events, node);
-
-	return 0;
-}
-
-static void module_destroy(void *data)
-{
-	struct impl *impl = data;
-	struct node *n, *t;
-
-	spa_hook_remove(&impl->module_listener);
-	spa_hook_remove(&impl->core_listener);
-
-	spa_list_for_each_safe(n, t, &impl->node_list, link)
-		pw_node_destroy(n->node);
-
-	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 void
-core_global_added(void *data, struct pw_global *global)
-{
-	on_global(data, global);
-}
-
-static const struct pw_core_events core_events = {
-	PW_VERSION_CORE_EVENTS,
-        .global_added = core_global_added,
-};
-
-static int module_init(struct pw_module *module, struct pw_properties *properties)
-{
-	struct pw_core *core = pw_module_get_core(module);
-	struct impl *impl;
-
-	impl = calloc(1, sizeof(struct impl));
-	if (impl == NULL)
-		return -ENOMEM;
-
-	pw_log_debug("module %p: new", impl);
-
-	impl->core = core;
-	impl->t = pw_core_get_type(core);
-	impl->module = module;
-	impl->properties = properties;
-
-	init_type(&impl->type, core->type.map);
-
-	spa_list_init(&impl->node_list);
-
-	pw_core_for_each_global(core, on_global, impl);
-
-	pw_core_add_listener(core, &impl->core_listener, &core_events, impl);
-	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
-
-	return 0;
-}
-
-int pipewire__module_init(struct pw_module *module, const char *args)
-{
-	return module_init(module, NULL);
 }

src/modules/module-audio-session/audio-dsp.h

@@ -0,0 +1,44 @@
+/* PipeWire
+ * Copyright (C) 2018 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __PIPEWIRE_AUDIO_DSP_H__
+#define __PIPEWIRE_AUDIO_DSP_H__
+
+#include <pipewire/core.h>
+#include <pipewire/node.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct pw_node *
+pw_audio_dsp_new(struct pw_core *core,
+		 const struct pw_properties *properties,
+		 enum pw_direction direction,
+		 uint32_t channels,
+		 uint32_t max_buffer_size,
+		 size_t user_data_size);
+
+void *pw_audio_dsp_get_user_data(struct pw_node *node);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __PIPEWIRE_AUDIO_DSP_H__ */