connection: add test and fix a bug

Add a unit test for the connection

When we consumed all packets in a buffer, try to get more data instead
of failing right away.

src/modules/meson.build

@@ -73,6 +73,15 @@ pipewire_module_protocol_native = shared_library('pipewire-module-protocol-nativ
   dependencies : pipewire_module_protocol_native_deps,
 )
 
+test('pw-test-protocol-native',
+	executable('pw-test-protocol-native',
+		[ 'module-protocol-native/test-connection.c',
+		  'module-protocol-native/connection.c' ],
+			c_args : libpipewire_c_args,
+			include_directories : [configinc, spa_inc ],
+			dependencies : [pipewire_dep],
+			install : false))
+
 pipewire_module_audio_dsp = shared_library('pipewire-module-audio-dsp',
   [ 'module-audio-dsp.c',
     'module-audio-dsp/audio-dsp.c',

src/modules/module-protocol-native/connection.c

@@ -61,14 +61,14 @@ struct buffer {
 
 struct impl {
 	struct pw_protocol_native_connection this;
+	struct pw_core *core;
 
 	struct buffer in, out;
 
 	uint32_t dest_id;
 	uint8_t opcode;
-	struct spa_pod_builder builder;
-	struct pw_core *core;
 	uint32_t seq;
+	struct spa_pod_builder builder;
 };
 
 /** \endcond */
@@ -128,7 +128,9 @@ static void *connection_ensure_size(struct pw_protocol_native_connection *conn,
 		buf->buffer_data = realloc(buf->buffer_data, buf->buffer_maxsize);
 		if (buf->buffer_data == NULL) {
 			buf->buffer_maxsize = 0;
-			spa_hook_list_call(&conn->listener_list, struct pw_protocol_native_connection_events, error, 0, -ENOMEM);
+			spa_hook_list_call(&conn->listener_list,
+					struct pw_protocol_native_connection_events,
+					error, 0, -ENOMEM);
 			return NULL;
 		}
 		pw_log_warn("connection %p: resize buffer to %zd %zd %zd",
@@ -214,6 +216,7 @@ struct pw_protocol_native_connection *pw_protocol_native_connection_new(struct p
 		return NULL;
 
 	debug_messages = pw_debug_is_category_enabled("connection");
+	impl->core = core;
 
 	this = &impl->this;
 
@@ -227,7 +230,6 @@ struct pw_protocol_native_connection *pw_protocol_native_connection_new(struct p
 	impl->in.buffer_data = calloc(1, MAX_BUFFER_SIZE);
 	impl->in.buffer_maxsize = MAX_BUFFER_SIZE;
 	impl->in.update = true;
-	impl->core = core;
 
 	if (impl->out.buffer_data == NULL || impl->in.buffer_data == NULL)
 		goto no_mem;
@@ -307,7 +309,7 @@ pw_protocol_native_connection_get_next(struct pw_protocol_native_connection *con
 	if (buf->offset >= size) {
 		clear_buffer(buf);
 		buf->update = true;
-		return false;
+		goto again;
 	}
 
 	data += buf->offset;

src/modules/module-protocol-native/test-connection.c

@@ -0,0 +1,141 @@
+/* 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 <spa/pod/parser.h>
+
+#include <pipewire/pipewire.h>
+
+#include "connection.h"
+
+static void test_create(struct pw_protocol_native_connection *conn)
+{
+	uint8_t opcode;
+	uint32_t dest_id, size;
+	void *data;
+	int res, seq;
+
+	res = pw_protocol_native_connection_get_next(conn,
+			&opcode, &dest_id, &data, &size, &seq);
+	spa_assert(res == false);
+
+	res = pw_protocol_native_connection_get_fd(conn, 0);
+	spa_assert(res == -1);
+
+	res = pw_protocol_native_connection_flush(conn);
+	spa_assert(res == 0);
+
+	res = pw_protocol_native_connection_clear(conn);
+	spa_assert(res == 0);
+}
+
+static void write_message(struct pw_protocol_native_connection *conn, int fd)
+{
+	struct spa_pod_builder *b;
+	int seq = -1, res;
+
+	b = pw_protocol_native_connection_begin(conn, 1, 5, &seq);
+	spa_assert(b != NULL);
+	spa_assert(seq != -1);
+
+	spa_pod_builder_add_struct(b,
+			SPA_POD_Int(42),
+			SPA_POD_Id(SPA_TYPE_Object),
+			SPA_POD_Int(pw_protocol_native_connection_add_fd(conn, fd)));
+
+	res = pw_protocol_native_connection_end(conn, b);
+	spa_assert(seq == res);
+}
+
+static int read_message(struct pw_protocol_native_connection *conn)
+{
+        struct spa_pod_parser prs;
+	uint8_t opcode;
+	uint32_t dest_id, size;
+	void *data;
+	int res, seq, fd;
+	uint32_t v_int, v_id, fdidx;
+
+	res = pw_protocol_native_connection_get_next(conn,
+			&opcode, &dest_id, &data, &size, &seq);
+	if (!res)
+		return -1;
+
+	spa_assert(opcode == 5);
+	spa_assert(dest_id == 1);
+	spa_assert(data != NULL);
+	spa_assert(size > 0);
+
+	spa_pod_parser_init(&prs, data, size);
+	if (spa_pod_parser_get_struct(&prs,
+                        SPA_POD_Int(&v_int),
+                        SPA_POD_Id(&v_id),
+                        SPA_POD_Int(&fdidx)) < 0)
+                spa_assert_not_reached();
+
+	fd = pw_protocol_native_connection_get_fd(conn, fdidx);
+	pw_log_debug("got fd %d", fd);
+	spa_assert(fd != -1);
+	return 0;
+}
+
+static void test_read_write(struct pw_protocol_native_connection *in,
+		struct pw_protocol_native_connection *out)
+{
+	write_message(out, 1);
+	pw_protocol_native_connection_flush(out);
+	write_message(out, 2);
+	pw_protocol_native_connection_flush(out);
+	spa_assert(read_message(in) == 0);
+	spa_assert(read_message(in) == 0);
+	spa_assert(read_message(in) == -1);
+}
+
+int main(int argc, char *argv[])
+{
+	struct pw_main_loop *loop;
+	struct pw_core *core;
+	struct pw_protocol_native_connection *in, *out;
+	int fds[2];
+
+	pw_init(&argc, &argv);
+
+	loop = pw_main_loop_new(NULL);
+	core = pw_core_new(pw_main_loop_get_loop(loop), NULL, 0);
+
+	if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) {
+		spa_assert_not_reached();
+		return -1;
+	}
+
+	in = pw_protocol_native_connection_new(core, fds[0]);
+	spa_assert(in != NULL);
+	out = pw_protocol_native_connection_new(core, fds[1]);
+	spa_assert(out != NULL);
+
+	test_create(in);
+	test_create(out);
+	test_read_write(in, out);
+
+	return 0;
+}