--- /dev/null
+/* GStreamer concat element
+ *
+ * Copyright (c) 2014 Sebastian Dröge <sebastian@centricular.com>
+ *
+ * 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 "gstconcat.h"
+
+GST_DEBUG_CATEGORY_STATIC (gst_concat_debug);
+#define GST_CAT_DEFAULT gst_concat_debug
+
+G_GNUC_INTERNAL GType gst_concat_pad_get_type (void);
+
+#define GST_TYPE_CONCAT_PAD (gst_concat_pad_get_type())
+#define GST_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_CONCAT_PAD, GstConcatPad))
+#define GST_CONCAT_PAD_CAST(obj) ((GstConcatPad *)(obj))
+#define GST_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_CONCAT_PAD, GstConcatPadClass))
+#define GST_IS_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_CONCAT_PAD))
+#define GST_IS_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_CONCAT_PAD))
+
+typedef struct _GstConcatPad GstConcatPad;
+typedef struct _GstConcatPadClass GstConcatPadClass;
+
+struct _GstConcatPad
+{
+ GstPad parent;
+
+ GstSegment segment;
+
+ /* Protected by the concat lock */
+ gboolean flushing;
+};
+
+struct _GstConcatPadClass
+{
+ GstPadClass parent;
+};
+
+G_DEFINE_TYPE (GstConcatPad, gst_concat_pad, GST_TYPE_PAD);
+
+static void
+gst_concat_pad_class_init (GstConcatPadClass * klass)
+{
+}
+
+static void
+gst_concat_pad_init (GstConcatPad * self)
+{
+ gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
+ self->flushing = FALSE;
+}
+
+static GstStaticPadTemplate concat_sink_template =
+GST_STATIC_PAD_TEMPLATE ("sink_%u",
+ GST_PAD_SINK,
+ GST_PAD_REQUEST,
+ GST_STATIC_CAPS_ANY);
+
+static GstStaticPadTemplate concat_src_template =
+GST_STATIC_PAD_TEMPLATE ("src",
+ GST_PAD_SRC,
+ GST_PAD_ALWAYS,
+ GST_STATIC_CAPS_ANY);
+
+#define _do_init \
+ GST_DEBUG_CATEGORY_INIT (gst_concat_debug, "concat", 0, "concat element");
+#define gst_concat_parent_class parent_class
+G_DEFINE_TYPE_WITH_CODE (GstConcat, gst_concat, GST_TYPE_ELEMENT, _do_init);
+
+static void gst_concat_dispose (GObject * object);
+static void gst_concat_finalize (GObject * object);
+
+static GstStateChangeReturn gst_concat_change_state (GstElement * element,
+ GstStateChange transition);
+static GstPad *gst_concat_request_new_pad (GstElement * element,
+ GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
+static void gst_concat_release_pad (GstElement * element, GstPad * pad);
+
+static GstFlowReturn gst_concat_sink_chain (GstPad * pad, GstObject * parent,
+ GstBuffer * buffer);
+static gboolean gst_concat_sink_event (GstPad * pad, GstObject * parent,
+ GstEvent * event);
+static gboolean gst_concat_sink_query (GstPad * pad, GstObject * parent,
+ GstQuery * query);
+
+static gboolean gst_concat_src_event (GstPad * pad, GstObject * parent,
+ GstEvent * event);
+static gboolean gst_concat_src_query (GstPad * pad, GstObject * parent,
+ GstQuery * query);
+
+static gboolean gst_concat_switch_pad (GstConcat * self);
+
+static void
+gst_concat_class_init (GstConcatClass * klass)
+{
+ GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+ GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
+
+ gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_concat_dispose);
+ gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_concat_finalize);
+
+ gst_element_class_set_static_metadata (gstelement_class,
+ "Concat", "Generic", "Concatenate multiple streams",
+ "Sebastian Dröge <sebastian@centricular.com>");
+
+ gst_element_class_add_pad_template (gstelement_class,
+ gst_static_pad_template_get (&concat_sink_template));
+ gst_element_class_add_pad_template (gstelement_class,
+ gst_static_pad_template_get (&concat_src_template));
+
+ gstelement_class->request_new_pad =
+ GST_DEBUG_FUNCPTR (gst_concat_request_new_pad);
+ gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_concat_release_pad);
+ gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_concat_change_state);
+}
+
+static void
+gst_concat_init (GstConcat * self)
+{
+ g_mutex_init (&self->lock);
+ g_cond_init (&self->cond);
+
+ self->srcpad = gst_pad_new_from_static_template (&concat_src_template, "src");
+ gst_pad_set_event_function (self->srcpad,
+ GST_DEBUG_FUNCPTR (gst_concat_src_event));
+ gst_pad_set_query_function (self->srcpad,
+ GST_DEBUG_FUNCPTR (gst_concat_src_query));
+ gst_pad_use_fixed_caps (self->srcpad);
+
+ gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
+}
+
+static void
+gst_concat_dispose (GObject * object)
+{
+ GstConcat *self = GST_CONCAT (object);
+ GList *item;
+
+ gst_object_replace ((GstObject **) & self->current_sinkpad, NULL);
+
+restart:
+ for (item = GST_ELEMENT_PADS (object); item; item = g_list_next (item)) {
+ GstPad *pad = GST_PAD (item->data);
+
+ if (GST_PAD_IS_SINK (pad)) {
+ gst_element_release_request_pad (GST_ELEMENT (object), pad);
+ goto restart;
+ }
+ }
+
+ G_OBJECT_CLASS (parent_class)->dispose (object);
+}
+
+static void
+gst_concat_finalize (GObject * object)
+{
+ GstConcat *self = GST_CONCAT (object);
+
+ g_mutex_clear (&self->lock);
+ g_cond_clear (&self->cond);
+
+ G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static GstPad *
+gst_concat_request_new_pad (GstElement * element, GstPadTemplate * templ,
+ const gchar * name, const GstCaps * caps)
+{
+ GstConcat *self = GST_CONCAT (element);
+ GstPad *sinkpad;
+ gchar *pad_name;
+
+ GST_DEBUG_OBJECT (element, "requesting pad");
+
+ g_mutex_lock (&self->lock);
+ pad_name = g_strdup_printf ("sink_%u", self->pad_count);
+ self->pad_count++;
+ g_mutex_unlock (&self->lock);
+
+ sinkpad = GST_PAD_CAST (g_object_new (GST_TYPE_CONCAT_PAD,
+ "name", pad_name, "direction", templ->direction, "template", templ,
+ NULL));
+ g_free (pad_name);
+
+ gst_pad_set_chain_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_concat_sink_chain));
+ gst_pad_set_event_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_concat_sink_event));
+ gst_pad_set_query_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_concat_sink_query));
+ GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_CAPS);
+ GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_ALLOCATION);
+
+ gst_pad_set_active (sinkpad, TRUE);
+
+ g_mutex_lock (&self->lock);
+ self->sinkpads = g_list_prepend (self->sinkpads, gst_object_ref (sinkpad));
+ if (!self->current_sinkpad)
+ self->current_sinkpad = gst_object_ref (sinkpad);
+ g_mutex_unlock (&self->lock);
+
+ gst_element_add_pad (element, sinkpad);
+
+ return sinkpad;
+}
+
+static void
+gst_concat_release_pad (GstElement * element, GstPad * pad)
+{
+ GstConcat *self = GST_CONCAT (element);
+ GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
+ GList *l;
+ gboolean current_pad_removed = FALSE;
+ gboolean eos = FALSE;
+
+ GST_DEBUG_OBJECT (self, "releasing pad");
+
+ g_mutex_lock (&self->lock);
+ spad->flushing = TRUE;
+ g_cond_broadcast (&self->cond);
+ g_mutex_unlock (&self->lock);
+
+ gst_pad_set_active (pad, FALSE);
+
+ /* Now the pad is definitely not running anymore */
+
+ g_mutex_lock (&self->lock);
+ if (self->current_sinkpad == GST_PAD_CAST (spad)) {
+ eos = ! !gst_concat_switch_pad (self);
+ current_pad_removed = TRUE;
+ }
+
+ for (l = self->sinkpads; l; l = l->next) {
+ if ((gpointer) spad == l->data) {
+ gst_object_unref (spad);
+ self->sinkpads = g_list_delete_link (self->sinkpads, l);
+ break;
+ }
+ }
+ g_mutex_unlock (&self->lock);
+
+ gst_element_remove_pad (GST_ELEMENT_CAST (self), pad);
+
+ if (GST_STATE (self) > GST_STATE_READY) {
+ if (current_pad_removed && !eos)
+ gst_element_post_message (GST_ELEMENT_CAST (self),
+ gst_message_new_duration_changed (GST_OBJECT_CAST (self)));
+
+ /* FIXME: Sending EOS from application thread */
+ if (eos)
+ gst_pad_push_event (self->srcpad, gst_event_new_eos ());
+ }
+}
+
+/* Returns FALSE if flushing
+ * Must be called from the pad's streaming thread
+ */
+static gboolean
+gst_concat_pad_wait (GstConcatPad * spad, GstConcat * self)
+{
+ g_mutex_lock (&self->lock);
+ if (spad->flushing) {
+ g_mutex_unlock (&self->lock);
+ GST_DEBUG_OBJECT (spad, "Flushing");
+ return FALSE;
+ }
+
+ while (spad != GST_CONCAT_PAD_CAST (self->current_sinkpad)) {
+ GST_TRACE_OBJECT (spad, "Not the current sinkpad - waiting");
+ g_cond_wait (&self->cond, &self->lock);
+ if (spad->flushing) {
+ g_mutex_unlock (&self->lock);
+ GST_DEBUG_OBJECT (spad, "Flushing");
+ return FALSE;
+ }
+ }
+ /* This pad can only become not the current sinkpad from
+ * a) This streaming thread (we hold the stream lock)
+ * b) Releasing the pad (takes the stream lock, see above)
+ *
+ * Unlocking here is thus safe and we can safely push
+ * serialized data to our srcpad
+ */
+ GST_DEBUG_OBJECT (spad, "Now the current sinkpad");
+ g_mutex_unlock (&self->lock);
+
+ return TRUE;
+}
+
+static GstFlowReturn
+gst_concat_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
+{
+ GstFlowReturn ret;
+ GstConcat *self = GST_CONCAT (parent);
+ GstConcatPad *spad = GST_CONCAT_PAD (pad);
+
+ GST_LOG_OBJECT (pad, "received buffer %p", buffer);
+
+ if (!gst_concat_pad_wait (spad, self))
+ return GST_FLOW_FLUSHING;
+
+ if (self->last_stop == GST_CLOCK_TIME_NONE)
+ self->last_stop = spad->segment.start;
+
+ if (self->format == GST_FORMAT_TIME) {
+ GstClockTime start_time = GST_BUFFER_TIMESTAMP (buffer);
+ GstClockTime end_time = GST_CLOCK_TIME_NONE;
+
+ if (start_time != GST_CLOCK_TIME_NONE)
+ end_time = start_time;
+ if (GST_BUFFER_DURATION_IS_VALID (buffer))
+ end_time += GST_BUFFER_DURATION (buffer);
+
+ if (end_time != GST_CLOCK_TIME_NONE && end_time > self->last_stop)
+ self->last_stop = end_time;
+ } else {
+ self->last_stop += gst_buffer_get_size (buffer);
+ }
+
+ ret = gst_pad_push (self->srcpad, buffer);
+
+ GST_LOG_OBJECT (pad, "handled buffer %s", gst_flow_get_name (ret));
+
+ return ret;
+}
+
+/* Returns FALSE if no further pad, must be called with concat lock */
+static gboolean
+gst_concat_switch_pad (GstConcat * self)
+{
+ GList *l;
+ gboolean next;
+ GstSegment segment;
+ gint64 last_stop;
+
+ segment = GST_CONCAT_PAD (self->current_sinkpad)->segment;
+
+ last_stop = self->last_stop;
+ if (last_stop == GST_CLOCK_TIME_NONE)
+ last_stop = segment.stop;
+ if (last_stop == GST_CLOCK_TIME_NONE)
+ last_stop = segment.start;
+ g_assert (last_stop != GST_CLOCK_TIME_NONE);
+
+ if (last_stop > segment.stop)
+ last_stop = segment.stop;
+
+ if (segment.format == GST_FORMAT_TIME)
+ last_stop =
+ gst_segment_to_running_time (&segment, segment.format, last_stop);
+ else
+ last_stop += segment.start;
+
+ self->current_start_offset += last_stop;
+
+ for (l = self->sinkpads; l; l = l->next) {
+ if ((gpointer) self->current_sinkpad == l->data) {
+ l = l->prev;
+ GST_DEBUG_OBJECT (self,
+ "Switching from pad %" GST_PTR_FORMAT " to %" GST_PTR_FORMAT,
+ self->current_sinkpad, l ? l->data : NULL);
+ gst_object_unref (self->current_sinkpad);
+ self->current_sinkpad = l ? gst_object_ref (l->data) : NULL;
+ g_cond_broadcast (&self->cond);
+ break;
+ }
+ }
+
+ next = self->current_sinkpad != NULL;
+
+ self->last_stop = GST_CLOCK_TIME_NONE;
+
+ return next;
+}
+
+static gboolean
+gst_concat_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
+{
+ GstConcat *self = GST_CONCAT (parent);
+ GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
+ gboolean ret = TRUE;
+
+ GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
+
+ switch (GST_EVENT_TYPE (event)) {
+ case GST_EVENT_STREAM_START:{
+ if (!gst_concat_pad_wait (spad, self)) {
+ ret = FALSE;
+ gst_event_unref (event);
+ } else {
+ ret = gst_pad_event_default (pad, parent, event);
+ }
+ break;
+ }
+ case GST_EVENT_SEGMENT:{
+ /* Drop segment event, we create our own one */
+ gst_event_copy_segment (event, &spad->segment);
+ gst_event_unref (event);
+
+ g_mutex_lock (&self->lock);
+ if (self->format == GST_FORMAT_UNDEFINED) {
+ if (spad->segment.format != GST_FORMAT_TIME
+ && spad->segment.format != GST_FORMAT_BYTES) {
+ g_mutex_unlock (&self->lock);
+ GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
+ ("Can only operate in TIME or BYTES format"));
+ ret = FALSE;
+ break;
+ }
+ self->format = spad->segment.format;
+ GST_DEBUG_OBJECT (self, "Operating in %s format",
+ gst_format_get_name (self->format));
+ g_mutex_unlock (&self->lock);
+ } else if (self->format != spad->segment.format) {
+ g_mutex_unlock (&self->lock);
+ GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
+ ("Operating in %s format but new pad has %s",
+ gst_format_get_name (self->format),
+ gst_format_get_name (spad->segment.format)));
+ ret = FALSE;
+ } else {
+ g_mutex_unlock (&self->lock);
+ }
+
+ if (!gst_concat_pad_wait (spad, self)) {
+ ret = FALSE;
+ } else {
+ GstSegment segment = spad->segment;
+
+ /* We know no duration */
+ segment.duration = -1;
+
+ /* Update segment values to be contiguous with last stream */
+ if (self->format == GST_FORMAT_TIME) {
+ segment.base += self->current_start_offset;
+ } else {
+ /* Shift start/stop byte position */
+ segment.start += self->current_start_offset;
+ if (segment.stop != -1)
+ segment.stop += self->current_start_offset;
+ }
+ gst_pad_push_event (self->srcpad, gst_event_new_segment (&segment));
+ }
+ break;
+ }
+ case GST_EVENT_EOS:{
+ gst_event_unref (event);
+ if (!gst_concat_pad_wait (spad, self)) {
+ ret = FALSE;
+ } else {
+ gboolean next;
+
+ g_mutex_lock (&self->lock);
+ next = gst_concat_switch_pad (self);
+ g_mutex_unlock (&self->lock);
+ ret = TRUE;
+
+ if (!next) {
+ gst_pad_push_event (self->srcpad, gst_event_new_eos ());
+ } else {
+ gst_element_post_message (GST_ELEMENT_CAST (self),
+ gst_message_new_duration_changed (GST_OBJECT_CAST (self)));
+ }
+ }
+ break;
+ }
+ case GST_EVENT_FLUSH_START:{
+ g_mutex_lock (&self->lock);
+ spad->flushing = TRUE;
+ g_cond_broadcast (&self->cond);
+ g_mutex_unlock (&self->lock);
+ break;
+ }
+ case GST_EVENT_FLUSH_STOP:{
+ gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
+ spad->flushing = FALSE;
+ break;
+ }
+ default:{
+ /* Wait for other serialized events before forwarding */
+ if (GST_EVENT_IS_SERIALIZED (event) && !gst_concat_pad_wait (spad, self)) {
+ gst_event_unref (event);
+ ret = FALSE;
+ } else {
+ ret = gst_pad_event_default (pad, parent, event);
+ }
+ break;
+ }
+ }
+
+ return ret;
+}
+
+static gboolean
+gst_concat_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
+{
+ GstConcat *self = GST_CONCAT (parent);
+ GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
+ gboolean ret = TRUE;
+
+ GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
+
+ switch (GST_QUERY_TYPE (query)) {
+ default:
+ /* Wait for other serialized queries before forwarding */
+ if (GST_QUERY_IS_SERIALIZED (query) && !gst_concat_pad_wait (spad, self)) {
+ ret = FALSE;
+ } else {
+ ret = gst_pad_query_default (pad, parent, query);
+ }
+ break;
+ }
+
+ return ret;
+}
+
+static gboolean
+gst_concat_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
+{
+ GstConcat *self = GST_CONCAT (parent);
+ gboolean ret = TRUE;
+
+ GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
+
+ switch (GST_EVENT_TYPE (event)) {
+ case GST_EVENT_SEEK:{
+ /* We don't support seeking */
+ gst_event_unref (event);
+ ret = FALSE;
+ break;
+ }
+ case GST_EVENT_QOS:{
+ GstQOSType type;
+ GstClockTimeDiff diff;
+ GstClockTime timestamp;
+ gdouble proportion;
+
+ gst_event_parse_qos (event, &type, &proportion, &diff, ×tamp);
+ gst_event_unref (event);
+
+ if (timestamp != GST_CLOCK_TIME_NONE
+ && timestamp > self->current_start_offset) {
+ timestamp -= self->current_start_offset;
+ event = gst_event_new_qos (type, proportion, diff, timestamp);
+ ret = gst_pad_push_event (self->current_sinkpad, event);
+ } else {
+ ret = FALSE;
+ }
+ break;
+ }
+ default:
+ ret = gst_pad_event_default (pad, parent, event);
+ break;
+ }
+
+ return ret;
+}
+
+static gboolean
+gst_concat_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
+{
+ gboolean ret = TRUE;
+
+ GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
+
+ switch (GST_QUERY_TYPE (query)) {
+ default:
+ ret = gst_pad_query_default (pad, parent, query);
+ break;
+ }
+
+ return ret;
+}
+
+static void
+reset_pad (const GValue * data, gpointer user_data)
+{
+ GstPad *pad = g_value_get_object (data);
+ GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
+
+ gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
+ spad->flushing = FALSE;
+}
+
+static void
+unblock_pad (const GValue * data, gpointer user_data)
+{
+ GstPad *pad = g_value_get_object (data);
+ GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
+
+ spad->flushing = TRUE;
+}
+
+static GstStateChangeReturn
+gst_concat_change_state (GstElement * element, GstStateChange transition)
+{
+ GstConcat *self = GST_CONCAT (element);
+ GstStateChangeReturn ret;
+
+ switch (transition) {
+ case GST_STATE_CHANGE_READY_TO_PAUSED:{
+ self->format = GST_FORMAT_UNDEFINED;
+ self->current_start_offset = 0;
+ self->last_stop = GST_CLOCK_TIME_NONE;
+ break;
+ }
+ case GST_STATE_CHANGE_PAUSED_TO_READY:{
+ GstIterator *iter = gst_element_iterate_sink_pads (element);
+ GstIteratorResult res;
+
+ g_mutex_lock (&self->lock);
+ do {
+ res = gst_iterator_foreach (iter, unblock_pad, NULL);
+ } while (res == GST_ITERATOR_RESYNC);
+
+ gst_iterator_free (iter);
+ g_cond_broadcast (&self->cond);
+ g_mutex_unlock (&self->lock);
+
+ if (res == GST_ITERATOR_ERROR)
+ return GST_STATE_CHANGE_FAILURE;
+
+ break;
+ }
+ default:
+ break;
+ }
+
+ ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+ if (ret == GST_STATE_CHANGE_FAILURE)
+ return ret;
+
+ switch (transition) {
+ case GST_STATE_CHANGE_PAUSED_TO_READY:{
+ GstIterator *iter = gst_element_iterate_sink_pads (element);
+ GstIteratorResult res;
+
+ do {
+ res = gst_iterator_foreach (iter, reset_pad, NULL);
+ } while (res == GST_ITERATOR_RESYNC);
+
+ gst_iterator_free (iter);
+
+ if (res == GST_ITERATOR_ERROR)
+ return GST_STATE_CHANGE_FAILURE;
+
+ break;
+ }
+ default:
+ break;
+ }
+
+ return ret;
+}
projects / platform / upstream / gstreamer.git / commitdiff