Index: libs/gst/base/gstcollectpads.c
===================================================================
--- libs/gst/base/gstcollectpads.c	(revision 376)
+++ libs/gst/base/gstcollectpads.c	(working copy)
@@ -42,7 +42,7 @@
  *     performing a pull_range.
  *   </para></listitem>
  *   <listitem><para>
- *     When data is queued on all pads, the callback function is called.
+ *     When data is queued on all pads in waiting mode, the callback function is called.
  *   </para></listitem>
  *   <listitem><para>
  *     Data can be dequeued from the pad with the gst_collect_pads_pop() method.
@@ -67,6 +67,12 @@
  *     elements that start a #GstTask to drive the collect_pads. This feature is however
  *     not yet implemented.
  *   </para></listitem>
+ *   <listitem><para>
+ *     gst_collect_pads_set_waiting() sets a pad to waiting or non-waiting mode.
+ *     CollectPads element is not waiting for data to be collected on non-waiting pads.
+ *     Thus these pads may but need not have data when the callback is called.
+ *     All pads are in waiting mode by default.
+ *   </para></listitem>
  * </itemizedlist>
  *
  * Last reviewed on 2006-05-10 (0.10.6)
@@ -113,6 +119,9 @@
   pads->eospads = 0;
   pads->started = FALSE;
 
+  pads->func = NULL;
+  pads->user_data = NULL;
+
   /* members to manage the pad list */
   pads->abidata.ABI.pad_lock = g_mutex_new ();
   pads->abidata.ABI.pad_cookie = 0;
@@ -239,6 +248,36 @@
 }
 
 /**
+ * gst_collect_pads_set_event_function:
+ * @pads: the collectspads to use
+ * @func: the function to set
+ * @user_data: user data passed to the function
+ *
+ * Set the event callback function and user data that will be called after
+ * collectpads has processed and event originating from one of the collected
+ * pads.  Note that this callback is called without @pads LOCK held, contrary
+ * to the collected function.  As this lock should held when calling
+ * a number of CollectPads functions, it should be acquired if so (unusually)
+ * needed.
+ *
+ * MT safe.
+ */
+void
+gst_collect_pads_set_event_function (GstCollectPads *pads,
+    GstCollectPadsEventFunction func, gpointer user_data)
+{
+  g_return_if_fail (pads != NULL);
+  g_return_if_fail (GST_IS_COLLECT_PADS (pads));
+
+  GST_OBJECT_LOCK (pads);
+  /* FIXME: Ugly hacks as we are short on fields in GstCollectPads */
+  g_object_set_data (G_OBJECT (pads), "gst-collect-pads-event-func", func);
+  g_object_set_data (G_OBJECT (pads), "gst-collect-pads-event-user-data", 
+      user_data);
+  GST_OBJECT_UNLOCK (pads);
+}
+
+/**
  * gst_collect_pads_add_pad:
  * @pads: the collectspads to use
  * @pad: the pad to add
@@ -320,8 +359,7 @@
   data->buffer = NULL;
   data->pos = 0;
   gst_segment_init (&data->segment, GST_FORMAT_UNDEFINED);
-  data->abidata.ABI.flushing = FALSE;
-  data->abidata.ABI.new_segment = FALSE;
+  data->abidata.ABI.state = GST_COLLECT_PADS_STATE_WAITING;
   data->abidata.ABI.eos = FALSE;
   data->abidata.ABI.refcount = 1;
 
@@ -534,7 +572,10 @@
         GST_PAD_SET_FLUSHING (cdata->pad);
       else
         GST_PAD_UNSET_FLUSHING (cdata->pad);
-      cdata->abidata.ABI.flushing = flushing;
+      if (flushing)
+        cdata->abidata.ABI.state |= GST_COLLECT_PADS_STATE_FLUSHING;
+      else
+        cdata->abidata.ABI.state &= ~GST_COLLECT_PADS_STATE_FLUSHING;
       gst_collect_pads_clear (pads, cdata);
       GST_OBJECT_UNLOCK (cdata->pad);
     }
@@ -718,7 +759,8 @@
     data->buffer = NULL;
     data->pos = 0;
     /* one less pad with queued data now */
-    pads->queuedpads--;
+    if (data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING)
+      pads->queuedpads--;
   }
 
   GST_COLLECT_PADS_BROADCAST (pads);
@@ -890,6 +932,51 @@
   return flushsize;
 }
 
+/**
+ * gst_collect_pads_set_waiting:
+ * @pads: the collectspads
+ * @data: the data to use
+ * @waiting: boolean indicating whether this pad should operate
+ *           in waiting or non-waiting mode
+ *
+ * Sets a pad to waiting or non-waiting mode.
+ *
+ * This function should be called with @pads LOCK held, such as
+ * in the callback.
+ *
+ * MT safe.
+ */
+void
+gst_collect_pads_set_waiting (GstCollectPads * pads, GstCollectData * data,
+    gboolean waiting)
+{
+  g_return_if_fail (pads != NULL);
+  g_return_if_fail (GST_IS_COLLECT_PADS (pads));
+  g_return_if_fail (data != NULL);
+
+  GST_DEBUG ("Setting pad %s to waiting %d", GST_PAD_NAME (data->pad),
+      waiting);
+
+  /* Do something only on changes */
+  if (!!(data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING) != waiting) {
+    /* Set waiting state for this pad */
+    if (waiting)
+      data->abidata.ABI.state |= GST_COLLECT_PADS_STATE_WAITING;
+    else
+      data->abidata.ABI.state &= ~GST_COLLECT_PADS_STATE_WAITING;
+    /* Update number of queued pads if needed */
+    if (!data->buffer && !data->abidata.ABI.eos) {
+      if (waiting)
+        pads->queuedpads--;
+      else
+        pads->queuedpads++;
+    }
+
+    /* signal waiters because something changed */
+    GST_COLLECT_PADS_BROADCAST (pads);
+  }
+}
+
 /* see if pads were added or removed and update our stats. Any pad
  * added after releasing the PAD_LOCK will get collected in the next
  * round.
@@ -922,10 +1009,11 @@
       /* update the stats */
       pads->numpads++;
       data = collected->data;
-      if (data->buffer)
-        pads->queuedpads++;
       if (data->abidata.ABI.eos)
         pads->eospads++;
+      else if (data->buffer ||
+               !(data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING))
+        pads->queuedpads++;
 
       /* add to the list of pads to collect */
       pads->data = g_slist_prepend (pads->data, data);
@@ -991,9 +1079,11 @@
 static gboolean
 gst_collect_pads_event (GstPad * pad, GstEvent * event)
 {
-  gboolean res;
+  gboolean res = FALSE;
   GstCollectData *data;
   GstCollectPads *pads;
+  GstCollectPadsEventFunction event_func;
+  gpointer event_user_data;
 
   /* some magic to get the managing collect_pads */
   GST_OBJECT_LOCK (pad);
@@ -1010,42 +1100,60 @@
   GST_DEBUG ("Got %s event on pad %s:%s", GST_EVENT_TYPE_NAME (event),
       GST_DEBUG_PAD_NAME (data->pad));
 
+  /* get a local and consistent view on event callback */
+  /* FIXME should need object LOCK to be really consistent in extreme cases,
+   * but that could deadlock as collectpads also (ab)uses
+   * LOCK for some inter-pad execution serialization,
+   * e.g. holds this throughout collected callback,
+   * which could end up quite far (up to sinks)
+   * In short, no simple object property protecting lock left */
+  event_func = g_object_get_data (G_OBJECT (pads),
+      "gst-collect-pads-event-func");
+  event_user_data = g_object_get_data (G_OBJECT (pads),
+      "gst-collect-pads-event-user-data");
+
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_FLUSH_START:
     {
       /* forward event to unblock check_collected */
-      gst_pad_event_default (pad, event);
+      if (event_func)
+        res = event_func (pads, data, event, event_user_data);
+      if (!res)
+        res = gst_pad_event_default (pad, event);
 
       /* now unblock the chain function.
        * no cond per pad, so they all unblock, 
        * non-flushing block again */
       GST_OBJECT_LOCK (pads);
-      data->abidata.ABI.flushing = TRUE;
+      data->abidata.ABI.state |= GST_COLLECT_PADS_STATE_FLUSHING;
       gst_collect_pads_clear (pads, data);
       GST_OBJECT_UNLOCK (pads);
 
       /* event already cleaned up by forwarding */
+      res = TRUE;
       goto done;
     }
     case GST_EVENT_FLUSH_STOP:
     {
       /* flush the 1 buffer queue */
       GST_OBJECT_LOCK (pads);
-      data->abidata.ABI.flushing = FALSE;
+      data->abidata.ABI.state &= ~GST_COLLECT_PADS_STATE_FLUSHING;
       gst_collect_pads_clear (pads, data);
       /* we need new segment info after the flush */
       gst_segment_init (&data->segment, GST_FORMAT_UNDEFINED);
-      data->abidata.ABI.new_segment = FALSE;
+      data->abidata.ABI.state &= ~GST_COLLECT_PADS_STATE_NEW_SEGMENT;
       /* if the pad was EOS, remove the EOS flag and
        * decrement the number of eospads */
       if (G_UNLIKELY (data->abidata.ABI.eos == TRUE)) {
+        if (!(data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING))
+          pads->queuedpads++;
         pads->eospads--;
         data->abidata.ABI.eos = FALSE;
       }
       GST_OBJECT_UNLOCK (pads);
 
       /* forward event */
-      goto forward;
+      goto forward_or_default;
     }
     case GST_EVENT_EOS:
     {
@@ -1054,6 +1162,8 @@
        * have one more eospad */
       if (G_LIKELY (data->abidata.ABI.eos == FALSE)) {
         data->abidata.ABI.eos = TRUE;
+        if (!(data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING))
+          pads->queuedpads--;
         pads->eospads++;
       }
       /* check if we need collecting anything, we ignore the
@@ -1061,10 +1171,7 @@
       gst_collect_pads_check_collected (pads);
       GST_OBJECT_UNLOCK (pads);
 
-      /* We eat this event, element should do something
-       * in the collected callback. */
-      gst_event_unref (event);
-      goto done;
+      goto forward_or_eat;
     }
     case GST_EVENT_NEWSEGMENT:
     {
@@ -1083,31 +1190,33 @@
       gst_segment_set_newsegment_full (&data->segment, update, rate, arate,
           format, start, stop, time);
 
-      data->abidata.ABI.new_segment = TRUE;
+      data->abidata.ABI.state |= GST_COLLECT_PADS_STATE_NEW_SEGMENT;
 
       /* we must not forward this event since multiple segments will be 
        * accumulated and this is certainly not what we want. */
-      gst_event_unref (event);
-      /* FIXME: collect-pads based elements need to create their own newsegment
-       * event (and only one really)
-       * (a) make the segment part of the GstCollectData structure of each pad,
-       * so you can just check that once you have a buffer queued on that pad,
-       * (b) you can override a pad's event function with your own,
-       * catch the newsegment event and then pass it on to the original
-       * gstcollectpads event function
-       * (that's what avimux does for something IIRC)
-       * see #340060
-       */
-      goto done;
+      goto forward_or_eat;
     }
     default:
       /* forward other events */
-      goto forward;
+      goto forward_or_default;
   }
 
-forward:
-  res = gst_pad_event_default (pad, event);
+forward_or_default:
+  if (event_func)
+    res = event_func (pads, data, event, event_user_data);
+  if (!res)
+    res = gst_pad_event_default (pad, event);
+  goto done;
 
+forward_or_eat:
+  if (event_func)
+    res = event_func (pads, data, event, event_user_data);
+  if (!res) {
+    gst_event_unref (event);
+    res = TRUE;
+  }
+  goto done;
+
 done:
   unref_data (data);
   return res;
@@ -1154,7 +1263,7 @@
   if (G_UNLIKELY (!pads->started))
     goto not_started;
   /* check if this pad is flushing */
-  if (G_UNLIKELY (data->abidata.ABI.flushing))
+  if (G_UNLIKELY (data->abidata.ABI.state & GST_COLLECT_PADS_STATE_FLUSHING))
     goto flushing;
   /* pad was EOS, we can refuse this data */
   if (G_UNLIKELY (data->abidata.ABI.eos))
@@ -1164,7 +1273,8 @@
       GST_DEBUG_PAD_NAME (pad));
 
   /* One more pad has data queued */
-  pads->queuedpads++;
+  if (data->abidata.ABI.state & GST_COLLECT_PADS_STATE_WAITING)
+      pads->queuedpads++;
   buffer_p = &data->buffer;
   gst_buffer_replace (buffer_p, buffer);
 
@@ -1214,7 +1324,7 @@
     if (G_UNLIKELY (!pads->started))
       goto not_started;
     /* check if this pad is flushing */
-    if (G_UNLIKELY (data->abidata.ABI.flushing))
+    if (G_UNLIKELY (data->abidata.ABI.state & GST_COLLECT_PADS_STATE_FLUSHING))
       goto flushing;
   }
   while (data->buffer != NULL);
Index: libs/gst/base/gstcollectpads.h
===================================================================
--- libs/gst/base/gstcollectpads.h	(revision 376)
+++ libs/gst/base/gstcollectpads.h	(working copy)
@@ -50,6 +50,22 @@
 typedef void (*GstCollectDataDestroyNotify) (GstCollectData *data);
 
 /**
+ * GstCollectPadsStateFlags:
+ * @GST_COLLECT_PADS_STATE_FLUSHING:   Set if collectdata's pad is flushing.
+ * @GST_COLLECT_PADS_STATE_FLUSHING:   Set if collectdata's pad received a 
+ *                                     new_segment event.
+ * @GST_COLLECT_PADS_STATE_WAITING:    Set if collectdata's pad must be waited 
+ *                                     for when collecting.
+ *
+ * #GstCollectPadsStateFlags indicate private state of a collectdata('s pad).
+ */
+typedef enum {
+  GST_COLLECT_PADS_STATE_FLUSHING = 1 << 0,
+  GST_COLLECT_PADS_STATE_NEW_SEGMENT = 1 << 1,
+  GST_COLLECT_PADS_STATE_WAITING = 1 << 2
+} GstCollectPadsStateFlags;
+
+/**
  * GstCollectData:
  * @collect: owner #GstCollectPads
  * @pad: #GstPad managed by this data
@@ -71,8 +87,9 @@
   /*< private >*/
   union {
     struct {
-      gboolean           flushing;
-      gboolean           new_segment;
+      /* bitfield to save on space and performance; 
+       * flushing, new_segment, waiting */
+      guint              state;
       gboolean           eos;
       gint               refcount;
     } ABI;
@@ -96,6 +113,19 @@
 #define GST_COLLECT_PADS_PAD_LOCK(pads)     (g_mutex_lock(GST_COLLECT_PADS_GET_PAD_LOCK (pads)))
 #define GST_COLLECT_PADS_PAD_UNLOCK(pads)   (g_mutex_unlock(GST_COLLECT_PADS_GET_PAD_LOCK (pads)))
 
+/**
+ * GstCollectPadsEventFunction:
+ * @pads: the #GstCollectPads that trigered the callback
+ * @pad: the @GstPad that received an event
+ * @event: the #GstEvent received
+ * @user_data: user data passed to gst_collect_pads_set_event_function()
+ *
+ * A function that will be called after collectpads has processed the event.
+ *
+ * Returns: %TRUE if the pad could handle the event
+ */
+typedef gboolean (*GstCollectPadsEventFunction) (GstCollectPads *pads, GstCollectData * pad, GstEvent * event, gpointer user_data);
+
 #define GST_COLLECT_PADS_GET_COND(pads) (((GstCollectPads *)pads)->cond)
 #define GST_COLLECT_PADS_WAIT(pads)     (g_cond_wait (GST_COLLECT_PADS_GET_COND (pads), GST_OBJECT_GET_LOCK (pads)))
 #define GST_COLLECT_PADS_SIGNAL(pads)   (g_cond_signal (GST_COLLECT_PADS_GET_COND (pads)))
@@ -155,10 +185,13 @@
 /* creating the object */
 GstCollectPads*	gst_collect_pads_new	 	(void);
 
-/* set the callback */
+/* set the callbacks */
 void		gst_collect_pads_set_function 	(GstCollectPads *pads, GstCollectPadsFunction func,
 						 gpointer user_data);
 
+void            gst_collect_pads_set_event_function (GstCollectPads *pads, GstCollectPadsEventFunction func,
+						 gpointer user_data);
+
 /* pad management */
 GstCollectData*	gst_collect_pads_add_pad	(GstCollectPads *pads, GstPad *pad, guint size);
 GstCollectData* gst_collect_pads_add_pad_full   (GstCollectPads *pads, GstPad *pad, guint size, GstCollectDataDestroyNotify destroy_notify);
@@ -184,6 +217,11 @@
 guint 		gst_collect_pads_flush 		(GstCollectPads *pads, GstCollectData *data,
 						 guint size);
 
+/* setting and unsetting waiting mode */
+void		gst_collect_pads_set_waiting	(GstCollectPads *pads, GstCollectData *data,
+						 gboolean waiting);
+
+
 G_END_DECLS
 
 #endif /* __GST_COLLECT_PADS_H__ */
Index: docs/libs/gstreamer-libs-sections.txt
===================================================================
--- docs/libs/gstreamer-libs-sections.txt	(revision 424)
+++ docs/libs/gstreamer-libs-sections.txt	(working copy)
@@ -243,9 +243,11 @@
 GstCollectData
 GstCollectPads
 GstCollectPadsFunction
+GstCollectPadsEventFunction
 GstCollectDataDestroyNotify
 gst_collect_pads_new
 gst_collect_pads_set_function
+gst_collect_pads_set_event_function
 gst_collect_pads_add_pad
 gst_collect_pads_add_pad_full
 gst_collect_pads_remove_pad
@@ -255,6 +257,7 @@
 gst_collect_pads_start
 gst_collect_pads_stop
 gst_collect_pads_set_flushing
+gst_collect_pads_set_waiting
 gst_collect_pads_peek
 gst_collect_pads_pop
 gst_collect_pads_available
@@ -270,6 +273,7 @@
 GST_COLLECT_PADS_GET_CLASS
 <SUBSECTION Private>
 gst_collect_pads_get_type
+GstCollectPadsStateFlags
 GST_COLLECT_PADS_GET_PAD_LOCK
 GST_COLLECT_PADS_PAD_LOCK
 GST_COLLECT_PADS_PAD_UNLOCK
