From 6f231f89d6eb814872b4773d1e7cc3835cd3fa93 Mon Sep 17 00:00:00 2001 From: =?utf8?q?Sebastian=20Dr=C3=B6ge?= Date: Fri, 28 Oct 2011 09:18:55 +0200 Subject: [PATCH] base: Add collectpads2 This handles muxing of sparse/subtitle streams and has lots of cleanup. Still missing is special support for live streams but this can be added later without breaking API/ABI. Based on the version from the videomixer plugin. https://bugzilla.gnome.org/show_bug.cgi?id=415754 --- libs/gst/base/Makefile.am | 2 + libs/gst/base/gstcollectpads2.c | 1975 +++++++++++++++++++++++++++++++++++++++ libs/gst/base/gstcollectpads2.h | 362 +++++++ 3 files changed, 2339 insertions(+) create mode 100644 libs/gst/base/gstcollectpads2.c create mode 100644 libs/gst/base/gstcollectpads2.h diff --git a/libs/gst/base/Makefile.am b/libs/gst/base/Makefile.am index f6fa3ef..4c8bccf 100644 --- a/libs/gst/base/Makefile.am +++ b/libs/gst/base/Makefile.am @@ -12,6 +12,7 @@ libgstbase_@GST_MAJORMINOR@_la_SOURCES = \ gstbytereader.c \ gstbytewriter.c \ gstcollectpads.c \ + gstcollectpads2.c \ gstpushsrc.c \ gsttypefindhelper.c \ gstdataqueue.c @@ -33,6 +34,7 @@ libgstbase_@GST_MAJORMINOR@include_HEADERS = \ gstbytereader.h \ gstbytewriter.h \ gstcollectpads.h \ + gstcollectpads2.h \ gstpushsrc.h \ gsttypefindhelper.h \ gstdataqueue.h diff --git a/libs/gst/base/gstcollectpads2.c b/libs/gst/base/gstcollectpads2.c new file mode 100644 index 0000000..428638e --- /dev/null +++ b/libs/gst/base/gstcollectpads2.c @@ -0,0 +1,1975 @@ +/* GStreamer + * Copyright (C) 2005 Wim Taymans + * Copyright (C) 2008 Mark Nauwelaerts + * Copyright (C) 2010 Sebastian Dröge + * + * gstcollectpads2.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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ +/** + * SECTION:gstcollectpads2 + * @short_description: manages a set of pads that operate in collect mode + * @see_also: + * + * Manages a set of pads that operate in collect mode. This means that control + * is given to the manager of this object when all pads have data. + * + * + * Collectpads are created with gst_collect_pads2_new(). A callback should then + * be installed with gst_collect_pads2_set_function (). + * + * + * Pads are added to the collection with gst_collect_pads2_add_pad()/ + * gst_collect_pads2_remove_pad(). The pad + * has to be a sinkpad. The chain and event functions of the pad are + * overridden. The element_private of the pad is used to store + * private information for the collectpads. + * + * + * For each pad, data is queued in the _chain function or by + * performing a pull_range. + * + * + * When data is queued on all pads in waiting mode, the callback function is called. + * + * + * Data can be dequeued from the pad with the gst_collect_pads2_pop() method. + * One can peek at the data with the gst_collect_pads2_peek() function. + * These functions will return NULL if the pad received an EOS event. When all + * pads return NULL from a gst_collect_pads2_peek(), the element can emit an EOS + * event itself. + * + * + * Data can also be dequeued in byte units using the gst_collect_pads2_available(), + * gst_collect_pads2_read() and gst_collect_pads2_flush() calls. + * + * + * Elements should call gst_collect_pads2_start() and gst_collect_pads2_stop() in + * their state change functions to start and stop the processing of the collectpads. + * The gst_collect_pads2_stop() call should be called before calling the parent + * element state change function in the PAUSED_TO_READY state change to ensure + * no pad is blocked and the element can finish streaming. + * + * + * gst_collect_pads2_collect() and gst_collect_pads2_collect_range() can be used by + * elements that start a #GstTask to drive the collect_pads2. This feature is however + * not yet implemented. + * + * + * gst_collect_pads2_set_waiting() sets a pad to waiting or non-waiting mode. + * CollectPads element is not waiting for data to be collected on non-waiting pads. + * Thus these pads may but need not have data when the callback is called. + * All pads are in waiting mode by default. + * + * + * + * Last reviewed on 2008-03-14 (0.10.17) + * + * Since: 0.10.36 + */ + +#include "gstcollectpads2.h" + +GST_DEBUG_CATEGORY_STATIC (collect_pads2_debug); +#define GST_CAT_DEFAULT collect_pads2_debug + +GST_BOILERPLATE (GstCollectPads2, gst_collect_pads2, GstObject, + GST_TYPE_OBJECT); + +static void gst_collect_pads2_clear (GstCollectPads2 * pads, + GstCollectData2 * data); +static GstFlowReturn gst_collect_pads2_chain (GstPad * pad, GstBuffer * buffer); +static gboolean gst_collect_pads2_event (GstPad * pad, GstEvent * event); +static void gst_collect_pads2_finalize (GObject * object); +static GstFlowReturn gst_collect_pads2_default_collected (GstCollectPads2 * + pads, gpointer user_data); +static gint gst_collect_pads2_default_compare_func (GstCollectPads2 * pads, + GstCollectData2 * data1, GstClockTime timestamp1, GstCollectData2 * data2, + GstClockTime timestamp2, gpointer user_data); +static gboolean gst_collect_pads2_recalculate_full (GstCollectPads2 * pads); +static void ref_data (GstCollectData2 * data); +static void unref_data (GstCollectData2 * data); + +/* Some properties are protected by LOCK, others by STREAM_LOCK + * However, manipulating either of these partitions may require + * to signal/wake a _WAIT, so use a separate (sort of) event to prevent races + * Alternative implementations are possible, e.g. some low-level re-implementing + * of the 2 above locks to drop both of them atomically when going into _WAIT. + */ +#define GST_COLLECT_PADS2_GET_EVT_COND(pads) (((GstCollectPads2 *)pads)->evt_cond) +#define GST_COLLECT_PADS2_GET_EVT_LOCK(pads) (((GstCollectPads2 *)pads)->evt_lock) +#define GST_COLLECT_PADS2_EVT_WAIT(pads, cookie) G_STMT_START { \ + g_mutex_lock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ + /* should work unless a lot of event'ing and thread starvation */\ + while (cookie == ((GstCollectPads2 *) pads)->evt_cookie) \ + g_cond_wait (GST_COLLECT_PADS2_GET_EVT_COND (pads), \ + GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ + cookie = ((GstCollectPads2 *) pads)->evt_cookie; \ + g_mutex_unlock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ +} G_STMT_END +#define GST_COLLECT_PADS2_EVT_WAIT_TIMED(pads, cookie, timeout) G_STMT_START { \ + GTimeVal __tv; \ + \ + g_get_current_time (&tv); \ + g_time_val_add (&tv, timeout); \ + \ + g_mutex_lock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ + /* should work unless a lot of event'ing and thread starvation */\ + while (cookie == ((GstCollectPads2 *) pads)->evt_cookie) \ + g_cond_timed_wait (GST_COLLECT_PADS2_GET_EVT_COND (pads), \ + GST_COLLECT_PADS2_GET_EVT_LOCK (pads), &tv); \ + cookie = ((GstCollectPads2 *) pads)->evt_cookie; \ + g_mutex_unlock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ +} G_STMT_END +#define GST_COLLECT_PADS2_EVT_BROADCAST(pads) G_STMT_START { \ + g_mutex_lock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ + /* never mind wrap-around */ \ + ++(((GstCollectPads2 *) pads)->evt_cookie); \ + g_cond_broadcast (GST_COLLECT_PADS2_GET_EVT_COND (pads)); \ + g_mutex_unlock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ +} G_STMT_END +#define GST_COLLECT_PADS2_EVT_INIT(cookie) G_STMT_START { \ + g_mutex_lock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ + cookie = ((GstCollectPads2 *) pads)->evt_cookie; \ + g_mutex_unlock (GST_COLLECT_PADS2_GET_EVT_LOCK (pads)); \ +} G_STMT_END + +static void +gst_collect_pads2_base_init (gpointer g_class) +{ + /* Do nothing here */ +} + +static void +gst_collect_pads2_class_init (GstCollectPads2Class * klass) +{ + GObjectClass *gobject_class = (GObjectClass *) klass; + + GST_DEBUG_CATEGORY_INIT (collect_pads2_debug, "collectpads2", 0, + "GstCollectPads2"); + + gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_collect_pads2_finalize); +} + +static void +gst_collect_pads2_init (GstCollectPads2 * pads, GstCollectPads2Class * g_class) +{ + pads->data = NULL; + pads->cookie = 0; + pads->numpads = 0; + pads->queuedpads = 0; + pads->eospads = 0; + pads->started = FALSE; + + g_static_rec_mutex_init (&pads->stream_lock); + + pads->func = gst_collect_pads2_default_collected; + pads->user_data = NULL; + pads->event_func = NULL; + pads->event_user_data = NULL; + + pads->prepare_buffer_func = NULL; + pads->prepare_buffer_user_data = NULL; + + /* members for default muxing */ + pads->buffer_func = NULL; + pads->buffer_user_data = NULL; + pads->compare_func = gst_collect_pads2_default_compare_func; + pads->compare_user_data = NULL; + pads->earliest_data = NULL; + pads->earliest_time = GST_CLOCK_TIME_NONE; + + /* members to manage the pad list */ + pads->pad_cookie = 0; + pads->pad_list = NULL; + + /* members for event */ + pads->evt_lock = g_mutex_new (); + pads->evt_cond = g_cond_new (); + pads->evt_cookie = 0; +} + +static void +gst_collect_pads2_finalize (GObject * object) +{ + GstCollectPads2 *pads = GST_COLLECT_PADS2 (object); + + GST_DEBUG_OBJECT (object, "finalize"); + + g_static_rec_mutex_free (&pads->stream_lock); + + g_cond_free (pads->evt_cond); + g_mutex_free (pads->evt_lock); + + /* Remove pads and free pads list */ + g_slist_foreach (pads->pad_list, (GFunc) unref_data, NULL); + g_slist_foreach (pads->data, (GFunc) unref_data, NULL); + g_slist_free (pads->data); + g_slist_free (pads->pad_list); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +/** + * gst_collect_pads2_new: + * + * Create a new instance of #GstCollectsPads. + * + * Returns: a new #GstCollectPads2, or NULL in case of an error. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstCollectPads2 * +gst_collect_pads2_new (void) +{ + GstCollectPads2 *newcoll; + + newcoll = g_object_new (GST_TYPE_COLLECT_PADS2, NULL); + + return newcoll; +} + +/** + * gst_collect_pads2_set_prepare_buffer_function: + * @pads: the collectpads to use + * @func: the function to set + * @user_data: user data passed to the function + * + * Set the callback function and user data that will be called + * for every buffer that arrives. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_prepare_buffer_function (GstCollectPads2 * pads, + GstCollectPads2BufferFunction func, gpointer user_data) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_OBJECT_LOCK (pads); + pads->prepare_buffer_func = func; + pads->prepare_buffer_user_data = user_data; + GST_OBJECT_UNLOCK (pads); +} + +/* Must be called with GstObject lock! */ +static void +gst_collect_pads2_set_buffer_function_locked (GstCollectPads2 * pads, + GstCollectPads2BufferFunction func, gpointer user_data) +{ + pads->buffer_func = func; + pads->buffer_user_data = user_data; +} + +/** + * gst_collect_pads2_set_buffer_function: + * @pads: the collectpads to use + * @func: the function to set + * @user_data: user data passed to the function + * + * Set the callback function and user data that will be called with + * the oldest buffer when all pads have been collected. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_buffer_function (GstCollectPads2 * pads, + GstCollectPads2BufferFunction func, gpointer user_data) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_OBJECT_LOCK (pads); + gst_collect_pads2_set_buffer_function_locked (pads, func, user_data); + GST_OBJECT_UNLOCK (pads); +} + +/** + * gst_collect_pads2_set_compare_function: + * @pads: the pads to use + * @func: the function to set + * @user_data: user data passed to the function + * + * Set the timestamp comparison function. + * + * MT safe. + * + * Since: 0.10.36 + */ +/* NOTE allowing to change comparison seems not advisable; +no known use-case, and collaboration with default algorithm is unpredictable. +If custom compairing/operation is needed, just use a collect function of +your own */ +void +gst_collect_pads2_set_compare_function (GstCollectPads2 * pads, + GstCollectPads2CompareFunction func, gpointer user_data) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_OBJECT_LOCK (pads); + pads->compare_func = func; + pads->compare_user_data = user_data; + GST_OBJECT_UNLOCK (pads); +} + +/** + * gst_collect_pads2_set_function: + * @pads: the collectspads to use + * @func: the function to set + * @user_data: user data passed to the function + * + * CollectPads provides a default collection algorithm that will determine + * the oldest buffer available on all of its pads, and then delegate + * to a configured callback. + * However, if circumstances are more complicated and/or more control + * is desired, this sets a callback that will be invoked instead when + * all the pads added to the collection have buffers queued. + * Evidently, this callback is not compatible with + * gst_collect_pads2_set_buffer_function() callback. + * If this callback is set, the former will be unset. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_function (GstCollectPads2 * pads, + GstCollectPads2Function func, gpointer user_data) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_OBJECT_LOCK (pads); + pads->func = func; + pads->user_data = user_data; + gst_collect_pads2_set_buffer_function_locked (pads, NULL, NULL); + GST_OBJECT_UNLOCK (pads); +} + +static void +ref_data (GstCollectData2 * data) +{ + g_assert (data != NULL); + + g_atomic_int_inc (&(data->refcount)); +} + +static void +unref_data (GstCollectData2 * data) +{ + g_assert (data != NULL); + g_assert (data->refcount > 0); + + if (!g_atomic_int_dec_and_test (&(data->refcount))) + return; + + if (data->destroy_notify) + data->destroy_notify (data); + + g_object_unref (data->pad); + if (data->buffer) { + gst_buffer_unref (data->buffer); + } + g_free (data); +} + +/** + * gst_collect_pads2_set_event_function: + * @pads: the collectspads to use + * @func: the function to set + * @user_data: user data passed to the function + * + * Set the event callback function and user data that will be called after + * collectpads has processed and event originating from one of the collected + * pads. If the event being processed is a serialized one, this callback is + * called with @pads STREAM_LOCK held, otherwise not. As this lock should be + * held when calling a number of CollectPads functions, it should be acquired + * if so (unusually) needed. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_event_function (GstCollectPads2 * pads, + GstCollectPads2EventFunction func, gpointer user_data) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_OBJECT_LOCK (pads); + pads->event_func = func; + pads->event_user_data = user_data; + GST_OBJECT_UNLOCK (pads); +} + +/** + * gst_collect_pads2_add_pad: + * @pads: the collectspads to use + * @pad: the pad to add + * @size: the size of the returned #GstCollectData2 structure + * + * Add a pad to the collection of collect pads. The pad has to be + * a sinkpad. The refcount of the pad is incremented. Use + * gst_collect_pads2_remove_pad() to remove the pad from the collection + * again. + * + * You specify a size for the returned #GstCollectData2 structure + * so that you can use it to store additional information. + * + * The pad will be automatically activated in push mode when @pads is + * started. + * + * This function calls gst_collect_pads2_add_pad() passing a value of NULL + * for destroy_notify and TRUE for locked. + * + * Returns: a new #GstCollectData2 to identify the new pad. Or NULL + * if wrong parameters are supplied. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstCollectData2 * +gst_collect_pads2_add_pad (GstCollectPads2 * pads, GstPad * pad, guint size) +{ + return gst_collect_pads2_add_pad_full (pads, pad, size, NULL, TRUE); +} + +/** + * gst_collect_pads2_add_pad_full: + * @pads: the collectspads to use + * @pad: the pad to add + * @size: the size of the returned #GstCollectData2 structure + * @destroy_notify: function to be called before the returned #GstCollectData2 + * structure is freed + * @lock: whether to lock this pad in usual waiting state + * + * Add a pad to the collection of collect pads. The pad has to be + * a sinkpad. The refcount of the pad is incremented. Use + * gst_collect_pads2_remove_pad() to remove the pad from the collection + * again. + * + * You specify a size for the returned #GstCollectData2 structure + * so that you can use it to store additional information. + * + * You can also specify a #GstCollectData2DestroyNotify that will be called + * just before the #GstCollectData2 structure is freed. It is passed the + * pointer to the structure and should free any custom memory and resources + * allocated for it. + * + * Keeping a pad locked in waiting state is only relevant when using + * the default collection algorithm (providing the oldest buffer). + * It ensures a buffer must be available on this pad for a collection + * to take place. This is of typical use to a muxer element where + * non-subtitle streams should always be in waiting state, + * e.g. to assure that caps information is available on all these streams + * when initial headers have to be written. + * + * The pad will be automatically activated in push mode when @pads is + * started. + * + * Since: 0.10.36 + * + * Returns: a new #GstCollectData2 to identify the new pad. Or NULL + * if wrong parameters are supplied. + * + * MT safe. + */ +GstCollectData2 * +gst_collect_pads2_add_pad_full (GstCollectPads2 * pads, GstPad * pad, + guint size, GstCollectData2DestroyNotify destroy_notify, gboolean lock) +{ + GstCollectData2 *data; + + g_return_val_if_fail (pads != NULL, NULL); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), NULL); + g_return_val_if_fail (pad != NULL, NULL); + g_return_val_if_fail (GST_PAD_IS_SINK (pad), NULL); + g_return_val_if_fail (size >= sizeof (GstCollectData2), NULL); + + GST_DEBUG_OBJECT (pads, "adding pad %s:%s", GST_DEBUG_PAD_NAME (pad)); + + data = g_malloc0 (size); + data->collect = pads; + data->pad = gst_object_ref (pad); + data->buffer = NULL; + data->pos = 0; + gst_segment_init (&data->segment, GST_FORMAT_UNDEFINED); + data->state = GST_COLLECT_PADS2_STATE_WAITING; + data->state |= lock ? GST_COLLECT_PADS2_STATE_LOCKED : 0; + data->refcount = 1; + data->destroy_notify = destroy_notify; + + GST_OBJECT_LOCK (pads); + GST_OBJECT_LOCK (pad); + gst_pad_set_element_private (pad, data); + GST_OBJECT_UNLOCK (pad); + pads->pad_list = g_slist_append (pads->pad_list, data); + gst_pad_set_chain_function (pad, GST_DEBUG_FUNCPTR (gst_collect_pads2_chain)); + gst_pad_set_event_function (pad, GST_DEBUG_FUNCPTR (gst_collect_pads2_event)); + /* backward compat, also add to data if stopped, so that the element already + * has this in the public data list before going PAUSED (typically) + * this can only be done when we are stopped because we don't take the + * STREAM_LOCK to protect the pads->data list. */ + if (!pads->started) { + pads->data = g_slist_append (pads->data, data); + ref_data (data); + } + /* activate the pad when needed */ + if (pads->started) + gst_pad_set_active (pad, TRUE); + pads->pad_cookie++; + GST_OBJECT_UNLOCK (pads); + + return data; +} + +static gint +find_pad (GstCollectData2 * data, GstPad * pad) +{ + if (data->pad == pad) + return 0; + return 1; +} + +/** + * gst_collect_pads2_remove_pad: + * @pads: the collectspads to use + * @pad: the pad to remove + * + * Remove a pad from the collection of collect pads. This function will also + * free the #GstCollectData2 and all the resources that were allocated with + * gst_collect_pads2_add_pad(). + * + * The pad will be deactivated automatically when @pads is stopped. + * + * Returns: %TRUE if the pad could be removed. + * + * MT safe. + * + * Since: 0.10.36 + */ +gboolean +gst_collect_pads2_remove_pad (GstCollectPads2 * pads, GstPad * pad) +{ + GstCollectData2 *data; + GSList *list; + + g_return_val_if_fail (pads != NULL, FALSE); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), FALSE); + g_return_val_if_fail (pad != NULL, FALSE); + g_return_val_if_fail (GST_IS_PAD (pad), FALSE); + + GST_DEBUG_OBJECT (pads, "removing pad %s:%s", GST_DEBUG_PAD_NAME (pad)); + + GST_OBJECT_LOCK (pads); + list = g_slist_find_custom (pads->pad_list, pad, (GCompareFunc) find_pad); + if (!list) + goto unknown_pad; + + data = (GstCollectData2 *) list->data; + + GST_DEBUG_OBJECT (pads, "found pad %s:%s at %p", GST_DEBUG_PAD_NAME (pad), + data); + + /* clear the stuff we configured */ + gst_pad_set_chain_function (pad, NULL); + gst_pad_set_event_function (pad, NULL); + GST_OBJECT_LOCK (pad); + gst_pad_set_element_private (pad, NULL); + GST_OBJECT_UNLOCK (pad); + + /* backward compat, also remove from data if stopped, note that this function + * can only be called when we are stopped because we don't take the + * STREAM_LOCK to protect the pads->data list. */ + if (!pads->started) { + GSList *dlist; + + dlist = g_slist_find_custom (pads->data, pad, (GCompareFunc) find_pad); + if (dlist) { + GstCollectData2 *pdata = dlist->data; + + pads->data = g_slist_delete_link (pads->data, dlist); + unref_data (pdata); + } + } + /* remove from the pad list */ + pads->pad_list = g_slist_delete_link (pads->pad_list, list); + pads->pad_cookie++; + + /* signal waiters because something changed */ + GST_COLLECT_PADS2_EVT_BROADCAST (pads); + + /* deactivate the pad when needed */ + if (!pads->started) + gst_pad_set_active (pad, FALSE); + + /* clean and free the collect data */ + unref_data (data); + + GST_OBJECT_UNLOCK (pads); + + return TRUE; + +unknown_pad: + { + GST_WARNING_OBJECT (pads, "cannot remove unknown pad %s:%s", + GST_DEBUG_PAD_NAME (pad)); + GST_OBJECT_UNLOCK (pads); + return FALSE; + } +} + +/** + * gst_collect_pads2_is_active: + * @pads: the collectspads to use + * @pad: the pad to check + * + * Check if a pad is active. + * + * This function is currently not implemented. + * + * Returns: %TRUE if the pad is active. + * + * MT safe. + * + * Since: 0.10.36 + */ +gboolean +gst_collect_pads2_is_active (GstCollectPads2 * pads, GstPad * pad) +{ + g_return_val_if_fail (pads != NULL, FALSE); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), FALSE); + g_return_val_if_fail (pad != NULL, FALSE); + g_return_val_if_fail (GST_IS_PAD (pad), FALSE); + + g_warning ("gst_collect_pads2_is_active() is not implemented"); + + return FALSE; +} + +/** + * gst_collect_pads2_collect: + * @pads: the collectspads to use + * + * Collect data on all pads. This function is usually called + * from a #GstTask function in an element. + * + * This function is currently not implemented. + * + * Returns: #GstFlowReturn of the operation. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstFlowReturn +gst_collect_pads2_collect (GstCollectPads2 * pads) +{ + g_return_val_if_fail (pads != NULL, GST_FLOW_ERROR); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), GST_FLOW_ERROR); + + g_warning ("gst_collect_pads2_collect() is not implemented"); + + return GST_FLOW_NOT_SUPPORTED; +} + +/** + * gst_collect_pads2_collect_range: + * @pads: the collectspads to use + * @offset: the offset to collect + * @length: the length to collect + * + * Collect data with @offset and @length on all pads. This function + * is typically called in the getrange function of an element. + * + * This function is currently not implemented. + * + * Returns: #GstFlowReturn of the operation. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstFlowReturn +gst_collect_pads2_collect_range (GstCollectPads2 * pads, guint64 offset, + guint length) +{ + g_return_val_if_fail (pads != NULL, GST_FLOW_ERROR); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), GST_FLOW_ERROR); + + g_warning ("gst_collect_pads2_collect_range() is not implemented"); + + return GST_FLOW_NOT_SUPPORTED; +} + +/* + * Must be called with STREAM_LOCK. + */ +static void +gst_collect_pads2_set_flushing_unlocked (GstCollectPads2 * pads, + gboolean flushing) +{ + GSList *walk = NULL; + + /* Update the pads flushing flag */ + for (walk = pads->data; walk; walk = g_slist_next (walk)) { + GstCollectData2 *cdata = walk->data; + + if (GST_IS_PAD (cdata->pad)) { + GST_OBJECT_LOCK (cdata->pad); + if (flushing) + GST_PAD_SET_FLUSHING (cdata->pad); + else + GST_PAD_UNSET_FLUSHING (cdata->pad); + if (flushing) + GST_COLLECT_PADS2_STATE_SET (cdata, GST_COLLECT_PADS2_STATE_FLUSHING); + else + GST_COLLECT_PADS2_STATE_UNSET (cdata, GST_COLLECT_PADS2_STATE_FLUSHING); + gst_collect_pads2_clear (pads, cdata); + GST_OBJECT_UNLOCK (cdata->pad); + } + } + + /* inform _chain of changes */ + GST_COLLECT_PADS2_EVT_BROADCAST (pads); +} + +/** + * gst_collect_pads2_set_flushing: + * @pads: the collectspads to use + * @flushing: desired state of the pads + * + * Change the flushing state of all the pads in the collection. No pad + * is able to accept anymore data when @flushing is %TRUE. Calling this + * function with @flushing %FALSE makes @pads accept data again. + * Caller must ensure that downstream streaming (thread) is not blocked, + * e.g. by sending a FLUSH_START downstream. + * + * MT safe. + * + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_flushing (GstCollectPads2 * pads, gboolean flushing) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + /* NOTE since this eventually calls _pop, some (STREAM_)LOCK is needed here */ + GST_COLLECT_PADS2_STREAM_LOCK (pads); + gst_collect_pads2_set_flushing_unlocked (pads, flushing); + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); +} + +/** + * gst_collect_pads2_start: + * @pads: the collectspads to use + * + * Starts the processing of data in the collect_pads2. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_start (GstCollectPads2 * pads) +{ + GSList *collected; + + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_DEBUG_OBJECT (pads, "starting collect pads"); + + /* make sure stop and collect cannot be called anymore */ + GST_COLLECT_PADS2_STREAM_LOCK (pads); + + /* make pads streamable */ + GST_OBJECT_LOCK (pads); + + /* loop over the master pad list and reset the segment */ + collected = pads->pad_list; + for (; collected; collected = g_slist_next (collected)) { + GstCollectData2 *data; + + data = collected->data; + gst_segment_init (&data->segment, GST_FORMAT_UNDEFINED); + } + + gst_collect_pads2_set_flushing_unlocked (pads, FALSE); + + /* Start collect pads */ + pads->started = TRUE; + GST_OBJECT_UNLOCK (pads); + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); +} + +/** + * gst_collect_pads2_stop: + * @pads: the collectspads to use + * + * Stops the processing of data in the collect_pads2. this function + * will also unblock any blocking operations. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_stop (GstCollectPads2 * pads) +{ + GSList *collected; + + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + + GST_DEBUG_OBJECT (pads, "stopping collect pads"); + + /* make sure collect and start cannot be called anymore */ + GST_COLLECT_PADS2_STREAM_LOCK (pads); + + /* make pads not accept data anymore */ + GST_OBJECT_LOCK (pads); + gst_collect_pads2_set_flushing_unlocked (pads, TRUE); + + /* Stop collect pads */ + pads->started = FALSE; + pads->eospads = 0; + pads->queuedpads = 0; + + /* loop over the master pad list and flush buffers */ + collected = pads->pad_list; + for (; collected; collected = g_slist_next (collected)) { + GstCollectData2 *data; + GstBuffer **buffer_p; + + data = collected->data; + if (data->buffer) { + buffer_p = &data->buffer; + gst_buffer_replace (buffer_p, NULL); + data->pos = 0; + } + GST_COLLECT_PADS2_STATE_UNSET (data, GST_COLLECT_PADS2_STATE_EOS); + } + + if (pads->earliest_data) + unref_data (pads->earliest_data); + pads->earliest_data = NULL; + pads->earliest_time = GST_CLOCK_TIME_NONE; + + GST_OBJECT_UNLOCK (pads); + /* Wake them up so they can end the chain functions. */ + GST_COLLECT_PADS2_EVT_BROADCAST (pads); + + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); +} + +/** + * gst_collect_pads2_peek: + * @pads: the collectspads to peek + * @data: the data to use + * + * Peek at the buffer currently queued in @data. This function + * should be called with the @pads STREAM_LOCK held, such as in the callback + * handler. + * + * Returns: The buffer in @data or NULL if no buffer is queued. + * should unref the buffer after usage. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstBuffer * +gst_collect_pads2_peek (GstCollectPads2 * pads, GstCollectData2 * data) +{ + GstBuffer *result; + + g_return_val_if_fail (pads != NULL, NULL); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), NULL); + g_return_val_if_fail (data != NULL, NULL); + + if ((result = data->buffer)) + gst_buffer_ref (result); + + GST_DEBUG_OBJECT (pads, "Peeking at pad %s:%s: buffer=%p", + GST_DEBUG_PAD_NAME (data->pad), result); + + return result; +} + +/** + * gst_collect_pads2_pop: + * @pads: the collectspads to pop + * @data: the data to use + * + * Pop the buffer currently queued in @data. This function + * should be called with the @pads STREAM_LOCK held, such as in the callback + * handler. + * + * Returns: The buffer in @data or NULL if no buffer was queued. + * You should unref the buffer after usage. + * + * MT safe. + * + * Since: 0.10.36 + */ +GstBuffer * +gst_collect_pads2_pop (GstCollectPads2 * pads, GstCollectData2 * data) +{ + GstBuffer *result; + + g_return_val_if_fail (pads != NULL, NULL); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), NULL); + g_return_val_if_fail (data != NULL, NULL); + + if ((result = data->buffer)) { + data->buffer = NULL; + data->pos = 0; + /* one less pad with queued data now */ + if (GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_WAITING)) + pads->queuedpads--; + } + + GST_COLLECT_PADS2_EVT_BROADCAST (pads); + + GST_DEBUG_OBJECT (pads, "Pop buffer on pad %s:%s: buffer=%p", + GST_DEBUG_PAD_NAME (data->pad), result); + + return result; +} + +/* pop and unref the currently queued buffer, should be called with STREAM_LOCK + * held */ +static void +gst_collect_pads2_clear (GstCollectPads2 * pads, GstCollectData2 * data) +{ + GstBuffer *buf; + + if ((buf = gst_collect_pads2_pop (pads, data))) + gst_buffer_unref (buf); +} + +/** + * gst_collect_pads2_available: + * @pads: the collectspads to query + * + * Query how much bytes can be read from each queued buffer. This means + * that the result of this call is the maximum number of bytes that can + * be read from each of the pads. + * + * This function should be called with @pads STREAM_LOCK held, such as + * in the callback. + * + * Returns: The maximum number of bytes queued on all pads. This function + * returns 0 if a pad has no queued buffer. + * + * MT safe. + * + * Since: 0.10.36 + */ +/* we might pre-calculate this in some struct field, + * but would then have to maintain this in _chain and particularly _pop, etc, + * even if element is never interested in this information */ +guint +gst_collect_pads2_available (GstCollectPads2 * pads) +{ + GSList *collected; + guint result = G_MAXUINT; + + g_return_val_if_fail (pads != NULL, 0); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), 0); + + collected = pads->data; + for (; collected; collected = g_slist_next (collected)) { + GstCollectData2 *pdata; + GstBuffer *buffer; + gint size; + + pdata = (GstCollectData2 *) collected->data; + + /* ignore pad with EOS */ + if (G_UNLIKELY (GST_COLLECT_PADS2_STATE_IS_SET (pdata, + GST_COLLECT_PADS2_STATE_EOS))) { + GST_DEBUG_OBJECT (pads, "pad %p is EOS", pdata); + continue; + } + + /* an empty buffer without EOS is weird when we get here.. */ + if (G_UNLIKELY ((buffer = pdata->buffer) == NULL)) { + GST_WARNING_OBJECT (pads, "pad %p has no buffer", pdata); + goto not_filled; + } + + /* this is the size left of the buffer */ + size = GST_BUFFER_SIZE (buffer) - pdata->pos; + GST_DEBUG_OBJECT (pads, "pad %p has %d bytes left", pdata, size); + + /* need to return the min of all available data */ + if (size < result) + result = size; + } + /* nothing changed, all must be EOS then, return 0 */ + if (G_UNLIKELY (result == G_MAXUINT)) + result = 0; + + return result; + +not_filled: + { + return 0; + } +} + +/** + * gst_collect_pads2_read: + * @pads: the collectspads to query + * @data: the data to use + * @bytes: a pointer to a byte array + * @size: the number of bytes to read + * + * Get a pointer in @bytes where @size bytes can be read from the + * given pad data. + * + * This function should be called with @pads STREAM_LOCK held, such as + * in the callback. + * + * Returns: The number of bytes available for consumption in the + * memory pointed to by @bytes. This can be less than @size and + * is 0 if the pad is end-of-stream. + * + * MT safe. + * + * Since: 0.10.36 + */ +guint +gst_collect_pads2_read (GstCollectPads2 * pads, GstCollectData2 * data, + guint8 ** bytes, guint size) +{ + guint readsize; + GstBuffer *buffer; + + g_return_val_if_fail (pads != NULL, 0); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), 0); + g_return_val_if_fail (data != NULL, 0); + g_return_val_if_fail (bytes != NULL, 0); + + /* no buffer, must be EOS */ + if ((buffer = data->buffer) == NULL) + return 0; + + readsize = MIN (size, GST_BUFFER_SIZE (buffer) - data->pos); + + *bytes = GST_BUFFER_DATA (buffer) + data->pos; + + return readsize; +} + +/** + * gst_collect_pads2_flush: + * @pads: the collectspads to query + * @data: the data to use + * @size: the number of bytes to flush + * + * Flush @size bytes from the pad @data. + * + * This function should be called with @pads STREAM_LOCK held, such as + * in the callback. + * + * Returns: The number of bytes flushed This can be less than @size and + * is 0 if the pad was end-of-stream. + * + * MT safe. + * + * Since: 0.10.36 + */ +guint +gst_collect_pads2_flush (GstCollectPads2 * pads, GstCollectData2 * data, + guint size) +{ + guint flushsize; + GstBuffer *buffer; + + g_return_val_if_fail (pads != NULL, 0); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), 0); + g_return_val_if_fail (data != NULL, 0); + + /* no buffer, must be EOS */ + if ((buffer = data->buffer) == NULL) + return 0; + + /* this is what we can flush at max */ + flushsize = MIN (size, GST_BUFFER_SIZE (buffer) - data->pos); + + data->pos += size; + + if (data->pos >= GST_BUFFER_SIZE (buffer)) + /* _clear will also reset data->pos to 0 */ + gst_collect_pads2_clear (pads, data); + + return flushsize; +} + +/** + * gst_collect_pads2_read_buffer: + * @pads: the collectspads to query + * @data: the data to use + * @size: the number of bytes to read + * + * Get a subbuffer of @size bytes from the given pad @data. + * + * This function should be called with @pads STREAM_LOCK held, such as in the + * callback. + * + * Since: 0.10.36 + * + * Returns: A sub buffer. The size of the buffer can be less that requested. + * A return of NULL signals that the pad is end-of-stream. + * Unref the buffer after use. + * + * MT safe. + */ +GstBuffer * +gst_collect_pads2_read_buffer (GstCollectPads2 * pads, GstCollectData2 * data, + guint size) +{ + guint readsize; + GstBuffer *buffer; + + g_return_val_if_fail (pads != NULL, NULL); + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), NULL); + g_return_val_if_fail (data != NULL, NULL); + + /* no buffer, must be EOS */ + if ((buffer = data->buffer) == NULL) + return NULL; + + readsize = MIN (size, GST_BUFFER_SIZE (buffer) - data->pos); + + return gst_buffer_create_sub (buffer, data->pos, readsize); +} + +/** + * gst_collect_pads2_take_buffer: + * @pads: the collectspads to query + * @data: the data to use + * @size: the number of bytes to read + * + * Get a subbuffer of @size bytes from the given pad @data. Flushes the amount + * of read bytes. + * + * This function should be called with @pads STREAM_LOCK held, such as in the + * callback. + * + * Since: 0.10.36 + * + * Returns: A sub buffer. The size of the buffer can be less that requested. + * A return of NULL signals that the pad is end-of-stream. + * Unref the buffer after use. + * + * MT safe. + */ +GstBuffer * +gst_collect_pads2_take_buffer (GstCollectPads2 * pads, GstCollectData2 * data, + guint size) +{ + GstBuffer *buffer = gst_collect_pads2_read_buffer (pads, data, size); + + if (buffer) { + gst_collect_pads2_flush (pads, data, GST_BUFFER_SIZE (buffer)); + } + return buffer; +} + +/** + * gst_collect_pads2_set_waiting: + * @pads: the collectspads + * @data: the data to use + * @waiting: boolean indicating whether this pad should operate + * in waiting or non-waiting mode + * + * Sets a pad to waiting or non-waiting mode, if at least this pad + * has not been created with locked waiting state, + * in which case nothing happens. + * + * This function should be called with @pads STREAM_LOCK held, such as + * in the callback. + * + * MT safe. + * + * Since: 0.10.36 + */ +void +gst_collect_pads2_set_waiting (GstCollectPads2 * pads, GstCollectData2 * data, + gboolean waiting) +{ + g_return_if_fail (pads != NULL); + g_return_if_fail (GST_IS_COLLECT_PADS2 (pads)); + g_return_if_fail (data != NULL); + + GST_DEBUG_OBJECT (pads, "Setting pad %s to waiting %d, locked %d", + GST_PAD_NAME (data->pad), waiting, + GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_LOCKED)); + + /* Do something only on a change and if not locked */ + if (!GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_LOCKED) && + (GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_WAITING) != + ! !waiting)) { + /* Set waiting state for this pad */ + if (waiting) + GST_COLLECT_PADS2_STATE_SET (data, GST_COLLECT_PADS2_STATE_WAITING); + else + GST_COLLECT_PADS2_STATE_UNSET (data, GST_COLLECT_PADS2_STATE_WAITING); + /* Update number of queued pads if needed */ + if (!data->buffer && + !GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_EOS)) { + if (waiting) + pads->queuedpads--; + else + pads->queuedpads++; + } + + /* signal waiters because something changed */ + GST_COLLECT_PADS2_EVT_BROADCAST (pads); + } +} + +/* see if pads were added or removed and update our stats. Any pad + * added after releasing the LOCK will get collected in the next + * round. + * + * We can do a quick check by checking the cookies, that get changed + * whenever the pad list is updated. + * + * Must be called with STREAM_LOCK. + */ +static void +gst_collect_pads2_check_pads (GstCollectPads2 * pads) +{ + /* the master list and cookie are protected with LOCK */ + GST_OBJECT_LOCK (pads); + if (G_UNLIKELY (pads->pad_cookie != pads->cookie)) { + GSList *collected; + + /* clear list and stats */ + g_slist_foreach (pads->data, (GFunc) unref_data, NULL); + g_slist_free (pads->data); + pads->data = NULL; + pads->numpads = 0; + pads->queuedpads = 0; + pads->eospads = 0; + if (pads->earliest_data) + unref_data (pads->earliest_data); + pads->earliest_data = NULL; + pads->earliest_time = GST_CLOCK_TIME_NONE; + + /* loop over the master pad list */ + collected = pads->pad_list; + for (; collected; collected = g_slist_next (collected)) { + GstCollectData2 *data; + + /* update the stats */ + pads->numpads++; + data = collected->data; + if (GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_EOS)) + pads->eospads++; + else if (data->buffer || !GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_WAITING)) + pads->queuedpads++; + + /* add to the list of pads to collect */ + ref_data (data); + /* preserve order of adding/requesting pads */ + pads->data = g_slist_append (pads->data, data); + } + /* and update the cookie */ + pads->cookie = pads->pad_cookie; + } + GST_OBJECT_UNLOCK (pads); +} + +/* checks if all the pads are collected and call the collectfunction + * + * Should be called with STREAM_LOCK. + * + * Returns: The #GstFlowReturn of collection. + */ +static GstFlowReturn +gst_collect_pads2_check_collected (GstCollectPads2 * pads) +{ + GstFlowReturn flow_ret = GST_FLOW_OK; + GstCollectPads2Function func; + gpointer user_data; + + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), GST_FLOW_ERROR); + + GST_OBJECT_LOCK (pads); + func = pads->func; + user_data = pads->user_data; + GST_OBJECT_UNLOCK (pads); + + g_return_val_if_fail (pads->func != NULL, GST_FLOW_NOT_SUPPORTED); + + /* check for new pads, update stats etc.. */ + gst_collect_pads2_check_pads (pads); + + if (G_UNLIKELY (pads->eospads == pads->numpads)) { + /* If all our pads are EOS just collect once to let the element + * do its final EOS handling. */ + GST_DEBUG_OBJECT (pads, "All active pads (%d) are EOS, calling %s", + pads->numpads, GST_DEBUG_FUNCPTR_NAME (func)); + + flow_ret = func (pads, user_data); + } else { + gboolean collected = FALSE; + + /* We call the collected function as long as our condition matches. */ + while (((pads->queuedpads + pads->eospads) >= pads->numpads)) { + GST_DEBUG_OBJECT (pads, "All active pads (%d + %d >= %d) have data, " + "calling %s", pads->queuedpads, pads->eospads, pads->numpads, + GST_DEBUG_FUNCPTR_NAME (func)); + + flow_ret = func (pads, user_data); + collected = TRUE; + + /* break on error */ + if (flow_ret != GST_FLOW_OK) + break; + /* Don't keep looping after telling the element EOS or flushing */ + if (pads->queuedpads == 0) + break; + } + if (!collected) + GST_DEBUG_OBJECT (pads, "Not all active pads (%d) have data, continuing", + pads->numpads); + } + return flow_ret; +} + + +/* General overview: + * - only pad with a buffer can determine earliest_data (and earliest_time) + * - only segment info determines (non-)waiting state + * - ? perhaps use _stream_time for comparison + * (which muxers might have use as well ?) + */ + +/* + * Function to recalculate the waiting state of all pads. + * + * Must be called with STREAM_LOCK. + * + * Returns TRUE if a pad was set to waiting + * (from non-waiting state). + */ +static gboolean +gst_collect_pads2_recalculate_waiting (GstCollectPads2 * pads) +{ + GSList *collected; + gboolean result = FALSE; + + /* If earliest time is not known, there is nothing to do. */ + if (pads->earliest_data == NULL) + return FALSE; + + for (collected = pads->data; collected; collected = g_slist_next (collected)) { + GstCollectData2 *data = (GstCollectData2 *) collected->data; + int cmp_res; + + /* check if pad has a segment */ + if (data->segment.format == GST_FORMAT_UNDEFINED) + continue; + + /* check segment format */ + if (data->segment.format != GST_FORMAT_TIME) { + GST_ERROR_OBJECT (pads, "GstCollectPads2 can handle only time segments."); + continue; + } + + /* check if the waiting state should be changed */ + cmp_res = pads->compare_func (pads, data, data->segment.start, + pads->earliest_data, pads->earliest_time, pads->compare_user_data); + if (cmp_res > 0) + /* stop waiting */ + gst_collect_pads2_set_waiting (pads, data, FALSE); + else { + if (!GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_WAITING)) { + /* start waiting */ + gst_collect_pads2_set_waiting (pads, data, TRUE); + result = TRUE; + } + } + } + + return result; +} + +/** + * gst_collect_pads2_find_best_pad: + * @pads: the collectpads to use + * @data: returns the collectdata for earliest data + * @time: returns the earliest available buffertime + * + * Find the oldest/best pad, i.e. pad holding the oldest buffer and + * and return the corresponding #GstCollectData2 and buffertime. + * + * This function should be called with STREAM_LOCK held, + * such as in the callback. + * + * Since: 0.10.36 + */ +static void +gst_collect_pads2_find_best_pad (GstCollectPads2 * pads, + GstCollectData2 ** data, GstClockTime * time) +{ + GSList *collected; + GstCollectData2 *best = NULL; + GstClockTime best_time = GST_CLOCK_TIME_NONE; + + g_return_if_fail (data != NULL); + g_return_if_fail (time != NULL); + + for (collected = pads->data; collected; collected = g_slist_next (collected)) { + GstBuffer *buffer; + GstCollectData2 *data = (GstCollectData2 *) collected->data; + GstClockTime timestamp; + + buffer = gst_collect_pads2_peek (pads, data); + /* if we have a buffer check if it is better then the current best one */ + if (buffer != NULL) { + timestamp = GST_BUFFER_TIMESTAMP (buffer); + gst_buffer_unref (buffer); + if (best == NULL || pads->compare_func (pads, data, timestamp, + best, best_time, pads->compare_user_data) < 0) { + best = data; + best_time = timestamp; + } + } + } + + /* set earliest time */ + *data = best; + *time = best_time; + + GST_DEBUG_OBJECT (pads, "best pad %s, best time %" GST_TIME_FORMAT, + best ? GST_PAD_NAME (((GstCollectData2 *) best)->pad) : "(nil)", + GST_TIME_ARGS (best_time)); +} + +/* + * Function to recalculate earliest_data and earliest_timestamp. This also calls + * gst_collect_pads2_recalculate_waiting + * + * Must be called with STREAM_LOCK. + */ +static gboolean +gst_collect_pads2_recalculate_full (GstCollectPads2 * pads) +{ + if (pads->earliest_data) + unref_data (pads->earliest_data); + gst_collect_pads2_find_best_pad (pads, &pads->earliest_data, + &pads->earliest_time); + if (pads->earliest_data) + ref_data (pads->earliest_data); + return gst_collect_pads2_recalculate_waiting (pads); +} + +/* + * Default collect callback triggered when #GstCollectPads2 gathered all data. + * + * Called with STREAM_LOCK. + */ +static GstFlowReturn +gst_collect_pads2_default_collected (GstCollectPads2 * pads, gpointer user_data) +{ + GstCollectData2 *best = NULL; + GstBuffer *buffer; + GstFlowReturn ret = GST_FLOW_OK; + GstCollectPads2BufferFunction func; + gpointer buffer_user_data; + + g_return_val_if_fail (GST_IS_COLLECT_PADS2 (pads), GST_FLOW_ERROR); + + GST_OBJECT_LOCK (pads); + func = pads->buffer_func; + buffer_user_data = pads->buffer_user_data; + GST_OBJECT_UNLOCK (pads); + + g_return_val_if_fail (func != NULL, GST_FLOW_NOT_SUPPORTED); + + /* Find the oldest pad at all cost */ + if (gst_collect_pads2_recalculate_full (pads)) { + /* waiting was switched on, + * so give another thread a chance to deliver a possibly + * older buffer; don't charge on yet with the current oldest */ + ret = GST_FLOW_OK; + } + + best = pads->earliest_data; + + /* No data collected means EOS. */ + if (G_UNLIKELY (best == NULL)) { + ret = func (pads, best, NULL, buffer_user_data); + if (ret == GST_FLOW_OK) + ret = GST_FLOW_UNEXPECTED; + goto done; + } + + /* make sure that the pad we take a buffer from is waiting; + * otherwise popping a buffer will seem not to have happened + * and collectpads can get into a busy loop */ + gst_collect_pads2_set_waiting (pads, best, TRUE); + + /* Send buffer */ + buffer = gst_collect_pads2_pop (pads, best); + ret = func (pads, best, buffer, buffer_user_data); + +done: + return ret; +} + +/* + * Default timestamp compare function. + */ +static gint +gst_collect_pads2_default_compare_func (GstCollectPads2 * pads, + GstCollectData2 * data1, GstClockTime timestamp1, + GstCollectData2 * data2, GstClockTime timestamp2, gpointer user_data) +{ + + GST_LOG_OBJECT (pads, "comparing %" GST_TIME_FORMAT + " and %" GST_TIME_FORMAT, GST_TIME_ARGS (timestamp1), + GST_TIME_ARGS (timestamp2)); + /* non-valid timestamps go first as they are probably headers or so */ + if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (timestamp1))) + return GST_CLOCK_TIME_IS_VALID (timestamp2) ? -1 : 0; + + if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (timestamp2))) + return 1; + + /* compare timestamp */ + if (timestamp1 < timestamp2) + return -1; + + if (timestamp1 > timestamp2) + return 1; + + return 0; +} + +static gboolean +gst_collect_pads2_event (GstPad * pad, GstEvent * event) +{ + gboolean res = FALSE, need_unlock = FALSE; + GstCollectData2 *data; + GstCollectPads2 *pads; + GstCollectPads2EventFunction event_func; + GstCollectPads2BufferFunction buffer_func; + gpointer event_user_data; + + /* some magic to get the managing collect_pads2 */ + GST_OBJECT_LOCK (pad); + data = (GstCollectData2 *) gst_pad_get_element_private (pad); + if (G_UNLIKELY (data == NULL)) + goto pad_removed; + ref_data (data); + GST_OBJECT_UNLOCK (pad); + + res = FALSE; + + pads = data->collect; + + GST_DEBUG ("Got %s event on pad %s:%s", GST_EVENT_TYPE_NAME (event), + GST_DEBUG_PAD_NAME (data->pad)); + + GST_OBJECT_LOCK (pads); + event_func = pads->event_func; + event_user_data = pads->event_user_data; + buffer_func = pads->buffer_func; + GST_OBJECT_UNLOCK (pads); + + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_FLUSH_START: + { + /* forward event to unblock check_collected */ + if (event_func) + res = event_func (pads, data, event, event_user_data); + if (!res) + res = gst_pad_event_default (pad, event); + + /* now unblock the chain function. + * no cond per pad, so they all unblock, + * non-flushing block again */ + GST_COLLECT_PADS2_STREAM_LOCK (pads); + GST_COLLECT_PADS2_STATE_SET (data, GST_COLLECT_PADS2_STATE_FLUSHING); + gst_collect_pads2_clear (pads, data); + + /* cater for possible default muxing functionality */ + if (buffer_func) { + /* restore to initial state */ + gst_collect_pads2_set_waiting (pads, data, TRUE); + /* if the current pad is affected, reset state, recalculate later */ + if (pads->earliest_data == data) { + unref_data (data); + pads->earliest_data = NULL; + pads->earliest_time = GST_CLOCK_TIME_NONE; + } + } + + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + + /* event already cleaned up by forwarding */ + res = TRUE; + goto done; + } + case GST_EVENT_FLUSH_STOP: + { + /* flush the 1 buffer queue */ + GST_COLLECT_PADS2_STREAM_LOCK (pads); + GST_COLLECT_PADS2_STATE_UNSET (data, GST_COLLECT_PADS2_STATE_FLUSHING); + gst_collect_pads2_clear (pads, data); + /* we need new segment info after the flush */ + gst_segment_init (&data->segment, GST_FORMAT_UNDEFINED); + GST_COLLECT_PADS2_STATE_UNSET (data, GST_COLLECT_PADS2_STATE_NEW_SEGMENT); + /* if the pad was EOS, remove the EOS flag and + * decrement the number of eospads */ + if (G_UNLIKELY (GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_EOS))) { + if (!GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_WAITING)) + pads->queuedpads++; + pads->eospads--; + GST_COLLECT_PADS2_STATE_UNSET (data, GST_COLLECT_PADS2_STATE_EOS); + } + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + + /* forward event */ + goto forward_or_default; + } + case GST_EVENT_EOS: + { + GST_COLLECT_PADS2_STREAM_LOCK (pads); + /* if the pad was not EOS, make it EOS and so we + * have one more eospad */ + if (G_LIKELY (!GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_EOS))) { + GST_COLLECT_PADS2_STATE_SET (data, GST_COLLECT_PADS2_STATE_EOS); + if (!GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_WAITING)) + pads->queuedpads--; + pads->eospads++; + } + /* check if we need collecting anything, we ignore the result. */ + gst_collect_pads2_check_collected (pads); + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + + goto forward_or_eat; + } + case GST_EVENT_NEWSEGMENT: + { + gint64 start, stop, time; + gdouble rate, arate; + GstFormat format; + gboolean update; + gint cmp_res; + + GST_COLLECT_PADS2_STREAM_LOCK (pads); + + gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format, + &start, &stop, &time); + + GST_DEBUG_OBJECT (data->pad, "got newsegment, start %" GST_TIME_FORMAT + ", stop %" GST_TIME_FORMAT, GST_TIME_ARGS (start), + GST_TIME_ARGS (stop)); + + gst_segment_set_newsegment_full (&data->segment, update, rate, arate, + format, start, stop, time); + + GST_COLLECT_PADS2_STATE_SET (data, GST_COLLECT_PADS2_STATE_NEW_SEGMENT); + + /* default muxing functionality */ + if (!buffer_func) + goto newsegment_done; + + /* default collection can not handle other segment formats than time */ + if (format != GST_FORMAT_TIME) { + GST_ERROR_OBJECT (pads, "GstCollectPads2 default collecting " + "can only handle time segments."); + goto newsegment_done; + } + + /* If oldest time is not known, or current pad got newsegment; + * recalculate the state */ + if (!pads->earliest_data || pads->earliest_data == data) { + gst_collect_pads2_recalculate_full (pads); + goto newsegment_done; + } + + /* Check if the waiting state of the pad should change. */ + cmp_res = pads->compare_func (pads, data, start, pads->earliest_data, + pads->earliest_time, pads->compare_user_data); + + if (cmp_res > 0) + /* Stop waiting */ + gst_collect_pads2_set_waiting (pads, data, FALSE); + + newsegment_done: + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + /* we must not forward this event since multiple segments will be + * accumulated and this is certainly not what we want. */ + goto forward_or_eat; + } + default: + /* forward other events */ + goto forward_or_default; + } + +forward_or_default: + if (GST_EVENT_IS_SERIALIZED (event)) { + GST_COLLECT_PADS2_STREAM_LOCK (pads); + need_unlock = TRUE; + } + if (event_func) + res = event_func (pads, data, event, event_user_data); + if (!res) + res = gst_pad_event_default (pad, event); + if (need_unlock) + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + goto done; + +forward_or_eat: + if (GST_EVENT_IS_SERIALIZED (event)) { + GST_COLLECT_PADS2_STREAM_LOCK (pads); + need_unlock = TRUE; + } + if (event_func) + res = event_func (pads, data, event, event_user_data); + if (!res) { + gst_event_unref (event); + res = TRUE; + } + if (need_unlock) + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + goto done; + +done: + unref_data (data); + return res; + + /* ERRORS */ +pad_removed: + { + GST_DEBUG ("%s got removed from collectpads", GST_OBJECT_NAME (pad)); + GST_OBJECT_UNLOCK (pad); + return FALSE; + } +} + +/* For each buffer we receive we check if our collected condition is reached + * and if so we call the collected function. When this is done we check if + * data has been unqueued. If data is still queued we wait holding the stream + * lock to make sure no EOS event can happen while we are ready to be + * collected + */ +static GstFlowReturn +gst_collect_pads2_chain (GstPad * pad, GstBuffer * buffer) +{ + GstCollectData2 *data; + GstCollectPads2 *pads; + GstFlowReturn ret; + GstBuffer **buffer_p; + guint32 cookie; + GstCollectPads2BufferFunction prepare_buffer_func; + gpointer prepare_buffer_user_data; + + GST_DEBUG ("Got buffer for pad %s:%s", GST_DEBUG_PAD_NAME (pad)); + + /* some magic to get the managing collect_pads2 */ + GST_OBJECT_LOCK (pad); + data = (GstCollectData2 *) gst_pad_get_element_private (pad); + if (G_UNLIKELY (data == NULL)) + goto no_data; + ref_data (data); + GST_OBJECT_UNLOCK (pad); + + pads = data->collect; + GST_OBJECT_LOCK (pads); + prepare_buffer_func = pads->prepare_buffer_func; + prepare_buffer_user_data = pads->prepare_buffer_user_data; + GST_OBJECT_UNLOCK (pads); + + GST_COLLECT_PADS2_STREAM_LOCK (pads); + /* if not started, bail out */ + if (G_UNLIKELY (!pads->started)) + goto not_started; + /* check if this pad is flushing */ + if (G_UNLIKELY (GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_FLUSHING))) + goto flushing; + /* pad was EOS, we can refuse this data */ + if (G_UNLIKELY (GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_EOS))) + goto unexpected; + + GST_DEBUG_OBJECT (pads, "Queuing buffer %p for pad %s:%s", buffer, + GST_DEBUG_PAD_NAME (pad)); + + if (prepare_buffer_func) { + ret = prepare_buffer_func (pads, data, buffer, prepare_buffer_user_data); + if (ret == GST_COLLECT_PADS2_FLOW_DROP) { + GST_DEBUG_OBJECT (pads, "Dropping buffer as requested"); + ret = GST_FLOW_OK; + goto unlock_done; + } else if (ret == GST_FLOW_UNEXPECTED) { + goto unexpected; + } else if (ret != GST_FLOW_OK) { + goto error; + } + } + + /* One more pad has data queued */ + if (GST_COLLECT_PADS2_STATE_IS_SET (data, GST_COLLECT_PADS2_STATE_WAITING)) + pads->queuedpads++; + buffer_p = &data->buffer; + gst_buffer_replace (buffer_p, buffer); + + /* update segment last position if in TIME */ + if (G_LIKELY (data->segment.format == GST_FORMAT_TIME)) { + GstClockTime timestamp = GST_BUFFER_TIMESTAMP (buffer); + + if (GST_CLOCK_TIME_IS_VALID (timestamp)) + gst_segment_set_last_stop (&data->segment, GST_FORMAT_TIME, timestamp); + } + + /* While we have data queued on this pad try to collect stuff */ + do { + /* Check if our collected condition is matched and call the collected + * function if it is */ + ret = gst_collect_pads2_check_collected (pads); + /* when an error occurs, we want to report this back to the caller ASAP + * without having to block if the buffer was not popped */ + if (G_UNLIKELY (ret != GST_FLOW_OK)) + goto error; + + /* data was consumed, we can exit and accept new data */ + if (data->buffer == NULL) + break; + + /* Having the _INIT here means we don't care about any broadcast up to here + * (most of which occur with STREAM_LOCK held, so could not have happened + * anyway). We do care about e.g. a remove initiated broadcast as of this + * point. Putting it here also makes this thread ignores any evt it raised + * itself (as is a usual WAIT semantic). + */ + GST_COLLECT_PADS2_EVT_INIT (cookie); + + /* pad could be removed and re-added */ + unref_data (data); + GST_OBJECT_LOCK (pad); + if (G_UNLIKELY ((data = gst_pad_get_element_private (pad)) == NULL)) + goto pad_removed; + ref_data (data); + GST_OBJECT_UNLOCK (pad); + + GST_DEBUG_OBJECT (pads, "Pad %s:%s has a buffer queued, waiting", + GST_DEBUG_PAD_NAME (pad)); + + /* wait to be collected, this must happen from another thread triggered + * by the _chain function of another pad. We release the lock so we + * can get stopped or flushed as well. We can however not get EOS + * because we still hold the STREAM_LOCK. + */ + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + GST_COLLECT_PADS2_EVT_WAIT (pads, cookie); + GST_COLLECT_PADS2_STREAM_LOCK (pads); + + GST_DEBUG_OBJECT (pads, "Pad %s:%s resuming", GST_DEBUG_PAD_NAME (pad)); + + /* after a signal, we could be stopped */ + if (G_UNLIKELY (!pads->started)) + goto not_started; + /* check if this pad is flushing */ + if (G_UNLIKELY (GST_COLLECT_PADS2_STATE_IS_SET (data, + GST_COLLECT_PADS2_STATE_FLUSHING))) + goto flushing; + } + while (data->buffer != NULL); + +unlock_done: + GST_COLLECT_PADS2_STREAM_UNLOCK (pads); + unref_data (data); + gst_buffer_unref (buffer); + return ret; + +pad_removed: + { + GST_WARNING ("%s got removed from collectpads", GST_OBJECT_NAME (pad)); + GST_OBJECT_UNLOCK (pad); + ret = GST_FLOW_NOT_LINKED; + goto unlock_done; + } + /* ERRORS */ +no_data: + { + GST_DEBUG ("%s got removed from collectpads", GST_OBJECT_NAME (pad)); + GST_OBJECT_UNLOCK (pad); + gst_buffer_unref (buffer); + return GST_FLOW_NOT_LINKED; + } +not_started: + { + GST_DEBUG ("not started"); + gst_collect_pads2_clear (pads, data); + ret = GST_FLOW_WRONG_STATE; + goto unlock_done; + } +flushing: + { + GST_DEBUG ("pad %s:%s is flushing", GST_DEBUG_PAD_NAME (pad)); + gst_collect_pads2_clear (pads, data); + ret = GST_FLOW_WRONG_STATE; + goto unlock_done; + } +unexpected: + { + /* we should not post an error for this, just inform upstream that + * we don't expect anything anymore */ + GST_DEBUG ("pad %s:%s is eos", GST_DEBUG_PAD_NAME (pad)); + ret = GST_FLOW_UNEXPECTED; + goto unlock_done; + } +error: + { + /* we print the error, the element should post a reasonable error + * message for fatal errors */ + GST_DEBUG ("collect failed, reason %d (%s)", ret, gst_flow_get_name (ret)); + gst_collect_pads2_clear (pads, data); + goto unlock_done; + } +} diff --git a/libs/gst/base/gstcollectpads2.h b/libs/gst/base/gstcollectpads2.h new file mode 100644 index 0000000..bb3b921 --- /dev/null +++ b/libs/gst/base/gstcollectpads2.h @@ -0,0 +1,362 @@ +/* GStreamer + * Copyright (C) 2005 Wim Taymans + * Copyright (C) 2008 Mark Nauwelaerts + * + * gstcollectpads2.h: + * + * 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., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +#ifndef __GST_COLLECT_PADS2_H__ +#define __GST_COLLECT_PADS2_H__ + +#include + +G_BEGIN_DECLS + +#define GST_TYPE_COLLECT_PADS2 (gst_collect_pads2_get_type()) +#define GST_COLLECT_PADS2(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_COLLECT_PADS2,GstCollectPads2)) +#define GST_COLLECT_PADS2_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_COLLECT_PADS2,GstCollectPads2Class)) +#define GST_COLLECT_PADS2_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS ((obj),GST_TYPE_COLLECT_PADS2,GstCollectPads2Class)) +#define GST_IS_COLLECT_PADS2(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_COLLECT_PADS2)) +#define GST_IS_COLLECT_PADS2_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_COLLECT_PADS2)) + +typedef struct _GstCollectData2 GstCollectData2; +typedef struct _GstCollectPads2 GstCollectPads2; +typedef struct _GstCollectPads2Class GstCollectPads2Class; + +/** + * GstCollectData2DestroyNotify: + * @data: the #GstCollectData2 that will be freed + * + * A function that will be called when the #GstCollectData2 will be freed. + * It is passed the pointer to the structure and should free any custom + * memory and resources allocated for it. + * + * Since: 0.10.36 + */ +typedef void (*GstCollectData2DestroyNotify) (GstCollectData2 *data); + +/** + * GstCollectPads2StateFlags: + * @GST_COLLECT_PADS2_STATE_EOS: Set if collectdata's pad is EOS. + * @GST_COLLECT_PADS2_STATE_FLUSHING: Set if collectdata's pad is flushing. + * @GST_COLLECT_PADS2_STATE_NEW_SEGMENT: Set if collectdata's pad received a + * new_segment event. + * @GST_COLLECT_PADS2_STATE_WAITING: Set if collectdata's pad must be waited + * for when collecting. + * @GST_COLLECT_PADS2_STATE_LOCKED: Set collectdata's pad WAITING state must + * not be changed. + * #GstCollectPads2StateFlags indicate private state of a collectdata('s pad). + * + * Since: 0.10.36 + */ +typedef enum { + GST_COLLECT_PADS2_STATE_EOS = 1 << 0, + GST_COLLECT_PADS2_STATE_FLUSHING = 1 << 1, + GST_COLLECT_PADS2_STATE_NEW_SEGMENT = 1 << 2, + GST_COLLECT_PADS2_STATE_WAITING = 1 << 3, + GST_COLLECT_PADS2_STATE_LOCKED = 1 << 4 +} GstCollectPads2StateFlags; + +/** + * GST_COLLECT_PADS2_STATE: + * @data: a #GstCollectData2. + * + * A flags word containing #GstCollectPads2StateFlags flags set + * on this collected pad. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STATE(data) (((GstCollectData2 *) data)->state) +/** + * GST_COLLECT_PADS2_STATE_IS_SET: + * @data: a #GstCollectData2. + * @flag: the #GstCollectPads2StateFlags to check. + * + * Gives the status of a specific flag on a collected pad. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STATE_IS_SET(data,flag) !!(GST_COLLECT_PADS2_STATE (data) & flag) +/** + * GST_COLLECT_PADS2_STATE_SET: + * @data: a #GstCollectData2. + * @flag: the #GstCollectPads2StateFlags to set. + * + * Sets a state flag on a collected pad. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STATE_SET(data,flag) (GST_COLLECT_PADS2_STATE (data) |= flag) +/** + * GST_COLLECT_PADS2_STATE_UNSET: + * @data: a #GstCollectData2. + * @flag: the #GstCollectPads2StateFlags to clear. + * + * Clears a state flag on a collected pad. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STATE_UNSET(data,flag) (GST_COLLECT_PADS2_STATE (data) &= ~(flag)) + +#define GST_COLLECT_PADS2_FLOW_DROP GST_FLOW_CUSTOM_SUCCESS + +/** + * GstCollectData2: + * @collect: owner #GstCollectPads2 + * @pad: #GstPad managed by this data + * @buffer: currently queued buffer. + * @pos: position in the buffer + * @segment: last segment received. + * + * Structure used by the collect_pads2. + * + * Since: 0.10.36 + */ +struct _GstCollectData2 +{ + /* with STREAM_LOCK of @collect */ + GstCollectPads2 *collect; + GstPad *pad; + GstBuffer *buffer; + guint pos; + GstSegment segment; + + /*< private >*/ + /* state: bitfield for easier extension; + * eos, flushing, new_segment, waiting */ + guint state; + + /* refcounting for struct, and destroy callback */ + GstCollectData2DestroyNotify destroy_notify; + gint refcount; + + gpointer _gst_reserved[GST_PADDING]; +}; + +/** + * GstCollectPads2Function: + * @pads: the #GstCollectPads2 that trigered the callback + * @user_data: user data passed to gst_collect_pads2_set_function() + * + * A function that will be called when all pads have received data. + * + * Returns: #GST_FLOW_OK for success + * + * Since: 0.10.36 + */ +typedef GstFlowReturn (*GstCollectPads2Function) (GstCollectPads2 *pads, gpointer user_data); + +/** + * GstCollectPads2BufferFunction: + * @pads: the #GstCollectPads2 that trigered the callback + * @data: the #GstCollectData2 of pad that has received the buffer + * @buffer: the #GstBuffer + * @user_data: user data passed to gst_collect_pads2_set_buffer_function() + * + * A function that will be called when a (considered oldest) buffer can be muxed. + * If all pads have reached EOS, this function is called with NULL @buffer + * and NULL @data. + * + * Returns: #GST_FLOW_OK for success + * + * Since: 0.10.36 + */ +typedef GstFlowReturn (*GstCollectPads2BufferFunction) (GstCollectPads2 *pads, GstCollectData2 *data, + GstBuffer *buffer, gpointer user_data); + +/** + * GstCollectPads2CompareFunction: + * @pads: the #GstCollectPads that is comparing the timestamps + * @data1: the first #GstCollectData2 + * @timestamp1: the first timestamp + * @data2: the second #GstCollectData2 + * @timestamp2: the second timestamp + * @user_data: user data passed to gst_collect_pads2_set_compare_function() + * + * A function for comparing two timestamps of buffers or newsegments collected on one pad. + * + * Returns: Integer less than zero when first timestamp is deemed older than the second one. + * Zero if the timestamps are deemed equally old. + * Integer greate than zero when second timestamp is deemed older than the first one. + * + * Since: 0.10.36 + */ +typedef gint (*GstCollectPads2CompareFunction) (GstCollectPads2 *pads, + GstCollectData2 * data1, GstClockTime timestamp1, + GstCollectData2 * data2, GstClockTime timestamp2, + gpointer user_data); + +/** + * GstCollectPads2EventFunction: + * @pads: the #GstCollectPads2 that trigered the callback + * @pad: the #GstPad that received an event + * @event: the #GstEvent received + * @user_data: user data passed to gst_collect_pads2_set_event_function() + * + * A function that will be called after collectpads has processed the event. + * + * Returns: %TRUE if the pad could handle the event + * + * Since: 0.10.36 + */ +typedef gboolean (*GstCollectPads2EventFunction) (GstCollectPads2 *pads, GstCollectData2 * pad, + GstEvent * event, gpointer user_data); + +/** + * GST_COLLECT_PADS2_GET_STREAM_LOCK: + * @pads: a #GstCollectPads2 + * + * Get the stream lock of @pads. The stream lock is used to coordinate and + * serialize execution among the various streams being collected, and in + * protecting the resources used to accomplish this. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_GET_STREAM_LOCK(pads) (&((GstCollectPads2 *)pads)->stream_lock) +/** + * GST_COLLECT_PADS2_STREAM_LOCK: + * @pads: a #GstCollectPads2 + * + * Lock the stream lock of @pads. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STREAM_LOCK(pads) (g_static_rec_mutex_lock(GST_COLLECT_PADS2_GET_STREAM_LOCK (pads))) +/** + * GST_COLLECT_PADS2_STREAM_UNLOCK: + * @pads: a #GstCollectPads2 + * + * Unlock the stream lock of @pads. + * + * Since: 0.10.36 + */ +#define GST_COLLECT_PADS2_STREAM_UNLOCK(pads) (g_static_rec_mutex_unlock(GST_COLLECT_PADS2_GET_STREAM_LOCK (pads))) + +/** + * GstCollectPads2: + * @data: #GList of #GstCollectData2 managed by this #GstCollectPads2. + * + * Collectpads object. + * + * Since: 0.10.36 + */ +struct _GstCollectPads2 { + GstObject object; + + /*< public >*/ /* with LOCK and/or STREAM_LOCK */ + GSList *data; /* list of CollectData items */ + + /*< private >*/ + GStaticRecMutex stream_lock; /* used to serialize collection among several streams */ + /* with LOCK and/or STREAM_LOCK*/ + gboolean started; + + /* with STREAM_LOCK */ + guint32 cookie; /* @data list cookie */ + guint numpads; /* number of pads in @data */ + guint queuedpads; /* number of pads with a buffer */ + guint eospads; /* number of pads that are EOS */ + GstClockTime earliest_time; /* Current earliest time */ + GstCollectData2 *earliest_data; /* Pad data for current earliest time */ + + /* with LOCK */ + GSList *pad_list; /* updated pad list */ + guint32 pad_cookie; /* updated cookie */ + + GstCollectPads2Function func; /* function and user_data for callback */ + gpointer user_data; + GstCollectPads2BufferFunction prepare_buffer_func; /* function and user_data for prepare buffer callback */ + gpointer prepare_buffer_user_data; + GstCollectPads2BufferFunction buffer_func; /* function and user_data for buffer callback */ + gpointer buffer_user_data; + GstCollectPads2CompareFunction compare_func; + gpointer compare_user_data; + GstCollectPads2EventFunction event_func; /* function and data for event callback */ + gpointer event_user_data; + + /* no other lock needed */ + GMutex *evt_lock; /* these make up sort of poor man's event signaling */ + GCond *evt_cond; + guint32 evt_cookie; + + gpointer _gst_reserved[GST_PADDING + 0]; + +}; + +struct _GstCollectPads2Class { + GstObjectClass parent_class; + + /*< private >*/ + gpointer _gst_reserved[GST_PADDING]; +}; + +GType gst_collect_pads2_get_type(void); + +/* creating the object */ +GstCollectPads2* gst_collect_pads2_new (void); + +/* set the callbacks */ +void gst_collect_pads2_set_function (GstCollectPads2 *pads, GstCollectPads2Function func, + gpointer user_data); +void gst_collect_pads2_set_prepare_buffer_function (GstCollectPads2 *pads, + GstCollectPads2BufferFunction func, gpointer user_data); +void gst_collect_pads2_set_buffer_function (GstCollectPads2 *pads, + GstCollectPads2BufferFunction func, gpointer user_data); +void gst_collect_pads2_set_event_function (GstCollectPads2 *pads, + GstCollectPads2EventFunction func, gpointer user_data); +void gst_collect_pads2_set_compare_function (GstCollectPads2 *pads, + GstCollectPads2CompareFunction func, gpointer user_data); + +/* pad management */ +GstCollectData2* gst_collect_pads2_add_pad (GstCollectPads2 *pads, GstPad *pad, guint size); +GstCollectData2* gst_collect_pads2_add_pad_full (GstCollectPads2 *pads, GstPad *pad, guint size, GstCollectData2DestroyNotify destroy_notify, + gboolean lock); +gboolean gst_collect_pads2_remove_pad (GstCollectPads2 *pads, GstPad *pad); +gboolean gst_collect_pads2_is_active (GstCollectPads2 *pads, GstPad *pad); + +/* start/stop collection */ +GstFlowReturn gst_collect_pads2_collect (GstCollectPads2 *pads); +GstFlowReturn gst_collect_pads2_collect_range (GstCollectPads2 *pads, guint64 offset, guint length); + +void gst_collect_pads2_start (GstCollectPads2 *pads); +void gst_collect_pads2_stop (GstCollectPads2 *pads); +void gst_collect_pads2_set_flushing (GstCollectPads2 *pads, gboolean flushing); + +/* get collected buffers */ +GstBuffer* gst_collect_pads2_peek (GstCollectPads2 *pads, GstCollectData2 *data); +GstBuffer* gst_collect_pads2_pop (GstCollectPads2 *pads, GstCollectData2 *data); + +/* get collected bytes */ +guint gst_collect_pads2_available (GstCollectPads2 *pads); +guint gst_collect_pads2_read (GstCollectPads2 *pads, GstCollectData2 *data, + guint8 **bytes, guint size); +guint gst_collect_pads2_flush (GstCollectPads2 *pads, GstCollectData2 *data, + guint size); +GstBuffer* gst_collect_pads2_read_buffer (GstCollectPads2 * pads, GstCollectData2 * data, + guint size); +GstBuffer* gst_collect_pads2_take_buffer (GstCollectPads2 * pads, GstCollectData2 * data, + guint size); + +/* setting and unsetting waiting mode */ +void gst_collect_pads2_set_waiting (GstCollectPads2 *pads, GstCollectData2 *data, + gboolean waiting); + + +G_END_DECLS + +#endif /* __GST_COLLECT_PADS22_H__ */ -- 2.7.4