diff --git a/ChangeLog b/ChangeLog
index f81034669d..6c66d79fe8 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,17 @@
+2007-06-06  Tim-Philipp Müller  <tim at centricular dot net>
+
+	* libs/gst/base/gstdataqueue.c:
+	* libs/gst/base/gstdataqueue.h:
+	* plugins/elements/gstmultiqueue.c: (gst_single_queue_push_one),
+	(gst_multi_queue_item_new), (gst_multi_queue_chain),
+	(gst_multi_queue_sink_event):
+	* tests/check/elements/multiqueue.c: (multiqueue_suite):
+	  Fix multiqueue leaking buffers and events when downstream or the
+	  queue are flushing. Make refcounting assumptions explicit and
+	  document them (shouldn't break existing code that uses it other than
+	  maybe leak miniobjects, but that already happens anyway). Add unit
+	  test for the most common flushing case. Fixes #423700.
+	  
 2007-06-06  Sebastian Dröge  <slomo@circular-chaos.org>
 
 	* libs/gst/controller/gstcontroller.c:
diff --git a/libs/gst/base/gstdataqueue.c b/libs/gst/base/gstdataqueue.c
index 5c009d4610..a3bfde2973 100644
--- a/libs/gst/base/gstdataqueue.c
+++ b/libs/gst/base/gstdataqueue.c
@@ -397,6 +397,11 @@ gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
  * available, OR the @queue is set to flushing state.
  * MT safe.
  *
+ * Note that this function has slightly different semantics than gst_pad_push()
+ * and gst_pad_push_event(): this function only takes ownership of @item and
+ * the #GstMiniObject contained in @item if the push was successful. If FALSE
+ * is returned, the caller is responsible for freeing @item and its contents.
+ *
  * Returns: #TRUE if the @item was successfully pushed on the @queue.
  */
 gboolean
diff --git a/libs/gst/base/gstdataqueue.h b/libs/gst/base/gstdataqueue.h
index b3cd87b554..503e44c74d 100644
--- a/libs/gst/base/gstdataqueue.h
+++ b/libs/gst/base/gstdataqueue.h
@@ -48,7 +48,9 @@ typedef struct _GstDataQueueItem GstDataQueueItem;
  * @duration: the duration in #GstClockTime of the miniobject. Can not be
  * #GST_CLOCK_TIME_NONE.
  * @visible: #TRUE if @object should be considered as a visible object.
- * @destroy: The #GDestroyNotify to use to free the #GstDataQueueItem.
+ * @destroy: The #GDestroyNotify function to use to free the #GstDataQueueItem.
+ * This function should also drop the reference to @object the owner of the
+ * #GstDataQueueItem is assumed to hold.
  *
  * Structure used by #GstDataQueue. You can supply a different structure, as
  * long as the top of the structure is identical to this structure.
diff --git a/plugins/elements/gstmultiqueue.c b/plugins/elements/gstmultiqueue.c
index 3aca642669..d4208af11f 100644
--- a/plugins/elements/gstmultiqueue.c
+++ b/plugins/elements/gstmultiqueue.c
@@ -427,7 +427,10 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
     GstMiniObject * object)
 {
   if (GST_IS_BUFFER (object)) {
-    sq->srcresult = gst_pad_push (sq->srcpad, GST_BUFFER (object));
+    GstBuffer *buf;
+
+    buf = gst_buffer_ref (GST_BUFFER_CAST (object));
+    sq->srcresult = gst_pad_push (sq->srcpad, buf);
 
     if ((sq->srcresult != GST_FLOW_OK)
         && (sq->srcresult != GST_FLOW_NOT_LINKED)) {
@@ -437,7 +440,10 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
       gst_pad_pause_task (sq->srcpad);
     }
   } else if (GST_IS_EVENT (object)) {
-    if (GST_EVENT_TYPE (object) == GST_EVENT_EOS) {
+    GstEvent *event;
+
+    event = gst_event_ref (GST_EVENT_CAST (object));
+    if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
       sq->srcresult = GST_FLOW_UNEXPECTED;
 
       GST_DEBUG_OBJECT (mq, "GstSingleQueue %d : pausing queue, got EOS",
@@ -445,7 +451,7 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
       gst_data_queue_set_flushing (sq->queue, TRUE);
       gst_pad_pause_task (sq->srcpad);
     }
-    gst_pad_push_event (sq->srcpad, GST_EVENT (object));
+    gst_pad_push_event (sq->srcpad, event);
   } else {
     g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
         sq->id);
@@ -461,13 +467,14 @@ gst_multi_queue_item_destroy (GstMultiQueueItem * item)
   g_free (item);
 }
 
+/* takes ownership of passed mini object! */
 static GstMultiQueueItem *
 gst_multi_queue_item_new (GstMiniObject * object)
 {
   GstMultiQueueItem *item;
 
   item = g_new0 (GstMultiQueueItem, 1);
-  item->object = gst_mini_object_ref (object);
+  item->object = object;
   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
 
   if (GST_IS_BUFFER (object)) {
@@ -621,7 +628,6 @@ gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
         sq->id, gst_flow_get_name (sq->srcresult));
     gst_multi_queue_item_destroy (item);
-    gst_buffer_unref (buffer);
     ret = sq->srcresult;
   }
 
@@ -718,7 +724,6 @@ gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
         sq->id, gst_flow_get_name (sq->srcresult));
     gst_multi_queue_item_destroy (item);
-    gst_event_unref (event);
   }
 
 done:
diff --git a/tests/check/elements/multiqueue.c b/tests/check/elements/multiqueue.c
index ddf01ae01f..e6826e2c31 100644
--- a/tests/check/elements/multiqueue.c
+++ b/tests/check/elements/multiqueue.c
@@ -259,11 +259,7 @@ multiqueue_suite (void)
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_simple_create_destroy);
   tcase_add_test (tc_chain, test_simple_pipeline);
-
-  if (0) {
-    /* FIXME: this leaks buffers, disabled for now */
-    tcase_add_test (tc_chain, test_simple_shutdown_while_running);
-  }
+  tcase_add_test (tc_chain, test_simple_shutdown_while_running);
 
   /* FIXME: test_request_pads() needs some more fixes, see comments there */
   tcase_add_test (tc_chain, test_request_pads);