WIP basesink: rework event handlingwip-basesink-event

2 files changed, 131 insertions, 168 deletions

diff --git a/libs/gst/base/gstbasesink.c b/libs/gst/base/gstbasesink.c
index 2d4026bb8..0cccac79d 100644
--- a/libs/gst/base/gstbasesink.c
+++ b/libs/gst/base/gstbasesink.c
@@ -389,6 +389,8 @@ static void gst_base_sink_loop (GstPad * pad);
 static gboolean gst_base_sink_pad_activate (GstPad * pad, GstObject * parent);
 static gboolean gst_base_sink_pad_activate_mode (GstPad * pad,
     GstObject * parent, GstPadMode mode, gboolean active);
+static gboolean gst_base_sink_default_event (GstBaseSink * basesink,
+    GstEvent * event);
 static gboolean gst_base_sink_event (GstPad * pad, GstObject * parent,
     GstEvent * event);
 
@@ -546,6 +548,7 @@ gst_base_sink_class_init (GstBaseSinkClass * klass)
       GST_DEBUG_FUNCPTR (gst_base_sink_default_activate_pull);
   klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
   klass->query = GST_DEBUG_FUNCPTR (default_sink_query);
+  klass->event = GST_DEBUG_FUNCPTR (gst_base_sink_default_event);
 
   /* Registering debug symbols for function pointers */
   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_fixate);
@@ -1777,19 +1780,19 @@ gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
         /* EOS event needs syncing */
       case GST_EVENT_EOS:
       {
-        if (basesink->segment.rate >= 0.0) {
+        if (segment->rate >= 0.0) {
           sstart = sstop = priv->current_sstop;
           if (!GST_CLOCK_TIME_IS_VALID (sstart)) {
             /* we have not seen a buffer yet, use the segment values */
-            sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
-                basesink->segment.format, basesink->segment.stop);
+            sstart = sstop = gst_segment_to_stream_time (segment,
+                segment->format, segment->stop);
           }
         } else {
           sstart = sstop = priv->current_sstart;
           if (!GST_CLOCK_TIME_IS_VALID (sstart)) {
             /* we have not seen a buffer yet, use the segment values */
-            sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
-                basesink->segment.format, basesink->segment.start);
+            sstart = sstop = gst_segment_to_stream_time (segment,
+                segment->format, segment->start);
           }
         }
 
@@ -2008,8 +2011,8 @@ gst_base_sink_wait_clock (GstBaseSink * sink, GstClockTime time,
   /* FIXME: Casting to GstClockEntry only works because the types
    * are the same */
   if (G_LIKELY (sink->priv->cached_clock_id != NULL
-          && GST_CLOCK_ENTRY_CLOCK ((GstClockEntry *) sink->priv->
-              cached_clock_id) == clock)) {
+          && GST_CLOCK_ENTRY_CLOCK ((GstClockEntry *) sink->
+              priv->cached_clock_id) == clock)) {
     if (!gst_clock_single_shot_id_reinit (clock, sink->priv->cached_clock_id,
             time)) {
       gst_clock_id_unref (sink->priv->cached_clock_id);
@@ -2276,7 +2279,7 @@ flushing:
  * does not take ownership of obj.
  */
 static GstFlowReturn
-gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
+gst_base_sink_do_sync (GstBaseSink * basesink,
     GstMiniObject * obj, gboolean * late, gboolean * step_end, guint8 obj_type)
 {
   GstClockTimeDiff jitter = 0;
@@ -2764,7 +2767,7 @@ gst_base_sink_do_render_stats (GstBaseSink * basesink, gboolean start)
  * takes ownership of obj.
  */
 static GstFlowReturn
-gst_base_sink_render_object (GstBaseSink * basesink, GstPad * pad,
+gst_base_sink_render_object (GstBaseSink * basesink,
     guint8 obj_type, gpointer obj)
 {
   GstFlowReturn ret;
@@ -2776,10 +2779,8 @@ gst_base_sink_render_object (GstBaseSink * basesink, GstPad * pad,
   priv = basesink->priv;
 
   if (OBJ_IS_BUFFERLIST (obj_type)) {
-    /*
-     * If buffer list, use the first group buffer within the list
-     * for syncing
-     */
+    /* If buffer list, use the first group buffer within the list
+     * for syncing */
     sync_obj = gst_buffer_list_get (GST_BUFFER_LIST_CAST (obj), 0);
     g_assert (NULL != sync_obj);
   } else {
@@ -2792,9 +2793,7 @@ again:
 
   /* synchronize this object, non syncable objects return OK
    * immediately. */
-  ret =
-      gst_base_sink_do_sync (basesink, pad, sync_obj, &late, &step_end,
-      obj_type);
+  ret = gst_base_sink_do_sync (basesink, sync_obj, &late, &step_end, obj_type);
   if (G_UNLIKELY (ret != GST_FLOW_OK))
     goto sync_failed;
 
@@ -2848,91 +2847,6 @@ again:
 
       priv->rendered++;
     }
-  } else if (G_LIKELY (OBJ_IS_EVENT (obj_type))) {
-    GstEvent *event = GST_EVENT_CAST (obj);
-    gboolean event_res = TRUE;
-    GstEventType type;
-
-    bclass = GST_BASE_SINK_GET_CLASS (basesink);
-
-    type = GST_EVENT_TYPE (event);
-
-    GST_DEBUG_OBJECT (basesink, "rendering event %p, type %s", obj,
-        gst_event_type_get_name (type));
-
-    if (bclass->event)
-      event_res = bclass->event (basesink, event);
-
-    /* when we get here we could be flushing again when the event handler calls
-     * _wait_eos(). We have to ignore this object in that case. */
-    if (G_UNLIKELY (basesink->flushing))
-      goto flushing;
-
-    if (G_LIKELY (event_res)) {
-      guint32 seqnum;
-
-      seqnum = basesink->priv->seqnum = gst_event_get_seqnum (event);
-      GST_DEBUG_OBJECT (basesink, "Got seqnum #%" G_GUINT32_FORMAT, seqnum);
-
-      switch (type) {
-        case GST_EVENT_EOS:
-        {
-          GstMessage *message;
-
-          /* the EOS event is completely handled so we mark
-           * ourselves as being in the EOS state. eos is also
-           * protected by the object lock so we can read it when
-           * answering the POSITION query. */
-          GST_OBJECT_LOCK (basesink);
-          basesink->eos = TRUE;
-          GST_OBJECT_UNLOCK (basesink);
-
-          /* ok, now we can post the message */
-          GST_DEBUG_OBJECT (basesink, "Now posting EOS");
-
-          message = gst_message_new_eos (GST_OBJECT_CAST (basesink));
-          gst_message_set_seqnum (message, seqnum);
-          gst_element_post_message (GST_ELEMENT_CAST (basesink), message);
-          break;
-        }
-        case GST_EVENT_SEGMENT:
-          /* configure the segment */
-          /* The segment is protected with both the STREAM_LOCK and the OBJECT_LOCK.
-           * We protect with the OBJECT_LOCK so that we can use the values to
-           * safely answer a POSITION query. */
-          GST_OBJECT_LOCK (basesink);
-          /* the newsegment event is needed to bring the buffer timestamps to the
-           * stream time and to drop samples outside of the playback segment. */
-          gst_event_copy_segment (event, &basesink->segment);
-          GST_DEBUG_OBJECT (basesink, "configured SEGMENT %" GST_SEGMENT_FORMAT,
-              &basesink->segment);
-          basesink->have_newsegment = TRUE;
-          GST_OBJECT_UNLOCK (basesink);
-          break;
-        case GST_EVENT_TAG:
-        {
-          GstTagList *taglist;
-
-          gst_event_parse_tag (event, &taglist);
-
-          gst_element_post_message (GST_ELEMENT_CAST (basesink),
-              gst_message_new_tag (GST_OBJECT_CAST (basesink),
-                  gst_tag_list_copy (taglist)));
-          break;
-        }
-        case GST_EVENT_SINK_MESSAGE:
-        {
-          GstMessage *msg = NULL;
-
-          gst_event_parse_sink_message (event, &msg);
-
-          if (msg)
-            gst_element_post_message (GST_ELEMENT_CAST (basesink), msg);
-        }
-        default:
-          break;
-      }
-    }
   } else {
     g_return_val_if_reached (GST_FLOW_ERROR);
   }
@@ -3070,8 +2984,10 @@ stopping:
 }
 
 static void
-gst_base_sink_flush_start (GstBaseSink * basesink, GstPad * pad)
+gst_base_sink_flush_start (GstBaseSink * basesink)
 {
+  GstPad *pad = basesink->sinkpad;
+
   /* make sure we are not blocked on the clock also clear any pending
    * eos state. */
   gst_base_sink_set_flushing (basesink, pad, TRUE);
@@ -3097,9 +3013,10 @@ gst_base_sink_flush_start (GstBaseSink * basesink, GstPad * pad)
 }
 
 static void
-gst_base_sink_flush_stop (GstBaseSink * basesink, GstPad * pad,
-    gboolean reset_time)
+gst_base_sink_flush_stop (GstBaseSink * basesink, gboolean reset_time)
 {
+  GstPad *pad = basesink->sinkpad;
+
   /* unset flushing so we can accept new data, this also flushes out any EOS
    * event. */
   gst_base_sink_set_flushing (basesink, pad, FALSE);
@@ -3124,42 +3041,57 @@ gst_base_sink_flush_stop (GstBaseSink * basesink, GstPad * pad,
 }
 
 static gboolean
-gst_base_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
+handle_serialized_events (GstBaseSink * basesink, GstEvent * event)
 {
-  GstBaseSink *basesink;
+  GstFlowReturn ret;
   gboolean result = TRUE;
   GstBaseSinkClass *bclass;
 
-  basesink = GST_BASE_SINK (parent);
   bclass = GST_BASE_SINK_GET_CLASS (basesink);
 
-  GST_DEBUG_OBJECT (basesink, "received event %p %" GST_PTR_FORMAT, event,
-      event);
-
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_EOS:
     {
-      GstFlowReturn ret;
-
-      GST_BASE_SINK_PREROLL_LOCK (basesink);
-      if (G_UNLIKELY (basesink->flushing))
-        goto flushing;
-
-      if (G_UNLIKELY (basesink->priv->received_eos))
-        goto after_eos;
+      GstMessage *message;
+      gboolean late, step_end;
+      guint32 seqnum;
 
       /* we set the received EOS flag here so that we can use it when testing if
        * we are prerolled and to refuse more buffers. */
       basesink->priv->received_eos = TRUE;
 
-      /* EOS is a prerollable object, we call the unlocked version because it
-       * does not check the received_eos flag. */
-      ret = gst_base_sink_render_object (basesink, pad,
-          _PR_IS_EVENT, GST_MINI_OBJECT_CAST (event));
-      if (G_UNLIKELY (ret != GST_FLOW_OK))
+      ret = gst_base_sink_do_sync (basesink, GST_MINI_OBJECT_CAST (event),
+          &late, &step_end, _PR_IS_EVENT);
+      if (G_UNLIKELY (ret != GST_FLOW_OK)) {
+        result = FALSE;
+        goto done;
+      }
+
+      /* let the subclass wait for EOS */
+      if (bclass->wait_eos)
+        ret = bclass->wait_eos (basesink);
+      if (G_UNLIKELY (ret != GST_FLOW_OK)) {
         result = FALSE;
+        goto done;
+      }
 
-      GST_BASE_SINK_PREROLL_UNLOCK (basesink);
+      /* the EOS event is completely handled so we mark
+       * ourselves as being in the EOS state. eos is also
+       * protected by the object lock so we can read it when
+       * answering the POSITION query. */
+      GST_OBJECT_LOCK (basesink);
+      basesink->eos = TRUE;
+      GST_OBJECT_UNLOCK (basesink);
+
+      /* ok, now we can post the message */
+      GST_DEBUG_OBJECT (basesink, "Now posting EOS");
+
+      seqnum = basesink->priv->seqnum = gst_event_get_seqnum (event);
+      GST_DEBUG_OBJECT (basesink, "Got seqnum #%" G_GUINT32_FORMAT, seqnum);
+
+      message = gst_message_new_eos (GST_OBJECT_CAST (basesink));
+      gst_message_set_seqnum (message, seqnum);
+      gst_element_post_message (GST_ELEMENT_CAST (basesink), message);
       break;
     }
     case GST_EVENT_CAPS:
@@ -3177,63 +3109,81 @@ gst_base_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
         gst_caps_replace (&basesink->priv->caps, caps);
         GST_OBJECT_UNLOCK (basesink);
       }
-      gst_event_unref (event);
       break;
     }
     case GST_EVENT_SEGMENT:
+      /* configure the segment */
+      /* The segment is protected with both the STREAM_LOCK and the OBJECT_LOCK.
+       * We protect with the OBJECT_LOCK so that we can use the values to
+       * safely answer a POSITION query. */
+      GST_OBJECT_LOCK (basesink);
+      /* the newsegment event is needed to bring the buffer timestamps to the
+       * stream time and to drop samples outside of the playback segment. */
+      gst_event_copy_segment (event, &basesink->segment);
+      GST_DEBUG_OBJECT (basesink, "configured SEGMENT %" GST_SEGMENT_FORMAT,
+          &basesink->segment);
+      basesink->have_newsegment = TRUE;
+      GST_OBJECT_UNLOCK (basesink);
+      GST_BASE_SINK_PREROLL_UNLOCK (basesink);
+      break;
+    case GST_EVENT_TAG:
     {
-      GstFlowReturn ret;
+      GstTagList *taglist;
 
-      GST_DEBUG_OBJECT (basesink, "segment %p", event);
+      gst_event_parse_tag (event, &taglist);
 
-      GST_BASE_SINK_PREROLL_LOCK (basesink);
-      if (G_UNLIKELY (basesink->flushing))
-        goto flushing;
-
-      if (G_UNLIKELY (basesink->priv->received_eos))
-        goto after_eos;
-
-      ret =
-          gst_base_sink_render_object (basesink, pad,
-          _PR_IS_EVENT, GST_MINI_OBJECT_CAST (event));
-      if (G_UNLIKELY (ret != GST_FLOW_OK))
-        result = FALSE;
+      gst_element_post_message (GST_ELEMENT_CAST (basesink),
+          gst_message_new_tag (GST_OBJECT_CAST (basesink),
+              gst_tag_list_copy (taglist)));
+      break;
+    }
+    case GST_EVENT_SINK_MESSAGE:
+    {
+      GstMessage *msg = NULL;
 
-      GST_BASE_SINK_PREROLL_UNLOCK (basesink);
+      gst_event_parse_sink_message (event, &msg);
+      if (msg)
+        gst_element_post_message (GST_ELEMENT_CAST (basesink), msg);
       break;
     }
-    case GST_EVENT_FLUSH_START:
-      if (bclass->event)
-        bclass->event (basesink, event);
+    default:
+      break;
+  }
+done:
+  gst_event_unref (event);
 
-      GST_DEBUG_OBJECT (basesink, "flush-start %p", event);
+  return result;
+}
 
-      gst_base_sink_flush_start (basesink, pad);
+static gboolean
+gst_base_sink_default_event (GstBaseSink * basesink, GstEvent * event)
+{
+  gboolean result = TRUE;
 
+  GST_DEBUG_OBJECT (basesink, "received event %p %" GST_PTR_FORMAT, event,
+      event);
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_FLUSH_START:
+    {
+      GST_DEBUG_OBJECT (basesink, "flush-start %p", event);
+      gst_base_sink_flush_start (basesink);
       gst_event_unref (event);
       break;
+    }
     case GST_EVENT_FLUSH_STOP:
     {
       gboolean reset_time;
 
-      if (bclass->event)
-        bclass->event (basesink, event);
-
       gst_event_parse_flush_stop (event, &reset_time);
       GST_DEBUG_OBJECT (basesink, "flush-stop %p, reset_time: %d", event,
           reset_time);
-
-      gst_base_sink_flush_stop (basesink, pad, reset_time);
-
+      gst_base_sink_flush_stop (basesink, reset_time);
       gst_event_unref (event);
       break;
     }
     default:
-      /* other events are sent to queue or subclass depending on if they
-       * are serialized. */
       if (GST_EVENT_IS_SERIALIZED (event)) {
-        GstFlowReturn ret;
-
         GST_BASE_SINK_PREROLL_LOCK (basesink);
         if (G_UNLIKELY (basesink->flushing))
           goto flushing;
@@ -3241,16 +3191,9 @@ gst_base_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
         if (G_UNLIKELY (basesink->priv->received_eos))
           goto after_eos;
 
-        ret = gst_base_sink_render_object (basesink, pad, _PR_IS_EVENT,
-            GST_MINI_OBJECT_CAST (event));
-        if (G_UNLIKELY (ret != GST_FLOW_OK))
-          result = FALSE;
+        result = handle_serialized_events (basesink, event);
 
         GST_BASE_SINK_PREROLL_UNLOCK (basesink);
-      } else {
-        if (bclass->event)
-          bclass->event (basesink, event);
-        gst_event_unref (event);
       }
       break;
   }
@@ -3262,8 +3205,8 @@ flushing:
   {
     GST_DEBUG_OBJECT (basesink, "we are flushing");
     GST_BASE_SINK_PREROLL_UNLOCK (basesink);
-    result = FALSE;
     gst_event_unref (event);
+    result = FALSE;
     goto done;
   }
 
@@ -3271,12 +3214,30 @@ after_eos:
   {
     GST_DEBUG_OBJECT (basesink, "Event received after EOS, dropping");
     GST_BASE_SINK_PREROLL_UNLOCK (basesink);
-    result = FALSE;
     gst_event_unref (event);
+    result = FALSE;
     goto done;
   }
 }
 
+static gboolean
+gst_base_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
+{
+  GstBaseSink *basesink;
+  GstBaseSinkClass *bclass;
+  gboolean result = FALSE;
+
+  basesink = GST_BASE_SINK (parent);
+  bclass = GST_BASE_SINK_GET_CLASS (basesink);
+
+  if (bclass->event)
+    result = bclass->event (basesink, event);
+  else
+    gst_event_unref (event);
+
+  return result;
+}
+
 /* default implementation to calculate the start and end
  * timestamps on a buffer, subclasses can override
  */
@@ -3396,7 +3357,7 @@ gst_base_sink_chain_unlocked (GstBaseSink * basesink, GstPad * pad,
 
   /* now we can process the buffer in the queue, this function takes ownership
    * of the buffer */
-  result = gst_base_sink_render_object (basesink, pad, obj_type, obj);
+  result = gst_base_sink_render_object (basesink, obj_type, obj);
   return result;
 
   /* ERRORS */
@@ -3610,7 +3571,7 @@ gst_base_sink_perform_seek (GstBaseSink * sink, GstPad * pad, GstEvent * event)
   if (flush) {
     GST_DEBUG_OBJECT (sink, "flushing upstream");
     gst_pad_push_event (pad, gst_event_new_flush_start ());
-    gst_base_sink_flush_start (sink, pad);
+    gst_base_sink_flush_start (sink);
   } else {
     GST_DEBUG_OBJECT (sink, "pausing pulling thread");
   }
@@ -3659,7 +3620,7 @@ gst_base_sink_perform_seek (GstBaseSink * sink, GstPad * pad, GstEvent * event)
   if (flush) {
     GST_DEBUG_OBJECT (sink, "stop flushing upstream");
     gst_pad_push_event (pad, gst_event_new_flush_stop (TRUE));
-    gst_base_sink_flush_stop (sink, pad, TRUE);
+    gst_base_sink_flush_stop (sink, TRUE);
   } else if (res && sink->running) {
     /* we are running the current segment and doing a non-flushing seek,
      * close the segment first based on the position. */
diff --git a/libs/gst/base/gstbasesink.h b/libs/gst/base/gstbasesink.h
index 5237b4526..d14908303 100644
--- a/libs/gst/base/gstbasesink.h
+++ b/libs/gst/base/gstbasesink.h
@@ -175,6 +175,8 @@ struct _GstBaseSinkClass {
 
   /* notify subclass of event */
   gboolean      (*event)        (GstBaseSink *sink, GstEvent *event);
+  GstFlowReturn (*wait_eos)     (GstBaseSink *sink);
+
   /* notify subclass of preroll buffer or real buffer */
   GstFlowReturn (*preroll)      (GstBaseSink *sink, GstBuffer *buffer);
   GstFlowReturn (*render)       (GstBaseSink *sink, GstBuffer *buffer);