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
 * @see_also: #GstFunnel
 *
 * Concatenates streams together to one continous 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 continous for %GST_FORMAT_TIME segments or
 * keeping the segment continous for %GST_FORMAT_BYTES segments.
 *
 * Streams are switched in the order in which the sinkpads were requested.
 *
 * <refsect2>
 * <title>Example launch line</title>
 * |[
 * 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.
 * </refsect2>
 */

#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 continous 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:{
      GstPad *sinkpad = NULL;

      g_mutex_lock (&self->lock);
      if ((sinkpad = self->current_sinkpad))
        gst_object_ref (sinkpad);
      g_mutex_unlock (&self->lock);
      if (sinkpad) {
        ret = gst_pad_push_event (sinkpad, event);
        gst_object_unref (sinkpad);
      } else {
        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, &timestamp);
      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;
}