gst/elements/gstfakesink.c (gst_fakesink_render): Er, emit with

Original commit message from CVS:
2005-05-10  Andy Wingo  <wingo@pobox.com>

* gst/elements/gstfakesink.c (gst_fakesink_render): Er, emit with
*all* the arguments.

* gst/base/gstbasetransform.c (gst_base_transform_event): Grab the
stream lock if it's a FLUSH_DONE; normal flushes don't get the
lock (according to the docs -- if this is wrong change the docs).

* gst/gstpipeline.c (gst_pipeline_change_state): Set the bus to
flush messages in the NULL state.

* gst/gstbus.c (gst_bus_post): If a bus is flushing, unref the
message immediately and return.
(gst_bus_set_flushing): New function. If a bus is flushing, it
flushes out any queued messages and immediately unrefs new
messages. This is so when an element goes to NULL, all of the
unhandled messages coming from it can be freed, and their
references to the element dropped. In other words: message source
ref considered harmful :P

* gst/gstbin.c (gst_bin_change_state): Unref peer element when
we're finished with it.

* gst/gstmessage.c (gst_message_new_state_changed):

ChangeLog

@@ -1,3 +1,29 @@
+2005-05-10  Andy Wingo  <wingo@pobox.com>
+
+	* gst/elements/gstfakesink.c (gst_fakesink_render): Er, emit with
+	*all* the arguments.
+
+	* gst/base/gstbasetransform.c (gst_base_transform_event): Grab the
+	stream lock if it's a FLUSH_DONE; normal flushes don't get the
+	lock (according to the docs -- if this is wrong change the docs).
+
+	* gst/gstpipeline.c (gst_pipeline_change_state): Set the bus to
+	flush messages in the NULL state.
+
+	* gst/gstbus.c (gst_bus_post): If a bus is flushing, unref the
+	message immediately and return.
+	(gst_bus_set_flushing): New function. If a bus is flushing, it
+	flushes out any queued messages and immediately unrefs new
+	messages. This is so when an element goes to NULL, all of the
+	unhandled messages coming from it can be freed, and their
+	references to the element dropped. In other words: message source
+	ref considered harmful :P
+
+	* gst/gstbin.c (gst_bin_change_state): Unref peer element when
+	we're finished with it.
+
+	* gst/gstmessage.c (gst_message_new_state_changed): 
+
 2005-05-10  Wim Taymans  <wim@fluendo.com>
 
 	* gst/gstvalue.c: (gst_value_compare_flags),

gst/base/gstbasetransform.c

@@ -236,9 +236,9 @@ gst_base_transform_event (GstPad * pad, GstEvent * event)
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_FLUSH:
       if (GST_EVENT_FLUSH_DONE (event)) {
+        GST_STREAM_LOCK (pad);
+        unlock = TRUE;
       }
-      GST_STREAM_LOCK (pad);
-      unlock = TRUE;
       break;
     case GST_EVENT_EOS:
       GST_STREAM_LOCK (pad);

gst/elements/gstfakesink.c

@@ -323,7 +323,7 @@ gst_fakesink_render (GstBaseSink * bsink, GstBuffer * buf)
   }
   if (sink->signal_handoffs)
     g_signal_emit (G_OBJECT (sink), gst_fakesink_signals[SIGNAL_HANDOFF], 0,
-        buf);
+        buf, bsink->sinkpad);
 
   if (sink->dump) {
     gst_util_dump_mem (GST_BUFFER_DATA (buf), GST_BUFFER_SIZE (buf));

gst/gstbin.c

@@ -949,6 +949,7 @@ restart:
             GST_CAT_DEBUG_OBJECT (GST_CAT_STATES, element,
                 "not adding element %s to queue, it is in another bin",
                 GST_ELEMENT_NAME (peer_elem));
+            gst_object_unref (GST_OBJECT_CAST (peer_elem));
           }
           if (parent) {
             gst_object_unref (GST_OBJECT_CAST (parent));

gst/gstbus.c

@@ -207,8 +207,16 @@ gst_bus_post (GstBus * bus, GstMessage * message)
   GST_DEBUG_OBJECT (bus, "posting message on bus");
 
   GST_LOCK (bus);
+
+  if (GST_FLAG_IS_SET (bus, GST_BUS_FLUSHING)) {
+    gst_message_unref (message);
+    GST_UNLOCK (bus);
+    return FALSE;
+  }
+
   handler = bus->sync_handler;
   handler_data = bus->sync_handler_data;
+
   GST_UNLOCK (bus);
 
   /* first call the sync handler if it is installed */
@@ -323,6 +331,36 @@ gst_bus_have_pending (GstBus * bus)
   return (length > 0);
 }
 
+/**
+ * gst_bus_set_flushing:
+ * @bus: a #GstBus
+ * @flushing: whether or not to flush the bus
+ *
+ * If @flushing, flush out and unref any messages queued in the bus. Releases
+ * references to the message origin objects. Will flush future messages until
+ * gst_bus_set_flushing() sets @flushing to #FALSE.
+ *
+ * MT safe.
+ */
+void
+gst_bus_set_flushing (GstBus * bus, gboolean flushing)
+{
+  GstMessage *message;
+
+  GST_LOCK (bus);
+
+  if (flushing) {
+    GST_FLAG_SET (bus, GST_BUS_FLUSHING);
+
+    while ((message = gst_bus_pop (bus)))
+      gst_message_unref (message);
+  } else {
+    GST_FLAG_UNSET (bus, GST_BUS_FLUSHING);
+  }
+
+  GST_UNLOCK (bus);
+}
+
 /**
  * gst_bus_pop:
  * @bus: a #GstBus to pop

gst/gstbus.h

@@ -37,6 +37,12 @@ G_BEGIN_DECLS
 #define GST_BUS_GET_CLASS(bus)    (G_TYPE_INSTANCE_GET_CLASS ((bus), GST_TYPE_BUS, GstBusClass))
 #define GST_BUS_CAST(bus)         ((GstBus*)(bus))
 
+typedef enum {
+  GST_BUS_FLUSHING		= GST_OBJECT_FLAG_LAST,
+
+  GST_BUS_FLAG_LAST		= GST_OBJECT_FLAG_LAST + 1
+} GstBusFlags;
+
 typedef enum
 {
   GST_BUS_DROP = 0,             /* drop message */
@@ -82,6 +88,7 @@ gboolean 		gst_bus_post 			(GstBus * bus, GstMessage * message);
 gboolean 		gst_bus_have_pending 		(GstBus * bus);
 GstMessage *		gst_bus_peek 			(GstBus * bus);
 GstMessage *		gst_bus_pop 			(GstBus * bus);
+void			gst_bus_set_flushing		(GstBus * bus, gboolean flushing);
 
 void 			gst_bus_set_sync_handler 	(GstBus * bus, GstBusSyncHandler func,
     							 gpointer data);

gst/gstpipeline.c

@@ -370,6 +370,8 @@ gst_pipeline_change_state (GstElement * element)
 
   switch (transition) {
     case GST_STATE_NULL_TO_READY:
+      if (element->bus)
+        gst_bus_set_flushing (element->bus, FALSE);
       gst_scheduler_setup (GST_ELEMENT_SCHEDULER (pipeline));
       break;
     case GST_STATE_READY_TO_PAUSED:
@@ -425,6 +427,9 @@ gst_pipeline_change_state (GstElement * element)
     case GST_STATE_PAUSED_TO_READY:
       break;
     case GST_STATE_READY_TO_NULL:
+      if (element->bus) {
+        gst_bus_set_flushing (element->bus, TRUE);
+      }
       break;
   }
 

libs/gst/base/gstbasetransform.c

@@ -236,9 +236,9 @@ gst_base_transform_event (GstPad * pad, GstEvent * event)
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_FLUSH:
       if (GST_EVENT_FLUSH_DONE (event)) {
+        GST_STREAM_LOCK (pad);
+        unlock = TRUE;
       }
-      GST_STREAM_LOCK (pad);
-      unlock = TRUE;
       break;
     case GST_EVENT_EOS:
       GST_STREAM_LOCK (pad);

plugins/elements/gstfakesink.c

@@ -323,7 +323,7 @@ gst_fakesink_render (GstBaseSink * bsink, GstBuffer * buf)
   }
   if (sink->signal_handoffs)
     g_signal_emit (G_OBJECT (sink), gst_fakesink_signals[SIGNAL_HANDOFF], 0,
-        buf);
+        buf, bsink->sinkpad);
 
   if (sink->dump) {
     gst_util_dump_mem (GST_BUFFER_DATA (buf), GST_BUFFER_SIZE (buf));