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, 117 insertions, 7 deletions

diff --git a/src/input_stream.c b/src/input_stream.c
index 164df9860..60a1559ba 100644
--- a/src/input_stream.c
+++ b/src/input_stream.c
@@ -33,10 +33,13 @@ input_quark(void)
 }
 
 struct input_stream *
-input_stream_open(const char *url, GError **error_r)
+input_stream_open(const char *url,
+		  GMutex *mutex, GCond *cond,
+		  GError **error_r)
 {
 	GError *error = NULL;
 
+	assert(mutex != NULL);
 	assert(error_r == NULL || *error_r == NULL);
 
 	for (unsigned i = 0; input_plugins[i] != NULL; ++i) {
@@ -46,7 +49,7 @@ input_stream_open(const char *url, GError **error_r)
 		if (!input_plugins_enabled[i])
 			continue;
 
-		is = plugin->open(url, &error);
+		is = plugin->open(url, mutex, cond, &error);
 		if (is != NULL) {
 			assert(is->plugin != NULL);
 			assert(is->plugin->close != NULL);
@@ -87,26 +90,108 @@ input_stream_update(struct input_stream *is)
 		is->plugin->update(is);
 }
 
+void
+input_stream_wait_ready(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->mutex != NULL);
+	assert(is->cond != NULL);
+
+	while (true) {
+		input_stream_update(is);
+		if (is->ready)
+			break;
+
+		g_cond_wait(is->cond, is->mutex);
+	}
+}
+
+void
+input_stream_lock_wait_ready(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->mutex != NULL);
+	assert(is->cond != NULL);
+
+	g_mutex_lock(is->mutex);
+	input_stream_wait_ready(is);
+	g_mutex_unlock(is->mutex);
+}
+
 bool
 input_stream_seek(struct input_stream *is, goffset offset, int whence,
 		  GError **error_r)
 {
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
 	if (is->plugin->seek == NULL)
 		return false;
 
 	return is->plugin->seek(is, offset, whence, error_r);
 }
 
+bool
+input_stream_lock_seek(struct input_stream *is, goffset offset, int whence,
+		       GError **error_r)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->plugin->seek == NULL)
+		return false;
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_seek(is, offset, whence, error_r);
+
+	g_mutex_lock(is->mutex);
+	bool success = input_stream_seek(is, offset, whence, error_r);
+	g_mutex_unlock(is->mutex);
+	return success;
+}
+
 struct tag *
 input_stream_tag(struct input_stream *is)
 {
 	assert(is != NULL);
+	assert(is->plugin != NULL);
 
 	return is->plugin->tag != NULL
 		? is->plugin->tag(is)
 		: NULL;
 }
 
+struct tag *
+input_stream_lock_tag(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->plugin->tag == NULL)
+		return false;
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_tag(is);
+
+	g_mutex_lock(is->mutex);
+	struct tag *tag = input_stream_tag(is);
+	g_mutex_unlock(is->mutex);
+	return tag;
+}
+
+bool
+input_stream_available(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	return is->plugin->available != NULL
+		? is->plugin->available(is)
+		: true;
+}
+
 size_t
 input_stream_read(struct input_stream *is, void *ptr, size_t size,
 		  GError **error_r)
@@ -117,6 +202,23 @@ input_stream_read(struct input_stream *is, void *ptr, size_t size,
 	return is->plugin->read(is, ptr, size, error_r);
 }
 
+size_t
+input_stream_lock_read(struct input_stream *is, void *ptr, size_t size,
+		       GError **error_r)
+{
+	assert(ptr != NULL);
+	assert(size > 0);
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_read(is, ptr, size, error_r);
+
+	g_mutex_lock(is->mutex);
+	size_t nbytes = input_stream_read(is, ptr, size, error_r);
+	g_mutex_unlock(is->mutex);
+	return nbytes;
+}
+
 void input_stream_close(struct input_stream *is)
 {
 	is->plugin->close(is);
@@ -127,11 +229,19 @@ bool input_stream_eof(struct input_stream *is)
 	return is->plugin->eof(is);
 }
 
-int
-input_stream_buffer(struct input_stream *is, GError **error_r)
+bool
+input_stream_lock_eof(struct input_stream *is)
 {
-	if (is->plugin->buffer == NULL)
-		return 0;
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_eof(is);
 
-	return is->plugin->buffer(is, error_r);
+	g_mutex_lock(is->mutex);
+	bool eof = input_stream_eof(is);
+	g_mutex_unlock(is->mutex);
+	return eof;
 }
+