--- plugins/daap/rb-daap-src.c	2009-10-23 21:32:36.112399284 +0100
+++ plugins/daap/rb-daap-src.c	2009-12-06 16:04:42.069637296 +0000
@@ -46,8 +46,8 @@
 
 #include <glib/gi18n.h>
 #include <gst/gst.h>
+#include <gst/base/gstadapter.h>
 #include <gst/base/gstbasesrc.h>
-#include <gst/base/gstpushsrc.h>
 
 #include "rb-daap-source.h"
 #include "rb-daap-src.h"
@@ -73,7 +73,7 @@
 
 struct _RBDAAPSrc
 {
-	GstPushSrc parent;
+	GstBaseSrc parent;
 
 	/* uri */
 	gchar *daap_uri;
@@ -87,17 +87,20 @@
 	gboolean chunked;
 	gboolean first_chunk;
 
-	gint64 size;
+	gint64 bytes_left;
+	gint64 bytes_total;
 
 	/* Seek stuff */
 	gint64 curoffset;
 	gint64 seek_bytes;
-	gboolean do_seek;
+
+	/* Buffers data for arbitrary length create() calls. */
+	GstAdapter *adapter;
 };
 
 struct _RBDAAPSrcClass
 {
-	GstPushSrcClass parent_class;
+	GstBaseSrcClass parent_class;
 };
 
 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
@@ -134,7 +137,7 @@
 			&urihandler_info);
 }
 
-GST_BOILERPLATE_FULL (RBDAAPSrc, rb_daap_src, GstElement, GST_TYPE_PUSH_SRC, _do_init);
+GST_BOILERPLATE_FULL (RBDAAPSrc, rb_daap_src, GstElement, GST_TYPE_BASE_SRC, _do_init);
 
 static void rb_daap_src_finalize (GObject *object);
 static void rb_daap_src_set_property (GObject *object,
@@ -150,8 +153,7 @@
 static gboolean rb_daap_src_stop (GstBaseSrc *bsrc);
 static gboolean rb_daap_src_is_seekable (GstBaseSrc *bsrc);
 static gboolean rb_daap_src_get_size (GstBaseSrc *src, guint64 *size);
-static gboolean rb_daap_src_do_seek (GstBaseSrc *src, GstSegment *segment);
-static GstFlowReturn rb_daap_src_create (GstPushSrc *psrc, GstBuffer **outbuf);
+static GstFlowReturn rb_daap_src_create (GstBaseSrc *bsrc, guint64 offset, guint size, GstBuffer **outbuf);
 
 void
 rb_daap_src_set_plugin (RBPlugin *plugin)
@@ -183,14 +185,12 @@
 	GObjectClass *gobject_class;
 	GstElementClass *gstelement_class;
 	GstBaseSrcClass *gstbasesrc_class;
-	GstPushSrcClass *gstpushsrc_class;
 
 	gobject_class = G_OBJECT_CLASS (klass);
 	gstelement_class = GST_ELEMENT_CLASS (klass);
 	gstbasesrc_class = (GstBaseSrcClass *) klass;
-	gstpushsrc_class = (GstPushSrcClass *) klass;
 
-	parent_class = g_type_class_ref (GST_TYPE_PUSH_SRC);
+	parent_class = g_type_class_ref (GST_TYPE_BASE_SRC);
 
 	gobject_class->set_property = rb_daap_src_set_property;
 	gobject_class->get_property = rb_daap_src_get_property;
@@ -207,9 +207,7 @@
 	gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (rb_daap_src_stop);
 	gstbasesrc_class->is_seekable = GST_DEBUG_FUNCPTR (rb_daap_src_is_seekable);
 	gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (rb_daap_src_get_size);
-	gstbasesrc_class->do_seek = GST_DEBUG_FUNCPTR (rb_daap_src_do_seek);
-
-	gstpushsrc_class->create = GST_DEBUG_FUNCPTR (rb_daap_src_create);
+	gstbasesrc_class->create = GST_DEBUG_FUNCPTR (rb_daap_src_create);
 }
 
 static void
@@ -219,6 +217,9 @@
 	src->sock_fd = -1;
 	src->curoffset = 0;
 	src->bytes_per_read = 4096 * 2;
+	src->bytes_left = 0;
+	src->bytes_total = 0;
+	src->adapter = gst_adapter_new ();
 }
 
 static void
@@ -235,6 +236,8 @@
 		src->sock_fd = -1;
 	}
 
+	g_object_unref (src->adapter);
+
 	G_OBJECT_CLASS (parent_class)->finalize (object);
 }
 
@@ -585,15 +588,21 @@
 				src->chunked = FALSE;
 				if (len_str) {
 					char *e;
-					src->size = strtoul (len_str, &e, 10);
+					src->bytes_left = strtoul (len_str, &e, 10);
 					if (e == len_str) {
 						GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 								   ("Couldn't read HTTP content length \"%s\"", len_str));
 						ok = FALSE;
 					}
+
+					/* In addition to the number of bytes remaining, record the total
+					 * stream length - since we are operating in pull mode. */
+					src->bytes_total = src->bytes_left + src->seek_bytes;
+
 				} else {
 					GST_DEBUG_OBJECT (src, "Response doesn't have a content length");
-					src->size = 0;
+					src->bytes_left = 0;
+					src->bytes_total = 0;
 				}
 			}
 
@@ -641,7 +650,8 @@
 		src->curoffset = src->seek_bytes;
 		if (src->chunked) {
 			src->first_chunk = TRUE;
-			src->size = 0;
+			src->bytes_left = 0;
+			src->bytes_total = 0;
 		}
 		return TRUE;
 	} else {
@@ -659,98 +669,106 @@
 }
 
 static GstFlowReturn
-rb_daap_src_create (GstPushSrc *psrc, GstBuffer **outbuf)
+rb_daap_src_create (GstBaseSrc *bsrc, guint64 offset, guint size, GstBuffer **outbuf)
 {
-	RBDAAPSrc *src;
-	size_t readsize;
-	GstBuffer *buf = NULL;
+	RBDAAPSrc *src = RB_DAAP_SRC (bsrc);
 
-	src = RB_DAAP_SRC (psrc);
-	if (src->do_seek) {
-		if (src->sock_fd != -1) {
-			close (src->sock_fd);
-			src->sock_fd = -1;
-		}
-		if (!rb_daap_src_start (GST_BASE_SRC (src)))
-			return GST_FLOW_ERROR;
-		src->do_seek = FALSE;
-	}
+	/* Restart the stream at the requested offset if it differs from the
+	 * current offset adjusted for buffered data. */
+	if (offset != src->curoffset - gst_adapter_available (src->adapter)) {
+		src->seek_bytes = offset;
 
-	/* get a new chunk, if we need one */
-	if (src->chunked && src->size == 0) {
-		if (!rb_daap_src_read_chunk_size (src, src->first_chunk, &src->size)) {
+		if (!rb_daap_src_start (GST_BASE_SRC (src))) {
 			return GST_FLOW_ERROR;
-		} else if (src->size == 0) {
-			/* EOS */
-			return GST_FLOW_UNEXPECTED;
 		}
-		src->first_chunk = FALSE;
+
+		/* Flush any buffers accumulated before this seek. */
+		gst_adapter_clear (src->adapter);
 	}
 
-	readsize = src->bytes_per_read;
-	if (src->chunked && readsize > src->size)
-		readsize = src->size;
+	/* Fill the adapter until it has enough data. */
+	while (gst_adapter_available (src->adapter) < size) {
+		GstBuffer *buf;
+		size_t bytes_to_read, bytes_read;
+
+		/* Get a new chunk if we need one. */
+		if (src->chunked && src->bytes_left == 0) {
+			if (!rb_daap_src_read_chunk_size (src, src->first_chunk, &src->bytes_left)) {
+				return GST_FLOW_ERROR;
+			} else if (src->bytes_left == 0) {
+				/* EOS */
+				return GST_FLOW_UNEXPECTED;
+			}
+			src->first_chunk = FALSE;
+		}
 
-	buf = gst_buffer_new_and_alloc (readsize);
+		/* Read pre-sized data into a local buffer. */
+		bytes_to_read = src->bytes_per_read;
+		if (src->chunked && bytes_to_read > src->bytes_left) {
+			bytes_to_read = src->bytes_left;
+		}
 
-	GST_LOG_OBJECT (src, "Reading %d bytes", readsize);
-	readsize = rb_daap_src_read (src, GST_BUFFER_DATA (buf), readsize);
-	if (readsize < 0) {
-		GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
-		gst_buffer_unref (buf);
-		return GST_FLOW_ERROR;
-	}
+		buf = gst_buffer_new_and_alloc (bytes_to_read);
 
-	if (readsize == 0) {
-		GST_DEBUG ("blocking read returns 0, EOS");
-		gst_buffer_unref (buf);
-		return GST_FLOW_UNEXPECTED;
+		GST_LOG_OBJECT (src, "Reading %d bytes", bytes_to_read);
+		bytes_read = rb_daap_src_read (src, GST_BUFFER_DATA (buf), bytes_to_read);
+		if (bytes_read < 0) {
+			GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
+			gst_buffer_unref (buf);
+			return GST_FLOW_ERROR;
+		}
+
+		if (bytes_read == 0) {
+			GST_DEBUG ("blocking read returns 0, EOS");
+			gst_buffer_unref (buf);
+			return GST_FLOW_UNEXPECTED;
+		}
+
+		/* Update stream position. */
+		if (src->chunked) {
+			src->bytes_left -= bytes_read;
+		}
+		src->curoffset += bytes_read;
+
+		/* Pass buffer to the adapter. */
+		gst_adapter_push (src->adapter, buf);
 	}
 
-	if (src->chunked)
-		src->size -= readsize;
+	/* Take the requested number of bytes from the adapter. */
+	*outbuf = gst_adapter_take_buffer (src->adapter, size);
 
-	GST_BUFFER_OFFSET (buf) = src->curoffset;
-	GST_BUFFER_SIZE (buf) = readsize;
-	GST_BUFFER_TIMESTAMP (buf) = GST_CLOCK_TIME_NONE;
-	src->curoffset += readsize;
+	/* Annotate buffer with stream information. */
+	GST_BUFFER_OFFSET (*outbuf) = offset;
+	GST_BUFFER_SIZE (*outbuf) = size;
+	GST_BUFFER_TIMESTAMP (*outbuf) = GST_CLOCK_TIME_NONE;
 
 	GST_LOG_OBJECT (src,
 			"Returning buffer from _get of size %d, ts %"
 			GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
 			", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
-			GST_BUFFER_SIZE (buf), GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
-			GST_TIME_ARGS (GST_BUFFER_DURATION (buf)),
-			GST_BUFFER_OFFSET (buf), GST_BUFFER_OFFSET_END (buf));
-	*outbuf = buf;
+			GST_BUFFER_SIZE (*outbuf), GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)),
+			GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)),
+			GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf));
+
 	return GST_FLOW_OK;
 }
 
 gboolean
 rb_daap_src_is_seekable (GstBaseSrc *bsrc)
 {
-	return TRUE;
-}
-
-gboolean
-rb_daap_src_do_seek (GstBaseSrc *bsrc, GstSegment *segment)
-{
-	RBDAAPSrc *src = RB_DAAP_SRC (bsrc);
-	if (segment->format == GST_FORMAT_BYTES) {
-		src->do_seek = TRUE;
-		src->seek_bytes = segment->start;
-		return TRUE;
-	} else {
-		return FALSE;
-	}
+	/* Only report seekable if the DAAP server provided us with a stream length.
+	 * In some cases, e.g. with transcoded sources, this data is not available
+	 * and must force the stream into an unseekable state. */
+	return (RB_DAAP_SRC(bsrc)->bytes_total > 0);
 }
 
 gboolean
 rb_daap_src_get_size (GstBaseSrc *bsrc, guint64 *size)
 {
 	RBDAAPSrc *src = RB_DAAP_SRC (bsrc);
-	if (src->chunked == FALSE && src->size > 0) {
-		*size = src->size;
+
+	if (src->chunked == FALSE && src->bytes_total > 0) {
+		*size = src->bytes_total;
 		return TRUE;
 	}
 	return FALSE;
