struct _GstAggregatorPrivate
{
- gint padcount;
+ gint max_padserial;
/* Our state is >= PAUSED */
gboolean running; /* protected by src_lock */
gboolean peer_latency_live; /* protected by src_lock */
GstClockTime peer_latency_min; /* protected by src_lock */
GstClockTime peer_latency_max; /* protected by src_lock */
- gboolean has_peer_latency;
+ gboolean has_peer_latency; /* protected by src_lock */
GstClockTime sub_latency_min; /* protected by src_lock */
GstClockTime sub_latency_max; /* protected by src_lock */
GMutex src_lock;
GCond src_cond;
- gboolean first_buffer;
+ gboolean first_buffer; /* protected by object lock */
GstAggregatorStartTimeSelection start_time_selection;
GstClockTime start_time;
{
GstAggregatorPad *pad;
GList *l, *sinkpads;
+ gboolean have_data = TRUE;
GST_LOG_OBJECT (self, "checking pads");
PAD_LOCK (pad);
- /* In live mode, having a single pad with buffers is enough to
- * generate a start time from it. In non-live mode all pads need
- * to have a buffer
- */
- if (self->priv->peer_latency_live &&
- !gst_aggregator_pad_queue_is_empty (pad))
- self->priv->first_buffer = FALSE;
+ if (gst_aggregator_pad_queue_is_empty (pad)) {
+ if (!pad->priv->eos) {
+ have_data = FALSE;
- if (gst_aggregator_pad_queue_is_empty (pad) && !pad->priv->eos) {
- PAD_UNLOCK (pad);
- goto pad_not_ready;
+ /* If not live we need data on all pads, so leave the loop */
+ if (!self->priv->peer_latency_live) {
+ PAD_UNLOCK (pad);
+ goto pad_not_ready;
+ }
+ }
+ } else if (self->priv->peer_latency_live) {
+ /* In live mode, having a single pad with buffers is enough to
+ * generate a start time from it. In non-live mode all pads need
+ * to have a buffer
+ */
+ self->priv->first_buffer = FALSE;
}
- PAD_UNLOCK (pad);
+ PAD_UNLOCK (pad);
}
+ if (!have_data)
+ goto pad_not_ready;
+
self->priv->first_buffer = FALSE;
GST_OBJECT_UNLOCK (self);
* and if a pad does not have a buffer in time we ignore
* that pad.
*/
+ GST_OBJECT_LOCK (self);
if (!GST_CLOCK_TIME_IS_VALID (latency) ||
!GST_IS_CLOCK (GST_ELEMENT_CLOCK (self)) ||
!GST_CLOCK_TIME_IS_VALID (start) ||
* then check if we're ready now. If we return FALSE,
* we will be directly called again.
*/
+ GST_OBJECT_UNLOCK (self);
SRC_WAIT (self);
} else {
GstClockTime base_time, time;
GST_DEBUG_OBJECT (self, "got subclass start time: %" GST_TIME_FORMAT,
GST_TIME_ARGS (start));
- GST_OBJECT_LOCK (self);
base_time = GST_ELEMENT_CAST (self)->base_time;
- clock = GST_ELEMENT_CLOCK (self);
- if (clock)
- gst_object_ref (clock);
+ clock = gst_object_ref (GST_ELEMENT_CLOCK (self));
GST_OBJECT_UNLOCK (self);
time = base_time + start;
GST_TIME_FORMAT " (base %" GST_TIME_FORMAT " start %" GST_TIME_FORMAT
" latency %" GST_TIME_FORMAT " current %" GST_TIME_FORMAT ")",
GST_TIME_ARGS (time),
- GST_TIME_ARGS (GST_ELEMENT_CAST (self)->base_time),
+ GST_TIME_ARGS (base_time),
GST_TIME_ARGS (start), GST_TIME_ARGS (latency),
GST_TIME_ARGS (gst_clock_get_time (clock)));
}
GST_DEBUG_OBJECT (self,
- "clock returned %d (jitter: %s%" GST_TIME_FORMAT ")",
- status, (jitter < 0 ? "-" : " "),
- GST_TIME_ARGS ((jitter < 0 ? -jitter : jitter)));
+ "clock returned %d (jitter: %" GST_STIME_FORMAT ")",
+ status, GST_STIME_ARGS (jitter));
/* we timed out */
if (status == GST_CLOCK_OK || status == GST_CLOCK_EARLY) {
GstAggregatorClass *klass;
gboolean result;
- self->priv->running = TRUE;
self->priv->send_stream_start = TRUE;
self->priv->send_segment = TRUE;
self->priv->send_eos = TRUE;
GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
{
GstAggregatorPad *agg_pad;
- GstElementClass *klass = GST_ELEMENT_GET_CLASS (self);
GstAggregatorPrivate *priv = self->priv;
+ gint serial = 0;
+ gchar *name = NULL;
- if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
- gint serial = 0;
- gchar *name = NULL;
+ if (templ->direction != GST_PAD_SINK ||
+ g_strcmp0 (templ->name_template, "sink_%u") != 0)
+ goto not_sink;
- GST_OBJECT_LOCK (self);
- if (req_name == NULL || strlen (req_name) < 6
- || !g_str_has_prefix (req_name, "sink_")) {
- /* no name given when requesting the pad, use next available int */
- priv->padcount++;
- } else {
- /* parse serial number from requested padname */
- serial = g_ascii_strtoull (&req_name[5], NULL, 10);
- if (serial >= priv->padcount)
- priv->padcount = serial;
- }
+ GST_OBJECT_LOCK (self);
+ if (req_name == NULL || strlen (req_name) < 6
+ || !g_str_has_prefix (req_name, "sink_")) {
+ /* no name given when requesting the pad, use next available int */
+ serial = ++priv->max_padserial;
+ } else {
+ /* parse serial number from requested padname */
+ serial = g_ascii_strtoull (&req_name[5], NULL, 10);
+ if (serial > priv->max_padserial)
+ priv->max_padserial = serial;
+ }
- name = g_strdup_printf ("sink_%u", priv->padcount);
- agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
- "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
- g_free (name);
+ name = g_strdup_printf ("sink_%u", serial);
+ agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
+ "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
+ g_free (name);
- GST_OBJECT_UNLOCK (self);
+ GST_OBJECT_UNLOCK (self);
- return agg_pad;
- } else {
+ return agg_pad;
+
+ /* errors */
+not_sink:
+ {
+ GST_WARNING_OBJECT (self, "request new pad that is not a SINK pad\n");
return NULL;
}
}
else
max = GST_CLOCK_TIME_NONE;
- if (live && min > max) {
- GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
- ("%s", "Latency too big"),
- ("The requested latency value is too big for the current pipeline. "
- "Limiting to %" G_GINT64_FORMAT, max));
- min = max;
- /* FIXME: This could in theory become negative, but in
- * that case all is lost anyway */
- self->priv->latency -= min - max;
- /* FIXME: shouldn't we g_object_notify() the change here? */
- }
-
SRC_BROADCAST (self);
GST_DEBUG_OBJECT (self, "configured latency live:%s min:%" G_GINT64_FORMAT
* Gets the latency value. See gst_aggregator_set_latency for
* more details.
*
- * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
+ * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
* before a pad is deemed unresponsive. A value of -1 means an
* unlimited time.
*/
gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
g_return_if_fail (pad_template != NULL);
- priv->padcount = -1;
+ priv->max_padserial = -1;
priv->tags_changed = FALSE;
self->priv->peer_latency_live = FALSE;
if (aggpad->priv->pending_eos == TRUE)
goto eos;
- flow_return = aggpad->priv->flow_return;
- if (flow_return != GST_FLOW_OK)
- goto flushing;
-
PAD_UNLOCK (aggpad);
if (aggclass->clip && head) {
for (;;) {
SRC_LOCK (self);
+ GST_OBJECT_LOCK (self);
PAD_LOCK (aggpad);
if (gst_aggregator_pad_has_space (self, aggpad)
&& aggpad->priv->flow_return == GST_FLOW_OK) {
flow_return = aggpad->priv->flow_return;
if (flow_return != GST_FLOW_OK) {
+ GST_OBJECT_UNLOCK (self);
SRC_UNLOCK (self);
goto flushing;
}
GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
+ GST_OBJECT_UNLOCK (self);
SRC_UNLOCK (self);
PAD_WAIT_EVENT (aggpad);
start_time = 0;
break;
case GST_AGGREGATOR_START_TIME_SELECTION_FIRST:
+ GST_OBJECT_LOCK (aggpad);
if (aggpad->segment.format == GST_FORMAT_TIME) {
start_time = buf_pts;
if (start_time != -1) {
"as the segment is a %s segment instead of a time segment",
gst_format_get_name (aggpad->segment.format));
}
+ GST_OBJECT_UNLOCK (aggpad);
break;
case GST_AGGREGATOR_START_TIME_SELECTION_SET:
start_time = self->priv->start_time;
}
PAD_UNLOCK (aggpad);
+ GST_OBJECT_UNLOCK (self);
SRC_UNLOCK (self);
done:
return FALSE;
}
-static gboolean
+static GstFlowReturn
gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
GstEvent * event)
{
+ GstFlowReturn ret = GST_FLOW_OK;
GstAggregator *self = GST_AGGREGATOR (parent);
GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
PAD_LOCK (aggpad);
if (aggpad->priv->flow_return != GST_FLOW_OK
- && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP)
+ && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
+ ret = aggpad->priv->flow_return;
goto flushing;
+ }
if (GST_EVENT_TYPE (event) == GST_EVENT_SEGMENT) {
GST_OBJECT_LOCK (aggpad);
SRC_UNLOCK (self);
}
- if (event)
- return klass->sink_event (self, aggpad, event);
- else
- return TRUE;
+ 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;
+ }
+ }
+
+ return ret;
flushing:
GST_DEBUG_OBJECT (aggpad, "Pad is %s, dropping event",
if (GST_EVENT_IS_STICKY (event))
gst_pad_store_sticky_event (pad, event);
gst_event_unref (event);
- return FALSE;
+
+ return ret;
}
static gboolean
gst_pad_set_chain_function (pad,
GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
- gst_pad_set_event_function (pad,
- GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func));
+ gst_pad_set_event_full_function_full (pad,
+ GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func),
+ NULL, NULL);
gst_pad_set_query_function (pad,
GST_DEBUG_FUNCPTR (gst_aggregator_pad_query_func));
gst_pad_set_activatemode_function (pad,
projects / platform / upstream / gstreamer.git / blobdiff