filter: add filter API

The filter API is meant for making generic audio and video filters.
It's like a pw_stream but with many in/out ports and no conversion.
Eventually the pw_stream will be implemented with the filter.

Add some example audio and video filters.

src/examples/audio-dsp-filter.c

@@ -0,0 +1,147 @@
+/* PipeWire
+ *
+ * Copyright © 2019 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdio.h>
+#include <errno.h>
+#include <time.h>
+#include <math.h>
+#include <sys/mman.h>
+
+#include <spa/param/audio/format-utils.h>
+#include <spa/param/props.h>
+
+#include <pipewire/pipewire.h>
+#include <pipewire/filter.h>
+
+struct data;
+
+struct port {
+	struct data *data;
+};
+
+struct data {
+	struct pw_main_loop *loop;
+	struct pw_filter *filter;
+	struct port *in_port;
+	struct port *out_port;
+};
+
+/* our data processing function is in general:
+ *
+ *  struct pw_buffer *b;
+ *  in = pw_filter_dequeue_buffer(filter, in_port);
+ *  out = pw_filter_dequeue_buffer(filter, out_port);
+ *
+ *  .. do stuff with buffers ...
+ *
+ *  pw_filter_queue_buffer(filter, in_port, in);
+ *  pw_filter_queue_buffer(filter, out_port, out);
+ *
+ *  For DSP ports, there is a shortcut to directly dequeue, get
+ *  the data and requeue the buffer with pw_filter_get_dsp_buffer().
+ *
+ *
+ */
+static void on_process(void *userdata, struct spa_io_position *position)
+{
+	struct data *data = userdata;
+	float *in, *out;
+	uint32_t n_samples = position->clock.duration;
+
+	pw_log_trace("do process %d", n_samples);
+
+	in = pw_filter_get_dsp_buffer(data->in_port, n_samples);
+	out = pw_filter_get_dsp_buffer(data->out_port, n_samples);
+
+	memcpy(out, in, n_samples * sizeof(float));
+}
+
+static const struct pw_filter_events filter_events = {
+	PW_VERSION_FILTER_EVENTS,
+	.process = on_process,
+};
+
+int main(int argc, char *argv[])
+{
+	struct data data = { 0, };
+
+	pw_init(&argc, &argv);
+
+	/* make a main loop. If you already have another main loop, you can add
+	 * the fd of this pipewire mainloop to it. */
+	data.loop = pw_main_loop_new(NULL);
+
+	/* Create a simple filter, the simple filter manages the core and remote
+	 * objects for you if you don't need to deal with them.
+	 *
+	 * Pass your events and a user_data pointer as the last arguments. This
+	 * will inform you about the filter state. The most important event
+	 * you need to listen to is the process event where you need to process
+	 * the data.
+	 */
+	data.filter = pw_filter_new_simple(
+			pw_main_loop_get_loop(data.loop),
+			"audio-filter",
+			pw_properties_new(
+				PW_KEY_MEDIA_TYPE, "Audio",
+				PW_KEY_MEDIA_CATEGORY, "Filter",
+				PW_KEY_MEDIA_ROLE, "DSP",
+				NULL),
+			&filter_events,
+			&data);
+
+	/* make an audio DSP input port */
+	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, "32 bit float mono audio",
+				PW_KEY_PORT_NAME, "input",
+				NULL),
+			NULL, 0);
+
+	/* make an audio DSP output port */
+	data.out_port = pw_filter_add_port(data.filter,
+			PW_DIRECTION_OUTPUT,
+			PW_FILTER_PORT_FLAG_MAP_BUFFERS,
+			sizeof(struct port),
+			pw_properties_new(
+				PW_KEY_FORMAT_DSP, "32 bit float mono audio",
+				PW_KEY_PORT_NAME, "output",
+				NULL),
+			NULL, 0);
+
+	/* Now connect this filter. We ask that our process function is
+	 * called in a realtime thread. */
+	pw_filter_connect(data.filter, PW_FILTER_FLAG_RT_PROCESS, NULL, 0);
+
+	/* and wait while we let things run */
+	pw_main_loop_run(data.loop);
+
+	pw_filter_destroy(data.filter);
+	pw_main_loop_destroy(data.loop);
+
+	return 0;
+}

src/examples/meson.build

@@ -23,6 +23,13 @@ executable('export-source',
   dependencies : [pipewire_dep, mathlib],
 )
 
+executable('audio-dsp-filter',
+  'audio-dsp-filter.c',
+  c_args : [ '-D_GNU_SOURCE' ],
+  install: false,
+  dependencies : [pipewire_dep, mathlib],
+)
+
 executable('export-spa',
   'export-spa.c',
   c_args : [ '-D_GNU_SOURCE' ],
@@ -58,6 +65,13 @@ if sdl_dep.found()
     install: false,
     dependencies : [pipewire_dep, sdl_dep, mathlib],
   )
+  executable('video-dsp-play',
+    'video-dsp-play.c',
+    c_args : [ '-D_GNU_SOURCE' ],
+    install: false,
+    dependencies : [pipewire_dep, sdl_dep, mathlib],
+  )
+
   executable('local-v4l2',
     'local-v4l2.c',
     c_args : [ '-D_GNU_SOURCE' ],

src/examples/sdl.h

@@ -77,7 +77,7 @@ static struct {
 #endif
 };
 
-static uint32_t sdl_format_to_id(Uint32 format)
+static inline uint32_t sdl_format_to_id(Uint32 format)
 {
 	size_t i;
 	for (i = 0; i < SPA_N_ELEMENTS(sdl_video_formats); i++) {
@@ -87,7 +87,7 @@ static uint32_t sdl_format_to_id(Uint32 format)
 	return SPA_VIDEO_FORMAT_UNKNOWN;
 }
 
-static Uint32 id_to_sdl_format(uint32_t id)
+static inline Uint32 id_to_sdl_format(uint32_t id)
 {
 	size_t i;
 	for (i = 0; i < SPA_N_ELEMENTS(sdl_video_formats); i++) {
@@ -98,7 +98,7 @@ static Uint32 id_to_sdl_format(uint32_t id)
 }
 
 
-static struct spa_pod *sdl_build_formats(SDL_RendererInfo *info, struct spa_pod_builder *b)
+static inline struct spa_pod *sdl_build_formats(SDL_RendererInfo *info, struct spa_pod_builder *b)
 {
 	uint32_t i, c;
 	struct spa_pod_frame f[2];

src/examples/video-dsp-play.c

@@ -0,0 +1,316 @@
+/* PipeWire
+ *
+ * Copyright © 2019 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/mman.h>
+
+#include <spa/param/video/format-utils.h>
+#include <spa/param/props.h>
+#include <spa/debug/format.h>
+
+#include <pipewire/pipewire.h>
+#include <pipewire/filter.h>
+
+#define WIDTH   640
+#define HEIGHT  480
+#define BPP    3
+
+#define MAX_BUFFERS	64
+
+#include "sdl.h"
+
+struct pixel {
+	float r, g, b, a;
+};
+
+struct data {
+	const char *path;
+
+	SDL_Renderer *renderer;
+	SDL_Window *window;
+	SDL_Texture *texture;
+	SDL_Texture *cursor;
+
+	struct pw_main_loop *loop;
+
+	struct pw_filter *filter;
+	struct spa_hook filter_listener;
+
+	void *in_port;
+
+	struct spa_video_info_raw format;
+	int32_t stride;
+
+	int counter;
+	SDL_Rect rect;
+	SDL_Rect cursor_rect;
+};
+
+static void handle_events(struct data *data)
+{
+	SDL_Event event;
+	while (SDL_PollEvent(&event)) {
+		switch (event.type) {
+		case SDL_QUIT:
+			pw_main_loop_quit(data->loop);
+			break;
+		}
+	}
+}
+
+/* our data processing function is in general:
+ *
+ *  struct pw_buffer *b;
+ *  b = pw_filter_dequeue_buffer(port);
+ *
+ *  .. do stuff with buffer ...
+ *
+ *  pw_filter_queue_buffer(port, b);
+ */
+static void
+on_process(void *_data, struct spa_io_position *position)
+{
+	struct data *data = _data;
+	struct pw_buffer *b;
+	struct spa_buffer *buf;
+	void *sdata, *ddata;
+	int sstride, dstride;
+	uint32_t i, j;
+	uint8_t *src, *dst;
+
+	b = pw_filter_dequeue_buffer(data->in_port);
+	if (b == NULL)
+		return;
+
+	buf = b->buffer;
+
+	pw_log_trace("new buffer %p %dx%d", buf, data->format.size.width, data->format.size.height);
+
+	handle_events(data);
+
+	if ((sdata = buf->datas[0].data) == NULL) {
+		pw_log_error("no buffer data");
+		goto done;
+	}
+
+	if (SDL_LockTexture(data->texture, NULL, &ddata, &dstride) < 0) {
+		pw_log_error("Couldn't lock texture: %s", SDL_GetError());
+		goto done;
+	}
+
+	/* copy video image in texture */
+	sstride = buf->datas[0].chunk->stride;
+
+	src = sdata;
+	dst = ddata;
+
+	for (i = 0; i < data->format.size.height; i++) {
+		struct pixel *p = (struct pixel *) src;
+		for (j = 0; j < data->format.size.width; j++) {
+			dst[j * 4 + 0] = SPA_CLAMP(lrintf(p[j].r * 255.0f), 0, 255);
+			dst[j * 4 + 1] = SPA_CLAMP(lrintf(p[j].g * 255.0f), 0, 255);
+			dst[j * 4 + 2] = SPA_CLAMP(lrintf(p[j].b * 255.0f), 0, 255);
+			dst[j * 4 + 3] = SPA_CLAMP(lrintf(p[j].a * 255.0f), 0, 255);
+		}
+		src += sstride;
+		dst += dstride;
+	}
+	SDL_UnlockTexture(data->texture);
+
+	SDL_RenderClear(data->renderer);
+	SDL_RenderCopy(data->renderer, data->texture, &data->rect, NULL);
+	SDL_RenderPresent(data->renderer);
+
+      done:
+	pw_filter_queue_buffer(data->in_port, b);
+}
+
+static void on_filter_state_changed(void *_data, enum pw_filter_state old,
+				    enum pw_filter_state state, const char *error)
+{
+	struct data *data = _data;
+	fprintf(stderr, "filter state: \"%s\"\n", pw_filter_state_as_string(state));
+	switch (state) {
+	case PW_FILTER_STATE_UNCONNECTED:
+		pw_main_loop_quit(data->loop);
+		break;
+	case PW_FILTER_STATE_PAUSED:
+		/* because we started inactive, activate ourselves now */
+		pw_filter_set_active(data->filter, true);
+		break;
+	default:
+		break;
+	}
+}
+
+static void
+on_filter_param_changed(void *_data, void *port_data, uint32_t id, const struct spa_pod *param)
+{
+	struct data *data = _data;
+	struct pw_filter *filter = data->filter;
+	uint8_t params_buffer[1024];
+	struct spa_pod_builder b = SPA_POD_BUILDER_INIT(params_buffer, sizeof(params_buffer));
+	const struct spa_pod *params[5];
+	Uint32 sdl_format;
+	void *d;
+
+	if (id != SPA_PARAM_Format)
+		return;
+
+	/* NULL means to clear the format */
+	if (param == NULL) {
+		pw_filter_update_params(filter, port_data, 0, NULL, 0);
+		return;
+	}
+
+	fprintf(stderr, "got format:\n");
+	spa_debug_format(2, NULL, param);
+
+	/* call a helper function to parse the format for us. */
+	spa_format_video_raw_parse(param, &data->format);
+
+	if (data->format.format == SPA_VIDEO_FORMAT_RGBA_F32)
+		sdl_format = SDL_PIXELFORMAT_RGBA32;
+	else
+		sdl_format = SDL_PIXELFORMAT_UNKNOWN;
+
+	if (sdl_format == SDL_PIXELFORMAT_UNKNOWN) {
+		pw_filter_update_params(filter, port_data, -EINVAL, NULL, 0);
+		return;
+	}
+
+	data->texture = SDL_CreateTexture(data->renderer,
+					  sdl_format,
+					  SDL_TEXTUREACCESS_STREAMING,
+					  data->format.size.width,
+					  data->format.size.height);
+	SDL_LockTexture(data->texture, NULL, &d, &data->stride);
+	SDL_UnlockTexture(data->texture);
+
+	data->rect.x = 0;
+	data->rect.y = 0;
+	data->rect.w = data->format.size.width;
+	data->rect.h = data->format.size.height;
+
+	/* a SPA_TYPE_OBJECT_ParamBuffers object defines the acceptable size,
+	 * number, stride etc of the buffers */
+	params[0] = spa_pod_builder_add_object(&b,
+		SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers,
+		SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(8, 2, MAX_BUFFERS),
+		SPA_PARAM_BUFFERS_blocks,  SPA_POD_Int(1),
+		SPA_PARAM_BUFFERS_size,    SPA_POD_Int(data->stride * data->format.size.height),
+		SPA_PARAM_BUFFERS_stride,  SPA_POD_Int(data->stride),
+		SPA_PARAM_BUFFERS_align,   SPA_POD_Int(16));
+
+	/* we are done */
+	pw_filter_update_params(filter, port_data, 0, params, 1);
+}
+
+/* these are the filter events we listen for */
+static const struct pw_filter_events filter_events = {
+	PW_VERSION_FILTER_EVENTS,
+	.state_changed = on_filter_state_changed,
+	.param_changed = on_filter_param_changed,
+	.process = on_process,
+};
+
+int main(int argc, char *argv[])
+{
+	struct data data = { 0, };
+	const struct spa_pod *params[1];
+
+	pw_init(&argc, &argv);
+
+	/* create a main loop */
+	data.loop = pw_main_loop_new(NULL);
+
+	/* create a simple filter, the simple filter manages to core and remote
+	 * objects for you if you don't need to deal with them
+	 *
+	 * If you plan to autoconnect your filter, you need to provide at least
+	 * media, category and role properties
+	 *
+	 * Pass your events and a user_data pointer as the last arguments. This
+	 * will inform you about the filter state. The most important event
+	 * you need to listen to is the process event where you need to consume
+	 * the data provided to you.
+	 */
+	data.filter = pw_filter_new_simple(
+			pw_main_loop_get_loop(data.loop),
+			"video-dsp-play",
+			pw_properties_new(
+				PW_KEY_MEDIA_TYPE, "Video",
+				PW_KEY_MEDIA_CATEGORY, "Capture",
+				PW_KEY_MEDIA_ROLE, "DSP",
+				NULL),
+			&filter_events,
+			&data);
+
+	data.path = argc > 1 ? argv[1] : NULL;
+
+	if (SDL_Init(SDL_INIT_VIDEO) < 0) {
+		fprintf(stderr, "can't initialize SDL: %s\n", SDL_GetError());
+		return -1;
+	}
+
+	if (SDL_CreateWindowAndRenderer
+	    (WIDTH, HEIGHT, SDL_WINDOW_RESIZABLE, &data.window, &data.renderer)) {
+		fprintf(stderr, "can't create window: %s\n", SDL_GetError());
+		return -1;
+	}
+
+	/* build the extra parameters to connect with. To connect, we can provide
+	 * a list of supported formats.  We use a builder that writes the param
+	 * object to the stack. */
+
+	data.in_port = pw_filter_add_port(data.filter,
+			PW_DIRECTION_INPUT,
+			PW_FILTER_PORT_FLAG_MAP_BUFFERS,
+			0,
+			pw_properties_new(
+				PW_KEY_FORMAT_DSP, "32 bit float RGBA video",
+				PW_KEY_PORT_NAME, "input",
+				NULL),
+			params, 1);
+
+	pw_filter_connect(data.filter,
+			0,
+			//PW_FILTER_FLAG_RT_PROCESS,
+			NULL, 0);
+
+	/* do things until we quit the mainloop */
+	pw_main_loop_run(data.loop);
+
+	pw_filter_destroy(data.filter);
+	pw_main_loop_destroy(data.loop);
+
+	SDL_DestroyTexture(data.texture);
+	if (data.cursor)
+		SDL_DestroyTexture(data.cursor);
+	SDL_DestroyRenderer(data.renderer);
+	SDL_DestroyWindow(data.window);
+
+	return 0;
+}

src/pipewire/filter.h

@@ -0,0 +1,235 @@
+/* PipeWire
+ *
+ * Copyright © 2019 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef PIPEWIRE_FILTER_H
+#define PIPEWIRE_FILTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** \class pw_filter
+ *
+ * \brief PipeWire filter object class
+ *
+ * The filter object provides a convenient way to implement
+ * processing filters.
+ *
+ * See also \ref page_filters and \ref page_core_api
+ */
+struct pw_filter;
+
+#include <spa/buffer/buffer.h>
+#include <spa/node/io.h>
+#include <spa/param/param.h>
+
+#include <pipewire/remote.h>
+
+/** \enum pw_filter_state The state of a filter \memberof pw_filter */
+enum pw_filter_state {
+	PW_FILTER_STATE_ERROR = -1,		/**< the strean is in error */
+	PW_FILTER_STATE_UNCONNECTED = 0,	/**< unconnected */
+	PW_FILTER_STATE_CONNECTING = 1,		/**< connection is in progress */
+	PW_FILTER_STATE_PAUSED = 2,		/**< filter is connected and paused */
+	PW_FILTER_STATE_STREAMING = 3		/**< filter is streaming */
+};
+
+#if 0
+struct pw_buffer {
+	struct spa_buffer *buffer;	/**< the spa buffer */
+	void *user_data;		/**< user data attached to the buffer */
+	uint64_t size;			/**< For input ports, this field is set by pw_filter
+					  *  with the duration of the buffer in ticks.
+					  *  For output ports, this field is set by the user.
+					  *  This field is added for all queued buffers and
+					  *  returned in the time info. */
+};
+#endif
+
+/** Events for a filter. These events are always called from the mainloop
+ * unless explicitly documented otherwise. */
+struct pw_filter_events {
+#define PW_VERSION_FILTER_EVENTS	0
+	uint32_t version;
+
+	void (*destroy) (void *data);
+	/** when the filter state changes */
+	void (*state_changed) (void *data, enum pw_filter_state old,
+				enum pw_filter_state state, const char *error);
+
+	/** when io changed on a port of the filter (when port_data is NULL). */
+	void (*io_changed) (void *data, void *port_data,
+			uint32_t id, void *area, uint32_t size);
+	/** when a parameter changed on a port of the filter (when port_data is NULL). */
+	void (*param_changed) (void *data, void *port_data,
+			uint32_t id, const struct spa_pod *param);
+
+        /** when a new buffer was created for a port */
+        void (*add_buffer) (void *data, void *port_data, struct pw_buffer *buffer);
+        /** when a buffer was destroyed for a port */
+        void (*remove_buffer) (void *data, void *port_data, struct pw_buffer *buffer);
+
+        /** do processing. This is normally called from the
+	 *  mainloop but can also be called directly from the realtime data
+	 *  thread if the user is prepared to deal with this. */
+        void (*process) (void *data, struct spa_io_position *position);
+
+	/** The filter is drained */
+        void (*drained) (void *data);
+};
+
+/** Convert a filter state to a readable string \memberof pw_filter */
+const char * pw_filter_state_as_string(enum pw_filter_state state);
+
+/** \enum pw_filter_flags Extra flags that can be used in \ref pw_filter_connect() \memberof pw_filter */
+enum pw_filter_flags {
+	PW_FILTER_FLAG_NONE = 0,			/**< no flags */
+	PW_FILTER_FLAG_INACTIVE		= (1 << 0),	/**< start the filter inactive,
+							  *  pw_filter_set_active() needs to be
+							  *  called explicitly */
+	PW_FILTER_FLAG_DRIVER		= (1 << 1),	/**< be a driver */
+	PW_FILTER_FLAG_RT_PROCESS	= (1 << 2),	/**< call process from the realtime
+							  *  thread */
+};
+
+enum pw_filter_port_flags {
+	PW_FILTER_PORT_FLAG_NONE		= 0,		/**< no flags */
+	PW_FILTER_PORT_FLAG_MAP_BUFFERS		= (1 << 0),	/**< mmap the buffers */
+	PW_FILTER_PORT_FLAG_ALLOC_BUFFERS	= (1 << 1),	/**< the application will allocate buffer
+								  *  memory. In the add_buffer event, the
+								  *  data of the buffer should be set */
+};
+
+/** Create a new unconneced \ref pw_filter \memberof pw_filter
+ * \return a newly allocated \ref pw_filter */
+struct pw_filter *
+pw_filter_new(struct pw_remote *remote,		/**< a \ref pw_remote */
+	      const char *name,			/**< a filter media name */
+	      struct pw_properties *props	/**< filter properties, ownership is taken */);
+
+struct pw_filter *
+pw_filter_new_simple(struct pw_loop *loop,		/**< a \ref pw_loop to use */
+		     const char *name,			/**< a filter media name */
+		     struct pw_properties *props,	/**< filter properties, ownership is taken */
+		     const struct pw_filter_events *events,	/**< filter events */
+		     void *data					/**< data passed to events */);
+
+/** Destroy a filter \memberof pw_filter */
+void pw_filter_destroy(struct pw_filter *filter);
+
+void pw_filter_add_listener(struct pw_filter *filter,
+			    struct spa_hook *listener,
+			    const struct pw_filter_events *events,
+			    void *data);
+
+enum pw_filter_state pw_filter_get_state(struct pw_filter *filter, const char **error);
+
+const char *pw_stream_get_name(struct pw_stream *stream);
+
+struct pw_remote *pw_filter_get_remote(struct pw_filter *filter);
+
+/** Connect a filter for processing. \memberof pw_filter
+ * \return 0 on success < 0 on error.
+ *
+ * You should connect to the process event and use pw_filter_dequeue_buffer()
+ * to get the latest metadata and data. */
+int
+pw_filter_connect(struct pw_filter *filter,		/**< a \ref pw_filter */
+		  enum pw_filter_flags flags,		/**< filter flags */
+		  const struct spa_pod **params,	/**< an array with params. */
+		  uint32_t n_params			/**< number of items in \a params */);
+
+/** Get the node ID of the filter. \memberof pw_filter
+ * \return node ID. */
+uint32_t
+pw_filter_get_node_id(struct pw_filter *filter);
+
+/** Disconnect \a filter \memberof pw_filter */
+int pw_filter_disconnect(struct pw_filter *filter);
+
+/** add a port to the filter, returns user data of port_data_size. */
+void *pw_filter_add_port(struct pw_filter *filter,
+		enum pw_direction direction,		/**< port direction */
+		enum pw_filter_port_flags flags,	/**< port flags */
+		size_t port_data_size,			/**< allocated and given to the user as port_data */
+		struct pw_properties *props,		/**< port properties, ownership is taken */
+		const struct spa_pod **params,		/**< an array of params. The params should
+							  *  ideally contain the supported formats */
+		uint32_t n_params			/**< number of elements in \a params */);
+
+/** remove a port from the filter */
+int pw_filter_remove_port(void *port_data		/**< data associated with port */);
+
+/** get properties, port_data of NULL will give global properties */
+const struct pw_properties *pw_filter_get_properties(struct pw_filter *filter,
+		void *port_data);
+
+/** Update properties, use NULL port_data for global filter properties */
+int pw_filter_update_properties(struct pw_filter *filter,
+		void *port_data, const struct spa_dict *dict);
+
+/** Update params, use NULL port_data for global filter params */
+int
+pw_filter_update_params(struct pw_filter *filter,	/**< a \ref pw_filter */
+			void *port_data,		/**< data associated with port */
+			int res,			/**< a result code */
+			const struct spa_pod **params,	/**< an array of params. */
+			uint32_t n_params		/**< number of elements in \a params */);
+
+
+#if 0
+/** A time structure \memberof pw_filter */
+struct pw_time {
+	int64_t now;			/**< the monotonic time */
+	struct spa_fraction rate;	/**< the rate of \a ticks and delay */
+	uint64_t ticks;			/**< the ticks at \a now. This is the current time that
+					  *  the remote end is reading/writing. */
+};
+#endif
+
+/** Query the time on the filter \memberof pw_filter */
+int pw_filter_get_time(struct pw_filter *filter, struct pw_time *time);
+
+/** Get a buffer that can be filled for output ports or consumed
+ * for input ports.  */
+struct pw_buffer *pw_filter_dequeue_buffer(void *port_data);
+
+/** Submit a buffer for playback or recycle a buffer for capture. */
+int pw_filter_queue_buffer(void *port_data, struct pw_buffer *buffer);
+
+/** Get a data pointer to the buffer data */
+void *pw_filter_get_dsp_buffer(void *port_data, uint32_t n_samples);
+
+/** Activate or deactivate the filter \memberof pw_filter */
+int pw_filter_set_active(struct pw_filter *filter, bool active);
+
+/** Flush a filter. When \a drain is true, the drained callback will
+ * be called when all data is played or recorded */
+int pw_filter_flush(struct pw_filter *filter, bool drain);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PIPEWIRE_FILTER_H */

src/pipewire/meson.build

@@ -6,6 +6,7 @@ pipewire_headers = [
   'core.h',
   'data-loop.h',
   'device.h',
+  'filter.h',
   'global.h',
   'interfaces.h',
   'introspect.h',
@@ -41,6 +42,7 @@ pipewire_sources = [
   'core.c',
   'data-loop.c',
   'device.c',
+  'filter.c',
   'global.c',
   'introspect.c',
   'link.c',

src/pipewire/private.h

@@ -39,6 +39,7 @@ extern "C" {
 #include "pipewire/introspect.h"
 #include "pipewire/interfaces.h"
 #include "pipewire/stream.h"
+#include "pipewire/filter.h"
 #include "pipewire/log.h"
 
 #include <spa/support/plugin.h>
@@ -733,6 +734,7 @@ struct pw_remote {
 	struct pw_client_proxy *client_proxy;	/**< proxy for the client object */
 
 	struct spa_list stream_list;		/**< list of \ref pw_stream objects */
+	struct spa_list filter_list;		/**< list of \ref pw_stream objects */
 
 	struct pw_protocol_client *conn;	/**< the protocol client connection */
 	int recv_seq;				/**< last received sequence number */
@@ -780,6 +782,40 @@ struct pw_stream {
 	struct spa_list controls;
 };
 
+#define pw_filter_emit(s,m,v,...) spa_hook_list_call(&(s)->listener_list, struct pw_filter_events, m, v, ##__VA_ARGS__)
+#define pw_filter_emit_destroy(s)		pw_filter_emit(s, destroy, 0)
+#define pw_filter_emit_state_changed(s,o,n,e)	pw_filter_emit(s, state_changed,0,o,n,e)
+#define pw_filter_emit_io_changed(s,p,i,d,t)	pw_filter_emit(s, io_changed,0,p,i,d,t)
+#define pw_filter_emit_param_changed(s,p,i,f)	pw_filter_emit(s, param_changed,0,p,i,f)
+#define pw_filter_emit_add_buffer(s,p,b)	pw_filter_emit(s, add_buffer, 0, p, b)
+#define pw_filter_emit_remove_buffer(s,p,b)	pw_filter_emit(s, remove_buffer, 0, p, b)
+#define pw_filter_emit_process(s,p)		pw_filter_emit(s, process, 0, p)
+#define pw_filter_emit_drained(s)		pw_filter_emit(s, drained, 0)
+
+
+struct pw_filter {
+	struct pw_remote *remote;		/**< the owner remote */
+	struct spa_list link;			/**< link in the remote */
+
+	char *name;				/**< the name of the filter */
+	struct pw_properties *properties;	/**< properties of the filter */
+
+	uint32_t node_id;			/**< node id for remote node, available from
+						  *  CONFIGURE state and higher */
+	enum pw_filter_state state;		/**< filter state */
+	char *error;				/**< error reason when state is in error */
+
+	struct spa_hook_list listener_list;
+
+	struct pw_proxy *proxy;
+	struct spa_hook proxy_listener;
+
+	struct pw_node_proxy *node;
+	struct spa_hook node_listener;
+
+	struct spa_list controls;
+};
+
 #define pw_factory_emit(s,m,v,...) spa_hook_list_call(&s->listener_list, struct pw_factory_events, m, v, ##__VA_ARGS__)
 
 #define pw_factory_emit_destroy(s)		pw_factory_emit(s, destroy, 0)

src/pipewire/remote.c

@@ -215,6 +215,7 @@ struct pw_remote *pw_remote_new(struct pw_core *core,
 	pw_map_init(&this->objects, 64, 32);
 
 	spa_list_init(&this->stream_list);
+	spa_list_init(&this->filter_list);
 
 	spa_hook_list_init(&this->listener_list);
 
@@ -272,6 +273,7 @@ void pw_remote_destroy(struct pw_remote *remote)
 {
 	struct remote *impl = SPA_CONTAINER_OF(remote, struct remote, this);
 	struct pw_stream *stream;
+	struct pw_filter *filter;
 
 	pw_log_debug(NAME" %p: destroy", remote);
 	pw_remote_emit_destroy(remote);
@@ -281,6 +283,8 @@ void pw_remote_destroy(struct pw_remote *remote)
 
 	spa_list_consume(stream, &remote->stream_list, link)
 		pw_stream_destroy(stream);
+	spa_list_consume(filter, &remote->filter_list, link)
+		pw_filter_destroy(filter);
 
 	pw_protocol_client_destroy(remote->conn);