gboolean pending_flush_stop;
gboolean pending_eos;
+ gboolean first_buffer;
+
GQueue buffers;
guint num_buffers;
GstClockTime head_position;
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;
GstEvent *event;
gboolean result;
gboolean flush;
+ gboolean only_to_active_pads;
gboolean one_actually_seeked;
} EventData;
GstAggregatorClass *klass;
gboolean result;
- self->priv->running = TRUE;
self->priv->send_stream_start = TRUE;
self->priv->send_segment = TRUE;
self->priv->send_eos = TRUE;
GST_OBJECT_UNLOCK (self);
}
+ aggpad->priv->first_buffer = TRUE;
+
/* We never forward the event */
goto eat;
}
SRC_UNLOCK (self);
}
-static GstPad *
-gst_aggregator_request_new_pad (GstElement * element,
+static GstAggregatorPad *
+gst_aggregator_default_create_new_pad (GstAggregator * self,
GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
{
- GstAggregator *self;
GstAggregatorPad *agg_pad;
-
- GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
- GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
-
- self = GST_AGGREGATOR (element);
+ GstElementClass *klass = GST_ELEMENT_GET_CLASS (self);
+ GstAggregatorPrivate *priv = self->priv;
if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
gint serial = 0;
gchar *name = NULL;
- GST_OBJECT_LOCK (element);
+ 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 */
"name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
g_free (name);
- GST_OBJECT_UNLOCK (element);
+ GST_OBJECT_UNLOCK (self);
+ return agg_pad;
} else {
return NULL;
}
+}
+
+static GstPad *
+gst_aggregator_request_new_pad (GstElement * element,
+ GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
+{
+ GstAggregator *self;
+ GstAggregatorPad *agg_pad;
+ GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (element);
+ GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
+
+ self = GST_AGGREGATOR (element);
+
+ agg_pad = klass->create_new_pad (self, templ, req_name, caps);
+ if (!agg_pad) {
+ GST_ERROR_OBJECT (element, "Couldn't create new pad");
+ return NULL;
+ }
GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
self->priv->has_peer_latency = FALSE;
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
GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
if (peer) {
- ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
- GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
- gst_object_unref (peer);
+ if (evdata->only_to_active_pads && aggpad->priv->first_buffer) {
+ GST_DEBUG_OBJECT (pad, "not sending event to inactive pad");
+ ret = TRUE;
+ } else {
+ ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
+ GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
+ gst_object_unref (peer);
+ }
}
if (ret == FALSE) {
- if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
- GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
-
if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
+ GST_DEBUG_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
+
if (gst_pad_query (peer, seeking)) {
gboolean seekable;
static EventData
gst_aggregator_forward_event_to_all_sinkpads (GstAggregator * self,
- GstEvent * event, gboolean flush)
+ GstEvent * event, gboolean flush, gboolean only_to_active_pads)
{
EventData evdata;
evdata.result = TRUE;
evdata.flush = flush;
evdata.one_actually_seeked = FALSE;
+ evdata.only_to_active_pads = only_to_active_pads;
/* We first need to set all pads as flushing in a first pass
* as flush_start flush_stop is sometimes sent synchronously
GST_OBJECT_UNLOCK (self);
/* forward the seek upstream */
- evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, flush);
+ evdata =
+ gst_aggregator_forward_event_to_all_sinkpads (self, event, flush, FALSE);
event = NULL;
if (!evdata.result || !evdata.one_actually_seeked) {
}
}
- evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE);
+ /* Don't forward QOS events to pads that had no active buffer yet. Otherwise
+ * they will receive a QOS event that has earliest_time=0 (because we can't
+ * have negative timestamps), and consider their buffer as too late */
+ evdata =
+ gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE,
+ GST_EVENT_TYPE (event) == GST_EVENT_QOS);
res = evdata.result;
done:
klass->src_event = gst_aggregator_default_src_event;
klass->src_query = gst_aggregator_default_src_query;
+ klass->create_new_pad = gst_aggregator_default_create_new_pad;
+
gstelement_class->request_new_pad =
GST_DEBUG_FUNCPTR (gst_aggregator_request_new_pad);
gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_aggregator_send_event);
buf_pts = GST_BUFFER_PTS (actual_buf);
+ aggpad->priv->first_buffer = FALSE;
+
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;
self->segment.position = start_time;
else
self->segment.position = MIN (start_time, self->segment.position);
- self->segment.start = MIN (start_time, self->segment.start);
- self->segment.time = MIN (start_time, self->segment.time);
GST_DEBUG_OBJECT (self, "Selecting start time %" GST_TIME_FORMAT,
GST_TIME_ARGS (start_time));
}
PAD_UNLOCK (aggpad);
+ GST_OBJECT_UNLOCK (self);
SRC_UNLOCK (self);
done:
g_mutex_init (&pad->priv->flush_lock);
g_mutex_init (&pad->priv->lock);
+
+ pad->priv->first_buffer = TRUE;
}
/**
projects / platform / upstream / gstreamer.git / blobdiff