aggregator: Hide GstAggregatorPad buffer and EOS fileds

And add a getter for the EOS.

The user should always use the various getters to access
those fields

https://bugzilla.gnome.org/show_bug.cgi?id=742684

gst-libs/gst/base/gstaggregator.c

@@ -166,6 +166,10 @@ struct _GstAggregatorPadPrivate
   gboolean pending_eos;
   gboolean flushing;
 
+  /* Protected by the pad lock */
+  GstBuffer *buffer;
+  gboolean eos;
+
   GCond event_cond;
 
   GMutex stream_lock;
@@ -177,7 +181,7 @@ gst_aggregator_pad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 
   PAD_LOCK (aggpad);
-  aggpad->eos = FALSE;
+  aggpad->priv->eos = FALSE;
   aggpad->priv->flushing = FALSE;
   PAD_UNLOCK (aggpad);
 
@@ -345,8 +349,8 @@ gst_aggregator_check_pads_ready (GstAggregator * self)
     pad = l->data;
 
     PAD_LOCK (pad);
-    if (pad->buffer == NULL && !pad->eos) {
+    if (pad->priv->buffer == NULL && !pad->priv->eos) {
-      GST_OBJECT_UNLOCK (pad);
+      PAD_UNLOCK (pad);
       goto pad_not_ready;
     }
     PAD_UNLOCK (pad);
@@ -862,8 +866,8 @@ gst_aggregator_default_sink_event (GstAggregator * self,
        */
       SRC_STREAM_LOCK (self);
       PAD_LOCK (aggpad);
-      if (!aggpad->buffer) {
+      if (!aggpad->priv->buffer) {
-        aggpad->eos = TRUE;
+        aggpad->priv->eos = TRUE;
       } else {
         aggpad->priv->pending_eos = TRUE;
       }
@@ -1770,8 +1774,8 @@ gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
     goto eos;
 
   PAD_LOCK (aggpad);
-
+  while (aggpad->priv->buffer
-  while (aggpad->buffer && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
+      && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
     PAD_WAIT_EVENT (aggpad);
   }
@@ -1786,9 +1790,9 @@ gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
 
   SRC_STREAM_LOCK (self);
   PAD_LOCK (aggpad);
-  if (aggpad->buffer)
+  if (aggpad->priv->buffer)
-    gst_buffer_unref (aggpad->buffer);
+    gst_buffer_unref (aggpad->priv->buffer);
-  aggpad->buffer = actual_buf;
+  aggpad->priv->buffer = actual_buf;
   PAD_UNLOCK (aggpad);
   PAD_STREAM_UNLOCK (aggpad);
 
@@ -1836,7 +1840,7 @@ gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
       goto flushing;
     }
 
-    while (aggpad->buffer
+    while (aggpad->priv->buffer
         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
       PAD_WAIT_EVENT (aggpad);
@@ -1872,7 +1876,7 @@ gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
       goto flushing;
     }
 
-    while (aggpad->buffer
+    while (aggpad->priv->buffer
         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
       PAD_WAIT_EVENT (aggpad);
@@ -1904,7 +1908,7 @@ gst_aggregator_pad_activate_mode_func (GstPad * pad,
   if (active == FALSE) {
     PAD_LOCK (aggpad);
     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
-    gst_buffer_replace (&aggpad->buffer, NULL);
+    gst_buffer_replace (&aggpad->priv->buffer, NULL);
     PAD_BROADCAST_EVENT (aggpad);
     PAD_UNLOCK (aggpad);
   } else {
@@ -1980,7 +1984,7 @@ gst_aggregator_pad_init (GstAggregatorPad * pad)
       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
       GstAggregatorPadPrivate);
 
-  pad->buffer = NULL;
+  pad->priv->buffer = NULL;
   g_cond_init (&pad->priv->event_cond);
 
   g_mutex_init (&pad->priv->stream_lock);
@@ -2002,13 +2006,13 @@ gst_aggregator_pad_steal_buffer_unlocked (GstAggregatorPad * pad)
 {
   GstBuffer *buffer = NULL;
 
-  if (pad->buffer) {
+  if (pad->priv->buffer) {
     GST_TRACE_OBJECT (pad, "Consuming buffer");
-    buffer = pad->buffer;
+    buffer = pad->priv->buffer;
-    pad->buffer = NULL;
+    pad->priv->buffer = NULL;
     if (pad->priv->pending_eos) {
       pad->priv->pending_eos = FALSE;
-      pad->eos = TRUE;
+      pad->priv->eos = TRUE;
     }
     PAD_BROADCAST_EVENT (pad);
     GST_DEBUG_OBJECT (pad, "Consumed: %" GST_PTR_FORMAT, buffer);
@@ -2052,13 +2056,25 @@ gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
   GstBuffer *buffer = NULL;
 
   PAD_LOCK (pad);
-  if (pad->buffer)
+  if (pad->priv->buffer)
-    buffer = gst_buffer_ref (pad->buffer);
+    buffer = gst_buffer_ref (pad->priv->buffer);
   PAD_UNLOCK (pad);
 
   return buffer;
 }
 
+gboolean
+gst_aggregator_pad_is_eos (GstAggregatorPad * pad)
+{
+  gboolean is_eos;
+
+  PAD_LOCK (pad);
+  is_eos = pad->priv->eos;
+  PAD_UNLOCK (pad);
+
+  return is_eos;
+}
+
 /**
  * gst_aggregator_merge_tags:
  * @self: a #GstAggregator

gst-libs/gst/base/gstaggregator.h

@@ -70,9 +70,7 @@ struct _GstAggregatorPad
   GstPad                       parent;
 
   /* Protected by the pad's object lock */
-  GstBuffer                 *  buffer;
   GstSegment                   segment;
-  gboolean                     eos;
 
   /* < Private > */
   GstAggregatorPadPrivate   *  priv;
@@ -107,6 +105,7 @@ GType gst_aggregator_pad_get_type           (void);
 GstBuffer * gst_aggregator_pad_steal_buffer (GstAggregatorPad *  pad);
 GstBuffer * gst_aggregator_pad_steal_buffer_unlocked (GstAggregatorPad *  pad);
 GstBuffer * gst_aggregator_pad_get_buffer   (GstAggregatorPad *  pad);
+gboolean    gst_aggregator_pad_is_eos       (GstAggregatorPad *  pad);
 
 /*********************
  * GstAggregator API *

gst-libs/gst/video/gstvideoaggregator.c

@@ -966,8 +966,9 @@ gst_videoaggregator_fill_queues (GstVideoAggregator * vagg,
     bpad = GST_AGGREGATOR_PAD (pad);
     GST_OBJECT_LOCK (bpad);
     segment = bpad->segment;
-    is_eos = bpad->eos;
     GST_OBJECT_UNLOCK (bpad);
+    is_eos = gst_aggregator_pad_is_eos (bpad);
+
     if (!is_eos)
       eos = FALSE;
     buf = gst_aggregator_pad_get_buffer (bpad);
@@ -1550,6 +1551,7 @@ gst_videoaggregator_sink_clip (GstAggregator * agg,
 {
   GstVideoAggregatorPad *pad = GST_VIDEO_AGGREGATOR_PAD (bpad);
   GstClockTime start_time, end_time;
+  GstBuffer *pbuf;
 
   start_time = GST_BUFFER_TIMESTAMP (buf);
   if (start_time == -1) {
@@ -1586,10 +1588,15 @@ gst_videoaggregator_sink_clip (GstAggregator * agg,
     end_time *= ABS (agg->segment.rate);
   }
 
-  if (bpad->buffer != NULL && end_time < pad->priv->end_time) {
+  pbuf = gst_aggregator_pad_get_buffer (bpad);
-    gst_buffer_unref (buf);
+  if (pbuf != NULL) {
-    *outbuf = NULL;
+    gst_buffer_unref (pbuf);
-    goto done;
+
+    if (end_time < pad->priv->end_time) {
+      gst_buffer_unref (buf);
+      *outbuf = NULL;
+      goto done;
+    }
   }
 
   *outbuf = buf;

gst/audiomixer/gstaudiomixer.c

@@ -1469,7 +1469,7 @@ gst_audiomixer_aggregate (GstAggregator * agg, gboolean timeout)
     GstAudioMixerPad *pad = GST_AUDIO_MIXER_PAD (iter->data);
     GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (iter->data);
 
-    if (!GST_AGGREGATOR_PAD (pad)->eos)
+    if (!gst_aggregator_pad_is_eos (aggpad))
       is_eos = FALSE;
 
     inbuf = gst_aggregator_pad_get_buffer (aggpad);
@@ -1489,7 +1489,7 @@ gst_audiomixer_aggregate (GstAggregator * agg, gboolean timeout)
       gst_buffer_unref (inbuf);
     }
 
-    if (!pad->buffer && !dropped && GST_AGGREGATOR_PAD (pad)->eos) {
+    if (!pad->buffer && !dropped && gst_aggregator_pad_is_eos (aggpad)) {
       GST_DEBUG_OBJECT (aggpad, "Pad is in EOS state");
       continue;
     }

tests/check/libs/aggregator.c

@@ -84,7 +84,7 @@ gst_test_aggregator_aggregate (GstAggregator * aggregator, gboolean timeout)
       case GST_ITERATOR_OK:
         pad = g_value_get_object (&value);
 
-        if (pad->eos == FALSE)
+        if (gst_aggregator_pad_is_eos (pad) == FALSE)
           all_eos = FALSE;
         buffer = gst_aggregator_pad_steal_buffer (pad);
         gst_buffer_replace (&buffer, NULL);