From 55c65d079f7fcdecdd5500a249507ed78b80bb0a Mon Sep 17 00:00:00 2001 From: Thibault Saunier Date: Thu, 22 May 2014 19:44:37 +0200 Subject: [PATCH] aggregator: Add new GstAggregator base class This base class has been added to a newly created libgstbadbase library Co-Authored by: Mathieu Duponchelle https://bugzilla.gnome.org/show_bug.cgi?id=731917 --- configure.ac | 3 + docs/libs/gst-plugins-bad-libs-docs.sgml | 6 + docs/libs/gst-plugins-bad-libs-sections.txt | 35 + gst-libs/gst/Makefile.am | 4 +- gst-libs/gst/base/Makefile.am | 21 + gst-libs/gst/base/gstaggregator.c | 1323 ++++++++++++++++++++++++ gst-libs/gst/base/gstaggregator.h | 256 +++++ pkgconfig/gstreamer-bad-base-uninstalled.pc.in | 15 + pkgconfig/gstreamer-bad-base.pc.in | 15 + tests/check/Makefile.am | 11 + tests/check/libs/aggregator.c | 1017 ++++++++++++++++++ 11 files changed, 2704 insertions(+), 2 deletions(-) create mode 100644 gst-libs/gst/base/Makefile.am create mode 100644 gst-libs/gst/base/gstaggregator.c create mode 100644 gst-libs/gst/base/gstaggregator.h create mode 100644 pkgconfig/gstreamer-bad-base-uninstalled.pc.in create mode 100644 pkgconfig/gstreamer-bad-base.pc.in create mode 100644 tests/check/libs/aggregator.c diff --git a/configure.ac b/configure.ac index b0fdd31..a28b9bd 100644 --- a/configure.ac +++ b/configure.ac @@ -3221,6 +3221,7 @@ gst-libs/gst/codecparsers/Makefile gst-libs/gst/mpegts/Makefile gst-libs/gst/uridownloader/Makefile gst-libs/gst/wayland/Makefile +gst-libs/gst/base/Makefile sys/Makefile sys/dshowdecwrapper/Makefile sys/acmenc/Makefile @@ -3359,6 +3360,8 @@ pkgconfig/gstreamer-mpegts.pc pkgconfig/gstreamer-mpegts-uninstalled.pc pkgconfig/gstreamer-wayland.pc pkgconfig/gstreamer-wayland-uninstalled.pc +pkgconfig/gstreamer-bad-base.pc +pkgconfig/gstreamer-bad-base-uninstalled.pc tools/Makefile m4/Makefile ) diff --git a/docs/libs/gst-plugins-bad-libs-docs.sgml b/docs/libs/gst-plugins-bad-libs-docs.sgml index 44c2cd6..7ef65da 100644 --- a/docs/libs/gst-plugins-bad-libs-docs.sgml +++ b/docs/libs/gst-plugins-bad-libs-docs.sgml @@ -57,6 +57,12 @@ + + GStreamer Base classes from gst-plugins-bad + + + + OpenGL Helper Library diff --git a/docs/libs/gst-plugins-bad-libs-sections.txt b/docs/libs/gst-plugins-bad-libs-sections.txt index c54978e..de81686 100644 --- a/docs/libs/gst-plugins-bad-libs-sections.txt +++ b/docs/libs/gst-plugins-bad-libs-sections.txt @@ -1290,3 +1290,38 @@ gst_gl_window_x11_get_type GstGLWindowX11Private x11_event_source_new + +
+gstaggregator +GstAggregator +GstAggregator +GstAggregatorClass +gst_aggregator_finish_buffer +gst_aggregator_set_src_caps +gst_aggregator_iterate_sinkpads + +GST_IS_AGGREGATOR +GST_IS_AGGREGATOR_CLASS +GST_TYPE_AGGREGATOR +GST_AGGREGATOR +GST_AGGREGATOR_CLASS +GST_AGGREGATOR_GET_CLASS +gst_aggregator_get_type +
+ +
+gstaggregatorpad +GstAggregatorPad +GstAggregatorPad +GstAggregatorPadClass +gst_aggregator_pad_steal_buffer +gst_aggregator_pad_get_buffer + +GST_IS_AGGREGATOR_PAD +GST_IS_AGGREGATOR_PAD_CLASS +GST_TYPE_AGGREGATOR_PAD +GST_AGGREGATOR_PAD +GST_AGGREGATOR_PAD_CLASS +GST_AGGREGATOR_PAD_GET_CLASS +gst_aggregator_pad_get_type +
diff --git a/gst-libs/gst/Makefile.am b/gst-libs/gst/Makefile.am index 93abc17..7932f41 100644 --- a/gst-libs/gst/Makefile.am +++ b/gst-libs/gst/Makefile.am @@ -11,8 +11,8 @@ WAYLAND_DIR=wayland endif SUBDIRS = interfaces basecamerabinsrc codecparsers \ - insertbin uridownloader mpegts $(GL_DIR) $(WAYLAND_DIR) + insertbin uridownloader mpegts base $(GL_DIR) $(WAYLAND_DIR) noinst_HEADERS = gst-i18n-plugin.h gettext.h glib-compat-private.h DIST_SUBDIRS = interfaces gl basecamerabinsrc codecparsers \ - insertbin uridownloader mpegts wayland + insertbin uridownloader mpegts wayland base diff --git a/gst-libs/gst/base/Makefile.am b/gst-libs/gst/base/Makefile.am new file mode 100644 index 0000000..c71197b --- /dev/null +++ b/gst-libs/gst/base/Makefile.am @@ -0,0 +1,21 @@ +lib_LTLIBRARIES = libgstbadbase-@GST_API_VERSION@.la + +libgstbadbase_@GST_API_VERSION@_la_SOURCES = \ + gstaggregator.c + +libgstbadbase_@GST_API_VERSION@_la_CFLAGS = $(GST_CFLAGS) +libgstbadbase_@GST_API_VERSION@_la_LIBADD = $(GST_LIBS) +libgstbadbase_@GST_API_VERSION@_la_LDFLAGS = $(GST_LDFLAGS) + +libgstbadbase_@GST_API_VERSION@includedir = $(includedir)/gstreamer-@GST_API_VERSION@/gst/badbase/ +libgstbadbase_@GST_API_VERSION@include_HEADERS = \ + gstaggregator.h + +EXTRA_DIST = + +CLEANFILES = *.gcno *.gcda *.gcov + +%.c.gcov: .libs/libgstbadbase_@GST_API_VERSION@_la-%.gcda %.c + $(GCOV) -b -f -o $^ > $@.out + +gcov: $(libgstbadbase_@GST_API_VERSION@_la_SOURCES:=.gcov) diff --git a/gst-libs/gst/base/gstaggregator.c b/gst-libs/gst/base/gstaggregator.c new file mode 100644 index 0000000..c566384 --- /dev/null +++ b/gst-libs/gst/base/gstaggregator.c @@ -0,0 +1,1323 @@ +/* GStreamer + * Copyright (C) 2014 Mathieu Duponchelle + * Copyright (C) 2014 Thibault Saunier + * + * gstaggregator.c: + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, + * Boston, MA 02110-1301, USA. + */ +/** + * SECTION: gstaggregator + * @short_description: manages a set of pads with the purpose of + * aggregating their buffers. + * @see_also: gstcollectpads for historical reasons. + * + * Manages a set of pads with the purpose of aggregating their buffers. + * Control is given to the subclass when all pads have data. + * + * + * Base class for mixers and muxers. Implementers should at least implement + * the aggregate () vmethod. + * + * + * When data is queued on all pads, tha 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 + * on that pad. + * + * + * If the subclass wishes to push a buffer downstream in its aggregate + * implementation, it should do so through the + * gst_aggregator_finish_buffer () method. This method will take care + * of sending and ordering mandatory events such as stream start, caps + * and segment. + * + * + * Same goes for EOS events, which should not be pushed directly by the + * subclass, it should instead return GST_FLOW_EOS in its aggregate + * implementation. + * + * + */ + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include /* strlen */ + +#include "gstaggregator.h" + + +/* Might become API */ +static void gst_aggregator_merge_tags (GstAggregator * aggregator, + const GstTagList * tags, GstTagMergeMode mode); + +GST_DEBUG_CATEGORY_STATIC (aggregator_debug); +#define GST_CAT_DEFAULT aggregator_debug + +/* GstAggregatorPad definitions */ +#define PAD_LOCK_EVENT(pad) G_STMT_START { \ + GST_LOG_OBJECT (pad, "Taking EVENT lock from thread %p", \ + g_thread_self()); \ + g_mutex_lock(&pad->priv->event_lock); \ + GST_LOG_OBJECT (pad, "Took EVENT lock from thread %p", \ + g_thread_self()); \ + } G_STMT_END + +#define PAD_UNLOCK_EVENT(pad) G_STMT_START { \ + GST_LOG_OBJECT (pad, "Releasing EVENT lock from thread %p", \ + g_thread_self()); \ + g_mutex_unlock(&pad->priv->event_lock); \ + GST_LOG_OBJECT (pad, "Release EVENT lock from thread %p", \ + g_thread_self()); \ + } G_STMT_END + + +#define PAD_WAIT_EVENT(pad) G_STMT_START { \ + GST_LOG_OBJECT (pad, "Waiting for EVENT on thread %p", \ + g_thread_self()); \ + g_cond_wait(&(((GstAggregatorPad* )pad)->priv->event_cond), \ + &(pad->priv->event_lock)); \ + GST_LOG_OBJECT (pad, "DONE Waiting for EVENT on thread %p", \ + g_thread_self()); \ + } G_STMT_END + +#define PAD_BROADCAST_EVENT(pad) { \ + GST_LOG_OBJECT (pad, "Signaling EVENT from thread %p", \ + g_thread_self()); \ + g_cond_broadcast(&(((GstAggregatorPad* )pad)->priv->event_cond)); \ + } + +struct _GstAggregatorPadPrivate +{ + gboolean pending_flush_start; + gboolean pending_flush_stop; + gboolean pending_eos; + gboolean flushing; + + GMutex event_lock; + GCond event_cond; +}; + +static gboolean +_aggpad_flush (GstAggregatorPad * aggpad, GstAggregator * agg) +{ + GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad); + + aggpad->eos = FALSE; + aggpad->priv->flushing = FALSE; + + if (klass->flush) + return klass->flush (aggpad, agg); + + return TRUE; +} + +/************************************* + * GstAggregator implementation * + *************************************/ +static GstElementClass *aggregator_parent_class = NULL; + +#define MAIN_CONTEXT_LOCK(self) G_STMT_START { \ + GST_LOG_OBJECT (self, "Getting MAIN_CONTEXT_LOCK in thread %p", \ + g_thread_self()); \ + g_mutex_lock(&((GstAggregator*)self)->priv->mcontext_lock); \ + GST_LOG_OBJECT (self, "Got MAIN_CONTEXT_LOCK in thread %p", \ + g_thread_self()); \ +} G_STMT_END + +#define MAIN_CONTEXT_UNLOCK(self) G_STMT_START { \ + g_mutex_unlock(&((GstAggregator*)self)->priv->mcontext_lock); \ + GST_LOG_OBJECT (self, "Unlocked MAIN_CONTEXT_LOCK in thread %p", \ + g_thread_self()); \ +} G_STMT_END + +struct _GstAggregatorPrivate +{ + gint padcount; + + GMainContext *mcontext; + + /* Our state is >= PAUSED */ + gboolean running; + + /* Ensure that when we remove all sources from the maincontext + * we can not add any source, avoiding: + * "g_source_attach: assertion '!SOURCE_DESTROYED (source)' failed" */ + GMutex mcontext_lock; + + gboolean send_stream_start; + gboolean send_segment; + gboolean flush_seeking; + gboolean pending_flush_start; + gboolean send_eos; + GstFlowReturn flow_return; + + GstCaps *srccaps; + + GstTagList *tags; + gboolean tags_changed; +}; + +typedef struct +{ + GstEvent *event; + gboolean result; + gboolean flush; +} EventData; + +/** + * gst_aggregator_iterate_sinkpads: + * @self: The #GstAggregator + * @func: The function to call. + * @user_data: The data to pass to @func. + * + * Iterate the sinkpads of aggregator to call a function on them. + * + * This method guarantees that @func will be called only once for each + * sink pad. + */ +gboolean +gst_aggregator_iterate_sinkpads (GstAggregator * self, + GstAggregatorPadForeachFunc func, gpointer user_data) +{ + gboolean result = FALSE; + GstIterator *iter; + gboolean done = FALSE; + GValue item = { 0, }; + GList *seen_pads = NULL; + + iter = gst_element_iterate_sink_pads (GST_ELEMENT (self)); + + if (!iter) + goto no_iter; + + while (!done) { + switch (gst_iterator_next (iter, &item)) { + case GST_ITERATOR_OK: + { + GstPad *pad; + + pad = g_value_get_object (&item); + + /* if already pushed, skip. FIXME, find something faster to tag pads */ + if (pad == NULL || g_list_find (seen_pads, pad)) { + g_value_reset (&item); + break; + } + + GST_LOG_OBJECT (self, "calling function on pad %s:%s", + GST_DEBUG_PAD_NAME (pad)); + result = func (self, pad, user_data); + + done = !result; + + seen_pads = g_list_prepend (seen_pads, pad); + + g_value_reset (&item); + break; + } + case GST_ITERATOR_RESYNC: + gst_iterator_resync (iter); + break; + case GST_ITERATOR_ERROR: + GST_ERROR_OBJECT (self, + "Could not iterate over internally linked pads"); + done = TRUE; + break; + case GST_ITERATOR_DONE: + done = TRUE; + break; + } + } + g_value_unset (&item); + gst_iterator_free (iter); + + if (seen_pads == NULL) { + GST_DEBUG_OBJECT (self, "No pad seen"); + return FALSE; + } + + g_list_free (seen_pads); + +no_iter: + return result; +} + +static inline gboolean +_check_all_pads_with_data_or_eos (GstAggregator * self, + GstAggregatorPad * aggpad) +{ + if (aggpad->buffer || aggpad->eos) { + return TRUE; + } + + GST_LOG_OBJECT (aggpad, "Not ready to be aggregated"); + + return FALSE; +} + +/** + * gst_aggregator_set_src_caps: + * @self: The #GstAggregator + * @caps: The #GstCaps to set later on the src pad. + * + * Sets the caps to be used on the src pad. + */ +void +gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps) +{ + gst_caps_replace (&self->priv->srccaps, caps); +} + +static void +_reset_flow_values (GstAggregator * self) +{ + self->priv->send_stream_start = TRUE; + self->priv->send_segment = TRUE; + gst_segment_init (&self->segment, GST_FORMAT_TIME); +} + +static inline void +_push_mandatory_events (GstAggregator * self) +{ + GstAggregatorPrivate *priv = self->priv; + + if (g_atomic_int_get (&self->priv->send_stream_start)) { + gchar s_id[32]; + + 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))) { + GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed"); + } + g_atomic_int_set (&self->priv->send_stream_start, FALSE); + } + + if (self->priv->srccaps) { + + GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT, + self->priv->srccaps); + if (!gst_pad_push_event (self->srcpad, + gst_event_new_caps (self->priv->srccaps))) { + GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed"); + } + self->priv->srccaps = NULL; + } + + if (g_atomic_int_get (&self->priv->send_segment)) { + if (!g_atomic_int_get (&self->priv->flush_seeking)) { + GST_INFO_OBJECT (self, "pushing segment"); + gst_pad_push_event (self->srcpad, gst_event_new_segment (&self->segment)); + g_atomic_int_set (&self->priv->send_segment, FALSE); + } + } + + if (priv->tags && priv->tags_changed) { + gst_pad_push_event (self->srcpad, + gst_event_new_tag (gst_tag_list_ref (priv->tags))); + priv->tags_changed = FALSE; + } +} + +/** + * gst_aggregator_finish_buffer: + * @self: The #GstAggregator + * @buffer: the #GstBuffer to push. + * + * This method will take care of sending mandatory events before pushing + * the provided buffer. + */ +GstFlowReturn +gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer) +{ + _push_mandatory_events (self); + + if (!g_atomic_int_get (&self->priv->flush_seeking) && + gst_pad_is_active (self->srcpad)) { + GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer); + return gst_pad_push (self->srcpad, buffer); + } else { + GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)", + g_atomic_int_get (&self->priv->flush_seeking), + gst_pad_is_active (self->srcpad)); + return GST_FLOW_OK; + } +} + +static void +_push_eos (GstAggregator * self) +{ + _push_mandatory_events (self); + + self->priv->send_eos = FALSE; + gst_pad_push_event (self->srcpad, gst_event_new_eos ()); +} + +static void +_remove_all_sources (GstAggregator * self) +{ + GSource *source; + + MAIN_CONTEXT_LOCK (self); + while ((source = + g_main_context_find_source_by_user_data (self->priv->mcontext, + self))) { + g_source_destroy (source); + } + MAIN_CONTEXT_UNLOCK (self); +} + +static gboolean +aggregate_func (GstAggregator * self) +{ + GstAggregatorPrivate *priv = self->priv; + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self); + + GST_LOG_OBJECT (self, "Checking aggregate"); + while (priv->send_eos && gst_aggregator_iterate_sinkpads (self, + (GstAggregatorPadForeachFunc) _check_all_pads_with_data_or_eos, + NULL) && priv->running) { + GST_TRACE_OBJECT (self, "Actually aggregating!"); + + priv->flow_return = klass->aggregate (self); + + if (priv->flow_return == GST_FLOW_EOS) { + g_main_context_wakeup (self->priv->mcontext); + _remove_all_sources (self); + _push_eos (self); + } + + if (priv->flow_return == GST_FLOW_FLUSHING && + g_atomic_int_get (&priv->flush_seeking)) + priv->flow_return = GST_FLOW_OK; + + GST_LOG_OBJECT (self, "flow return is %s", + gst_flow_get_name (priv->flow_return)); + + if (priv->flow_return != GST_FLOW_OK) + break; + } + + return G_SOURCE_REMOVE; +} + +static void +iterate_main_context_func (GstAggregator * self) +{ + if (self->priv->running == FALSE) { + GST_DEBUG_OBJECT (self, "Not running anymore"); + + return; + } + + g_main_context_iteration (self->priv->mcontext, TRUE); +} + +static gboolean +_start (GstAggregator * self) +{ + self->priv->running = TRUE; + self->priv->send_stream_start = TRUE; + self->priv->send_segment = TRUE; + self->priv->send_eos = TRUE; + self->priv->srccaps = NULL; + + return TRUE; +} + +static gboolean +_check_pending_flush_stop (GstAggregatorPad * pad) +{ + return (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start); +} + +static gboolean +_stop_srcpad_task (GstAggregator * self, GstEvent * flush_start) +{ + gboolean res = TRUE; + + GST_INFO_OBJECT (self, "%s srcpad task", + flush_start ? "Pausing" : "Stopping"); + + self->priv->running = FALSE; + + /* Clean the stack of GSource set on the MainContext */ + g_main_context_wakeup (self->priv->mcontext); + _remove_all_sources (self); + if (flush_start) { + res = gst_pad_push_event (self->srcpad, flush_start); + } + + gst_pad_stop_task (self->srcpad); + + return res; +} + +static void +_start_srcpad_task (GstAggregator * self) +{ + GST_INFO_OBJECT (self, "Starting srcpad task"); + + self->priv->running = TRUE; + gst_pad_start_task (GST_PAD (self->srcpad), + (GstTaskFunction) iterate_main_context_func, self, NULL); +} + +static inline void +_add_aggregate_gsource (GstAggregator * self) +{ + MAIN_CONTEXT_LOCK (self); + g_main_context_invoke (self->priv->mcontext, (GSourceFunc) aggregate_func, + self); + MAIN_CONTEXT_UNLOCK (self); +} + +static GstFlowReturn +_flush (GstAggregator * self) +{ + GstFlowReturn ret = GST_FLOW_OK; + GstAggregatorPrivate *priv = self->priv; + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self); + + GST_DEBUG_OBJECT (self, "Flushing everything"); + g_atomic_int_set (&priv->send_segment, TRUE); + g_atomic_int_set (&priv->flush_seeking, FALSE); + g_atomic_int_set (&priv->tags_changed, FALSE); + if (klass->flush) + ret = klass->flush (self); + + return ret; +} + +static gboolean +_all_flush_stop_received (GstAggregator * self) +{ + GList *tmp; + GstAggregatorPad *tmppad; + + GST_OBJECT_LOCK (self); + for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) { + tmppad = (GstAggregatorPad *) tmp->data; + + if (_check_pending_flush_stop (tmppad) == FALSE) { + GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i", + tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop); + GST_OBJECT_UNLOCK (self); + return FALSE; + } + } + GST_OBJECT_UNLOCK (self); + + return TRUE; +} + +/* GstAggregator vmethods default implementations */ +static gboolean +_sink_event (GstAggregator * self, GstAggregatorPad * aggpad, GstEvent * event) +{ + gboolean res = TRUE; + GstPad *pad = GST_PAD (aggpad); + GstAggregatorPrivate *priv = self->priv; + GstAggregatorPadPrivate *padpriv = aggpad->priv; + + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_FLUSH_START: + { + GstBuffer *tmpbuf; + + g_atomic_int_set (&aggpad->priv->flushing, TRUE); + /* Remove pad buffer and wake up the streaming thread */ + tmpbuf = gst_aggregator_pad_steal_buffer (aggpad); + gst_buffer_replace (&tmpbuf, NULL); + if (g_atomic_int_compare_and_exchange (&padpriv->pending_flush_start, + TRUE, FALSE) == TRUE) { + GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now"); + g_atomic_int_set (&padpriv->pending_flush_stop, TRUE); + } + + if (g_atomic_int_get (&priv->flush_seeking)) { + /* If flush_seeking we forward the first FLUSH_START */ + if (g_atomic_int_compare_and_exchange (&priv->pending_flush_start, + TRUE, FALSE) == TRUE) { + + GST_DEBUG_OBJECT (self, "Flushing, pausing srcpad task"); + priv->flow_return = GST_FLOW_OK; + _stop_srcpad_task (self, event); + + GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking"); + GST_PAD_STREAM_LOCK (self->srcpad); + GST_LOG_OBJECT (self, "GOT STREAM_LOCK"); + event = NULL; + goto eat; + } + } + + /* We forward only in one case: right after flush_seeking */ + goto eat; + } + case GST_EVENT_FLUSH_STOP: + { + GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP"); + + _aggpad_flush (aggpad, self); + if (g_atomic_int_get (&priv->flush_seeking)) { + g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE); + + if (g_atomic_int_get (&priv->flush_seeking)) { + if (_all_flush_stop_received (self)) { + /* That means we received FLUSH_STOP/FLUSH_STOP on + * all sinkpads -- Seeking is Done... sending FLUSH_STOP */ + _flush (self); + gst_pad_push_event (self->srcpad, event); + priv->send_eos = TRUE; + event = NULL; + _add_aggregate_gsource (self); + + GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK"); + GST_PAD_STREAM_UNLOCK (self->srcpad); + _start_srcpad_task (self); + } + } + } + + /* We never forward the event */ + goto eat; + } + case GST_EVENT_EOS: + { + GST_DEBUG_OBJECT (aggpad, "EOS"); + + /* We still have a buffer, and we don't want the subclass to have to + * check for it. Mark pending_eos, eos will be set when steal_buffer is + * called + */ + PAD_LOCK_EVENT (aggpad); + if (!aggpad->buffer) { + aggpad->eos = TRUE; + } else { + aggpad->priv->pending_eos = TRUE; + } + PAD_UNLOCK_EVENT (aggpad); + + _add_aggregate_gsource (self); + goto eat; + } + case GST_EVENT_SEGMENT: + { + gst_event_copy_segment (event, &aggpad->segment); + PAD_UNLOCK_EVENT (aggpad); + + goto eat; + } + case GST_EVENT_STREAM_START: + { + 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; + } + default: + { + break; + } + } + + GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event); + return gst_pad_event_default (pad, GST_OBJECT (self), event); + +eat: + GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event); + if (event) + gst_event_unref (event); + + return res; +} + +static gboolean +_stop (GstAggregator * agg) +{ + _reset_flow_values (agg); + + return TRUE; +} + +/* GstElement vmethods implementations */ +static GstStateChangeReturn +_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn ret; + GstAggregator *self = GST_AGGREGATOR (element); + GstAggregatorClass *agg_class = GST_AGGREGATOR_GET_CLASS (self); + + + switch (transition) { + case GST_STATE_CHANGE_READY_TO_PAUSED: + agg_class->start (self); + break; + default: + break; + } + + if ((ret = + GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element, + transition)) == GST_STATE_CHANGE_FAILURE) + goto failure; + + + switch (transition) { + case GST_STATE_CHANGE_PAUSED_TO_READY: + agg_class->stop (self); + break; + default: + break; + } + + return ret; + +failure: + { + GST_ERROR_OBJECT (element, "parent failed state change"); + return ret; + } +} + +static void +_release_pad (GstElement * element, GstPad * pad) +{ + GstBuffer *tmpbuf; + + GstAggregator *self = GST_AGGREGATOR (element); + GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad); + + GST_INFO_OBJECT (pad, "Removing pad"); + + g_atomic_int_set (&aggpad->priv->flushing, TRUE); + tmpbuf = gst_aggregator_pad_steal_buffer (aggpad); + gst_buffer_replace (&tmpbuf, NULL); + gst_element_remove_pad (element, pad); + + /* Something changed make sure we try to aggregate */ + _add_aggregate_gsource (self); +} + +static GstPad * +_request_new_pad (GstElement * element, + 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); + + if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) { + guint serial = 0; + gchar *name = NULL; + + GST_OBJECT_LOCK (element); + 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 + 1; + } + + 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); + GST_OBJECT_UNLOCK (element); + + } else { + return NULL; + } + + GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad)); + + if (priv->running) + gst_pad_set_active (GST_PAD (agg_pad), TRUE); + + /* add the pad to the element */ + gst_element_add_pad (element, GST_PAD (agg_pad)); + + return GST_PAD (agg_pad); +} + +static gboolean +_src_query (GstAggregator * self, GstQuery * query) +{ + gboolean res = TRUE; + + switch (GST_QUERY_TYPE (query)) { + case GST_QUERY_SEEKING: + { + GstFormat format; + + /* don't pass it along as some (file)sink might claim it does + * whereas with a collectpads in between that will not likely work */ + gst_query_parse_seeking (query, &format, NULL, NULL, NULL); + gst_query_set_seeking (query, format, FALSE, 0, -1); + res = TRUE; + + goto discard; + } + default: + break; + } + + return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query); + +discard: + return res; +} + +static gboolean +event_forward_func (GstPad * pad, EventData * evdata) +{ + gboolean ret = TRUE; + GstPad *peer = gst_pad_get_peer (pad); + GstAggregatorPadPrivate *padpriv = GST_AGGREGATOR_PAD (pad)->priv; + + 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); + } + + evdata->result &= ret; + + if (ret == FALSE) { + if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) + GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event); + else + GST_INFO_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event); + + if (evdata->flush) { + padpriv->pending_flush_start = FALSE; + padpriv->pending_flush_stop = FALSE; + } + } + + /* Always send to all pads */ + return FALSE; +} + +static gboolean +_set_flush_pending (GstAggregator * self, GstAggregatorPad * pad, + gpointer udata) +{ + pad->priv->pending_flush_start = TRUE; + pad->priv->pending_flush_stop = FALSE; + + return TRUE; +} + +static gboolean +_forward_event_to_all_sinkpads (GstAggregator * self, GstEvent * event, + gboolean flush) +{ + EventData evdata; + + evdata.event = event; + evdata.result = TRUE; + evdata.flush = flush; + + /* We first need to set all pads as flushing in a first pass + * as flush_start flush_stop is sometimes sent synchronously + * while we send the seek event */ + if (flush) + gst_aggregator_iterate_sinkpads (self, + (GstAggregatorPadForeachFunc) _set_flush_pending, NULL); + gst_pad_forward (self->srcpad, (GstPadForwardFunction) event_forward_func, + &evdata); + + gst_event_unref (event); + + return evdata.result; +} + +static gboolean +_do_seek (GstAggregator * self, GstEvent * event) +{ + gdouble rate; + GstFormat fmt; + GstSeekFlags flags; + GstSeekType start_type, stop_type; + gint64 start, stop; + gboolean flush; + gboolean res; + GstAggregatorPrivate *priv = self->priv; + + gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type, + &start, &stop_type, &stop); + + GST_INFO_OBJECT (self, "starting SEEK"); + + flush = flags & GST_SEEK_FLAG_FLUSH; + + if (flush) { + g_atomic_int_set (&priv->pending_flush_start, TRUE); + g_atomic_int_set (&priv->flush_seeking, TRUE); + } + + gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start, + stop_type, stop, NULL); + + /* forward the seek upstream */ + res = _forward_event_to_all_sinkpads (self, event, flush); + event = NULL; + + if (!res) { + g_atomic_int_set (&priv->flush_seeking, FALSE); + g_atomic_int_set (&priv->pending_flush_start, FALSE); + } + + GST_INFO_OBJECT (self, "seek done, result: %d", res); + + return res; +} + +static gboolean +_src_event (GstAggregator * self, GstEvent * event) +{ + gboolean res = TRUE; + + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_SEEK: + { + res = _do_seek (self, event); + event = NULL; + goto done; + } + case GST_EVENT_NAVIGATION: + { + /* navigation is rather pointless. */ + res = FALSE; + gst_event_unref (event); + goto done; + } + default: + { + break; + } + } + + return _forward_event_to_all_sinkpads (self, event, FALSE); + +done: + return res; +} + +static gboolean +src_event_func (GstPad * pad, GstObject * parent, GstEvent * event) +{ + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent); + + return klass->src_event (GST_AGGREGATOR (parent), event); +} + +static gboolean +src_query_func (GstPad * pad, GstObject * parent, GstQuery * query) +{ + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent); + + return klass->src_query (GST_AGGREGATOR (parent), query); +} + +static gboolean +src_activate_mode (GstPad * pad, + GstObject * parent, GstPadMode mode, gboolean active) +{ + GstAggregator *self = GST_AGGREGATOR (parent); + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent); + + if (klass->src_activate) { + if (klass->src_activate (self, mode, active) == FALSE) { + return FALSE; + } + } + + if (active == TRUE) { + switch (mode) { + case GST_PAD_MODE_PUSH: + { + GST_INFO_OBJECT (pad, "Activating pad!"); + _start_srcpad_task (self); + return TRUE; + } + default: + { + GST_ERROR_OBJECT (pad, "Only supported mode is PUSH"); + return FALSE; + } + } + } + + /* deactivating */ + GST_INFO_OBJECT (self, "Deactivating srcpad"); + _stop_srcpad_task (self, FALSE); + + return TRUE; +} + +static gboolean +_sink_query (GstAggregator * self, GstAggregatorPad * aggpad, GstQuery * query) +{ + GstPad *pad = GST_PAD (aggpad); + + return gst_pad_query_default (pad, GST_OBJECT (self), query); +} + +/* GObject vmethods implementations */ +static void +gst_aggregator_class_init (GstAggregatorClass * klass) +{ + 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"); + + klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD; + klass->start = _start; + klass->stop = _stop; + + klass->sink_event = _sink_event; + klass->sink_query = _sink_query; + + klass->src_event = _src_event; + klass->src_query = _src_query; + + gstelement_class->request_new_pad = GST_DEBUG_FUNCPTR (_request_new_pad); + gstelement_class->release_pad = GST_DEBUG_FUNCPTR (_release_pad); + gstelement_class->change_state = GST_DEBUG_FUNCPTR (_change_state); +} + +static void +gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass) +{ + GstPadTemplate *pad_template; + GstAggregatorPrivate *priv; + + g_return_if_fail (klass->aggregate != NULL); + + self->priv = + G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR, + GstAggregatorPrivate); + + priv = self->priv; + + pad_template = + gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src"); + g_return_if_fail (pad_template != NULL); + + priv->padcount = -1; + priv->tags_changed = FALSE; + _reset_flow_values (self); + + priv->mcontext = g_main_context_new (); + self->srcpad = gst_pad_new_from_template (pad_template, "src"); + + gst_pad_set_event_function (self->srcpad, + GST_DEBUG_FUNCPTR ((GstPadEventFunction) src_event_func)); + gst_pad_set_query_function (self->srcpad, + GST_DEBUG_FUNCPTR ((GstPadQueryFunction) src_query_func)); + gst_pad_set_activatemode_function (self->srcpad, + GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) src_activate_mode)); + + gst_element_add_pad (GST_ELEMENT (self), self->srcpad); +} + +/* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init + * method to get to the padtemplates */ +GType +gst_aggregator_get_type (void) +{ + static volatile gsize type = 0; + + if (g_once_init_enter (&type)) { + GType _type; + static const GTypeInfo info = { + sizeof (GstAggregatorClass), + NULL, + NULL, + (GClassInitFunc) gst_aggregator_class_init, + NULL, + NULL, + sizeof (GstAggregator), + 0, + (GInstanceInitFunc) gst_aggregator_init, + }; + + _type = g_type_register_static (GST_TYPE_ELEMENT, + "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT); + g_once_init_leave (&type, _type); + } + return type; +} + +static GstFlowReturn +_chain (GstPad * pad, GstObject * object, GstBuffer * buffer) +{ + GstBuffer *actual_buf = buffer; + GstAggregator *self = GST_AGGREGATOR (object); + GstAggregatorPrivate *priv = self->priv; + GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad); + GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object); + + GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer); + + if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE) + goto flushing; + + if (g_atomic_int_get (&aggpad->priv->pending_eos) == TRUE) + goto eos; + + PAD_LOCK_EVENT (aggpad); + if (aggpad->buffer) { + GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed"); + PAD_WAIT_EVENT (aggpad); + } + PAD_UNLOCK_EVENT (aggpad); + + if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE) + goto flushing; + + + if (aggclass->clip) { + aggclass->clip (self, aggpad, buffer, &actual_buf); + } + + PAD_LOCK_EVENT (aggpad); + if (aggpad->buffer) + gst_buffer_unref (aggpad->buffer); + aggpad->buffer = actual_buf; + PAD_UNLOCK_EVENT (aggpad); + + _add_aggregate_gsource (self); + + GST_DEBUG_OBJECT (aggpad, "Done chaining"); + + return priv->flow_return; + +flushing: + + GST_DEBUG_OBJECT (aggpad, "We are flushing"); + + return GST_FLOW_FLUSHING; + +eos: + + GST_DEBUG_OBJECT (pad, "We are EOS already..."); + + return GST_FLOW_EOS; +} + +static gboolean +pad_query_func (GstPad * pad, GstObject * parent, GstQuery * query) +{ + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent); + + return klass->sink_query (GST_AGGREGATOR (parent), + GST_AGGREGATOR_PAD (pad), query); +} + +static gboolean +pad_event_func (GstPad * pad, GstObject * parent, GstEvent * event) +{ + GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent); + + return klass->sink_event (GST_AGGREGATOR (parent), + GST_AGGREGATOR_PAD (pad), event); +} + +static gboolean +pad_activate_mode_func (GstPad * pad, + GstObject * parent, GstPadMode mode, gboolean active) +{ + GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad); + + if (active == FALSE) { + PAD_LOCK_EVENT (aggpad); + g_atomic_int_set (&aggpad->priv->flushing, TRUE); + gst_buffer_replace (&aggpad->buffer, NULL); + PAD_BROADCAST_EVENT (aggpad); + PAD_UNLOCK_EVENT (aggpad); + } else { + g_atomic_int_set (&aggpad->priv->flushing, FALSE); + PAD_LOCK_EVENT (aggpad); + PAD_BROADCAST_EVENT (aggpad); + PAD_UNLOCK_EVENT (aggpad); + } + + return TRUE; +} + +/*********************************** + * GstAggregatorPad implementation * + ************************************/ +G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD); + +static void +_pad_constructed (GObject * object) +{ + GstPad *pad = GST_PAD (object); + + gst_pad_set_chain_function (pad, + GST_DEBUG_FUNCPTR ((GstPadChainFunction) _chain)); + gst_pad_set_event_function (pad, + GST_DEBUG_FUNCPTR ((GstPadEventFunction) pad_event_func)); + gst_pad_set_query_function (pad, + GST_DEBUG_FUNCPTR ((GstPadQueryFunction) pad_query_func)); + gst_pad_set_activatemode_function (pad, + GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) pad_activate_mode_func)); +} + +static void +gst_aggregator_pad_class_init (GstAggregatorPadClass * klass) +{ + GObjectClass *gobject_class = (GObjectClass *) klass; + + g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate)); + + gobject_class->constructed = GST_DEBUG_FUNCPTR (_pad_constructed); +} + +static void +gst_aggregator_pad_init (GstAggregatorPad * pad) +{ + pad->priv = + G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD, + GstAggregatorPadPrivate); + + pad->buffer = NULL; + g_mutex_init (&pad->priv->event_lock); + g_cond_init (&pad->priv->event_cond); + +} + +/** + * gst_aggregator_pad_steal_buffer: + * @pad: the pad to get buffer from + * + * Steal the ref to the buffer currently queued in @pad. + * + * Returns: (transfer full): The buffer in @pad or NULL if no buffer was + * queued. You should unref the buffer after usage. + */ +GstBuffer * +gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad) +{ + GstBuffer *buffer = NULL; + + PAD_LOCK_EVENT (pad); + if (pad->buffer) { + GST_TRACE_OBJECT (pad, "Consuming buffer"); + buffer = pad->buffer; + pad->buffer = NULL; + if (pad->priv->pending_eos) { + pad->priv->pending_eos = FALSE; + pad->eos = TRUE; + } + PAD_BROADCAST_EVENT (pad); + GST_DEBUG_OBJECT (pad, "Consummed: %" GST_PTR_FORMAT, buffer); + } + PAD_UNLOCK_EVENT (pad); + + return buffer; +} + +/** + * gst_aggregator_pad_get_buffer: + * @pad: the pad to get buffer from + * + * Returns: (transfer full): A reference to the buffer in @pad or + * NULL if no buffer was queued. You should unref the buffer after + * usage. + */ +GstBuffer * +gst_aggregator_pad_get_buffer (GstAggregatorPad * pad) +{ + GstBuffer *buffer = NULL; + + PAD_LOCK_EVENT (pad); + if (pad->buffer) + buffer = gst_buffer_ref (pad->buffer); + PAD_UNLOCK_EVENT (pad); + + return buffer; +} + +/** + * gst_aggregator_merge_tags: + * @self: a #GstAggregator + * @tags: a #GstTagList to merge + * @mode: the #GstTagMergeMode to use + * + * Adds tags to so-called pending tags, which will be processed + * before pushing out data downstream. + * + * Note that this is provided for convenience, and the subclass is + * not required to use this and can still do tag handling on its own. + * + * MT safe. + */ +void +gst_aggregator_merge_tags (GstAggregator * self, + const GstTagList * tags, GstTagMergeMode mode) +{ + GstTagList *otags; + + g_return_if_fail (GST_IS_AGGREGATOR (self)); + g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags)); + + /* FIXME Check if we can use OBJECT lock here! */ + GST_OBJECT_LOCK (self); + if (tags) + GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags); + otags = self->priv->tags; + self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode); + if (otags) + gst_tag_list_unref (otags); + self->priv->tags_changed = TRUE; + GST_OBJECT_UNLOCK (self); +} diff --git a/gst-libs/gst/base/gstaggregator.h b/gst-libs/gst/base/gstaggregator.h new file mode 100644 index 0000000..3965ccc --- /dev/null +++ b/gst-libs/gst/base/gstaggregator.h @@ -0,0 +1,256 @@ +/* GStreamer + * Copyright (C) 2014 Mathieu Duponchelle + * Copyright (C) 2014 Thibault Saunier + * + * gstaggregator.c: + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, + * Boston, MA 02110-1301, USA. + */ + +#ifndef __GST_AGGREGATOR_H__ +#define __GST_AGGREGATOR_H__ + +#include + +G_BEGIN_DECLS + +/************************** + * GstAggregator Structs * + *************************/ + +typedef struct _GstAggregator GstAggregator; +typedef struct _GstAggregatorPrivate GstAggregatorPrivate; +typedef struct _GstAggregatorClass GstAggregatorClass; + +/************************ + * GstAggregatorPad API * + ***********************/ + +#define GST_TYPE_AGGREGATOR_PAD (gst_aggregator_pad_get_type()) +#define GST_AGGREGATOR_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPad)) +#define GST_AGGREGATOR_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPadClass)) +#define GST_AGGREGATOR_PAD_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS ((obj),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPadClass)) +#define GST_IS_AGGREGATOR_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AGGREGATOR_PAD)) +#define GST_IS_AGGREGATOR_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AGGREGATOR_PAD)) + +/**************************** + * GstAggregatorPad Structs * + ***************************/ + +typedef struct _GstAggregatorPad GstAggregatorPad; +typedef struct _GstAggregatorPadClass GstAggregatorPadClass; +typedef struct _GstAggregatorPadPrivate GstAggregatorPadPrivate; + +/** + * GstAggregatorPad: + * @buffer: currently queued buffer. + * @segment: last segment received. + * + * The implementation the GstPad to use with #GstAggregator + */ +struct _GstAggregatorPad +{ + GstPad parent; + + GstBuffer * buffer; + GstSegment segment; + gboolean eos; + + /* < Private > */ + GstAggregatorPadPrivate * priv; + + gpointer _gst_reserved[GST_PADDING]; +}; + +/** + * GstAggregatorPadClass: + * @flush: Optional + * Called when the pad has received a flush stop, this is the place + * to flush any information specific to the pad, it allows for individual + * pads to be flushed while others might not be. + * + */ +struct _GstAggregatorPadClass +{ + GstPadClass parent_class; + + GstFlowReturn (*flush) (GstAggregatorPad * aggpad, GstAggregator * aggregator); + + /*< private >*/ + gpointer _gst_reserved[GST_PADDING]; +}; + +GType gst_aggregator_pad_get_type (void); + +/**************************** + * GstAggregatorPad methods * + ***************************/ + +GstBuffer * gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad); +GstBuffer * gst_aggregator_pad_get_buffer (GstAggregatorPad * pad); + +/********************* + * GstAggregator API * + ********************/ + +#define GST_TYPE_AGGREGATOR (gst_aggregator_get_type()) +#define GST_AGGREGATOR(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AGGREGATOR,GstAggregator)) +#define GST_AGGREGATOR_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_AGGREGATOR,GstAggregatorClass)) +#define GST_AGGREGATOR_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS ((obj),GST_TYPE_AGGREGATOR,GstAggregatorClass)) +#define GST_IS_AGGREGATOR(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AGGREGATOR)) +#define GST_IS_AGGREGATOR_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AGGREGATOR)) + +#define GST_FLOW_CUSTOM_SUCCESS GST_FLOW_NOT_HANDLED + +/** + * GstAggregator: + * @aggregator_pads: #GList of #GstAggregatorPad managed by this #GstAggregator. + * + * Collectpads object. + */ +struct _GstAggregator +{ + GstElement parent; + + GstPad * srcpad; + + GstSegment segment; + + /*< private >*/ + GstAggregatorPrivate * priv; + + gpointer _gst_reserved[GST_PADDING]; +}; + +/** + * GstAggregatorClass: + * @sinkpads_type: Optional. + * The type of the pads that should be created when + * GstElement.request_new_pad is called. + * @flush: Optional. + * Called after a succesful flushing seek, once all the flush + * stops have been received. Flush pad-specific data in + * #GstAggregatorPad->flush. + * @clip: Optional. + * Called when a buffer is received on a sink pad, the task + * of clipping it and translating it to the current segment + * falls on the subclass. + * @sink_event: Optional. + * Called when an event is received on a sink pad, the subclass + * should always chain up. + * @sink_query: Optional. + * Called when a query is received on a sink pad, the subclass + * should always chain up. + * @src_event: Optional. + * Called when an event is received on the src pad, the subclass + * should always chain up. + * @src_query: Optional. + * Called when a query is received on the src pad, the subclass + * should always chain up. + * @src_activate: Optional. + * Called when the src pad is activated, it will start/stop its + * pad task right after that call. + * @aggregate: Mandatory. + * Called when buffers are queued on all sinkpads. Classes + * should iterate the GstElement->sinkpads and peek or steal + * buffers from the #GstAggregatorPads. If the subclass returns + * GST_FLOW_EOS, sending of the eos event will be taken care + * of. Once / if a buffer has been constructed from the + * aggregated buffers, the subclass should call _finish_buffer. + * @stop: Optional. + * Should be linked up first. Called when the + * element goes from PAUSED to READY. The subclass should free + * all resources and reset its state. + * @start: Optional. + * Should be linked up first. Called when the element goes from + * READY to PAUSED. The subclass should get ready to process + * aggregated buffers. + * + * The aggregator base class will handle in a thread-safe way all manners of + * concurrent flushes, seeks, pad additions and removals, leaving to the + * subclass the responsibility of clipping buffers, and aggregating buffers in + * the way the implementor sees fit. + * + * It will also take care of event ordering (stream-start, segment, eos). + * + * Basically, a basic implementation will override @aggregate, and call + * _finish_buffer from inside that function. + */ +struct _GstAggregatorClass { + GstElementClass parent_class; + + GType sinkpads_type; + + GstFlowReturn (*flush) (GstAggregator * aggregator); + + GstFlowReturn (*clip) (GstAggregator * agg, + GstAggregatorPad * bpad, + GstBuffer * buf, + GstBuffer ** outbuf); + + /* sinkpads virtual methods */ + gboolean (*sink_event) (GstAggregator * aggregate, + GstAggregatorPad * bpad, + GstEvent * event); + + gboolean (*sink_query) (GstAggregator * aggregate, + GstAggregatorPad * bpad, + GstQuery * query); + + /* srcpad virtual methods */ + gboolean (*src_event) (GstAggregator * aggregate, + GstEvent * event); + + gboolean (*src_query) (GstAggregator * aggregate, + GstQuery * query); + + gboolean (*src_activate) (GstAggregator * aggregator, + GstPadMode mode, + gboolean active); + + GstFlowReturn (*aggregate) (GstAggregator * aggregator); + + gboolean (*stop) (GstAggregator * aggregator); + + gboolean (*start) (GstAggregator * aggregator); + + /*< private >*/ + gpointer _gst_reserved[GST_PADDING]; +}; + +/************************* + * GstAggregator methods * + ************************/ + +GstFlowReturn gst_aggregator_finish_buffer (GstAggregator * agg, + GstBuffer * buffer); +void gst_aggregator_set_src_caps (GstAggregator * agg, + GstCaps * caps); + +GType gst_aggregator_get_type(void); + +/* API that should eventually land in GstElement itself*/ +typedef gboolean (*GstAggregatorPadForeachFunc) (GstAggregator * self, + GstPad * pad, + gpointer user_data); +gboolean gst_aggregator_iterate_sinkpads (GstAggregator * self, + GstAggregatorPadForeachFunc func, + gpointer user_data); + + +G_END_DECLS + +#endif /* __GST_AGGREGATOR_H__ */ diff --git a/pkgconfig/gstreamer-bad-base-uninstalled.pc.in b/pkgconfig/gstreamer-bad-base-uninstalled.pc.in new file mode 100644 index 0000000..560eb13 --- /dev/null +++ b/pkgconfig/gstreamer-bad-base-uninstalled.pc.in @@ -0,0 +1,15 @@ +prefix= +exec_prefix= +libdir=${pcfiledir}/../gst-libs/gst/base +includedir=${pcfiledir}/../gst-libs +pluginsdir=${pcfiledir}/.. + +Name: GStreamer bad base classes, uninstalled +Description: Bad base classes for GStreamer elements, Not Installed +Version: @VERSION@ +Requires: gstreamer-@GST_API_VERSION@ + +Libs: -L${libdir} ${libdir}/libgstbadbase-@GST_API_VERSION@.la @BADBASE_LIBS@ +Cflags: -I${includedir} -I@srcdir@/.. @BADBASE_CFLAGS@ + +libraries=badbase diff --git a/pkgconfig/gstreamer-bad-base.pc.in b/pkgconfig/gstreamer-bad-base.pc.in new file mode 100644 index 0000000..a03013b --- /dev/null +++ b/pkgconfig/gstreamer-bad-base.pc.in @@ -0,0 +1,15 @@ +prefix=@prefix@ +exec_prefix=@exec_prefix@ +libdir=@libdir@ +includedir=@includedir@/gstreamer-@GST_API_VERSION@ +pluginsdir=@libdir@/gstreamer-@GST_API_VERSION@ + +Name: GStreamer bad base classes, uninstalled +Description: Bad base classes for GStreamer elements, Not Installed +Version: @VERSION@ +Requires: gstreamer-@GST_API_VERSION@ + +Libs: -L${libdir} ${libdir}/libgstbadbase-@GST_API_VERSION@.la @BADBASE_LIBS@ +Cflags: -I${includedir} -I@srcdir@/.. @BADBASE_CFLAGS@ + +libraries=badbase diff --git a/tests/check/Makefile.am b/tests/check/Makefile.am index e7b77ed..6233d013 100644 --- a/tests/check/Makefile.am +++ b/tests/check/Makefile.am @@ -244,6 +244,7 @@ check_PROGRAMS = \ libs/mpegts \ libs/h264parser \ libs/vp8parser \ + libs/aggregator \ $(check_uvch264) \ libs/vc1parser \ $(check_schro) \ @@ -455,6 +456,16 @@ libs_gstglupload_LDADD = \ $(GST_PLUGINS_BASE_LIBS) -lgstvideo-$(GST_API_VERSION) \ $(GST_BASE_LIBS) $(GST_LIBS) $(LDADD) +libs_aggregator_LDADD = \ + $(top_builddir)/gst-libs/gst/base/libgstbadbase-@GST_API_VERSION@.la \ + $(GST_PLUGINS_BASE_LIBS) \ + $(GST_BASE_LIBS) $(GST_LIBS) $(LDADD) + +libs_aggregator_CFLAGS = \ + $(GST_PLUGINS_BAD_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS) \ + -DGST_USE_UNSTABLE_API \ + $(GST_BASE_CFLAGS) $(GST_CFLAGS) $(AM_CFLAGS) + distclean-local-orc: rm -rf orc diff --git a/tests/check/libs/aggregator.c b/tests/check/libs/aggregator.c new file mode 100644 index 0000000..c0b0bdb --- /dev/null +++ b/tests/check/libs/aggregator.c @@ -0,0 +1,1017 @@ +/* + * aggregator.c - GstAggregator testsuite + * Copyright (C) 2006 Alessandro Decina + * Copyright (C) 2014 Mathieu Duponchelle + * Copyright (C) 2014 Thibault Saunier + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, + * Boston, MA 02110-1301, USA. + */ + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include +#include + +/* dummy aggregator based element */ + +#define GST_TYPE_TEST_AGGREGATOR (gst_test_aggregator_get_type ()) +#define GST_TEST_AGGREGATOR(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_TEST_AGGREGATOR, GstTestAggregator)) +#define GST_TEST_AGGREGATOR_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_TEST_AGGREGATOR, GstTestAggregatorClass)) +#define GST_TEST_AGGREGATOR_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_TEST_AGGREGATOR, GstTestAggregatorClass)) + +#define fail_error_message(msg) \ + G_STMT_START { \ + GError *error; \ + gst_message_parse_error(msg, &error, NULL); \ + fail_unless(FALSE, "Error Message from %s : %s", \ + GST_OBJECT_NAME (GST_MESSAGE_SRC(msg)), error->message); \ + g_error_free (error); \ + } G_STMT_END; + +typedef struct _GstTestAggregator GstTestAggregator; +typedef struct _GstTestAggregatorClass GstTestAggregatorClass; + +static GType gst_test_aggregator_get_type (void); + +#define BUFFER_DURATION 100000000 /* 10 frames per second */ + +struct _GstTestAggregator +{ + GstAggregator parent; + + guint64 timestamp; +}; + +struct _GstTestAggregatorClass +{ + GstAggregatorClass parent_class; +}; + +static GstFlowReturn +gst_test_aggregator_aggregate (GstAggregator * aggregator) +{ + GstIterator *iter; + gboolean all_eos = TRUE; + GstTestAggregator *testagg; + GstBuffer *buf; + + gboolean done_iterating = FALSE; + + testagg = GST_TEST_AGGREGATOR (aggregator); + + iter = gst_element_iterate_sink_pads (GST_ELEMENT (testagg)); + while (!done_iterating) { + GstBuffer *buffer; + GValue value = { 0, }; + GstAggregatorPad *pad; + + switch (gst_iterator_next (iter, &value)) { + case GST_ITERATOR_OK: + pad = g_value_get_object (&value); + + if (pad->eos == FALSE) + all_eos = FALSE; + buffer = gst_aggregator_pad_steal_buffer (pad); + gst_buffer_replace (&buffer, NULL); + + g_value_reset (&value); + break; + case GST_ITERATOR_RESYNC: + gst_iterator_resync (iter); + break; + case GST_ITERATOR_ERROR: + GST_WARNING_OBJECT (testagg, "Sinkpads iteration error"); + done_iterating = TRUE; + break; + case GST_ITERATOR_DONE: + done_iterating = TRUE; + break; + } + } + gst_iterator_free (iter); + + if (all_eos == TRUE) { + GST_ERROR_OBJECT (testagg, "no data available, must be EOS"); + gst_pad_push_event (aggregator->srcpad, gst_event_new_eos ()); + return GST_FLOW_EOS; + } + + buf = gst_buffer_new (); + GST_BUFFER_TIMESTAMP (buf) = testagg->timestamp; + GST_BUFFER_DURATION (buf) = BUFFER_DURATION; + testagg->timestamp += BUFFER_DURATION; + + gst_aggregator_finish_buffer (aggregator, buf); + + /* We just check finish_frame return FLOW_OK */ + return GST_FLOW_OK; +} + +#define gst_test_aggregator_parent_class parent_class +G_DEFINE_TYPE (GstTestAggregator, gst_test_aggregator, GST_TYPE_AGGREGATOR); + +static void +gst_test_aggregator_class_init (GstTestAggregatorClass * klass) +{ + GstElementClass *gstelement_class = (GstElementClass *) klass; + GstAggregatorClass *base_aggregator_class = (GstAggregatorClass *) klass; + + static GstStaticPadTemplate _src_template = + GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC, GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + + static GstStaticPadTemplate _sink_template = + GST_STATIC_PAD_TEMPLATE ("sink_%u", GST_PAD_SINK, GST_PAD_REQUEST, + GST_STATIC_CAPS_ANY); + + gst_element_class_add_pad_template (gstelement_class, + gst_static_pad_template_get (&_src_template)); + + gst_element_class_add_pad_template (gstelement_class, + gst_static_pad_template_get (&_sink_template)); + + gst_element_class_set_static_metadata (gstelement_class, "Aggregator", + "Testing", "Combine N buffers", "Stefan Sauer "); + + base_aggregator_class->aggregate = + GST_DEBUG_FUNCPTR (gst_test_aggregator_aggregate); +} + +static void +gst_test_aggregator_init (GstTestAggregator * self) +{ + GstAggregator *agg = GST_AGGREGATOR (self); + gst_segment_init (&agg->segment, GST_FORMAT_BYTES); + self->timestamp = 0; +} + +static gboolean +gst_test_aggregator_plugin_init (GstPlugin * plugin) +{ + return gst_element_register (plugin, "testaggregator", GST_RANK_NONE, + GST_TYPE_TEST_AGGREGATOR); +} + +static gboolean +gst_test_aggregator_plugin_register (void) +{ + return gst_plugin_register_static (GST_VERSION_MAJOR, + GST_VERSION_MINOR, + "testaggregator", + "Combine buffers", + gst_test_aggregator_plugin_init, + VERSION, GST_LICENSE, PACKAGE, GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN); +} + +typedef struct +{ + GstEvent *event; + GstBuffer *buffer; + GstElement *aggregator; + GstPad *sinkpad, *srcpad; + GstFlowReturn expected_result; + + /* ------------------ + * ----------- --------|-- | + * | srcpad | -- | sinkpad | aggregator | + * ----------- --------|-- | + * ------------------ + * This is for 1 Chain, we can have several + */ +} ChainData; + +typedef struct +{ + GMainLoop *ml; + GstPad *srcpad, /* srcpad of the GstAggregator */ + *sinkpad; /* fake sinkpad to which GstAggregator.srcpad is linked */ + guint timeout_id; + GstElement *aggregator; + + /* -----------------| + * | ---------- ----------- + * | aggregator | srcpad | -- | sinkpad | + * | ---------- ----------- + * -----------------| + */ + + gint flush_start_events, flush_stop_events; +} TestData; + +static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src", + GST_PAD_SRC, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +static gpointer +push_buffer (gpointer user_data) +{ + GstFlowReturn flow; + GstCaps *caps; + ChainData *chain_data = (ChainData *) user_data; + GstSegment segment; + + gst_pad_push_event (chain_data->srcpad, gst_event_new_stream_start ("test")); + + caps = gst_caps_new_empty_simple ("foo/x-bar"); + gst_pad_push_event (chain_data->srcpad, gst_event_new_caps (caps)); + gst_caps_unref (caps); + + gst_segment_init (&segment, GST_FORMAT_TIME); + gst_pad_push_event (chain_data->srcpad, gst_event_new_segment (&segment)); + + GST_DEBUG ("Pushing buffer on pad: %s:%s", + GST_DEBUG_PAD_NAME (chain_data->sinkpad)); + flow = gst_pad_push (chain_data->srcpad, chain_data->buffer); + fail_unless (flow == chain_data->expected_result, + "got flow %s instead of %s on %s:%s", gst_flow_get_name (flow), + gst_flow_get_name (chain_data->expected_result), + GST_DEBUG_PAD_NAME (chain_data->sinkpad)); + chain_data->buffer = NULL; + + return NULL; +} + +static gpointer +push_event (gpointer user_data) +{ + ChainData *chain_data = (ChainData *) user_data; + + GST_ERROR_OBJECT (chain_data->srcpad, "Pushing event: %" + GST_PTR_FORMAT, chain_data->event); + fail_unless (gst_pad_push_event (chain_data->srcpad, + chain_data->event) == TRUE); + + return NULL; +} + +static gboolean +_aggregate_timeout (GMainLoop * ml) +{ + g_main_loop_quit (ml); + + fail_unless ("No buffer found on aggregator.srcpad -> TIMEOUT" == NULL); + + return FALSE; +} + +static gboolean +_quit (GMainLoop * ml) +{ + GST_DEBUG ("QUITING ML"); + g_main_loop_quit (ml); + + return G_SOURCE_REMOVE; +} + +static GstPadProbeReturn +_aggregated_cb (GstPad * pad, GstPadProbeInfo * info, GMainLoop * ml) +{ + GST_DEBUG ("SHould quit ML"); + g_idle_add ((GSourceFunc) _quit, ml); + + return GST_PAD_PROBE_REMOVE; +} + +static GstPadProbeReturn +downstream_probe_cb (GstPad * pad, GstPadProbeInfo * info, TestData * test) +{ + GST_DEBUG ("PROBING "); + if (info->type & GST_PAD_PROBE_TYPE_EVENT_FLUSH) { + if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_EVENT (info)) == + GST_EVENT_FLUSH_START) { + + g_atomic_int_inc (&test->flush_start_events); + GST_DEBUG ("==========> FLUSH: %i", test->flush_start_events); + } else if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_EVENT (info)) == + GST_EVENT_FLUSH_STOP) + g_atomic_int_inc (&test->flush_stop_events); + } + + return GST_PAD_PROBE_DROP; +} + +/* + * Not thread safe, will create a new ChainData which contains + * an activated src pad linked to a requested sink pad of @agg, and + * a newly allocated buffer ready to be pushed. Caller needs to + * clear with _chain_data_clear after. + */ +static void +_chain_data_init (ChainData * data, GstElement * agg) +{ + static gint num_src_pads = 0; + gchar *pad_name = g_strdup_printf ("src%d", num_src_pads); + + num_src_pads += 1; + + data->srcpad = gst_pad_new_from_static_template (&srctemplate, pad_name); + g_free (pad_name); + gst_pad_set_active (data->srcpad, TRUE); + data->aggregator = agg; + data->buffer = gst_buffer_new (); + data->sinkpad = gst_element_get_request_pad (agg, "sink_%u"); + fail_unless (GST_IS_PAD (data->sinkpad)); + fail_unless (gst_pad_link (data->srcpad, data->sinkpad) == GST_PAD_LINK_OK); +} + +static void +_chain_data_clear (ChainData * data) +{ + if (data->buffer) + gst_buffer_unref (data->buffer); + if (data->srcpad) + gst_object_unref (data->srcpad); + if (data->sinkpad) + gst_object_unref (data->sinkpad); +} + +static void +_test_data_init (TestData * test, gboolean needs_flushing) +{ + test->aggregator = gst_element_factory_make ("testaggregator", NULL); + gst_element_set_state (test->aggregator, GST_STATE_PLAYING); + test->ml = g_main_loop_new (NULL, TRUE); + test->srcpad = GST_AGGREGATOR (test->aggregator)->srcpad; + + GST_DEBUG ("Srcpad: %p", test->srcpad); + + if (needs_flushing) { + static gint num_sink_pads = 0; + gchar *pad_name = g_strdup_printf ("sink%d", num_sink_pads); + + num_sink_pads += 1; + test->sinkpad = gst_pad_new_from_static_template (&sinktemplate, pad_name); + g_free (pad_name); + fail_unless (gst_pad_link (test->srcpad, test->sinkpad) == GST_PAD_LINK_OK); + gst_pad_add_probe (test->srcpad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | + GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM | + GST_PAD_PROBE_TYPE_EVENT_FLUSH, + (GstPadProbeCallback) downstream_probe_cb, test, NULL); + } else { + gst_pad_add_probe (test->srcpad, GST_PAD_PROBE_TYPE_BUFFER, + (GstPadProbeCallback) _aggregated_cb, test->ml, NULL); + } + + + test->timeout_id = + g_timeout_add (1000, (GSourceFunc) _aggregate_timeout, test->ml); +} + +static void +_test_data_clear (TestData * test) +{ + gst_element_set_state (test->aggregator, GST_STATE_NULL); + gst_object_unref (test->aggregator); + + g_main_loop_unref (test->ml); +} + +GST_START_TEST (test_aggregate) +{ + GThread *thread1, *thread2; + + ChainData data1 = { 0, }; + ChainData data2 = { 0, }; + TestData test = { 0, }; + + _test_data_init (&test, FALSE); + _chain_data_init (&data1, test.aggregator); + _chain_data_init (&data2, test.aggregator); + + thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL); + thread2 = g_thread_try_new ("gst-check", push_buffer, &data2, NULL); + + g_main_loop_run (test.ml); + g_source_remove (test.timeout_id); + + + /* these will return immediately as when the data is popped the threads are + * unlocked and will terminate */ + g_thread_join (thread1); + g_thread_join (thread2); + + _chain_data_clear (&data1); + _chain_data_clear (&data2); + _test_data_clear (&test); +} + +GST_END_TEST; + +GST_START_TEST (test_aggregate_eos) +{ + GThread *thread1, *thread2; + + ChainData data1 = { 0, }; + ChainData data2 = { 0, }; + TestData test = { 0, }; + + _test_data_init (&test, FALSE); + _chain_data_init (&data1, test.aggregator); + _chain_data_init (&data2, test.aggregator); + + data2.event = gst_event_new_eos (); + + thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL); + thread2 = g_thread_try_new ("gst-check", push_event, &data2, NULL); + + g_main_loop_run (test.ml); + g_source_remove (test.timeout_id); + + /* these will return immediately as when the data is popped the threads are + * unlocked and will terminate */ + g_thread_join (thread1); + g_thread_join (thread2); + + _chain_data_clear (&data1); + _chain_data_clear (&data2); + _test_data_clear (&test); +} + +GST_END_TEST; + +#define NUM_BUFFERS 3 +static void +handoff (GstElement * fakesink, GstBuffer * buf, GstPad * pad, guint * count) +{ + *count = *count + 1; + GST_DEBUG ("HANDOFF: %i", *count); +} + +/* Test a linear pipeline using aggregator */ +GST_START_TEST (test_linear_pipeline) +{ + GstBus *bus; + GstMessage *msg; + GstElement *pipeline, *src, *agg, *sink; + + gint count = 0; + + pipeline = gst_pipeline_new ("pipeline"); + src = gst_check_setup_element ("fakesrc"); + g_object_set (src, "num-buffers", NUM_BUFFERS, "sizetype", 2, "sizemax", 4, + NULL); + agg = gst_check_setup_element ("testaggregator"); + sink = gst_check_setup_element ("fakesink"); + g_object_set (sink, "signal-handoffs", TRUE, NULL); + g_signal_connect (sink, "handoff", (GCallback) handoff, &count); + + fail_unless (gst_bin_add (GST_BIN (pipeline), src)); + fail_unless (gst_bin_add (GST_BIN (pipeline), agg)); + fail_unless (gst_bin_add (GST_BIN (pipeline), sink)); + fail_unless (gst_element_link (src, agg)); + fail_unless (gst_element_link (agg, sink)); + + bus = gst_element_get_bus (pipeline); + fail_if (bus == NULL); + gst_element_set_state (pipeline, GST_STATE_PLAYING); + + msg = gst_bus_poll (bus, GST_MESSAGE_EOS | GST_MESSAGE_ERROR, -1); + fail_if (GST_MESSAGE_TYPE (msg) != GST_MESSAGE_EOS); + gst_message_unref (msg); + + fail_unless_equals_int (count, NUM_BUFFERS); + + gst_element_set_state (pipeline, GST_STATE_NULL); + gst_object_unref (bus); + gst_object_unref (pipeline); +} + +GST_END_TEST; + +GST_START_TEST (test_two_src_pipeline) +{ + GstBus *bus; + GstMessage *msg; + GstElement *pipeline, *src, *src1, *agg, *sink; + + gint count = 0; + + pipeline = gst_pipeline_new ("pipeline"); + src = gst_element_factory_make ("fakesrc", NULL); + g_object_set (src, "num-buffers", NUM_BUFFERS, "sizetype", 2, "sizemax", 4, + NULL); + + src1 = gst_element_factory_make ("fakesrc", NULL); + g_object_set (src1, "num-buffers", NUM_BUFFERS + 1, "sizetype", 2, "sizemax", + 4, NULL); + + agg = gst_check_setup_element ("testaggregator"); + sink = gst_check_setup_element ("fakesink"); + g_object_set (sink, "signal-handoffs", TRUE, NULL); + g_signal_connect (sink, "handoff", (GCallback) handoff, &count); + + fail_unless (gst_bin_add (GST_BIN (pipeline), src)); + fail_unless (gst_bin_add (GST_BIN (pipeline), src1)); + fail_unless (gst_bin_add (GST_BIN (pipeline), agg)); + fail_unless (gst_bin_add (GST_BIN (pipeline), sink)); + fail_unless (gst_element_link (src, agg)); + fail_unless (gst_element_link (src1, agg)); + fail_unless (gst_element_link (agg, sink)); + + bus = gst_element_get_bus (pipeline); + fail_if (bus == NULL); + gst_element_set_state (pipeline, GST_STATE_PLAYING); + + msg = gst_bus_poll (bus, GST_MESSAGE_EOS | GST_MESSAGE_ERROR, -1); + fail_if (GST_MESSAGE_TYPE (msg) != GST_MESSAGE_EOS); + gst_message_unref (msg); + + fail_unless_equals_int (count, NUM_BUFFERS + 1); + + gst_element_set_state (pipeline, GST_STATE_NULL); + gst_object_unref (bus); + gst_object_unref (pipeline); +} + +GST_END_TEST; + +GST_START_TEST (test_flushing_seek) +{ + GstEvent *event; + GThread *thread1, *thread2; + + ChainData data1 = { 0, }; + ChainData data2 = { 0, }; + TestData test = { 0, }; + + _test_data_init (&test, TRUE); + + /* Queue a buffer in agg:sink_1. Then do a flushing seek and check that the + * new flushing seek logic is triggered. On the first FLUSH_START call the + * buffers queued in collectpads should get flushed. Only one FLUSH_START and + * one FLUSH_STOP should be forwarded downstream. + */ + _chain_data_init (&data1, test.aggregator); + _chain_data_init (&data2, test.aggregator); + GST_BUFFER_TIMESTAMP (data2.buffer) = 0; + + gst_segment_init (&GST_AGGREGATOR (test.aggregator)->segment, + GST_FORMAT_TIME); + /* now do a successful flushing seek */ + event = gst_event_new_seek (1, GST_FORMAT_TIME, GST_SEEK_FLAG_FLUSH, + GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, 10 * GST_SECOND); + fail_unless (gst_pad_send_event (test.srcpad, event)); + + /* flushing starts once one of the upstream elements sends the first + * FLUSH_START */ + fail_unless_equals_int (test.flush_start_events, 0); + fail_unless_equals_int (test.flush_stop_events, 0); + + /* flush ogg:sink_0. This flushs collectpads, calls ::flush() and sends + * FLUSH_START downstream */ + GST_DEBUG ("Flushing: %s:%s", GST_DEBUG_PAD_NAME (data2.sinkpad)); + fail_unless (gst_pad_push_event (data2.srcpad, gst_event_new_flush_start ())); + + /* expect this buffer to be flushed */ + data2.expected_result = GST_FLOW_FLUSHING; + thread2 = g_thread_try_new ("gst-check", push_buffer, &data2, NULL); + + fail_unless (gst_pad_push_event (data1.srcpad, gst_event_new_flush_start ())); + fail_unless_equals_int (test.flush_start_events, 1); + fail_unless_equals_int (test.flush_stop_events, 0); + + /* the first FLUSH_STOP is not forwarded downstream */ + fail_unless (gst_pad_push_event (data1.srcpad, + gst_event_new_flush_stop (TRUE))); + fail_unless_equals_int (test.flush_start_events, 1); + fail_unless_equals_int (test.flush_stop_events, 0); + + /* at this point even the other pad agg:sink_1 should be flushing so thread2 + * should have stopped */ + g_thread_join (thread2); + + /* push a buffer on agg:sink_0 to trigger one collect after flushing to verify + * that flushing completes once all the pads have been flushed */ + thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL); + + /* flush agg:sink_1 as well. This completes the flushing seek so a FLUSH_STOP is + * sent downstream */ + gst_pad_push_event (data2.srcpad, gst_event_new_flush_stop (TRUE)); + + /* and the last FLUSH_STOP is forwarded downstream */ + fail_unless_equals_int (test.flush_start_events, 1); + + /* Check collected */ + gst_pad_add_probe (test.srcpad, GST_PAD_PROBE_TYPE_BUFFER, + (GstPadProbeCallback) _aggregated_cb, test.ml, NULL); + + data2.event = gst_event_new_eos (); + thread2 = g_thread_try_new ("gst-check", push_event, &data2, NULL); + + g_main_loop_run (test.ml); + g_source_remove (test.timeout_id); + + fail_unless_equals_int (test.flush_stop_events, 1); + + /* these will return immediately as at this point the threads have been + * unlocked and are finished */ + g_thread_join (thread1); + g_thread_join (thread2); + + _test_data_clear (&test); + +} + +GST_END_TEST; + +static void +infinite_seek (guint num_srcs, guint num_seeks) +{ + GstBus *bus; + GstMessage *message; + GstElement *pipeline, *src, *agg, *sink; + + gint count = 0, i; + gboolean seek_res, carry_on = TRUE; + + gst_init (NULL, NULL); + + pipeline = gst_pipeline_new ("pipeline"); + + agg = gst_check_setup_element ("testaggregator"); + sink = gst_check_setup_element ("fakesink"); + + fail_unless (gst_bin_add (GST_BIN (pipeline), agg)); + fail_unless (gst_bin_add (GST_BIN (pipeline), sink)); + fail_unless (gst_element_link (agg, sink)); + + for (i = 0; i < num_srcs; i++) { + src = gst_element_factory_make ("fakesrc", NULL); + g_object_set (src, "sizetype", 2, "sizemax", 4, NULL); + fail_unless (gst_bin_add (GST_BIN (pipeline), src)); + fail_unless (gst_element_link (src, agg)); + } + + bus = gst_element_get_bus (pipeline); + fail_if (bus == NULL); + gst_element_set_state (pipeline, GST_STATE_PLAYING); + while (count < num_seeks && carry_on) { + message = gst_bus_poll (bus, GST_MESSAGE_ANY, GST_SECOND / 10); + if (message) { + switch (GST_MESSAGE_TYPE (message)) { + case GST_MESSAGE_EOS: + { + /* we should check if we really finished here */ + GST_WARNING ("Got an EOS"); + carry_on = FALSE; + break; + } + case GST_MESSAGE_STATE_CHANGED: + { + GstState new; + + if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) { + gst_message_parse_state_changed (message, NULL, &new, NULL); + + if (new != GST_STATE_PLAYING) + break; + + GST_INFO ("Seeking (num: %i)", count); + seek_res = + gst_element_seek_simple (sink, GST_FORMAT_BYTES, + GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, 0); + GST_INFO ("seek result is : %d", seek_res); + fail_unless (seek_res != 0); + count++; + } + + break; + } + case GST_MESSAGE_ERROR: + GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message); + carry_on = FALSE; + fail_error_message (message); + break; + default: + break; + } + gst_message_unref (message); + } + } + + gst_element_set_state (pipeline, GST_STATE_NULL); + gst_object_unref (bus); + gst_object_unref (pipeline); +} + +GST_START_TEST (test_infinite_seek) +{ + infinite_seek (2, 500); +} + +GST_END_TEST; + +GST_START_TEST (test_infinite_seek_50_src) +{ + infinite_seek (50, 100); +} + +GST_END_TEST; + +typedef struct +{ + GstElement *agg, *src, *pipeline; + GCond *cond; + GMutex *lock; +} RemoveElementData; + +static GstPadProbeReturn +pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, RemoveElementData * data) +{ + GstPad *peer; + + GST_INFO_OBJECT (pad, "Removing pad"); + + peer = gst_pad_get_peer (pad); + gst_pad_unlink (pad, peer); + gst_element_release_request_pad (data->agg, peer); + fail_unless (gst_bin_remove (GST_BIN (data->pipeline), data->src)); + + g_mutex_lock (data->lock); + g_cond_broadcast (data->cond); + g_mutex_unlock (data->lock); + + return GST_PAD_PROBE_OK; +} + +GST_START_TEST (test_add_remove) +{ + /* Used to notify that we removed the pad from */ + GCond cond; + GMutex lock; + + GstBus *bus; + GstState state; + GstMessage *message; + gboolean carry_on = TRUE; + guint num_iterations = 100; + + GstPad *pad; + GstElement *pipeline, *src, *src1 = NULL, *agg, *sink; + + gint count = 0; + + gst_init (NULL, NULL); + g_mutex_init (&lock); + g_cond_init (&cond); + + pipeline = gst_pipeline_new ("pipeline"); + + agg = gst_check_setup_element ("testaggregator"); + sink = gst_check_setup_element ("fakesink"); + + fail_unless (gst_bin_add (GST_BIN (pipeline), agg)); + fail_unless (gst_bin_add (GST_BIN (pipeline), sink)); + fail_unless (gst_element_link (agg, sink)); + + bus = gst_element_get_bus (pipeline); + while (count < num_iterations) { + + src = gst_element_factory_make ("fakesrc", NULL); + g_object_set (src, "num-buffers", 100000, "sizetype", 2, "sizemax", 4, + NULL); + gst_element_set_locked_state (src, TRUE); + fail_unless (gst_bin_add (GST_BIN (pipeline), src)); + fail_unless (gst_element_link (src, agg)); + gst_element_set_locked_state (src, FALSE); + fail_unless (gst_element_sync_state_with_parent (src)); + + if (count == 0) + gst_element_set_state (pipeline, GST_STATE_PLAYING); + + /* Now make sure the seek happend */ + carry_on = TRUE; + do { + message = gst_bus_timed_pop (bus, -1); + switch (GST_MESSAGE_TYPE (message)) { + case GST_MESSAGE_EOS: + { + /* we should check if we really finished here */ + GST_WARNING ("Got an EOS"); + carry_on = FALSE; + break; + } + case GST_MESSAGE_STATE_CHANGED: + { + if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) { + gst_message_parse_state_changed (message, NULL, &state, NULL); + + if (state == GST_STATE_PLAYING) { + RemoveElementData data; + + carry_on = FALSE; + if (count == 0) { + GST_DEBUG ("First run, not removing any element yet"); + + break; + } + + data.src = gst_object_ref (src1); + data.agg = agg; + data.lock = &lock; + data.cond = &cond; + data.pipeline = pipeline; + pad = gst_element_get_static_pad (data.src, "src"); + + g_mutex_lock (&lock); + gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM, + (GstPadProbeCallback) pad_probe_cb, &data, NULL); + GST_INFO ("Waiting for %" GST_PTR_FORMAT " %s", pad, + gst_element_state_get_name (GST_STATE (data.src))); + g_cond_wait (&cond, &lock); + g_mutex_unlock (&lock); + + /* We can not set state from the streaming thread so we + * need to make sure that the source has been removed + * before setting its state to NULL */ + gst_element_set_state (data.src, GST_STATE_NULL); + + gst_object_unref (data.src); + } + } + + break; + } + case GST_MESSAGE_ERROR: + { + GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message); + carry_on = FALSE; + fail_error_message (message); + break; + } + default: + break; + } + + gst_message_unref (message); + } while (carry_on); + + GST_INFO ("Seeking"); + fail_unless (gst_element_seek_simple (pipeline, GST_FORMAT_BYTES, + GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, 0)); + + count++; + src1 = src; + } + gst_element_set_state (pipeline, GST_STATE_NULL); + gst_object_unref (bus); + gst_object_unref (pipeline); +} + +GST_END_TEST; + +GST_START_TEST (test_change_state_intensive) +{ + GstBus *bus; + GstMessage *message; + GstElement *pipeline, *src, *agg, *sink; + + gint i, state_i = 0, num_srcs = 3; + gboolean carry_on = TRUE, ready = FALSE; + GstStateChangeReturn state_return; + GstState wanted_state, wanted_states[] = { + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY, + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY, + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY, + GST_STATE_PAUSED, GST_STATE_READY, GST_STATE_PAUSED, GST_STATE_READY, + GST_STATE_PAUSED, GST_STATE_READY, GST_STATE_PAUSED, GST_STATE_NULL, + GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_NULL, + GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_NULL, + GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL, + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL, + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL, + GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL, + }; + + gst_init (NULL, NULL); + + pipeline = gst_pipeline_new ("pipeline"); + + agg = gst_check_setup_element ("testaggregator"); + sink = gst_check_setup_element ("fakesink"); + + fail_unless (gst_bin_add (GST_BIN (pipeline), agg)); + fail_unless (gst_bin_add (GST_BIN (pipeline), sink)); + fail_unless (gst_element_link (agg, sink)); + + for (i = 0; i < num_srcs; i++) { + src = gst_element_factory_make ("fakesrc", NULL); + g_object_set (src, "sizetype", 2, "sizemax", 4, NULL); + fail_unless (gst_bin_add (GST_BIN (pipeline), src)); + fail_unless (gst_element_link (src, agg)); + } + + bus = gst_element_get_bus (pipeline); + fail_if (bus == NULL); + + wanted_state = wanted_states[state_i++]; + state_return = gst_element_set_state (pipeline, wanted_state); + + while (state_i < G_N_ELEMENTS (wanted_states) && carry_on) { + if (state_return == GST_STATE_CHANGE_SUCCESS && ready) { + wanted_state = wanted_states[state_i++]; + fail_unless (gst_element_set_state (pipeline, wanted_state), + GST_STATE_CHANGE_SUCCESS); + GST_INFO ("Wanted state: %s", gst_element_state_get_name (wanted_state)); + } + + message = gst_bus_poll (bus, GST_MESSAGE_ANY, GST_SECOND / 10); + if (message) { + switch (GST_MESSAGE_TYPE (message)) { + case GST_MESSAGE_EOS: + { + /* we should check if we really finished here */ + GST_WARNING ("Got an EOS"); + carry_on = FALSE; + break; + } + case GST_MESSAGE_STATE_CHANGED: + { + GstState new; + + if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) { + gst_message_parse_state_changed (message, NULL, &new, NULL); + + if (new != wanted_state) { + ready = FALSE; + break; + } + + GST_DEBUG ("State %s reached", + gst_element_state_get_name (wanted_state)); + wanted_state = wanted_states[state_i++]; + GST_DEBUG ("Wanted state: %s", + gst_element_state_get_name (wanted_state)); + state_return = gst_element_set_state (pipeline, wanted_state); + fail_unless (state_return == GST_STATE_CHANGE_SUCCESS || + state_return == GST_STATE_CHANGE_ASYNC); + ready = TRUE; + } + + break; + } + case GST_MESSAGE_ERROR: + GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message); + carry_on = FALSE; + break; + default: + break; + } + gst_message_unref (message); + } + } + + gst_element_set_state (pipeline, GST_STATE_NULL); + gst_object_unref (bus); + gst_object_unref (pipeline); +} + +GST_END_TEST; + +static Suite * +gst_aggregator_suite (void) +{ + Suite *suite; + TCase *general; + + gst_test_aggregator_plugin_register (); + + suite = suite_create ("GstAggregator"); + + general = tcase_create ("general"); + suite_add_tcase (suite, general); + tcase_add_test (general, test_aggregate); + tcase_add_test (general, test_aggregate_eos); + tcase_add_test (general, test_flushing_seek); + tcase_add_test (general, test_infinite_seek); + tcase_add_test (general, test_infinite_seek_50_src); + tcase_add_test (general, test_linear_pipeline); + tcase_add_test (general, test_two_src_pipeline); + tcase_add_test (general, test_add_remove); + tcase_add_test (general, test_change_state_intensive); + + return suite; +} + +GST_CHECK_MAIN (gst_aggregator); -- 2.7.4