[rhythmbox] xfade: don't link network streams until fully buffered

[Jonathan Matthew <jmatthew src gnome org>]

commit 5816ace1055ea860189eff23650dbabd79aa73f4
Author: Jonathan Matthew <jonathan d14n org>
Date:   Fri Apr 13 22:46:45 2012 +1000

    xfade: don't link network streams until fully buffered
    
    This doesn't introduce any new stream states, we just pretend the
    stream is playing whether it's actually linked and playing or
    unlinked and buffering.

 backends/gstreamer/rb-player-gst-xfade.c |  100 +++++++++++++++++++++++++++++-
 1 files changed, 97 insertions(+), 3 deletions(-)
---
diff --git a/backends/gstreamer/rb-player-gst-xfade.c b/backends/gstreamer/rb-player-gst-xfade.c
index a74ef2e..e6e1b73 100644
--- a/backends/gstreamer/rb-player-gst-xfade.c
+++ b/backends/gstreamer/rb-player-gst-xfade.c
@@ -227,6 +227,11 @@ enum
 	LAST_SIGNAL
 };
 
+/* copied from gsturidecodebin.c:stream_uris */
+static const char *stream_schemes[] = {
+	"http", "https", "mms", "mmsh", "mmsu", "mmst", "ssh", "ftp", "sftp"
+};
+
 static guint signals[LAST_SIGNAL] = { 0 };
 
 struct _RBPlayerGstXFadePrivate
@@ -343,6 +348,7 @@ typedef struct
 	gint64 crossfade;
 	gboolean fading;
 	gboolean starting_eos;
+	gboolean use_buffering;
 
 	gulong adjust_probe_id;
 
@@ -363,6 +369,7 @@ typedef struct
 #define RB_IS_XFADE_STREAM(obj)	(G_TYPE_CHECK_INSTANCE_TYPE ((obj), RB_TYPE_XFADE_STREAM))
 
 static void adjust_stream_base_time (RBXFadeStream *stream);
+static gboolean actually_start_stream (RBXFadeStream *stream, GError **error);
 
 static void rb_xfade_stream_class_init (RBXFadeStreamClass *klass);
 
@@ -831,6 +838,8 @@ post_stream_playing_message (RBXFadeStream *stream, gboolean fake)
 
 	if (fake == FALSE) {
 		stream->emitted_playing = TRUE;
+	} else {
+		stream->emitted_fake_playing = TRUE;
 	}
 }
 
@@ -1699,12 +1708,14 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 			GstTagList *tags;
 			gst_message_parse_tag (message, &tags);
 
+			g_mutex_lock (stream->lock);
 			if (stream->emitted_playing) {
 				gst_tag_list_foreach (tags, (GstTagForeachFunc) process_tag, stream);
 				gst_tag_list_free (tags);
 			} else {
 				stream->tags = g_list_append (stream->tags, tags);
 			}
+			g_mutex_unlock (stream->lock);
 		}
 		break;
 
@@ -1733,13 +1744,19 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 		if (stream == NULL) {
 			rb_debug ("got application message %s for unknown stream", name);
 		} else if (strcmp (name, STREAM_PLAYING_MESSAGE) == 0) {
+			GList *l;
 			GList *t;
 
 			rb_debug ("got stream playing message for %s", stream->uri);
 			_rb_player_emit_playing_stream (RB_PLAYER (player), stream->stream_data);
 
 			/* process any buffered tag lists we received while prerolling the stream */
-			for (t = stream->tags; t != NULL; t = t->next) {
+			g_mutex_lock (stream->lock);
+			l = stream->tags;
+			stream->tags = NULL;
+			g_mutex_unlock (stream->lock);
+
+			for (t = l; t != NULL; t = t->next) {
 				GstTagList *tags;
 
 				tags = (GstTagList *)t->data;
@@ -1747,8 +1764,7 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 				gst_tag_list_foreach (tags, (GstTagForeachFunc) process_tag, stream);
 				gst_tag_list_free (tags);
 			}
-			g_list_free (stream->tags);
-			stream->tags = NULL;
+			g_list_free (l);
 
 		} else if (strcmp (name, FADE_IN_DONE_MESSAGE) == 0) {
 			/* do something? */
@@ -1757,7 +1773,9 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 			case FADING_OUT:
 				/* stop the stream and dispose of it */
 				rb_debug ("got fade-out-done for stream %s -> PENDING_REMOVE", stream->uri);
+				g_mutex_lock (stream->lock);
 				stream->state = PENDING_REMOVE;
+				g_mutex_unlock (stream->lock);
 				schedule_stream_reap (player);
 				break;
 
@@ -1766,6 +1784,8 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 					/* try to seek back a bit to account for the fade */
 					GstFormat format = GST_FORMAT_TIME;
 					gint64 pos = -1;
+
+					g_mutex_lock (stream->lock);
 					gst_element_query_position (stream->volume, &format, &pos);
 					if (pos != -1) {
 						stream->seek_target = pos > PAUSE_FADE_LENGTH ? pos - PAUSE_FADE_LENGTH : 0;
@@ -1777,6 +1797,7 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 						rb_debug ("got fade-out-done for stream %s -> PAUSED (position query failed)",
 							  stream->uri);
 					}
+					g_mutex_unlock (stream->lock);
 				}
 				unlink_and_block_stream (stream);
 				break;
@@ -1825,6 +1846,57 @@ rb_player_gst_xfade_bus_cb (GstBus *bus, GstMessage *message, RBPlayerGstXFade *
 			break;
 		}
 
+		g_mutex_lock (stream->lock);
+		if (progress >= 100) {
+			GError *error = NULL;
+			switch (stream->state) {
+			case PREROLLING:
+				rb_debug ("stream %s is buffered, now waiting", stream->uri);
+				stream->state = WAITING;
+				break;
+
+			case WAITING_EOS:
+				/* hmm, not sure */
+				break;
+
+			case PREROLL_PLAY:
+				rb_debug ("stream %s is buffered, now playing", stream->uri);
+				if (actually_start_stream (stream, &error) == FALSE) {
+					emit_stream_error (stream, error);
+				}
+				break;
+
+			default:
+				rb_debug ("stream %s is buffered, resuming", stream->uri);
+				link_and_unblock_stream (stream, &error);
+				if (error) {
+					g_warning ("couldn't restart newly buffered stream: %s", error->message);
+					g_clear_error (&error);
+				}
+				break;
+			}
+		} else {
+			switch (stream->state) {
+			case PREROLLING:
+			case WAITING:
+				rb_debug ("still buffering, %d", progress);
+				stream->state = PREROLLING;
+				break;
+
+			case WAITING_EOS:
+				/* not sure */
+				break;
+
+			default:
+				if (stream->adder_pad != NULL) {
+					rb_debug ("stream buffering, stopping playback");
+					unlink_and_block_stream (stream);
+				}
+				break;
+			}
+		}
+		g_mutex_unlock (stream->lock);
+
 		if (stream == NULL) {
 			rb_debug ("got buffering message for unknown stream (%d)", progress);
 		} else {
@@ -2008,6 +2080,7 @@ create_stream (RBPlayerGstXFade *player, const char *uri, gpointer stream_data,
 	GValueArray *stream_filters = NULL;
 	GstElement *tail;
 	GstController *controller;
+	int i;
 
 	rb_debug ("creating new stream for %s (stream data %p)", uri, stream_data);
 	stream = g_object_new (RB_TYPE_XFADE_STREAM, NULL, NULL);
@@ -2017,6 +2090,14 @@ create_stream (RBPlayerGstXFade *player, const char *uri, gpointer stream_data,
 	stream->uri = g_strdup (uri);
 	stream->state = WAITING;
 
+	stream->use_buffering = FALSE;
+	for (i = 0; i < G_N_ELEMENTS (stream_schemes); i++) {
+		if (gst_uri_has_protocol (uri, stream_schemes[i])) {
+			stream->use_buffering = TRUE;
+			break;
+		}
+	}
+
 	/* kill the floating reference */
 	g_object_ref (stream);
 	gst_object_sink (stream);
@@ -2403,6 +2484,19 @@ stream_src_blocked_cb (GstPad *pad, gboolean blocked, RBXFadeStream *stream)
 		      "max-size-buffers", 200,		/* back to normal value */
 		      NULL);
 
+	if (stream->use_buffering) {
+		rb_debug ("stream %s requires buffering", stream->uri);
+		switch (stream->state) {
+		case PREROLL_PLAY:
+			post_stream_playing_message (stream, TRUE);
+			break;
+		default:
+			break;
+		}
+		g_mutex_unlock (stream->lock);
+		return;
+	}
+
 	/* update stream state */
 	switch (stream->state) {
 	case PREROLLING: