gstreamer/gst/liveadder/liveadder.c

/*
 * GStreamer
 *
 *  Copyright 2012 Collabora Ltd
 *  Copyright 2008 Nokia Corporation
 *   @author: Olivier Crete <olivier.crete@collabora.co.uk>
 *
 * With parts copied from the adder plugin which is
 * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
 *                    2001 Thomas <thomas@apestaart.org>
 *               2005,2006 Wim Taymans <wim@fluendo.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 *
 */
/**
 * SECTION:element-liveadder
 * @see_also: adder
 *
 * The live adder allows to mix several streams into one by adding the data.
 * Mixed data is clamped to the min/max values of the data format.
 *
 * Unlike the adder, the liveadder mixes the streams according the their
 * timestamps and waits for some milli-seconds before trying doing the mixing.
 */

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

#include "liveadder.h"

#include <gst/audio/audio.h>

#include <string.h>

#define DEFAULT_LATENCY_MS 60

GST_DEBUG_CATEGORY_STATIC (live_adder_debug);
#define GST_CAT_DEFAULT (live_adder_debug)

static GstStaticPadTemplate gst_live_adder_sink_template =
GST_STATIC_PAD_TEMPLATE ("sink_%u",
    GST_PAD_SINK,
    GST_PAD_REQUEST,
    GST_STATIC_CAPS (GST_AUDIO_CAPS_MAKE ("{ S8, U8, "
            GST_AUDIO_NE (S16) "," GST_AUDIO_NE (U16) ","
            GST_AUDIO_NE (S32) "," GST_AUDIO_NE (U32) ","
            GST_AUDIO_NE (F32) "," GST_AUDIO_NE (F64) "}"))
    );

static GstStaticPadTemplate gst_live_adder_src_template =
GST_STATIC_PAD_TEMPLATE ("src",
    GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS (GST_AUDIO_CAPS_MAKE ("{ S8, U8, "
            GST_AUDIO_NE (S16) "," GST_AUDIO_NE (U16) ","
            GST_AUDIO_NE (S32) "," GST_AUDIO_NE (U32) ","
            GST_AUDIO_NE (F32) "," GST_AUDIO_NE (F64) "}"))
    );

/* Valve signals and args */
enum
{
  /* FILL ME */
  LAST_SIGNAL
};

enum
{
  PROP_0,
  PROP_LATENCY,
};

typedef struct _GstLiveAdderPadPrivate
{
  GstSegment segment;
  gboolean eos;

  GstClockTime expected_timestamp;

} GstLiveAdderPadPrivate;

G_DEFINE_TYPE (GstLiveAdder, gst_live_adder, GST_TYPE_ELEMENT);

static void gst_live_adder_finalize (GObject * object);
static void
gst_live_adder_set_property (GObject * object,
    guint prop_id, const GValue * value, GParamSpec * pspec);
static void
gst_live_adder_get_property (GObject * object,
    guint prop_id, GValue * value, GParamSpec * pspec);

static GstPad *gst_live_adder_request_new_pad (GstElement * element,
    GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
static void gst_live_adder_release_pad (GstElement * element, GstPad * pad);
static GstStateChangeReturn
gst_live_adder_change_state (GstElement * element, GstStateChange transition);

static gboolean gst_live_adder_setcaps (GstLiveAdder * adder, GstPad * pad,
    GstCaps * caps);
static GstCaps *gst_live_adder_sink_getcaps (GstLiveAdder * adder, GstPad * pad,
    GstCaps * filter);
static gboolean gst_live_adder_src_activate_mode (GstPad * pad,
    GstObject * parent, GstPadMode mode, gboolean active);
static gboolean gst_live_adder_src_event (GstPad * pad, GstObject * parent,
    GstEvent * event);

static void gst_live_adder_loop (gpointer data);
static gboolean gst_live_adder_src_query (GstPad * pad, GstObject * parent,
    GstQuery * query);
static gboolean gst_live_adder_sink_query (GstPad * pad, GstObject * parent,
    GstQuery * query);
static gboolean gst_live_adder_sink_event (GstPad * pad, GstObject * parent,
    GstEvent * event);


static void reset_pad_private (GstPad * pad);

/* clipping versions */
#define MAKE_FUNC(name,type,ttype,min,max)                      \
static void name (type *out, type *in, gint bytes) {            \
  gint i;                                                       \
  for (i = 0; i < bytes / sizeof (type); i++)                   \
    out[i] = CLAMP ((ttype)out[i] + (ttype)in[i], min, max);    \
}

/* non-clipping versions (for float) */
#define MAKE_FUNC_NC(name,type,ttype)                           \
static void name (type *out, type *in, gint bytes) {            \
  gint i;                                                       \
  for (i = 0; i < bytes / sizeof (type); i++)                   \
    out[i] = (ttype)out[i] + (ttype)in[i];                      \
}

/* *INDENT-OFF* */
MAKE_FUNC (add_int32, gint32, gint64, G_MININT32, G_MAXINT32)
MAKE_FUNC (add_int16, gint16, gint32, G_MININT16, G_MAXINT16)
MAKE_FUNC (add_int8, gint8, gint16, G_MININT8, G_MAXINT8)
MAKE_FUNC (add_uint32, guint32, guint64, 0, G_MAXUINT32)
MAKE_FUNC (add_uint16, guint16, guint32, 0, G_MAXUINT16)
MAKE_FUNC (add_uint8, guint8, guint16, 0, G_MAXUINT8)
MAKE_FUNC_NC (add_float64, gdouble, gdouble)
MAKE_FUNC_NC (add_float32, gfloat, gfloat)
/* *INDENT-ON* */


static void
gst_live_adder_class_init (GstLiveAdderClass * klass)
{
  GObjectClass *gobject_class = (GObjectClass *) klass;
  GstElementClass *gstelement_class = (GstElementClass *) klass;

  GST_DEBUG_CATEGORY_INIT (live_adder_debug, "liveadder", 0, "Live Adder");

  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&gst_live_adder_src_template));
  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&gst_live_adder_sink_template));
  gst_element_class_set_static_metadata (gstelement_class, "Live Adder element",
      "Generic/Audio",
      "Mixes live/discontinuous audio streams",
      "Olivier Crete <olivier.crete@collabora.co.uk>");

  gobject_class->finalize = gst_live_adder_finalize;
  gobject_class->set_property = gst_live_adder_set_property;
  gobject_class->get_property = gst_live_adder_get_property;

  gstelement_class->request_new_pad = gst_live_adder_request_new_pad;
  gstelement_class->release_pad = gst_live_adder_release_pad;
  gstelement_class->change_state = gst_live_adder_change_state;

  g_object_class_install_property (gobject_class, PROP_LATENCY,
      g_param_spec_uint ("latency", "Buffering latency",
          "Amount of data to buffer (in milliseconds)",
          0, G_MAXUINT, DEFAULT_LATENCY_MS,
          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
}

static void
gst_live_adder_init (GstLiveAdder * adder)
{
  adder->srcpad =
      gst_pad_new_from_static_template (&gst_live_adder_src_template, "src");
  gst_pad_set_query_function (adder->srcpad,
      GST_DEBUG_FUNCPTR (gst_live_adder_src_query));
  gst_pad_set_event_function (adder->srcpad,
      GST_DEBUG_FUNCPTR (gst_live_adder_src_event));
  gst_pad_set_activatemode_function (adder->srcpad,
      GST_DEBUG_FUNCPTR (gst_live_adder_src_activate_mode));
  gst_element_add_pad (GST_ELEMENT (adder), adder->srcpad);

  adder->padcount = 0;
  adder->func = NULL;
  g_cond_init (&adder->not_empty_cond);

  adder->next_timestamp = GST_CLOCK_TIME_NONE;

  adder->latency_ms = DEFAULT_LATENCY_MS;

  adder->buffers = g_queue_new ();
}


static void
gst_live_adder_finalize (GObject * object)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (object);

  g_cond_clear (&adder->not_empty_cond);

  g_queue_foreach (adder->buffers, (GFunc) gst_mini_object_unref, NULL);
  while (g_queue_pop_head (adder->buffers)) {
  }
  g_queue_free (adder->buffers);

  g_list_free (adder->sinkpads);

  G_OBJECT_CLASS (gst_live_adder_parent_class)->finalize (object);
}


static void
gst_live_adder_set_property (GObject * object,
    guint prop_id, const GValue * value, GParamSpec * pspec)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (object);

  switch (prop_id) {
    case PROP_LATENCY:
    {
      guint64 new_latency, old_latency;

      new_latency = g_value_get_uint (value);

      GST_OBJECT_LOCK (adder);
      old_latency = adder->latency_ms;
      adder->latency_ms = new_latency;
      GST_OBJECT_UNLOCK (adder);

      /* post message if latency changed, this will inform the parent pipeline
       * that a latency reconfiguration is possible/needed. */
      if (new_latency != old_latency) {
        GST_DEBUG_OBJECT (adder, "latency changed to: %" GST_TIME_FORMAT,
            GST_TIME_ARGS (new_latency));

        gst_element_post_message (GST_ELEMENT_CAST (adder),
            gst_message_new_latency (GST_OBJECT_CAST (adder)));
      }
      break;
    }
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}


static void
gst_live_adder_get_property (GObject * object,
    guint prop_id, GValue * value, GParamSpec * pspec)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (object);

  switch (prop_id) {
    case PROP_LATENCY:
      GST_OBJECT_LOCK (adder);
      g_value_set_uint (value, adder->latency_ms);
      GST_OBJECT_UNLOCK (adder);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}


/* we can only accept caps that we and downstream can handle. */
static GstCaps *
gst_live_adder_sink_getcaps (GstLiveAdder * adder, GstPad * pad,
    GstCaps * filter)
{
  GstCaps *result, *peercaps, *sinkcaps;

  /* get the downstream possible caps */
  peercaps = gst_pad_peer_query_caps (adder->srcpad, filter);
  /* get the allowed caps on this sinkpad, we use the fixed caps function so
   * that it does not call recursively in this function. */
  sinkcaps = gst_pad_get_current_caps (pad);
  if (!sinkcaps)
    sinkcaps = gst_pad_get_pad_template_caps (pad);
  if (peercaps) {
    /* if the peer has caps, intersect */
    GST_DEBUG_OBJECT (adder, "intersecting peer and template caps");
    result = gst_caps_intersect (peercaps, sinkcaps);
    gst_caps_unref (sinkcaps);
    gst_caps_unref (peercaps);
  } else {
    /* the peer has no caps (or there is no peer), just use the allowed caps
     * of this sinkpad. */
    GST_DEBUG_OBJECT (adder, "no peer caps, using sinkcaps");
    result = sinkcaps;
  }

  return result;
}

struct SetCapsIterCtx
{
  GstPad *pad;
  GstCaps *caps;
  gboolean all_valid;
};

static void
check_other_caps (const GValue * item, gpointer user_data)
{
  GstPad *otherpad = GST_PAD (g_value_get_object (item));
  struct SetCapsIterCtx *ctx = user_data;

  if (otherpad == ctx->pad)
    return;

  if (!gst_pad_peer_query_accept_caps (otherpad, ctx->caps))
    ctx->all_valid = FALSE;
}

static void
set_other_caps (const GValue * item, gpointer user_data)
{
  GstPad *otherpad = GST_PAD (g_value_get_object (item));
  struct SetCapsIterCtx *ctx = user_data;

  if (otherpad == ctx->pad)
    return;

  if (!gst_pad_set_caps (otherpad, ctx->caps))
    ctx->all_valid = FALSE;
}

/* the first caps we receive on any of the sinkpads will define the caps for all
 * the other sinkpads because we can only mix streams with the same caps.
 * */
static gboolean
gst_live_adder_setcaps (GstLiveAdder * adder, GstPad * pad, GstCaps * caps)
{
  GstIterator *iter;
  struct SetCapsIterCtx ctx;

  GST_LOG_OBJECT (adder, "setting caps on pad %p,%s to %" GST_PTR_FORMAT, pad,
      GST_PAD_NAME (pad), caps);

  /* FIXME, see if the other pads can accept the format. Also lock the
   * format on the other pads to this new format. */
  iter = gst_element_iterate_sink_pads (GST_ELEMENT (adder));
  ctx.pad = pad;
  ctx.caps = caps;
  ctx.all_valid = TRUE;
  while (gst_iterator_foreach (iter, check_other_caps, &ctx) ==
      GST_ITERATOR_RESYNC) {
    ctx.all_valid = TRUE;
    gst_iterator_resync (iter);
  }
  if (!ctx.all_valid) {
    GST_WARNING_OBJECT (adder, "Caps are not acceptable by other sinkpads");
    gst_iterator_free (iter);
    return FALSE;
  }

  while (gst_iterator_foreach (iter, set_other_caps, &ctx) ==
      GST_ITERATOR_RESYNC) {
    ctx.all_valid = TRUE;
    gst_iterator_resync (iter);
  }
  gst_iterator_free (iter);

  if (!ctx.all_valid) {
    GST_WARNING_OBJECT (adder, "Could not set caps on the other sink pads");
    return FALSE;
  }

  if (!gst_pad_set_caps (adder->srcpad, caps)) {
    GST_WARNING_OBJECT (adder, "Could not set caps downstream");
    return FALSE;
  }

  GST_OBJECT_LOCK (adder);
  /* parse caps now */
  if (!gst_audio_info_from_caps (&adder->info, caps))
    goto not_supported;

  if (GST_AUDIO_INFO_IS_INTEGER (&adder->info)) {
    switch (GST_AUDIO_INFO_WIDTH (&adder->info)) {
      case 8:
        adder->func = GST_AUDIO_INFO_IS_SIGNED (&adder->info) ?
            (GstLiveAdderFunction) add_int8 : (GstLiveAdderFunction) add_uint8;
        break;
      case 16:
        adder->func = GST_AUDIO_INFO_IS_SIGNED (&adder->info) ?
            (GstLiveAdderFunction) add_int16 : (GstLiveAdderFunction)
            add_uint16;
        break;
      case 32:
        adder->func = GST_AUDIO_INFO_IS_SIGNED (&adder->info) ?
            (GstLiveAdderFunction) add_int32 : (GstLiveAdderFunction)
            add_uint32;
        break;
      default:
        goto not_supported;
    }
  } else if (GST_AUDIO_INFO_IS_FLOAT (&adder->info)) {
    switch (GST_AUDIO_INFO_WIDTH (&adder->info)) {
      case 32:
        adder->func = (GstLiveAdderFunction) add_float32;
        break;
      case 64:
        adder->func = (GstLiveAdderFunction) add_float64;
        break;
      default:
        goto not_supported;
    }
  } else {
    goto not_supported;
  }

  GST_OBJECT_UNLOCK (adder);
  return TRUE;

  /* ERRORS */
not_supported:
  {
    GST_OBJECT_UNLOCK (adder);
    GST_DEBUG_OBJECT (adder, "unsupported format set as caps");
    return FALSE;
  }
}

static void
gst_live_adder_flush_start (GstLiveAdder * adder)
{
  GST_DEBUG_OBJECT (adder, "Disabling pop on queue");

  GST_OBJECT_LOCK (adder);
  /* mark ourselves as flushing */
  adder->srcresult = GST_FLOW_FLUSHING;

  /* Empty the queue */
  g_queue_foreach (adder->buffers, (GFunc) gst_mini_object_unref, NULL);
  while (g_queue_pop_head (adder->buffers));

  /* unlock clock, we just unschedule, the entry will be released by the
   * locking streaming thread. */
  if (adder->clock_id)
    gst_clock_id_unschedule (adder->clock_id);

  g_cond_broadcast (&adder->not_empty_cond);
  GST_OBJECT_UNLOCK (adder);
}

static gboolean
gst_live_adder_src_activate_mode (GstPad * pad, GstObject * parent,
    GstPadMode mode, gboolean active)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  gboolean result = TRUE;

  if (mode == GST_PAD_MODE_PULL)
    return FALSE;

  if (active) {
    /* Mark as non flushing */
    GST_OBJECT_LOCK (adder);
    adder->srcresult = GST_FLOW_OK;
    GST_OBJECT_UNLOCK (adder);

    /* start pushing out buffers */
    GST_DEBUG_OBJECT (adder, "Starting task on srcpad");
    gst_pad_start_task (adder->srcpad,
        (GstTaskFunction) gst_live_adder_loop, adder, NULL);
  } else {
    /* make sure all data processing stops ASAP */
    gst_live_adder_flush_start (adder);

    /* NOTE this will hardlock if the state change is called from the src pad
     * task thread because we will _join() the thread. */
    GST_DEBUG_OBJECT (adder, "Stopping task on srcpad");
    result = gst_pad_stop_task (pad);
  }

  return result;
}

static gboolean
gst_live_adder_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  GstLiveAdderPadPrivate *padprivate = NULL;
  gboolean ret = TRUE;

  padprivate = gst_pad_get_element_private (pad);

  if (!padprivate)
    return FALSE;

  GST_LOG_OBJECT (adder, "received %s", GST_EVENT_TYPE_NAME (event));

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_CAPS:
    {
      GstCaps *caps;

      gst_event_parse_caps (event, &caps);
      ret = gst_live_adder_setcaps (adder, pad, caps);
      gst_event_unref (event);
      break;
    }
    case GST_EVENT_SEGMENT:
    {
      const GstSegment *segment;
      GstSegment livesegment;

      gst_event_parse_segment (event, &segment);

      /* we need time for now */
      if (segment->format != GST_FORMAT_TIME)
        goto newseg_wrong_format;

      /* now configure the values, we need these to time the release of the
       * buffers on the srcpad. */
      GST_OBJECT_LOCK (adder);
      gst_segment_copy_into (segment, &padprivate->segment);
      GST_OBJECT_UNLOCK (adder);
      gst_event_unref (event);

      gst_segment_init (&livesegment, GST_FORMAT_TIME);
      gst_pad_push_event (adder->srcpad, gst_event_new_segment (&livesegment));
      break;
    }
    case GST_EVENT_FLUSH_START:
      gst_live_adder_flush_start (adder);
      ret = gst_pad_push_event (adder->srcpad, event);
      break;
    case GST_EVENT_FLUSH_STOP:
      GST_OBJECT_LOCK (adder);
      adder->next_timestamp = GST_CLOCK_TIME_NONE;
      reset_pad_private (pad);
      GST_OBJECT_UNLOCK (adder);
      ret = gst_pad_push_event (adder->srcpad, event);
      ret = gst_live_adder_src_activate_mode (adder->srcpad, GST_OBJECT (adder),
          GST_PAD_MODE_PUSH, TRUE);
      break;
    case GST_EVENT_EOS:
    {
      GST_OBJECT_LOCK (adder);

      ret = adder->srcresult == GST_FLOW_OK;
      if (ret && !padprivate->eos) {
        GST_DEBUG_OBJECT (adder, "queuing EOS");
        padprivate->eos = TRUE;
        g_cond_broadcast (&adder->not_empty_cond);
      } else if (padprivate->eos) {
        GST_DEBUG_OBJECT (adder, "dropping EOS, we are already EOS");
      } else {
        GST_DEBUG_OBJECT (adder, "dropping EOS, reason %s",
            gst_flow_get_name (adder->srcresult));
      }

      GST_OBJECT_UNLOCK (adder);

      gst_event_unref (event);
      break;
    }
    default:
      ret = gst_pad_push_event (adder->srcpad, event);
      break;
  }

done:

  return ret;

  /* ERRORS */
newseg_wrong_format:
  {
    GST_DEBUG_OBJECT (adder, "received non TIME segment");
    ret = FALSE;
    goto done;
  }
}

static gboolean
gst_live_adder_query_pos_dur (GstLiveAdder * adder, GstFormat format,
    gboolean position, gint64 * outvalue)
{
  GValue item = { 0 };
  gint64 max = G_MININT64;
  gboolean res = TRUE;
  GstIterator *it;
  gboolean done = FALSE;

  it = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (adder));
  while (!done) {
    switch (gst_iterator_next (it, &item)) {
      case GST_ITERATOR_DONE:
        done = TRUE;
        break;
      case GST_ITERATOR_OK:
      {
        GstPad *pad = GST_PAD_CAST (g_value_get_object (&item));
        gint64 value;
        gboolean curres;

        /* ask sink peer for duration */
        if (position)
          curres = gst_pad_peer_query_position (pad, format, &value);
        else
          curres = gst_pad_peer_query_duration (pad, format, &value);

        /* take max from all valid return values */
        /* Only if the format is the one we requested, otherwise ignore it ?
         */

        if (curres) {
          res &= curres;

          /* valid unknown length, stop searching */
          if (value == -1) {
            max = value;
            done = TRUE;
          } else if (value > max) {
            max = value;
          }
        }
        g_value_reset (&item);
        break;
      }
      case GST_ITERATOR_RESYNC:
        max = -1;
        res = TRUE;
        break;
      default:
        res = FALSE;
        done = TRUE;
        break;
    }
  }

  g_value_unset (&item);
  gst_iterator_free (it);

  if (res)
    *outvalue = max;

  return res;
}

/* FIXME:
 *
 * When we add a new stream (or remove a stream) the duration might
 * also become invalid again and we need to post a new DURATION
 * message to notify this fact to the parent.
 * For now we take the max of all the upstream elements so the simple
 * cases work at least somewhat.
 */
static gboolean
gst_live_adder_query_duration (GstLiveAdder * adder, GstQuery * query)
{
  GstFormat format;
  gint64 max;
  gboolean res;

  /* parse format */
  gst_query_parse_duration (query, &format, NULL);

  res = gst_live_adder_query_pos_dur (adder, format, FALSE, &max);

  if (res) {
    /* and store the max */
    gst_query_set_duration (query, format, max);
  }

  return res;
}

static gboolean
gst_live_adder_query_position (GstLiveAdder * adder, GstQuery * query)
{
  GstFormat format;
  gint64 max;
  gboolean res;

  /* parse format */
  gst_query_parse_position (query, &format, NULL);

  res = gst_live_adder_query_pos_dur (adder, format, TRUE, &max);

  if (res) {
    /* and store the max */
    gst_query_set_position (query, format, max);
  }

  return res;
}



static gboolean
gst_live_adder_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  gboolean res = FALSE;

  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_LATENCY:
    {
      /* We need to send the query upstream and add the returned latency to our
       * own */
      res = gst_pad_query_default (pad, parent, query);

      if (res) {
        GstClockTime my_latency = adder->latency_ms * GST_MSECOND;
        GstClockTime min_latency, max_latency;
        gboolean live;

        gst_query_parse_latency (query, &live, &min_latency, &max_latency);

        GST_OBJECT_LOCK (adder);
        adder->peer_latency = min_latency;
        min_latency += my_latency;
        GST_OBJECT_UNLOCK (adder);

        /* Make sure we don't risk an overflow */
        if (max_latency < G_MAXUINT64 - my_latency)
          max_latency += my_latency;
        else
          max_latency = G_MAXUINT64;
        gst_query_set_latency (query, TRUE, min_latency, max_latency);
        GST_DEBUG_OBJECT (adder, "Calculated total latency : min %"
            GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
            GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
      }
      break;
    }
    case GST_QUERY_DURATION:
      res = gst_live_adder_query_duration (adder, query);
      break;
    case GST_QUERY_POSITION:
      res = gst_live_adder_query_position (adder, query);
      break;
    default:
      res = gst_pad_query_default (pad, parent, query);
      break;
  }

  return res;
}

static gboolean
gst_live_adder_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  gboolean res;

  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_CAPS:
    {
      GstCaps *filter;
      GstCaps *result;

      gst_query_parse_caps (query, &filter);
      result = gst_live_adder_sink_getcaps (adder, pad, filter);
      gst_query_set_caps_result (query, result);
      gst_caps_unref (result);
      res = TRUE;
      break;
    }
    default:
      res = gst_pad_query_default (pad, parent, query);
      break;
  }

  return res;
}

static gboolean
forward_event_func (const GValue * item, GValue * ret, gpointer user_data)
{
  GstPad *pad = GST_PAD (g_value_get_object (item));
  GstEvent *event = user_data;

  gst_event_ref (event);
  GST_LOG_OBJECT (pad, "About to send event %s", GST_EVENT_TYPE_NAME (event));
  if (!gst_pad_push_event (pad, event)) {
    g_value_set_boolean (ret, FALSE);
    GST_WARNING_OBJECT (pad, "Sending event  %p (%s) failed.",
        event, GST_EVENT_TYPE_NAME (event));
  } else {
    GST_LOG_OBJECT (pad, "Sent event  %p (%s).",
        event, GST_EVENT_TYPE_NAME (event));
  }
  return TRUE;
}

/* forwards the event to all sinkpads, takes ownership of the
 * event
 *
 * Returns: TRUE if the event could be forwarded on all
 * sinkpads.
 */
static gboolean
forward_event (GstLiveAdder * adder, GstEvent * event)
{
  gboolean ret;
  GstIterator *it;
  GValue vret = { 0 };

  GST_LOG_OBJECT (adder, "Forwarding event %p (%s)", event,
      GST_EVENT_TYPE_NAME (event));

  ret = TRUE;

  g_value_init (&vret, G_TYPE_BOOLEAN);
  g_value_set_boolean (&vret, TRUE);
  it = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (adder));
  gst_iterator_fold (it, forward_event_func, &vret, event);
  gst_iterator_free (it);

  ret = g_value_get_boolean (&vret);

  return ret;
}


static gboolean
gst_live_adder_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  gboolean result;

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_QOS:
      /* TODO : QoS might be tricky */
      result = FALSE;
      break;
    case GST_EVENT_NAVIGATION:
      /* TODO : navigation is rather pointless. */
      result = FALSE;
      break;
    default:
      /* just forward the rest for now */
      result = forward_event (adder, event);
      break;
  }

  gst_event_unref (event);

  return result;
}

static guint
gst_live_adder_length_from_duration (GstLiveAdder * adder,
    GstClockTime duration)
{
  guint64 ret = GST_AUDIO_INFO_BPF (&adder->info) *
      gst_util_uint64_scale_int_round (duration,
      GST_AUDIO_INFO_RATE (&adder->info), GST_SECOND);

  return (guint) ret;
}

static GstFlowReturn
gst_live_live_adder_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (parent);
  GstLiveAdderPadPrivate *padprivate = NULL;
  GstFlowReturn ret = GST_FLOW_OK;
  GList *item = NULL;
  GstClockTime skip = 0;
  gint64 drift = 0;             /* Positive if new buffer after old buffer */

  GST_OBJECT_LOCK (adder);

  ret = adder->srcresult;

  GST_DEBUG ("Incoming buffer time:%" GST_TIME_FORMAT " duration:%"
      GST_TIME_FORMAT, GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
      GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));

  if (ret != GST_FLOW_OK) {
    GST_DEBUG_OBJECT (adder, "Passing non-ok result from src: %s",
        gst_flow_get_name (ret));
    gst_buffer_unref (buffer);
    goto out;
  }

  padprivate = gst_pad_get_element_private (pad);

  if (!padprivate) {
    ret = GST_FLOW_NOT_LINKED;
    gst_buffer_unref (buffer);
    goto out;
  }

  if (padprivate->eos) {
    GST_DEBUG_OBJECT (adder, "Received buffer after EOS");
    ret = GST_FLOW_EOS;
    gst_buffer_unref (buffer);
    goto out;
  }

  if (!GST_BUFFER_TIMESTAMP_IS_VALID (buffer))
    goto invalid_timestamp;

  if (padprivate->segment.format == GST_FORMAT_UNDEFINED) {
    GST_WARNING_OBJECT (adder, "No new-segment received,"
        " initializing segment with time 0..-1");
    gst_segment_init (&padprivate->segment, GST_FORMAT_TIME);
  }

  buffer = gst_buffer_make_writable (buffer);

  drift = GST_BUFFER_TIMESTAMP (buffer) - padprivate->expected_timestamp;

  /* Just see if we receive invalid timestamp/durations */
  if (GST_CLOCK_TIME_IS_VALID (padprivate->expected_timestamp) &&
      !GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DISCONT) &&
      (drift != 0)) {
    GST_LOG_OBJECT (adder,
        "Timestamp discontinuity without the DISCONT flag set"
        " (expected %" GST_TIME_FORMAT ", got %" GST_TIME_FORMAT
        " drift:%" G_GINT64_FORMAT "ms)",
        GST_TIME_ARGS (padprivate->expected_timestamp),
        GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)), drift / GST_MSECOND);

    /* We accept drifts of 10ms */
    if (ABS (drift) < (10 * GST_MSECOND)) {
      GST_DEBUG ("Correcting minor drift");
      GST_BUFFER_TIMESTAMP (buffer) = padprivate->expected_timestamp;
    }
  }


  /* If there is no duration, lets set one */
  if (!GST_BUFFER_DURATION_IS_VALID (buffer)) {
    GST_BUFFER_DURATION (buffer) = (gst_buffer_get_size (buffer) * GST_SECOND) /
        (GST_AUDIO_INFO_BPF (&adder->info) *
        GST_AUDIO_INFO_RATE (&adder->info));
    padprivate->expected_timestamp = GST_CLOCK_TIME_NONE;
  } else {
    padprivate->expected_timestamp = GST_BUFFER_TIMESTAMP (buffer) +
        GST_BUFFER_DURATION (buffer);
  }


  /*
   * Lets clip the buffer to the segment (so we don't have to worry about
   * cliping afterwards).
   * This should also guarantee us that we'll have valid timestamps and
   * durations afterwards
   */

  buffer = gst_audio_buffer_clip (buffer, &padprivate->segment,
      GST_AUDIO_INFO_RATE (&adder->info), GST_AUDIO_INFO_BPF (&adder->info));

  /* buffer can be NULL if it's completely outside of the segment */
  if (!buffer) {
    GST_DEBUG ("Buffer completely outside of configured segment, dropping it");
    goto out;
  }

  /*
   * Make sure all incoming buffers share the same timestamping
   */
  GST_BUFFER_TIMESTAMP (buffer) =
      gst_segment_to_running_time (&padprivate->segment,
      padprivate->segment.format, GST_BUFFER_TIMESTAMP (buffer));


  if (GST_CLOCK_TIME_IS_VALID (adder->next_timestamp) &&
      GST_BUFFER_TIMESTAMP (buffer) < adder->next_timestamp) {
    if (GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer) <
        adder->next_timestamp) {
      GST_DEBUG_OBJECT (adder, "Buffer is late, dropping (ts: %" GST_TIME_FORMAT
          " duration: %" GST_TIME_FORMAT ")",
          GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
          GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
      gst_buffer_unref (buffer);
      goto out;
    } else {
      skip = adder->next_timestamp - GST_BUFFER_TIMESTAMP (buffer);
      GST_DEBUG_OBJECT (adder, "Buffer is partially late, skipping %"
          GST_TIME_FORMAT, GST_TIME_ARGS (skip));
    }
  }

  /* If our new buffer's head is higher than the queue's head, lets wake up,
   * we may not have to wait for as long
   */
  if (adder->clock_id &&
      g_queue_peek_head (adder->buffers) != NULL &&
      GST_BUFFER_TIMESTAMP (buffer) + skip <
      GST_BUFFER_TIMESTAMP (g_queue_peek_head (adder->buffers)))
    gst_clock_id_unschedule (adder->clock_id);

  for (item = g_queue_peek_head_link (adder->buffers);
      item; item = g_list_next (item)) {
    GstBuffer *oldbuffer = item->data;
    GstClockTime old_skip = 0;
    GstClockTime mix_duration = 0;
    GstClockTime mix_start = 0;
    GstClockTime mix_end = 0;
    GstMapInfo oldmap, map;

    /* We haven't reached our place yet */
    if (GST_BUFFER_TIMESTAMP (buffer) + skip >=
        GST_BUFFER_TIMESTAMP (oldbuffer) + GST_BUFFER_DURATION (oldbuffer))
      continue;

    /* We're past our place, lets insert ouselves here */
    if (GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer) <=
        GST_BUFFER_TIMESTAMP (oldbuffer))
      break;

    /* if we reach this spot, we have overlap, so we must mix */

    /* First make a subbuffer with the non-overlapping part */
    if (GST_BUFFER_TIMESTAMP (buffer) + skip < GST_BUFFER_TIMESTAMP (oldbuffer)) {
      GstBuffer *subbuffer = NULL;
      GstClockTime subbuffer_duration = GST_BUFFER_TIMESTAMP (oldbuffer) -
          (GST_BUFFER_TIMESTAMP (buffer) + skip);

      subbuffer = gst_buffer_copy_region (buffer, GST_BUFFER_COPY_ALL,
          gst_live_adder_length_from_duration (adder, skip),
          gst_live_adder_length_from_duration (adder, subbuffer_duration));

      GST_BUFFER_TIMESTAMP (subbuffer) = GST_BUFFER_TIMESTAMP (buffer) + skip;
      GST_BUFFER_DURATION (subbuffer) = subbuffer_duration;

      skip += subbuffer_duration;

      g_queue_insert_before (adder->buffers, item, subbuffer);
    }

    /* Now we are on the overlapping part */
    oldbuffer = gst_buffer_make_writable (oldbuffer);
    item->data = oldbuffer;

    old_skip = GST_BUFFER_TIMESTAMP (buffer) + skip -
        GST_BUFFER_TIMESTAMP (oldbuffer);

    mix_start = GST_BUFFER_TIMESTAMP (oldbuffer) + old_skip;

    if (GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer) <
        GST_BUFFER_TIMESTAMP (oldbuffer) + GST_BUFFER_DURATION (oldbuffer))
      mix_end = GST_BUFFER_TIMESTAMP (buffer) + GST_BUFFER_DURATION (buffer);
    else
      mix_end = GST_BUFFER_TIMESTAMP (oldbuffer) +
          GST_BUFFER_DURATION (oldbuffer);

    mix_duration = mix_end - mix_start;

    if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_GAP)) {
      GST_BUFFER_FLAG_UNSET (oldbuffer, GST_BUFFER_FLAG_GAP);
      gst_buffer_map (oldbuffer, &oldmap, GST_MAP_WRITE);
      gst_buffer_map (buffer, &map, GST_MAP_READ);
      adder->func (oldmap.data +
          gst_live_adder_length_from_duration (adder, old_skip),
          map.data +
          gst_live_adder_length_from_duration (adder, skip),
          gst_live_adder_length_from_duration (adder, mix_duration));
      gst_buffer_unmap (oldbuffer, &oldmap);
      gst_buffer_unmap (buffer, &map);
    }
    skip += mix_duration;
  }

  g_cond_broadcast (&adder->not_empty_cond);

  if (skip == GST_BUFFER_DURATION (buffer)) {
    gst_buffer_unref (buffer);
  } else {
    if (skip) {
      GstClockTime subbuffer_duration = GST_BUFFER_DURATION (buffer) - skip;
      GstClockTime subbuffer_ts = GST_BUFFER_TIMESTAMP (buffer) + skip;
      GstBuffer *new_buffer = gst_buffer_copy_region (buffer,
          GST_BUFFER_COPY_ALL,
          gst_live_adder_length_from_duration (adder, skip),
          gst_live_adder_length_from_duration (adder, subbuffer_duration));
      gst_buffer_unref (buffer);
      buffer = new_buffer;
      GST_BUFFER_PTS (buffer) = subbuffer_ts;
      GST_BUFFER_DURATION (buffer) = subbuffer_duration;
    }

    if (item)
      g_queue_insert_before (adder->buffers, item, buffer);
    else
      g_queue_push_tail (adder->buffers, buffer);
  }

out:

  GST_OBJECT_UNLOCK (adder);

  return ret;

invalid_timestamp:

  GST_OBJECT_UNLOCK (adder);
  gst_buffer_unref (buffer);
  GST_ELEMENT_ERROR (adder, STREAM, FAILED,
      ("Buffer without a valid timestamp received"),
      ("Invalid timestamp received on buffer"));

  return GST_FLOW_ERROR;
}

/*
 * This only works because the GstObject lock is taken
 *
 * It checks if all sink pads are EOS
 */
static gboolean
check_eos_locked (GstLiveAdder * adder)
{
  GList *item;

  /* We can't be EOS if we have no sinkpads */
  if (adder->sinkpads == NULL)
    return FALSE;

  for (item = adder->sinkpads; item; item = g_list_next (item)) {
    GstPad *pad = item->data;
    GstLiveAdderPadPrivate *padprivate = gst_pad_get_element_private (pad);

    if (padprivate && padprivate->eos != TRUE)
      return FALSE;
  }

  return TRUE;
}

static void
gst_live_adder_loop (gpointer data)
{
  GstLiveAdder *adder = GST_LIVE_ADDER (data);
  GstClockTime buffer_timestamp = 0;
  GstClockTime sync_time = 0;
  GstClock *clock = NULL;
  GstClockID id = NULL;
  GstClockReturn ret;
  GstBuffer *buffer = NULL;
  GstFlowReturn result;

  GST_OBJECT_LOCK (adder);

again:

  for (;;) {
    if (adder->srcresult != GST_FLOW_OK)
      goto flushing;
    if (!g_queue_is_empty (adder->buffers))
      break;
    if (check_eos_locked (adder))
      goto eos;
    g_cond_wait (&adder->not_empty_cond, GST_OBJECT_GET_LOCK (adder));
  }

  buffer_timestamp = GST_BUFFER_TIMESTAMP (g_queue_peek_head (adder->buffers));

  clock = GST_ELEMENT_CLOCK (adder);

  /* If we have no clock, then we can't do anything.. error */
  if (!clock) {
    if (adder->playing)
      goto no_clock;
    else
      goto push_buffer;
  }

  GST_DEBUG_OBJECT (adder, "sync to timestamp %" GST_TIME_FORMAT,
      GST_TIME_ARGS (buffer_timestamp));

  sync_time = buffer_timestamp + GST_ELEMENT_CAST (adder)->base_time;
  /* add latency, this includes our own latency and the peer latency. */
  sync_time += adder->latency_ms * GST_MSECOND;
  sync_time += adder->peer_latency;

  /* create an entry for the clock */
  id = adder->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
  GST_OBJECT_UNLOCK (adder);

  ret = gst_clock_id_wait (id, NULL);

  GST_OBJECT_LOCK (adder);

  /* and free the entry */
  gst_clock_id_unref (id);
  adder->clock_id = NULL;

  /* at this point, the clock could have been unlocked by a timeout, a new
   * head element was added to the queue or because we are shutting down. Check
   * for shutdown first. */

  if (adder->srcresult != GST_FLOW_OK)
    goto flushing;

  if (ret == GST_CLOCK_UNSCHEDULED) {
    GST_DEBUG_OBJECT (adder,
        "Wait got unscheduled, will retry to push with new buffer");
    goto again;
  }

  if (ret != GST_CLOCK_OK && ret != GST_CLOCK_EARLY)
    goto clock_error;

push_buffer:

  buffer = g_queue_pop_head (adder->buffers);

  if (!buffer)
    goto again;

  /*
   * We make sure the timestamps are exactly contiguous
   * If its only small skew (due to rounding errors), we correct it
   * silently. Otherwise we put the discont flag
   */
  if (GST_CLOCK_TIME_IS_VALID (adder->next_timestamp) &&
      GST_BUFFER_TIMESTAMP (buffer) != adder->next_timestamp) {
    GstClockTimeDiff diff = GST_CLOCK_DIFF (GST_BUFFER_TIMESTAMP (buffer),
        adder->next_timestamp);
    if (diff < 0)
      diff = -diff;

    if (diff < GST_SECOND / GST_AUDIO_INFO_RATE (&adder->info)) {
      GST_BUFFER_TIMESTAMP (buffer) = adder->next_timestamp;
      GST_DEBUG_OBJECT (adder, "Correcting slight skew");
      GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
    } else {
      GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
      GST_DEBUG_OBJECT (adder, "Expected buffer at %" GST_TIME_FORMAT
          ", but is at %" GST_TIME_FORMAT ", setting discont",
          GST_TIME_ARGS (adder->next_timestamp),
          GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)));
    }
  } else {
    GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
  }

  GST_BUFFER_OFFSET (buffer) = GST_BUFFER_OFFSET_NONE;
  GST_BUFFER_OFFSET_END (buffer) = GST_BUFFER_OFFSET_NONE;

  if (GST_BUFFER_DURATION_IS_VALID (buffer))
    adder->next_timestamp = GST_BUFFER_TIMESTAMP (buffer) +
        GST_BUFFER_DURATION (buffer);
  else
    adder->next_timestamp = GST_CLOCK_TIME_NONE;
  GST_OBJECT_UNLOCK (adder);

  GST_LOG_OBJECT (adder, "About to push buffer time:%" GST_TIME_FORMAT
      " duration:%" GST_TIME_FORMAT,
      GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
      GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));

  result = gst_pad_push (adder->srcpad, buffer);
  if (result != GST_FLOW_OK)
    goto pause;

  return;

flushing:
  {
    GST_DEBUG_OBJECT (adder, "we are flushing");
    gst_pad_pause_task (adder->srcpad);
    GST_OBJECT_UNLOCK (adder);
    return;
  }

clock_error:
  {
    gst_pad_pause_task (adder->srcpad);
    GST_OBJECT_UNLOCK (adder);
    GST_ELEMENT_ERROR (adder, STREAM, MUX, ("Error with the clock"),
        ("Error with the clock: %d", ret));
    GST_ERROR_OBJECT (adder, "Error with the clock: %d", ret);
    return;
  }

no_clock:
  {
    gst_pad_pause_task (adder->srcpad);
    GST_OBJECT_UNLOCK (adder);
    GST_ELEMENT_ERROR (adder, STREAM, MUX, ("No available clock"),
        ("No available clock"));
    GST_ERROR_OBJECT (adder, "No available clock");
    return;
  }

pause:
  {
    GST_DEBUG_OBJECT (adder, "pausing task, reason %s",
        gst_flow_get_name (result));

    GST_OBJECT_LOCK (adder);

    /* store result */
    adder->srcresult = result;
    /* we don't post errors or anything because upstream will do that for us
     * when we pass the return value upstream. */
    gst_pad_pause_task (adder->srcpad);
    GST_OBJECT_UNLOCK (adder);
    return;
  }

eos:
  {
    /* store result, we are flushing now */
    GST_DEBUG_OBJECT (adder, "We are EOS, pushing EOS downstream");
    adder->srcresult = GST_FLOW_EOS;
    gst_pad_pause_task (adder->srcpad);
    GST_OBJECT_UNLOCK (adder);
    gst_pad_push_event (adder->srcpad, gst_event_new_eos ());
    return;
  }
}

static GstPad *
gst_live_adder_request_new_pad (GstElement * element, GstPadTemplate * templ,
    const gchar * ignored_name, const GstCaps * caps)
{
  gchar *name;
  GstLiveAdder *adder;
  GstPad *newpad;
  gint padcount;
  GstLiveAdderPadPrivate *padprivate = NULL;

  if (templ->direction != GST_PAD_SINK)
    goto not_sink;

  adder = GST_LIVE_ADDER (element);

  /* increment pad counter */
  padcount = g_atomic_int_add (&adder->padcount, 1);

  name = g_strdup_printf ("sink_%u", padcount);
  newpad = gst_pad_new_from_template (templ, name);
  GST_DEBUG_OBJECT (adder, "request new pad %s", name);
  g_free (name);

  gst_pad_set_event_function (newpad,
      GST_DEBUG_FUNCPTR (gst_live_adder_sink_event));
  gst_pad_set_query_function (newpad,
      GST_DEBUG_FUNCPTR (gst_live_adder_sink_query));

  padprivate = g_new0 (GstLiveAdderPadPrivate, 1);

  gst_segment_init (&padprivate->segment, GST_FORMAT_UNDEFINED);
  padprivate->eos = FALSE;
  padprivate->expected_timestamp = GST_CLOCK_TIME_NONE;

  gst_pad_set_element_private (newpad, padprivate);

  gst_pad_set_chain_function (newpad, gst_live_live_adder_chain);


  if (!gst_pad_set_active (newpad, TRUE))
    goto could_not_activate;

  /* takes ownership of the pad */
  if (!gst_element_add_pad (GST_ELEMENT (adder), newpad))
    goto could_not_add;

  GST_OBJECT_LOCK (adder);
  adder->sinkpads = g_list_prepend (adder->sinkpads, newpad);
  GST_OBJECT_UNLOCK (adder);

  return newpad;

  /* errors */
not_sink:
  {
    g_warning ("gstadder: request new pad that is not a SINK pad\n");
    return NULL;
  }
could_not_add:
  {
    GST_DEBUG_OBJECT (adder, "could not add pad");
    g_free (padprivate);
    gst_object_unref (newpad);
    return NULL;
  }
could_not_activate:
  {
    GST_DEBUG_OBJECT (adder, "could not activate new pad");
    g_free (padprivate);
    gst_object_unref (newpad);
    return NULL;
  }
}

static void
gst_live_adder_release_pad (GstElement * element, GstPad * pad)
{
  GstLiveAdder *adder;
  GstLiveAdderPadPrivate *padprivate;

  adder = GST_LIVE_ADDER (element);

  GST_DEBUG_OBJECT (adder, "release pad %s:%s", GST_DEBUG_PAD_NAME (pad));

  GST_OBJECT_LOCK (element);
  padprivate = gst_pad_get_element_private (pad);
  gst_pad_set_element_private (pad, NULL);
  adder->sinkpads = g_list_remove_all (adder->sinkpads, pad);
  GST_OBJECT_UNLOCK (element);

  g_free (padprivate);

  gst_element_remove_pad (element, pad);
}

static void
reset_pad_private (GstPad * pad)
{
  GstLiveAdderPadPrivate *padprivate;

  padprivate = gst_pad_get_element_private (pad);

  if (!padprivate)
    return;

  gst_segment_init (&padprivate->segment, GST_FORMAT_UNDEFINED);

  padprivate->expected_timestamp = GST_CLOCK_TIME_NONE;
  padprivate->eos = FALSE;
}

static GstStateChangeReturn
gst_live_adder_change_state (GstElement * element, GstStateChange transition)
{
  GstLiveAdder *adder;
  GstStateChangeReturn ret;

  adder = GST_LIVE_ADDER (element);

  switch (transition) {
    case GST_STATE_CHANGE_READY_TO_PAUSED:
      GST_OBJECT_LOCK (adder);
      adder->segment_pending = TRUE;
      adder->peer_latency = 0;
      adder->next_timestamp = GST_CLOCK_TIME_NONE;
      g_list_foreach (adder->sinkpads, (GFunc) reset_pad_private, NULL);
      GST_OBJECT_UNLOCK (adder);
      break;
    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
      GST_OBJECT_LOCK (adder);
      adder->playing = FALSE;
      GST_OBJECT_UNLOCK (adder);
      break;
    default:
      break;
  }

  ret = GST_ELEMENT_CLASS (gst_live_adder_parent_class)->change_state (element,
      transition);

  switch (transition) {
    case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
      GST_OBJECT_LOCK (adder);
      adder->playing = TRUE;
      GST_OBJECT_UNLOCK (adder);
      break;
    default:
      break;
  }

  return ret;
}


static gboolean
plugin_init (GstPlugin * plugin)
{
  if (!gst_element_register (plugin, "liveadder", GST_RANK_NONE,
          GST_TYPE_LIVE_ADDER)) {
    return FALSE;
  }

  return TRUE;
}

GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
    GST_VERSION_MINOR,
    liveadder,
    "Adds multiple live discontinuous streams",
    plugin_init, VERSION, GST_LICENSE, GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN)