GstClockTime time_level;
GstSegment head_segment; /* segment before the queue */
+ gboolean negotiated;
+
gboolean eos;
GMutex lock;
*
* This method guarantees that @func will be called only once for each
* sink pad.
+ *
+ * Returns: %FALSE if there are no sinkpads or if @func returned %FALSE
*/
gboolean
gst_aggregator_iterate_sinkpads (GstAggregator * self,
check_events (GstAggregator * self, GstAggregatorPad * pad, gpointer user_data)
{
GstEvent *event = NULL;
+ GstQuery *query = NULL;
GstAggregatorClass *klass = NULL;
gboolean *processed_event = user_data;
pad->priv->eos = TRUE;
}
if (pad->priv->clipped_buffer == NULL &&
- GST_IS_EVENT (g_queue_peek_tail (&pad->priv->buffers))) {
- event = g_queue_pop_tail (&pad->priv->buffers);
- PAD_BROADCAST_EVENT (pad);
+ !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) {
+ if (event || query) {
+ gboolean ret;
+
if (processed_event)
*processed_event = TRUE;
if (klass == NULL)
klass = GST_AGGREGATOR_GET_CLASS (self);
- GST_LOG_OBJECT (pad, "Processing %" GST_PTR_FORMAT, event);
- klass->sink_event (self, pad, event);
+ 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);
GST_EVENT_TYPE (item->data) == GST_EVENT_EOS ||
GST_EVENT_TYPE (item->data) == GST_EVENT_SEGMENT ||
!GST_EVENT_IS_STICKY (item->data)) {
- gst_mini_object_unref (item->data);
+ if (!GST_IS_QUERY (item->data))
+ gst_mini_object_unref (item->data);
g_queue_delete_link (&aggpad->priv->buffers, item);
}
item = next;
PAD_LOCK (aggpad);
GST_OBJECT_LOCK (aggpad);
gst_event_copy_segment (event, &aggpad->segment);
+ /* We've got a new segment, tail_position is now meaningless
+ * and may interfere with the time_level calculation
+ */
+ aggpad->priv->tail_position = GST_CLOCK_TIME_NONE;
update_time_level (aggpad, FALSE);
GST_OBJECT_UNLOCK (aggpad);
PAD_UNLOCK (aggpad);
{
gst_aggregator_pad_flush (pad, self);
+ PAD_LOCK (pad);
+ pad->priv->flow_return = GST_FLOW_FLUSHING;
+ pad->priv->negotiated = FALSE;
+ PAD_BROADCAST_EVENT (pad);
+ PAD_UNLOCK (pad);
+
return TRUE;
}
gint serial = 0;
gchar *name = NULL;
- if (templ->direction != GST_PAD_SINK ||
- g_strcmp0 (templ->name_template, "sink_%u") != 0)
+ if (templ->direction != GST_PAD_SINK)
goto not_sink;
+ if (templ->presence != GST_PAD_REQUEST)
+ goto not_request;
+
GST_OBJECT_LOCK (self);
if (req_name == NULL || strlen (req_name) < 6
|| !g_str_has_prefix (req_name, "sink_")) {
/* errors */
not_sink:
{
- GST_WARNING_OBJECT (self, "request new pad that is not a SINK pad\n");
+ GST_WARNING_OBJECT (self, "request new pad that is not a SINK pad");
+ return NULL;
+ }
+not_request:
+ {
+ GST_WARNING_OBJECT (self, "request new pad that is not a REQUEST pad");
return NULL;
}
}
{
GstPad *pad = GST_PAD (aggpad);
+ if (GST_QUERY_TYPE (query) == GST_QUERY_ALLOCATION) {
+ GstQuery *decide_query = NULL;
+ GstAggregatorClass *agg_class;
+ gboolean ret;
+
+ GST_OBJECT_LOCK (self);
+ PAD_LOCK (aggpad);
+ if (G_UNLIKELY (!aggpad->priv->negotiated)) {
+ GST_DEBUG_OBJECT (self,
+ "not negotiated yet, can't answer ALLOCATION query");
+ PAD_UNLOCK (aggpad);
+ GST_OBJECT_UNLOCK (self);
+
+ return FALSE;
+ }
+
+ if ((decide_query = self->priv->allocation_query))
+ gst_query_ref (decide_query);
+ PAD_UNLOCK (aggpad);
+ GST_OBJECT_UNLOCK (self);
+
+ GST_DEBUG_OBJECT (self,
+ "calling propose allocation with query %" GST_PTR_FORMAT, decide_query);
+
+ agg_class = GST_AGGREGATOR_GET_CLASS (self);
+
+ /* pass the query to the propose_allocation vmethod if any */
+ if (agg_class->propose_allocation)
+ ret = agg_class->propose_allocation (self, aggpad, decide_query, query);
+ else
+ ret = FALSE;
+
+ if (decide_query)
+ gst_query_unref (decide_query);
+
+ GST_DEBUG_OBJECT (self, "ALLOCATION ret %d, %" GST_PTR_FORMAT, ret, query);
+ return ret;
+ }
+
return gst_pad_query_default (pad, GST_OBJECT (self), query);
}
gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
GstQuery * query)
{
+ GstAggregator *self = GST_AGGREGATOR (parent);
GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
if (GST_QUERY_IS_SERIALIZED (query)) {
+ GstStructure *s;
+ gboolean ret = FALSE;
+
+ SRC_LOCK (self);
PAD_LOCK (aggpad);
+ if (aggpad->priv->flow_return != GST_FLOW_OK) {
+ SRC_UNLOCK (self);
+ goto flushing;
+ }
+
+ g_queue_push_head (&aggpad->priv->buffers, query);
+ SRC_BROADCAST (self);
+ SRC_UNLOCK (self);
+
while (!gst_aggregator_pad_queue_is_empty (aggpad)
&& aggpad->priv->flow_return == GST_FLOW_OK) {
GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
PAD_WAIT_EVENT (aggpad);
}
+ s = gst_query_writable_structure (query);
+ if (gst_structure_get_boolean (s, "gst-aggregator-retval", &ret))
+ gst_structure_remove_field (s, "gst-aggregator-retval");
+ else
+ g_queue_remove (&aggpad->priv->buffers, query);
+
if (aggpad->priv->flow_return != GST_FLOW_OK)
goto flushing;
PAD_UNLOCK (aggpad);
+
+ return ret;
}
- return klass->sink_query (GST_AGGREGATOR (parent),
- GST_AGGREGATOR_PAD (pad), query);
+ return klass->sink_query (self, aggpad, query);
flushing:
GST_DEBUG_OBJECT (aggpad, "Pad is %s, dropping query",
gst_flow_get_name (aggpad->priv->flow_return));
PAD_UNLOCK (aggpad);
+
return FALSE;
}
}
if (event) {
- gboolean is_caps = (GST_EVENT_TYPE (event) == GST_EVENT_CAPS);
-
if (!klass->sink_event (self, aggpad, event)) {
/* Copied from GstPad to convert boolean to a GstFlowReturn in
* the event handling func */
- ret = is_caps ? GST_FLOW_NOT_NEGOTIATED : GST_FLOW_ERROR;
+ ret = GST_FLOW_ERROR;
}
}
g_mutex_init (&pad->priv->lock);
gst_aggregator_pad_reset_unlocked (pad);
+ pad->priv->negotiated = FALSE;
}
/* Must be called with the PAD_LOCK held */
gst_aggregator_pad_clip_buffer_unlocked (GstAggregatorPad * pad)
{
GstAggregator *self = NULL;
- GstAggregatorClass *aggclass;
+ GstAggregatorClass *aggclass = NULL;
GstBuffer *buffer = NULL;
while (pad->priv->clipped_buffer == NULL &&
projects / platform / upstream / gstreamer.git / blobdiff