input_stream: non-blocking I/O

Add GMutex, GCond attributes which will be used by callers to
conditionally wait on the stream.

Remove the (now-useless) plugin method buffer(), wait on GCond
instead.  Lock the input_stream before each method call.  Do the same
with the playlist plugins.

1 files changed, 18 insertions, 12 deletions

diff --git a/test/run_input.c b/test/run_input.c
index 651d36480..0fe5a01f0 100644
--- a/test/run_input.c
+++ b/test/run_input.c
@@ -53,20 +53,17 @@ dump_input_stream(struct input_stream *is)
 	size_t num_read;
 	ssize_t num_written;
 
+	g_mutex_lock(is->mutex);
+
 	/* wait until the stream becomes ready */
 
-	while (!is->ready) {
-		int ret = input_stream_buffer(is, &error);
-		if (ret < 0) {
-			/* error */
-			g_warning("%s", error->message);
-			g_error_free(error);
-			return 2;
-		}
+	input_stream_wait_ready(is);
 
-		if (ret == 0)
-			/* nothing was buffered - wait */
-			g_usleep(10000);
+	if (!input_stream_check(is, &error)) {
+		g_warning("%s", error->message);
+		g_error_free(error);
+		g_mutex_unlock(is->mutex);
+		return EXIT_FAILURE;
 	}
 
 	/* print meta data */
@@ -103,9 +100,12 @@ dump_input_stream(struct input_stream *is)
 	if (!input_stream_check(is, &error)) {
 		g_warning("%s", error->message);
 		g_error_free(error);
+		g_mutex_unlock(is->mutex);
 		return EXIT_FAILURE;
 	}
 
+	g_mutex_unlock(is->mutex);
+
 	return 0;
 }
 
@@ -149,7 +149,10 @@ int main(int argc, char **argv)
 
 	/* open the stream and dump it */
 
-	is = input_stream_open(argv[1], &error);
+	GMutex *mutex = g_mutex_new();
+	GCond *cond = g_cond_new();
+
+	is = input_stream_open(argv[1], mutex, cond, &error);
 	if (is != NULL) {
 		ret = dump_input_stream(is);
 		input_stream_close(is);
@@ -162,6 +165,9 @@ int main(int argc, char **argv)
 		ret = 2;
 	}
 
+	g_cond_free(cond);
+	g_mutex_free(mutex);
+
 	/* deinitialize everything */
 
 	input_stream_global_finish();