aggregator: add simple support for caps handling

Modelled off the videoaggregator caps handling as that seems the most
mature aggregtor-using implementation that has caps handling there is.

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

gst-libs/gst/audio/gstaudioaggregator.c

@@ -144,8 +144,6 @@ struct _GstAudioAggregatorPrivate
 {
   GMutex mutex;
 
-  gboolean send_caps;           /* aagg lock */
-
   /* All three properties are unprotected, can't be modified while streaming */
   /* Size in frames that is output per buffer */
   GstClockTime output_buffer_duration;
@@ -189,6 +187,8 @@ static GstFlowReturn gst_audio_aggregator_aggregate (GstAggregator * agg,
     gboolean timeout);
 static gboolean sync_pad_values (GstAudioAggregator * aagg,
     GstAudioAggregatorPad * pad);
+static gboolean gst_audio_aggregator_negotiated_src_caps (GstAggregator * agg,
+    GstCaps * caps);
 
 #define DEFAULT_OUTPUT_BUFFER_DURATION (10 * GST_MSECOND)
 #define DEFAULT_ALIGNMENT_THRESHOLD   (40 * GST_MSECOND)
@@ -251,6 +251,8 @@ gst_audio_aggregator_class_init (GstAudioAggregatorClass * klass)
       GST_DEBUG_FUNCPTR (gst_audio_aggregator_aggregate);
   gstaggregator_class->clip = GST_DEBUG_FUNCPTR (gst_audio_aggregator_do_clip);
   gstaggregator_class->get_next_time = gst_audio_aggregator_get_next_time;
+  gstaggregator_class->negotiated_src_caps =
+      gst_audio_aggregator_negotiated_src_caps;
 
   klass->create_output_buffer = gst_audio_aggregator_create_output_buffer;
 
@@ -656,9 +658,10 @@ gst_audio_aggregator_set_sink_caps (GstAudioAggregator * aagg,
 }
 
 
-gboolean
-gst_audio_aggregator_set_src_caps (GstAudioAggregator * aagg, GstCaps * caps)
+static gboolean
+gst_audio_aggregator_negotiated_src_caps (GstAggregator * agg, GstCaps * caps)
 {
+  GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
   GstAudioInfo info;
 
   if (!gst_audio_info_from_caps (&info, caps)) {
@@ -674,8 +677,6 @@ gst_audio_aggregator_set_src_caps (GstAudioAggregator * aagg, GstCaps * caps)
     gst_caps_replace (&aagg->current_caps, caps);
 
     memcpy (&aagg->info, &info, sizeof (info));
-    aagg->priv->send_caps = TRUE;
-
   }
 
   GST_OBJECT_UNLOCK (aagg);
@@ -683,7 +684,9 @@ gst_audio_aggregator_set_src_caps (GstAudioAggregator * aagg, GstCaps * caps)
 
   /* send caps event later, after stream-start event */
 
-  return TRUE;
+  return
+      GST_AGGREGATOR_CLASS
+      (gst_audio_aggregator_parent_class)->negotiated_src_caps (agg, caps);
 }
 
 
@@ -1132,21 +1135,13 @@ gst_audio_aggregator_aggregate (GstAggregator * agg, gboolean timeout)
 
       GST_OBJECT_UNLOCK (agg);
       GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
-      return GST_FLOW_OK;
+      return GST_AGGREGATOR_FLOW_NEED_DATA;
     } else {
       GST_OBJECT_UNLOCK (agg);
       goto not_negotiated;
     }
   }
 
-  if (aagg->priv->send_caps) {
-    GST_OBJECT_UNLOCK (agg);
-    gst_aggregator_set_src_caps (agg, aagg->current_caps);
-    GST_OBJECT_LOCK (agg);
-
-    aagg->priv->send_caps = FALSE;
-  }
-
   rate = GST_AUDIO_INFO_RATE (&aagg->info);
   bpf = GST_AUDIO_INFO_BPF (&aagg->info);
 
@@ -1296,7 +1291,7 @@ gst_audio_aggregator_aggregate (GstAggregator * agg, gboolean timeout)
     /* We dropped a buffer, retry */
     GST_LOG_OBJECT (aagg, "A pad dropped a buffer, wait for the next one");
     GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
-    return GST_FLOW_OK;
+    return GST_AGGREGATOR_FLOW_NEED_DATA;
   }
 
   if (!is_done && !is_eos) {
@@ -1304,7 +1299,7 @@ gst_audio_aggregator_aggregate (GstAggregator * agg, gboolean timeout)
     GST_LOG_OBJECT (aagg,
         "We're not done yet for the current offset, waiting for more data");
     GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
-    return GST_FLOW_OK;
+    return GST_AGGREGATOR_FLOW_NEED_DATA;
   }
 
   if (is_eos) {

gst-libs/gst/audio/gstaudioaggregator.h

@@ -162,9 +162,6 @@ void
 gst_audio_aggregator_set_sink_caps (GstAudioAggregator * aagg,
     GstAudioAggregatorPad * pad, GstCaps * caps);
 
-gboolean
-gst_audio_aggregator_set_src_caps (GstAudioAggregator * aagg, GstCaps * caps);
-
 
 G_END_DECLS
 

gst/audiomixer/gstaudiointerleave.c

@@ -432,10 +432,10 @@ gst_audio_interleave_setcaps (GstAudioInterleave * self, GstPad * pad,
     GST_DEBUG_OBJECT (self, "setting sinkcaps %" GST_PTR_FORMAT, sinkcaps);
 
     gst_caps_replace (&self->sinkcaps, sinkcaps);
+    gst_pad_mark_reconfigure (GST_AGGREGATOR_SRC_PAD (aagg));
 
     gst_caps_unref (sinkcaps);
     new = TRUE;
-    self->new_caps = TRUE;
   }
 
   if (self->channel_positions_from_input
@@ -504,52 +504,40 @@ gst_audio_interleave_sink_event (GstAggregator * agg, GstAggregatorPad * aggpad,
 }
 
 static GstFlowReturn
-gst_audio_interleave_aggregate (GstAggregator * aggregator, gboolean timeout)
+gst_audio_interleave_update_src_caps (GstAggregator * agg, GstCaps * caps,
+    GstCaps ** ret)
 {
-  GstAudioInterleave *self = GST_AUDIO_INTERLEAVE (aggregator);
-  GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (aggregator);
+  GstAudioInterleave *self = GST_AUDIO_INTERLEAVE (agg);
+  GstStructure *s;
 
-  GST_OBJECT_LOCK (aggregator);
-  if (self->new_caps) {
-    GstCaps *srccaps;
-    GstStructure *s;
-    gboolean ret;
+  /* This means that either no caps have been set on the sink pad (if
+   * sinkcaps is NULL) or that there is no sink pad (if channels == 0).
+   */
+  if (self->sinkcaps == NULL || self->channels == 0)
+    return GST_FLOW_NOT_NEGOTIATED;
 
-    if (self->sinkcaps == NULL || self->channels == 0) {
-      /* In this case, let the base class handle it */
-      goto not_negotiated;
-    }
+  *ret = gst_caps_copy (self->sinkcaps);
+  s = gst_caps_get_structure (*ret, 0);
 
-    srccaps = gst_caps_copy (self->sinkcaps);
-    s = gst_caps_get_structure (srccaps, 0);
+  gst_structure_set (s, "channels", G_TYPE_INT, self->channels, "layout",
+      G_TYPE_STRING, "interleaved", "channel-mask", GST_TYPE_BITMASK,
+      gst_audio_interleave_get_channel_mask (self), NULL);
 
-    gst_structure_set (s, "channels", G_TYPE_INT, self->channels, "layout",
-        G_TYPE_STRING, "interleaved", "channel-mask", GST_TYPE_BITMASK,
-        gst_audio_interleave_get_channel_mask (self), NULL);
+  return GST_FLOW_OK;
+}
 
+static gboolean
+gst_audio_interleave_negotiated_src_caps (GstAggregator * agg, GstCaps * caps)
+{
+  GstAudioInterleave *self = GST_AUDIO_INTERLEAVE (agg);
+  GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (self);
 
-    GST_OBJECT_UNLOCK (aggregator);
-    ret = gst_audio_aggregator_set_src_caps (aagg, srccaps);
-    gst_caps_unref (srccaps);
+  if (!GST_AGGREGATOR_CLASS (parent_class)->negotiated_src_caps (agg, caps))
+    return FALSE;
 
-    if (!ret)
-      goto src_did_not_accept;
+  gst_audio_interleave_set_process_function (self, &aagg->info);
 
-    GST_OBJECT_LOCK (aggregator);
-
-    gst_audio_interleave_set_process_function (self, &aagg->info);
-
-    self->new_caps = FALSE;
-  }
-
-not_negotiated:
-  GST_OBJECT_UNLOCK (aggregator);
-
-  return GST_AGGREGATOR_CLASS (parent_class)->aggregate (aggregator, timeout);
-
-src_did_not_accept:
-  GST_WARNING_OBJECT (self, "src did not accept setcaps()");
-  return GST_FLOW_NOT_NEGOTIATED;;
+  return TRUE;
 }
 
 static void
@@ -586,7 +574,8 @@ gst_audio_interleave_class_init (GstAudioInterleaveClass * klass)
   agg_class->sink_query = GST_DEBUG_FUNCPTR (gst_audio_interleave_sink_query);
   agg_class->sink_event = GST_DEBUG_FUNCPTR (gst_audio_interleave_sink_event);
   agg_class->stop = gst_audio_interleave_stop;
-  agg_class->aggregate = gst_audio_interleave_aggregate;
+  agg_class->update_src_caps = gst_audio_interleave_update_src_caps;
+  agg_class->negotiated_src_caps = gst_audio_interleave_negotiated_src_caps;
 
   aagg_class->aggregate_one_buffer = gst_audio_interleave_aggregate_one_buffer;
 
@@ -720,7 +709,6 @@ gst_audio_interleave_stop (GstAggregator * agg)
   if (!GST_AGGREGATOR_CLASS (parent_class)->stop (agg))
     return FALSE;
 
-  self->new_caps = FALSE;
   gst_caps_replace (&self->sinkcaps, NULL);
 
   return TRUE;
@@ -765,9 +753,7 @@ gst_audio_interleave_request_new_pad (GstElement * element,
   g_value_unset (&val);
 
   /* Update the src caps if we already have them */
-  GST_OBJECT_LOCK (self);
-  self->new_caps = TRUE;
-  GST_OBJECT_UNLOCK (self);
+  gst_pad_mark_reconfigure (GST_AGGREGATOR_SRC_PAD (self));
 
   return GST_PAD_CAST (newpad);
 
@@ -804,7 +790,7 @@ gst_audio_interleave_release_pad (GstElement * element, GstPad * pad)
       ipad->channel--;
   }
 
-  self->new_caps = TRUE;
+  gst_pad_mark_reconfigure (GST_AGGREGATOR_SRC_PAD (self));
   GST_OBJECT_UNLOCK (self);
 
 

gst/audiomixer/gstaudiointerleave.h

@@ -58,7 +58,6 @@ struct _GstAudioInterleave {
   gint padcounter;
   guint channels;
 
-  gboolean new_caps;
   GstCaps *sinkcaps;
 
   GValueArray *channel_positions;

gst/audiomixer/gstaudiomixer.c

@@ -348,7 +348,6 @@ gst_audiomixer_setcaps (GstAudioMixer * audiomixer, GstPad * pad,
   GstAudioInfo info;
   GstStructure *s;
   gint channels = 0;
-  gboolean ret;
 
   caps = gst_caps_copy (orig_caps);
 
@@ -405,20 +404,21 @@ gst_audiomixer_setcaps (GstAudioMixer * audiomixer, GstPad * pad,
       gst_caps_unref (caps);
       return FALSE;
     }
+  } else {
+    gst_caps_replace (&aagg->current_caps, caps);
+    aagg->info = info;
+    gst_pad_mark_reconfigure (GST_AGGREGATOR_SRC_PAD (agg));
   }
   GST_OBJECT_UNLOCK (audiomixer);
 
-  ret = gst_audio_aggregator_set_src_caps (aagg, caps);
-
-  if (ret)
-    gst_audio_aggregator_set_sink_caps (aagg, GST_AUDIO_AGGREGATOR_PAD (pad),
-        orig_caps);
+  gst_audio_aggregator_set_sink_caps (aagg, GST_AUDIO_AGGREGATOR_PAD (pad),
+      orig_caps);
 
   GST_INFO_OBJECT (pad, "handle caps change to %" GST_PTR_FORMAT, caps);
 
   gst_caps_unref (caps);
 
-  return ret;
+  return TRUE;
 
   /* ERRORS */
 invalid_format:
@@ -429,6 +429,20 @@ invalid_format:
   }
 }
 
+static GstFlowReturn
+gst_audiomixer_update_src_caps (GstAggregator * agg, GstCaps * caps,
+    GstCaps ** ret)
+{
+  GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
+
+  if (aagg->current_caps == NULL)
+    return GST_AGGREGATOR_FLOW_NEED_DATA;
+
+  *ret = gst_caps_ref (aagg->current_caps);
+
+  return GST_FLOW_OK;
+}
+
 static gboolean
 gst_audiomixer_sink_event (GstAggregator * agg, GstAggregatorPad * aggpad,
     GstEvent * event)
@@ -495,6 +509,8 @@ gst_audiomixer_class_init (GstAudioMixerClass * klass)
 
   agg_class->sink_query = GST_DEBUG_FUNCPTR (gst_audiomixer_sink_query);
   agg_class->sink_event = GST_DEBUG_FUNCPTR (gst_audiomixer_sink_event);
+  agg_class->update_src_caps =
+      GST_DEBUG_FUNCPTR (gst_audiomixer_update_src_caps);
 
   aagg_class->aggregate_one_buffer = gst_audiomixer_aggregate_one_buffer;
 }

tests/check/elements/audiointerleave.c

@@ -1043,7 +1043,9 @@ GST_START_TEST (test_audiointerleave_2ch_smallbuf)
   gst_caps_unref (caps);
   gst_event_unref (ev);
 
-  for (i = 0; i < 24; i++)
+  /* eat the caps processing */
+  gst_harness_crank_single_clock_wait (h);
+  for (i = 0; i < 23; i++)
     gst_harness_crank_single_clock_wait (h);
   fail_unless_equals_uint64 (gst_clock_get_time (GST_ELEMENT_CLOCK
           (h->element)), 750 * GST_MSECOND);