s/GstBuffer/GstData/ in the API where you can pass events. Fix the plugins to deal with that. Fixes #113488. Also inc...

Original commit message from CVS:
s/GstBuffer/GstData/ in the API where you can pass events. Fix the plugins to deal with that. Fixes #113488. Also includes scheduler patches, and probably fixes some queue bugs relating to events and buffers.

16 files changed, 119 insertions, 104 deletions

diff --git a/plugins/elements/gstaggregator.c b/plugins/elements/gstaggregator.c
index 0ac2bd90a..3c3ba5d8a 100644
--- a/plugins/elements/gstaggregator.c
+++ b/plugins/elements/gstaggregator.c
@@ -92,7 +92,7 @@ static void 	gst_aggregator_set_property 	(GObject *object, guint prop_id,
 static void 	gst_aggregator_get_property 	(GObject *object, guint prop_id, 
 						 GValue *value, GParamSpec *pspec);
 
-static void  	gst_aggregator_chain 		(GstPad *pad, GstBuffer *buf);
+static void  	gst_aggregator_chain 		(GstPad *pad, GstData *_data);
 static void 	gst_aggregator_loop 		(GstElement *element);
 
 static GstElementClass *parent_class = NULL;
@@ -286,7 +286,7 @@ gst_aggregator_push (GstAggregator *aggregator, GstPad *pad, GstBuffer *buf, guc
     g_object_notify (G_OBJECT (aggregator), "last_message");
   }
 
-  gst_pad_push (aggregator->srcpad, buf);
+  gst_pad_push (aggregator->srcpad, GST_DATA (buf));
 }
 
 static void 
@@ -313,7 +313,7 @@ gst_aggregator_loop (GstElement *element)
        * and that the peer pad is also enabled.
        */
       if (GST_PAD_IS_USABLE (pad)) {
-        buf = gst_pad_pull (pad);
+        buf = GST_BUFFER (gst_pad_pull (pad));
         debug = "loop";
 
 	/* then push it forward */
@@ -328,7 +328,7 @@ gst_aggregator_loop (GstElement *element)
       debug = "loop_select";
 
       pad = gst_pad_select (aggregator->sinkpads);
-      buf = gst_pad_pull (pad);
+      buf = GST_BUFFER (gst_pad_pull (pad));
 
       gst_aggregator_push (aggregator, pad, buf, debug);
     }
@@ -346,8 +346,9 @@ gst_aggregator_loop (GstElement *element)
  * Chain a buffer on a pad.
  */
 static void 
-gst_aggregator_chain (GstPad *pad, GstBuffer *buf) 
+gst_aggregator_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstAggregator *aggregator;
 
   g_return_if_fail (pad != NULL);
diff --git a/plugins/elements/gstfakesink.c b/plugins/elements/gstfakesink.c
index 0050939ae..d9a94a582 100644
--- a/plugins/elements/gstfakesink.c
+++ b/plugins/elements/gstfakesink.c
@@ -102,7 +102,7 @@ static void	gst_fakesink_get_property	(GObject *object, guint prop_id,
 static GstElementStateReturn
 		gst_fakesink_change_state 	(GstElement *element);
 
-static void	gst_fakesink_chain		(GstPad *pad, GstBuffer *buf);
+static void	gst_fakesink_chain		(GstPad *pad, GstData *_data);
 
 static GstElementClass *parent_class = NULL;
 static guint gst_fakesink_signals[LAST_SIGNAL] = { 0 };
@@ -298,8 +298,9 @@ gst_fakesink_get_property (GObject *object, guint prop_id, GValue *value, GParam
 }
 
 static void 
-gst_fakesink_chain (GstPad *pad, GstBuffer *buf) 
+gst_fakesink_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstFakeSink *fakesink;
 
   g_return_if_fail (pad != NULL);
diff --git a/plugins/elements/gstfakesrc.c b/plugins/elements/gstfakesrc.c
index 9d088beac..e992ee50f 100644
--- a/plugins/elements/gstfakesrc.c
+++ b/plugins/elements/gstfakesrc.c
@@ -170,7 +170,7 @@ static void		gst_fakesrc_get_property	(GObject *object, guint prop_id,
 
 static GstElementStateReturn gst_fakesrc_change_state 	(GstElement *element);
 
-static GstBuffer*	gst_fakesrc_get			(GstPad *pad);
+static GstData*	gst_fakesrc_get			(GstPad *pad);
 static void 		gst_fakesrc_loop		(GstElement *element);
 
 static GstElementClass *parent_class = NULL;
@@ -734,7 +734,7 @@ gst_fakesrc_create_buffer (GstFakeSrc *src)
   return buf;
 }
 
-static GstBuffer *
+static GstData *
 gst_fakesrc_get(GstPad *pad)
 {
   GstFakeSrc *src;
@@ -748,22 +748,22 @@ gst_fakesrc_get(GstPad *pad)
 
   if (src->need_flush) {
     src->need_flush = FALSE;
-    return GST_BUFFER(gst_event_new (GST_EVENT_FLUSH));
+    return GST_DATA(gst_event_new (GST_EVENT_FLUSH));
   }
 
   if (src->buffer_count == src->segment_end) {
     if (src->segment_loop) {
-      return GST_BUFFER(gst_event_new (GST_EVENT_SEGMENT_DONE));
+      return GST_DATA(gst_event_new (GST_EVENT_SEGMENT_DONE));
     }
     else {
       gst_element_set_eos (GST_ELEMENT (src));
-      return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+      return GST_DATA(gst_event_new (GST_EVENT_EOS));
     }
   }
 
   if (src->rt_num_buffers == 0) {
     gst_element_set_eos (GST_ELEMENT (src));
-    return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+    return GST_DATA(gst_event_new (GST_EVENT_EOS));
   }
   else {
     if (src->rt_num_buffers > 0)
@@ -772,7 +772,7 @@ gst_fakesrc_get(GstPad *pad)
 
   if (src->eos) {
     GST_INFO ( "fakesrc is setting eos on pad");
-    return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+    return GST_DATA(gst_event_new (GST_EVENT_EOS));
   }
 
   buf = gst_fakesrc_create_buffer (src);
@@ -795,7 +795,7 @@ gst_fakesrc_get(GstPad *pad)
     GST_LOG_OBJECT (src, "post handoff emit");
   }
 
-  return buf;
+  return GST_DATA (buf);
 }
 
 /**
@@ -819,10 +819,10 @@ gst_fakesrc_loop(GstElement *element)
 
   while (pads) {
     GstPad *pad = GST_PAD (pads->data);
-    GstBuffer *buf;
+    GstData *data;
 
-    buf = gst_fakesrc_get (pad);
-    gst_pad_push (pad, buf);
+    data = gst_fakesrc_get (pad);
+    gst_pad_push (pad, data);
 
     if (src->eos) {
       return;
diff --git a/plugins/elements/gstfdsink.c b/plugins/elements/gstfdsink.c
index b25f377d8..b3f215912 100644
--- a/plugins/elements/gstfdsink.c
+++ b/plugins/elements/gstfdsink.c
@@ -61,7 +61,7 @@ static void 	gst_fdsink_set_property	(GObject *object, guint prop_id,
 static void 	gst_fdsink_get_property	(GObject *object, guint prop_id, 
 					 GValue *value, GParamSpec *pspec);
 
-static void 	gst_fdsink_chain	(GstPad *pad,GstBuffer *buf);
+static void 	gst_fdsink_chain	(GstPad *pad,GstData *_data);
 
 static GstElementClass *parent_class = NULL;
 /*static guint gst_fdsink_signals[LAST_SIGNAL] = { 0 };*/
@@ -115,8 +115,9 @@ gst_fdsink_init (GstFdSink *fdsink)
 }
 
 static void 
-gst_fdsink_chain (GstPad *pad, GstBuffer *buf) 
+gst_fdsink_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstFdSink *fdsink;
 
   g_return_if_fail (pad != NULL);
diff --git a/plugins/elements/gstfdsrc.c b/plugins/elements/gstfdsrc.c
index 5b2bc91c2..997039987 100644
--- a/plugins/elements/gstfdsrc.c
+++ b/plugins/elements/gstfdsrc.c
@@ -71,7 +71,7 @@ static void		gst_fdsrc_set_property	(GObject *object, guint prop_id,
 static void		gst_fdsrc_get_property	(GObject *object, guint prop_id, 
 						 GValue *value, GParamSpec *pspec);
 
-static GstBuffer *	gst_fdsrc_get		(GstPad *pad);
+static GstData *	gst_fdsrc_get		(GstPad *pad);
 
 
 static GstElementClass *parent_class = NULL;
@@ -177,7 +177,7 @@ gst_fdsrc_get_property (GObject *object, guint prop_id, GValue *value, GParamSpe
   }
 }
 
-static GstBuffer *
+static GstData *
 gst_fdsrc_get(GstPad *pad)
 {
   GstFdSrc *src;
@@ -195,14 +195,14 @@ gst_fdsrc_get(GstPad *pad)
   /* if nothing was read, we're in eos */
   if (readbytes == 0) {
     gst_element_set_eos (GST_ELEMENT (src));
-    return GST_BUFFER (gst_event_new (GST_EVENT_EOS));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
   }
 
   if (readbytes == -1) {
-	  g_error ("Error reading from file descriptor. Ending stream.\n");
-	  gst_element_set_eos (GST_ELEMENT (src));
-	  return GST_BUFFER (gst_event_new (GST_EVENT_EOS));
-	  }
+    g_error ("Error reading from file descriptor. Ending stream.\n");
+    gst_element_set_eos (GST_ELEMENT (src));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
+  }
   
   GST_BUFFER_OFFSET (buf) = src->curoffset;
   GST_BUFFER_SIZE (buf) = readbytes;
@@ -210,5 +210,5 @@ gst_fdsrc_get(GstPad *pad)
   src->curoffset += readbytes;
 
   /* we're done, return the buffer */
-  return buf;
+  return GST_DATA (buf);
 }
diff --git a/plugins/elements/gstfilesink.c b/plugins/elements/gstfilesink.c
index ac8fb14cc..a3ced9954 100644
--- a/plugins/elements/gstfilesink.c
+++ b/plugins/elements/gstfilesink.c
@@ -83,7 +83,7 @@ static void 	gst_filesink_close_file 	(GstFileSink *sink);
 static gboolean gst_filesink_handle_event       (GstPad *pad, GstEvent *event);
 static gboolean	gst_filesink_pad_query		(GstPad *pad, GstQueryType type,
 						 GstFormat *format, gint64 *value);
-static void	gst_filesink_chain		(GstPad *pad,GstBuffer *buf);
+static void	gst_filesink_chain		(GstPad *pad,GstData *_data);
 
 static GstElementStateReturn gst_filesink_change_state (GstElement *element);
 
@@ -361,8 +361,9 @@ gst_filesink_handle_event (GstPad *pad, GstEvent *event)
  * take the buffer from the pad and write to file if it's open
  */
 static void 
-gst_filesink_chain (GstPad *pad, GstBuffer *buf) 
+gst_filesink_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstFileSink *filesink;
 
   g_return_if_fail (pad != NULL);
diff --git a/plugins/elements/gstfilesrc.c b/plugins/elements/gstfilesrc.c
index 8ceba5883..f0de20d0a 100644
--- a/plugins/elements/gstfilesrc.c
+++ b/plugins/elements/gstfilesrc.c
@@ -142,7 +142,7 @@ static void		gst_filesrc_set_property	(GObject *object, guint prop_id,
 static void		gst_filesrc_get_property	(GObject *object, guint prop_id, 
 							 GValue *value, GParamSpec *pspec);
 
-static GstBuffer *	gst_filesrc_get			(GstPad *pad);
+static GstData *	gst_filesrc_get			(GstPad *pad);
 static gboolean 	gst_filesrc_srcpad_event 	(GstPad *pad, GstEvent *event);
 static gboolean 	gst_filesrc_srcpad_query 	(GstPad *pad, GstQueryType type,
 		         				 GstFormat *format, gint64 *value);
@@ -650,7 +650,7 @@ gst_filesrc_get_read (GstFileSrc *src)
   return buf;
 }
 
-static GstBuffer *
+static GstData *
 gst_filesrc_get (GstPad *pad)
 {
   GstFileSrc *src;
@@ -667,13 +667,13 @@ gst_filesrc_get (GstPad *pad)
     GST_DEBUG ("filesrc sending discont");
     event = gst_event_new_discontinuous (FALSE, GST_FORMAT_BYTES, src->curoffset, NULL);
     src->need_flush = FALSE;
-    return GST_BUFFER (event);
+    return GST_DATA (event);
   }
   /* check for flush */
   if (src->need_flush) {
     src->need_flush = FALSE;
     GST_DEBUG ("filesrc sending flush");
-    return GST_BUFFER (gst_event_new_flush ());
+    return GST_DATA (gst_event_new_flush ());
   }
 
   /* check for EOF */
@@ -681,13 +681,13 @@ gst_filesrc_get (GstPad *pad)
     GST_DEBUG ("filesrc eos %" G_GINT64_FORMAT" %" G_GINT64_FORMAT,
                src->curoffset, src->filelen);
     gst_element_set_eos (GST_ELEMENT (src));
-    return GST_BUFFER (gst_event_new (GST_EVENT_EOS));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
   }
 
   if (src->using_mmap){
-    return gst_filesrc_get_mmap (src);
+    return GST_DATA (gst_filesrc_get_mmap (src));
   }else{
-    return gst_filesrc_get_read (src);
+    return GST_DATA (gst_filesrc_get_read (src));
   }
 }
 
diff --git a/plugins/elements/gstidentity.c b/plugins/elements/gstidentity.c
index df42d70bd..b6b77d6f2 100644
--- a/plugins/elements/gstidentity.c
+++ b/plugins/elements/gstidentity.c
@@ -70,7 +70,7 @@ static void gst_identity_init		(GstIdentity *identity);
 static void gst_identity_set_property	(GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
 static void gst_identity_get_property	(GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
 
-static void gst_identity_chain		(GstPad *pad, GstBuffer *buf);
+static void gst_identity_chain		(GstPad *pad, GstData *_data);
 
 static GstElementClass *parent_class = NULL;
 static guint gst_identity_signals[LAST_SIGNAL] = { 0 };
@@ -224,8 +224,9 @@ gst_identity_init (GstIdentity *identity)
 }
 
 static void 
-gst_identity_chain (GstPad *pad, GstBuffer *buf) 
+gst_identity_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstIdentity *identity;
   guint i;
 
@@ -286,7 +287,7 @@ gst_identity_chain (GstPad *pad, GstBuffer *buf)
     if (i>1) 
       gst_buffer_ref (buf);
 
-    gst_pad_push (identity->srcpad, buf);
+    gst_pad_push (identity->srcpad, GST_DATA (buf));
 
     if (identity->sleep_time)
       g_usleep (identity->sleep_time);
@@ -304,7 +305,7 @@ gst_identity_loop (GstElement *element)
 
   identity = GST_IDENTITY (element);
   
-  buf = gst_pad_pull (identity->sinkpad);
+  buf = GST_BUFFER (gst_pad_pull (identity->sinkpad));
   if (GST_IS_EVENT (buf)) {
     GstEvent *event = GST_EVENT (buf);
 
@@ -316,7 +317,7 @@ gst_identity_loop (GstElement *element)
     }
   }
   else {
-    gst_identity_chain (identity->sinkpad, buf);
+    gst_identity_chain (identity->sinkpad, GST_DATA (buf));
   }
 }
 
diff --git a/plugins/elements/gstmd5sink.c b/plugins/elements/gstmd5sink.c
index e5781506b..114d78daf 100644
--- a/plugins/elements/gstmd5sink.c
+++ b/plugins/elements/gstmd5sink.c
@@ -63,7 +63,7 @@ static void			gst_md5sink_init		(GstMD5Sink *md5sink);
 static void			gst_md5sink_get_property	(GObject *object, guint prop_id, 
 								 GValue *value, GParamSpec *pspec);
 
-static void			gst_md5sink_chain		(GstPad *pad, GstBuffer *buf);
+static void			gst_md5sink_chain		(GstPad *pad, GstData *_data);
 static GstElementStateReturn	gst_md5sink_change_state	(GstElement *element);
 
 /* variables */
@@ -489,8 +489,9 @@ gst_md5sink_get_property (GObject *object, guint prop_id, GValue *value, GParamS
 }
 
 static void 
-gst_md5sink_chain (GstPad *pad, GstBuffer *buf) 
+gst_md5sink_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstMD5Sink *md5sink;
 
   g_return_if_fail (pad != NULL);
diff --git a/plugins/elements/gstmultidisksrc.c b/plugins/elements/gstmultidisksrc.c
index 5a27bdf3a..c2a0e6fdb 100644
--- a/plugins/elements/gstmultidisksrc.c
+++ b/plugins/elements/gstmultidisksrc.c
@@ -63,7 +63,7 @@ static void		gst_multidisksrc_init		(GstMultiDiskSrc *disksrc);
 static void		gst_multidisksrc_set_property	(GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
 static void		gst_multidisksrc_get_property	(GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
 
-static GstBuffer *	gst_multidisksrc_get		(GstPad *pad);
+static GstData *	gst_multidisksrc_get		(GstPad *pad);
 /*static GstBuffer *	gst_multidisksrc_get_region	(GstPad *pad,GstRegionType type,guint64 offset,guint64 len);*/
 
 static GstElementStateReturn	gst_multidisksrc_change_state	(GstElement *element);
@@ -195,7 +195,7 @@ gst_multidisksrc_get_property (GObject *object, guint prop_id, GValue *value, GP
  *
  * Push a new buffer from the disksrc at the current offset.
  */
-static GstBuffer *
+static GstData *
 gst_multidisksrc_get (GstPad *pad)
 {
   GstMultiDiskSrc *src;
@@ -209,7 +209,7 @@ gst_multidisksrc_get (GstPad *pad)
     gst_multidisksrc_close_file(src);
 
   if (!src->listptr) {
-    return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
   }
 
   list = src->listptr;
@@ -240,7 +240,7 @@ gst_multidisksrc_get (GstPad *pad)
   }
 
   /* we're done, return the buffer */
-  return buf;
+  return GST_DATA (buf);
 }
 
 /* open the file and mmap it, necessary to go to READY state */
diff --git a/plugins/elements/gstmultifilesrc.c b/plugins/elements/gstmultifilesrc.c
index 5a27bdf3a..c2a0e6fdb 100644
--- a/plugins/elements/gstmultifilesrc.c
+++ b/plugins/elements/gstmultifilesrc.c
@@ -63,7 +63,7 @@ static void		gst_multidisksrc_init		(GstMultiDiskSrc *disksrc);
 static void		gst_multidisksrc_set_property	(GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
 static void		gst_multidisksrc_get_property	(GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
 
-static GstBuffer *	gst_multidisksrc_get		(GstPad *pad);
+static GstData *	gst_multidisksrc_get		(GstPad *pad);
 /*static GstBuffer *	gst_multidisksrc_get_region	(GstPad *pad,GstRegionType type,guint64 offset,guint64 len);*/
 
 static GstElementStateReturn	gst_multidisksrc_change_state	(GstElement *element);
@@ -195,7 +195,7 @@ gst_multidisksrc_get_property (GObject *object, guint prop_id, GValue *value, GP
  *
  * Push a new buffer from the disksrc at the current offset.
  */
-static GstBuffer *
+static GstData *
 gst_multidisksrc_get (GstPad *pad)
 {
   GstMultiDiskSrc *src;
@@ -209,7 +209,7 @@ gst_multidisksrc_get (GstPad *pad)
     gst_multidisksrc_close_file(src);
 
   if (!src->listptr) {
-    return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
   }
 
   list = src->listptr;
@@ -240,7 +240,7 @@ gst_multidisksrc_get (GstPad *pad)
   }
 
   /* we're done, return the buffer */
-  return buf;
+  return GST_DATA (buf);
 }
 
 /* open the file and mmap it, necessary to go to READY state */
diff --git a/plugins/elements/gstpipefilter.c b/plugins/elements/gstpipefilter.c
index 65a2a7fc2..e31e0e255 100644
--- a/plugins/elements/gstpipefilter.c
+++ b/plugins/elements/gstpipefilter.c
@@ -68,8 +68,8 @@ static void 			gst_pipefilter_init		(GstPipefilter *pipefilter);
 static void 			gst_pipefilter_set_property	(GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
 static void 			gst_pipefilter_get_property	(GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
 
-static GstBuffer*		gst_pipefilter_get		(GstPad *pad);
-static void 			gst_pipefilter_chain		(GstPad *pad, GstBuffer *buf);
+static GstData*		gst_pipefilter_get		(GstPad *pad);
+static void 			gst_pipefilter_chain		(GstPad *pad, GstData *_data);
 static gboolean 		gst_pipefilter_handle_event 	(GstPad *pad, GstEvent *event);
 
 static GstElementStateReturn 	gst_pipefilter_change_state	(GstElement *element);
@@ -155,7 +155,7 @@ gst_pipefilter_handle_event (GstPad *pad, GstEvent *event)
   return TRUE;
 }
 
-static GstBuffer* 
+static GstData* 
 gst_pipefilter_get (GstPad *pad)
 {
   GstPipefilter *pipefilter;
@@ -184,7 +184,7 @@ gst_pipefilter_get (GstPad *pad)
   }
   /* if we didn't get as many bytes as we asked for, we're at EOF */
   if (readbytes == 0) {
-    return GST_BUFFER(gst_event_new (GST_EVENT_EOS));
+    return GST_DATA (gst_event_new (GST_EVENT_EOS));
 
   }
 
@@ -192,12 +192,13 @@ gst_pipefilter_get (GstPad *pad)
   GST_BUFFER_SIZE(newbuf) = readbytes;
   pipefilter->curoffset += readbytes;
 
-  return newbuf;
+  return GST_DATA (newbuf);
 }
 
 static void
-gst_pipefilter_chain (GstPad *pad,GstBuffer *buf)
+gst_pipefilter_chain (GstPad *pad,GstData *_data)
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstPipefilter *pipefilter;
   glong writebytes;
   guchar *data;
diff --git a/plugins/elements/gstqueue.c b/plugins/elements/gstqueue.c
index b20163968..c1924242e 100644
--- a/plugins/elements/gstqueue.c
+++ b/plugins/elements/gstqueue.c
@@ -72,8 +72,8 @@ static void			gst_queue_set_property		(GObject *object, guint prop_id,
 static void			gst_queue_get_property		(GObject *object, guint prop_id, 
 								 GValue *value, GParamSpec *pspec);
 
-static void			gst_queue_chain			(GstPad *pad, GstBuffer *buf);
-static GstBuffer *		gst_queue_get			(GstPad *pad);
+static void			gst_queue_chain			(GstPad *pad, GstData *data);
+static GstData *		gst_queue_get			(GstPad *pad);
 static GstBufferPool* 		gst_queue_get_bufferpool 	(GstPad *pad);
 	
 static gboolean 		gst_queue_handle_src_event 	(GstPad *pad, GstEvent *event);
@@ -300,13 +300,13 @@ gst_queue_locked_flush (GstQueue *queue)
 }
 
 static void
-gst_queue_chain (GstPad *pad, GstBuffer *buf)
+gst_queue_chain (GstPad *pad, GstData *data)
 {
   GstQueue *queue;
 
   g_return_if_fail (pad != NULL);
   g_return_if_fail (GST_IS_PAD (pad));
-  g_return_if_fail (buf != NULL);
+  g_return_if_fail (data != NULL);
 
   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
   
@@ -319,33 +319,35 @@ gst_queue_chain (GstPad *pad, GstBuffer *buf)
     GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "event sent\n");
   }
   g_async_queue_unlock(queue->events);
-
+  
 restart:
   /* we have to lock the queue since we span threads */
   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "locking t:%p", g_thread_self ());
   g_mutex_lock (queue->qlock);
   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "locked t:%p", g_thread_self ());
-
+  
   /* assume don't need to flush this buffer when the queue is filled */
   queue->flush = FALSE;
-
-  if (GST_IS_EVENT (buf)) {
-    switch (GST_EVENT_TYPE (buf)) {
+  
+  if (GST_IS_EVENT (data)) {
+    switch (GST_EVENT_TYPE (data)) {
       case GST_EVENT_FLUSH:
         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "FLUSH event, flushing queue\n");
         gst_queue_locked_flush (queue);
 	break;
       case GST_EVENT_EOS:
 	GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "eos in on %s %d\n", 
-			   GST_ELEMENT_NAME (queue), queue->level_buffers);
+                              GST_ELEMENT_NAME (queue), queue->level_buffers);
 	break;
       default:
 	/* we put the event in the queue, we don't have to act ourselves */
 	break;
     }
   }
-
-  GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "adding buffer %p of size %d",buf,GST_BUFFER_SIZE(buf));
+  
+  if (GST_IS_BUFFER (data))
+    GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, 
+                        "adding buffer %p of size %d", data, GST_BUFFER_SIZE (data));
 
   if (queue->level_buffers == queue->size_buffers) {
     g_mutex_unlock (queue->qlock);
@@ -358,10 +360,10 @@ restart:
       /* if we leak on the upstream side, drop the current buffer */
       if (queue->leaky == GST_QUEUE_LEAK_UPSTREAM) {
         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "queue is full, leaking buffer on upstream end");
-        if (GST_IS_EVENT (buf))
+        if (GST_IS_EVENT (data))
           fprintf(stderr, "Error: queue [%s] leaked an event, type:%d\n",
               GST_ELEMENT_NAME(GST_ELEMENT(queue)),
-              GST_EVENT_TYPE(GST_EVENT(buf)));
+              GST_EVENT_TYPE(GST_EVENT(data)));
         /* now we have to clean up and exit right away */
         g_mutex_unlock (queue->qlock);
         goto out_unref;
@@ -369,21 +371,23 @@ restart:
       /* otherwise we have to push a buffer off the other end */
       else {
         gpointer front;
-        GstBuffer *leakbuf;
+        GstData *leak;
 
         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "queue is full, leaking buffer on downstream end");
 
         front = g_queue_pop_head (queue->queue);
-        leakbuf = (GstBuffer *)(front);
+        leak = GST_DATA (front);
 
-        if (GST_IS_EVENT (leakbuf)) {
+        queue->level_buffers--;
+        if (GST_IS_EVENT (leak)) {
           fprintf(stderr, "Error: queue [%s] leaked an event, type:%d\n",
               GST_ELEMENT_NAME(GST_ELEMENT(queue)),
-              GST_EVENT_TYPE(GST_EVENT(leakbuf)));
-	}
-        queue->level_buffers--;
-        queue->level_bytes -= GST_BUFFER_SIZE(leakbuf);
-        gst_data_unref (GST_DATA (leakbuf));
+              GST_EVENT_TYPE(GST_EVENT(leak)));
+	} else {
+          queue->level_bytes -= GST_BUFFER_SIZE(leak);
+        }
+
+        gst_data_unref (leak);
       }
     }
 
@@ -412,7 +416,7 @@ restart:
 	/* try to signal to resolve the error */
 	if (!queue->may_deadlock) {
           g_mutex_unlock (queue->qlock);
-          gst_data_unref (GST_DATA (buf));
+          gst_data_unref (data);
           gst_element_error (GST_ELEMENT (queue), "deadlock found, source pad elements are shut down");
 	  /* we don't want to goto out_unref here, since we want to clean up before calling gst_element_error */
 	  return;
@@ -428,14 +432,15 @@ restart:
       GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "got not_full signal");
     }
     GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "post full wait, level:%d/%d buffers, %d bytes",
-        queue->level_buffers, queue->size_buffers, queue->level_bytes);
+                        queue->level_buffers, queue->size_buffers, queue->level_bytes);
   }
 
   /* put the buffer on the tail of the list */
-  g_queue_push_tail (queue->queue, buf);
+  g_queue_push_tail (queue->queue, data);
 
   queue->level_buffers++;
-  queue->level_bytes += GST_BUFFER_SIZE(buf);
+  if (GST_IS_BUFFER (data))
+    queue->level_bytes += GST_BUFFER_SIZE (data);
 
   /* this assertion _has_ to hold */
   g_assert (queue->queue->length == queue->level_buffers);
@@ -451,15 +456,15 @@ restart:
   return;
 
 out_unref:
-  gst_data_unref (GST_DATA (buf));
+  gst_data_unref (data);
   return;
 }
 
-static GstBuffer *
+static GstData *
 gst_queue_get (GstPad *pad)
 {
   GstQueue *queue;
-  GstBuffer *buf = NULL;
+  GstData *data = NULL;
   gpointer front;
 
   g_assert(pad != NULL);
@@ -486,7 +491,7 @@ restart:
       GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "interrupted!!");
       g_mutex_unlock (queue->qlock);
       if (gst_scheduler_interrupt (gst_pad_get_scheduler (queue->srcpad), GST_ELEMENT (queue)))
-        return GST_BUFFER (gst_event_new (GST_EVENT_INTERRUPT));
+        return GST_DATA (gst_event_new (GST_EVENT_INTERRUPT));
       goto restart;
     }
     if (GST_STATE (queue) != GST_STATE_PLAYING) {
@@ -512,7 +517,7 @@ restart:
       if (!g_cond_timed_wait (queue->not_empty, queue->qlock, &timeout)){
         g_mutex_unlock (queue->qlock);
 	g_warning ("filler");
-        return GST_BUFFER(gst_event_new_filler());
+        return GST_DATA (gst_event_new_filler());
       }
     }
     else {
@@ -524,11 +529,12 @@ restart:
     queue->level_buffers, queue->size_buffers, queue->level_bytes);
 
   front = g_queue_pop_head (queue->queue);
-  buf = (GstBuffer *)(front);
-  GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "retrieved buffer %p from queue", buf);
+  data = GST_DATA (front);
+  GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "retrieved data %p from queue", data);
 
   queue->level_buffers--;
-  queue->level_bytes -= GST_BUFFER_SIZE(buf);
+  if (GST_IS_BUFFER (data))
+    queue->level_bytes -= GST_BUFFER_SIZE (data);
 
   GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, queue, "(%s:%s)- level:%d/%d buffers, %d bytes",
       GST_DEBUG_PAD_NAME(pad),
@@ -543,8 +549,8 @@ restart:
   g_mutex_unlock (queue->qlock);
 
   /* FIXME where should this be? locked? */
-  if (GST_IS_EVENT(buf)) {
-    GstEvent *event = GST_EVENT(buf);
+  if (GST_IS_EVENT (data)) {
+    GstEvent *event = GST_EVENT (data);
     switch (GST_EVENT_TYPE(event)) {
       case GST_EVENT_EOS:
         GST_CAT_DEBUG_OBJECT (GST_CAT_DATAFLOW, queue, "queue \"%s\" eos", GST_ELEMENT_NAME (queue));
@@ -555,7 +561,7 @@ restart:
     }
   }
 
-  return buf;
+  return data;
 }
 
 
diff --git a/plugins/elements/gstshaper.c b/plugins/elements/gstshaper.c
index 7d163e25c..57fcf561b 100644
--- a/plugins/elements/gstshaper.c
+++ b/plugins/elements/gstshaper.c
@@ -286,14 +286,14 @@ gst_shaper_loop (GstElement *element)
     if (connection->buffer == NULL && GST_PAD_IS_USABLE (connection->sinkpad)) {
       GstBuffer *buffer;
 
-      buffer = gst_pad_pull (connection->sinkpad);
+      buffer = GST_BUFFER (gst_pad_pull (connection->sinkpad));
 
       /* events are simply pushed ASAP */
       if (GST_IS_EVENT (buffer)) {
 	/* save event type as it will be unreffed after the next push */
 	GstEventType type = GST_EVENT_TYPE (buffer);
 
-	gst_pad_push (connection->srcpad, buffer);
+	gst_pad_push (connection->srcpad, GST_DATA (buffer));
 
 	switch (type) {
           /* on EOS we disable the pad so that we don't pull on
@@ -322,7 +322,7 @@ gst_shaper_loop (GstElement *element)
   }
   /* if we have a connection with a buffer, push it */
   if (min != NULL && min->buffer) {
-    gst_pad_push (min->srcpad, min->buffer);
+    gst_pad_push (min->srcpad, GST_DATA (min->buffer));
     min->buffer = NULL;
     /* since we pushed a buffer, it's not EOS */
     eos = FALSE;
diff --git a/plugins/elements/gststatistics.c b/plugins/elements/gststatistics.c
index a18e64b2c..3cbcebc35 100644
--- a/plugins/elements/gststatistics.c
+++ b/plugins/elements/gststatistics.c
@@ -67,7 +67,7 @@ static void gst_statistics_init		(GstStatistics *statistics);
 static void gst_statistics_set_property	(GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
 static void gst_statistics_get_property	(GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
 
-static void gst_statistics_chain		(GstPad *pad, GstBuffer *buf);
+static void gst_statistics_chain		(GstPad *pad, GstData *_data);
 static void gst_statistics_reset		(GstStatistics *statistics);
 static void gst_statistics_print		(GstStatistics *statistics);
 
@@ -256,8 +256,9 @@ gst_statistics_print (GstStatistics *statistics)
 }
 
 static void 
-gst_statistics_chain (GstPad *pad, GstBuffer *buf) 
+gst_statistics_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstStatistics *statistics;
   gboolean update = FALSE;
 
@@ -313,7 +314,7 @@ gst_statistics_chain (GstPad *pad, GstBuffer *buf)
       gst_statistics_print(statistics);
     }
   }
-  gst_pad_push (statistics->srcpad, buf);
+  gst_pad_push (statistics->srcpad, GST_DATA (buf));
 }
 
 static void 
diff --git a/plugins/elements/gsttee.c b/plugins/elements/gsttee.c
index f29a258ca..bfc4f8fa2 100644
--- a/plugins/elements/gsttee.c
+++ b/plugins/elements/gsttee.c
@@ -71,7 +71,7 @@ static void 	gst_tee_set_property 	(GObject *object, guint prop_id,
 static void 	gst_tee_get_property 	(GObject *object, guint prop_id, 
 					 GValue *value, GParamSpec *pspec);
 
-static void  	gst_tee_chain 		(GstPad *pad, GstBuffer *buf);
+static void  	gst_tee_chain 		(GstPad *pad, GstData *_data);
 
 
 static GstElementClass *parent_class = NULL;
@@ -342,8 +342,9 @@ gst_tee_get_property (GObject *object, guint prop_id, GValue *value, GParamSpec
  * Chain a buffer on a pad.
  */
 static void 
-gst_tee_chain (GstPad *pad, GstBuffer *buf) 
+gst_tee_chain (GstPad *pad, GstData *_data) 
 {
+  GstBuffer *buf = GST_BUFFER (_data);
   GstTee *tee;
   const GList *pads;
 
@@ -373,7 +374,7 @@ gst_tee_chain (GstPad *pad, GstBuffer *buf)
     }
 
     if (GST_PAD_IS_USABLE (outpad))
-      gst_pad_push (outpad, buf);
+      gst_pad_push (outpad, GST_DATA (buf));
     else
       gst_buffer_unref (buf);
   }