* * When data is queued on all pads, the aggregate vmethod is called.
*
* * One can peek at the data on any given GstAggregatorPad with the
- * gst_aggregator_pad_get_buffer () method, and take ownership of it
- * with the gst_aggregator_pad_steal_buffer () method. When a buffer
- * has been taken with steal_buffer (), a new buffer can be queued
+ * gst_aggregator_pad_peek_buffer () method, and remove it from the pad
+ * with the gst_aggregator_pad_pop_buffer () method. When a buffer
+ * has been taken with pop_buffer (), a new buffer can be queued
* on that pad.
*
* * If the subclass wishes to push a buffer downstream in its aggregate
* flag these buffers with GST_BUFFER_FLAG_GAP and GST_BUFFER_FLAG_DROPPABLE
* to ease their identification and subsequent processing.
*
+ * * Subclasses must use (a subclass of) #GstAggregatorPad for both their
+ * sink and source pads.
+ * See gst_element_class_add_static_pad_template_with_gtype().
+ *
* This class used to live in gst-plugins-bad and was moved to core.
*
* Since: 1.14
}
/* Might become API */
+#if 0
static void gst_aggregator_merge_tags (GstAggregator * aggregator,
const GstTagList * tags, GstTagMergeMode mode);
+#endif
static void gst_aggregator_set_latency_property (GstAggregator * agg,
GstClockTime latency);
static GstClockTime gst_aggregator_get_latency_property (GstAggregator * agg);
static GstClockTime gst_aggregator_get_latency_unlocked (GstAggregator * self);
+static void gst_aggregator_pad_buffer_consumed (GstAggregatorPad * pad);
+
GST_DEBUG_CATEGORY_STATIC (aggregator_debug);
#define GST_CAT_DEFAULT aggregator_debug
* GstAggregator implementation *
*************************************/
static GstElementClass *aggregator_parent_class = NULL;
+static gint aggregator_private_offset = 0;
/* All members are protected by the object lock unless otherwise noted */
static gboolean
gst_aggregator_check_pads_ready (GstAggregator * self)
{
- GstAggregatorPad *pad;
+ GstAggregatorPad *pad = NULL;
GList *l, *sinkpads;
gboolean have_buffer = TRUE;
gboolean have_event_or_query = FALSE;
GST_OBJECT_LOCK (self);
self->priv->send_stream_start = TRUE;
self->priv->send_segment = TRUE;
- gst_segment_init (&self->segment, GST_FORMAT_TIME);
+ gst_segment_init (&GST_AGGREGATOR_PAD (self->srcpad)->segment,
+ GST_FORMAT_TIME);
self->priv->first_buffer = TRUE;
GST_OBJECT_UNLOCK (self);
}
GST_INFO_OBJECT (self, "pushing stream start");
/* stream-start (FIXME: create id based on input ids) */
g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
- if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
+ if (!gst_pad_push_event (GST_PAD (self->srcpad),
+ gst_event_new_stream_start (s_id))) {
GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
}
self->priv->send_stream_start = FALSE;
GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
self->priv->srccaps);
- if (!gst_pad_push_event (self->srcpad,
+ if (!gst_pad_push_event (GST_PAD (self->srcpad),
gst_event_new_caps (self->priv->srccaps))) {
GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
}
GST_OBJECT_LOCK (self);
if (self->priv->send_segment && !self->priv->flush_seeking) {
- segment = gst_event_new_segment (&self->segment);
+ segment =
+ gst_event_new_segment (&GST_AGGREGATOR_PAD (self->srcpad)->segment);
if (!self->priv->seqnum)
/* This code-path is in preparation to be able to run without a source
GST_PAD_STREAM_UNLOCK (self->srcpad);
}
-/**
- * gst_aggregator_finish_buffer:
- * @self: The #GstAggregator
- * @buffer: (transfer full): the #GstBuffer to push.
- *
- * This method will push the provided output buffer downstream. If needed,
- * mandatory events such as stream-start, caps, and segment events will be
- * sent before pushing the buffer.
- */
-GstFlowReturn
-gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
+static GstFlowReturn
+gst_aggregator_default_finish_buffer (GstAggregator * self, GstBuffer * buffer)
{
gst_aggregator_push_mandatory_events (self);
}
}
+/**
+ * gst_aggregator_finish_buffer:
+ * @aggregator: The #GstAggregator
+ * @buffer: (transfer full): the #GstBuffer to push.
+ *
+ * This method will push the provided output buffer downstream. If needed,
+ * mandatory events such as stream-start, caps, and segment events will be
+ * sent before pushing the buffer.
+ */
+GstFlowReturn
+gst_aggregator_finish_buffer (GstAggregator * aggregator, GstBuffer * buffer)
+{
+ GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (aggregator);
+
+ g_assert (klass->finish_buffer != NULL);
+
+ return klass->finish_buffer (aggregator, buffer);
+}
+
static void
gst_aggregator_push_eos (GstAggregator * self)
{
return TRUE;
}
+static gboolean
+gst_aggregator_pad_skip_buffers (GstElement * self, GstPad * epad,
+ gpointer user_data)
+{
+ GList *item;
+ GstAggregatorPad *aggpad = (GstAggregatorPad *) epad;
+ GstAggregator *agg = (GstAggregator *) self;
+ GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
+
+ if (!klass->skip_buffer)
+ return FALSE;
+
+ PAD_LOCK (aggpad);
+
+ item = g_queue_peek_head_link (&aggpad->priv->data);
+ while (item) {
+ GList *next = item->next;
+
+ if (GST_IS_BUFFER (item->data)
+ && klass->skip_buffer (aggpad, agg, item->data)) {
+ GST_LOG_OBJECT (aggpad, "Skipping %" GST_PTR_FORMAT, item->data);
+ gst_aggregator_pad_buffer_consumed (aggpad);
+ gst_buffer_unref (item->data);
+ g_queue_delete_link (&aggpad->priv->data, item);
+ } else {
+ break;
+ }
+
+ item = next;
+ }
+
+ PAD_UNLOCK (aggpad);
+
+ return TRUE;
+}
+
static void
gst_aggregator_pad_set_flushing (GstAggregatorPad * aggpad,
GstFlowReturn flow_return, gboolean full)
gst_element_foreach_sink_pad (GST_ELEMENT_CAST (self),
gst_aggregator_do_events_and_queries, NULL);
+ if (self->priv->peer_latency_live)
+ gst_element_foreach_sink_pad (GST_ELEMENT_CAST (self),
+ gst_aggregator_pad_skip_buffers, NULL);
+
/* Ensure we have buffers ready (either in clipped_buffer or at the head of
* the queue */
if (!gst_aggregator_wait_and_check (self, &timeout))
goto eat;
}
case GST_EVENT_TAG:
- {
- GstTagList *tags;
-
- gst_event_parse_tag (event, &tags);
-
- if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
- gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
- gst_event_unref (event);
- event = NULL;
- goto eat;
- }
- break;
- }
+ goto eat;
default:
{
break;
&start, &stop_type, &stop);
GST_OBJECT_LOCK (self);
- gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
- stop_type, stop, NULL);
+ gst_segment_do_seek (&GST_AGGREGATOR_PAD (self->srcpad)->segment, rate, fmt,
+ flags, start_type, start, stop_type, stop, NULL);
self->priv->seqnum = gst_event_get_seqnum (event);
self->priv->first_buffer = FALSE;
GST_OBJECT_UNLOCK (self);
} 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);
}
}
evdata->result &= ret;
+ if (peer)
+ gst_object_unref (peer);
+
/* Always send to all pads */
return FALSE;
}
priv->flush_seeking = TRUE;
}
- gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
- stop_type, stop, NULL);
+ gst_segment_do_seek (&GST_AGGREGATOR_PAD (self->srcpad)->segment, rate, fmt,
+ flags, start_type, start, stop_type, stop, NULL);
/* Seeking sets a position */
self->priv->first_buffer = FALSE;
GstElementClass *gstelement_class = (GstElementClass *) klass;
aggregator_parent_class = g_type_class_peek_parent (klass);
- g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
GST_DEBUG_FG_MAGENTA, "GstAggregator");
+ if (aggregator_private_offset != 0)
+ g_type_class_adjust_private_offset (klass, &aggregator_private_offset);
+
+ klass->finish_buffer = gst_aggregator_default_finish_buffer;
+
klass->sink_event = gst_aggregator_default_sink_event;
klass->sink_query = gst_aggregator_default_sink_query;
"Start time to use if start-time-selection=set", 0,
G_MAXUINT64,
DEFAULT_START_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+}
- GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_do_events_and_queries);
+static inline gpointer
+gst_aggregator_get_instance_private (GstAggregator * self)
+{
+ return (G_STRUCT_MEMBER_P (self, aggregator_private_offset));
}
static void
g_return_if_fail (klass->aggregate != NULL);
- self->priv =
- G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
- GstAggregatorPrivate);
+ self->priv = gst_aggregator_get_instance_private (self);
priv = self->priv;
self->priv->peer_latency_min = self->priv->sub_latency_min = 0;
self->priv->peer_latency_max = self->priv->sub_latency_max = 0;
self->priv->has_peer_latency = FALSE;
- gst_aggregator_reset_flow_values (self);
self->srcpad = gst_pad_new_from_template (pad_template, "src");
+ gst_aggregator_reset_flow_values (self);
+
gst_pad_set_event_function (self->srcpad,
GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_event_func));
gst_pad_set_query_function (self->srcpad,
_type = g_type_register_static (GST_TYPE_ELEMENT,
"GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
+
+ aggregator_private_offset =
+ g_type_add_instance_private (_type, sizeof (GstAggregatorPrivate));
+
g_once_init_leave (&type, _type);
}
return type;
GstFlowReturn flow_return;
GstClockTime buf_pts;
- GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
-
PAD_LOCK (aggpad);
flow_return = aggpad->priv->flow_return;
if (flow_return != GST_FLOW_OK)
if (self->priv->first_buffer) {
GstClockTime start_time;
+ GstAggregatorPad *srcpad = GST_AGGREGATOR_PAD (self->srcpad);
switch (self->priv->start_time_selection) {
case GST_AGGREGATOR_START_TIME_SELECTION_ZERO:
}
if (start_time != -1) {
- if (self->segment.position == -1)
- self->segment.position = start_time;
+ if (srcpad->segment.position == -1)
+ srcpad->segment.position = start_time;
else
- self->segment.position = MIN (start_time, self->segment.position);
+ srcpad->segment.position = MIN (start_time, srcpad->segment.position);
GST_DEBUG_OBJECT (self, "Selecting start time %" GST_TIME_FORMAT,
GST_TIME_ARGS (start_time));
/***********************************
* GstAggregatorPad implementation *
************************************/
-G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
+G_DEFINE_TYPE_WITH_PRIVATE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
static void
gst_aggregator_pad_constructed (GObject * object)
{
GstPad *pad = GST_PAD (object);
- gst_pad_set_chain_function (pad,
- GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
- 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,
- GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
+ if (GST_PAD_IS_SINK (pad)) {
+ gst_pad_set_chain_function (pad,
+ GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
+ 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,
+ GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
+ }
}
static void
{
GObjectClass *gobject_class = (GObjectClass *) klass;
- g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
-
gobject_class->constructed = gst_aggregator_pad_constructed;
gobject_class->finalize = gst_aggregator_pad_finalize;
gobject_class->dispose = gst_aggregator_pad_dispose;
static void
gst_aggregator_pad_init (GstAggregatorPad * pad)
{
- pad->priv =
- G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
- GstAggregatorPadPrivate);
+ pad->priv = gst_aggregator_pad_get_instance_private (pad);
g_queue_init (&pad->priv->data);
g_cond_init (&pad->priv->event_cond);
}
/**
- * gst_aggregator_pad_steal_buffer:
+ * gst_aggregator_pad_pop_buffer:
* @pad: the pad to get buffer from
*
* Steal the ref to the buffer currently queued in @pad.
* queued. You should unref the buffer after usage.
*/
GstBuffer *
-gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
+gst_aggregator_pad_pop_buffer (GstAggregatorPad * pad)
{
GstBuffer *buffer;
{
GstBuffer *buf;
- buf = gst_aggregator_pad_steal_buffer (pad);
+ buf = gst_aggregator_pad_pop_buffer (pad);
if (buf == NULL)
return FALSE;
}
/**
- * gst_aggregator_pad_get_buffer:
+ * gst_aggregator_pad_peek_buffer:
* @pad: the pad to get buffer from
*
* Returns: (transfer full): A reference to the buffer in @pad or
* usage.
*/
GstBuffer *
-gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
+gst_aggregator_pad_peek_buffer (GstAggregatorPad * pad)
{
GstBuffer *buffer;
}
/**
+ * gst_aggregator_pad_has_buffer:
+ * @pad: the pad to check the buffer on
+ *
+ * This checks if a pad has a buffer available that will be returned by
+ * a call to gst_aggregator_pad_peek_buffer() or
+ * gst_aggregator_pad_pop_buffer().
+ *
+ * Returns: %TRUE if the pad has a buffer available as the next thing.
+ *
+ * Since: 1.14.1
+ */
+gboolean
+gst_aggregator_pad_has_buffer (GstAggregatorPad * pad)
+{
+ gboolean has_buffer;
+
+ PAD_LOCK (pad);
+ gst_aggregator_pad_clip_buffer_unlocked (pad);
+ has_buffer = (pad->priv->clipped_buffer != NULL);
+ PAD_UNLOCK (pad);
+
+ return has_buffer;
+}
+
+/**
* gst_aggregator_pad_is_eos:
* @pad: an aggregator pad
*
return is_eos;
}
-/**
+#if 0
+/*
* gst_aggregator_merge_tags:
* @self: a #GstAggregator
* @tags: a #GstTagList to merge
self->priv->tags_changed = TRUE;
GST_OBJECT_UNLOCK (self);
}
+#endif
/**
* gst_aggregator_set_latency:
if (params)
*params = self->priv->allocation_params;
}
+
+/**
+ * gst_aggregator_simple_get_next_time:
+ * @self: A #GstAggregator
+ *
+ * This is a simple #GstAggregator::get_next_time implementation that
+ * just looks at the #GstSegment on the srcpad of the aggregator and bases
+ * the next time on the running there there.
+ *
+ * This is the desired behaviour in most cases where you have a live source
+ * and you have a dead line based aggregator subclass.
+ *
+ * Returns: The running time based on the position
+ *
+ * Since: 1.16
+ */
+GstClockTime
+gst_aggregator_simple_get_next_time (GstAggregator * self)
+{
+ GstClockTime next_time;
+ GstAggregatorPad *srcpad = GST_AGGREGATOR_PAD (self->srcpad);
+ GstSegment *segment = &srcpad->segment;
+
+ GST_OBJECT_LOCK (self);
+ if (segment->position == -1 || segment->position < segment->start)
+ next_time = segment->start;
+ else
+ next_time = segment->position;
+
+ if (segment->stop != -1 && next_time > segment->stop)
+ next_time = segment->stop;
+
+ next_time = gst_segment_to_running_time (segment, GST_FORMAT_TIME, next_time);
+ GST_OBJECT_UNLOCK (self);
+
+ return next_time;
+}