/* 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 "gstplayback.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_STREAM 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 auxilliary 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.

 *

 * <emphasis>decodebin3 is still experimental API and a technology preview.

 * Its behaviour and exposed API is subject to change.</emphasis>

 *

 */

/**

 * Global design

 *

 * 1) From sink pad to elementary streams (GstParseBin)

 *

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

}

typedef struct _GstDecodebin3 GstDecodebin3;

typedef struct _GstDecodebin3Class GstDecodebin3Class;

typedef struct _DecodebinInputStream DecodebinInputStream;

typedef struct _DecodebinInput DecodebinInput;

typedef struct _DecodebinOutputStream DecodebinOutputStream;

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) */

  /* FIXME : Needs to be resetted appropriately (when upstream changes ?) */

  guint32 current_group_id;

  /* End of variables protected by input_lock */

  GstElement *multiqueue;

  /* 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;

  /* pending list of streams to select (from downstream) */

  GList *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 */

  /* List of pending collections.

   * FIXME : Is this really needed ? */

  GList *pending_collection;

  /* 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;

};

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 */

  guint group_id;

  GstElement *parsebin;

  gulong pad_added_sigid;

  gulong pad_removed_sigid;

  gulong drained_sigid;

  /* TRUE if the input got drained

   * FIXME : When do we reset it if re-used ?

   */

  gboolean drained;

  /* HACK : Remove these fields */

  /* List of PendingPad structures */

  GList *pending_pads;

};

/* 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;

  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;

  /* keyframe dropping probe */

  gulong drop_probe_id;

};

/* Pending pads from parsebin */

typedef struct _PendingPad

{

  GstDecodebin3 *dbin;

  DecodebinInput *input;

  GstPad *pad;

  gulong buffer_probe;

  gulong event_probe;

  gboolean saw_eos;

} PendingPad;

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

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

#if 0

static gboolean have_factory (GstDecodebin3 * dbin, GstCaps * caps,

    GstElementFactoryListType ftype);

#endif

static void free_input (GstDecodebin3 * dbin, DecodebinInput * input);

static void free_input_async (GstDecodebin3 * dbin, DecodebinInput * input);

static DecodebinInput *create_new_input (GstDecodebin3 * dbin, gboolean main);

static gboolean set_input_group_id (DecodebinInput * input, guint32 * group_id);

static void reconfigure_output_stream (DecodebinOutputStream * output,

    MultiQueueSlot * slot);

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 *get_slot_for_input (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);

/* FIXME: Really make all the parser stuff a self-contained helper object */

#include "gstdecodebin3-parse.c"

static gboolean

_gst_int_accumulator (GSignalInvocationHint * ihint,

    GValue * return_accu, const GValue * handler_return, gpointer dummy)

{

  gint res = g_value_get_int (handler_return);

  if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))

    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->set_property = gst_decodebin3_set_property;

  gobject_klass->get_property = gst_decodebin3_get_property;

  /* FIXME : ADD PROPERTIES ! */

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

  /* FIXME : ADD SIGNALS ! */

  /**

   * 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.

   *

   * 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, g_cclosure_marshal_generic,

      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 anothe 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, g_cclosure_marshal_generic, 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);

  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 = create_new_input (dbin, TRUE);

  dbin->multiqueue = gst_element_factory_make ("multiqueue", NULL);

  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_dispose (GObject * object)

{

  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  GList *walk, *next;

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

  g_list_free_full (dbin->requested_selection, g_free);

  g_list_free (dbin->active_selection);

  g_list_free (dbin->to_activate);

  g_list_free (dbin->pending_select_streams);

  g_clear_object (&dbin->collection);

  free_input (dbin, dbin->main_input);

  for (walk = dbin->other_inputs; walk; walk = next) {

    DecodebinInput *input = walk->data;

    next = g_list_next (walk);

    free_input (dbin, input);

    dbin->other_inputs = g_list_delete_link (dbin->other_inputs, walk);

  }

  G_OBJECT_CLASS (parent_class)->dispose (object);

}

static void

gst_decodebin3_set_property (GObject * object, guint prop_id,

    const GValue * value, GParamSpec * pspec)

{

  GstDecodebin3 *dbin = (GstDecodebin3 *) object;

  /* FIXME : IMPLEMENT */

  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;

  /* FIXME : IMPLEMENT */

  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;

  }

}

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;

}

/* This method should be called whenever a STREAM_START event

 * comes out of a given parsebin.

 * The caller shall replace the group_id if the function returns TRUE */

static gboolean

set_input_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) {

    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;

    }

    *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_WARNING_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_WARNING_OBJECT (dbin, "All inputs drained. Posting about-to-finish");

    g_signal_emit (dbin, gst_decodebin3_signals[SIGNAL_ABOUT_TO_FINISH], 0,

        NULL);

  }

}

/* Call with INPUT_LOCK taken */

static gboolean

ensure_input_parsebin (GstDecodebin3 * dbin, DecodebinInput * input)

{

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

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

  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;

  GstPadLinkReturn res = GST_PAD_LINK_OK;

  DecodebinInput *input;

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

      ". Creating parsebin if needed", pad);

  if ((input = g_object_get_data (G_OBJECT (pad), "decodebin.input")) == NULL)

    goto fail;

  INPUT_LOCK (dbin);

  if (!ensure_input_parsebin (dbin, input))

    res = GST_PAD_LINK_REFUSED;

  INPUT_UNLOCK (dbin);

  return res;

fail:

  GST_ERROR_OBJECT (parent, "Failed to retrieve input state from ghost pad");

  return GST_PAD_LINK_REFUSED;

}

/* 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;

}

static void

gst_decodebin3_input_pad_unlink (GstPad * pad, GstObject * parent)

{

  GstDecodebin3 *dbin = (GstDecodebin3 *) parent;

  DecodebinInput *input;

  GST_LOG_OBJECT (parent, "Got unlink on input pad %" GST_PTR_FORMAT

      ". Removing parsebin.", pad);

  if ((input = g_object_get_data (G_OBJECT (pad), "decodebin.input")) == NULL)

    goto fail;

  INPUT_LOCK (dbin);

  if (input->parsebin == NULL) {

    INPUT_UNLOCK (dbin);

    return;

  }

  if (GST_OBJECT_PARENT (GST_OBJECT (input->parsebin)) == GST_OBJECT (dbin)) {

    GstStreamCollection *collection = NULL;