path: root/subprojects/gst-plugins-base/gst/playback/gstdecodebin3.c

/* GStreamer
 *
 * Copyright (C) <2015> Centricular Ltd
 *  @author: Edward Hervey <edward@centricular.com>
 *  @author: Jan Schmidt <jan@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 <glib.h>
#include <glib-object.h>
#include <glib/gprintf.h>
#include <gst/gst.h>
#include <gst/pbutils/pbutils.h>

#include "gstplaybackelements.h"
#include "gstplay-enum.h"
#include "gstrawcaps.h"

/**
 * SECTION:element-decodebin3
 * @title: decodebin3
 *
 * #GstBin that auto-magically constructs a decoding pipeline using available
 * decoders and demuxers via auto-plugging. The output is raw audio, video
 * or subtitle streams.
 *
 * decodebin3 differs from the previous decodebin (decodebin2) in important ways:
 *
 * * supports publication and selection of stream information via
 * GstStreamCollection messages and #GST_EVENT_SELECT_STREAMS events.
 *
 * * dynamically switches stream connections internally, and
 * reuses decoder elements when stream selections change, so that in
 * the normal case it maintains 1 decoder of each type (video/audio/subtitle)
 * and only creates new elements when streams change and an existing decoder
 * is not capable of handling the new format.
 *
 * * supports multiple input pads for the parallel decoding of auxiliary streams
 * not muxed with the primary stream.
 *
 * * does not handle network stream buffering. decodebin3 expects that network stream
 * buffering is handled upstream, before data is passed to it.
 *
 */

/*
 * Global design
 *
 * 1) From sink pad to elementary streams (GstParseBin or identity)
 *
 * Note : If the incoming streams are push-based-only and are compatible with
 * either the output caps or a potential decoder, the usage of parsebin is
 * replaced by a simple passthrough identity element.
 *
 * The input sink pads are fed to GstParseBin. GstParseBin will feed them
 * through typefind. When the caps are detected (or changed) we recursively
 * figure out which demuxer, parser or depayloader is needed until we get to
 * elementary streams.
 *
 * All elementary streams (whether decoded or not, whether exposed or not) are
 * fed through multiqueue. There is only *one* multiqueue in decodebin3.
 *
 * => MultiQueue is the cornerstone.
 * => No buffering before multiqueue
 *
 * 2) Elementary streams
 *
 * After GstParseBin, there are 3 main components:
 *  1) Input Streams (provided by GstParseBin)
 *  2) Multiqueue slots
 *  3) Output Streams
 *
 * Input Streams correspond to the stream coming from GstParseBin and that gets
 * fed into a multiqueue slot.
 *
 * Output Streams correspond to the combination of a (optional) decoder and an
 * output ghostpad. Output Streams can be moved from one multiqueue slot to
 * another, can reconfigure itself (different decoders), and can be
 * added/removed depending on the configuration (all streams outputted, only one
 * of each type, ...).
 *
 * Multiqueue slots correspond to a pair of sink/src pad from multiqueue. For
 * each 'active' Input Stream there is a corresponding slot.
 * Slots might have different streams on input and output (due to internal
 * buffering).
 *
 * Due to internal queuing/buffering/..., all those components (might) behave
 * asynchronously. Therefore probes will be used on each component source pad to
 * detect various key-points:
 *  * EOS :
 *     the stream is done => Mark that component as done, optionally freeing/removing it
 *  * STREAM_START :
 *     a new stream is starting => link it further if needed
 *
 * 3) Gradual replacement
 *
 * If the caps change at any point in decodebin (input sink pad, demuxer output,
 * multiqueue output, ..), we gradually replace (if needed) the following elements.
 *
 * This is handled by the probes in various locations:
 *  a) typefind output
 *  b) multiqueue input (source pad of Input Streams)
 *  c) multiqueue output (source pad of Multiqueue Slots)
 *  d) final output (target of source ghostpads)
 *
 * When CAPS event arrive at those points, one of three things can happen:
 * a) There is no elements downstream yet, just create/link-to following elements
 * b) There are downstream elements, do a ACCEPT_CAPS query
 *  b.1) The new CAPS are accepted, keep current configuration
 *  b.2) The new CAPS are not accepted, remove following elements then do a)
 *
 *    Components:
 *
 *                                                   MultiQ     Output
 *                     Input(s)                      Slots      Streams
 *  /-------------------------------------------\   /-----\  /------------- \
 *
 * +-------------------------------------------------------------------------+
 * |                                                                         |
 * | +---------------------------------------------+                         |
 * | |   GstParseBin(s)                            |                         |
 * | |                +--------------+             |  +-----+                |
 * | |                |              |---[parser]-[|--| Mul |---[ decoder ]-[|
 * |]--[ typefind ]---|  demuxer(s)  |------------[|  | ti  |                |
 * | |                |  (if needed) |---[parser]-[|--| qu  |                |
 * | |                |              |---[parser]-[|--| eu  |---[ decoder ]-[|
 * | |                +--------------+             |  +------             ^  |
 * | +---------------------------------------------+        ^             |  |
 * |                                               ^        |             |  |
 * +-----------------------------------------------+--------+-------------+--+
 *                                                 |        |             |
 *                                                 |        |             |
 *                                       Probes  --/--------/-------------/
 *
 * ATOMIC SWITCHING
 *
 * We want to ensure we re-use decoders when switching streams. This takes place
 * at the multiqueue output level.
 *
 * MAIN CONCEPTS
 *  1) Activating a stream (i.e. linking a slot to an output) is only done within
 *    the streaming thread in the multiqueue_src_probe() and only if the
 *    stream is in the REQUESTED selection.
 *  2) Deactivating a stream (i.e. unlinking a slot from an output) is also done
 *    within the stream thread, but only in a purposefully called IDLE probe
 *    that calls reassign_slot().
 *
 * Based on those two principles, 3 "selection" of streams (stream-id) are used:
 * 1) requested_selection
 *    All streams within that list should be activated
 * 2) active_selection
 *    List of streams that are exposed by decodebin
 * 3) to_activate
 *    List of streams that will be moved to requested_selection in the
 *    reassign_slot() method (i.e. once a stream was deactivated, and the output
 *    was retargetted)
 */


GST_DEBUG_CATEGORY_STATIC (decodebin3_debug);
#define GST_CAT_DEFAULT decodebin3_debug

#define GST_TYPE_DECODEBIN3	 (gst_decodebin3_get_type ())

#define EXTRA_DEBUG 1

#define CUSTOM_FINAL_EOS_QUARK _custom_final_eos_quark_get ()
#define CUSTOM_FINAL_EOS_QUARK_DATA "custom-final-eos"
static GQuark
_custom_final_eos_quark_get (void)
{
  static gsize g_quark;

  if (g_once_init_enter (&g_quark)) {
    gsize quark =
        (gsize) g_quark_from_static_string ("decodebin3-custom-final-eos");
    g_once_init_leave (&g_quark, quark);
  }
  return g_quark;
}

#define CUSTOM_EOS_QUARK _custom_eos_quark_get ()
#define CUSTOM_EOS_QUARK_DATA "custom-eos"
static GQuark
_custom_eos_quark_get (void)
{
  static gsize g_quark;

  if (g_once_init_enter (&g_quark)) {
    gsize quark = (gsize) g_quark_from_static_string ("decodebin3-custom-eos");
    g_once_init_leave (&g_quark, quark);
  }
  return g_quark;
}

typedef struct _GstDecodebin3 GstDecodebin3;
typedef struct _GstDecodebin3Class GstDecodebin3Class;

typedef struct _DecodebinInputStream DecodebinInputStream;
typedef struct _DecodebinInput DecodebinInput;
typedef struct _DecodebinOutputStream DecodebinOutputStream;

typedef struct
{
  GstElement *element;
  GstMessage *error;            // Last error message seen for that element
  GstMessage *latency;          // Last latency message seen for that element
} CandidateDecoder;

struct _GstDecodebin3
{
  GstBin bin;

  /* input_lock protects the following variables */
  GMutex input_lock;
  /* Main input (static sink pad) */
  DecodebinInput *main_input;
  /* Supplementary input (request sink pads) */
  GList *other_inputs;
  /* counter for input */
  guint32 input_counter;
  /* Current stream group_id (default : GST_GROUP_ID_INVALID) */
  guint32 current_group_id;
  /* End of variables protected by input_lock */

  GstElement *multiqueue;
  GstClockTime default_mq_min_interleave;
  GstClockTime current_mq_min_interleave;

  /* selection_lock protects access to following variables */
  GMutex selection_lock;
  GList *input_streams;         /* List of DecodebinInputStream for active collection */
  GList *output_streams;        /* List of DecodebinOutputStream used for output */
  GList *slots;                 /* List of MultiQueueSlot */
  guint slot_id;

  /* Active collection */
  GstStreamCollection *collection;
  /* requested selection of stream-id to activate post-multiqueue */
  GList *requested_selection;
  /* list of stream-id currently activated in output */
  GList *active_selection;
  /* List of stream-id that need to be activated (after a stream switch for ex) */
  GList *to_activate;
  /* Pending select streams event */
  guint32 select_streams_seqnum;
  /* There is a pending GST_EVENT_SELECT_STREAMS being handled */
  gboolean pending_select_streams;
  /* TRUE if requested_selection was updated, will become FALSE once
   * it has fully transitioned to active */
  gboolean selection_updated;
  /* End of variables protected by selection_lock */

  /* Upstream handles stream selection */
  gboolean upstream_handles_selection;

  /* Factories */
  GMutex factories_lock;
  guint32 factories_cookie;
  /* All DECODABLE factories */
  GList *factories;
  /* Only DECODER factories */
  GList *decoder_factories;
  /* DECODABLE but not DECODER factories */
  GList *decodable_factories;

  /* counters for pads */
  guint32 apadcount, vpadcount, tpadcount, opadcount;

  /* Properties */
  GstCaps *caps;

  GList *candidate_decoders;
};

struct _GstDecodebin3Class
{
  GstBinClass class;

    gint (*select_stream) (GstDecodebin3 * dbin,
      GstStreamCollection * collection, GstStream * stream);
};

/* Input of decodebin, controls input pad and parsebin */
struct _DecodebinInput
{
  GstDecodebin3 *dbin;

  gboolean is_main;

  GstPad *ghost_sink;
  GstPad *parsebin_sink;

  GstStreamCollection *collection;      /* Active collection */
  gboolean upstream_selected;

  guint group_id;

  /* Either parsebin or identity is used */
  GstElement *parsebin;
  GstElement *identity;

  gulong pad_added_sigid;
  gulong pad_removed_sigid;
  gulong drained_sigid;

  /* TRUE if the input got drained */
  gboolean drained;

  /* TEMPORARY HACK for knowing if upstream is already parsed and identity can
   * be avoided */
  gboolean input_is_parsed;
};

/* Streams that come from parsebin or identity */
/* FIXME : All this is hardcoded. Switch to tree of chains */
struct _DecodebinInputStream
{
  GstDecodebin3 *dbin;

  GstStream *active_stream;

  DecodebinInput *input;

  GstPad *srcpad;               /* From parsebin or identity */

  /* id of the pad event probe */
  gulong output_event_probe_id;

  /* id of the buffer blocking probe on the parsebin srcpad pad */
  gulong buffer_probe_id;

  /* Whether we saw an EOS on input. This should be treated accordingly
   * when the stream is no longer used */
  gboolean saw_eos;
};

/* Multiqueue Slots */
typedef struct _MultiQueueSlot
{
  guint id;

  GstDecodebin3 *dbin;
  /* Type of stream handled by this slot */
  GstStreamType type;

  /* Linked input and output */
  DecodebinInputStream *input;

  /* pending => last stream received on sink pad */
  GstStream *pending_stream;
  /* active => last stream outputted on source pad */
  GstStream *active_stream;

  GstPad *sink_pad, *src_pad;

  /* id of the MQ src_pad event probe */
  gulong probe_id;

  /* TRUE if EOS was pushed out by multiqueue */
  gboolean is_drained;

  DecodebinOutputStream *output;
} MultiQueueSlot;

/* Streams that are exposed downstream (i.e. output) */
struct _DecodebinOutputStream
{
  GstDecodebin3 *dbin;
  /* The type of stream handled by this output stream */
  GstStreamType type;

  /* The slot to which this output stream is currently connected to */
  MultiQueueSlot *slot;

  GstElement *decoder;          /* Optional */
  GstPad *decoder_sink, *decoder_src;
  gboolean linked;

  /* ghostpad */
  GstPad *src_pad;
  /* Flag if ghost pad is exposed */
  gboolean src_exposed;

  /* Reported decoder latency */
  GstClockTime decoder_latency;

  /* keyframe dropping probe */
  gulong drop_probe_id;
};

/* properties */
enum
{
  PROP_0,
  PROP_CAPS
};

/* signals */
enum
{
  SIGNAL_SELECT_STREAM,
  SIGNAL_ABOUT_TO_FINISH,
  LAST_SIGNAL
};
static guint gst_decodebin3_signals[LAST_SIGNAL] = { 0 };

#define SELECTION_LOCK(dbin) G_STMT_START {				\
    GST_LOG_OBJECT (dbin,						\
		    "selection locking from thread %p",			\
		    g_thread_self ());					\
    g_mutex_lock (&dbin->selection_lock);				\
    GST_LOG_OBJECT (dbin,						\
		    "selection locked from thread %p",			\
		    g_thread_self ());					\
  } G_STMT_END

#define SELECTION_UNLOCK(dbin) G_STMT_START {				\
    GST_LOG_OBJECT (dbin,						\
		    "selection unlocking from thread %p",		\
		    g_thread_self ());					\
    g_mutex_unlock (&dbin->selection_lock);				\
  } G_STMT_END

#define INPUT_LOCK(dbin) G_STMT_START {				\
    GST_LOG_OBJECT (dbin,						\
		    "input locking from thread %p",			\
		    g_thread_self ());					\
    g_mutex_lock (&dbin->input_lock);				\
    GST_LOG_OBJECT (dbin,						\
		    "input locked from thread %p",			\
		    g_thread_self ());					\
  } G_STMT_END

#define INPUT_UNLOCK(dbin) G_STMT_START {				\
    GST_LOG_OBJECT (dbin,						\
		    "input unlocking from thread %p",		\
		    g_thread_self ());					\
    g_mutex_unlock (&dbin->input_lock);				\
  } G_STMT_END

GType gst_decodebin3_get_type (void);
#define gst_decodebin3_parent_class parent_class
G_DEFINE_TYPE (GstDecodebin3, gst_decodebin3, GST_TYPE_BIN);
#define _do_init \
    GST_DEBUG_CATEGORY_INIT (decodebin3_debug, "decodebin3", 0, "decoder bin");\
    playback_element_init (plugin);
GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (decodebin3, "decodebin3", GST_RANK_NONE,
    GST_TYPE_DECODEBIN3, _do_init);

static GstStaticCaps default_raw_caps = GST_STATIC_CAPS (DEFAULT_RAW_CAPS);

static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate request_sink_template =
GST_STATIC_PAD_TEMPLATE ("sink_%u",
    GST_PAD_SINK,
    GST_PAD_REQUEST,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate video_src_template =
GST_STATIC_PAD_TEMPLATE ("video_%u",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate audio_src_template =
GST_STATIC_PAD_TEMPLATE ("audio_%u",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate text_src_template =
GST_STATIC_PAD_TEMPLATE ("text_%u",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src_%u",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS_ANY);


static void gst_decodebin3_dispose (GObject * object);
static void gst_decodebin3_finalize (GObject * object);
static void gst_decodebin3_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec);
static void gst_decodebin3_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec);

static gboolean parsebin_autoplug_continue_cb (GstElement *
    parsebin, GstPad * pad, GstCaps * caps, GstDecodebin3 * dbin);

static gint
gst_decodebin3_select_stream (GstDecodebin3 * dbin,
    GstStreamCollection * collection, GstStream * stream)
{
  GST_LOG_OBJECT (dbin, "default select-stream, returning -1");

  return -1;
}

static GstPad *gst_decodebin3_request_new_pad (GstElement * element,
    GstPadTemplate * temp, const gchar * name, const GstCaps * caps);
static void gst_decodebin3_release_pad (GstElement * element, GstPad * pad);
static GstMessage *handle_stream_collection_locked (GstDecodebin3 * dbin,
    GstStreamCollection * collection, DecodebinInput * input);
static void gst_decodebin3_handle_message (GstBin * bin, GstMessage * message);
static GstStateChangeReturn gst_decodebin3_change_state (GstElement * element,
    GstStateChange transition);
static gboolean gst_decodebin3_send_event (GstElement * element,
    GstEvent * event);

static void gst_decode_bin_update_factories_list (GstDecodebin3 * dbin);

static void gst_decodebin_input_reset (DecodebinInput * input);
static void gst_decodebin_input_free (DecodebinInput * input);
static DecodebinInput *gst_decodebin_input_new (GstDecodebin3 * dbin,
    gboolean main);
static gboolean gst_decodebin_input_set_group_id (DecodebinInput * input,
    guint32 * group_id);
static void gst_decodebin_input_unblock_streams (DecodebinInput * input,
    gboolean unblock_other_inputs);

static gboolean reconfigure_output_stream (DecodebinOutputStream * output,
    MultiQueueSlot * slot, GstMessage ** msg);
static void free_output_stream (GstDecodebin3 * dbin,
    DecodebinOutputStream * output);
static DecodebinOutputStream *create_output_stream (GstDecodebin3 * dbin,
    GstStreamType type);

static GstPadProbeReturn slot_unassign_probe (GstPad * pad,
    GstPadProbeInfo * info, MultiQueueSlot * slot);
static gboolean reassign_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot);
static MultiQueueSlot
    * gst_decodebin_get_slot_for_input_stream_locked (GstDecodebin3 * dbin,
    DecodebinInputStream * input);
static void link_input_to_slot (DecodebinInputStream * input,
    MultiQueueSlot * slot);
static void free_multiqueue_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot);
static void free_multiqueue_slot_async (GstDecodebin3 * dbin,
    MultiQueueSlot * slot);

static GstStreamCollection *get_merged_collection (GstDecodebin3 * dbin);
static void update_requested_selection (GstDecodebin3 * dbin);

static void parsebin_pad_added_cb (GstElement * demux, GstPad * pad,
    DecodebinInput * input);
static void parsebin_pad_removed_cb (GstElement * demux, GstPad * pad,
    DecodebinInput * input);

static gboolean
_gst_int_accumulator (GSignalInvocationHint * ihint,
    GValue * return_accu, const GValue * handler_return, gpointer dummy)
{
  gint res = g_value_get_int (handler_return);

  g_value_set_int (return_accu, res);

  if (res == -1)
    return TRUE;

  return FALSE;
}

static void
gst_decodebin3_class_init (GstDecodebin3Class * klass)
{
  GObjectClass *gobject_klass = (GObjectClass *) klass;
  GstElementClass *element_class = (GstElementClass *) klass;
  GstBinClass *bin_klass = (GstBinClass *) klass;

  gobject_klass->dispose = gst_decodebin3_dispose;
  gobject_klass->finalize = gst_decodebin3_finalize;
  gobject_klass->set_property = gst_decodebin3_set_property;
  gobject_klass->get_property = gst_decodebin3_get_property;

  g_object_class_install_property (gobject_klass, PROP_CAPS,
      g_param_spec_boxed ("caps", "Caps",
          "The caps on which to stop decoding. (NULL = default)",
          GST_TYPE_CAPS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  /**
   * GstDecodebin3::select-stream
   * @decodebin: a #GstDecodebin3
   * @collection: a #GstStreamCollection
   * @stream: a #GstStream
   *
   * This signal is emitted whenever @decodebin needs to decide whether
   * to expose a @stream of a given @collection.
   *
   * Note that the prefered way to select streams is to listen to
   * GST_MESSAGE_STREAM_COLLECTION on the bus and send a
   * GST_EVENT_SELECT_STREAMS with the streams the user wants.
   *
   * Returns: 1 if the stream should be selected, 0 if it shouldn't be selected.
   * A value of -1 (default) lets @decodebin decide what to do with the stream.
   * */
  gst_decodebin3_signals[SIGNAL_SELECT_STREAM] =
      g_signal_new ("select-stream", G_TYPE_FROM_CLASS (klass),
      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstDecodebin3Class, select_stream),
      _gst_int_accumulator, NULL, NULL,
      G_TYPE_INT, 2, GST_TYPE_STREAM_COLLECTION, GST_TYPE_STREAM);

  /**
   * GstDecodebin3::about-to-finish:
   *
   * This signal is emitted when the data for the selected URI is
   * entirely buffered and it is safe to specify another URI.
   */
  gst_decodebin3_signals[SIGNAL_ABOUT_TO_FINISH] =
      g_signal_new ("about-to-finish", G_TYPE_FROM_CLASS (klass),
      G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_NONE, 0, G_TYPE_NONE);


  element_class->request_new_pad =
      GST_DEBUG_FUNCPTR (gst_decodebin3_request_new_pad);
  element_class->change_state = GST_DEBUG_FUNCPTR (gst_decodebin3_change_state);
  element_class->send_event = GST_DEBUG_FUNCPTR (gst_decodebin3_send_event);
  element_class->release_pad = GST_DEBUG_FUNCPTR (gst_decodebin3_release_pad);

  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&sink_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&request_sink_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&video_src_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&audio_src_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&text_src_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&src_template));

  gst_element_class_set_static_metadata (element_class,
      "Decoder Bin 3", "Generic/Bin/Decoder",
      "Autoplug and decode to raw media",
      "Edward Hervey <edward@centricular.com>");

  bin_klass->handle_message = gst_decodebin3_handle_message;

  klass->select_stream = gst_decodebin3_select_stream;
}

static void
gst_decodebin3_init (GstDecodebin3 * dbin)
{
  /* Create main input */
  dbin->main_input = gst_decodebin_input_new (dbin, TRUE);

  dbin->multiqueue = gst_element_factory_make ("multiqueue", NULL);
  g_object_get (dbin->multiqueue, "min-interleave-time",
      &dbin->default_mq_min_interleave, NULL);
  dbin->current_mq_min_interleave = dbin->default_mq_min_interleave;
  g_object_set (dbin->multiqueue, "sync-by-running-time", TRUE,
      "max-size-buffers", 0, "use-interleave", TRUE, NULL);
  gst_bin_add ((GstBin *) dbin, dbin->multiqueue);

  dbin->current_group_id = GST_GROUP_ID_INVALID;

  g_mutex_init (&dbin->factories_lock);
  g_mutex_init (&dbin->selection_lock);
  g_mutex_init (&dbin->input_lock);

  dbin->caps = gst_static_caps_get (&default_raw_caps);

  GST_OBJECT_FLAG_SET (dbin, GST_BIN_FLAG_STREAMS_AWARE);
}

static void
gst_decodebin3_reset (GstDecodebin3 * dbin)
{
  GList *tmp;

  GST_DEBUG_OBJECT (dbin, "Resetting");

  /* Free output streams */
  for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
    DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
    free_output_stream (dbin, output);
  }
  g_list_free (dbin->output_streams);
  dbin->output_streams = NULL;

  /* Free multiqueue slots */
  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    free_multiqueue_slot (dbin, slot);
  }
  g_list_free (dbin->slots);
  dbin->slots = NULL;
  dbin->current_group_id = GST_GROUP_ID_INVALID;

  /* Reset the inputs */
  gst_decodebin_input_reset (dbin->main_input);
  for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
    gst_decodebin_input_reset (tmp->data);
  }

  /* Reset multiqueue to default interleave */
  g_object_set (dbin->multiqueue, "min-interleave-time",
      dbin->default_mq_min_interleave, NULL);
  dbin->current_mq_min_interleave = dbin->default_mq_min_interleave;
  dbin->upstream_handles_selection = FALSE;

  if (dbin->collection) {
    gst_clear_object (&dbin->collection);
  }

  g_list_free_full (dbin->requested_selection, g_free);
  dbin->requested_selection = NULL;

  g_list_free_full (dbin->active_selection, g_free);
  dbin->active_selection = NULL;

  g_list_free (dbin->to_activate);
  dbin->to_activate = NULL;

  dbin->select_streams_seqnum = GST_SEQNUM_INVALID;

  dbin->pending_select_streams = FALSE;

  dbin->selection_updated = FALSE;
}

static void
gst_decodebin3_dispose (GObject * object)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  gst_decodebin3_reset (dbin);

  g_mutex_lock (&dbin->factories_lock);
  if (dbin->factories) {
    gst_plugin_feature_list_free (dbin->factories);
    dbin->factories = NULL;
  }
  if (dbin->decoder_factories) {
    g_list_free (dbin->decoder_factories);
    dbin->decoder_factories = NULL;
  }
  if (dbin->decodable_factories) {
    g_list_free (dbin->decodable_factories);
    dbin->decodable_factories = NULL;
  }
  g_mutex_unlock (&dbin->factories_lock);

  SELECTION_LOCK (dbin);
  if (dbin->collection) {
    gst_clear_object (&dbin->collection);
  }
  SELECTION_UNLOCK (dbin);

  INPUT_LOCK (dbin);
  if (dbin->main_input) {
    gst_decodebin_input_free (dbin->main_input);
    dbin->main_input = NULL;
  }

  g_list_free_full (dbin->other_inputs,
      (GDestroyNotify) gst_decodebin_input_free);
  dbin->other_inputs = NULL;
  INPUT_UNLOCK (dbin);

  gst_clear_caps (&dbin->caps);

  G_OBJECT_CLASS (parent_class)->dispose (object);
}

static void
gst_decodebin3_finalize (GObject * object)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  g_mutex_clear (&dbin->factories_lock);
  g_mutex_clear (&dbin->selection_lock);
  g_mutex_clear (&dbin->input_lock);

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

static void
gst_decodebin3_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  switch (prop_id) {
    case PROP_CAPS:
      GST_OBJECT_LOCK (dbin);
      if (dbin->caps)
        gst_caps_unref (dbin->caps);
      dbin->caps = g_value_dup_boxed (value);
      GST_OBJECT_UNLOCK (dbin);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static void
gst_decodebin3_get_property (GObject * object, guint prop_id, GValue * value,
    GParamSpec * pspec)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  switch (prop_id) {
    case PROP_CAPS:
      GST_OBJECT_LOCK (dbin);
      g_value_set_boxed (value, dbin->caps);
      GST_OBJECT_UNLOCK (dbin);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

/* WITH SELECTION_LOCK TAKEN! */
static gboolean
all_input_streams_are_eos (GstDecodebin3 * dbin)
{
  GList *tmp;

  /* First check input streams */
  for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
    DecodebinInputStream *input = (DecodebinInputStream *) tmp->data;
    if (input->saw_eos == FALSE)
      return FALSE;
  }

  GST_DEBUG_OBJECT (dbin, "All input streams are EOS");
  return TRUE;
}

/** check_all_input_streams_for_eos:
 * @dbin: A #GstDecodebin3
 * @eos_event: (transfer none): The GST_EVENT_EOS that triggered this check
 *
 * Check if all inputs streams are EOS. If they are propagates the @eos_event to all
 * #DecodebinInputstream pads.
 *
 * Returns: #TRUE if all pads are EOS and the event was propagated, else #FALSE.
 */
static gboolean
check_all_input_streams_for_eos (GstDecodebin3 * dbin, GstEvent * eos_event)
{
  GList *tmp;
  GList *outputpads = NULL;

  SELECTION_LOCK (dbin);

  if (!all_input_streams_are_eos (dbin)) {
    SELECTION_UNLOCK (dbin);
    return FALSE;
  }

  /* We know all streams are EOS, properly clean up everything */

  /* We grab all peer pads *while* the selection lock is taken and then we will
     push EOS downstream with the selection lock released */
  for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
    DecodebinInputStream *input = (DecodebinInputStream *) tmp->data;
    GstPad *peer = gst_pad_get_peer (input->srcpad);

    /* Keep a reference to the peer pad */
    if (peer)
      outputpads = g_list_append (outputpads, peer);
  }

  SELECTION_UNLOCK (dbin);

  for (tmp = outputpads; tmp; tmp = tmp->next) {
    GstPad *peer = (GstPad *) tmp->data;

    /* Send EOS and then remove elements */
    gst_pad_send_event (peer, gst_event_ref (eos_event));
    GST_FIXME_OBJECT (peer, "Remove input stream");
    gst_object_unref (peer);
  }

  g_list_free (outputpads);

  return TRUE;
}

/* Get the intersection of parser caps and available (sorted) decoders */
static GstCaps *
get_parser_caps_filter (GstDecodebin3 * dbin, GstCaps * caps)
{
  GList *tmp;
  GstCaps *filter_caps;

  /* If no filter was provided, it can handle anything */
  if (!caps || gst_caps_is_any (caps))
    return gst_caps_new_any ();

  filter_caps = gst_caps_new_empty ();

  g_mutex_lock (&dbin->factories_lock);
  gst_decode_bin_update_factories_list (dbin);
  for (tmp = dbin->decoder_factories; tmp; tmp = tmp->next) {
    GstElementFactory *factory = (GstElementFactory *) tmp->data;
    GstCaps *tcaps, *intersection;
    const GList *tmps;

    GST_LOG ("Trying factory %s",
        gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
    for (tmps = gst_element_factory_get_static_pad_templates (factory); tmps;
        tmps = tmps->next) {
      GstStaticPadTemplate *st = (GstStaticPadTemplate *) tmps->data;
      if (st->direction != GST_PAD_SINK || st->presence != GST_PAD_ALWAYS)
        continue;
      tcaps = gst_static_pad_template_get_caps (st);
      intersection =
          gst_caps_intersect_full (tcaps, caps, GST_CAPS_INTERSECT_FIRST);
      filter_caps = gst_caps_merge (filter_caps, intersection);
      gst_caps_unref (tcaps);
    }
  }
  g_mutex_unlock (&dbin->factories_lock);
  GST_DEBUG_OBJECT (dbin, "Got filter caps %" GST_PTR_FORMAT, filter_caps);
  return filter_caps;
}

static gboolean
check_parser_caps_filter (GstDecodebin3 * dbin, GstCaps * caps)
{
  GList *tmp;
  gboolean res = FALSE;
  GstCaps *default_raw = gst_static_caps_get (&default_raw_caps);

  if (gst_caps_can_intersect (caps, default_raw)) {
    GST_INFO_OBJECT (dbin, "Dealing with raw stream from the demuxer, "
        " we can handle them even if we won't expose then");
    gst_caps_unref (default_raw);

    return TRUE;
  }
  gst_caps_unref (default_raw);

  g_mutex_lock (&dbin->factories_lock);
  gst_decode_bin_update_factories_list (dbin);
  for (tmp = dbin->decoder_factories; tmp; tmp = tmp->next) {
    GstElementFactory *factory = (GstElementFactory *) tmp->data;
    GstCaps *tcaps;
    const GList *tmps;

    GST_LOG ("Trying factory %s",
        gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
    for (tmps = gst_element_factory_get_static_pad_templates (factory); tmps;
        tmps = tmps->next) {
      GstStaticPadTemplate *st = (GstStaticPadTemplate *) tmps->data;
      if (st->direction != GST_PAD_SINK || st->presence != GST_PAD_ALWAYS)
        continue;
      tcaps = gst_static_pad_template_get_caps (st);
      if (gst_caps_can_intersect (tcaps, caps)) {
        res = TRUE;
        gst_caps_unref (tcaps);
        goto beach;
      }
      gst_caps_unref (tcaps);
    }
  }
beach:
  g_mutex_unlock (&dbin->factories_lock);
  GST_DEBUG_OBJECT (dbin, "Can intersect %" GST_PTR_FORMAT ": %d", caps, res);
  return res;
}

/* Probe on the output of a decodebin input stream (from parsebin or identity) */
static GstPadProbeReturn
gst_decodebin_input_stream_src_probe (GstPad * pad, GstPadProbeInfo * info,
    DecodebinInputStream * input)
{
  GstPadProbeReturn ret = GST_PAD_PROBE_OK;

  if (GST_IS_EVENT (GST_PAD_PROBE_INFO_DATA (info))) {
    GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);

    GST_DEBUG_OBJECT (pad, "Got event %s", GST_EVENT_TYPE_NAME (ev));
    switch (GST_EVENT_TYPE (ev)) {
      case GST_EVENT_STREAM_START:
      {
        GstStream *stream = NULL;
        guint group_id = GST_GROUP_ID_INVALID;

        if (!gst_event_parse_group_id (ev, &group_id)) {
          GST_FIXME_OBJECT (pad,
              "Consider implementing group-id handling on stream-start event");
          group_id = gst_util_group_id_next ();
        }

        GST_DEBUG_OBJECT (pad, "Got stream-start, group_id:%d, input %p",
            group_id, input->input);
        if (gst_decodebin_input_set_group_id (input->input, &group_id)) {
          ev = gst_event_make_writable (ev);
          gst_event_set_group_id (ev, group_id);
          GST_PAD_PROBE_INFO_DATA (info) = ev;
        }
        input->saw_eos = FALSE;

        gst_event_parse_stream (ev, &stream);
        /* FIXME : Would we ever end up with a stream already set on the input ?? */
        if (stream) {
          if (input->active_stream != stream) {
            MultiQueueSlot *slot;
            if (input->active_stream)
              gst_object_unref (input->active_stream);
            input->active_stream = stream;
            /* We have the beginning of a stream, get a multiqueue slot and link to it */
            SELECTION_LOCK (input->dbin);
            slot =
                gst_decodebin_get_slot_for_input_stream_locked (input->dbin,
                input);
            link_input_to_slot (input, slot);
            SELECTION_UNLOCK (input->dbin);
          } else
            gst_object_unref (stream);
        }
      }
        break;
      case GST_EVENT_GAP:
      {
        /* If we are still waiting to be unblocked and we get a gap, unblock */
        if (input->buffer_probe_id) {
          GST_DEBUG_OBJECT (pad, "Got a gap event! Unblocking input(s) !");
          gst_decodebin_input_unblock_streams (input->input, TRUE);
        }
        break;
      }
      case GST_EVENT_CAPS:
      {
        GstCaps *caps = NULL;
        gst_event_parse_caps (ev, &caps);
        GST_DEBUG_OBJECT (pad, "caps %" GST_PTR_FORMAT, caps);
        if (caps && input->active_stream)
          gst_stream_set_caps (input->active_stream, caps);
      }
        break;
      case GST_EVENT_EOS:
      {
        GST_DEBUG_OBJECT (pad, "Marking input as EOS");
        input->saw_eos = TRUE;

        /* If not all pads are EOS yet, we send our custom EOS (which will be
         * handled/dropped downstream of multiqueue) */
        if (!check_all_input_streams_for_eos (input->dbin, ev)) {
          GstPad *peer = gst_pad_get_peer (input->srcpad);
          if (peer) {
            /* Send custom-eos event to multiqueue slot */
            GstEvent *event;

            GST_DEBUG_OBJECT (pad,
                "Got EOS end of input stream, post custom-eos");
            event = gst_event_new_eos ();
            gst_event_set_seqnum (event, gst_event_get_seqnum (ev));
            gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (event),
                CUSTOM_EOS_QUARK, (gchar *) CUSTOM_EOS_QUARK_DATA, NULL);
            gst_pad_send_event (peer, event);
            gst_object_unref (peer);
          } else {
            GST_FIXME_OBJECT (pad, "No peer, what should we do ?");
          }
        }
        ret = GST_PAD_PROBE_DROP;
      }
        break;
      case GST_EVENT_FLUSH_STOP:
        GST_DEBUG_OBJECT (pad, "Clear saw_eos flag");
        input->saw_eos = FALSE;
      default:
        break;
    }
  } else if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
    if (input->input && input->input->identity) {
      GST_DEBUG_OBJECT (pad, "Letting query through");
    } else {
      GstQuery *q = GST_PAD_PROBE_INFO_QUERY (info);
      GST_DEBUG_OBJECT (pad, "Seeing query %" GST_PTR_FORMAT, q);
      /* If we have a parser, we want to reply to the caps query */
      /* FIXME: Set a flag when the input stream is created for
       * streams where we shouldn't reply to these queries */
      if (GST_QUERY_TYPE (q) == GST_QUERY_CAPS
          && (info->type & GST_PAD_PROBE_TYPE_PULL)) {
        GstCaps *filter = NULL;
        GstCaps *allowed;
        gst_query_parse_caps (q, &filter);
        allowed = get_parser_caps_filter (input->dbin, filter);
        GST_DEBUG_OBJECT (pad,
            "Intercepting caps query, setting %" GST_PTR_FORMAT, allowed);
        gst_query_set_caps_result (q, allowed);
        gst_caps_unref (allowed);
        ret = GST_PAD_PROBE_HANDLED;
      } else if (GST_QUERY_TYPE (q) == GST_QUERY_ACCEPT_CAPS) {
        GstCaps *prop = NULL;
        gst_query_parse_accept_caps (q, &prop);
        /* Fast check against target caps */
        if (gst_caps_can_intersect (prop, input->dbin->caps)) {
          gst_query_set_accept_caps_result (q, TRUE);
        } else {
          gboolean accepted = check_parser_caps_filter (input->dbin, prop);
          /* check against caps filter */
          gst_query_set_accept_caps_result (q, accepted);
          GST_DEBUG_OBJECT (pad, "ACCEPT_CAPS query, returning %d", accepted);
        }
        ret = GST_PAD_PROBE_HANDLED;
      }
    }
  }

  return ret;
}

static GstPadProbeReturn
gst_decodebin_input_stream_buffer_probe (GstPad * pad, GstPadProbeInfo * info,
    DecodebinInput * input)
{
  /* We have at least one buffer pending, unblock parsebin/identity pads */
  GST_DEBUG_OBJECT (pad, "Got a buffer ! unblocking");
  gst_decodebin_input_unblock_streams (input, TRUE);

  return GST_PAD_PROBE_OK;
}

/** gst_decodebin_input_add_stream:
 * @input: A #DecodebinInput
 * @pad: (transfer none): The #GstPad that generates this stream
 * @stream: (allow none) (transfer full): The #GstStream if known
 *
 * Creates a new #DecodebinInputstream for the given @pad and @stream, and adds
 * it to the list of #GstDecodebin3 input_streams.
 *
 * Returns: The new #DecodebinInputstream.
 */
/* Call with selection lock */
static DecodebinInputStream *
gst_decodebin_input_add_stream (DecodebinInput * input, GstPad * pad,
    GstStream * stream)
{
  GstDecodebin3 *dbin = input->dbin;
  DecodebinInputStream *res = g_new0 (DecodebinInputStream, 1);

  GST_DEBUG_OBJECT (dbin, "Creating input stream for %" GST_PTR_FORMAT, pad);

  res->dbin = dbin;
  res->input = input;
  res->srcpad = gst_object_ref (pad);
  res->active_stream = stream;

  /* Put probe on output source pad (for detecting EOS/STREAM_START/FLUSH) */
  res->output_event_probe_id =
      gst_pad_add_probe (pad,
      GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM
      | GST_PAD_PROBE_TYPE_EVENT_FLUSH,
      (GstPadProbeCallback) gst_decodebin_input_stream_src_probe, res, NULL);

  /* Install a blocking buffer probe */
  res->buffer_probe_id =
      gst_pad_add_probe (pad,
      GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_BUFFER,
      (GstPadProbeCallback) gst_decodebin_input_stream_buffer_probe, input,
      NULL);

  SELECTION_LOCK (dbin);
  /* Add to list of current input streams */
  dbin->input_streams = g_list_append (dbin->input_streams, res);
  SELECTION_UNLOCK (dbin);
  GST_DEBUG_OBJECT (pad, "Done creating input stream");

  return res;
}

/* WITH SELECTION_LOCK TAKEN! */
static void
remove_input_stream (GstDecodebin3 * dbin, DecodebinInputStream * stream)
{
  MultiQueueSlot *slot;

  GST_DEBUG_OBJECT (dbin, "Removing input stream %p (%s)", stream,
      stream->active_stream ? gst_stream_get_stream_id (stream->active_stream) :
      "<NONE>");

  gst_object_replace ((GstObject **) & stream->active_stream, NULL);

  /* Unlink from slot */
  if (stream->srcpad) {
    GstPad *peer;
    peer = gst_pad_get_peer (stream->srcpad);
    if (peer) {
      gst_pad_unlink (stream->srcpad, peer);
      gst_object_unref (peer);
    }
    if (stream->buffer_probe_id)
      gst_pad_remove_probe (stream->srcpad, stream->buffer_probe_id);
    gst_object_unref (stream->srcpad);
  }

  slot = gst_decodebin_get_slot_for_input_stream_locked (dbin, stream);
  if (slot) {
    slot->pending_stream = NULL;
    slot->input = NULL;
    GST_DEBUG_OBJECT (dbin, "slot %p cleared", slot);
  }

  dbin->input_streams = g_list_remove (dbin->input_streams, stream);

  g_free (stream);
}

/** gst_decodebin_input_unblock_streams:
 * @input: A #DecodebinInput
 * @unblock_other_inputs: Whether to also unblock other #DecodebinInput targetting the same #GstStreamCollection
 *
 * Unblock all #DecodebinInputstream for the given @input. If
 * @unblock_other_inputs is TRUE, it will also unblock other #DecodebinInput
 * targetting the same #GstStreamCollection.
 */
static void
gst_decodebin_input_unblock_streams (DecodebinInput * input,
    gboolean unblock_other_inputs)
{
  GstDecodebin3 *dbin = input->dbin;
  GList *tmp, *unused_slot = NULL;

  GST_DEBUG_OBJECT (dbin,
      "DecodebinInput for %" GST_PTR_FORMAT " , unblock_other_inputs:%d",
      input->parsebin, unblock_other_inputs);

  /* Re-use existing streams if/when possible */
  GST_FIXME_OBJECT (dbin, "Re-use existing input streams if/when possible");

  /* Unblock all input streams and link to a slot if needed */
  SELECTION_LOCK (dbin);
  tmp = dbin->input_streams;
  while (tmp != NULL) {
    DecodebinInputStream *input_stream = (DecodebinInputStream *) tmp->data;
    GList *next = tmp->next;
    MultiQueueSlot *slot;

    if (input_stream->input != input) {
      tmp = next;
      continue;
    }

    GST_DEBUG_OBJECT (dbin, "Checking input stream %p", input_stream);

    if (!input_stream->active_stream)
      input_stream->active_stream = gst_pad_get_stream (input_stream->srcpad);

    /* Ensure the stream has an associated slot */
    slot = gst_decodebin_get_slot_for_input_stream_locked (dbin, input_stream);
    if (slot->input != input_stream)
      link_input_to_slot (input_stream, slot);

    if (input_stream->buffer_probe_id) {
      GST_DEBUG_OBJECT (dbin,
          "Removing pad block on input %p pad %" GST_PTR_FORMAT, input_stream,
          input_stream->srcpad);
      gst_pad_remove_probe (input_stream->srcpad,
          input_stream->buffer_probe_id);
      input_stream->buffer_probe_id = 0;
    }

    if (input_stream->saw_eos) {
      GST_DEBUG_OBJECT (dbin, "Removing EOS'd stream");
      remove_input_stream (dbin, input_stream);
      tmp = dbin->input_streams;
    } else
      tmp = next;
  }

  /* Weed out unused multiqueue slots */
  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    GST_LOG_OBJECT (dbin, "Slot %d input:%p", slot->id, slot->input);
    if (slot->input == NULL) {
      unused_slot =
          g_list_append (unused_slot, gst_object_ref (slot->sink_pad));
    }
  }
  SELECTION_UNLOCK (dbin);

  if (unused_slot) {
    for (tmp = unused_slot; tmp; tmp = tmp->next) {
      GstPad *sink_pad = (GstPad *) tmp->data;
      GST_DEBUG_OBJECT (sink_pad, "Sending EOS to unused slot");
      gst_pad_send_event (sink_pad, gst_event_new_eos ());
    }
    g_list_free_full (unused_slot, (GDestroyNotify) gst_object_unref);
  }

  if (unblock_other_inputs) {
    GList *tmp;
    /* If requrested, unblock inputs which are targetting the same collection */
    if (dbin->main_input != input) {
      if (dbin->main_input->collection == input->collection) {
        GST_DEBUG_OBJECT (dbin, "Unblock main input");
        gst_decodebin_input_unblock_streams (dbin->main_input, FALSE);
      }
    }
    for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
      DecodebinInput *other = tmp->data;
      if (other->collection == input->collection) {
        GST_DEBUG_OBJECT (dbin, "Unblock other input");
        gst_decodebin_input_unblock_streams (other, FALSE);
      }
    }
  }
}

static void
parsebin_pad_added_cb (GstElement * demux, GstPad * pad, DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;

  GST_DEBUG_OBJECT (dbin, "New pad %s:%s (input:%p)", GST_DEBUG_PAD_NAME (pad),
      input);

  gst_decodebin_input_add_stream (input, pad, NULL);
}

/* WITH SELECTION_LOCK TAKEN! */
static DecodebinInputStream *
find_input_stream_for_pad (GstDecodebin3 * dbin, GstPad * pad)
{
  GList *tmp;

  for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
    DecodebinInputStream *cand = (DecodebinInputStream *) tmp->data;
    if (cand->srcpad == pad)
      return cand;
  }
  return NULL;
}

static void
parsebin_pad_removed_cb (GstElement * demux, GstPad * pad, DecodebinInput * inp)
{
  GstDecodebin3 *dbin = inp->dbin;
  DecodebinInputStream *input = NULL;
  MultiQueueSlot *slot;

  if (!GST_PAD_IS_SRC (pad))
    return;

  SELECTION_LOCK (dbin);

  GST_DEBUG_OBJECT (pad, "removed");
  input = find_input_stream_for_pad (dbin, pad);

  if (input == NULL) {
    GST_DEBUG_OBJECT (pad,
        "Input stream not found, it was cleaned-up earlier after receiving EOS");
    SELECTION_UNLOCK (dbin);
    return;
  }

  /* If there are no pending pads, this means we will definitely not need this
   * stream anymore */

  GST_DEBUG_OBJECT (pad, "Remove input stream %p", input);

  slot = gst_decodebin_get_slot_for_input_stream_locked (dbin, input);
  remove_input_stream (dbin, input);

  if (slot && g_list_find (dbin->slots, slot) && slot->is_drained) {
    /* if slot is still there and already drained, remove it in here */
    if (slot->output) {
      DecodebinOutputStream *output = slot->output;
      GST_DEBUG_OBJECT (pad, "Multiqueue was drained, Remove output stream");

      dbin->output_streams = g_list_remove (dbin->output_streams, output);
      free_output_stream (dbin, output);
    }
    GST_DEBUG_OBJECT (pad, "No pending pad, Remove multiqueue slot");
    if (slot->probe_id)
      gst_pad_remove_probe (slot->src_pad, slot->probe_id);
    slot->probe_id = 0;
    dbin->slots = g_list_remove (dbin->slots, slot);
    free_multiqueue_slot_async (dbin, slot);
  }
  SELECTION_UNLOCK (dbin);
}

static gboolean
parsebin_autoplug_continue_cb (GstElement * parsebin, GstPad * pad,
    GstCaps * caps, GstDecodebin3 * dbin)
{
  GST_DEBUG_OBJECT (pad, "caps %" GST_PTR_FORMAT, caps);

  /* If it matches our target caps, expose it */
  if (gst_caps_can_intersect (caps, dbin->caps))
    return FALSE;

  return TRUE;
}

/** gst_decodebin_input_set_group_id:
 * @input: A #DecodebinInput
 * @group_id: The new group_id received on that input
 *
 * This method should be called whenever a STREAM_START event comes out of a
 * given input (via parsebin or identity).
 *
 * It will update the input group_id if needed, and also compute and update the
 * current_group_id of #GstDecodebin3.
 *
 * Returns: TRUE if the caller shall replace the group_id
*/
static gboolean
gst_decodebin_input_set_group_id (DecodebinInput * input, guint32 * group_id)
{
  GstDecodebin3 *dbin = input->dbin;

  if (input->group_id != *group_id) {
    if (input->group_id != GST_GROUP_ID_INVALID)
      GST_WARNING_OBJECT (dbin,
          "Group id changed (%" G_GUINT32_FORMAT " -> %" G_GUINT32_FORMAT
          ") on input %p ", input->group_id, *group_id, input);
    input->group_id = *group_id;
  }

  if (*group_id != dbin->current_group_id) {
    /* The input is being re-used with a different incoming stream, we do want
     * to change/unify to this new group-id */
    if (dbin->current_group_id == GST_GROUP_ID_INVALID) {
      GST_DEBUG_OBJECT (dbin,
          "Setting current group id to %" G_GUINT32_FORMAT, *group_id);
      dbin->current_group_id = *group_id;
    } else {
      GST_DEBUG_OBJECT (dbin, "Returning global group id %" G_GUINT32_FORMAT,
          dbin->current_group_id);
    }
    *group_id = dbin->current_group_id;
    return TRUE;
  }

  return FALSE;
}

static void
parsebin_drained_cb (GstElement * parsebin, DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;
  gboolean all_drained;
  GList *tmp;

  GST_INFO_OBJECT (dbin, "input %p drained", input);
  input->drained = TRUE;

  all_drained = dbin->main_input->drained;
  for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
    DecodebinInput *data = (DecodebinInput *) tmp->data;

    all_drained &= data->drained;
  }

  if (all_drained) {
    GST_INFO_OBJECT (dbin, "All inputs drained. Posting about-to-finish");
    g_signal_emit (dbin, gst_decodebin3_signals[SIGNAL_ABOUT_TO_FINISH], 0,
        NULL);
  }
}

static gboolean
clear_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
{
  GST_DEBUG_OBJECT (pad, "clearing sticky event %" GST_PTR_FORMAT, *event);
  gst_event_unref (*event);
  *event = NULL;
  return TRUE;
}

static gboolean
copy_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
{
  GstPad *gpad = GST_PAD_CAST (user_data);

  GST_DEBUG_OBJECT (gpad, "store sticky event %" GST_PTR_FORMAT, *event);
  gst_pad_store_sticky_event (gpad, *event);

  return TRUE;
}

static gboolean
decode_pad_set_target (GstGhostPad * pad, GstPad * target)
{
  gboolean res = gst_ghost_pad_set_target (pad, target);
  if (!res)
    return res;

  if (target == NULL)
    gst_pad_sticky_events_foreach (GST_PAD_CAST (pad), clear_sticky_events,
        NULL);
  else
    gst_pad_sticky_events_foreach (target, copy_sticky_events, pad);

  return res;
}

static CandidateDecoder *
add_candidate_decoder (GstDecodebin3 * dbin, GstElement * element)
{
  GST_OBJECT_LOCK (dbin);
  CandidateDecoder *candidate;
  candidate = g_new0 (CandidateDecoder, 1);
  candidate->element = element;
  dbin->candidate_decoders =
      g_list_prepend (dbin->candidate_decoders, candidate);
  GST_OBJECT_UNLOCK (dbin);
  return candidate;
}

static void
remove_candidate_decoder (GstDecodebin3 * dbin, CandidateDecoder * candidate)
{
  GST_OBJECT_LOCK (dbin);
  dbin->candidate_decoders =
      g_list_remove (dbin->candidate_decoders, candidate);
  if (candidate->error)
    gst_message_unref (candidate->error);
  g_free (candidate);
  GST_OBJECT_UNLOCK (dbin);
}

/** gst_decodebin_input_ensure_parsebin:
 * @input: A #DecodebinInput
 *
 * Ensure the given @input has a parsebin properly setup for it.
 *
 * Returns: TRUE if parsebin could be properly set, else FALSE.
 */
/* Call with INPUT_LOCK taken */
static gboolean
gst_decodebin_input_ensure_parsebin (DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;
  gboolean set_state = FALSE;

  if (input->parsebin == NULL) {
    input->parsebin = gst_element_factory_make ("parsebin", NULL);
    if (input->parsebin == NULL)
      goto no_parsebin;
    input->parsebin = gst_object_ref (input->parsebin);
    input->parsebin_sink = gst_element_get_static_pad (input->parsebin, "sink");
    input->pad_added_sigid =
        g_signal_connect (input->parsebin, "pad-added",
        (GCallback) parsebin_pad_added_cb, input);
    input->pad_removed_sigid =
        g_signal_connect (input->parsebin, "pad-removed",
        (GCallback) parsebin_pad_removed_cb, input);
    input->drained_sigid =
        g_signal_connect (input->parsebin, "drained",
        (GCallback) parsebin_drained_cb, input);
    g_signal_connect (input->parsebin, "autoplug-continue",
        (GCallback) parsebin_autoplug_continue_cb, dbin);
  }

  if (GST_OBJECT_PARENT (GST_OBJECT (input->parsebin)) != GST_OBJECT (dbin)) {
    /* The state lock is taken so that we ensure we are the one (de)activating
     * parsebin. We need to do this to ensure any activation taking place in
     * parsebin (including by elements doing upstream activation) are done
     * within the same thread. */
    GST_STATE_LOCK (input->parsebin);
    gst_bin_add (GST_BIN (dbin), input->parsebin);
    set_state = TRUE;
  }

  gst_ghost_pad_set_target (GST_GHOST_PAD (input->ghost_sink),
      input->parsebin_sink);

  if (set_state) {
    gst_element_sync_state_with_parent (input->parsebin);
    GST_STATE_UNLOCK (input->parsebin);
  }

  return TRUE;

  /* ERRORS */
no_parsebin:
  {
    gst_element_post_message ((GstElement *) dbin,
        gst_missing_element_message_new ((GstElement *) dbin, "parsebin"));
    return FALSE;
  }
}

static GstPadLinkReturn
gst_decodebin3_input_pad_link (GstPad * pad, GstObject * parent, GstPad * peer)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) parent;
  GstQuery *query;
  gboolean pull_mode = FALSE;
  gboolean has_caps = TRUE;
  GstPadLinkReturn res = GST_PAD_LINK_OK;
  DecodebinInput *input = g_object_get_data (G_OBJECT (pad), "decodebin.input");

  g_return_val_if_fail (input, GST_PAD_LINK_REFUSED);

  GST_LOG_OBJECT (parent, "Got link on input pad %" GST_PTR_FORMAT, pad);

  query = gst_query_new_scheduling ();
  if (gst_pad_query (peer, query)
      && gst_query_has_scheduling_mode_with_flags (query, GST_PAD_MODE_PULL,
          GST_SCHEDULING_FLAG_SEEKABLE))
    pull_mode = TRUE;
  gst_query_unref (query);

  GST_DEBUG_OBJECT (dbin, "Upstream can do pull-based : %d", pull_mode);

  if (!pull_mode) {
    /* If push-based, query if it will provide some caps */
    query = gst_query_new_caps (NULL);
    if (gst_pad_query (peer, query)) {
      GstCaps *rescaps = NULL;
      gst_query_parse_caps_result (query, &rescaps);
      if (!rescaps || gst_caps_is_any (rescaps) || gst_caps_is_empty (rescaps)) {
        GST_DEBUG_OBJECT (dbin, "Upstream can't provide caps");
        has_caps = FALSE;
      }
    }
    gst_query_unref (query);
  }

  /* If upstream *can* do pull-based OR it doesn't have any caps, we always use
   * a parsebin. If not, we will delay that decision to a later stage
   * (caps/stream/collection event processing) to figure out if one is really
   * needed or whether an identity element will be enough */
  INPUT_LOCK (dbin);
  if (pull_mode || !has_caps) {
    if (!gst_decodebin_input_ensure_parsebin (input))
      res = GST_PAD_LINK_REFUSED;
    else if (input->identity) {
      GST_ERROR_OBJECT (dbin,
          "Can't reconfigure input from push-based to pull-based");
      res = GST_PAD_LINK_REFUSED;
    }
  }

  /* Clear stream-collection corresponding to current INPUT.  We do not
   * recalculate the global one yet, it will be done when at least one
   * collection is received/computed for this input.
   */
  if (input->collection) {
    GST_DEBUG_OBJECT (pad, "Clearing input collection");
    gst_object_unref (input->collection);
    input->collection = NULL;
  }

  INPUT_UNLOCK (dbin);

  return res;
}

/* Drop duration query during _input_pad_unlink */
static GstPadProbeReturn
query_duration_drop_probe (GstPad * pad, GstPadProbeInfo * info,
    DecodebinInput * input)
{
  GstPadProbeReturn ret = GST_PAD_PROBE_OK;

  if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
    GstQuery *query = GST_PAD_PROBE_INFO_QUERY (info);
    if (GST_QUERY_TYPE (query) == GST_QUERY_DURATION) {
      GST_LOG_OBJECT (pad, "stop forwarding query duration");
      ret = GST_PAD_PROBE_HANDLED;
    }
  }

  return ret;
}

/* CALL with INPUT LOCK */
static void
recalculate_group_id (GstDecodebin3 * dbin)
{
  guint32 common_group_id;
  GList *iter;

  GST_DEBUG_OBJECT (dbin,
      "recalculating, current global group_id: %" G_GUINT32_FORMAT,
      dbin->current_group_id);

  common_group_id = dbin->main_input->group_id;

  for (iter = dbin->other_inputs; iter; iter = iter->next) {
    DecodebinInput *input = iter->data;

    if (input->group_id != common_group_id) {
      if (common_group_id != GST_GROUP_ID_INVALID)
        return;

      common_group_id = input->group_id;
    }
  }

  if (common_group_id == dbin->current_group_id) {
    GST_DEBUG_OBJECT (dbin, "Global group_id hasn't changed");
  } else {
    GST_DEBUG_OBJECT (dbin, "Updating global group_id to %" G_GUINT32_FORMAT,
        common_group_id);
    dbin->current_group_id = common_group_id;
  }
}

/** gst_decodebin_input_reset_parsebin:
 * @dbin:
 * @input: A #DecodebinInput
 *
 * Reset the parsebin of @input (if any) by resetting all associated variables,
 * inputstreams and elements.
 *
 * Call with INPUT LOCK taken
 */
static void
gst_decodebin_input_reset_parsebin (GstDecodebin3 * dbin,
    DecodebinInput * input)
{
  GList *iter;

  if (input->parsebin == NULL)
    return;

  GST_DEBUG_OBJECT (dbin, "Resetting %" GST_PTR_FORMAT, input->parsebin);

  GST_STATE_LOCK (dbin);
  gst_element_set_state (input->parsebin, GST_STATE_NULL);
  input->drained = FALSE;
  input->group_id = GST_GROUP_ID_INVALID;
  recalculate_group_id (dbin);
  for (iter = dbin->input_streams; iter; iter = iter->next) {
    DecodebinInputStream *istream = iter->data;
    if (istream->input == input)
      istream->saw_eos = TRUE;
  }
  gst_element_sync_state_with_parent (input->parsebin);
  GST_STATE_UNLOCK (dbin);
}


static void
gst_decodebin3_input_pad_unlink (GstPad * pad, GstPad * peer,
    DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;

  g_return_if_fail (input);

  GST_LOG_OBJECT (dbin, "Got unlink on input pad %" GST_PTR_FORMAT, pad);

  INPUT_LOCK (dbin);

  if (input->parsebin && GST_PAD_MODE (pad) == GST_PAD_MODE_PULL) {
    GST_DEBUG_OBJECT (dbin, "Resetting parsebin since it's pull-based");
    gst_decodebin_input_reset_parsebin (dbin, input);
  }
  /* In all cases we will be receiving new stream-start and data */
  input->group_id = GST_GROUP_ID_INVALID;
  input->drained = FALSE;
  recalculate_group_id (dbin);

  INPUT_UNLOCK (dbin);
}

static void
gst_decodebin3_release_pad (GstElement * element, GstPad * pad)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) element;
  DecodebinInput *input = g_object_get_data (G_OBJECT (pad), "decodebin.input");
  gulong probe_id = 0;
  GstMessage *msg;

  g_return_if_fail (input);
  GST_LOG_OBJECT (dbin, "Releasing pad %" GST_PTR_FORMAT, pad);

  INPUT_LOCK (dbin);

  /* Clear stream-collection corresponding to current INPUT and post new
   * stream-collection message, if needed */
  if (input->collection) {
    gst_object_unref (input->collection);
    input->collection = NULL;
  }

  msg = handle_stream_collection_locked (dbin, NULL, input);

  if (msg) {
    if (input->parsebin)
      /* Drop duration queries that the application might be doing while this
       * message is posted */
      probe_id =
          gst_pad_add_probe (input->parsebin_sink,
          GST_PAD_PROBE_TYPE_QUERY_UPSTREAM,
          (GstPadProbeCallback) query_duration_drop_probe, input, NULL);

    gst_element_post_message (GST_ELEMENT_CAST (dbin), msg);
    update_requested_selection (dbin);

    if (input->parsebin) {
      gst_pad_remove_probe (input->parsebin_sink, probe_id);
    }
  }

  if (!input->is_main) {
    dbin->other_inputs = g_list_remove (dbin->other_inputs, input);
    gst_decodebin_input_free (input);
  } else
    gst_decodebin_input_reset (input);

  INPUT_UNLOCK (dbin);
  return;
}

/** gst_decodebin_input_reset:
 * @input: The #DecodebinInput to reset
 *
 * Resets the @input for re-used. Call with the INPUT_LOCK.
 */
static void
gst_decodebin_input_reset (DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;

  g_return_if_fail (dbin);

  GST_LOG_OBJECT (dbin, "Resetting input %p", input);

  gst_ghost_pad_set_target (GST_GHOST_PAD (input->ghost_sink), NULL);

  if (input->parsebin) {
    g_signal_handler_disconnect (input->parsebin, input->pad_removed_sigid);
    g_signal_handler_disconnect (input->parsebin, input->pad_added_sigid);
    g_signal_handler_disconnect (input->parsebin, input->drained_sigid);
    gst_element_set_state (input->parsebin, GST_STATE_NULL);
    gst_bin_remove (GST_BIN (dbin), input->parsebin);
    gst_clear_object (&input->parsebin);
    gst_clear_object (&input->parsebin_sink);
  }
  if (input->identity) {
    GstPad *idpad = gst_element_get_static_pad (input->identity, "src");
    DecodebinInputStream *stream;

    SELECTION_LOCK (dbin);
    stream = find_input_stream_for_pad (dbin, idpad);
    remove_input_stream (dbin, stream);
    SELECTION_UNLOCK (dbin);

    gst_object_unref (idpad);

    gst_element_set_state (input->identity, GST_STATE_NULL);
    gst_bin_remove (GST_BIN (dbin), input->identity);
    gst_clear_object (&input->identity);
  }
  if (input->collection)
    gst_clear_object (&input->collection);

  input->group_id = GST_GROUP_ID_INVALID;
}

/** gst_decodebin_input_free:
 * @input: The #DecodebinInput to free
 *
 * Frees the @input and removes the ghost pad from decodebin
 *
 * Call with INPUT LOCK taken
 */
static void
gst_decodebin_input_free (DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;

  g_return_if_fail (dbin);

  gst_decodebin_input_reset (input);

  GST_LOG_OBJECT (dbin, "Freeing input %p", input);

  INPUT_UNLOCK (dbin);
  gst_element_remove_pad (GST_ELEMENT (dbin), input->ghost_sink);
  INPUT_LOCK (dbin);

  g_free (input);
}

static gboolean
sink_query_function (GstPad * sinkpad, GstDecodebin3 * dbin, GstQuery * query)
{
  DecodebinInput *input =
      g_object_get_data (G_OBJECT (sinkpad), "decodebin.input");

  g_return_val_if_fail (input, FALSE);

  GST_DEBUG_OBJECT (sinkpad, "query %" GST_PTR_FORMAT, query);

  /* We accept any caps, since we will reconfigure ourself internally if the new
   * stream is incompatible */
  if (GST_QUERY_TYPE (query) == GST_QUERY_ACCEPT_CAPS) {
    GST_DEBUG_OBJECT (dbin, "Accepting ACCEPT_CAPS query");
    gst_query_set_accept_caps_result (query, TRUE);
    return TRUE;
  }
  return gst_pad_query_default (sinkpad, GST_OBJECT (dbin), query);
}

/** gst_decodebin_input_requires_parsebin:
 * @input: A #DecodebinInput
 * @newcaps: The new incoming #GstCaps
 *
 * Returns: #TRUE if @input requires setting up a `parsebin` element for the
 * incoming stream and @newcaps.
 */
static gboolean
gst_decodebin_input_requires_parsebin (DecodebinInput * input,
    GstCaps * newcaps)
{
  GstDecodebin3 *dbin = input->dbin;
  GstPad *sinkpad = input->ghost_sink;
  gboolean parsebin_needed = TRUE;
  GstStream *stream;

  stream = gst_pad_get_stream (sinkpad);

  if (stream == NULL) {
    /* If upstream didn't provide a `GstStream` we will need to create a
     * parsebin to handle that stream */
    GST_DEBUG_OBJECT (sinkpad,
        "Need to create parsebin since upstream doesn't provide GstStream");
  } else if (gst_caps_can_intersect (newcaps, dbin->caps)) {
    /* If the incoming caps match decodebin3 output, no processing is needed */
    GST_FIXME_OBJECT (sinkpad, "parsebin not needed (matches output caps) !");
    parsebin_needed = FALSE;
  } else if (input->input_is_parsed) {
    GST_DEBUG_OBJECT (sinkpad, "input is parsed, no parsebin needed");
    parsebin_needed = FALSE;
  } else {
    GList *decoder_list;
    /* If the incoming caps are compatible with a decoder, we don't need to
     * process it before */
    g_mutex_lock (&dbin->factories_lock);
    gst_decode_bin_update_factories_list (dbin);
    decoder_list =
        gst_element_factory_list_filter (dbin->decoder_factories, newcaps,
        GST_PAD_SINK, TRUE);
    g_mutex_unlock (&dbin->factories_lock);
    if (decoder_list) {
      GST_FIXME_OBJECT (sinkpad, "parsebin not needed (available decoders) !");
      gst_plugin_feature_list_free (decoder_list);
      parsebin_needed = FALSE;
    }
  }
  if (stream)
    gst_object_unref (stream);

  return parsebin_needed;
}

/** gst_decodebin_input_setup_identity:
 * @input: A #DecodebinInput
 *
 * Sets up @input to receive a single elementary stream with `identity`.
 **/
static void
gst_decodebin_input_setup_identity (DecodebinInput * input)
{
  GstDecodebin3 *dbin = input->dbin;
  GstPad *idsrc, *idsink;

  GST_DEBUG_OBJECT (input->ghost_sink, "Adding identity for new input stream");

  input->identity = gst_element_factory_make ("identity", NULL);
  /* We drop allocation queries due to our usage of multiqueue just
   * afterwards. It is just too dangerous.
   *
   * If application users want to have optimal raw source <=> sink allocations
   * they should not use decodebin3
   */
  g_object_set (input->identity, "drop-allocation", TRUE, NULL);
  input->identity = gst_object_ref (input->identity);
  idsink = gst_element_get_static_pad (input->identity, "sink");
  idsrc = gst_element_get_static_pad (input->identity, "src");
  gst_bin_add (GST_BIN (dbin), input->identity);

  /* Forward any existing GstStream directly on the input stream */
  gst_decodebin_input_add_stream (input, idsrc,
      gst_pad_get_stream (input->ghost_sink));

  gst_object_unref (idsrc);
  gst_object_unref (idsink);
  gst_ghost_pad_set_target (GST_GHOST_PAD (input->ghost_sink), idsink);
  gst_element_sync_state_with_parent (input->identity);
}

static gboolean
sink_event_function (GstPad * sinkpad, GstDecodebin3 * dbin, GstEvent * event)
{
  DecodebinInput *input =
      g_object_get_data (G_OBJECT (sinkpad), "decodebin.input");

  g_return_val_if_fail (input, FALSE);

  GST_DEBUG_OBJECT (sinkpad, "event %" GST_PTR_FORMAT, event);

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_STREAM_START:
    {
      GstQuery *q = gst_query_new_selectable ();
      const GstStructure *s = gst_event_get_structure (event);

      /* Query whether upstream can handle stream selection or not */
      if (gst_pad_peer_query (sinkpad, q)) {
        gst_query_parse_selectable (q, &input->upstream_selected);
        GST_DEBUG_OBJECT (sinkpad, "Upstream is selectable : %d",
            input->upstream_selected);
      } else {
        input->upstream_selected = FALSE;
        GST_DEBUG_OBJECT (sinkpad, "Upstream does not handle SELECTABLE query");
      }
      gst_query_unref (q);

      /* FIXME : We force `decodebin3` to upstream selection mode if *any* of the
         inputs is. This means things might break if there's a mix */
      if (input->upstream_selected)
        dbin->upstream_handles_selection = TRUE;

      input->input_is_parsed = s
          && gst_structure_has_field (s, "urisourcebin-parsed-data");
      if (input->input_is_parsed) {
        /* We remove the custom field from stream-start so as not to pollute
         * downstream */
        event = gst_event_make_writable (event);
        s = gst_event_get_structure (event);
        gst_structure_remove_field ((GstStructure *) s,
            "urisourcebin-parsed-data");
      }

      /* Make sure group ids will be recalculated */
      input->group_id = GST_GROUP_ID_INVALID;
      INPUT_LOCK (dbin);
      recalculate_group_id (dbin);
      INPUT_UNLOCK (dbin);
      break;
    }
    case GST_EVENT_STREAM_COLLECTION:
    {
      GstStreamCollection *collection = NULL;

      gst_event_parse_stream_collection (event, &collection);
      if (collection) {
        GstMessage *collection_msg;
        INPUT_LOCK (dbin);
        collection_msg =
            handle_stream_collection_locked (dbin, collection, input);
        gst_object_unref (collection);
        INPUT_UNLOCK (dbin);
        if (collection_msg)
          gst_element_post_message (GST_ELEMENT_CAST (dbin), collection_msg);

        /* In all cases we want to make sure the selection is valid */
        update_requested_selection (dbin);
      }

      /* If we are waiting to create an identity passthrough, do it now */
      if (!input->parsebin && !input->identity)
        gst_decodebin_input_setup_identity (input);
      break;
    }
    case GST_EVENT_CAPS:
    {
      GstCaps *newcaps = NULL;

      gst_event_parse_caps (event, &newcaps);
      if (!newcaps)
        break;
      GST_DEBUG_OBJECT (sinkpad, "new caps %" GST_PTR_FORMAT, newcaps);

      /* No parsebin or identity present, check if we can avoid creating one */
      if (!input->parsebin && !input->identity) {
        if (gst_decodebin_input_requires_parsebin (input, newcaps)) {
          GST_DEBUG_OBJECT (sinkpad, "parsebin is required for input");
          gst_decodebin_input_ensure_parsebin (input);
          break;
        }
        GST_DEBUG_OBJECT (sinkpad,
            "parsebin not required. Will create identity passthrough element once we get the collection");
        break;
      }

      if (input->identity) {
        if (gst_decodebin_input_requires_parsebin (input, newcaps)) {
          GST_ERROR_OBJECT (sinkpad,
              "Switching from passthrough to parsebin on inputs is not supported !");
          gst_event_unref (event);
          return FALSE;
        }
        /* Nothing else to do here */
        break;
      }

      /* Check if the parsebin present can handle the new caps */
      g_assert (input->parsebin);
      GST_DEBUG_OBJECT (sinkpad,
          "New caps, checking if they are compatible with existing parsebin");
      if (!gst_pad_query_accept_caps (input->parsebin_sink, newcaps)) {
        GST_DEBUG_OBJECT (sinkpad,
            "Parsebin doesn't accept the new caps %" GST_PTR_FORMAT, newcaps);
        /* Reset parsebin so that it reconfigures itself for the new stream format */
        INPUT_LOCK (dbin);
        gst_decodebin_input_reset_parsebin (dbin, input);
        INPUT_UNLOCK (dbin);
      } else {
        GST_DEBUG_OBJECT (sinkpad, "Parsebin accepts new caps");
      }
      break;
    }
    case GST_EVENT_SEGMENT:
    {
      const GstSegment *segment = NULL;
      gst_event_parse_segment (event, &segment);

      /* All data reaching multiqueue must be in time format. If it's not, we
       * need to use a parsebin on the incoming stream.
       */
      if (segment && segment->format != GST_FORMAT_TIME && !input->parsebin) {
        GST_DEBUG_OBJECT (sinkpad,
            "Got a non-time segment, forcing parsebin handling");
        gst_decodebin_input_ensure_parsebin (input);
      }
      break;
    }
    default:
      break;
  }

  /* Chain to parent function */
  return gst_pad_event_default (sinkpad, GST_OBJECT (dbin), event);
}

/** gst_decodebin_input_new:
 * @dbin: The decodebin instance
 * @main: Whether this is the main input (for the static "sink" sinkpad) or not
 *
 * Returns: A new #DecodebinInput
 */
static DecodebinInput *
gst_decodebin_input_new (GstDecodebin3 * dbin, gboolean main)
{
  DecodebinInput *input;

  input = g_new0 (DecodebinInput, 1);
  input->dbin = dbin;
  input->is_main = main;
  input->group_id = GST_GROUP_ID_INVALID;
  if (main)
    input->ghost_sink = gst_ghost_pad_new_no_target ("sink", GST_PAD_SINK);
  else {
    gchar *pad_name = g_strdup_printf ("sink_%u", dbin->input_counter++);
    input->ghost_sink = gst_ghost_pad_new_no_target (pad_name, GST_PAD_SINK);
    g_free (pad_name);
  }
  input->upstream_selected = FALSE;
  g_object_set_data (G_OBJECT (input->ghost_sink), "decodebin.input", input);
  gst_pad_set_event_function (input->ghost_sink,
      (GstPadEventFunction) sink_event_function);
  gst_pad_set_query_function (input->ghost_sink,
      (GstPadQueryFunction) sink_query_function);
  gst_pad_set_link_function (input->ghost_sink, gst_decodebin3_input_pad_link);
  g_signal_connect (input->ghost_sink, "unlinked",
      (GCallback) gst_decodebin3_input_pad_unlink, input);

  gst_pad_set_active (input->ghost_sink, TRUE);
  gst_element_add_pad ((GstElement *) dbin, input->ghost_sink);

  return input;

}

static GstPad *
gst_decodebin3_request_new_pad (GstElement * element, GstPadTemplate * temp,
    const gchar * name, const GstCaps * caps)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) element;
  DecodebinInput *input;
  GstPad *res = NULL;

  /* We are ignoring names for the time being, not sure it makes any sense
   * within the context of decodebin3 ... */
  input = gst_decodebin_input_new (dbin, FALSE);
  if (input) {
    INPUT_LOCK (dbin);
    dbin->other_inputs = g_list_append (dbin->other_inputs, input);
    res = input->ghost_sink;
    INPUT_UNLOCK (dbin);
  }

  return res;
}

/* Must be called with factories lock! */
static void
gst_decode_bin_update_factories_list (GstDecodebin3 * dbin)
{
  guint cookie;

  cookie = gst_registry_get_feature_list_cookie (gst_registry_get ());
  if (!dbin->factories || dbin->factories_cookie != cookie) {
    GList *tmp;
    if (dbin->factories)
      gst_plugin_feature_list_free (dbin->factories);
    if (dbin->decoder_factories)
      g_list_free (dbin->decoder_factories);
    if (dbin->decodable_factories)
      g_list_free (dbin->decodable_factories);
    dbin->factories =
        gst_element_factory_list_get_elements
        (GST_ELEMENT_FACTORY_TYPE_DECODABLE, GST_RANK_MARGINAL);
    dbin->factories =
        g_list_sort (dbin->factories, gst_plugin_feature_rank_compare_func);
    dbin->factories_cookie = cookie;

    /* Filter decoder and other decodables */
    dbin->decoder_factories = NULL;
    dbin->decodable_factories = NULL;
    for (tmp = dbin->factories; tmp; tmp = tmp->next) {
      GstElementFactory *fact = (GstElementFactory *) tmp->data;
      if (gst_element_factory_list_is_type (fact,
              GST_ELEMENT_FACTORY_TYPE_DECODER))
        dbin->decoder_factories = g_list_append (dbin->decoder_factories, fact);
      else
        dbin->decodable_factories =
            g_list_append (dbin->decodable_factories, fact);
    }
  }
}

/* Must be called with appropriate lock if list is a protected variable */
static const gchar *
stream_in_list (GList * list, const gchar * sid)
{
  GList *tmp;

#if EXTRA_DEBUG
  for (tmp = list; tmp; tmp = tmp->next) {
    gchar *osid = (gchar *) tmp->data;
    GST_DEBUG ("Checking %s against %s", sid, osid);
  }
#endif

  for (tmp = list; tmp; tmp = tmp->next) {
    const gchar *osid = (gchar *) tmp->data;
    if (!g_strcmp0 (sid, osid))
      return osid;
  }

  return NULL;
}

static GList *
remove_from_list (GList * list, const gchar * sid)
{
  GList *tmp;

  for (tmp = list; tmp; tmp = tmp->next) {
    gchar *osid = tmp->data;
    if (!g_strcmp0 (sid, osid)) {
      g_free (osid);
      return g_list_delete_link (list, tmp);
    }
  }
  return list;
}

static gboolean
stream_list_equal (GList * lista, GList * listb)
{
  GList *tmp;

  if (g_list_length (lista) != g_list_length (listb))
    return FALSE;

  for (tmp = lista; tmp; tmp = tmp->next) {
    gchar *osid = tmp->data;
    if (!stream_in_list (listb, osid))
      return FALSE;
  }

  return TRUE;
}

static void
update_requested_selection (GstDecodebin3 * dbin)
{
  guint i, nb;
  GList *tmp = NULL;
  gboolean all_user_selected = TRUE;
  GstStreamType used_types = 0;
  GstStreamCollection *collection;

  /* 1. Is there a pending SELECT_STREAMS we can return straight away since
   *  the switch handler will take care of the pending selection */
  SELECTION_LOCK (dbin);
  if (dbin->pending_select_streams) {
    GST_DEBUG_OBJECT (dbin,
        "No need to create pending selection, SELECT_STREAMS underway");
    goto beach;
  }

  collection = dbin->collection;
  if (G_UNLIKELY (collection == NULL)) {
    GST_DEBUG_OBJECT (dbin, "No current GstStreamCollection");
    goto beach;
  }
  nb = gst_stream_collection_get_size (collection);

  /* 2. If not, are we in EXPOSE_ALL_MODE ? If so, match everything */
  GST_FIXME_OBJECT (dbin, "Implement EXPOSE_ALL_MODE");

  /* 3. If not, check if we already have some of the streams in the
   * existing active/requested selection */
  for (i = 0; i < nb; i++) {
    GstStream *stream = gst_stream_collection_get_stream (collection, i);
    const gchar *sid = gst_stream_get_stream_id (stream);
    gint request = -1;
    /* Fire select-stream signal to see if outside components want to
     * hint at which streams should be selected */
    g_signal_emit (G_OBJECT (dbin),
        gst_decodebin3_signals[SIGNAL_SELECT_STREAM], 0, collection, stream,
        &request);
    GST_DEBUG_OBJECT (dbin, "stream %s , request:%d", sid, request);

    if (request == -1)
      all_user_selected = FALSE;
    if (request == 1 || (request == -1
            && (stream_in_list (dbin->requested_selection, sid)
                || stream_in_list (dbin->active_selection, sid)))) {
      GstStreamType curtype = gst_stream_get_stream_type (stream);
      if (request == 1)
        GST_DEBUG_OBJECT (dbin,
            "Using stream requested by 'select-stream' signal : %s", sid);
      else
        GST_DEBUG_OBJECT (dbin,
            "Re-using stream already present in requested or active selection : %s",
            sid);
      tmp = g_list_append (tmp, (gchar *) sid);
      used_types |= curtype;
    }
  }

  /* 4. If the user didn't explicitly selected all streams, match one stream of each type */
  if (!all_user_selected && dbin->select_streams_seqnum == GST_SEQNUM_INVALID) {
    for (i = 0; i < nb; i++) {
      GstStream *stream = gst_stream_collection_get_stream (collection, i);
      GstStreamType curtype = gst_stream_get_stream_type (stream);
      if (curtype != GST_STREAM_TYPE_UNKNOWN && !(used_types & curtype)) {
        const gchar *sid = gst_stream_get_stream_id (stream);
        GST_DEBUG_OBJECT (dbin,
            "Automatically selecting stream '%s' of type %s", sid,
            gst_stream_type_get_name (curtype));
        tmp = g_list_append (tmp, (gchar *) sid);
        used_types |= curtype;
      }
    }
  }

beach:
  if (stream_list_equal (tmp, dbin->requested_selection)) {
    /* If the selection is equal, there is nothign to do */
    GST_DEBUG_OBJECT (dbin, "Dropping duplicate selection");
    g_list_free (tmp);
    tmp = NULL;
  }

  if (tmp) {
    /* Finally set the requested selection */
    if (dbin->requested_selection) {
      GST_FIXME_OBJECT (dbin,
          "Replacing non-NULL requested_selection, what should we do ??");
      g_list_free_full (dbin->requested_selection, g_free);
    }
    dbin->requested_selection =
        g_list_copy_deep (tmp, (GCopyFunc) g_strdup, NULL);
    dbin->selection_updated = TRUE;
    g_list_free (tmp);
  }
  SELECTION_UNLOCK (dbin);
}

/* sort_streams:
 * GCompareFunc to use with lists of GstStream.
 * Sorts GstStreams by stream type and SELECT flag and stream-id
 * First video, then audio, then others.
 *
 * Return: negative if a<b, 0 if a==b, positive if a>b
 */
static gint
sort_streams (GstStream * sa, GstStream * sb)
{
  GstStreamType typea, typeb;
  GstStreamFlags flaga, flagb;
  const gchar *ida, *idb;
  gint ret = 0;

  typea = gst_stream_get_stream_type (sa);
  typeb = gst_stream_get_stream_type (sb);

  GST_LOG ("sa(%s), sb(%s)", gst_stream_get_stream_id (sa),
      gst_stream_get_stream_id (sb));

  /* Sort by stream type. First video, then audio, then others(text, container, unknown) */
  if (typea != typeb) {
    if (typea & GST_STREAM_TYPE_VIDEO)
      ret = -1;
    else if (typea & GST_STREAM_TYPE_AUDIO)
      ret = (!(typeb & GST_STREAM_TYPE_VIDEO)) ? -1 : 1;
    else if (typea & GST_STREAM_TYPE_TEXT)
      ret = (!(typeb & GST_STREAM_TYPE_VIDEO)
          && !(typeb & GST_STREAM_TYPE_AUDIO)) ? -1 : 1;
    else if (typea & GST_STREAM_TYPE_CONTAINER)
      ret = (typeb & GST_STREAM_TYPE_UNKNOWN) ? -1 : 1;
    else
      ret = 1;

    if (ret != 0) {
      GST_LOG ("Sort by stream-type: %d", ret);
      return ret;
    }
  }

  /* Sort by SELECT flag, if stream type is same. */
  flaga = gst_stream_get_stream_flags (sa);
  flagb = gst_stream_get_stream_flags (sb);

  ret =
      (flaga & GST_STREAM_FLAG_SELECT) ? ((flagb & GST_STREAM_FLAG_SELECT) ? 0 :
      -1) : ((flagb & GST_STREAM_FLAG_SELECT) ? 1 : 0);

  if (ret != 0) {
    GST_LOG ("Sort by SELECT flag: %d", ret);
    return ret;
  }

  /* Sort by stream-id, if otherwise the same. */
  ida = gst_stream_get_stream_id (sa);
  idb = gst_stream_get_stream_id (sb);
  ret = g_strcmp0 (ida, idb);

  GST_LOG ("Sort by stream-id: %d", ret);

  return ret;
}

/* Call with INPUT_LOCK taken */
static GstStreamCollection *
get_merged_collection (GstDecodebin3 * dbin)
{
  gboolean needs_merge = FALSE;
  GstStreamCollection *res = NULL;
  GList *tmp;
  GList *unsorted_streams = NULL;
  guint i, nb_stream;

  /* First check if we need to do a merge or just return the only collection */
  res = dbin->main_input->collection;

  for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
    DecodebinInput *input = (DecodebinInput *) tmp->data;
    GST_LOG_OBJECT (dbin, "Comparing res %p input->collection %p", res,
        input->collection);
    if (input->collection && input->collection != res) {
      if (res) {
        needs_merge = TRUE;
        break;
      }
      res = input->collection;
    }
  }

  if (!needs_merge) {
    GST_DEBUG_OBJECT (dbin, "No need to merge, returning %p", res);
    return res ? gst_object_ref (res) : NULL;
  }

  /* We really need to create a new collection */
  /* FIXME : Some numbering scheme maybe ?? */
  res = gst_stream_collection_new ("decodebin3");
  if (dbin->main_input->collection) {
    nb_stream = gst_stream_collection_get_size (dbin->main_input->collection);
    GST_DEBUG_OBJECT (dbin, "main input %p %d", dbin->main_input, nb_stream);
    for (i = 0; i < nb_stream; i++) {
      GstStream *stream =
          gst_stream_collection_get_stream (dbin->main_input->collection, i);
      unsorted_streams = g_list_append (unsorted_streams, stream);
    }
  }

  for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
    DecodebinInput *input = (DecodebinInput *) tmp->data;
    GST_DEBUG_OBJECT (dbin, "input %p , collection %p", input,
        input->collection);
    if (input->collection) {
      nb_stream = gst_stream_collection_get_size (input->collection);
      GST_DEBUG_OBJECT (dbin, "nb_stream : %d", nb_stream);
      for (i = 0; i < nb_stream; i++) {
        GstStream *stream =
            gst_stream_collection_get_stream (input->collection, i);
        /* Only add if not already present in the list */
        if (!g_list_find (unsorted_streams, stream))
          unsorted_streams = g_list_append (unsorted_streams, stream);
      }
    }
  }

  /* re-order streams : video, then audio, then others */
  unsorted_streams =
      g_list_sort (unsorted_streams, (GCompareFunc) sort_streams);
  for (tmp = unsorted_streams; tmp; tmp = tmp->next) {
    GstStream *stream = (GstStream *) tmp->data;
    GST_DEBUG_OBJECT (dbin, "Adding #stream(%s) to collection",
        gst_stream_get_stream_id (stream));
    gst_stream_collection_add_stream (res, gst_object_ref (stream));
  }

  if (unsorted_streams)
    g_list_free (unsorted_streams);

  return res;
}

/* Call with INPUT_LOCK taken */
static DecodebinInput *
find_message_parsebin (GstDecodebin3 * dbin, GstElement * child)
{
  DecodebinInput *input = NULL;
  GstElement *parent = gst_object_ref (child);
  GList *tmp;

  do {
    GstElement *next_parent;

    GST_DEBUG_OBJECT (dbin, "parent %s",
        parent ? GST_ELEMENT_NAME (parent) : "<NONE>");

    if (parent == dbin->main_input->parsebin) {
      input = dbin->main_input;
      break;
    }
    for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
      DecodebinInput *cur = (DecodebinInput *) tmp->data;
      if (parent == cur->parsebin) {
        input = cur;
        break;
      }
    }
    next_parent = (GstElement *) gst_element_get_parent (parent);
    gst_object_unref (parent);
    parent = next_parent;

  } while (parent && parent != (GstElement *) dbin);

  if (parent)
    gst_object_unref (parent);

  return input;
}

static const gchar *
stream_in_collection (GstDecodebin3 * dbin, gchar * sid)
{
  guint i, len;

  if (dbin->collection == NULL)
    return NULL;
  len = gst_stream_collection_get_size (dbin->collection);
  for (i = 0; i < len; i++) {
    GstStream *stream = gst_stream_collection_get_stream (dbin->collection, i);
    const gchar *osid = gst_stream_get_stream_id (stream);
    if (!g_strcmp0 (sid, osid))
      return osid;
  }

  return NULL;
}

/** handle_stream_collection_locked:
 * @dbin:
 * @collection: (transfer none): The new collection for @input. Can be %NULL.
 * @input: The #DecodebinInput
 *
 * Called with INPUT_LOCK taken.
 *
 * Handle a new (or updated) @collection for the given @input. If this results
 * in a different collection, the appropriate GST_MESSAGE_STREAM_COLLECTION to
 * be posted will be returned.
 *
 * Returns: A #GstMessage to be posted on the bus if a new collection was
 * generated, else %NULL.
 */
static GstMessage *
handle_stream_collection_locked (GstDecodebin3 * dbin,
    GstStreamCollection * collection, DecodebinInput * input)
{
  GstMessage *message = NULL;
#ifndef GST_DISABLE_GST_DEBUG
  const gchar *upstream_id;
  guint i;
#endif
  if (!input) {
    GST_DEBUG_OBJECT (dbin,
        "Couldn't find corresponding input, most likely shutting down");
    return NULL;
  }

  /* Replace collection in input */
  if (input->collection)
    gst_object_unref (input->collection);
  if (collection)
    input->collection = gst_object_ref (collection);
  GST_DEBUG_OBJECT (dbin, "Setting collection %p on input %p", collection,
      input);

  /* Merge collection if needed */
  collection = get_merged_collection (dbin);
  if (!collection)
    return NULL;

#ifndef GST_DISABLE_GST_DEBUG
  /* Just some debugging */
  upstream_id = gst_stream_collection_get_upstream_id (collection);
  GST_DEBUG ("Received Stream Collection. Upstream_id : %s", upstream_id);
  GST_DEBUG ("From input %p", input);
  GST_DEBUG ("  %d streams", gst_stream_collection_get_size (collection));
  for (i = 0; i < gst_stream_collection_get_size (collection); i++) {
    GstStream *stream = gst_stream_collection_get_stream (collection, i);
    GstTagList *taglist;
    GstCaps *caps;

    GST_DEBUG ("   Stream '%s'", gst_stream_get_stream_id (stream));
    GST_DEBUG ("     type  : %s",
        gst_stream_type_get_name (gst_stream_get_stream_type (stream)));
    GST_DEBUG ("     flags : 0x%x", gst_stream_get_stream_flags (stream));
    taglist = gst_stream_get_tags (stream);
    GST_DEBUG ("     tags  : %" GST_PTR_FORMAT, taglist);
    caps = gst_stream_get_caps (stream);
    GST_DEBUG ("     caps  : %" GST_PTR_FORMAT, caps);
    if (taglist)
      gst_tag_list_unref (taglist);
    if (caps)
      gst_caps_unref (caps);
  }
#endif

  /* Store collection for later usage */
  SELECTION_LOCK (dbin);
  if (dbin->collection == NULL) {
    GST_DEBUG_OBJECT (dbin, "Storing updated global collection");
    dbin->collection = collection;
    message =
        gst_message_new_stream_collection ((GstObject *) dbin, collection);
  } else if (dbin->collection == collection) {
    GST_DEBUG_OBJECT (dbin, "Collection didn't change");
    gst_object_unref (collection);
  } else {
    /* We need to check who emitted this collection (the owner).
     * If we already had a collection from that user, this one is an update,
     * that is to say that we need to figure out how we are going to re-use
     * the streams/slot */
    GST_FIXME_OBJECT (dbin, "New collection but already had one ...");
    /* FIXME : When do we switch from pending collection to active collection ?
     * When all streams from active collection are drained in multiqueue output ? */
    gst_object_unref (dbin->collection);
    dbin->collection = collection;
    message =
        gst_message_new_stream_collection ((GstObject *) dbin, collection);
  }
  if (message)
    dbin->select_streams_seqnum = GST_SEQNUM_INVALID;
  SELECTION_UNLOCK (dbin);

  return message;
}

/* Must be called with the selection lock taken */
static void
gst_decodebin3_update_min_interleave (GstDecodebin3 * dbin)
{
  GstClockTime max_latency = GST_CLOCK_TIME_NONE;
  GList *tmp;

  GST_DEBUG_OBJECT (dbin, "Recalculating max latency of decoders");
  for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
    DecodebinOutputStream *out = (DecodebinOutputStream *) tmp->data;
    if (GST_CLOCK_TIME_IS_VALID (out->decoder_latency)) {
      if (max_latency == GST_CLOCK_TIME_NONE
          || out->decoder_latency > max_latency)
        max_latency = out->decoder_latency;
    }
  }
  GST_DEBUG_OBJECT (dbin, "max latency of all decoders: %" GST_TIME_FORMAT,
      GST_TIME_ARGS (max_latency));

  if (!GST_CLOCK_TIME_IS_VALID (max_latency))
    return;

  /* Make sure we keep an extra overhead */
  max_latency += 100 * GST_MSECOND;
  if (max_latency == dbin->current_mq_min_interleave)
    return;

  dbin->current_mq_min_interleave = max_latency;
  GST_DEBUG_OBJECT (dbin, "Setting mq min-interleave to %" GST_TIME_FORMAT,
      GST_TIME_ARGS (dbin->current_mq_min_interleave));
  g_object_set (dbin->multiqueue, "min-interleave-time",
      dbin->current_mq_min_interleave, NULL);
}

static void
gst_decodebin3_handle_message (GstBin * bin, GstMessage * message)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) bin;
  gboolean posting_collection = FALSE;
  GList *l;

  GST_DEBUG_OBJECT (bin, "Got Message %s", GST_MESSAGE_TYPE_NAME (message));

  GST_OBJECT_LOCK (dbin);
  for (l = dbin->candidate_decoders; l; l = l->next) {
    CandidateDecoder *candidate = l->data;
    if (GST_OBJECT_CAST (candidate->element) == GST_MESSAGE_SRC (message)) {
      if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR) {
        if (candidate->error)
          gst_message_unref (candidate->error);
        candidate->error = message;
        GST_OBJECT_UNLOCK (dbin);
        return;
      } else if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_LATENCY) {
        if (candidate->latency)
          gst_message_unref (candidate->latency);
        GST_DEBUG_OBJECT (bin, "store latency message for %" GST_PTR_FORMAT,
            candidate->element);
        candidate->latency = message;
        GST_OBJECT_UNLOCK (dbin);
        return;
      }
      break;
    }
  }
  GST_OBJECT_UNLOCK (dbin);

  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_STREAM_COLLECTION:
    {
      GstStreamCollection *collection = NULL;
      DecodebinInput *input;
      GstMessage *collection_message;

      INPUT_LOCK (dbin);
      input =
          find_message_parsebin (dbin,
          (GstElement *) GST_MESSAGE_SRC (message));
      if (input == NULL) {
        GST_DEBUG_OBJECT (dbin,
            "Couldn't find corresponding input, most likely shutting down");
        INPUT_UNLOCK (dbin);
        break;
      }
      if (input->upstream_selected) {
        GST_DEBUG_OBJECT (dbin,
            "Upstream handles selection, not using/forwarding collection");
        INPUT_UNLOCK (dbin);
        goto drop_message;
      }
      gst_message_parse_stream_collection (message, &collection);
      if (!collection) {
        INPUT_UNLOCK (dbin);
        break;
      }

      collection_message =
          handle_stream_collection_locked (dbin, collection, input);
      INPUT_UNLOCK (dbin);

      if (collection_message) {
        posting_collection = TRUE;
        gst_message_unref (message);
        message = collection_message;
      }
      gst_object_unref (collection);
      break;
    }
    case GST_MESSAGE_LATENCY:
    {
      GList *tmp;
      /* Check if this is from one of our decoders */
      SELECTION_LOCK (dbin);
      for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
        DecodebinOutputStream *out = (DecodebinOutputStream *) tmp->data;
        if (out->decoder == (GstElement *) GST_MESSAGE_SRC (message)) {
          GstClockTime min, max;
          if (GST_IS_VIDEO_DECODER (out->decoder)) {
            gst_video_decoder_get_latency (GST_VIDEO_DECODER (out->decoder),
                &min, &max);
            GST_DEBUG_OBJECT (dbin,
                "Got latency update from one of our decoders. min: %"
                GST_TIME_FORMAT " max: %" GST_TIME_FORMAT, GST_TIME_ARGS (min),
                GST_TIME_ARGS (max));
            out->decoder_latency = min;
            /* Trigger recalculation */
            gst_decodebin3_update_min_interleave (dbin);
          }
          break;
        }
      }
      SELECTION_UNLOCK (dbin);
    }
    default:
      break;
  }

  GST_BIN_CLASS (parent_class)->handle_message (bin, message);

  if (posting_collection) {
    /* Figure out a selection for that collection */
    update_requested_selection (dbin);
  }

  return;

drop_message:
  {
    GST_DEBUG_OBJECT (bin, "dropping message");
    gst_message_unref (message);
  }
}

static void
handle_stored_latency_message (GstDecodebin3 * dbin,
    DecodebinOutputStream * output, CandidateDecoder * candidate)
{
  GstClockTime min, max;
  if (candidate->latency && GST_IS_VIDEO_DECODER (candidate->element)) {
    gst_video_decoder_get_latency (GST_VIDEO_DECODER (candidate->element),
        &min, &max);
    GST_DEBUG_OBJECT (dbin,
        "Got latency update from %" GST_PTR_FORMAT ". min: %"
        GST_TIME_FORMAT " max: %" GST_TIME_FORMAT, candidate->element,
        GST_TIME_ARGS (min), GST_TIME_ARGS (max));
    output->decoder_latency = min;
    /* Trigger recalculation */
    gst_decodebin3_update_min_interleave (dbin);

    GST_BIN_CLASS (parent_class)->handle_message (GST_BIN_CAST (dbin),
        candidate->latency);
  }
}

static DecodebinOutputStream *
find_free_compatible_output (GstDecodebin3 * dbin, GstStream * stream)
{
  GList *tmp;
  GstStreamType stype = gst_stream_get_stream_type (stream);

  for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
    DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
    if (output->type == stype && output->slot && output->slot->active_stream) {
      GstStream *tstream = output->slot->active_stream;
      if (!stream_in_list (dbin->requested_selection,
              (gchar *) gst_stream_get_stream_id (tstream))) {
        return output;
      }
    }
  }

  return NULL;
}

/* Give a certain slot, figure out if it should be linked to an
 * output stream
 * CALL WITH SELECTION LOCK TAKEN !*/
static DecodebinOutputStream *
get_output_for_slot (MultiQueueSlot * slot)
{
  GstDecodebin3 *dbin = slot->dbin;
  DecodebinOutputStream *output = NULL;
  const gchar *stream_id;
  GstCaps *caps;
  gchar *id_in_list = NULL;

  /* If we already have a configured output, just use it */
  if (slot->output != NULL)
    return slot->output;

  /*
   * FIXME
   *
   * This method needs to be split into multiple parts
   *
   * 1) Figure out whether stream should be exposed or not
   *   This is based on autoplug-continue, EXPOSE_ALL_MODE, or presence
   *   in the default stream attribution
   *
   * 2) Figure out whether an output stream should be created, whether
   *   we can re-use the output stream already linked to the slot, or
   *   whether we need to get re-assigned another (currently used) output
   *   stream.
   */

  stream_id = gst_stream_get_stream_id (slot->active_stream);
  caps = gst_stream_get_caps (slot->active_stream);
  GST_DEBUG_OBJECT (dbin, "stream %s , %" GST_PTR_FORMAT, stream_id, caps);
  gst_caps_unref (caps);

  /* 0. Emit autoplug-continue signal for pending caps ? */
  GST_FIXME_OBJECT (dbin, "emit autoplug-continue");

  /* 1. if in EXPOSE_ALL_MODE, just accept */
  GST_FIXME_OBJECT (dbin, "Handle EXPOSE_ALL_MODE");

  /* 3. In default mode check if we should expose */
  id_in_list = (gchar *) stream_in_list (dbin->requested_selection, stream_id);
  if (id_in_list || dbin->upstream_handles_selection) {
    /* Check if we can steal an existing output stream we could re-use.
     * that is:
     * * an output stream whose slot->stream is not in requested
     * * and is of the same type as this stream
     */
    output = find_free_compatible_output (dbin, slot->active_stream);
    if (output) {
      /* Move this output from its current slot to this slot */
      dbin->to_activate =
          g_list_append (dbin->to_activate, (gchar *) stream_id);
      dbin->requested_selection =
          g_list_remove (dbin->requested_selection, id_in_list);
      g_free (id_in_list);
      SELECTION_UNLOCK (dbin);
      gst_pad_add_probe (output->slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
          (GstPadProbeCallback) slot_unassign_probe, output->slot, NULL);
      SELECTION_LOCK (dbin);
      return NULL;
    }

    output = create_output_stream (dbin, slot->type);
    output->slot = slot;
    GST_DEBUG ("Linking slot %p to new output %p", slot, output);
    slot->output = output;
    GST_DEBUG ("Adding '%s' to active_selection", stream_id);
    dbin->active_selection =
        g_list_append (dbin->active_selection, (gchar *) g_strdup (stream_id));
  } else
    GST_DEBUG ("Not creating any output for slot %p", slot);

  return output;
}

/* Returns SELECTED_STREAMS message if active_selection is equal to
 * requested_selection, else NULL.
 * Must be called with LOCK taken */
static GstMessage *
is_selection_done (GstDecodebin3 * dbin)
{
  GList *tmp;
  GstMessage *msg;

  if (!dbin->selection_updated)
    return NULL;

  GST_LOG_OBJECT (dbin, "Checking");

  if (dbin->upstream_handles_selection) {
    GST_DEBUG ("Upstream handles stream selection, returning");
    return NULL;
  }

  if (dbin->to_activate != NULL) {
    GST_DEBUG ("Still have streams to activate");
    return NULL;
  }
  for (tmp = dbin->requested_selection; tmp; tmp = tmp->next) {
    GST_DEBUG ("Checking requested stream %s", (gchar *) tmp->data);
    if (!stream_in_list (dbin->active_selection, (gchar *) tmp->data)) {
      GST_DEBUG ("Not in active selection, returning");
      return NULL;
    }
  }

  GST_DEBUG_OBJECT (dbin, "Selection active, creating message");

  /* We are completely active */
  msg = gst_message_new_streams_selected ((GstObject *) dbin, dbin->collection);
  if (dbin->select_streams_seqnum != GST_SEQNUM_INVALID) {
    gst_message_set_seqnum (msg, dbin->select_streams_seqnum);
  }
  for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
    DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
    if (output->slot) {
      const gchar *output_streamid =
          gst_stream_get_stream_id (output->slot->active_stream);
      GST_DEBUG_OBJECT (dbin, "Adding stream %s", output_streamid);
      if (stream_in_list (dbin->requested_selection, output_streamid))
        gst_message_streams_selected_add (msg, output->slot->active_stream);
      else
        GST_WARNING_OBJECT (dbin,
            "Output slot still active for old selection ?");
    } else
      GST_WARNING_OBJECT (dbin, "No valid slot for output %p", output);
  }
  dbin->selection_updated = FALSE;
  return msg;
}

/** check_and_drain_multiqueue_locked:
 * @dbin: A #GstDecodebin3
 * @eos_event: The GST_EVENT_EOS that triggered this check.
 *
 * Check if all #DecodebinInputstream and #MultiqueueSlot are
 * emptied/drained. If that is the case, send the final sequence of final EOS
 * events based on the provided @eos_event.
 */
static void
check_and_drain_multiqueue_locked (GstDecodebin3 * dbin, GstEvent * eos_event)
{
  GList *iter;

  GST_DEBUG_OBJECT (dbin, "checking slots for eos");

  for (iter = dbin->slots; iter; iter = iter->next) {
    MultiQueueSlot *slot = iter->data;

    if (slot->output && !slot->is_drained) {
      GST_LOG_OBJECT (slot->sink_pad, "Not drained, not all slots are done");
      return;
    }
  }

  /* Also check with the inputs, data might be pending */
  if (!all_input_streams_are_eos (dbin))
    return;

  GST_DEBUG_OBJECT (dbin,
      "All active slots are drained, and no pending input, push EOS");

  for (iter = dbin->input_streams; iter; iter = iter->next) {
    GstEvent *stream_start, *eos;
    DecodebinInputStream *input = (DecodebinInputStream *) iter->data;
    GstPad *peer = gst_pad_get_peer (input->srcpad);

    if (!peer) {
      GST_DEBUG_OBJECT (input->srcpad, "Not linked to multiqueue");
      continue;
    }

    /* First forward a custom STREAM_START event to reset the EOS status (if
     * any) */
    stream_start =
        gst_pad_get_sticky_event (input->srcpad, GST_EVENT_STREAM_START, 0);
    if (stream_start) {
      GstStructure *s;
      GstEvent *custom_stream_start = gst_event_copy (stream_start);
      gst_event_unref (stream_start);
      s = (GstStructure *) gst_event_get_structure (custom_stream_start);
      gst_structure_set (s, "decodebin3-flushing-stream-start",
          G_TYPE_BOOLEAN, TRUE, NULL);
      gst_pad_send_event (peer, custom_stream_start);
    }
    /* Send EOS to all slots */
    eos = gst_event_new_eos ();
    gst_event_set_seqnum (eos, gst_event_get_seqnum (eos_event));
    gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (eos),
        CUSTOM_FINAL_EOS_QUARK, (gchar *) CUSTOM_FINAL_EOS_QUARK_DATA, NULL);
    gst_pad_send_event (peer, eos);
    gst_object_unref (peer);
  }
}

/*
 * Returns TRUE if there are no more streams to output and an ERROR message
 * should be posted
 */
static inline gboolean
no_more_streams_locked (GstDecodebin3 * dbin)
{
  return (!dbin->requested_selection && !dbin->active_selection
      && !dbin->to_activate);
}

static void
check_slot_reconfiguration (GstDecodebin3 * dbin, MultiQueueSlot * slot)
{
  DecodebinOutputStream *output;
  GstMessage *msg = NULL;
  gboolean no_more_streams;

  SELECTION_LOCK (dbin);
  output = get_output_for_slot (slot);
  if (!output) {
    no_more_streams = no_more_streams_locked (dbin);
    SELECTION_UNLOCK (dbin);
    if (no_more_streams)
      GST_ELEMENT_ERROR (slot->dbin, STREAM, FAILED, (NULL),
          ("No streams to output"));
    return;
  }

  if (!reconfigure_output_stream (output, slot, &msg)) {
    GST_DEBUG_OBJECT (dbin, "Removing failing stream from selection: %s ",
        gst_stream_get_stream_id (slot->active_stream));
    slot->dbin->requested_selection =
        remove_from_list (slot->dbin->requested_selection,
        gst_stream_get_stream_id (slot->active_stream));
    slot->dbin->active_selection =
        remove_from_list (slot->dbin->active_selection,
        gst_stream_get_stream_id (slot->active_stream));
    no_more_streams = no_more_streams_locked (dbin);
    dbin->selection_updated = TRUE;
    SELECTION_UNLOCK (dbin);
    if (msg)
      gst_element_post_message ((GstElement *) slot->dbin, msg);
    if (no_more_streams)
      GST_ELEMENT_ERROR (slot->dbin, CORE, MISSING_PLUGIN, (NULL),
          ("No suitable plugins found"));
    else
      GST_ELEMENT_WARNING (slot->dbin, CORE, MISSING_PLUGIN, (NULL),
          ("Some plugins were missing"));
    reassign_slot (dbin, slot);
  } else {
    GstMessage *selection_msg = is_selection_done (dbin);
    SELECTION_UNLOCK (dbin);
    if (selection_msg)
      gst_element_post_message ((GstElement *) slot->dbin, selection_msg);
  }
}

static GstPadProbeReturn
multiqueue_src_probe (GstPad * pad, GstPadProbeInfo * info,
    MultiQueueSlot * slot)
{
  GstPadProbeReturn ret = GST_PAD_PROBE_OK;
  GstDecodebin3 *dbin = slot->dbin;

  if (GST_IS_EVENT (GST_PAD_PROBE_INFO_DATA (info))) {
    GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);

    GST_DEBUG_OBJECT (pad, "Got event %p %s", ev, GST_EVENT_TYPE_NAME (ev));
    switch (GST_EVENT_TYPE (ev)) {
      case GST_EVENT_STREAM_START:
      {
        GstStream *stream = NULL;
        const GstStructure *s = gst_event_get_structure (ev);

        /* Drop STREAM_START events used to cleanup multiqueue */
        if (s
            && gst_structure_has_field (s,
                "decodebin3-flushing-stream-start")) {
          ret = GST_PAD_PROBE_HANDLED;
          gst_event_unref (ev);
          break;
        }

        gst_event_parse_stream (ev, &stream);
        if (stream == NULL) {
          GST_ERROR_OBJECT (pad,
              "Got a STREAM_START event without a GstStream");
          break;
        }
        slot->is_drained = FALSE;
        GST_DEBUG_OBJECT (pad, "Stream Start '%s'",
            gst_stream_get_stream_id (stream));
        if (slot->active_stream == NULL) {
          slot->active_stream = stream;
        } else if (slot->active_stream != stream) {
          gboolean stream_type_changed =
              gst_stream_get_stream_type (stream) !=
              gst_stream_get_stream_type (slot->active_stream);

          GST_DEBUG_OBJECT (pad, "Stream change (%s => %s) !",
              gst_stream_get_stream_id (slot->active_stream),
              gst_stream_get_stream_id (stream));
          gst_object_unref (slot->active_stream);
          slot->active_stream = stream;

          if (stream_type_changed) {
            /* The stream type has changed, we get rid of the current output. A
             * new one (targetting the new stream type) will be created once the
             * caps are received. */
            GST_DEBUG_OBJECT (pad,
                "Stream type change, discarding current output stream");
            if (slot->output) {
              DecodebinOutputStream *output = slot->output;
              SELECTION_LOCK (dbin);
              dbin->output_streams =
                  g_list_remove (dbin->output_streams, output);
              free_output_stream (dbin, output);
              SELECTION_UNLOCK (dbin);
            }
          }
        } else
          gst_object_unref (stream);
      }
        break;
      case GST_EVENT_CAPS:
      {
        /* Configure the output slot if needed */
        check_slot_reconfiguration (dbin, slot);
      }
        break;
      case GST_EVENT_EOS:
      {
        gboolean was_drained = slot->is_drained;
        slot->is_drained = TRUE;

        /* Custom EOS handling first */
        if (gst_mini_object_get_qdata (GST_MINI_OBJECT_CAST (ev),
                CUSTOM_EOS_QUARK)) {
          /* remove custom-eos */
          ev = gst_event_make_writable (ev);
          GST_PAD_PROBE_INFO_DATA (info) = ev;
          gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (ev),
              CUSTOM_EOS_QUARK, NULL, NULL);

          GST_LOG_OBJECT (pad, "Received custom EOS");
          ret = GST_PAD_PROBE_HANDLED;
          SELECTION_LOCK (dbin);
          if (slot->input == NULL) {
            GST_DEBUG_OBJECT (pad,
                "Got custom-eos from null input stream, remove output stream");
            /* Remove the output */
            if (slot->output) {
              DecodebinOutputStream *output = slot->output;
              dbin->output_streams =
                  g_list_remove (dbin->output_streams, output);
              free_output_stream (dbin, output);
              /* Reacalculate min interleave */
              gst_decodebin3_update_min_interleave (dbin);
            }
            slot->probe_id = 0;
            dbin->slots = g_list_remove (dbin->slots, slot);
            free_multiqueue_slot_async (dbin, slot);
            ret = GST_PAD_PROBE_REMOVE;
          } else if (!was_drained) {
            check_and_drain_multiqueue_locked (dbin, ev);
          }
          if (ret == GST_PAD_PROBE_HANDLED)
            gst_event_unref (ev);
          SELECTION_UNLOCK (dbin);
          break;
        }

        GST_FIXME_OBJECT (pad, "EOS on multiqueue source pad. input:%p",
            slot->input);
        if (slot->input == NULL) {
          GstPad *peer;
          GST_DEBUG_OBJECT (pad,
              "last EOS for input, forwarding and removing slot");
          peer = gst_pad_get_peer (pad);
          if (peer) {
            gst_pad_send_event (peer, gst_event_ref (ev));
            gst_object_unref (peer);
          }
          SELECTION_LOCK (dbin);
          /* FIXME : Shouldn't we try to re-assign the output instead of just
           * removing it ? */
          /* Remove the output */
          if (slot->output) {
            DecodebinOutputStream *output = slot->output;
            dbin->output_streams = g_list_remove (dbin->output_streams, output);
            free_output_stream (dbin, output);
          }
          slot->probe_id = 0;
          dbin->slots = g_list_remove (dbin->slots, slot);
          SELECTION_UNLOCK (dbin);

          /* FIXME: Removing the slot is async, which means actually
           * unlinking the pad is async. Other things like stream-start
           * might flow through this (now unprobed) link before it actually
           * gets released */
          free_multiqueue_slot_async (dbin, slot);
          ret = GST_PAD_PROBE_REMOVE;
        } else if (gst_mini_object_get_qdata (GST_MINI_OBJECT_CAST (ev),
                CUSTOM_FINAL_EOS_QUARK)) {
          GST_DEBUG_OBJECT (pad, "Got final eos, propagating downstream");
        } else {
          GST_DEBUG_OBJECT (pad, "Got regular eos (all_inputs_are_eos)");
          /* drop current event as eos will be sent in check_all_slot_for_eos
           * when all output streams are also eos */
          ret = GST_PAD_PROBE_DROP;
          SELECTION_LOCK (dbin);
          check_and_drain_multiqueue_locked (dbin, ev);
          SELECTION_UNLOCK (dbin);
        }
      }
        break;
      default:
        break;
    }
  } else if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
    GstQuery *query = GST_PAD_PROBE_INFO_QUERY (info);
    switch (GST_QUERY_TYPE (query)) {
      case GST_QUERY_CAPS:
      {
        GST_DEBUG_OBJECT (pad, "Intercepting CAPS query");
        gst_query_set_caps_result (query, GST_CAPS_ANY);
        ret = GST_PAD_PROBE_HANDLED;
      }
        break;

      case GST_QUERY_ACCEPT_CAPS:
      {
        GST_DEBUG_OBJECT (pad, "Intercepting Accept Caps query");
        /* If the current decoder doesn't accept caps, we'll reconfigure
         * on the actual caps event. So accept any caps. */
        gst_query_set_accept_caps_result (query, TRUE);
        ret = GST_PAD_PROBE_HANDLED;
      }
      default:
        break;
    }
  }

  return ret;
}

/* Create a new multiqueue slot for the given type
 *
 * It is up to the caller to know whether that slot is needed or not
 * (and release it when no longer needed) */
static MultiQueueSlot *
create_new_slot (GstDecodebin3 * dbin, GstStreamType type)
{
  MultiQueueSlot *slot;
  GstIterator *it = NULL;
  GValue item = { 0, };

  GST_DEBUG_OBJECT (dbin, "Creating new slot for type %s",
      gst_stream_type_get_name (type));
  slot = g_new0 (MultiQueueSlot, 1);
  slot->dbin = dbin;

  slot->id = dbin->slot_id++;

  slot->type = type;
  slot->sink_pad = gst_element_request_pad_simple (dbin->multiqueue, "sink_%u");
  if (slot->sink_pad == NULL)
    goto fail;

  it = gst_pad_iterate_internal_links (slot->sink_pad);
  if (!it || (gst_iterator_next (it, &item)) != GST_ITERATOR_OK
      || ((slot->src_pad = g_value_dup_object (&item)) == NULL)) {
    GST_ERROR ("Couldn't get srcpad from multiqueue for sink pad %s:%s",
        GST_DEBUG_PAD_NAME (slot->src_pad));
    goto fail;
  }
  gst_iterator_free (it);
  g_value_reset (&item);

  g_object_set (slot->sink_pad, "group-id", (guint) type, NULL);

  /* Add event probe */
  slot->probe_id =
      gst_pad_add_probe (slot->src_pad,
      GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM,
      (GstPadProbeCallback) multiqueue_src_probe, slot, NULL);

  GST_DEBUG ("Created new slot %u (%p) (%s:%s)", slot->id, slot,
      GST_DEBUG_PAD_NAME (slot->src_pad));

  dbin->slots = g_list_append (dbin->slots, slot);

  return slot;

  /* ERRORS */
fail:
  {
    if (slot->sink_pad)
      gst_element_release_request_pad (dbin->multiqueue, slot->sink_pad);
    g_free (slot);
    return NULL;
  }
}

/** gst_decodebin_get_slot_for_input_stream_locked:
 * @dbin: #GstDecodebin3
 * @input_stream: A #DecodebinInputstream
 *
 * Finds and returns the #MultiQueueslot for the given @input_stream. If needed
 * it will create a new one.
 *
 * Must be called with the SELECTION_LOCK taken
 */
static MultiQueueSlot *
gst_decodebin_get_slot_for_input_stream_locked (GstDecodebin3 * dbin,
    DecodebinInputStream * input_stream)
{
  GList *tmp;
  MultiQueueSlot *empty_slot = NULL;
  GstStreamType input_type = 0;
  gchar *stream_id = NULL;

  GST_DEBUG_OBJECT (dbin, "input %p (stream %p %s)",
      input_stream, input_stream->active_stream,
      input_stream->
      active_stream ? gst_stream_get_stream_id (input_stream->active_stream) :
      "");

  if (input_stream->active_stream) {
    input_type = gst_stream_get_stream_type (input_stream->active_stream);
    stream_id =
        (gchar *) gst_stream_get_stream_id (input_stream->active_stream);
  }

  /* Go over existing slots and check if there is already one for it */
  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    /* Already used input, return that one */
    if (slot->input == input_stream) {
      GST_DEBUG_OBJECT (dbin, "Returning already specified slot %d", slot->id);
      if (input_type && slot->type != input_type) {
        /* The input stream type has changed. It is the responsibility of the
         * user of decodebin3 to ensure that the inputs are coherent.
         *
         * The only case where the stream type will change is when switching
         * between sources which have non-intersecting stream types (ex:
         * switching from audio-only file to video-only file)
         *
         * NOTE : We need to change the slot type here, since it is notified as
         * soon as the *input* of the slot changes.
         */
        GST_DEBUG_OBJECT (dbin, "Changing multiqueue slot stream type");
        slot->type = input_type;
      }
      return slot;
    }
  }

  /* Go amongst all unused slots of the right type and try to find a candidate */
  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    if (slot->input == NULL && input_type == slot->type) {
      /* Remember this empty slot for later */
      empty_slot = slot;
      /* Check if available slot is of the same stream_id */
      GST_LOG_OBJECT (dbin, "Checking candidate slot %d (active_stream:%p)",
          slot->id, slot->active_stream);
      if (stream_id && slot->active_stream) {
        gchar *ostream_id =
            (gchar *) gst_stream_get_stream_id (slot->active_stream);
        GST_DEBUG_OBJECT (dbin, "Checking slot %d %s against %s", slot->id,
            ostream_id, stream_id);
        if (!g_strcmp0 (stream_id, ostream_id))
          break;
      }
    }
  }

  if (empty_slot) {
    GST_DEBUG_OBJECT (dbin, "Re-using existing unused slot %d", empty_slot->id);
    return empty_slot;
  }

  if (input_type)
    return create_new_slot (dbin, input_type);

  return NULL;
}

static void
link_input_to_slot (DecodebinInputStream * input, MultiQueueSlot * slot)
{
  if (slot->input != NULL && slot->input != input) {
    GST_ERROR_OBJECT (slot->dbin,
        "Trying to link input to an already used slot");
    return;
  }
  gst_pad_link_full (input->srcpad, slot->sink_pad, GST_PAD_LINK_CHECK_NOTHING);
  slot->pending_stream = input->active_stream;
  slot->input = input;
}

static GList *
create_decoder_factory_list (GstDecodebin3 * dbin, GstCaps * caps)
{
  GList *res;

  g_mutex_lock (&dbin->factories_lock);
  gst_decode_bin_update_factories_list (dbin);
  res = gst_element_factory_list_filter (dbin->decoder_factories,
      caps, GST_PAD_SINK, TRUE);
  g_mutex_unlock (&dbin->factories_lock);
  return res;
}

static GstPadProbeReturn
keyframe_waiter_probe (GstPad * pad, GstPadProbeInfo * info,
    DecodebinOutputStream * output)
{
  GstBuffer *buf = GST_PAD_PROBE_INFO_BUFFER (info);

  /* If we have a keyframe, remove the probe and let all data through */
  if (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT) ||
      GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_HEADER)) {
    GST_DEBUG_OBJECT (pad,
        "Buffer is keyframe or header, letting through and removing probe");
    output->drop_probe_id = 0;
    return GST_PAD_PROBE_REMOVE;
  }
  GST_DEBUG_OBJECT (pad, "Buffer is not a keyframe, dropping");
  return GST_PAD_PROBE_DROP;
}

static void
remove_decoder_link (DecodebinOutputStream * output, MultiQueueSlot * slot)
{
  GstDecodebin3 *dbin = output->dbin;

  if (gst_pad_is_linked (output->decoder_sink)) {
    gst_pad_unlink (slot->src_pad, output->decoder_sink);
  }
  if (output->drop_probe_id) {
    gst_pad_remove_probe (slot->src_pad, output->drop_probe_id);
    output->drop_probe_id = 0;
  }

  gst_element_set_locked_state (output->decoder, TRUE);
  gst_element_set_state (output->decoder, GST_STATE_NULL);

  gst_bin_remove ((GstBin *) dbin, output->decoder);
  output->decoder = NULL;
}

/* Returns FALSE if the output couldn't be properly configured and the
 * associated GstStreams should be disabled */
static gboolean
reconfigure_output_stream (DecodebinOutputStream * output,
    MultiQueueSlot * slot, GstMessage ** msg)
{
  GstDecodebin3 *dbin = output->dbin;
  GstCaps *new_caps = (GstCaps *) gst_stream_get_caps (slot->active_stream);
  gboolean needs_decoder;
  gboolean ret = TRUE;

  needs_decoder = gst_caps_can_intersect (new_caps, dbin->caps) != TRUE;

  GST_DEBUG_OBJECT (dbin,
      "Reconfiguring output %p to slot %p, needs_decoder:%d", output, slot,
      needs_decoder);

  /* FIXME : Maybe make the output un-hook itself automatically ? */
  if (output->slot != NULL && output->slot != slot) {
    GST_WARNING_OBJECT (dbin,
        "Output still linked to another slot (%p)", output->slot);
    gst_caps_unref (new_caps);
    return ret;
  }

  /* Check if existing config is reusable as-is by checking if
   * the existing decoder accepts the new caps, if not delete
   * it and create a new one */
  if (output->decoder) {
    gboolean can_reuse_decoder;

    if (needs_decoder) {
      can_reuse_decoder =
          gst_pad_query_accept_caps (output->decoder_sink, new_caps);
    } else
      can_reuse_decoder = FALSE;

    if (can_reuse_decoder) {
      if (output->type & GST_STREAM_TYPE_VIDEO && output->drop_probe_id == 0) {
        GST_DEBUG_OBJECT (dbin, "Adding keyframe-waiter probe");
        output->drop_probe_id =
            gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_BUFFER,
            (GstPadProbeCallback) keyframe_waiter_probe, output, NULL);
      }
      GST_DEBUG_OBJECT (dbin, "Reusing existing decoder for slot %p", slot);
      if (output->linked == FALSE) {
        gst_pad_link_full (slot->src_pad, output->decoder_sink,
            GST_PAD_LINK_CHECK_NOTHING);
        output->linked = TRUE;
      }
      gst_caps_unref (new_caps);
      return ret;
    }

    GST_DEBUG_OBJECT (dbin, "Removing old decoder for slot %p", slot);

    if (output->linked)
      gst_pad_unlink (slot->src_pad, output->decoder_sink);
    output->linked = FALSE;
    if (output->drop_probe_id) {
      gst_pad_remove_probe (slot->src_pad, output->drop_probe_id);
      output->drop_probe_id = 0;
    }

    if (!decode_pad_set_target ((GstGhostPad *) output->src_pad, NULL)) {
      GST_ERROR_OBJECT (dbin, "Could not release decoder pad");
      gst_caps_unref (new_caps);
      goto cleanup;
    }

    gst_element_set_locked_state (output->decoder, TRUE);
    gst_element_set_state (output->decoder, GST_STATE_NULL);

    gst_bin_remove ((GstBin *) dbin, output->decoder);
    output->decoder = NULL;
    output->decoder_latency = GST_CLOCK_TIME_NONE;
  } else if (output->linked) {
    /* Otherwise if we have no decoder yet but the output is linked make
     * sure that the ghost pad is really unlinked in case no decoder was
     * needed previously */
    if (!decode_pad_set_target ((GstGhostPad *) output->src_pad, NULL)) {
      GST_ERROR_OBJECT (dbin, "Could not release ghost pad");
      gst_caps_unref (new_caps);
      goto cleanup;
    }
  }

  gst_object_replace ((GstObject **) & output->decoder_sink, NULL);
  gst_object_replace ((GstObject **) & output->decoder_src, NULL);

  /* If a decoder is required, create one */
  if (needs_decoder) {
    GList *factories, *next_factory;

    factories = next_factory = create_decoder_factory_list (dbin, new_caps);
    if (!next_factory) {
      GST_DEBUG ("Could not find an element for caps %" GST_PTR_FORMAT,
          new_caps);
      g_assert (output->decoder == NULL);
      ret = FALSE;
      goto missing_decoder;
    }

    while (next_factory) {
      gboolean decoder_failed = FALSE;
      CandidateDecoder *candidate = NULL;

      /* If we don't have a decoder yet, instantiate one */
      output->decoder = gst_element_factory_create ((GstElementFactory *)
          next_factory->data, NULL);
      GST_DEBUG ("Trying decoder %" GST_PTR_FORMAT, output->decoder);

      if (output->decoder == NULL)
        goto try_next;

      if (!gst_bin_add ((GstBin *) dbin, output->decoder)) {
        GST_WARNING_OBJECT (dbin, "could not add decoder '%s' to pipeline",
            GST_ELEMENT_NAME (output->decoder));
        goto try_next;
      }
      output->decoder_sink =
          gst_element_get_static_pad (output->decoder, "sink");
      output->decoder_src = gst_element_get_static_pad (output->decoder, "src");
      if (output->type & GST_STREAM_TYPE_VIDEO) {
        GST_DEBUG_OBJECT (dbin, "Adding keyframe-waiter probe");
        output->drop_probe_id =
            gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_BUFFER,
            (GstPadProbeCallback) keyframe_waiter_probe, output, NULL);
      }

      candidate = add_candidate_decoder (dbin, output->decoder);
      if (gst_pad_link_full (slot->src_pad, output->decoder_sink,
              GST_PAD_LINK_CHECK_NOTHING) != GST_PAD_LINK_OK) {
        GST_WARNING_OBJECT (dbin, "could not link to %s:%s",
            GST_DEBUG_PAD_NAME (output->decoder_sink));
        decoder_failed = TRUE;
        goto try_next;
      }

      if (gst_element_set_state (output->decoder,
              GST_STATE_READY) == GST_STATE_CHANGE_FAILURE) {
        GST_WARNING_OBJECT (dbin,
            "Decoder '%s' failed to reach READY state",
            GST_ELEMENT_NAME (output->decoder));
        decoder_failed = TRUE;
        goto try_next;
      }

      if (!gst_pad_query_accept_caps (output->decoder_sink, new_caps)) {
        GST_DEBUG_OBJECT (dbin,
            "Decoder '%s' did not accept the caps, trying the next type",
            GST_ELEMENT_NAME (output->decoder));
        decoder_failed = TRUE;
        goto try_next;
      }

      if (gst_element_set_state (output->decoder,
              GST_STATE_PAUSED) == GST_STATE_CHANGE_FAILURE) {
        GST_WARNING_OBJECT (dbin, "Decoder '%s' failed to reach PAUSED state",
            GST_ELEMENT_NAME (output->decoder));
        decoder_failed = TRUE;
        goto try_next;
      } else {
        /* Everything went well */
        output->linked = TRUE;
        GST_DEBUG ("created decoder %" GST_PTR_FORMAT, output->decoder);

        handle_stored_latency_message (dbin, output, candidate);
        remove_candidate_decoder (dbin, candidate);
      }

      break;

    try_next:
      {
        if (decoder_failed)
          remove_decoder_link (output, slot);
        if (candidate)
          remove_candidate_decoder (dbin, candidate);

        if (!next_factory->next) {
          ret = FALSE;
          if (!decoder_failed)
            goto cleanup;
          if (output->decoder == NULL)
            goto missing_decoder;
        } else {
          next_factory = next_factory->next;
        }
      }
    }
    gst_plugin_feature_list_free (factories);
  } else {
    output->decoder_src = gst_object_ref (slot->src_pad);
    output->decoder_sink = NULL;
  }
  gst_caps_unref (new_caps);

  if (!decode_pad_set_target ((GstGhostPad *) output->src_pad,
          output->decoder_src)) {
    GST_ERROR_OBJECT (dbin, "Could not expose decoder pad");
    ret = FALSE;
    goto cleanup;
  }

  output->linked = TRUE;

  if (output->src_exposed == FALSE) {
    GstEvent *stream_start;

    stream_start = gst_pad_get_sticky_event (slot->src_pad,
        GST_EVENT_STREAM_START, 0);

    /* Ensure GstStream is accesiable from pad-added callback */
    if (stream_start) {
      gst_pad_store_sticky_event (output->src_pad, stream_start);
      gst_event_unref (stream_start);
    } else {
      GST_WARNING_OBJECT (slot->src_pad,
          "Pad has no stored stream-start event");
    }

    output->src_exposed = TRUE;
    gst_element_add_pad (GST_ELEMENT_CAST (dbin), output->src_pad);
  }

  if (output->decoder)
    gst_element_sync_state_with_parent (output->decoder);

  output->slot = slot;
  return ret;

missing_decoder:
  {
    GstCaps *caps;

    caps = gst_stream_get_caps (slot->active_stream);
    GST_DEBUG_OBJECT (slot->src_pad,
        "We are missing a decoder for %" GST_PTR_FORMAT, caps);
    *msg = gst_missing_decoder_message_new (GST_ELEMENT_CAST (dbin), caps);
    gst_caps_unref (caps);
  }

cleanup:
  {
    GST_DEBUG_OBJECT (dbin, "Cleanup");
    if (output->decoder_sink) {
      gst_object_unref (output->decoder_sink);
      output->decoder_sink = NULL;
    }
    if (output->decoder_src) {
      gst_object_unref (output->decoder_src);
      output->decoder_src = NULL;
    }
    if (output->decoder) {
      gst_element_set_state (output->decoder, GST_STATE_NULL);
      gst_bin_remove ((GstBin *) dbin, output->decoder);
      output->decoder = NULL;
    }
    return ret;
  }
}

static GstPadProbeReturn
idle_reconfigure (GstPad * pad, GstPadProbeInfo * info, MultiQueueSlot * slot)
{
  GstDecodebin3 *dbin = slot->dbin;

  check_slot_reconfiguration (dbin, slot);

  return GST_PAD_PROBE_REMOVE;
}

static MultiQueueSlot *
find_slot_for_stream_id (GstDecodebin3 * dbin, const gchar * sid)
{
  GList *tmp;

  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    const gchar *stream_id;
    if (slot->active_stream) {
      stream_id = gst_stream_get_stream_id (slot->active_stream);
      if (!g_strcmp0 (sid, stream_id))
        return slot;
    }
    if (slot->pending_stream && slot->pending_stream != slot->active_stream) {
      stream_id = gst_stream_get_stream_id (slot->pending_stream);
      if (!g_strcmp0 (sid, stream_id))
        return slot;
    }
  }

  return NULL;
}

/* This function handles the reassignment of a slot. Call this from
 * the streaming thread of a slot. */
static gboolean
reassign_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot)
{
  DecodebinOutputStream *output;
  MultiQueueSlot *target_slot = NULL;
  GList *tmp;
  const gchar *sid, *tsid;

  SELECTION_LOCK (dbin);
  output = slot->output;

  if (G_UNLIKELY (slot->active_stream == NULL)) {
    GST_DEBUG_OBJECT (slot->src_pad,
        "Called on inactive slot (active_stream == NULL)");
    SELECTION_UNLOCK (dbin);
    return FALSE;
  }

  if (G_UNLIKELY (output == NULL)) {
    GST_DEBUG_OBJECT (slot->src_pad,
        "Slot doesn't have any output to be removed");
    SELECTION_UNLOCK (dbin);
    return FALSE;
  }

  sid = gst_stream_get_stream_id (slot->active_stream);
  GST_DEBUG_OBJECT (slot->src_pad, "slot %s %p", sid, slot);

  /* Recheck whether this stream is still in the list of streams to deactivate */
  if (stream_in_list (dbin->requested_selection, sid)) {
    /* Stream is in the list of requested streams, don't remove */
    SELECTION_UNLOCK (dbin);
    GST_DEBUG_OBJECT (slot->src_pad,
        "Stream '%s' doesn't need to be deactivated", sid);
    return FALSE;
  }

  /* Unlink slot from output */
  /* FIXME : Handle flushing ? */
  /* FIXME : Handle outputs without decoders */
  GST_DEBUG_OBJECT (slot->src_pad, "Unlinking from decoder %p",
      output->decoder_sink);
  if (output->decoder_sink)
    gst_pad_unlink (slot->src_pad, output->decoder_sink);
  output->linked = FALSE;
  slot->output = NULL;
  output->slot = NULL;
  /* Remove sid from active selection */
  GST_DEBUG ("Removing '%s' from active_selection", sid);
  for (tmp = dbin->active_selection; tmp; tmp = tmp->next)
    if (!g_strcmp0 (sid, tmp->data)) {
      g_free (tmp->data);
      dbin->active_selection = g_list_delete_link (dbin->active_selection, tmp);
      break;
    }

  /* Can we re-assign this output to a requested stream ? */
  GST_DEBUG_OBJECT (slot->src_pad, "Attempting to re-assing output stream");
  for (tmp = dbin->to_activate; tmp; tmp = tmp->next) {
    MultiQueueSlot *tslot = find_slot_for_stream_id (dbin, tmp->data);
    GST_LOG_OBJECT (tslot->src_pad, "Checking slot %p (output:%p , stream:%s)",
        tslot, tslot->output, gst_stream_get_stream_id (tslot->active_stream));
    if (tslot && tslot->type == output->type && tslot->output == NULL) {
      GST_DEBUG_OBJECT (tslot->src_pad, "Using as reassigned slot");
      target_slot = tslot;
      tsid = tmp->data;
      /* Pass target stream id to requested selection */
      dbin->requested_selection =
          g_list_append (dbin->requested_selection, g_strdup (tmp->data));
      dbin->to_activate = g_list_delete_link (dbin->to_activate, tmp);
      break;
    }
  }

  if (target_slot) {
    GST_DEBUG_OBJECT (slot->src_pad, "Assigning output to slot %p '%s'",
        target_slot, tsid);
    target_slot->output = output;
    output->slot = target_slot;
    GST_DEBUG ("Adding '%s' to active_selection", tsid);
    dbin->active_selection =
        g_list_append (dbin->active_selection, (gchar *) g_strdup (tsid));
    SELECTION_UNLOCK (dbin);

    /* Wakeup the target slot so that it retries to send events/buffers
     * thereby triggering the output reconfiguration codepath */
    gst_pad_add_probe (target_slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
        (GstPadProbeCallback) idle_reconfigure, target_slot, NULL);
    /* gst_pad_send_event (target_slot->src_pad, gst_event_new_reconfigure ()); */
  } else {
    GstMessage *msg;

    dbin->output_streams = g_list_remove (dbin->output_streams, output);
    free_output_stream (dbin, output);
    msg = is_selection_done (slot->dbin);
    SELECTION_UNLOCK (dbin);

    if (msg)
      gst_element_post_message ((GstElement *) slot->dbin, msg);
  }

  return TRUE;
}

/* Idle probe called when a slot should be unassigned from its output stream.
 * This is needed to ensure nothing is flowing when unlinking the slot.
 *
 * Also, this method will search for a pending stream which could re-use
 * the output stream. */
static GstPadProbeReturn
slot_unassign_probe (GstPad * pad, GstPadProbeInfo * info,
    MultiQueueSlot * slot)
{
  GstDecodebin3 *dbin = slot->dbin;

  reassign_slot (dbin, slot);

  return GST_PAD_PROBE_REMOVE;
}

static gboolean
handle_stream_switch (GstDecodebin3 * dbin, GList * select_streams,
    guint32 seqnum)
{
  gboolean ret = TRUE;
  GList *tmp;
  /* List of slots to (de)activate. */
  GList *to_deactivate = NULL;
  GList *to_activate = NULL;
  /* List of unknown stream id, most likely means the event
   * should be sent upstream so that elements can expose the requested stream */
  GList *unknown = NULL;
  GList *to_reassign = NULL;
  GList *future_request_streams = NULL;
  GList *pending_streams = NULL;
  GList *slots_to_reassign = NULL;

  SELECTION_LOCK (dbin);
  if (G_UNLIKELY (seqnum != dbin->select_streams_seqnum)) {
    GST_DEBUG_OBJECT (dbin, "New SELECT_STREAMS has arrived in the meantime");
    SELECTION_UNLOCK (dbin);
    return TRUE;
  }
  /* Remove pending select_streams */
  dbin->pending_select_streams = FALSE;

  /* COMPARE the requested streams to the active and requested streams
   * on multiqueue. */

  /* First check the slots to activate and which ones are unknown */
  for (tmp = select_streams; tmp; tmp = tmp->next) {
    const gchar *sid = (const gchar *) tmp->data;
    MultiQueueSlot *slot;
    GST_DEBUG_OBJECT (dbin, "Checking for requested stream '%s'", sid);
    slot = find_slot_for_stream_id (dbin, sid);

    /* Find the multiqueue slot which is outputting, or will output, this stream
     * id */
    if (slot == NULL || slot->active_stream == NULL) {
      /* There is no slot on which this stream is active */
      if (stream_in_collection (dbin, (gchar *) sid)) {
        GST_DEBUG_OBJECT (dbin, "Adding to pending streams '%s'", sid);
        pending_streams = g_list_append (pending_streams, (gchar *) sid);
      } else {
        GST_DEBUG_OBJECT (dbin, "We don't have a slot for stream '%s'", sid);
        unknown = g_list_append (unknown, (gchar *) sid);
      }
    } else if (slot->output == NULL) {
      /* There is a slot on which this stream is active or pending */
      GST_DEBUG_OBJECT (dbin,
          "We need to activate slot %p %s:%s for stream '%s')", slot,
          GST_DEBUG_PAD_NAME (slot->src_pad), sid);
      to_activate = g_list_append (to_activate, slot);
    } else {
      GST_DEBUG_OBJECT (dbin,
          "Stream '%s' from slot %p is already active on output %p", sid, slot,
          slot->output);
      future_request_streams =
          g_list_append (future_request_streams, (gchar *) sid);
    }
  }

  for (tmp = dbin->slots; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    /* For slots that have an output, check if it's part of the streams to
     * be active */
    if (slot->output) {
      gboolean slot_to_deactivate = TRUE;

      if (slot->active_stream) {
        if (stream_in_list (select_streams,
                gst_stream_get_stream_id (slot->active_stream)))
          slot_to_deactivate = FALSE;
      }
      if (slot_to_deactivate && slot->pending_stream
          && slot->pending_stream != slot->active_stream) {
        if (stream_in_list (select_streams,
                gst_stream_get_stream_id (slot->pending_stream)))
          slot_to_deactivate = FALSE;
      }
      if (slot_to_deactivate) {
        GST_DEBUG_OBJECT (dbin,
            "Slot %p (%s) should be deactivated, no longer used", slot,
            slot->
            active_stream ? gst_stream_get_stream_id (slot->active_stream) :
            "NULL");
        to_deactivate = g_list_append (to_deactivate, slot);
      }
    }
  }

  if (to_deactivate != NULL) {
    GST_DEBUG_OBJECT (dbin, "Check if we can reassign slots");
    /* We need to compare what needs to be activated and deactivated in order
     * to determine whether there are outputs that can be transferred */
    /* Take the stream-id of the slots that are to be activated, for which there
     * is a slot of the same type that needs to be deactivated */
    tmp = to_deactivate;
    while (tmp) {
      MultiQueueSlot *slot_to_deactivate = (MultiQueueSlot *) tmp->data;
      gboolean removeit = FALSE;
      GList *tmp2, *next;
      GST_DEBUG_OBJECT (dbin,
          "Checking if slot to deactivate (%p) has a candidate slot to activate",
          slot_to_deactivate);
      for (tmp2 = to_activate; tmp2; tmp2 = tmp2->next) {
        MultiQueueSlot *slot_to_activate = (MultiQueueSlot *) tmp2->data;
        GST_DEBUG_OBJECT (dbin, "Comparing to slot %p", slot_to_activate);
        if (slot_to_activate->type == slot_to_deactivate->type) {
          GST_DEBUG_OBJECT (dbin, "Re-using");
          to_reassign = g_list_append (to_reassign, (gchar *)
              gst_stream_get_stream_id (slot_to_activate->active_stream));
          slots_to_reassign =
              g_list_append (slots_to_reassign, slot_to_deactivate);
          to_activate = g_list_remove (to_activate, slot_to_activate);
          removeit = TRUE;
          break;
        }
      }
      next = tmp->next;
      if (removeit)
        to_deactivate = g_list_delete_link (to_deactivate, tmp);
      tmp = next;
    }
  }

  for (tmp = to_deactivate; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    GST_DEBUG_OBJECT (dbin,
        "Really need to deactivate slot %p, but no available alternative",
        slot);

    slots_to_reassign = g_list_append (slots_to_reassign, slot);
  }

  /* The only slots left to activate are the ones that won't be reassigned and
   * therefore really need to have a new output created */
  for (tmp = to_activate; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    if (slot->active_stream)
      future_request_streams =
          g_list_append (future_request_streams,
          (gchar *) gst_stream_get_stream_id (slot->active_stream));
    else if (slot->pending_stream)
      future_request_streams =
          g_list_append (future_request_streams,
          (gchar *) gst_stream_get_stream_id (slot->pending_stream));
    else
      GST_ERROR_OBJECT (dbin, "No stream for slot %p !!", slot);
  }

  if (to_activate == NULL && pending_streams != NULL) {
    GST_DEBUG_OBJECT (dbin, "Stream switch requested for future collection");
    if (dbin->requested_selection)
      g_list_free_full (dbin->requested_selection, g_free);
    dbin->requested_selection =
        g_list_copy_deep (select_streams, (GCopyFunc) g_strdup, NULL);
    g_list_free (to_deactivate);
    g_list_free (pending_streams);
    to_deactivate = NULL;
    pending_streams = NULL;
  } else {
    if (dbin->requested_selection)
      g_list_free_full (dbin->requested_selection, g_free);
    dbin->requested_selection =
        g_list_copy_deep (future_request_streams, (GCopyFunc) g_strdup, NULL);
    dbin->requested_selection =
        g_list_concat (dbin->requested_selection,
        g_list_copy_deep (pending_streams, (GCopyFunc) g_strdup, NULL));
    if (dbin->to_activate)
      g_list_free (dbin->to_activate);
    dbin->to_activate = g_list_copy (to_reassign);
  }

  dbin->selection_updated = TRUE;
  SELECTION_UNLOCK (dbin);

  if (unknown) {
    GST_FIXME_OBJECT (dbin, "Got request for an unknown stream");
    g_list_free (unknown);
  }

  if (to_activate && !slots_to_reassign) {
    for (tmp = to_activate; tmp; tmp = tmp->next) {
      MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
      gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
          (GstPadProbeCallback) idle_reconfigure, slot, NULL);
    }
  }

  /* For all streams to deactivate, add an idle probe where we will do
   * the unassignment and switch over */
  for (tmp = slots_to_reassign; tmp; tmp = tmp->next) {
    MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
    gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
        (GstPadProbeCallback) slot_unassign_probe, slot, NULL);
  }

  if (to_deactivate)
    g_list_free (to_deactivate);
  if (to_activate)
    g_list_free (to_activate);
  if (to_reassign)
    g_list_free (to_reassign);
  if (future_request_streams)
    g_list_free (future_request_streams);
  if (pending_streams)
    g_list_free (pending_streams);
  if (slots_to_reassign)
    g_list_free (slots_to_reassign);

  return ret;
}

/*
 * * event : (transfer full): The select streams event
 *
 * Handles a GST_EVENT_SELECT_STREAMS (from application or downstream)
 *
 * Returns: TRUE if the event was handled, or FALSE if it should be forwarded to
 * the default handler.
 */
static gboolean
handle_select_streams (GstDecodebin3 * dbin, GstEvent * event)
{
  GList *streams = NULL;
  guint32 seqnum = gst_event_get_seqnum (event);

  if (dbin->upstream_handles_selection) {
    GST_DEBUG_OBJECT (dbin, "Letting select-streams event flow upstream");
    return FALSE;
  }

  SELECTION_LOCK (dbin);
  if (seqnum == dbin->select_streams_seqnum) {
    SELECTION_UNLOCK (dbin);
    GST_DEBUG_OBJECT (dbin,
        "Already handled/handling that SELECT_STREAMS event");
    gst_event_unref (event);
    return TRUE;
  }

  dbin->select_streams_seqnum = seqnum;
  if (dbin->pending_select_streams)
    GST_LOG_OBJECT (dbin, "Replacing pending select streams");
  dbin->pending_select_streams = TRUE;
  gst_event_parse_select_streams (event, &streams);
  SELECTION_UNLOCK (dbin);

  /* Finally handle the switch */
  if (streams) {
    handle_stream_switch (dbin, streams, seqnum);
    g_list_free_full (streams, g_free);
  }

  gst_event_unref (event);
  return TRUE;
}

static GstPadProbeReturn
ghost_pad_event_probe (GstPad * pad, GstPadProbeInfo * info,
    DecodebinOutputStream * output)
{
  GstPadProbeReturn ret = GST_PAD_PROBE_OK;
  GstDecodebin3 *dbin = output->dbin;
  GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);

  GST_DEBUG_OBJECT (pad, "Got event %p %s", event, GST_EVENT_TYPE_NAME (event));

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_SELECT_STREAMS:
    {
      if (handle_select_streams (dbin, event))
        ret = GST_PAD_PROBE_HANDLED;
    }
      break;
    default:
      break;
  }

  return ret;
}

static gboolean
gst_decodebin3_send_event (GstElement * element, GstEvent * event)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) element;

  GST_DEBUG_OBJECT (element, "event %s", GST_EVENT_TYPE_NAME (event));

  if (GST_EVENT_TYPE (event) == GST_EVENT_SELECT_STREAMS
      && handle_select_streams (dbin, event)) {
    return TRUE;
  }

  return GST_ELEMENT_CLASS (parent_class)->send_event (element, event);
}


static void
free_multiqueue_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot)
{
  if (slot->probe_id)
    gst_pad_remove_probe (slot->src_pad, slot->probe_id);
  if (slot->input) {
    if (slot->input->srcpad)
      gst_pad_unlink (slot->input->srcpad, slot->sink_pad);
  }

  gst_element_release_request_pad (dbin->multiqueue, slot->sink_pad);
  gst_object_replace ((GstObject **) & slot->sink_pad, NULL);
  gst_object_replace ((GstObject **) & slot->src_pad, NULL);
  gst_object_replace ((GstObject **) & slot->active_stream, NULL);
  g_free (slot);
}

static void
free_multiqueue_slot_async (GstDecodebin3 * dbin, MultiQueueSlot * slot)
{
  GST_LOG_OBJECT (dbin, "pushing multiqueue slot on thread pool to free");
  gst_element_call_async (GST_ELEMENT_CAST (dbin),
      (GstElementCallAsyncFunc) free_multiqueue_slot, slot, NULL);
}

/* Create a DecodebinOutputStream for a given type
 * Note: It will be empty initially, it needs to be configured
 * afterwards */
static DecodebinOutputStream *
create_output_stream (GstDecodebin3 * dbin, GstStreamType type)
{
  DecodebinOutputStream *res = g_new0 (DecodebinOutputStream, 1);
  gchar *pad_name;
  const gchar *prefix;
  GstStaticPadTemplate *templ;
  GstPadTemplate *ptmpl;
  guint32 *counter;
  GstPad *internal_pad;

  GST_DEBUG_OBJECT (dbin, "Created new output stream %p for type %s",
      res, gst_stream_type_get_name (type));

  res->type = type;
  res->dbin = dbin;
  res->decoder_latency = GST_CLOCK_TIME_NONE;

  if (type & GST_STREAM_TYPE_VIDEO) {
    templ = &video_src_template;
    counter = &dbin->vpadcount;
    prefix = "video";
  } else if (type & GST_STREAM_TYPE_AUDIO) {
    templ = &audio_src_template;
    counter = &dbin->apadcount;
    prefix = "audio";
  } else if (type & GST_STREAM_TYPE_TEXT) {
    templ = &text_src_template;
    counter = &dbin->tpadcount;
    prefix = "text";
  } else {
    templ = &src_template;
    counter = &dbin->opadcount;
    prefix = "src";
  }

  pad_name = g_strdup_printf ("%s_%u", prefix, *counter);
  *counter += 1;
  ptmpl = gst_static_pad_template_get (templ);
  res->src_pad = gst_ghost_pad_new_no_target_from_template (pad_name, ptmpl);
  gst_object_unref (ptmpl);
  g_free (pad_name);
  gst_pad_set_active (res->src_pad, TRUE);
  /* Put an event probe on the internal proxy pad to detect upstream
   * events */
  internal_pad =
      (GstPad *) gst_proxy_pad_get_internal ((GstProxyPad *) res->src_pad);
  gst_pad_add_probe (internal_pad, GST_PAD_PROBE_TYPE_EVENT_UPSTREAM,
      (GstPadProbeCallback) ghost_pad_event_probe, res, NULL);
  gst_object_unref (internal_pad);

  dbin->output_streams = g_list_append (dbin->output_streams, res);

  return res;
}

static void
free_output_stream (GstDecodebin3 * dbin, DecodebinOutputStream * output)
{
  if (output->slot) {
    if (output->decoder_sink && output->decoder)
      gst_pad_unlink (output->slot->src_pad, output->decoder_sink);

    output->slot->output = NULL;
    output->slot = NULL;
  }
  gst_object_replace ((GstObject **) & output->decoder_sink, NULL);
  decode_pad_set_target ((GstGhostPad *) output->src_pad, NULL);
  gst_object_replace ((GstObject **) & output->decoder_src, NULL);
  if (output->src_exposed) {
    gst_element_remove_pad ((GstElement *) dbin, output->src_pad);
  }
  if (output->decoder) {
    gst_element_set_locked_state (output->decoder, TRUE);
    gst_element_set_state (output->decoder, GST_STATE_NULL);
    gst_bin_remove ((GstBin *) dbin, output->decoder);
  }
  g_free (output);
}

static GstStateChangeReturn
gst_decodebin3_change_state (GstElement * element, GstStateChange transition)
{
  GstDecodebin3 *dbin = (GstDecodebin3 *) element;
  GstStateChangeReturn ret;

  /* Upwards */
  switch (transition) {
    default:
      break;
  }
  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
  if (ret == GST_STATE_CHANGE_FAILURE)
    goto beach;

  switch (transition) {
    case GST_STATE_CHANGE_PAUSED_TO_READY:
      gst_decodebin3_reset (dbin);
      break;
    default:
      break;
  }
beach:
  return ret;
}