gst/gstpad.c: implement enforcing discont events before buffers are passed. This allows state changes of only some el...

Original commit message from CVS:
* gst/gstpad.c: (gst_pad_set_active), (gst_pad_link_new),
(gst_pad_link_free), (gst_pad_link_try), (_invent_event),
(gst_pad_pull):
implement enforcing discont events before buffers are passed. This
allows state changes of only some elements and later correctly going
on where they left off (or in short: you can now set audio sinks to
NULL to release the device when the pipeline is paused)
* gst/gstpad.c: (gst_pad_call_chain_function),
(gst_pad_call_get_function):
* gst/gstpad.h:
add gst_pad_call_chain_function and gst_pad_call_get_function for
scheduler interaction. They are required because of the changes
above.
* gst/schedulers/entryscheduler.c: (get_buffer),
(gst_entry_scheduler_chain_wrapper),
(gst_entry_scheduler_get_wrapper),
(gst_entry_scheduler_state_transition),
(gst_entry_scheduler_pad_link):
* gst/schedulers/gstbasicscheduler.c:
(gst_basic_scheduler_chain_wrapper),
(gst_basic_scheduler_src_wrapper),
(gst_basic_scheduler_chainhandler_proxy),
(gst_basic_scheduler_gethandler_proxy),
(gst_basic_scheduler_cothreaded_chain),
(gst_basic_scheduler_chain_elements):
* gst/schedulers/gstoptimalscheduler.c:
(get_group_schedule_function), (pad_clear_queued),
(gst_opt_scheduler_pad_link):
use the new functions instead of calling get/chain-functions
directly.

ChangeLog

@@ -1,3 +1,36 @@
+2004-04-16  Benjamin Otte  <otte@gnome.org>
+
+	* gst/gstpad.c: (gst_pad_set_active), (gst_pad_link_new),
+	(gst_pad_link_free), (gst_pad_link_try), (_invent_event),
+	(gst_pad_pull):
+	  implement enforcing discont events before buffers are passed. This
+	  allows state changes of only some elements and later correctly going
+	  on where they left off (or in short: you can now set audio sinks to
+	  NULL to release the device when the pipeline is paused)
+	* gst/gstpad.c: (gst_pad_call_chain_function),
+	(gst_pad_call_get_function):
+	* gst/gstpad.h:
+	  add gst_pad_call_chain_function and gst_pad_call_get_function for
+	  scheduler interaction. They are required because of the changes
+	  above.
+	* gst/schedulers/entryscheduler.c: (get_buffer),
+	(gst_entry_scheduler_chain_wrapper),
+	(gst_entry_scheduler_get_wrapper),
+	(gst_entry_scheduler_state_transition),
+	(gst_entry_scheduler_pad_link):
+	* gst/schedulers/gstbasicscheduler.c:
+	(gst_basic_scheduler_chain_wrapper),
+	(gst_basic_scheduler_src_wrapper),
+	(gst_basic_scheduler_chainhandler_proxy),
+	(gst_basic_scheduler_gethandler_proxy),
+	(gst_basic_scheduler_cothreaded_chain),
+	(gst_basic_scheduler_chain_elements):
+	* gst/schedulers/gstoptimalscheduler.c:
+	(get_group_schedule_function), (pad_clear_queued),
+	(gst_opt_scheduler_pad_link):
+	  use the new functions instead of calling get/chain-functions
+	  directly.
+
 2004-04-15  David Schleef  <ds@schleef.org>
 
 	* docs/gst/gstreamer-sections.txt: Remove deprecated symbols.

gst/gstpad.c

@@ -36,6 +36,28 @@
 #define GST_CAT_DEFAULT GST_CAT_PADS
 
 
+struct _GstPadLink
+{
+  GType type;
+
+  gboolean bla;
+  gboolean srcnotify;
+  gboolean sinknotify;
+
+  GstPad *srcpad;
+  GstPad *sinkpad;
+
+  GstCaps *srccaps;
+  GstCaps *sinkcaps;
+  GstCaps *filtercaps;
+  GstCaps *caps;
+
+  GstPadFixateFunction app_fixate;
+
+  gboolean engaged;
+  GstData *temp_store;          /* used only when we invented a DISCONT */
+};
+
 enum
 {
   TEMPL_PAD_CREATED,
@@ -412,6 +434,7 @@ gst_pad_set_active (GstPad * pad, gboolean active)
 {
   GstRealPad *realpad;
   gboolean old;
+  GstPadLink *link;
 
   g_return_if_fail (GST_IS_PAD (pad));
 
@@ -431,6 +454,17 @@ gst_pad_set_active (GstPad * pad, gboolean active)
         GST_DEBUG_PAD_NAME (realpad));
     GST_FLAG_SET (realpad, GST_PAD_DISABLED);
   }
+  link = GST_RPAD_LINK (realpad);
+  if (link) {
+    link->engaged = FALSE;
+    if (link->temp_store) {
+      GST_CAT_INFO (GST_CAT_PADS,
+          "deleting cached buffer from bufpen of pad %s:%s",
+          GST_DEBUG_PAD_NAME (realpad));
+      gst_data_unref (link->temp_store);
+      link->temp_store = NULL;
+    }
+  }
 
   g_object_notify (G_OBJECT (realpad), "active");
 }
@@ -999,25 +1033,6 @@ gst_pad_is_linked (GstPad * pad)
   return GST_PAD_PEER (pad) != NULL;
 }
 
-struct _GstPadLink
-{
-  GType type;
-
-  gboolean bla;
-  gboolean srcnotify;
-  gboolean sinknotify;
-
-  GstPad *srcpad;
-  GstPad *sinkpad;
-
-  GstCaps *srccaps;
-  GstCaps *sinkcaps;
-  GstCaps *filtercaps;
-  GstCaps *caps;
-
-  GstPadFixateFunction app_fixate;
-};
-
 static gboolean
 gst_pad_check_schedulers (GstRealPad * realsrc, GstRealPad * realsink)
 {
@@ -1051,6 +1066,8 @@ gst_pad_link_new (void)
   link = g_new0 (GstPadLink, 1);
   link->sinknotify = TRUE;
   link->srcnotify = TRUE;
+
+  link->engaged = FALSE;
   return link;
 }
 
@@ -1065,6 +1082,8 @@ gst_pad_link_free (GstPadLink * link)
     gst_caps_free (link->filtercaps);
   if (link->caps)
     gst_caps_free (link->caps);
+  if (link->temp_store)
+    gst_data_unref (link->temp_store);
 #ifdef USE_POISONING
   memset (link, 0xff, sizeof (*link));
 #endif
@@ -1316,8 +1335,12 @@ gst_pad_link_try (GstPadLink * link)
 
   GST_RPAD_PEER (srcpad) = GST_REAL_PAD (link->sinkpad);
   GST_RPAD_PEER (sinkpad) = GST_REAL_PAD (link->srcpad);
-  if (oldlink)
+  if (oldlink) {
+    link->temp_store = oldlink->temp_store;
+    link->engaged = oldlink->engaged;
+    oldlink->temp_store = NULL;
     gst_pad_link_free (oldlink);
+  }
   GST_RPAD_LINK (srcpad) = link;
   GST_RPAD_LINK (sinkpad) = link;
   if (ret == GST_PAD_LINK_OK) {
@@ -3001,6 +3024,31 @@ gst_ghost_pad_save_thyself (GstPad * pad, xmlNodePtr parent)
 }
 #endif /* GST_DISABLE_LOADSAVE */
 
+static GstData *
+_invent_event (GstPad * pad, GstBuffer * buffer)
+{
+  GstData *data;
+
+  if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) {
+    data = GST_DATA (gst_event_new_discontinuous (TRUE, GST_FORMAT_TIME,
+            GST_BUFFER_TIMESTAMP (buffer), GST_BUFFER_OFFSET_IS_VALID (buffer) ?
+            GST_FORMAT_DEFAULT : 0, GST_BUFFER_OFFSET (buffer), 0));
+    GST_CAT_WARNING (GST_CAT_DATAFLOW,
+        "needed to invent a DISCONT (time %" G_GUINT64_FORMAT
+        ") for %s:%s => %s:%s", GST_BUFFER_TIMESTAMP (buffer),
+        GST_DEBUG_PAD_NAME (GST_PAD_PEER (pad)), GST_DEBUG_PAD_NAME (pad));
+  } else {
+    data = GST_DATA (gst_event_new_discontinuous (TRUE,
+            GST_BUFFER_OFFSET_IS_VALID (buffer) ? GST_FORMAT_DEFAULT : 0,
+            GST_BUFFER_OFFSET (buffer), 0));
+    GST_CAT_WARNING_OBJECT (GST_CAT_DATAFLOW,
+        "needed to invent a DISCONT (no time) for %s:%s => %s:%s",
+        GST_DEBUG_PAD_NAME (GST_PAD_PEER (pad)), GST_DEBUG_PAD_NAME (pad));
+  }
+
+  return data;
+}
+
 /**
  * gst_pad_push:
  * @pad: a source #GstPad.
@@ -3094,15 +3142,38 @@ gst_pad_pull (GstPad * pad)
   } else {
   restart:
     if (peer->gethandler) {
+      GstPadLink *link = GST_RPAD_LINK (pad);
       GstData *data;
 
       GST_CAT_LOG_OBJECT (GST_CAT_DATAFLOW, pad,
           "calling gethandler %s of peer pad %s:%s",
           GST_DEBUG_FUNCPTR_NAME (peer->gethandler), GST_DEBUG_PAD_NAME (peer));
 
-      data = (peer->gethandler) (GST_PAD (peer));
+      if (link->temp_store) {
+        g_assert (link->engaged);
+        data = link->temp_store;
+        link->temp_store = NULL;
+      } else {
+        data = (peer->gethandler) (GST_PAD (peer));
+        /* refetch - we might have been relinked */
+        link = GST_RPAD_LINK (pad);
+      }
 
       if (data) {
+        if (!link->engaged) {
+          g_assert (link->temp_store == NULL);
+          if (GST_IS_BUFFER (data)) {
+            link->temp_store = data;
+            link->engaged = TRUE;
+            data = _invent_event (pad, GST_BUFFER (data));
+          } else if (GST_IS_EVENT (data) &&
+              GST_EVENT_TYPE (data) == GST_EVENT_DISCONTINUOUS) {
+            link->engaged = TRUE;
+            GST_CAT_LOG (GST_CAT_DATAFLOW,
+                "link engaged by discont event for pad %s:%s",
+                GST_DEBUG_PAD_NAME (pad));
+          }
+        }
         if (!gst_probe_dispatcher_dispatch (&peer->probedisp, &data))
           goto restart;
         return data;
@@ -4140,3 +4211,71 @@ gst_pad_get_formats (GstPad * pad)
 
   return NULL;
 }
+
+#define CALL_CHAINFUNC(pad, data) G_STMT_START {\
+  if (GST_IS_EVENT (data) && \
+      !GST_FLAG_IS_SET (gst_pad_get_parent (pad), GST_ELEMENT_EVENT_AWARE)) { \
+    gst_pad_send_event (pad, GST_EVENT (data)); \
+  } else { \
+    GST_RPAD_CHAINFUNC (pad) (pad, data); \
+  } \
+}G_STMT_END
+/**
+ * gst_pad_call_chain_function:
+ * @pad: sink pad to call chain function on
+ * @data: data to call the chain function with
+ *
+ * Calls the chain function of the given pad while making sure the internal
+ * consistency is kept. Use this function inside schedulers instead of calling
+ * the chain function yourself.
+ */
+void
+gst_pad_call_chain_function (GstPad * pad, GstData * data)
+{
+  GstPadLink *link;
+
+  g_return_if_fail (GST_IS_REAL_PAD (pad));
+  g_return_if_fail (GST_PAD_IS_SINK (pad));
+  g_return_if_fail (data != NULL);
+  g_return_if_fail (GST_RPAD_CHAINFUNC (pad) != NULL);
+  g_return_if_fail (GST_RPAD_LINK (pad) != NULL);
+
+  link = GST_RPAD_LINK (pad);
+  if (!link->engaged) {
+    g_assert (link->temp_store == NULL);
+    if (GST_IS_BUFFER (data)) {
+      link->temp_store = data;
+      link->engaged = TRUE;
+      CALL_CHAINFUNC (pad, _invent_event (pad, GST_BUFFER (data)));
+      g_assert (link->temp_store == data);
+      link->temp_store = NULL;
+    } else if (GST_IS_EVENT (data) &&
+        GST_EVENT_TYPE (data) == GST_EVENT_DISCONTINUOUS) {
+      link->engaged = TRUE;
+      GST_CAT_LOG (GST_CAT_DATAFLOW,
+          "link engaged by discont event for pad %s:%s",
+          GST_DEBUG_PAD_NAME (pad));
+    }
+  }
+  CALL_CHAINFUNC (pad, data);
+}
+
+/**
+ * gst_pad_call_get_function:
+ * @pad: sink pad to call chain function on
+ *
+ * Calls the get function of the given pad while making sure the internal
+ * consistency is kept. Use this function inside schedulers instead of calling
+ * the get function yourself.
+ *
+ * Returns: the data provided by the pad or NULL if no data was available.
+ */
+GstData *
+gst_pad_call_get_function (GstPad * pad)
+{
+  g_return_val_if_fail (GST_IS_REAL_PAD (pad), NULL);
+  g_return_val_if_fail (GST_PAD_IS_SRC (pad), NULL);
+  g_return_val_if_fail (GST_RPAD_GETFUNC (pad) != NULL, NULL);
+
+  return GST_RPAD_GETFUNC (pad) (pad);
+}

gst/gstpad.h

@@ -518,6 +518,11 @@ xmlNodePtr              gst_ghost_pad_save_thyself   		(GstPad *pad,
 						     		 xmlNodePtr parent);
 #endif
 
+/* for schedulers only */
+void			gst_pad_call_chain_function		(GstPad *pad, GstData *data);
+GstData *		gst_pad_call_get_function		(GstPad *pad);
+
+
 G_END_DECLS
 
 

gst/schedulers/entryscheduler.c

@@ -114,8 +114,6 @@ typedef struct
   CothreadPrivate *sink;
   /* current data */
   GstData *bufpen;
-  /* if this link needs a discont - FIXME: remove when core ready for this */
-  gboolean need_discont;
 }
 LinkPrivate;
 
@@ -315,48 +313,14 @@ setup_loop (GstEntryScheduler * sched, GstElement * element)
  * CHAINBASED
  */
 
-/* this function returns the buffer currently in the bufpen
- * it's this complicated because we check that we don't need to invent 
- * a DISCONT event first. 
- * FIXME: The whole if part should go when the core supports this.
- */
 static GstData *
 get_buffer (GstEntryScheduler * sched, GstRealPad * pad)
 {
   LinkPrivate *priv = PAD_PRIVATE (pad);
-  GstData *data;
+  GstData *data = priv->bufpen;
 
-  g_assert (GST_PAD_IS_SINK (pad));
+  priv->bufpen = NULL;
-  if (priv->need_discont && GST_IS_BUFFER (priv->bufpen)) {
-    if (GST_BUFFER_TIMESTAMP_IS_VALID (priv->bufpen)) {
-      data =
-          GST_DATA (gst_event_new_discontinuous (TRUE, GST_FORMAT_TIME,
-              GST_BUFFER_TIMESTAMP (priv->bufpen),
-              GST_BUFFER_OFFSET_IS_VALID (priv->
-                  bufpen) ? GST_FORMAT_DEFAULT : 0,
-              GST_BUFFER_OFFSET (priv->bufpen), 0));
-      GST_WARNING_OBJECT (sched,
-          "needed to invent a DISCONT (time %" G_GUINT64_FORMAT
-          ") for %s:%s => %s:%s, fix it please",
-          GST_BUFFER_TIMESTAMP (priv->bufpen),
-          GST_DEBUG_PAD_NAME (GST_PAD_PEER (pad)), GST_DEBUG_PAD_NAME (pad));
-    } else {
-      data = GST_DATA (gst_event_new_discontinuous (TRUE,
-              GST_BUFFER_OFFSET_IS_VALID (priv->
-                  bufpen) ? GST_FORMAT_DEFAULT : 0,
-              GST_BUFFER_OFFSET (priv->bufpen), 0));
-      GST_WARNING_OBJECT (sched,
-          "needed to invent a DISCONT (no time) for %s:%s => %s:%s, fix it please",
-          GST_DEBUG_PAD_NAME (GST_PAD_PEER (pad)), GST_DEBUG_PAD_NAME (pad));
-    }
-    sched->schedule_now = g_list_prepend (sched->schedule_now, priv);
-  } else {
-    data = priv->bufpen;
-    priv->bufpen = NULL;
-  }
   g_assert (data);
-  if (GST_IS_EVENT (data) && GST_EVENT_TYPE (data) == GST_EVENT_DISCONTINUOUS)
-    priv->need_discont = FALSE;
   return data;
 }
 
@@ -378,12 +342,7 @@ gst_entry_scheduler_chain_wrapper (int argc, char **argv)
     if (pad->chainfunc) {
       GstData *data = get_buffer (priv->sched, pad);
 
-      if (GST_IS_EVENT (data)
+      gst_pad_call_chain_function (GST_PAD (pad), data);
-          && !GST_FLAG_IS_SET (element, GST_ELEMENT_EVENT_AWARE)) {
-        gst_pad_event_default (GST_PAD (pad), GST_EVENT (data));
-      } else {
-        pad->chainfunc (GST_PAD (pad), data);
-      }
       /* don't do anything after here with the pad, it might already be dead! 
          the element is still alive though */
     } else {
@@ -445,7 +404,7 @@ gst_entry_scheduler_get_wrapper (int argc, char **argv)
     GST_LOG_OBJECT (priv->sched, "calling getfunc for pad %s:%s",
         GST_DEBUG_PAD_NAME (pad));
     if (pad->getfunc) {
-      GstData *data = pad->getfunc (GST_PAD (pad));
+      GstData *data = gst_pad_call_get_function (GST_PAD (pad));
 
       /* make sure the pad still exists and is linked */
       if (!g_list_find (element->pads, pad)) {
@@ -470,7 +429,7 @@ gst_entry_scheduler_get_wrapper (int argc, char **argv)
           ("get-based element %s removed getfunc during processing",
               GST_OBJECT_NAME (element)));
     }
-    GST_LOG_OBJECT (priv->sched, "done calling chainfunc for element %s",
+    GST_LOG_OBJECT (priv->sched, "done calling getfunc for element %s",
         GST_ELEMENT_NAME (element));
 
     priv->wait = WAIT_FOR_PADS;
@@ -878,7 +837,6 @@ static GstElementStateReturn
 gst_entry_scheduler_state_transition (GstScheduler * scheduler,
     GstElement * element, gint transition)
 {
-  GList *list;
   GstEntryScheduler *sched = GST_ENTRY_SCHEDULER (scheduler);
 
   if (GST_FLAG_IS_SET (element, GST_ELEMENT_DECOUPLED))
@@ -889,15 +847,6 @@ gst_entry_scheduler_state_transition (GstScheduler * scheduler,
     case GST_STATE_NULL_TO_READY:
       break;
     case GST_STATE_READY_TO_PAUSED:
-      for (list = element->pads; list; list = g_list_next (list)) {
-        GstPad *pad = list->data;
-
-        if (!GST_IS_REAL_PAD (pad))
-          continue;
-        if (PAD_PRIVATE (pad)) {
-          PAD_PRIVATE (pad)->need_discont = TRUE;
-        }
-      }
       break;
     case GST_STATE_PAUSED_TO_PLAYING:
       break;
@@ -967,7 +916,6 @@ gst_entry_scheduler_pad_link (GstScheduler * scheduler, GstPad * srcpad,
 
   priv = g_new0 (LinkPrivate, 1);
   priv->entry.type = ENTRY_LINK;
-  priv->need_discont = TRUE;
   /* wrap srcpad */
   element = gst_pad_get_parent (srcpad);
   priv->srcpad = GST_REAL_PAD (srcpad);

gst/schedulers/gstbasicscheduler.c

@@ -362,7 +362,7 @@ gst_basic_scheduler_chain_wrapper (int argc, char **argv)
               GST_CAT_DEBUG (debug_dataflow,
                   "calling chain function of %s:%s %p", name,
                   GST_PAD_NAME (pad), data);
-              GST_RPAD_CHAINFUNC (realpad) (pad, data);
+              gst_pad_call_chain_function (GST_PAD (realpad), data);
               GST_CAT_DEBUG (debug_dataflow,
                   "calling chain function of element %s done", name);
             }
@@ -421,8 +421,7 @@ gst_basic_scheduler_src_wrapper (int argc, char **argv)
         inf_loop = FALSE;
         GST_CAT_DEBUG (debug_dataflow, "calling _getfunc for %s:%s",
             GST_DEBUG_PAD_NAME (realpad));
-        g_return_val_if_fail (GST_RPAD_GETFUNC (realpad) != NULL, 0);
+        data = gst_pad_call_get_function (GST_PAD (realpad));
-        data = GST_RPAD_GETFUNC (realpad) (GST_PAD (realpad));
         if (data) {
           GST_CAT_DEBUG (debug_dataflow, "calling gst_pad_push on pad %s:%s %p",
               GST_DEBUG_PAD_NAME (realpad), data);
@@ -691,11 +690,11 @@ gst_basic_scheduler_cothreaded_chain (GstBin * bin, GstSchedulerChain * chain)
             if (GST_RPAD_DIRECTION (peerpad) == GST_PAD_SINK) {
               GST_DEBUG ("copying chain func into push proxy for peer %s:%s",
                   GST_DEBUG_PAD_NAME (peerpad));
-              GST_RPAD_CHAINHANDLER (peerpad) = GST_RPAD_CHAINFUNC (peerpad);
+              GST_RPAD_CHAINHANDLER (peerpad) = gst_pad_call_chain_function;
             } else {
               GST_DEBUG ("copying get func into pull proxy for peer %s:%s",
                   GST_DEBUG_PAD_NAME (peerpad));
-              GST_RPAD_GETHANDLER (peerpad) = GST_RPAD_GETFUNC (peerpad);
+              GST_RPAD_GETHANDLER (peerpad) = gst_pad_call_get_function;
             }
           }
         }
@@ -709,11 +708,11 @@ gst_basic_scheduler_cothreaded_chain (GstBin * bin, GstSchedulerChain * chain)
         if (GST_RPAD_DIRECTION (pad) == GST_PAD_SINK) {
           GST_DEBUG ("copying chain function into push proxy for %s:%s",
               GST_DEBUG_PAD_NAME (pad));
-          GST_RPAD_CHAINHANDLER (pad) = GST_RPAD_CHAINFUNC (pad);
+          GST_RPAD_CHAINHANDLER (pad) = gst_pad_call_chain_function;
         } else {
           GST_DEBUG ("copying get function into pull proxy for %s:%s",
               GST_DEBUG_PAD_NAME (pad));
-          GST_RPAD_GETHANDLER (pad) = GST_RPAD_GETFUNC (pad);
+          GST_RPAD_GETHANDLER (pad) = gst_pad_call_get_function;
         }
       }
       /* otherwise we really are a cothread */

gst/schedulers/gstoptimalscheduler.c

@@ -261,7 +261,6 @@ static int unknown_group_schedule_function (int argc, char *argv[]);
  */
 static void gst_opt_scheduler_loop_wrapper (GstPad * sinkpad, GstData * data);
 static GstData *gst_opt_scheduler_get_wrapper (GstPad * srcpad);
-static void gst_opt_scheduler_chain_wrapper (GstPad * sinkpad, GstData * data);
 
 
 /*
@@ -1182,7 +1181,7 @@ get_group_schedule_function (int argc, char *argv[])
     GST_DEBUG ("doing get and push on pad \"%s:%s\" in group %p",
         GST_DEBUG_PAD_NAME (pad), group);
 
-    data = GST_RPAD_GETFUNC (pad) (pad);
+    data = gst_pad_call_get_function (pad);
     if (data) {
       if (GST_EVENT_IS_INTERRUPT (data)) {
         gst_event_unref (GST_EVENT (data));
@@ -1367,24 +1366,6 @@ gst_opt_scheduler_get_wrapper (GstPad * srcpad)
   return data;
 }
 
-/* this function is a chain wrapper for non-event-aware plugins,
- * it'll simply dispatch the events to the (default) event handler */
-static void
-gst_opt_scheduler_chain_wrapper (GstPad * sinkpad, GstData * data)
-{
-  if (GST_IS_EVENT (data)) {
-    gst_pad_send_event (sinkpad, GST_EVENT (data));
-  } else {
-    GST_RPAD_CHAINFUNC (sinkpad) (sinkpad, data);
-  }
-}
-
-static void
-clear_queued (GstData * data, gpointer user_data)
-{
-  gst_data_unref (data);
-}
-
 static void
 pad_clear_queued (GstPad * srcpad, gpointer user_data)
 {
@@ -1392,7 +1373,7 @@ pad_clear_queued (GstPad * srcpad, gpointer user_data)
 
   if (buflist) {
     GST_LOG ("need to clear some buffers");
-    g_list_foreach (buflist, (GFunc) clear_queued, NULL);
+    g_list_foreach (buflist, (GFunc) gst_data_unref, NULL);
     g_list_free (buflist);
     GST_PAD_BUFPEN (srcpad) = NULL;
   }
@@ -1908,11 +1889,8 @@ gst_opt_scheduler_pad_link (GstScheduler * sched, GstPad * srcpad,
       GST_LOG ("get to chain based link");
 
       /* setup get/chain handlers */
-      GST_RPAD_GETHANDLER (srcpad) = GST_RPAD_GETFUNC (srcpad);
+      GST_RPAD_GETHANDLER (srcpad) = gst_pad_call_get_function;
-      if (GST_ELEMENT_IS_EVENT_AWARE (sink_element))
+      GST_RPAD_CHAINHANDLER (sinkpad) = gst_pad_call_chain_function;
-        GST_RPAD_CHAINHANDLER (sinkpad) = GST_RPAD_CHAINFUNC (sinkpad);
-      else
-        GST_RPAD_CHAINHANDLER (sinkpad) = gst_opt_scheduler_chain_wrapper;
 
       /* the two elements should be put into the same group, 
        * this also means that they are in the same chain automatically */
@@ -1933,10 +1911,7 @@ gst_opt_scheduler_pad_link (GstScheduler * sched, GstPad * srcpad,
     case GST_OPT_CHAIN_TO_CHAIN:
       GST_LOG ("loop/chain to chain based link");
 
-      if (GST_ELEMENT_IS_EVENT_AWARE (sink_element))
+      GST_RPAD_CHAINHANDLER (sinkpad) = gst_pad_call_chain_function;
-        GST_RPAD_CHAINHANDLER (sinkpad) = GST_RPAD_CHAINFUNC (sinkpad);
-      else
-        GST_RPAD_CHAINHANDLER (sinkpad) = gst_opt_scheduler_chain_wrapper;
 
       /* the two elements should be put into the same group, this also means
        * that they are in the same chain automatically, in case of a loop-based
@@ -1948,7 +1923,7 @@ gst_opt_scheduler_pad_link (GstScheduler * sched, GstPad * srcpad,
     case GST_OPT_GET_TO_LOOP:
       GST_LOG ("get to loop based link");
 
-      GST_RPAD_GETHANDLER (srcpad) = GST_RPAD_GETFUNC (srcpad);
+      GST_RPAD_GETHANDLER (srcpad) = gst_pad_call_get_function;
 
       /* the two elements should be put into the same group, this also means
        * that they are in the same chain automatically, sink_element is