Add feedback port that sends out the filters control state when parameters change

src/modules/module-filter-chain.c

@@ -19,10 +19,13 @@
 #include <spa/param/tag-utils.h>
 #include <spa/param/audio/raw-json.h>
 #include <spa/pod/dynamic.h>
+#include <spa/utils/ringbuffer.h>
 #include <spa/filter-graph/filter-graph.h>
 
 #include <pipewire/impl.h>
 
+#include "pipewire/extensions/session-manager/interfaces.h"
+
 
 #define NAME "filter-chain"
 
@@ -1185,6 +1188,8 @@ static const struct spa_dict_item module_props[] = {
 #include <pipewire/pipewire.h>
 
 #define DEFAULT_RATE	48000
+#define DEFAULT_FEEDBACK_BUFFER_SIZE	49152
+#define DEFAULT_FEEDBACK_MAX_MESSAGE_SIZE	4096;
 
 struct impl {
 	struct pw_context *context;
@@ -1213,6 +1218,16 @@ struct impl {
 	struct spa_hook control_listener;
 	bool control_stream_active;
 
+	struct pw_properties *feedback_props;
+	struct pw_stream *feedback;
+	struct spa_hook feedback_listener;
+	bool feedback_stream_active;
+
+	struct spa_ringbuffer feedback_ringbuffer;
+	uint32_t feedback_ringbuffer_size;
+	uint32_t feedback_ringbuffer_max_message_size;
+	void *feedback_ringbuffer_data;
+
 	struct spa_audio_info_raw info;
 
 	struct spa_io_position *position;
@@ -1279,7 +1294,7 @@ static void control_process(void *d)
 
 	struct pw_buffer *control = pw_stream_dequeue_buffer(impl->control);
 	if (control == NULL)
-	  	goto done;
+		return;
 	if (control->buffer->n_datas > 0) {
 	  	struct spa_pod *pod = spa_pod_from_data(
 	    		control->buffer->datas[0].data,
@@ -1294,10 +1309,93 @@ static void control_process(void *d)
 		}
 	}
 
-done:
 	pw_stream_queue_buffer(impl->control, control);
 }
 
+static void feedback_process(void *d) {
+	struct impl *impl = d;
+
+	struct pw_buffer *feedback = pw_stream_dequeue_buffer(impl->feedback);
+	if (feedback == NULL) {
+		pw_log_debug("%p: out of control buffers: %m", &impl);
+		return;
+	}
+
+	struct spa_buffer *spa_buffer = feedback->buffer;
+	if (!spa_buffer || spa_buffer->n_datas < 1) {
+		pw_log_warn("control buffer has no data planes");
+		return;
+	}
+
+	struct spa_data *data = &spa_buffer->datas[0];
+	if (!data->data || !data->chunk || data->maxsize == 0) {
+		pw_log_warn("invalid control buffer data");
+		return;
+	}
+
+	struct spa_pod_builder b;
+	spa_pod_builder_init(&b, data->data, data->maxsize);
+
+	struct spa_pod_frame seq_f;
+	spa_pod_builder_push_sequence(&b, &seq_f, 0);
+
+	uint8_t message_buffer[4096];
+	uint32_t index;
+	while (spa_ringbuffer_get_read_index(&impl->feedback_ringbuffer, &index) > 0) {
+		fprintf(stdout, "reading from ringbuffer index: %d\n", index);
+
+		uint64_t pod_data_size;
+		spa_ringbuffer_read_data(
+			&impl->feedback_ringbuffer,
+			impl->feedback_ringbuffer_data,
+			impl->feedback_ringbuffer_size,
+			index % impl->feedback_ringbuffer_size,
+			&pod_data_size,
+			sizeof(uint64_t));
+
+		struct spa_pod_builder_state state;
+		spa_pod_builder_get_state(&b, &state);
+		fprintf(stdout, "pod_data_size: %lu\n", pod_data_size);
+		fprintf(stdout, "builder size:%u maxsize:%d offset:%d\n",
+		        b.size, data->maxsize, state.offset);
+
+		// next control size: round_up_8(sizeof(struct spa_pod_control) + data_size)
+		/*
+		if (pod_data_size + sizeof(struct spa_pod_control) + sizeof(
+			struct spa_pod_sequence) > b.size - state.offset) {
+			fprintf(stdout, "buffer too small\n");
+			pw_log_warn("not enough space in builder for pod data");
+			break;
+		}
+		 */
+
+		spa_ringbuffer_read_data(
+			&impl->feedback_ringbuffer,
+			impl->feedback_ringbuffer_data,
+			impl->feedback_ringbuffer_size,
+			(index + sizeof(uint64_t)) % impl->feedback_ringbuffer_size,
+			&message_buffer,
+			pod_data_size);
+
+		spa_ringbuffer_read_update(&impl->feedback_ringbuffer,
+			index + sizeof(uint64_t) + pod_data_size);
+
+		fprintf(stdout, "adding control data\n");
+		spa_pod_builder_control(&b, 0, SPA_CONTROL_Properties);
+		spa_pod_builder_raw(&b, message_buffer, pod_data_size);
+
+		fprintf(stdout, "builder size:%u maxsize:%d offset:%d\n",
+			b.size, data->maxsize, state.offset);
+	}
+
+	const struct spa_sequence *spa_sequence = spa_pod_builder_pop(&b, &seq_f);
+	const uint32_t seq_size = SPA_POD_SIZE(spa_sequence);
+	data->chunk->offset = 0;
+	data->chunk->size = seq_size;
+	data->chunk->stride = 0;
+	pw_stream_queue_buffer(impl->feedback, feedback);
+}
+
 static void playback_process(void *d)
 {
 	struct impl *impl = d;
@@ -1692,6 +1790,41 @@ static const struct pw_stream_events control_stream_events = {
 	.param_changed = NULL,
 };
 
+static void feedback_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->feedback_listener);
+	impl->feedback = NULL;
+}
+
+static void feedback_state_changed(void *data, enum pw_stream_state old,
+				enum pw_stream_state state, const char *error)
+{
+	struct impl *impl = data;
+
+	switch (state) {
+	case PW_STREAM_STATE_PAUSED:
+		pw_stream_flush(impl->feedback, false);
+		break;
+	case PW_STREAM_STATE_ERROR:
+		pw_log_info("module %p: error: %s", impl, error);
+		break;
+	case PW_STREAM_STATE_STREAMING:
+	default:
+		break;
+	}
+	return;
+}
+
+static const struct  pw_stream_events feedback_stream_events = {
+	.version = PW_VERSION_STREAM_EVENTS,
+	.destroy = feedback_destroy,
+	.process = feedback_process,
+	.io_changed = NULL,
+	.state_changed = feedback_state_changed,
+	.param_changed = NULL,
+};
+
 static int setup_streams(struct impl *impl)
 {
 	int res;
@@ -1734,6 +1867,39 @@ static int setup_streams(struct impl *impl)
 			param, 1);
 	}
 
+	if (impl->feedback_stream_active) {
+		impl->feedback = pw_stream_new(impl->core,
+		    "filter feedback", impl->feedback_props);
+		impl->feedback_props = NULL;
+		if (impl->feedback == NULL)
+		  return -errno;
+
+		pw_stream_add_listener(impl->feedback,
+			&impl->feedback_listener,
+			&feedback_stream_events, impl);
+
+		uint8_t buffer[256];
+		struct spa_pod_builder bt = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
+		const struct spa_pod *param[1];
+
+		param[0] = spa_pod_builder_add_object(&bt,
+			SPA_TYPE_OBJECT_Format,
+			SPA_PARAM_EnumFormat,
+			SPA_FORMAT_mediaType,
+			SPA_POD_Id(SPA_MEDIA_TYPE_application),
+			SPA_FORMAT_mediaSubtype,
+			SPA_POD_Id(SPA_MEDIA_SUBTYPE_control)
+		);
+
+		pw_stream_connect(impl->feedback,
+			PW_DIRECTION_OUTPUT,
+			PW_ID_ANY,
+			PW_STREAM_FLAG_AUTOCONNECT |
+			PW_STREAM_FLAG_MAP_BUFFERS |
+			PW_STREAM_FLAG_RT_PROCESS,
+			param, 1);
+	}
+
 	impl->capture = pw_stream_new(impl->core,
 			"filter capture", impl->capture_props);
 	impl->capture_props = NULL;
@@ -1895,6 +2061,39 @@ static void graph_props_changed(void *object, enum spa_direction direction)
 	spa_filter_graph_get_props(graph, &b.b, (struct spa_pod **)&params[0]);
 
 	pw_stream_update_params(impl->capture, params, 1);
+
+	fprintf(stdout, "graph props changed\n");
+	uint32_t index;
+	const uint32_t available = impl->feedback_ringbuffer_size
+		- spa_ringbuffer_get_write_index(&impl->feedback_ringbuffer, &index);
+	if (available >= impl->feedback_ringbuffer_max_message_size) {
+		fprintf(stdout, "writing to buffer\n");
+		const uint64_t pod_length = SPA_POD_SIZE(params[0]);
+		const uint64_t full_length = pod_length + sizeof(uint64_t);
+		if (full_length > impl->feedback_ringbuffer_max_message_size) {
+			fprintf(stdout, "feedback ringbuffer full\n");
+			pw_log_warn("feedback ringbuffer message too large");
+		} else {
+			fprintf(stdout, "writing data, pod length: %lu\nindex: %u\n",
+				pod_length, index);
+
+			spa_ringbuffer_write_data(&impl->feedback_ringbuffer,
+				impl->feedback_ringbuffer_data, impl->feedback_ringbuffer_size,
+				index % impl->feedback_ringbuffer_size, &pod_length,
+				sizeof(uint64_t));
+
+			spa_ringbuffer_write_data(&impl->feedback_ringbuffer,
+				impl->feedback_ringbuffer_data, impl->feedback_ringbuffer_size,
+				(index + sizeof(uint64_t)) % impl->feedback_ringbuffer_size, params[0],
+				SPA_POD_SIZE(params[0]));
+
+			spa_ringbuffer_write_update(&impl->feedback_ringbuffer,
+				index + sizeof(uint64_t) + SPA_POD_SIZE(params[0]));
+		}
+	} else {
+		fprintf(stdout, "not enough space in ringbuffer\n");
+	}
+
 	spa_pod_dynamic_builder_clean(&b);
 }
 
@@ -2039,7 +2238,9 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
 	impl->capture_props = pw_properties_new(NULL, NULL);
 	impl->playback_props = pw_properties_new(NULL, NULL);
 	impl->control_props = pw_properties_new(NULL, NULL);
-	if (impl->capture_props == NULL || impl->playback_props == NULL || impl->control_props == NULL) {
+	impl->feedback_props = pw_properties_new(NULL, NULL);
+	if (impl->capture_props == NULL || impl->playback_props == NULL ||
+	    impl->control_props == NULL || impl->feedback_props == NULL) {
 		res = -errno;
 		pw_log_error( "can't create properties: %m");
 		goto error;
@@ -2067,6 +2268,8 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
 		pw_properties_update_string(impl->playback_props, str, strlen(str));
 	if ((str = pw_properties_get(props, "control.props")) != NULL)
 		pw_properties_update_string(impl->control_props, str, strlen(str));
+	if ((str = pw_properties_get(props, "feedback.props")) != NULL)
+		pw_properties_update_string(impl->feedback_props, str, strlen(str));
 
 	copy_props(impl, props, PW_KEY_AUDIO_RATE);
 	copy_props(impl, props, PW_KEY_AUDIO_CHANNELS);
@@ -2165,6 +2368,14 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
 			&core_events, impl);
 
 	impl->control_stream_active = pw_properties_get_bool(impl->control_props, "enabled", false);
+	impl->feedback_stream_active = pw_properties_get_bool(impl->feedback_props, "enabled", false);
+
+	if (impl->feedback_stream_active) {
+		spa_ringbuffer_init(&impl->feedback_ringbuffer);
+		impl->feedback_ringbuffer_data = malloc(DEFAULT_FEEDBACK_BUFFER_SIZE);
+		impl->feedback_ringbuffer_size = DEFAULT_FEEDBACK_BUFFER_SIZE;
+		impl->feedback_ringbuffer_max_message_size = DEFAULT_FEEDBACK_MAX_MESSAGE_SIZE;
+	}
 
 	setup_streams(impl);
 

src/tools/meson.build

@@ -9,6 +9,7 @@ tools_sources = [
   [ 'pw-metadata', [ 'pw-metadata.c' ] ],
   [ 'pw-loopback', [ 'pw-loopback.c' ] ],
   [ 'pw-link', [ 'pw-link.c' ] ],
+  [ 'pw-ctrldump', [ 'pw-ctrldump.c' ] ],
 ]
 
 foreach t : tools_sources

src/tools/pw-ctrldump.c

@@ -0,0 +1,190 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2020 Wim Taymans */
+/* SPDX-License-Identifier: MIT */
+
+#include <stdio.h>
+#include <signal.h>
+#include <math.h>
+#include <getopt.h>
+#include <locale.h>
+
+#include <spa/utils/result.h>
+#include <spa/utils/defs.h>
+#include <spa/control/control.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/param/props.h>
+#include <spa/debug/mem.h>
+
+#include <pipewire/pipewire.h>
+#include <pipewire/filter.h>
+
+#include "midifile.h"
+
+struct data;
+
+struct port {
+	struct data *data;
+};
+
+struct data {
+	struct pw_main_loop *loop;
+	const char *opt_remote;
+	struct pw_filter *filter;
+	struct port *in_port;
+	int64_t clock_time;
+};
+
+static void on_process(void *_data, struct spa_io_position *position)
+{
+	struct data *data = _data;
+	struct pw_buffer *b;
+	struct spa_buffer *buf;
+	struct spa_data *d;
+	struct spa_pod_parser parser;
+	struct spa_pod_frame frame;
+	struct spa_pod_sequence seq;
+	const void *seq_body, *c_body;
+	struct spa_pod_control c;
+	uint64_t offset;
+
+	offset = data->clock_time;
+	data->clock_time += position->clock.duration;
+
+	b = pw_filter_dequeue_buffer(data->in_port);
+	if (b == NULL)
+		return;
+
+	buf = b->buffer;
+	d = &buf->datas[0];
+
+	if (d->data == NULL)
+		goto done;
+
+	spa_pod_parser_init_from_data(&parser, d->data, d->maxsize, d->chunk->offset, d->chunk->size);
+
+	if (spa_pod_parser_push_sequence_body(&parser, &frame, &seq, &seq_body) < 0)
+		goto done;
+
+	while (spa_pod_parser_get_control_body(&parser, &c, &c_body) >= 0) {
+		struct midi_event ev;
+
+		if (c.type != SPA_CONTROL_Properties)
+			continue;
+
+		fprintf(stdout, "received message");
+	}
+
+done:
+	pw_filter_queue_buffer(data->in_port, b);
+}
+
+static const struct pw_filter_events filter_events = {
+	PW_VERSION_FILTER_EVENTS,
+	.process = on_process,
+};
+
+static void do_quit(void *userdata, int signal_number)
+{
+	struct data *data = userdata;
+	pw_main_loop_quit(data->loop);
+}
+
+static int dump_filter(struct data *data)
+{
+	data->loop = pw_main_loop_new(NULL);
+	if (data->loop == NULL)
+		return -errno;
+
+	pw_loop_add_signal(pw_main_loop_get_loop(data->loop), SIGINT, do_quit, data);
+	pw_loop_add_signal(pw_main_loop_get_loop(data->loop), SIGTERM, do_quit, data);
+
+	data->filter = pw_filter_new_simple(
+			pw_main_loop_get_loop(data->loop),
+			"ctrl-dump",
+			pw_properties_new(
+				PW_KEY_REMOTE_NAME, data->opt_remote,
+				PW_KEY_MEDIA_TYPE, "Control",
+				PW_KEY_MEDIA_CATEGORY, "Filter",
+				PW_KEY_MEDIA_ROLE, "DSP",
+				NULL),
+			&filter_events,
+			data);
+
+	data->in_port = pw_filter_add_port(data->filter,
+			PW_DIRECTION_INPUT,
+			PW_FILTER_PORT_FLAG_MAP_BUFFERS,
+			sizeof(struct port),
+			pw_properties_new(
+				PW_KEY_FORMAT_DSP, "8 bit raw",
+				PW_KEY_PORT_NAME, "input",
+				NULL),
+			NULL, 0);
+
+	if (pw_filter_connect(data->filter, PW_FILTER_FLAG_RT_PROCESS, NULL, 0) < 0) {
+		fprintf(stderr, "can't connect\n");
+		return -1;
+	}
+
+	pw_main_loop_run(data->loop);
+
+	pw_filter_destroy(data->filter);
+	pw_main_loop_destroy(data->loop);
+
+	return 0;
+}
+
+static void show_help(const char *name, bool error)
+{
+	fprintf(error ? stderr : stdout, "%s [options] [FILE]\n"
+		"  -h, --help                            Show this help\n"
+		"      --version                         Show version\n"
+		"  -r, --remote                          Remote daemon name\n",
+		name);
+}
+
+int main(int argc, char *argv[])
+{
+	struct data data = { 0, };
+	int res = 0, c;
+	static const struct option long_options[] = {
+		{ "help",	no_argument,		NULL, 'h' },
+		{ "version",	no_argument,		NULL, 'V' },
+		{ "remote",	required_argument,	NULL, 'r' },
+		{ NULL,	0, NULL, 0}
+	};
+
+	setlocale(LC_ALL, "");
+	pw_init(&argc, &argv);
+
+	setlinebuf(stdout);
+
+	while ((c = getopt_long(argc, argv, "hVr:", long_options, NULL)) != -1) {
+		switch (c) {
+		case 'h':
+			show_help(argv[0], false);
+			return 0;
+		case 'V':
+			printf("%s\n"
+				"Compiled with libpipewire %s\n"
+				"Linked with libpipewire %s\n",
+				argv[0],
+				pw_get_headers_version(),
+				pw_get_library_version());
+			return 0;
+		case 'r':
+			data.opt_remote = optarg;
+			break;
+		default:
+			show_help(argv[0], true);
+			return -1;
+		}
+	}
+
+	if (optind < argc) {
+		fprintf(stderr, "write to file not possible\n");
+	} else {
+		res = dump_filter(&data);
+	}
+	pw_deinit();
+	return res;
+}