simpler implementation

basically based on

the audio-src example from https://docs.pipewire.org/audio-src_8c-example.html

but completely removing the on_process callback

plugout_pipewire.c

@@ -34,58 +34,10 @@ static const struct timespec buffer_fill_wait_time = {
 static struct pipewire_data {
 	struct pw_thread_loop *loop;
 	struct pw_stream *stream;
-	void *buffer;;
-	volatile size_t buffer_filled;
-	volatile size_t buffer_sent;
 } pipewire_data;
 
-static void pipewire_on_process(void *userdata) {
-	struct pipewire_data *data = userdata;
-        struct pw_buffer *b;
-        struct spa_buffer *buf;
-        int n_frames;
-        uint8_t *p;
-	int frames_to_send;
-
-	// wait until data can be sent by us
-	while (pipewire_data.buffer_filled == 0) {
-//		fprintf(stderr, " >> pipewire_on_process: wait for buffer\n");
-		nanosleep(&buffer_fill_wait_time, NULL);
-	}
-	frames_to_send = (data->buffer_filled - data->buffer_sent) / STRIDE;
-
-        if ((b = pw_stream_dequeue_buffer(data->stream)) == NULL) {
-                pw_log_warn("out of buffers: %m");
-                return;
-        }
-
-	buf = b->buffer;
-        if ((p = buf->datas[0].data) == NULL)
-                return;
-
-	n_frames = buf->datas[0].maxsize / STRIDE;
-        if (b->requested)
-                n_frames = SPA_MIN((int)b->requested, n_frames);
-	n_frames = SPA_MIN(n_frames, frames_to_send);
-
-	memcpy(p, data->buffer + data->buffer_sent, n_frames * STRIDE);
-
-	buf->datas[0].chunk->offset = 0;
-        buf->datas[0].chunk->stride = STRIDE;
-        buf->datas[0].chunk->size = n_frames * STRIDE;
-
-	fprintf(stderr, " >> pipewire_on_process: send %d bytes\n", n_frames * STRIDE);
-
-        pw_stream_queue_buffer(data->stream, b);
-
-	data->buffer_sent += n_frames * STRIDE;
-	if (data->buffer_sent >= data->buffer_filled)
-		data->buffer_filled = 0;
-}
-
 static const struct pw_stream_events pipewire_stream_events = {
         PW_VERSION_STREAM_EVENTS,
-        .process = pipewire_on_process,
 };
 
 static long pipewire_open(enum plugout_endian *endian, long rate, long *buffer_bytes)
@@ -104,12 +56,6 @@ static long pipewire_open(enum plugout_endian *endian, long rate, long *buffer_b
 	default:                    fmt = SPA_AUDIO_FORMAT_S16_NE; break;
 	}
 
-	// allocate buffers
-	if ((pipewire_data.buffer = malloc(*buffer_bytes)) == NULL) {
-		fprintf(stderr, _("Memory allocation for pipewire buffer failed!"));
-		return -1;
-	}
-
 	// init pipewire
 	pw_init(0, NULL);
 
@@ -162,17 +108,52 @@ static long pipewire_open(enum plugout_endian *endian, long rate, long *buffer_b
 
 static ssize_t pipewire_write(const void *buf, size_t count)
 {
-	// wait until buffer is writeable by us
-	while (pipewire_data.buffer_filled != 0) {
-//		fprintf(stderr, " >> pipewire_write: wait for buffer\n");
-		nanosleep(&buffer_fill_wait_time, NULL);
-	}
-
-	fprintf(stderr, ">> pipewire_write: fill buffer with %ld bytes\n", count);
-	memcpy(pipewire_data.buffer, buf, count);
+        struct pw_buffer *b;
+        struct spa_buffer *spa_buf;
+        int n_frames;
+        uint8_t *p;
+	size_t buf_sent = 0;
+	size_t bytes_to_send;
+
+	// repeat until the whole buffer is sent
+	while (buf_sent < count) {
+
+		const int frames_to_send = (count - buf_sent) / STRIDE;
+
+		// wait until data can be sent by us
+		if ((b = pw_stream_dequeue_buffer(pipewire_data.stream)) == NULL) {
+			pw_log_warn("out of buffers: %m");
+			while ((b = pw_stream_dequeue_buffer(pipewire_data.stream)) == NULL) {
+				nanosleep(&buffer_fill_wait_time, NULL);
+			}
+		}
+
+		// get send buffer
+		spa_buf = b->buffer;
+		if ((p = spa_buf->datas[0].data) == NULL) {
+			nanosleep(&buffer_fill_wait_time, NULL);
+			break;
+		}
+
+		// check how much we can send
+		n_frames = spa_buf->datas[0].maxsize / STRIDE;
+		if (b->requested)
+			n_frames = SPA_MIN((int)b->requested, n_frames);
+		n_frames = SPA_MIN(n_frames, frames_to_send);
+		bytes_to_send = n_frames * STRIDE;
+
+		// send audio data
+		memcpy(p, buf + buf_sent, bytes_to_send);
+
+		spa_buf->datas[0].chunk->offset = 0;
+		spa_buf->datas[0].chunk->stride = STRIDE;
+		spa_buf->datas[0].chunk->size = bytes_to_send;
+
+		pw_stream_queue_buffer(pipewire_data.stream, b);
 
-	pipewire_data.buffer_sent = 0;
-	pipewire_data.buffer_filled = count;
+		// remember what was sent
+		buf_sent += bytes_to_send;
+	}
 
 	return 0;
 }
@@ -185,13 +166,11 @@ static void pipewire_close()
 	pw_stream_destroy(pipewire_data.stream);
         pw_thread_loop_destroy(pipewire_data.loop);
 	pw_deinit();
-
-	free(pipewire_data.buffer);
 }
 
 const struct output_plugin plugout_pipewire = {
 	.name = "pipewire",
-	.description = "PIPEWIRE sound driver",
+	.description = "PipeWire sound driver",
 	.open = pipewire_open,
 	.write = pipewire_write,
 	.close = pipewire_close,