+static gboolean
+check_events (GstAggregator * self, GstAggregatorPad * pad, gpointer user_data)
+{
+ GstEvent *event = NULL;
+ GstQuery *query = NULL;
+ GstAggregatorClass *klass = NULL;
+ gboolean *processed_event = user_data;
+
+ do {
+ event = NULL;
+
+ PAD_LOCK (pad);
+ if (pad->priv->num_buffers == 0 && pad->priv->pending_eos) {
+ pad->priv->pending_eos = FALSE;
+ pad->priv->eos = TRUE;
+ }
+ if (pad->priv->clipped_buffer == NULL &&
+ !GST_IS_BUFFER (g_queue_peek_tail (&pad->priv->buffers))) {
+ if (GST_IS_EVENT (g_queue_peek_tail (&pad->priv->buffers)))
+ event = gst_event_ref (g_queue_peek_tail (&pad->priv->buffers));
+ if (GST_IS_QUERY (g_queue_peek_tail (&pad->priv->buffers)))
+ query = g_queue_peek_tail (&pad->priv->buffers);
+ }
+ PAD_UNLOCK (pad);
+ if (event || query) {
+ gboolean ret;
+
+ if (processed_event)
+ *processed_event = TRUE;
+ if (klass == NULL)
+ klass = GST_AGGREGATOR_GET_CLASS (self);
+
+ if (event) {
+ GST_LOG_OBJECT (pad, "Processing %" GST_PTR_FORMAT, event);
+ gst_event_ref (event);
+ ret = klass->sink_event (self, pad, event);
+
+ PAD_LOCK (pad);
+ if (GST_EVENT_TYPE (event) == GST_EVENT_CAPS)
+ pad->priv->negotiated = ret;
+ if (g_queue_peek_tail (&pad->priv->buffers) == event)
+ gst_event_unref (g_queue_pop_tail (&pad->priv->buffers));
+ gst_event_unref (event);
+ }
+
+ if (query) {
+ GST_LOG_OBJECT (pad, "Processing %" GST_PTR_FORMAT, query);
+ ret = klass->sink_query (self, pad, query);
+
+ PAD_LOCK (pad);
+ if (g_queue_peek_tail (&pad->priv->buffers) == query) {
+ GstStructure *s;
+
+ s = gst_query_writable_structure (query);
+ gst_structure_set (s, "gst-aggregator-retval", G_TYPE_BOOLEAN, ret,
+ NULL);
+ g_queue_pop_tail (&pad->priv->buffers);
+ }
+ }
+
+ PAD_BROADCAST_EVENT (pad);
+ PAD_UNLOCK (pad);
+ }
+ } while (event != NULL);
+
+ return TRUE;
+}
+
+static void
+gst_aggregator_pad_set_flushing (GstAggregatorPad * aggpad,
+ GstFlowReturn flow_return, gboolean full)
+{
+ GList *item;
+
+ PAD_LOCK (aggpad);
+ if (flow_return == GST_FLOW_NOT_LINKED)
+ aggpad->priv->flow_return = MIN (flow_return, aggpad->priv->flow_return);
+ else
+ aggpad->priv->flow_return = flow_return;
+
+ item = g_queue_peek_head_link (&aggpad->priv->buffers);
+ while (item) {
+ GList *next = item->next;
+
+ /* In partial flush, we do like the pad, we get rid of non-sticky events
+ * and EOS/SEGMENT.
+ */
+ if (full || GST_IS_BUFFER (item->data) ||
+ GST_EVENT_TYPE (item->data) == GST_EVENT_EOS ||
+ GST_EVENT_TYPE (item->data) == GST_EVENT_SEGMENT ||
+ !GST_EVENT_IS_STICKY (item->data)) {
+ if (!GST_IS_QUERY (item->data))
+ gst_mini_object_unref (item->data);
+ g_queue_delete_link (&aggpad->priv->buffers, item);
+ }
+ item = next;
+ }
+ aggpad->priv->num_buffers = 0;
+ gst_buffer_replace (&aggpad->priv->clipped_buffer, NULL);
+
+ PAD_BROADCAST_EVENT (aggpad);
+ PAD_UNLOCK (aggpad);
+}
+
+static GstFlowReturn
+gst_aggregator_default_update_src_caps (GstAggregator * agg, GstCaps * caps,
+ GstCaps ** ret)
+{
+ *ret = gst_caps_ref (caps);
+
+ return GST_FLOW_OK;
+}
+
+static GstCaps *
+gst_aggregator_default_fixate_src_caps (GstAggregator * agg, GstCaps * caps)
+{
+ caps = gst_caps_fixate (caps);
+
+ return caps;
+}
+
+static gboolean
+gst_aggregator_default_negotiated_src_caps (GstAggregator * agg, GstCaps * caps)
+{
+ return TRUE;
+}
+
+
+/* takes ownership of the pool, allocator and query */
+static gboolean
+gst_aggregator_set_allocation (GstAggregator * self,
+ GstBufferPool * pool, GstAllocator * allocator,
+ GstAllocationParams * params, GstQuery * query)
+{
+ GstAllocator *oldalloc;
+ GstBufferPool *oldpool;
+ GstQuery *oldquery;
+
+ GST_DEBUG ("storing allocation query");
+
+ GST_OBJECT_LOCK (self);
+ oldpool = self->priv->pool;
+ self->priv->pool = pool;
+
+ oldalloc = self->priv->allocator;
+ self->priv->allocator = allocator;
+
+ oldquery = self->priv->allocation_query;
+ self->priv->allocation_query = query;
+
+ if (params)
+ self->priv->allocation_params = *params;
+ else
+ gst_allocation_params_init (&self->priv->allocation_params);
+ GST_OBJECT_UNLOCK (self);
+
+ if (oldpool) {
+ GST_DEBUG_OBJECT (self, "deactivating old pool %p", oldpool);
+ gst_buffer_pool_set_active (oldpool, FALSE);
+ gst_object_unref (oldpool);
+ }
+ if (oldalloc) {
+ gst_object_unref (oldalloc);
+ }
+ if (oldquery) {
+ gst_query_unref (oldquery);
+ }
+ return TRUE;
+}
+
+
+static gboolean
+gst_aggregator_decide_allocation (GstAggregator * self, GstQuery * query)
+{
+ GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (self);
+
+ if (aggclass->decide_allocation)
+ if (!aggclass->decide_allocation (self, query))
+ return FALSE;
+
+ return TRUE;
+}
+
+static gboolean
+gst_aggregator_do_allocation (GstAggregator * self, GstCaps * caps)
+{
+ GstQuery *query;
+ gboolean result = TRUE;
+ GstBufferPool *pool = NULL;
+ GstAllocator *allocator;
+ GstAllocationParams params;
+
+ /* find a pool for the negotiated caps now */
+ GST_DEBUG_OBJECT (self, "doing allocation query");
+ query = gst_query_new_allocation (caps, TRUE);
+ if (!gst_pad_peer_query (self->srcpad, query)) {
+ /* not a problem, just debug a little */
+ GST_DEBUG_OBJECT (self, "peer ALLOCATION query failed");
+ }
+
+ GST_DEBUG_OBJECT (self, "calling decide_allocation");
+ result = gst_aggregator_decide_allocation (self, query);
+
+ GST_DEBUG_OBJECT (self, "ALLOCATION (%d) params: %" GST_PTR_FORMAT, result,
+ query);
+
+ if (!result)
+ goto no_decide_allocation;
+
+ /* we got configuration from our peer or the decide_allocation method,
+ * parse them */
+ if (gst_query_get_n_allocation_params (query) > 0) {
+ gst_query_parse_nth_allocation_param (query, 0, &allocator, ¶ms);
+ } else {
+ allocator = NULL;
+ gst_allocation_params_init (¶ms);
+ }
+
+ if (gst_query_get_n_allocation_pools (query) > 0)
+ gst_query_parse_nth_allocation_pool (query, 0, &pool, NULL, NULL, NULL);
+
+ /* now store */
+ result =
+ gst_aggregator_set_allocation (self, pool, allocator, ¶ms, query);
+
+ return result;
+
+ /* Errors */
+no_decide_allocation:
+ {
+ GST_WARNING_OBJECT (self, "Failed to decide allocation");
+ gst_query_unref (query);
+
+ return result;
+ }
+
+}
+
+/* WITH SRC_LOCK held */
+static GstFlowReturn
+gst_aggregator_update_src_caps (GstAggregator * self)
+{
+ GstAggregatorClass *agg_klass = GST_AGGREGATOR_GET_CLASS (self);
+ GstCaps *downstream_caps, *template_caps, *caps = NULL;
+ GstFlowReturn ret = GST_FLOW_OK;
+
+ template_caps = gst_pad_get_pad_template_caps (self->srcpad);
+ downstream_caps = gst_pad_peer_query_caps (self->srcpad, template_caps);
+
+ if (gst_caps_is_empty (downstream_caps)) {
+ GST_INFO_OBJECT (self, "Downstream caps (%"
+ GST_PTR_FORMAT ") not compatible with pad template caps (%"
+ GST_PTR_FORMAT ")", downstream_caps, template_caps);
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ goto done;
+ }
+
+ g_assert (agg_klass->update_src_caps);
+ GST_DEBUG_OBJECT (self, "updating caps from %" GST_PTR_FORMAT,
+ downstream_caps);
+ ret = agg_klass->update_src_caps (self, downstream_caps, &caps);
+ if (ret < GST_FLOW_OK) {
+ GST_WARNING_OBJECT (self, "Subclass failed to update provided caps");
+ goto done;
+ }
+ if ((caps == NULL || gst_caps_is_empty (caps)) && ret >= GST_FLOW_OK) {
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ goto done;
+ }
+ GST_DEBUG_OBJECT (self, " to %" GST_PTR_FORMAT, caps);
+
+#ifdef GST_ENABLE_EXTRA_CHECKS
+ if (!gst_caps_is_subset (caps, template_caps)) {
+ GstCaps *intersection;
+
+ GST_ERROR_OBJECT (self,
+ "update_src_caps returned caps %" GST_PTR_FORMAT
+ " which are not a real subset of the template caps %"
+ GST_PTR_FORMAT, caps, template_caps);
+ g_warning ("%s: update_src_caps returned caps which are not a real "
+ "subset of the filter caps", GST_ELEMENT_NAME (self));
+
+ intersection =
+ gst_caps_intersect_full (template_caps, caps, GST_CAPS_INTERSECT_FIRST);
+ gst_caps_unref (caps);
+ caps = intersection;
+ }
+#endif
+
+ if (gst_caps_is_any (caps)) {
+ goto done;
+ }
+
+ if (!gst_caps_is_fixed (caps)) {
+ g_assert (agg_klass->fixate_src_caps);
+
+ GST_DEBUG_OBJECT (self, "fixate caps from %" GST_PTR_FORMAT, caps);
+ if (!(caps = agg_klass->fixate_src_caps (self, caps))) {
+ GST_WARNING_OBJECT (self, "Subclass failed to fixate provided caps");
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ goto done;
+ }
+ GST_DEBUG_OBJECT (self, " to %" GST_PTR_FORMAT, caps);
+ }
+
+ if (agg_klass->negotiated_src_caps) {
+ if (!agg_klass->negotiated_src_caps (self, caps)) {
+ GST_WARNING_OBJECT (self, "Subclass failed to accept negotiated caps");
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ goto done;
+ }
+ }
+
+ gst_aggregator_set_src_caps (self, caps);
+
+ if (!gst_aggregator_do_allocation (self, caps)) {
+ GST_WARNING_OBJECT (self, "Allocation negotiation failed");
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ }
+
+done:
+ gst_caps_unref (downstream_caps);
+ gst_caps_unref (template_caps);
+
+ if (caps)
+ gst_caps_unref (caps);
+
+ return ret;
+}
+