gstreamer/plugins/elements/gstconcat.c

/* 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.
 *
 */
/**
 * SECTION:element-concat
 * @title: concat
 * @see_also: #GstFunnel
 *
 * Concatenates streams together to one continuous stream.
 *
 * All streams but the current one are blocked until the current one
 * finished with %GST_EVENT_EOS. Then the next stream is enabled, while
 * keeping the running time continuous for %GST_FORMAT_TIME segments or
 * keeping the segment continuous for %GST_FORMAT_BYTES segments.
 *
 * Streams are switched in the order in which the sinkpads were requested.
 *
 * By default, the stream segment's base values are adjusted to ensure
 * the segment transitions between streams are continuous. In some cases,
 * it may be desirable to turn off these adjustments (for example, because
 * another downstream element like a streamsynchronizer adjusts the base
 * values on its own). The adjust-base property can be used for this purpose.
 *
 * ## Example launch line
 * |[
 * gst-launch-1.0 concat name=c ! xvimagesink  videotestsrc num-buffers=100 ! c.   videotestsrc num-buffers=100 pattern=ball ! c.
 * ]| Plays two video streams one after another.
 *
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "gstconcat.h"
#include "gstcoreelementselements.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 sink_template = GST_STATIC_PAD_TEMPLATE ("sink_%u",
    GST_PAD_SINK,
    GST_PAD_REQUEST,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
    GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS_ANY);

enum
{
  PROP_0,
  PROP_ACTIVE_PAD,
  PROP_ADJUST_BASE
};

#define DEFAULT_ADJUST_BASE TRUE

#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);
GST_ELEMENT_REGISTER_DEFINE (concat, "concat", GST_RANK_NONE, GST_TYPE_CONCAT);

static void gst_concat_dispose (GObject * object);
static void gst_concat_finalize (GObject * object);
static void gst_concat_get_property (GObject * object,
    guint prop_id, GValue * value, GParamSpec * pspec);
static void gst_concat_set_property (GObject * object,
    guint prop_id, const GValue * value, GParamSpec * pspec);

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_notify_active_pad (GstConcat * self);

static GParamSpec *pspec_active_pad = NULL;

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);

  gobject_class->get_property = gst_concat_get_property;
  gobject_class->set_property = gst_concat_set_property;

  pspec_active_pad = g_param_spec_object ("active-pad", "Active pad",
      "Currently active sink pad", GST_TYPE_PAD, G_PARAM_READABLE |
      G_PARAM_STATIC_STRINGS);
  g_object_class_install_property (gobject_class, PROP_ACTIVE_PAD,
      pspec_active_pad);
  g_object_class_install_property (gobject_class, PROP_ADJUST_BASE,
      g_param_spec_boolean ("adjust-base", "Adjust segment base",
          "Adjust the base value of segments to ensure they are adjacent",
          DEFAULT_ADJUST_BASE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  gst_element_class_set_static_metadata (gstelement_class,
      "Concat", "Generic", "Concatenate multiple streams",
      "Sebastian Dröge <sebastian@centricular.com>");

  gst_element_class_add_static_pad_template (gstelement_class, &sink_template);
  gst_element_class_add_static_pad_template (gstelement_class, &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 (&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);

  self->adjust_base = DEFAULT_ADJUST_BASE;
}

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 void
gst_concat_get_property (GObject * object, guint prop_id, GValue * value,
    GParamSpec * pspec)
{
  GstConcat *self = GST_CONCAT (object);

  switch (prop_id) {
    case PROP_ACTIVE_PAD:{
      g_mutex_lock (&self->lock);
      g_value_set_object (value, self->current_sinkpad);
      g_mutex_unlock (&self->lock);
      break;
    }
    case PROP_ADJUST_BASE:{
      g_mutex_lock (&self->lock);
      g_value_set_boolean (value, self->adjust_base);
      g_mutex_unlock (&self->lock);
      break;
    }
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static void
gst_concat_set_property (GObject * object, guint prop_id, const GValue * value,
    GParamSpec * pspec)
{
  GstConcat *self = GST_CONCAT (object);

  switch (prop_id) {
    case PROP_ADJUST_BASE:{
      g_mutex_lock (&self->lock);
      self->adjust_base = g_value_get_boolean (value);
      g_mutex_unlock (&self->lock);
      break;
    }
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

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;
  gboolean do_notify = FALSE;

  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) {
    do_notify = TRUE;
    self->current_sinkpad = gst_object_ref (sinkpad);
  }
  g_mutex_unlock (&self->lock);

  gst_element_add_pad (element, sinkpad);

  if (do_notify)
    gst_concat_notify_active_pad (self);

  GST_DEBUG_OBJECT (sinkpad, "requested pad");

  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;
  gboolean do_notify = FALSE;

  GST_DEBUG_OBJECT (pad, "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;
    do_notify = 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 (do_notify)
    gst_concat_notify_active_pad (self);

  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");
    if (self->current_sinkpad == NULL && g_list_length (self->sinkpads) == 1) {
      GST_LOG_OBJECT (spad, "Sole pad waiting, switching");
      /* If we are the only sinkpad, take active pad ownership */
      self->current_sinkpad = gst_object_ref (self->sinkpads->data);
      break;
    }
    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 %" GST_PTR_FORMAT, 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, last_stop %" GST_TIME_FORMAT,
      gst_flow_get_name (ret), GST_TIME_ARGS (self->last_stop));

  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 void
gst_concat_notify_active_pad (GstConcat * self)
{
  g_object_notify_by_pspec ((GObject *) self, pspec_active_pad);
}

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;
  gboolean adjust_base;

  GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);

  g_mutex_lock (&self->lock);
  adjust_base = self->adjust_base;
  g_mutex_unlock (&self->lock);

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_STREAM_START:{
      if (!gst_concat_pad_wait (spad, self)) {
        ret = FALSE;
        gst_event_replace (&event, NULL);
      }
      break;
    }
    case GST_EVENT_SEGMENT:{
      /* Drop segment event, we create our own one */
      gst_event_copy_segment (event, &spad->segment);
      gst_event_replace (&event, NULL);

      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));
      } 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;
        break;
      }

      g_mutex_unlock (&self->lock);

      if (!gst_concat_pad_wait (spad, self)) {
        ret = FALSE;
      } else {
        GstSegment segment = spad->segment;

        g_mutex_lock (&self->lock);

        if (adjust_base) {
          /* We know no duration */
          segment.duration = -1;

          /* Update segment values to be continuous with last stream */
          if (self->format == GST_FORMAT_TIME) {
            GST_DEBUG_OBJECT (self,
                "Updating segment base %" GST_TIME_FORMAT " + %" GST_TIME_FORMAT
                " = %" GST_TIME_FORMAT, GST_TIME_ARGS (segment.base),
                GST_TIME_ARGS (self->current_start_offset),
                GST_TIME_ARGS (segment.base + self->current_start_offset));
            segment.base += self->current_start_offset;
          } else {
            /* Shift start/stop byte position */
            GST_DEBUG_OBJECT (self,
                "Updating segment start %" G_GUINT64_FORMAT " + %"
                G_GUINT64_FORMAT " = %" G_GUINT64_FORMAT, segment.start,
                self->current_start_offset,
                segment.start + self->current_start_offset);
            segment.start += self->current_start_offset;
            if (segment.stop != -1) {
              GST_DEBUG_OBJECT (self,
                  "Updating segment stop %" G_GUINT64_FORMAT " + %"
                  G_GUINT64_FORMAT " = %" G_GUINT64_FORMAT, segment.stop,
                  self->current_start_offset,
                  segment.stop + self->current_start_offset);
              segment.stop += self->current_start_offset;
            }
          }
        }
        event = gst_event_new_segment (&segment);
        gst_event_set_seqnum (event, gst_event_get_seqnum (event));

        g_mutex_unlock (&self->lock);
      }
      break;
    }
    case GST_EVENT_EOS:{
      gst_event_replace (&event, NULL);

      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;

        gst_concat_notify_active_pad (self);

        if (!next) {
          event = 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:{
      gboolean forward;

      g_mutex_lock (&self->lock);
      spad->flushing = TRUE;
      g_cond_broadcast (&self->cond);
      forward = (self->current_sinkpad == GST_PAD_CAST (spad));
      if (!forward && g_list_length (self->sinkpads) == 1)
        forward = TRUE;
      g_mutex_unlock (&self->lock);

      if (!forward)
        gst_event_replace (&event, NULL);
      break;
    }
    case GST_EVENT_FLUSH_STOP:{
      gboolean forward;

      gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
      spad->flushing = FALSE;

      g_mutex_lock (&self->lock);
      forward = (self->current_sinkpad == GST_PAD_CAST (spad));
      if (!forward && g_list_length (self->sinkpads) == 1)
        forward = TRUE;
      g_mutex_unlock (&self->lock);

      if (forward) {
        gboolean reset_time;

        gst_event_parse_flush_stop (event, &reset_time);
        if (reset_time) {
          GST_DEBUG_OBJECT (self,
              "resetting start offset to 0 after flushing with reset_time = TRUE");
          self->current_start_offset = 0;
          self->last_stop = GST_CLOCK_TIME_NONE;
        }
      } else {
        gst_event_replace (&event, NULL);
      }
      break;
    }
    default:{
      /* Wait for other serialized events before forwarding */
      if (GST_EVENT_IS_SERIALIZED (event) && !gst_concat_pad_wait (spad, self)) {
        gst_event_replace (&event, NULL);
        ret = FALSE;
      }
      break;
    }
  }

  if (event) {
    g_mutex_lock (&self->lock);
    if (self->adjust_base && self->format == GST_FORMAT_TIME) {
      gint64 offset;

      event = gst_event_make_writable (event);
      offset = gst_event_get_running_time_offset (event);
      offset += self->current_start_offset;

      gst_event_set_running_time_offset (event, offset);
    }
    g_mutex_unlock (&self->lock);
    ret = gst_pad_event_default (pad, parent, event);
  }

  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;
  GstPad *sinkpad = NULL;

  GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_SEEK:{
      g_mutex_lock (&self->lock);
      if ((sinkpad = self->current_sinkpad))
        gst_object_ref (sinkpad);
      /* If no current active sinkpad but only one sinkpad, try reactivating that pad */
      if (sinkpad == NULL && g_list_length (self->sinkpads) == 1) {
        sinkpad = gst_object_ref (self->sinkpads->data);
      }
      g_mutex_unlock (&self->lock);
      if (!sinkpad) {
        gst_event_replace (&event, NULL);
        ret = FALSE;
      }
      break;
    }
    case GST_EVENT_QOS:{
      GstPad *sinkpad = NULL;

      g_mutex_lock (&self->lock);
      if ((sinkpad = self->current_sinkpad))
        gst_object_ref (sinkpad);
      g_mutex_unlock (&self->lock);


      if (!sinkpad) {
        gst_event_replace (&event, NULL);
        ret = FALSE;
      }
      break;
    }
    case GST_EVENT_FLUSH_STOP:{
      gboolean reset_time;

      gst_event_parse_flush_stop (event, &reset_time);
      if (reset_time) {
        GST_DEBUG_OBJECT (self,
            "resetting start offset to 0 after flushing with reset_time = TRUE");
        self->current_start_offset = 0;
      }
      break;
    }
    default:
      break;
  }

  if (event) {
    g_mutex_lock (&self->lock);
    if (self->adjust_base && self->format == GST_FORMAT_TIME) {
      gint64 offset;

      event = gst_event_make_writable (event);
      offset = gst_event_get_running_time_offset (event);
      offset -= self->current_start_offset;

      gst_event_set_running_time_offset (event, offset);
    }
    g_mutex_unlock (&self->lock);

    if (sinkpad)
      ret = gst_pad_push_event (sinkpad, event);
    else
      ret = gst_pad_event_default (pad, parent, event);
  }

  if (sinkpad)
    gst_object_unref (sinkpad);

  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:{
      GstIterator *iter = gst_element_iterate_sink_pads (element);
      GstIteratorResult res;

      self->format = GST_FORMAT_UNDEFINED;
      self->current_start_offset = 0;
      self->last_stop = GST_CLOCK_TIME_NONE;

      while ((res =
              gst_iterator_foreach (iter, reset_pad,
                  NULL)) == GST_ITERATOR_RESYNC)
        gst_iterator_resync (iter);
      gst_iterator_free (iter);

      if (res == GST_ITERATOR_ERROR)
        return GST_STATE_CHANGE_FAILURE;
      break;
    }
    case GST_STATE_CHANGE_PAUSED_TO_READY:{
      GstIterator *iter = gst_element_iterate_sink_pads (element);
      GstIteratorResult res;

      g_mutex_lock (&self->lock);
      while ((res =
              gst_iterator_foreach (iter, unblock_pad,
                  NULL)) == GST_ITERATOR_RESYNC)
        gst_iterator_resync (iter);
      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);

  return ret;
}