- Use the eventhandler instead of the event function to send events.

Original commit message from CVS:
- Use the eventhandler instead of the event function to send events.
- make the scheduler setup the eventhandlers
- intercept flush events on pads links that can potentially queue data
and flush it.
- some more debugging info in spider.

I can think of one case where this flush might fail: unconnected pads where
the scheduler has not set up the eventhandler yet. I'll come up with a
solution for that soon.

gst/autoplug/gstspideridentity.c

@@ -190,6 +190,7 @@ gst_spider_identity_chain (GstPad *pad, GstBuffer *buf)
 
   if ((ident->src != NULL) && (GST_PAD_PEER (ident->src) != NULL)) {
     /* g_print("pushing buffer %p (refcount %d - buffersize %d) to pad %s:%s\n", buf, GST_BUFFER_REFCOUNT (buf), GST_BUFFER_SIZE (buf), GST_DEBUG_PAD_NAME (ident->src)); */
+    GST_DEBUG (0, "push %p %lld", buf, GST_BUFFER_OFFSET (buf));
     gst_pad_push (ident->src, buf);
   } else if (GST_IS_BUFFER (buf)) {
     gst_buffer_unref (buf);
@@ -401,7 +402,6 @@ gst_spider_identity_src_loop (GstSpiderIdentity *ident)
     gst_spider_identity_dumb_loop (ident);
     return;
   }
-
   gst_element_interrupt (GST_ELEMENT (ident));
 }
 /* This loop function is only needed when typefinding.
@@ -525,7 +525,7 @@ gst_spider_identity_handle_src_event (GstPad *pad, GstEvent *event)
   gboolean res = TRUE;
   GstSpiderIdentity *ident;
 
-  GST_DEBUG (0, "spider_identity src_event\n");
+  GST_DEBUG (0, "spider_identity src_event");
 
   ident = GST_SPIDER_IDENTITY (gst_pad_get_parent (pad));
 

gst/gstpad.c

@@ -2871,8 +2871,8 @@ gst_pad_send_event (GstPad *pad, GstEvent *event)
   GST_DEBUG (GST_CAT_EVENT, "have event %d on pad %s:%s",
 		  GST_EVENT_TYPE (event), GST_DEBUG_PAD_NAME (rpad));
 
-  if (GST_RPAD_EVENTFUNC (rpad))
-    success = GST_RPAD_EVENTFUNC (rpad) (GST_PAD_CAST (rpad), event);
+  if (GST_RPAD_EVENTHANDLER (rpad))
+    success = GST_RPAD_EVENTHANDLER (rpad) (GST_PAD_CAST (rpad), event);
   else {
     GST_DEBUG (GST_CAT_EVENT, "there's no event function for pad %s:%s", 
 	       GST_DEBUG_PAD_NAME (rpad));

gst/schedulers/gstbasicscheduler.c

@@ -523,6 +523,42 @@ gst_basic_scheduler_gethandler_proxy (GstPad * pad)
   return buf;
 }
 
+static gboolean
+gst_basic_scheduler_eventhandler_proxy (GstPad *srcpad, GstEvent *event)
+{
+  gboolean flush;
+
+  GST_INFO (GST_CAT_SCHEDULING, "intercepting event %d on pad %s:%s",
+                      GST_EVENT_TYPE (event), GST_DEBUG_PAD_NAME (srcpad));
+
+  /* figure out if we need to flush */
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_FLUSH:
+      flush = TRUE;
+      break;
+    case GST_EVENT_SEEK:
+    case GST_EVENT_SEEK_SEGMENT:
+      flush = GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH;
+      break;
+    default:
+      flush = FALSE;
+      break;
+  }
+
+  if (flush) {
+    GST_INFO (GST_CAT_SCHEDULING, "event is flush");
+    GstData *data = GST_DATA (GST_RPAD_BUFPEN (srcpad));
+
+    if (data) {
+      GST_INFO (GST_CAT_SCHEDULING, "need to clear some buffers");
+      
+      gst_data_unref (data);
+      GST_RPAD_BUFPEN (srcpad) = NULL;
+    }
+  }
+  return GST_RPAD_EVENTFUNC (srcpad) (srcpad, event);
+}
+
 static gboolean
 gst_basic_scheduler_cothreaded_chain (GstBin * bin, GstSchedulerChain * chain)
 {
@@ -627,6 +663,8 @@ gst_basic_scheduler_cothreaded_chain (GstBin * bin, GstSchedulerChain * chain)
 	    }
 	  }
 	}
+	/* in any case we need to copy the eventfunc into the handler */
+	GST_RPAD_EVENTHANDLER (peerpad) = GST_RPAD_EVENTFUNC (peerpad);
       }
 
       /* if the element is DECOUPLED or outside the manager, we have to chain */
@@ -652,12 +690,16 @@ gst_basic_scheduler_cothreaded_chain (GstBin * bin, GstSchedulerChain * chain)
 	             "setting cothreaded push proxy for sinkpad %s:%s",
 	     GST_DEBUG_PAD_NAME (pad));
 	  GST_RPAD_CHAINHANDLER (pad) = GST_DEBUG_FUNCPTR (gst_basic_scheduler_chainhandler_proxy);
+          GST_RPAD_EVENTHANDLER (pad) = GST_RPAD_EVENTFUNC (pad);
 	}
 	else {
 	  GST_DEBUG (GST_CAT_SCHEDULING, 
 	             "setting cothreaded pull proxy for srcpad %s:%s",
 	     GST_DEBUG_PAD_NAME (pad));
 	  GST_RPAD_GETHANDLER (pad) = GST_DEBUG_FUNCPTR (gst_basic_scheduler_gethandler_proxy);
+	  /* the gethandler proxy function can queue a buffer in the bufpen, we need
+	   * to remove this buffer when a flush event is sent on the pad */
+          GST_RPAD_EVENTHANDLER (pad) = GST_DEBUG_FUNCPTR (gst_basic_scheduler_eventhandler_proxy);
 	}
       }
     }

gst/schedulers/gstoptimalscheduler.c

@@ -922,6 +922,48 @@ gst_opt_scheduler_chain_wrapper (GstPad *sinkpad, GstBuffer *buffer)
   }
 }
 
+static void
+clear_queued (GstData *data, gpointer user_data)
+{
+  gst_data_unref (data);
+}
+
+static gboolean
+gst_opt_scheduler_event_wrapper (GstPad *srcpad, GstEvent *event)
+{
+  gboolean flush;
+
+  GST_INFO (GST_CAT_SCHEDULING, "intercepting event %d on pad %s:%s", 
+		  GST_EVENT_TYPE (event), GST_DEBUG_PAD_NAME (srcpad));
+  
+  /* figure out if this is a flush event */
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_FLUSH:
+      flush = TRUE;
+      break;
+    case GST_EVENT_SEEK:
+    case GST_EVENT_SEEK_SEGMENT:
+      flush = GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH;
+      break;
+    default:
+      flush = FALSE;
+      break;
+  }
+
+  if (flush) {
+    GST_INFO (GST_CAT_SCHEDULING, "event is flush");
+    GList *buflist = GST_PAD_BUFLIST (srcpad);
+
+    if (buflist) {
+      GST_INFO (GST_CAT_SCHEDULING, "need to clear some buffers");
+      g_list_foreach (buflist, (GFunc) clear_queued, NULL);
+      g_list_free (buflist);
+      GST_PAD_BUFLIST (srcpad) = NULL;
+    }
+  }
+  return GST_RPAD_EVENTFUNC (srcpad) (srcpad, event);
+}
+
 /* setup the scheduler context for a group. The right schedule function
  * is selected based on the group type and cothreads are created if 
  * needed */
@@ -1352,6 +1394,8 @@ gst_opt_scheduler_pad_link (GstScheduler *sched, GstPad *srcpad, GstPad *sinkpad
         GST_RPAD_CHAINHANDLER (sinkpad) = GST_RPAD_CHAINFUNC (sinkpad);
       else
         GST_RPAD_CHAINHANDLER (sinkpad) = gst_opt_scheduler_chain_wrapper;
+      GST_RPAD_EVENTHANDLER (srcpad) = GST_RPAD_EVENTFUNC (srcpad);
+      GST_RPAD_EVENTHANDLER (sinkpad) = GST_RPAD_EVENTFUNC (sinkpad);
 
       /* the two elements should be put into the same group, 
        * this also means that they are in the same chain automatically */
@@ -1376,6 +1420,8 @@ gst_opt_scheduler_pad_link (GstScheduler *sched, GstPad *srcpad, GstPad *sinkpad
         GST_RPAD_CHAINHANDLER (sinkpad) = GST_RPAD_CHAINFUNC (sinkpad);
       else
         GST_RPAD_CHAINHANDLER (sinkpad) = gst_opt_scheduler_chain_wrapper;
+      GST_RPAD_EVENTHANDLER (srcpad) = GST_RPAD_EVENTFUNC (srcpad);
+      GST_RPAD_EVENTHANDLER (sinkpad) = GST_RPAD_EVENTFUNC (sinkpad);
 
       /* the two elements should be put into the same group, 
        * this also means that they are in the same chain automatically, 
@@ -1387,6 +1433,8 @@ gst_opt_scheduler_pad_link (GstScheduler *sched, GstPad *srcpad, GstPad *sinkpad
       GST_INFO (GST_CAT_SCHEDULING, "get to loop based link");
 
       GST_RPAD_GETHANDLER (srcpad) = GST_RPAD_GETFUNC (srcpad);
+      GST_RPAD_EVENTHANDLER (srcpad) = GST_RPAD_EVENTFUNC (srcpad);
+      GST_RPAD_EVENTHANDLER (sinkpad) = GST_RPAD_EVENTFUNC (sinkpad);
 
       /* the two elements should be put into the same group, 
        * this also means that they are in the same chain automatically, 
@@ -1403,6 +1451,10 @@ gst_opt_scheduler_pad_link (GstScheduler *sched, GstPad *srcpad, GstPad *sinkpad
 
       GST_RPAD_CHAINHANDLER (sinkpad) = gst_opt_scheduler_loop_wrapper;
       GST_RPAD_GETHANDLER (srcpad) = gst_opt_scheduler_get_wrapper;
+      GST_RPAD_EVENTHANDLER (sinkpad) = GST_RPAD_EVENTFUNC (sinkpad);
+      /* events on the srcpad have to be intercepted as we might need to
+       * flush the buffer lists */
+      GST_RPAD_EVENTHANDLER (srcpad) = gst_opt_scheduler_event_wrapper;
 
       group1 = GST_ELEMENT_SCHED_GROUP (element1);
       group2 = GST_ELEMENT_SCHED_GROUP (element2);