gst/playback/gstdecodebin3.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) <2015> Centricular Ltd
   4  *  @author: Edward Hervey <edward@centricular.com>
   5  *  @author: Jan Schmidt <jan@centricular.com>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26
  27 #include <glib.h>
  28 #include <glib-object.h>
  29 #include <glib/gprintf.h>
  30 #include <gst/gst.h>
  31 #include <gst/pbutils/pbutils.h>
  32
  33 #include "gstplayback.h"
  34 #include "gstplay-enum.h"
  35 #include "gstrawcaps.h"
  36
  37
  38 /**
  39  * Global design
  40  *
  41  * 1) From sink pad to elementary streams (GstParseBin)
  42  *
  43  * The input sink pads are fed to GstParseBin. GstParseBin will feed them
  44  * through typefind. When the caps are detected (or changed) we recursively
  45  * figure out which demuxer, parser or depayloader is needed until we get to
  46  * elementary streams.
  47  *
  48  * All elementary streams (whether decoded or not, whether exposed or not) are
  49  * fed through multiqueue. There is only *one* multiqueue in decodebin3.
  50  *
  51  * => MultiQueue is the cornerstone.
  52  * => No buffering before multiqueue
  53  *
  54  * 2) Elementary streams
  55  *
  56  * After GstParseBin, there are 3 main components:
  57  *  1) Input Streams (provided by GstParseBin)
  58  *  2) Multiqueue slots
  59  *  3) Output Streams
  60  *
  61  * Input Streams correspond to the stream coming from GstParseBin and that gets
  62  * fed into a multiqueue slot.
  63  *
  64  * Output Streams correspond to the combination of a (optional) decoder and an
  65  * output ghostpad. Output Streams can be moved from one multiqueue slot to
  66  * another, can reconfigure itself (different decoders), and can be
  67  * added/removed depending on the configuration (all streams outputted, only one
  68  * of each type, ...).
  69  *
  70  * Multiqueue slots correspond to a pair of sink/src pad from multiqueue. For
  71  * each 'active' Input Stream there is a corresponding slot.
  72  * Slots might have different streams on input and output (due to internal
  73  * buffering).
  74  *
  75  * Due to internal queuing/buffering/..., all those components (might) behave
  76  * asynchronously. Therefore probes will be used on each component source pad to
  77  * detect various key-points:
  78  *  * EOS :
  79  *     the stream is done => Mark that component as done, optionally freeing/removing it
  80  *  * STREAM_START :
  81  *     a new stream is starting => link it further if needed
  82  *
  83  *
  84  * 3) Gradual replacement
  85  *
  86  * If the caps change at any point in decodebin (input sink pad, demuxer output,
  87  * multiqueue output, ..), we gradually replace (if needed) the following elements.
  88  *
  89  * This is handled by the probes in various locations:
  90  *  a) typefind output
  91  *  b) multiqueue input (source pad of Input Streams)
  92  *  c) multiqueue output (source pad of Multiqueue Slots)
  93  *  d) final output (target of source ghostpads)
  94  *
  95  * When CAPS event arrive at those points, one of three things can happen:
  96  * a) There is no elements downstream yet, just create/link-to following elements
  97  * b) There are downstream elements, do a ACCEPT_CAPS query
  98  *  b.1) The new CAPS are accepted, keep current configuration
  99  *  b.2) The new CAPS are not accepted, remove following elements then do a)
 100  *
 101  *
 102  *
 103  *    Components:
 104  *
 105  *                                                   MultiQ     Output
 106  *                     Input(s)                      Slots      Streams
 107  *  /-------------------------------------------\   /-----\  /------------- \
 108  *
 109  * +-------------------------------------------------------------------------+
 110  * |                                                                         |
 111  * | +---------------------------------------------+                         |
 112  * | |   GstParseBin(s)                            |                         |
 113  * | |                +--------------+             |  +-----+                |
 114  * | |                |              |---[parser]-[|--| Mul |---[ decoder ]-[|
 115  * |]--[ typefind ]---|  demuxer(s)  |------------[|  | ti  |                |
 116  * | |                |  (if needed) |---[parser]-[|--| qu  |                |
 117  * | |                |              |---[parser]-[|--| eu  |---[ decoder ]-[|
 118  * | |                +--------------+             |  +------             ^  |
 119  * | +---------------------------------------------+        ^             |  |
 120  * |                                               ^        |             |  |
 121  * +-----------------------------------------------+--------+-------------+--+
 122  *                                                 |        |             |
 123  *                                                 |        |             |
 124  *                                       Probes  --/--------/-------------/
 125  *
 126  * ATOMIC SWITCHING
 127  *
 128  * We want to ensure we re-use decoders when switching streams. This takes place
 129  * at the multiqueue output level.
 130  *
 131  * MAIN CONCEPTS
 132  *  1) Activating a stream (i.e. linking a slot to an output) is only done within
 133  *    the streaming thread in the multiqueue_src_probe() and only if the
 134       stream is in the REQUESTED selection.
 135  *  2) Deactivating a stream (i.e. unlinking a slot from an output) is also done
 136  *    within the stream thread, but only in a purposefully called IDLE probe
 137  *    that calls reassign_slot().
 138  *
 139  * Based on those two principles, 3 "selection" of streams (stream-id) are used:
 140  * 1) requested_selection
 141  *    All streams within that list should be activated
 142  * 2) active_selection
 143  *    List of streams that are exposed by decodebin
 144  * 3) to_activate
 145  *    List of streams that will be moved to requested_selection in the
 146  *    reassign_slot() method (i.e. once a stream was deactivated, and the output
 147  *    was retargetted)
 148  */
 149
 150
 151 GST_DEBUG_CATEGORY_STATIC (decodebin3_debug);
 152 #define GST_CAT_DEFAULT decodebin3_debug
 153
 154 #define GST_TYPE_DECODEBIN3      (gst_decodebin3_get_type ())
 155
 156 #define EXTRA_DEBUG 1
 157
 158 typedef struct _GstDecodebin3 GstDecodebin3;
 159 typedef struct _GstDecodebin3Class GstDecodebin3Class;
 160
 161 typedef struct _DecodebinInputStream DecodebinInputStream;
 162 typedef struct _DecodebinInput DecodebinInput;
 163 typedef struct _DecodebinOutputStream DecodebinOutputStream;
 164
 165 struct _GstDecodebin3
 166 {
 167   GstBin bin;
 168
 169   /* input_lock protects the following variables */
 170   GMutex input_lock;
 171   /* Main input (static sink pad) */
 172   DecodebinInput *main_input;
 173   /* Supplementary input (request sink pads) */
 174   GList *other_inputs;
 175   /* counter for input */
 176   guint32 input_counter;
 177   /* Current stream group_id (default : G_MAXUINT32) */
 178   /* FIXME : Needs to be resetted appropriately (when upstream changes ?) */
 179   guint32 current_group_id;
 180   /* End of variables protected by input_lock */
 181
 182   GstElement *multiqueue;
 183
 184   /* FIXME : Mutex for protecting values below */
 185   GstStreamCollection *collection;      /* Active collection */
 186
 187   GList *input_streams;         /* List of DecodebinInputStream for active collection */
 188   GList *output_streams;        /* List of DecodebinOutputStream used for output */
 189   GList *slots;                 /* List of MultiQueueSlot */
 190   guint slot_id;
 191
 192   /* selection_lock protects access to following variables */
 193   GMutex selection_lock;
 194   /* requested selection of stream-id to activate post-multiqueue */
 195   GList *requested_selection;
 196   /* list of stream-id currently activated in output */
 197   GList *active_selection;
 198   /* List of stream-id that need to be activated (after a stream switch for ex) */
 199   GList *to_activate;
 200   /* Pending select streams event */
 201   guint32 select_streams_seqnum;
 202   /* pending list of streams to select (from downstream) */
 203   GList *pending_select_streams;
 204   /* TRUE if requested_selection was updated, will become FALSE once
 205    * it has fully transitioned to active */
 206   gboolean selection_updated;
 207   /* End of variables protected by selection_lock */
 208
 209   /* List of pending collections.
 210    * FIXME : Is this really needed ? */
 211   GList *pending_collection;
 212
 213
 214   /* Factories */
 215   GMutex factories_lock;
 216   guint32 factories_cookie;
 217   /* All DECODABLE factories */
 218   GList *factories;
 219   /* Only DECODER factories */
 220   GList *decoder_factories;
 221   /* DECODABLE but not DECODER factories */
 222   GList *decodable_factories;
 223
 224   /* counters for pads */
 225   guint32 apadcount, vpadcount, tpadcount, opadcount;
 226
 227   /* Properties */
 228   GstCaps *caps;
 229 };
 230
 231 struct _GstDecodebin3Class
 232 {
 233   GstBinClass class;
 234
 235     gint (*select_stream) (GstDecodebin3 * dbin,
 236       GstStreamCollection * collection, GstStream * stream);
 237 };
 238
 239 /* Input of decodebin, controls input pad and parsebin */
 240 struct _DecodebinInput
 241 {
 242   GstDecodebin3 *dbin;
 243
 244   gboolean is_main;
 245
 246   GstPad *ghost_sink;
 247   GstPad *parsebin_sink;
 248
 249   GstStreamCollection *collection;      /* Active collection */
 250
 251   guint group_id;
 252
 253   GstElement *parsebin;
 254
 255   gulong pad_added_sigid;
 256   gulong pad_removed_sigid;
 257
 258   /* HACK : Remove these fields */
 259   /* List of PendingPad structures */
 260   GList *pending_pads;
 261 };
 262
 263 /* Multiqueue Slots */
 264 typedef struct _MultiQueueSlot
 265 {
 266   guint id;
 267
 268   GstDecodebin3 *dbin;
 269   /* Type of stream handled by this slot */
 270   GstStreamType type;
 271
 272   /* Linked input and output */
 273   DecodebinInputStream *input;
 274
 275   /* pending => last stream received on sink pad */
 276   GstStream *pending_stream;
 277   /* active => last stream outputted on source pad */
 278   GstStream *active_stream;
 279
 280   GstPad *sink_pad, *src_pad;
 281
 282   /* id of the MQ src_pad event probe */
 283   gulong probe_id;
 284
 285   gboolean drain_eos;
 286
 287   DecodebinOutputStream *output;
 288 } MultiQueueSlot;
 289
 290 /* Streams that are exposed downstream (i.e. output) */
 291 struct _DecodebinOutputStream
 292 {
 293   GstDecodebin3 *dbin;
 294   /* The type of stream handled by this output stream */
 295   GstStreamType type;
 296
 297   /* The slot to which this output stream is currently connected to */
 298   MultiQueueSlot *slot;
 299
 300   GstElement *decoder;          /* Optional */
 301   GstPad *decoder_sink, *decoder_src;
 302   gboolean linked;
 303
 304   /* ghostpad */
 305   GstPad *src_pad;
 306   /* Flag if ghost pad is exposed */
 307   gboolean src_exposed;
 308
 309   /* keyframe dropping probe */
 310   gulong drop_probe_id;
 311 };
 312
 313 /* Pending pads from parsebin */
 314 typedef struct _PendingPad
 315 {
 316   GstDecodebin3 *dbin;
 317   DecodebinInput *input;
 318   GstPad *pad;
 319
 320   gulong buffer_probe;
 321   gulong event_probe;
 322   gboolean saw_eos;
 323 } PendingPad;
 324
 325 /* properties */
 326 #define DEFAULT_CAPS (gst_static_caps_get (&default_raw_caps))
 327
 328 enum
 329 {
 330   PROP_0,
 331   PROP_CAPS
 332 };
 333
 334 /* signals */
 335 enum
 336 {
 337   SIGNAL_SELECT_STREAM,
 338   LAST_SIGNAL
 339 };
 340 static guint gst_decodebin3_signals[LAST_SIGNAL] = { 0 };
 341
 342 #define SELECTION_LOCK(dbin) G_STMT_START {                             \
 343     GST_LOG_OBJECT (dbin,                                               \
 344                     "selection locking from thread %p",                 \
 345                     g_thread_self ());                                  \
 346     g_mutex_lock (&dbin->selection_lock);                               \
 347     GST_LOG_OBJECT (dbin,                                               \
 348                     "selection locked from thread %p",                  \
 349                     g_thread_self ());                                  \
 350   } G_STMT_END
 351
 352 #define SELECTION_UNLOCK(dbin) G_STMT_START {                           \
 353     GST_LOG_OBJECT (dbin,                                               \
 354                     "selection unlocking from thread %p",               \
 355                     g_thread_self ());                                  \
 356     g_mutex_unlock (&dbin->selection_lock);                             \
 357   } G_STMT_END
 358
 359 #define INPUT_LOCK(dbin) G_STMT_START {                         \
 360     GST_LOG_OBJECT (dbin,                                               \
 361                     "input locking from thread %p",                     \
 362                     g_thread_self ());                                  \
 363     g_mutex_lock (&dbin->input_lock);                           \
 364     GST_LOG_OBJECT (dbin,                                               \
 365                     "input locked from thread %p",                      \
 366                     g_thread_self ());                                  \
 367   } G_STMT_END
 368
 369 #define INPUT_UNLOCK(dbin) G_STMT_START {                               \
 370     GST_LOG_OBJECT (dbin,                                               \
 371                     "input unlocking from thread %p",           \
 372                     g_thread_self ());                                  \
 373     g_mutex_unlock (&dbin->input_lock);                         \
 374   } G_STMT_END
 375
 376 GType gst_decodebin3_get_type (void);
 377 #define gst_decodebin3_parent_class parent_class
 378 G_DEFINE_TYPE (GstDecodebin3, gst_decodebin3, GST_TYPE_BIN);
 379
 380 static GstStaticCaps default_raw_caps = GST_STATIC_CAPS (DEFAULT_RAW_CAPS);
 381
 382 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
 383     GST_PAD_SINK,
 384     GST_PAD_ALWAYS,
 385     GST_STATIC_CAPS_ANY);
 386
 387 static GstStaticPadTemplate request_sink_template =
 388 GST_STATIC_PAD_TEMPLATE ("sink_%u",
 389     GST_PAD_SINK,
 390     GST_PAD_REQUEST,
 391     GST_STATIC_CAPS_ANY);
 392
 393 static GstStaticPadTemplate video_src_template =
 394 GST_STATIC_PAD_TEMPLATE ("video_%u",
 395     GST_PAD_SRC,
 396     GST_PAD_SOMETIMES,
 397     GST_STATIC_CAPS_ANY);
 398
 399 static GstStaticPadTemplate audio_src_template =
 400 GST_STATIC_PAD_TEMPLATE ("audio_%u",
 401     GST_PAD_SRC,
 402     GST_PAD_SOMETIMES,
 403     GST_STATIC_CAPS_ANY);
 404
 405 static GstStaticPadTemplate text_src_template =
 406 GST_STATIC_PAD_TEMPLATE ("text_%u",
 407     GST_PAD_SRC,
 408     GST_PAD_SOMETIMES,
 409     GST_STATIC_CAPS_ANY);
 410
 411 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src_%u",
 412     GST_PAD_SRC,
 413     GST_PAD_SOMETIMES,
 414     GST_STATIC_CAPS_ANY);
 415
 416
 417 static void gst_decodebin3_dispose (GObject * object);
 418 static void gst_decodebin3_set_property (GObject * object, guint prop_id,
 419     const GValue * value, GParamSpec * pspec);
 420 static void gst_decodebin3_get_property (GObject * object, guint prop_id,
 421     GValue * value, GParamSpec * pspec);
 422
 423 static gboolean parsebin_autoplug_continue_cb (GstElement *
 424     parsebin, GstPad * pad, GstCaps * caps, GstDecodebin3 * dbin);
 425
 426 static gint
 427 gst_decodebin3_select_stream (GstDecodebin3 * dbin,
 428     GstStreamCollection * collection, GstStream * stream)
 429 {
 430   GST_LOG_OBJECT (dbin, "default select-stream, returning -1");
 431
 432   return -1;
 433 }
 434
 435 static GstPad *gst_decodebin3_request_new_pad (GstElement * element,
 436     GstPadTemplate * temp, const gchar * name, const GstCaps * caps);
 437 static void gst_decodebin3_handle_message (GstBin * bin, GstMessage * message);
 438 static GstStateChangeReturn gst_decodebin3_change_state (GstElement * element,
 439     GstStateChange transition);
 440 static gboolean gst_decodebin3_send_event (GstElement * element,
 441     GstEvent * event);
 442
 443 static void gst_decode_bin_update_factories_list (GstDecodebin3 * dbin);
 444 #if 0
 445 static gboolean have_factory (GstDecodebin3 * dbin, GstCaps * caps,
 446     GstElementFactoryListType ftype);
 447 #endif
 448
 449 static void free_input (GstDecodebin3 * dbin, DecodebinInput * input);
 450 static DecodebinInput *create_new_input (GstDecodebin3 * dbin, gboolean main);
 451 static gboolean set_input_group_id (DecodebinInput * input, guint32 * group_id);
 452
 453 static void reconfigure_output_stream (DecodebinOutputStream * output,
 454     MultiQueueSlot * slot);
 455 static void free_output_stream (GstDecodebin3 * dbin,
 456     DecodebinOutputStream * output);
 457 static DecodebinOutputStream *create_output_stream (GstDecodebin3 * dbin,
 458     GstStreamType type);
 459
 460 static GstPadProbeReturn slot_unassign_probe (GstPad * pad,
 461     GstPadProbeInfo * info, MultiQueueSlot * slot);
 462 static gboolean reassign_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot);
 463 static MultiQueueSlot *get_slot_for_input (GstDecodebin3 * dbin,
 464     DecodebinInputStream * input);
 465 static void link_input_to_slot (DecodebinInputStream * input,
 466     MultiQueueSlot * slot);
 467 static void free_multiqueue_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot);
 468
 469 /* FIXME: Really make all the parser stuff a self-contained helper object */
 470 #include "gstdecodebin3-parse.c"
 471
 472 static gboolean
 473 _gst_int_accumulator (GSignalInvocationHint * ihint,
 474     GValue * return_accu, const GValue * handler_return, gpointer dummy)
 475 {
 476   gint res = g_value_get_int (handler_return);
 477
 478   if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))
 479     g_value_set_int (return_accu, res);
 480
 481   if (res == -1)
 482     return TRUE;
 483
 484   return FALSE;
 485 }
 486
 487 static void
 488 gst_decodebin3_class_init (GstDecodebin3Class * klass)
 489 {
 490   GObjectClass *gobject_klass = (GObjectClass *) klass;
 491   GstElementClass *element_class = (GstElementClass *) klass;
 492   GstBinClass *bin_klass = (GstBinClass *) klass;
 493
 494   gobject_klass->dispose = gst_decodebin3_dispose;
 495   gobject_klass->set_property = gst_decodebin3_set_property;
 496   gobject_klass->get_property = gst_decodebin3_get_property;
 497
 498   /* FIXME : ADD PROPERTIES ! */
 499   g_object_class_install_property (gobject_klass, PROP_CAPS,
 500       g_param_spec_boxed ("caps", "Caps",
 501           "The caps on which to stop decoding. (NULL = default)",
 502           GST_TYPE_CAPS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 503
 504   /* FIXME : ADD SIGNALS ! */
 505   /**
 506    * GstDecodebin3::select-stream
 507    * @decodebin: a #GstDecodebin3
 508    * @collection: a #GstStreamCollection
 509    * @stream: a #GstStream
 510    *
 511    * This signal is emitted whenever @decodebin needs to decide whether
 512    * to expose a @stream of a given @collection.
 513    *
 514    * Returns: 1 if the stream should be selected, 0 if it shouldn't be selected.
 515    * A value of -1 (default) lets @decodebin decide what to do with the stream.
 516    * */
 517   gst_decodebin3_signals[SIGNAL_SELECT_STREAM] =
 518       g_signal_new ("select-stream", G_TYPE_FROM_CLASS (klass),
 519       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstDecodebin3Class, select_stream),
 520       _gst_int_accumulator, NULL, g_cclosure_marshal_generic,
 521       G_TYPE_INT, 2, GST_TYPE_STREAM_COLLECTION, GST_TYPE_STREAM);
 522
 523
 524   element_class->request_new_pad =
 525       GST_DEBUG_FUNCPTR (gst_decodebin3_request_new_pad);
 526   element_class->change_state = GST_DEBUG_FUNCPTR (gst_decodebin3_change_state);
 527   element_class->send_event = GST_DEBUG_FUNCPTR (gst_decodebin3_send_event);
 528
 529   gst_element_class_add_pad_template (element_class,
 530       gst_static_pad_template_get (&sink_template));
 531   gst_element_class_add_pad_template (element_class,
 532       gst_static_pad_template_get (&request_sink_template));
 533   gst_element_class_add_pad_template (element_class,
 534       gst_static_pad_template_get (&video_src_template));
 535   gst_element_class_add_pad_template (element_class,
 536       gst_static_pad_template_get (&audio_src_template));
 537   gst_element_class_add_pad_template (element_class,
 538       gst_static_pad_template_get (&text_src_template));
 539   gst_element_class_add_pad_template (element_class,
 540       gst_static_pad_template_get (&src_template));
 541
 542   gst_element_class_set_static_metadata (element_class,
 543       "Decoder Bin 3", "Generic/Bin/Decoder",
 544       "Autoplug and decode to raw media",
 545       "Edward Hervey <edward@centricular.com>");
 546
 547   bin_klass->handle_message = gst_decodebin3_handle_message;
 548
 549   klass->select_stream = gst_decodebin3_select_stream;
 550 }
 551
 552 static void
 553 gst_decodebin3_init (GstDecodebin3 * dbin)
 554 {
 555   /* Create main input */
 556   dbin->main_input = create_new_input (dbin, TRUE);
 557
 558   dbin->multiqueue = gst_element_factory_make ("multiqueue", NULL);
 559   g_object_set (dbin->multiqueue, "sync-by-running-time", TRUE,
 560       "max-size-buffers", 0, "use-interleave", TRUE, NULL);
 561   gst_bin_add ((GstBin *) dbin, dbin->multiqueue);
 562
 563   dbin->current_group_id = G_MAXUINT32;
 564
 565   g_mutex_init (&dbin->factories_lock);
 566   g_mutex_init (&dbin->selection_lock);
 567   g_mutex_init (&dbin->input_lock);
 568
 569   dbin->caps = gst_static_caps_get (&default_raw_caps);
 570 }
 571
 572 static void
 573 gst_decodebin3_dispose (GObject * object)
 574 {
 575   GstDecodebin3 *dbin = (GstDecodebin3 *) object;
 576
 577   if (dbin->factories)
 578     gst_plugin_feature_list_free (dbin->factories);
 579   if (dbin->decoder_factories)
 580     g_list_free (dbin->decoder_factories);
 581   if (dbin->decodable_factories)
 582     g_list_free (dbin->decodable_factories);
 583   g_list_free (dbin->requested_selection);
 584   g_list_free (dbin->active_selection);
 585   g_list_free (dbin->to_activate);
 586   g_list_free (dbin->pending_select_streams);
 587
 588   free_input (dbin, dbin->main_input);
 589   /* FIXME : GO OVER INPUTS */
 590
 591   G_OBJECT_CLASS (parent_class)->dispose (object);
 592 }
 593
 594 static void
 595 gst_decodebin3_set_property (GObject * object, guint prop_id,
 596     const GValue * value, GParamSpec * pspec)
 597 {
 598   GstDecodebin3 *dbin = (GstDecodebin3 *) object;
 599
 600   /* FIXME : IMPLEMENT */
 601   switch (prop_id) {
 602     case PROP_CAPS:
 603       GST_OBJECT_LOCK (dbin);
 604       if (dbin->caps)
 605         gst_caps_unref (dbin->caps);
 606       dbin->caps = g_value_dup_boxed (value);
 607       GST_OBJECT_UNLOCK (dbin);
 608       break;
 609     default:
 610       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 611       break;
 612   }
 613 }
 614
 615 static void
 616 gst_decodebin3_get_property (GObject * object, guint prop_id, GValue * value,
 617     GParamSpec * pspec)
 618 {
 619   GstDecodebin3 *dbin = (GstDecodebin3 *) object;
 620
 621   /* FIXME : IMPLEMENT */
 622   switch (prop_id) {
 623     case PROP_CAPS:
 624       GST_OBJECT_LOCK (dbin);
 625       g_value_set_boxed (value, dbin->caps);
 626       GST_OBJECT_UNLOCK (dbin);
 627       break;
 628     default:
 629       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 630       break;
 631   }
 632 }
 633
 634 static gboolean
 635 parsebin_autoplug_continue_cb (GstElement * parsebin, GstPad * pad,
 636     GstCaps * caps, GstDecodebin3 * dbin)
 637 {
 638   GST_DEBUG_OBJECT (pad, "caps %" GST_PTR_FORMAT, caps);
 639
 640   /* If it matches our target caps, expose it */
 641   if (gst_caps_can_intersect (caps, dbin->caps))
 642     return FALSE;
 643
 644   return TRUE;
 645 }
 646
 647 /* This method should be called whenever a STREAM_START event
 648  * comes out of a given parsebin.
 649  * The caller shall replace the group_id if the function returns TRUE */
 650 static gboolean
 651 set_input_group_id (DecodebinInput * input, guint32 * group_id)
 652 {
 653   GstDecodebin3 *dbin = input->dbin;
 654
 655   if (input->group_id != *group_id) {
 656     if (input->group_id != G_MAXUINT32)
 657       GST_WARNING_OBJECT (dbin,
 658           "Group id changed (%" G_GUINT32_FORMAT " -> %" G_GUINT32_FORMAT
 659           ") on input %p ", input->group_id, *group_id, input);
 660     input->group_id = *group_id;
 661   }
 662
 663   if (*group_id != dbin->current_group_id) {
 664     if (dbin->current_group_id == G_MAXUINT32) {
 665       GST_DEBUG_OBJECT (dbin, "Setting current group id to %" G_GUINT32_FORMAT,
 666           *group_id);
 667       dbin->current_group_id = *group_id;
 668     }
 669     *group_id = dbin->current_group_id;
 670     return TRUE;
 671   }
 672
 673   return FALSE;
 674 }
 675
 676 /* Call with INPUT_LOCK taken */
 677 static gboolean
 678 ensure_input_parsebin (GstDecodebin3 * dbin, DecodebinInput * input)
 679 {
 680   gboolean set_state = FALSE;
 681
 682   if (input->parsebin == NULL) {
 683     input->parsebin = gst_element_factory_make ("parsebin", NULL);
 684     if (input->parsebin == NULL)
 685       goto no_parsebin;
 686     input->parsebin = gst_object_ref (input->parsebin);
 687     input->parsebin_sink = gst_element_get_static_pad (input->parsebin, "sink");
 688     input->pad_added_sigid =
 689         g_signal_connect (input->parsebin, "pad-added",
 690         (GCallback) parsebin_pad_added_cb, input);
 691     input->pad_removed_sigid =
 692         g_signal_connect (input->parsebin, "pad-removed",
 693         (GCallback) parsebin_pad_removed_cb, input);
 694     g_signal_connect (input->parsebin, "autoplug-continue",
 695         (GCallback) parsebin_autoplug_continue_cb, dbin);
 696   }
 697
 698   if (GST_OBJECT_PARENT (GST_OBJECT (input->parsebin)) != GST_OBJECT (dbin)) {
 699     gst_bin_add (GST_BIN (dbin), input->parsebin);
 700     set_state = TRUE;
 701   }
 702
 703   gst_ghost_pad_set_target (GST_GHOST_PAD (input->ghost_sink),
 704       input->parsebin_sink);
 705   if (set_state)
 706     gst_element_sync_state_with_parent (input->parsebin);
 707
 708   return TRUE;
 709
 710   /* ERRORS */
 711 no_parsebin:
 712   {
 713     gst_element_post_message ((GstElement *) dbin,
 714         gst_missing_element_message_new ((GstElement *) dbin, "parsebin"));
 715     return FALSE;
 716   }
 717 }
 718
 719 static GstPadLinkReturn
 720 gst_decodebin3_input_pad_link (GstPad * pad, GstObject * parent, GstPad * peer)
 721 {
 722   GstDecodebin3 *dbin = (GstDecodebin3 *) parent;
 723   GstPadLinkReturn res = GST_PAD_LINK_OK;
 724   DecodebinInput *input;
 725
 726   GST_LOG_OBJECT (parent, "Got link on input pad %" GST_PTR_FORMAT
 727       ". Creating parsebin if needed", pad);
 728
 729   if ((input = g_object_get_data (G_OBJECT (pad), "decodebin.input")) == NULL)
 730     goto fail;
 731
 732   INPUT_LOCK (dbin);
 733   if (!ensure_input_parsebin (dbin, input))
 734     res = GST_PAD_LINK_REFUSED;
 735   INPUT_UNLOCK (dbin);
 736
 737   return res;
 738 fail:
 739   GST_ERROR_OBJECT (parent, "Failed to retrieve input state from ghost pad");
 740   return GST_PAD_LINK_REFUSED;
 741 }
 742
 743 static void
 744 gst_decodebin3_input_pad_unlink (GstPad * pad, GstObject * parent)
 745 {
 746   GstDecodebin3 *dbin = (GstDecodebin3 *) parent;
 747   DecodebinInput *input;
 748
 749   GST_LOG_OBJECT (parent, "Got unlink on input pad %" GST_PTR_FORMAT
 750       ". Removing parsebin.", pad);
 751
 752   if ((input = g_object_get_data (G_OBJECT (pad), "decodebin.input")) == NULL)
 753     goto fail;
 754
 755   INPUT_LOCK (dbin);
 756   if (input->parsebin == NULL) {
 757     INPUT_UNLOCK (dbin);
 758     return;
 759   }
 760
 761   if (GST_OBJECT_PARENT (GST_OBJECT (input->parsebin)) == GST_OBJECT (dbin)) {
 762     gst_bin_remove (GST_BIN (dbin), input->parsebin);
 763     gst_element_set_state (input->parsebin, GST_STATE_NULL);
 764   }
 765   INPUT_UNLOCK (dbin);
 766   return;
 767
 768 fail:
 769   GST_ERROR_OBJECT (parent, "Failed to retrieve input state from ghost pad");
 770   return;
 771 }
 772
 773 static void
 774 free_input (GstDecodebin3 * dbin, DecodebinInput * input)
 775 {
 776   GST_DEBUG ("Freeing input %p", input);
 777   gst_ghost_pad_set_target (GST_GHOST_PAD (input->ghost_sink), NULL);
 778   gst_element_remove_pad (GST_ELEMENT (dbin), input->ghost_sink);
 779   if (input->parsebin) {
 780     g_signal_handler_disconnect (input->parsebin, input->pad_removed_sigid);
 781     g_signal_handler_disconnect (input->parsebin, input->pad_added_sigid);
 782     gst_element_set_state (input->parsebin, GST_STATE_NULL);
 783     gst_object_unref (input->parsebin);
 784     gst_object_unref (input->parsebin_sink);
 785   }
 786   g_free (input);
 787 }
 788
 789 /* Call with INPUT_LOCK taken */
 790 static DecodebinInput *
 791 create_new_input (GstDecodebin3 * dbin, gboolean main)
 792 {
 793   DecodebinInput *input;
 794
 795   input = g_new0 (DecodebinInput, 1);
 796   input->dbin = dbin;
 797   input->is_main = main;
 798   input->group_id = G_MAXUINT32;
 799   if (main)
 800     input->ghost_sink = gst_ghost_pad_new_no_target ("sink", GST_PAD_SINK);
 801   else {
 802     gchar *pad_name = g_strdup_printf ("sink_%u", dbin->input_counter++);
 803     input->ghost_sink = gst_ghost_pad_new_no_target (pad_name, GST_PAD_SINK);
 804     g_free (pad_name);
 805   }
 806   g_object_set_data (G_OBJECT (input->ghost_sink), "decodebin.input", input);
 807   gst_pad_set_link_function (input->ghost_sink, gst_decodebin3_input_pad_link);
 808   gst_pad_set_unlink_function (input->ghost_sink,
 809       gst_decodebin3_input_pad_unlink);
 810
 811   gst_pad_set_active (input->ghost_sink, TRUE);
 812   gst_element_add_pad ((GstElement *) dbin, input->ghost_sink);
 813
 814   return input;
 815
 816 }
 817
 818 static GstPad *
 819 gst_decodebin3_request_new_pad (GstElement * element, GstPadTemplate * temp,
 820     const gchar * name, const GstCaps * caps)
 821 {
 822   GstDecodebin3 *dbin = (GstDecodebin3 *) element;
 823   DecodebinInput *input;
 824   GstPad *res = NULL;
 825
 826   /* We are ignoring names for the time being, not sure it makes any sense
 827    * within the context of decodebin3 ... */
 828   INPUT_LOCK (dbin);
 829   input = create_new_input (dbin, FALSE);
 830   if (input) {
 831     dbin->other_inputs = g_list_append (dbin->other_inputs, input);
 832     res = input->ghost_sink;
 833   }
 834   INPUT_UNLOCK (dbin);
 835
 836   return res;
 837 }
 838
 839 /* Must be called with factories lock! */
 840 static void
 841 gst_decode_bin_update_factories_list (GstDecodebin3 * dbin)
 842 {
 843   guint cookie;
 844
 845   cookie = gst_registry_get_feature_list_cookie (gst_registry_get ());
 846   if (!dbin->factories || dbin->factories_cookie != cookie) {
 847     GList *tmp;
 848     if (dbin->factories)
 849       gst_plugin_feature_list_free (dbin->factories);
 850     if (dbin->decoder_factories)
 851       g_list_free (dbin->decoder_factories);
 852     if (dbin->decodable_factories)
 853       g_list_free (dbin->decodable_factories);
 854     dbin->factories =
 855         gst_element_factory_list_get_elements
 856         (GST_ELEMENT_FACTORY_TYPE_DECODABLE, GST_RANK_MARGINAL);
 857     dbin->factories =
 858         g_list_sort (dbin->factories, gst_plugin_feature_rank_compare_func);
 859     dbin->factories_cookie = cookie;
 860
 861     /* Filter decoder and other decodables */
 862     dbin->decoder_factories = NULL;
 863     dbin->decodable_factories = NULL;
 864     for (tmp = dbin->factories; tmp; tmp = tmp->next) {
 865       GstElementFactory *fact = (GstElementFactory *) tmp->data;
 866       if (gst_element_factory_list_is_type (fact,
 867               GST_ELEMENT_FACTORY_TYPE_DECODER))
 868         dbin->decoder_factories = g_list_append (dbin->decoder_factories, fact);
 869       else
 870         dbin->decodable_factories =
 871             g_list_append (dbin->decodable_factories, fact);
 872     }
 873   }
 874 }
 875
 876 /* Must be called with appropriate lock if list is a protected variable */
 877 static gboolean
 878 stream_in_list (GList * list, const gchar * sid)
 879 {
 880   GList *tmp;
 881
 882 #if EXTRA_DEBUG
 883   for (tmp = list; tmp; tmp = tmp->next) {
 884     gchar *osid = (gchar *) tmp->data;
 885     GST_DEBUG ("Checking %s against %s", sid, osid);
 886   }
 887 #endif
 888
 889   for (tmp = list; tmp; tmp = tmp->next) {
 890     gchar *osid = (gchar *) tmp->data;
 891     if (!g_strcmp0 (sid, osid))
 892       return TRUE;
 893   }
 894
 895   return FALSE;
 896 }
 897
 898 static void
 899 update_requested_selection (GstDecodebin3 * dbin,
 900     GstStreamCollection * collection)
 901 {
 902   guint i, nb;
 903   GList *tmp = NULL;
 904   GstStreamType used_types = 0;
 905
 906   nb = gst_stream_collection_get_size (collection);
 907
 908   /* 1. Is there a pending SELECT_STREAMS we can return straight away since
 909    *  the switch handler will take care of the pending selection */
 910   SELECTION_LOCK (dbin);
 911   if (dbin->pending_select_streams) {
 912     GST_DEBUG_OBJECT (dbin,
 913         "No need to create pending selection, SELECT_STREAMS underway");
 914     goto beach;
 915   }
 916
 917   /* 2. If not, are we in EXPOSE_ALL_MODE ? If so, match everything */
 918   GST_FIXME_OBJECT (dbin, "Implement EXPOSE_ALL_MODE");
 919
 920   /* 3. If not, check if we already have some of the streams in the
 921    * existing active/requested selection */
 922   for (i = 0; i < nb; i++) {
 923     GstStream *stream = gst_stream_collection_get_stream (collection, i);
 924     const gchar *sid = gst_stream_get_stream_id (stream);
 925     gint request = -1;
 926     /* Fire select-stream signal to see if outside components want to
 927      * hint at which streams should be selected */
 928     g_signal_emit (G_OBJECT (dbin),
 929         gst_decodebin3_signals[SIGNAL_SELECT_STREAM], 0, collection, stream,
 930         &request);
 931     GST_DEBUG_OBJECT (dbin, "stream %s , request:%d", sid, request);
 932     if (request == 1 || (request == -1
 933             && (stream_in_list (dbin->requested_selection, sid)
 934                 || stream_in_list (dbin->active_selection, sid)))) {
 935       GstStreamType curtype = gst_stream_get_stream_type (stream);
 936       if (request == 1)
 937         GST_DEBUG_OBJECT (dbin,
 938             "Using stream requested by 'select-stream' signal : %s", sid);
 939       else
 940         GST_DEBUG_OBJECT (dbin,
 941             "Re-using stream already present in requested or active selection : %s",
 942             sid);
 943       tmp = g_list_append (tmp, (gchar *) sid);
 944       used_types |= curtype;
 945     }
 946   }
 947
 948   /* 4. If not, match one stream of each type */
 949   for (i = 0; i < nb; i++) {
 950     GstStream *stream = gst_stream_collection_get_stream (collection, i);
 951     GstStreamType curtype = gst_stream_get_stream_type (stream);
 952     if (!(used_types & curtype)) {
 953       const gchar *sid = gst_stream_get_stream_id (stream);
 954       GST_DEBUG_OBJECT (dbin, "Selecting stream '%s' of type %s",
 955           sid, gst_stream_type_get_name (curtype));
 956       tmp = g_list_append (tmp, (gchar *) sid);
 957       used_types |= curtype;
 958     }
 959   }
 960
 961 beach:
 962   /* Finally set the requested selection */
 963   if (tmp) {
 964     if (dbin->requested_selection) {
 965       GST_FIXME_OBJECT (dbin,
 966           "Replacing non-NULL requested_selection, what should we do ??");
 967       g_list_free (dbin->requested_selection);
 968     }
 969     dbin->requested_selection = tmp;
 970     dbin->selection_updated = TRUE;
 971   }
 972   SELECTION_UNLOCK (dbin);
 973 }
 974
 975 /* Call with INPUT_LOCK taken */
 976 static GstStreamCollection *
 977 get_merged_collection (GstDecodebin3 * dbin)
 978 {
 979   gboolean needs_merge = FALSE;
 980   GstStreamCollection *res = NULL;
 981   GList *tmp;
 982   guint i, nb_stream;
 983
 984   /* First check if we need to do a merge or just return the only collection */
 985   res = dbin->main_input->collection;
 986
 987   for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
 988     DecodebinInput *input = (DecodebinInput *) tmp->data;
 989     if (input->collection) {
 990       if (res) {
 991         needs_merge = TRUE;
 992         break;
 993       }
 994       res = input->collection;
 995     }
 996   }
 997
 998   if (!needs_merge) {
 999     GST_DEBUG_OBJECT (dbin, "No need to merge, returning %p", res);
1000     return res;
1001   }
1002
1003   /* We really need to create a new collection */
1004   /* FIXME : Some numbering scheme maybe ?? */
1005   res = gst_stream_collection_new ("decodebin3");
1006   if (dbin->main_input->collection) {
1007     nb_stream = gst_stream_collection_get_size (dbin->main_input->collection);
1008     GST_DEBUG_OBJECT (dbin, "main input %p %d", dbin->main_input, nb_stream);
1009     for (i = 0; i < nb_stream; i++) {
1010       GstStream *stream =
1011           gst_stream_collection_get_stream (dbin->main_input->collection, i);
1012       gst_stream_collection_add_stream (res, stream);
1013     }
1014   }
1015
1016   for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
1017     DecodebinInput *input = (DecodebinInput *) tmp->data;
1018     GST_DEBUG_OBJECT (dbin, "input %p , collection %p", input,
1019         input->collection);
1020     if (input->collection) {
1021       nb_stream = gst_stream_collection_get_size (input->collection);
1022       GST_DEBUG_OBJECT (dbin, "nb_stream : %d", nb_stream);
1023       for (i = 0; i < nb_stream; i++) {
1024         GstStream *stream =
1025             gst_stream_collection_get_stream (input->collection, i);
1026         gst_stream_collection_add_stream (res, stream);
1027       }
1028     }
1029   }
1030
1031   return res;
1032 }
1033
1034 /* Call with INPUT_LOCK taken */
1035 static DecodebinInput *
1036 find_message_parsebin (GstDecodebin3 * dbin, GstElement * child)
1037 {
1038   DecodebinInput *input = NULL;
1039   GstElement *parent = gst_object_ref (child);
1040   GList *tmp;
1041
1042   do {
1043     GstElement *next_parent;
1044
1045     GST_DEBUG_OBJECT (dbin, "parent %s",
1046         parent ? GST_ELEMENT_NAME (parent) : "<NONE>");
1047
1048     if (parent == dbin->main_input->parsebin) {
1049       input = dbin->main_input;
1050       break;
1051     }
1052     for (tmp = dbin->other_inputs; tmp; tmp = tmp->next) {
1053       DecodebinInput *cur = (DecodebinInput *) tmp->data;
1054       if (parent == cur->parsebin) {
1055         input = cur;
1056         break;
1057       }
1058     }
1059     next_parent = (GstElement *) gst_element_get_parent (parent);
1060     gst_object_unref (parent);
1061     parent = next_parent;
1062
1063   } while (parent && parent != (GstElement *) dbin);
1064
1065   if (parent)
1066     gst_object_unref (parent);
1067
1068   return input;
1069 }
1070
1071 static gboolean
1072 stream_in_collection (GstDecodebin3 * dbin, gchar * sid)
1073 {
1074   guint i, len;
1075
1076   if (dbin->collection == NULL)
1077     return FALSE;
1078   len = gst_stream_collection_get_size (dbin->collection);
1079   for (i = 0; i < len; i++) {
1080     GstStream *stream = gst_stream_collection_get_stream (dbin->collection, i);
1081     const gchar *osid = gst_stream_get_stream_id (stream);
1082     if (!g_strcmp0 (sid, osid))
1083       return TRUE;
1084   }
1085
1086   return FALSE;
1087 }
1088
1089 /* Call with INPUT_LOCK taken */
1090 static void
1091 handle_stream_collection (GstDecodebin3 * dbin,
1092     GstStreamCollection * collection, GstElement * child)
1093 {
1094 #ifndef GST_DISABLE_GST_DEBUG
1095   const gchar *upstream_id;
1096   guint i;
1097 #endif
1098   DecodebinInput *input = find_message_parsebin (dbin, child);
1099
1100   if (!input) {
1101     GST_DEBUG_OBJECT (dbin,
1102         "Couldn't find corresponding input, most likely shutting down");
1103     return;
1104   }
1105
1106   /* Replace collection in input */
1107   if (input->collection)
1108     gst_object_unref (input->collection);
1109   input->collection = collection;
1110   GST_DEBUG_OBJECT (dbin, "Setting collection %p on input %p", collection,
1111       input);
1112
1113   /* Merge collection if needed */
1114   collection = get_merged_collection (dbin);
1115
1116 #ifndef GST_DISABLE_GST_DEBUG
1117   /* Just some debugging */
1118   upstream_id = gst_stream_collection_get_upstream_id (collection);
1119   GST_DEBUG ("Received Stream Collection. Upstream_id : %s", upstream_id);
1120   GST_DEBUG ("From input %p", input);
1121   GST_DEBUG ("  %d streams", gst_stream_collection_get_size (collection));
1122   for (i = 0; i < gst_stream_collection_get_size (collection); i++) {
1123     GstStream *stream = gst_stream_collection_get_stream (collection, i);
1124     const GstTagList *taglist;
1125     GstCaps *caps;
1126
1127     GST_DEBUG ("   Stream '%s'", gst_stream_get_stream_id (stream));
1128     GST_DEBUG ("     type  : %s",
1129         gst_stream_type_get_name (gst_stream_get_stream_type (stream)));
1130     GST_DEBUG ("     flags : 0x%x", gst_stream_get_stream_flags (stream));
1131     taglist = gst_stream_get_tags (stream);
1132     GST_DEBUG ("     tags  : %" GST_PTR_FORMAT, taglist);
1133     caps = gst_stream_get_caps (stream);
1134     GST_DEBUG ("     caps  : %" GST_PTR_FORMAT, caps);
1135     gst_caps_unref (caps);
1136   }
1137 #endif
1138
1139   /* Store collection for later usage */
1140   if (dbin->collection == NULL) {
1141     dbin->collection = collection;
1142   } else {
1143     /* We need to check who emitted this collection (the owner).
1144      * If we already had a collection from that user, this one is an update,
1145      * that is to say that we need to figure out how we are going to re-use
1146      * the streams/slot */
1147     GST_FIXME_OBJECT (dbin, "New collection but already had one ...");
1148     /* FIXME : When do we switch from pending collection to active collection ?
1149      * When all streams from active collection are drained in multiqueue output ? */
1150     gst_object_unref (dbin->collection);
1151     dbin->collection = collection;
1152     /* dbin->pending_collection = */
1153     /*     g_list_append (dbin->pending_collection, collection); */
1154   }
1155 }
1156
1157 static void
1158 gst_decodebin3_handle_message (GstBin * bin, GstMessage * message)
1159 {
1160   GstDecodebin3 *dbin = (GstDecodebin3 *) bin;
1161   gboolean posting_collection = FALSE;
1162
1163   GST_DEBUG_OBJECT (bin, "Got Message %s", GST_MESSAGE_TYPE_NAME (message));
1164
1165   switch (GST_MESSAGE_TYPE (message)) {
1166     case GST_MESSAGE_STREAM_COLLECTION:
1167     {
1168       GstStreamCollection *collection = NULL;
1169       gst_message_parse_stream_collection (message, &collection);
1170       if (collection) {
1171         INPUT_LOCK (dbin);
1172         handle_stream_collection (dbin, collection,
1173             (GstElement *) GST_MESSAGE_SRC (message));
1174         posting_collection = TRUE;
1175         INPUT_UNLOCK (dbin);
1176       }
1177       if (dbin->collection && collection != dbin->collection) {
1178         /* Replace collection message, we most likely aggregated it */
1179         GstMessage *new_msg;
1180         new_msg =
1181             gst_message_new_stream_collection ((GstObject *) dbin,
1182             dbin->collection);
1183         gst_message_unref (message);
1184         message = new_msg;
1185       }
1186       if (collection)
1187         gst_object_unref (collection);
1188       break;
1189     }
1190     default:
1191       break;
1192   }
1193
1194   GST_BIN_CLASS (parent_class)->handle_message (bin, message);
1195
1196   if (posting_collection) {
1197     /* Figure out a selection for that collection */
1198     update_requested_selection (dbin, dbin->collection);
1199   }
1200 }
1201
1202 static DecodebinOutputStream *
1203 find_free_compatible_output (GstDecodebin3 * dbin, GstStream * stream)
1204 {
1205   GList *tmp;
1206   GstStreamType stype = gst_stream_get_stream_type (stream);
1207
1208   for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
1209     DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
1210     if (output->type == stype && output->slot && output->slot->active_stream) {
1211       GstStream *tstream = output->slot->active_stream;
1212       if (!stream_in_list (dbin->requested_selection,
1213               (gchar *) gst_stream_get_stream_id (tstream))) {
1214         return output;
1215       }
1216     }
1217   }
1218
1219   return NULL;
1220 }
1221
1222 /* Give a certain slot, figure out if it should be linked to an
1223  * output stream
1224  * CALL WITH SELECTION LOCK TAKEN !*/
1225 static DecodebinOutputStream *
1226 get_output_for_slot (MultiQueueSlot * slot)
1227 {
1228   GstDecodebin3 *dbin = slot->dbin;
1229   DecodebinOutputStream *output = NULL;
1230   const gchar *stream_id;
1231   GstCaps *caps;
1232
1233   /* If we already have a configured output, just use it */
1234   if (slot->output != NULL)
1235     return slot->output;
1236
1237   /*
1238    * FIXME
1239    *
1240    * This method needs to be split into multiple parts
1241    *
1242    * 1) Figure out whether stream should be exposed or not
1243    *   This is based on autoplug-continue, EXPOSE_ALL_MODE, or presence
1244    *   in the default stream attribution
1245    *
1246    * 2) Figure out whether an output stream should be created, whether
1247    *   we can re-use the output stream already linked to the slot, or
1248    *   whether we need to get re-assigned another (currently used) output
1249    *   stream.
1250    */
1251
1252   stream_id = gst_stream_get_stream_id (slot->active_stream);
1253   caps = gst_stream_get_caps (slot->active_stream);
1254   GST_DEBUG_OBJECT (dbin, "stream %s , %" GST_PTR_FORMAT, stream_id, caps);
1255   gst_caps_unref (caps);
1256
1257   /* 0. Emit autoplug-continue signal for pending caps ? */
1258   GST_FIXME_OBJECT (dbin, "emit autoplug-continue");
1259
1260   /* 1. if in EXPOSE_ALL_MODE, just accept */
1261   GST_FIXME_OBJECT (dbin, "Handle EXPOSE_ALL_MODE");
1262
1263 #if 0
1264   /* FIXME : The idea around this was to avoid activating a stream for
1265    *     which we have no decoder. Unfortunately it is way too
1266    *     expensive. Need to figure out a better solution */
1267   /* 2. Is there a potential decoder (if one is required) */
1268   if (!gst_caps_can_intersect (caps, dbin->caps)
1269       && !have_factory (dbin, (GstCaps *) caps,
1270           GST_ELEMENT_FACTORY_TYPE_DECODER)) {
1271     GST_WARNING_OBJECT (dbin, "Don't have a decoder for %" GST_PTR_FORMAT,
1272         caps);
1273     SELECTION_UNLOCK (dbin);
1274     gst_element_post_message (GST_ELEMENT_CAST (dbin),
1275         gst_missing_decoder_message_new (GST_ELEMENT_CAST (dbin), caps));
1276     SELECTION_LOCK (dbin);
1277     return NULL;
1278   }
1279 #endif
1280
1281   /* 3. In default mode check if we should expose */
1282   if (stream_in_list (dbin->requested_selection, stream_id)) {
1283     /* Check if we can steal an existing output stream we could re-use.
1284      * that is:
1285      * * an output stream whose slot->stream is not in requested
1286      * * and is of the same type as this stream
1287      */
1288     output = find_free_compatible_output (dbin, slot->active_stream);
1289     if (output) {
1290       /* Move this output from its current slot to this slot */
1291       dbin->to_activate =
1292           g_list_append (dbin->to_activate, (gchar *) stream_id);
1293       dbin->requested_selection =
1294           g_list_remove (dbin->requested_selection, stream_id);
1295       SELECTION_UNLOCK (dbin);
1296       gst_pad_add_probe (output->slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
1297           (GstPadProbeCallback) slot_unassign_probe, output->slot, NULL);
1298       SELECTION_LOCK (dbin);
1299       return NULL;
1300     }
1301
1302     output = create_output_stream (dbin, slot->type);
1303     output->slot = slot;
1304     GST_DEBUG ("Linking slot %p to new output %p", slot, output);
1305     slot->output = output;
1306     dbin->active_selection =
1307         g_list_append (dbin->active_selection, (gchar *) stream_id);
1308   } else
1309     GST_DEBUG ("Not creating any output for slot %p", slot);
1310
1311   return output;
1312 }
1313
1314 /* Returns SELECTED_STREAMS message if active_selection is equal to
1315  * requested_selection, else NULL.
1316  * Must be called with LOCK taken */
1317 static GstMessage *
1318 is_selection_done (GstDecodebin3 * dbin)
1319 {
1320   GList *tmp;
1321   GstMessage *msg;
1322
1323   if (!dbin->selection_updated)
1324     return NULL;
1325
1326   GST_LOG_OBJECT (dbin, "Checking");
1327
1328   if (dbin->to_activate != NULL) {
1329     GST_DEBUG ("Still have streams to activate");
1330     return NULL;
1331   }
1332   for (tmp = dbin->requested_selection; tmp; tmp = tmp->next) {
1333     GST_DEBUG ("Checking requested stream %s", (gchar *) tmp->data);
1334     if (!stream_in_list (dbin->active_selection, (gchar *) tmp->data)) {
1335       GST_DEBUG ("Not in active selection, returning");
1336       return NULL;
1337     }
1338   }
1339
1340   GST_DEBUG_OBJECT (dbin, "Selection active, creating message");
1341
1342   /* We are completely active */
1343   msg = gst_message_new_streams_selected ((GstObject *) dbin, dbin->collection);
1344   GST_MESSAGE_SEQNUM (msg) = dbin->select_streams_seqnum;
1345   for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
1346     DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
1347     if (output->slot) {
1348       GST_DEBUG_OBJECT (dbin, "Adding stream %s",
1349           gst_stream_get_stream_id (output->slot->active_stream));
1350
1351       gst_message_streams_selected_add (msg, output->slot->active_stream);
1352     } else
1353       GST_WARNING_OBJECT (dbin, "No valid slot for output %p", output);
1354   }
1355   dbin->selection_updated = FALSE;
1356   return msg;
1357 }
1358
1359 static GstPadProbeReturn
1360 multiqueue_src_probe (GstPad * pad, GstPadProbeInfo * info,
1361     MultiQueueSlot * slot)
1362 {
1363   GstPadProbeReturn ret = GST_PAD_PROBE_OK;
1364   GstDecodebin3 *dbin = slot->dbin;
1365
1366   if (GST_IS_EVENT (GST_PAD_PROBE_INFO_DATA (info))) {
1367     GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
1368
1369     GST_DEBUG_OBJECT (pad, "Got event %p %s", ev, GST_EVENT_TYPE_NAME (ev));
1370     switch (GST_EVENT_TYPE (ev)) {
1371       case GST_EVENT_STREAM_START:
1372       {
1373         GstStream *stream = NULL;
1374         const gchar *stream_id;
1375
1376         gst_event_parse_stream (ev, &stream);
1377         if (stream == NULL) {
1378           GST_ERROR_OBJECT (pad,
1379               "Got a STREAM_START event without a GstStream");
1380           break;
1381         }
1382         stream_id = gst_stream_get_stream_id (stream);
1383         GST_DEBUG_OBJECT (pad, "Stream Start '%s'", stream_id);
1384         if (slot->active_stream == NULL) {
1385           slot->active_stream = stream;
1386         } else if (slot->active_stream != stream) {
1387           GST_FIXME_OBJECT (pad, "Handle stream changes (%s => %s) !",
1388               gst_stream_get_stream_id (slot->active_stream),
1389               gst_stream_get_stream_id (stream));
1390           gst_object_unref (slot->active_stream);
1391           slot->active_stream = stream;
1392         } else
1393           gst_object_unref (stream);
1394 #if 0                           /* Disabled because stream-start is pushed for every buffer on every unlinked pad */
1395         {
1396           gboolean is_active, is_requested;
1397           /* Quick check to see if we're in the current selection */
1398           /* FIXME : Re-check all slot<=>output mappings based on requested_selection */
1399           SELECTION_LOCK (dbin);
1400           GST_DEBUG_OBJECT (dbin, "Checking active selection");
1401           is_active = stream_in_list (dbin->active_selection, stream_id);
1402           GST_DEBUG_OBJECT (dbin, "Checking requested selection");
1403           is_requested = stream_in_list (dbin->requested_selection, stream_id);
1404           SELECTION_UNLOCK (dbin);
1405           if (is_active)
1406             GST_DEBUG_OBJECT (pad, "Slot in ACTIVE selection (output:%p)",
1407                 slot->output);
1408           if (is_requested)
1409             GST_DEBUG_OBJECT (pad, "Slot in REQUESTED selection (output:%p)",
1410                 slot->output);
1411           else if (slot->output) {
1412             GST_DEBUG_OBJECT (pad,
1413                 "Slot needs to be deactivated ? It's no longer in requested selection");
1414           } else if (!is_active)
1415             GST_DEBUG_OBJECT (pad,
1416                 "Slot in neither active nor requested selection");
1417         }
1418 #endif
1419       }
1420         break;
1421       case GST_EVENT_CAPS:
1422       {
1423         /* Configure the output slot if needed */
1424         DecodebinOutputStream *output;
1425         GstMessage *msg = NULL;
1426         SELECTION_LOCK (dbin);
1427         output = get_output_for_slot (slot);
1428         if (output) {
1429           reconfigure_output_stream (output, slot);
1430           msg = is_selection_done (dbin);
1431         }
1432         SELECTION_UNLOCK (dbin);
1433         if (msg)
1434           gst_element_post_message ((GstElement *) slot->dbin, msg);
1435       }
1436         break;
1437       case GST_EVENT_EOS:
1438         /* FIXME : Figure out */
1439         GST_FIXME_OBJECT (pad, "EOS on multiqueue source pad. input:%p",
1440             slot->input);
1441         if (slot->input == NULL) {
1442           GstPad *peer;
1443           GST_DEBUG_OBJECT (pad,
1444               "last EOS for input, forwarding and removing slot");
1445           peer = gst_pad_get_peer (pad);
1446           if (peer) {
1447             gst_pad_send_event (peer, ev);
1448             gst_object_unref (peer);
1449           }
1450           SELECTION_LOCK (dbin);
1451           /* FIXME : Shouldn't we try to re-assign the output instead of just
1452            * removing it ? */
1453           /* Remove the output */
1454           if (slot->output) {
1455             DecodebinOutputStream *output = slot->output;
1456             dbin->output_streams = g_list_remove (dbin->output_streams, output);
1457             free_output_stream (dbin, output);
1458           }
1459           SELECTION_UNLOCK (dbin);
1460           ret = GST_PAD_PROBE_HANDLED;
1461         }
1462         break;
1463       default:
1464         break;
1465     }
1466   } else if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
1467     GstQuery *query = GST_PAD_PROBE_INFO_QUERY (info);
1468     switch (GST_QUERY_TYPE (query)) {
1469       case GST_QUERY_CAPS:
1470       {
1471         GST_DEBUG_OBJECT (pad, "Intercepting CAPS query");
1472         gst_query_set_caps_result (query, GST_CAPS_ANY);
1473         ret = GST_PAD_PROBE_HANDLED;
1474       }
1475         break;
1476
1477       case GST_QUERY_ACCEPT_CAPS:
1478       {
1479         GST_DEBUG_OBJECT (pad, "Intercepting Accept Caps query");
1480         /* If the current decoder doesn't accept caps, we'll reconfigure
1481          * on the actual caps event. So accept any caps. */
1482         gst_query_set_accept_caps_result (query, TRUE);
1483         ret = GST_PAD_PROBE_HANDLED;
1484       }
1485       default:
1486         break;
1487     }
1488   }
1489
1490   return ret;
1491 }
1492
1493 /* Create a new multiqueue slot for the given type
1494  *
1495  * It is up to the caller to know whether that slot is needed or not
1496  * (and release it when no longer needed) */
1497 static MultiQueueSlot *
1498 create_new_slot (GstDecodebin3 * dbin, GstStreamType type)
1499 {
1500   MultiQueueSlot *slot;
1501   GstIterator *it = NULL;
1502   GValue item = { 0, };
1503
1504   GST_DEBUG_OBJECT (dbin, "Creating new slot for type %s",
1505       gst_stream_type_get_name (type));
1506   slot = g_new0 (MultiQueueSlot, 1);
1507   slot->dbin = dbin;
1508   slot->id = dbin->slot_id++;
1509   slot->type = type;
1510   slot->sink_pad = gst_element_get_request_pad (dbin->multiqueue, "sink_%u");
1511   if (slot->sink_pad == NULL)
1512     goto fail;
1513   it = gst_pad_iterate_internal_links (slot->sink_pad);
1514   if (!it || (gst_iterator_next (it, &item)) != GST_ITERATOR_OK
1515       || ((slot->src_pad = g_value_dup_object (&item)) == NULL)) {
1516     GST_ERROR ("Couldn't get srcpad from multiqueue for sink pad %s:%s",
1517         GST_DEBUG_PAD_NAME (slot->src_pad));
1518     goto fail;
1519   }
1520   gst_iterator_free (it);
1521   g_value_reset (&item);
1522
1523   g_object_set (slot->sink_pad, "group-id", (guint) type, NULL);
1524
1525   /* Add event probe */
1526   slot->probe_id =
1527       gst_pad_add_probe (slot->src_pad,
1528       GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM,
1529       (GstPadProbeCallback) multiqueue_src_probe, slot, NULL);
1530
1531   GST_DEBUG ("Created new slot %u (%p) (%s:%s)", slot->id, slot,
1532       GST_DEBUG_PAD_NAME (slot->src_pad));
1533   dbin->slots = g_list_append (dbin->slots, slot);
1534   return slot;
1535
1536   /* ERRORS */
1537 fail:
1538   {
1539     if (slot->sink_pad)
1540       gst_element_release_request_pad (dbin->multiqueue, slot->sink_pad);
1541     g_free (slot);
1542     return NULL;
1543   }
1544 }
1545
1546 static MultiQueueSlot *
1547 get_slot_for_input (GstDecodebin3 * dbin, DecodebinInputStream * input)
1548 {
1549   GList *tmp;
1550   MultiQueueSlot *empty_slot = NULL;
1551   GstStreamType input_type = 0;
1552   gchar *stream_id = NULL;
1553
1554   GST_DEBUG_OBJECT (dbin, "input %p (stream %p %s)",
1555       input, input->active_stream,
1556       input->
1557       active_stream ? gst_stream_get_stream_id (input->active_stream) : "");
1558
1559   if (input->active_stream) {
1560     input_type = gst_stream_get_stream_type (input->active_stream);
1561     stream_id = (gchar *) gst_stream_get_stream_id (input->active_stream);
1562   }
1563
1564   /* Go over existing slots and check if there is already one for it */
1565   for (tmp = dbin->slots; tmp; tmp = tmp->next) {
1566     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
1567     /* Already used input, return that one */
1568     if (slot->input == input) {
1569       GST_DEBUG_OBJECT (dbin, "Returning already specified slot %d", slot->id);
1570       return slot;
1571     }
1572   }
1573
1574   /* Go amongst all unused slots of the right type and try to find a candidate */
1575   for (tmp = dbin->slots; tmp; tmp = tmp->next) {
1576     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
1577     if (slot->input == NULL && input_type == slot->type) {
1578       /* Remember this empty slot for later */
1579       empty_slot = slot;
1580       /* Check if available slot is of the same stream_id */
1581       GST_LOG_OBJECT (dbin, "Checking candidate slot %d (active_stream:%p)",
1582           slot->id, slot->active_stream);
1583       if (stream_id && slot->active_stream) {
1584         gchar *ostream_id =
1585             (gchar *) gst_stream_get_stream_id (slot->active_stream);
1586         GST_DEBUG_OBJECT (dbin, "Checking slot %d %s against %s", slot->id,
1587             ostream_id, stream_id);
1588         if (!g_strcmp0 (stream_id, ostream_id))
1589           break;
1590       }
1591     }
1592   }
1593
1594   if (empty_slot) {
1595     GST_DEBUG_OBJECT (dbin, "Re-using existing unused slot %d", empty_slot->id);
1596     empty_slot->input = input;
1597     return empty_slot;
1598   }
1599
1600   if (input_type)
1601     return create_new_slot (dbin, input_type);
1602
1603   return NULL;
1604 }
1605
1606 static void
1607 link_input_to_slot (DecodebinInputStream * input, MultiQueueSlot * slot)
1608 {
1609   GstEvent *event;
1610   if (slot->input != NULL && slot->input != input) {
1611     GST_ERROR_OBJECT (slot->dbin,
1612         "Trying to link input to an already used slot");
1613     return;
1614   }
1615   gst_pad_link_full (input->srcpad, slot->sink_pad, GST_PAD_LINK_CHECK_NOTHING);
1616   slot->pending_stream = input->active_stream;
1617   slot->input = input;
1618   event = gst_pad_get_sticky_event (input->srcpad, GST_EVENT_STREAM_START, 0);
1619   if (event)
1620     gst_pad_send_event (slot->sink_pad, event);
1621 }
1622
1623 #if 0
1624 static gboolean
1625 have_factory (GstDecodebin3 * dbin, GstCaps * caps,
1626     GstElementFactoryListType ftype)
1627 {
1628   gboolean ret = FALSE;
1629   GList *res;
1630
1631   g_mutex_lock (&dbin->factories_lock);
1632   gst_decode_bin_update_factories_list (dbin);
1633   if (ftype == GST_ELEMENT_FACTORY_TYPE_DECODER)
1634     res =
1635         gst_element_factory_list_filter (dbin->decoder_factories,
1636         caps, GST_PAD_SINK, TRUE);
1637   else
1638     res =
1639         gst_element_factory_list_filter (dbin->decodable_factories,
1640         caps, GST_PAD_SINK, TRUE);
1641   g_mutex_unlock (&dbin->factories_lock);
1642
1643   if (res) {
1644     ret = TRUE;
1645     gst_plugin_feature_list_free (res);
1646   }
1647
1648   return ret;
1649 }
1650 #endif
1651
1652 static GstElement *
1653 create_element (GstDecodebin3 * dbin, GstStream * stream,
1654     GstElementFactoryListType ftype)
1655 {
1656   GList *res;
1657   GstElement *element = NULL;
1658   GstCaps *caps;
1659
1660   g_mutex_lock (&dbin->factories_lock);
1661   gst_decode_bin_update_factories_list (dbin);
1662   caps = gst_stream_get_caps (stream);
1663   if (ftype == GST_ELEMENT_FACTORY_TYPE_DECODER)
1664     res =
1665         gst_element_factory_list_filter (dbin->decoder_factories,
1666         caps, GST_PAD_SINK, TRUE);
1667   else
1668     res =
1669         gst_element_factory_list_filter (dbin->decodable_factories,
1670         caps, GST_PAD_SINK, TRUE);
1671   g_mutex_unlock (&dbin->factories_lock);
1672
1673   if (res) {
1674     element =
1675         gst_element_factory_create ((GstElementFactory *) res->data, NULL);
1676     GST_DEBUG ("Created element '%s'", GST_ELEMENT_NAME (element));
1677     gst_plugin_feature_list_free (res);
1678   } else {
1679     GST_DEBUG ("Could not find an element for caps %" GST_PTR_FORMAT, caps);
1680   }
1681
1682   gst_caps_unref (caps);
1683   return element;
1684 }
1685
1686 /* FIXME : VERY NAIVE. ASSUMING FIRST ONE WILL WORK */
1687 static GstElement *
1688 create_decoder (GstDecodebin3 * dbin, GstStream * stream)
1689 {
1690   return create_element (dbin, stream, GST_ELEMENT_FACTORY_TYPE_DECODER);
1691 }
1692
1693 static GstPadProbeReturn
1694 keyframe_waiter_probe (GstPad * pad, GstPadProbeInfo * info,
1695     DecodebinOutputStream * output)
1696 {
1697   GstBuffer *buf = GST_PAD_PROBE_INFO_BUFFER (info);
1698   /* If we have a keyframe, remove the probe and let all data through */
1699   /* FIXME : HANDLE HEADER BUFFER ?? */
1700   if (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT) ||
1701       GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_HEADER)) {
1702     GST_DEBUG_OBJECT (pad,
1703         "Buffer is keyframe or header, letting through and removing probe");
1704     output->drop_probe_id = 0;
1705     return GST_PAD_PROBE_REMOVE;
1706   }
1707   GST_DEBUG_OBJECT (pad, "Buffer is not a keyframe, dropping");
1708   return GST_PAD_PROBE_DROP;
1709 }
1710
1711 static void
1712 reconfigure_output_stream (DecodebinOutputStream * output,
1713     MultiQueueSlot * slot)
1714 {
1715   GstDecodebin3 *dbin = output->dbin;
1716   GstCaps *new_caps = (GstCaps *) gst_stream_get_caps (slot->active_stream);
1717   gboolean needs_decoder;
1718
1719   needs_decoder = gst_caps_can_intersect (new_caps, dbin->caps) != TRUE;
1720
1721   GST_DEBUG_OBJECT (dbin,
1722       "Reconfiguring output %p to slot %p, needs_decoder:%d", output, slot,
1723       needs_decoder);
1724
1725   /* FIXME : Maybe make the output un-hook itself automatically ? */
1726   if (output->slot != NULL && output->slot != slot) {
1727     GST_WARNING_OBJECT (dbin,
1728         "Output still linked to another slot (%p)", output->slot);
1729     gst_caps_unref (new_caps);
1730     return;
1731   }
1732
1733   /* Check if existing config is reusable as-is by checking if
1734    * the existing decoder accepts the new caps, if not delete
1735    * it and create a new one */
1736   if (output->decoder) {
1737     gboolean can_reuse_decoder;
1738
1739     if (needs_decoder) {
1740       GstQuery *q = gst_query_new_accept_caps (new_caps);
1741       can_reuse_decoder = gst_pad_query (output->decoder_sink, q);
1742     } else
1743       can_reuse_decoder = FALSE;
1744
1745     if (can_reuse_decoder) {
1746       if (output->type == GST_STREAM_TYPE_VIDEO && output->drop_probe_id == 0) {
1747         GST_DEBUG_OBJECT (dbin, "Adding keyframe-waiter probe");
1748         output->drop_probe_id =
1749             gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_BUFFER,
1750             (GstPadProbeCallback) keyframe_waiter_probe, output, NULL);
1751       }
1752       GST_DEBUG_OBJECT (dbin, "Reusing existing decoder for slot %p", slot);
1753       if (output->linked == FALSE) {
1754         gst_pad_link_full (slot->src_pad, output->decoder_sink,
1755             GST_PAD_LINK_CHECK_NOTHING);
1756         output->linked = TRUE;
1757       }
1758       gst_caps_unref (new_caps);
1759       return;
1760     }
1761
1762     GST_DEBUG_OBJECT (dbin, "Removing old decoder for slot %p", slot);
1763
1764     if (output->linked)
1765       gst_pad_unlink (slot->src_pad, output->decoder_sink);
1766     output->linked = FALSE;
1767     if (output->drop_probe_id) {
1768       gst_pad_remove_probe (slot->src_pad, output->drop_probe_id);
1769       output->drop_probe_id = 0;
1770     }
1771
1772     if (!gst_ghost_pad_set_target ((GstGhostPad *) output->src_pad, NULL)) {
1773       GST_ERROR_OBJECT (dbin, "Could not release decoder pad");
1774       gst_caps_unref (new_caps);
1775       goto cleanup;
1776     }
1777
1778     gst_object_replace ((GstObject **) & output->decoder_sink, NULL);
1779     gst_object_replace ((GstObject **) & output->decoder_src, NULL);
1780
1781     gst_element_set_locked_state (output->decoder, TRUE);
1782     gst_element_set_state (output->decoder, GST_STATE_NULL);
1783
1784     gst_bin_remove ((GstBin *) dbin, output->decoder);
1785     output->decoder = NULL;
1786   }
1787
1788   gst_caps_unref (new_caps);
1789
1790   /* If a decoder is required, create one */
1791   if (needs_decoder) {
1792     /* If we don't have a decoder yet, instantiate one */
1793     output->decoder = create_decoder (dbin, slot->active_stream);
1794     if (output->decoder == NULL) {
1795       GstCaps *caps;
1796
1797       SELECTION_UNLOCK (dbin);
1798       /* FIXME : Should we be smarter if there's a missing decoder ?
1799        * Should we deactivate that stream ? */
1800       caps = gst_stream_get_caps (slot->active_stream);
1801       gst_element_post_message (GST_ELEMENT_CAST (dbin),
1802           gst_missing_decoder_message_new (GST_ELEMENT_CAST (dbin), caps));
1803       gst_caps_unref (caps);
1804       SELECTION_LOCK (dbin);
1805       goto cleanup;
1806     }
1807     if (!gst_bin_add ((GstBin *) dbin, output->decoder)) {
1808       GST_ERROR_OBJECT (dbin, "could not add decoder to pipeline");
1809       goto cleanup;
1810     }
1811     output->decoder_sink = gst_element_get_static_pad (output->decoder, "sink");
1812     output->decoder_src = gst_element_get_static_pad (output->decoder, "src");
1813     if (output->type == GST_STREAM_TYPE_VIDEO) {
1814       GST_DEBUG_OBJECT (dbin, "Adding keyframe-waiter probe");
1815       output->drop_probe_id =
1816           gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_BUFFER,
1817           (GstPadProbeCallback) keyframe_waiter_probe, output, NULL);
1818     }
1819     if (gst_pad_link_full (slot->src_pad, output->decoder_sink,
1820             GST_PAD_LINK_CHECK_NOTHING) != GST_PAD_LINK_OK) {
1821       GST_ERROR_OBJECT (dbin, "could not link to %s:%s",
1822           GST_DEBUG_PAD_NAME (output->decoder_sink));
1823       goto cleanup;
1824     }
1825   } else {
1826     output->decoder_src = gst_object_ref (slot->src_pad);
1827     output->decoder_sink = NULL;
1828   }
1829   output->linked = TRUE;
1830   if (!gst_ghost_pad_set_target ((GstGhostPad *) output->src_pad,
1831           output->decoder_src)) {
1832     GST_ERROR_OBJECT (dbin, "Could not expose decoder pad");
1833     goto cleanup;
1834   }
1835   if (output->src_exposed == FALSE) {
1836     output->src_exposed = TRUE;
1837     gst_element_add_pad (GST_ELEMENT_CAST (dbin), output->src_pad);
1838   }
1839
1840   if (output->decoder)
1841     gst_element_sync_state_with_parent (output->decoder);
1842
1843   output->slot = slot;
1844   return;
1845
1846 cleanup:
1847   {
1848     GST_DEBUG_OBJECT (dbin, "Cleanup");
1849     if (output->decoder_sink) {
1850       gst_object_unref (output->decoder_sink);
1851       output->decoder_sink = NULL;
1852     }
1853     if (output->decoder_src) {
1854       gst_object_unref (output->decoder_src);
1855       output->decoder_src = NULL;
1856     }
1857     if (output->decoder) {
1858       gst_element_set_state (output->decoder, GST_STATE_NULL);
1859       gst_bin_remove ((GstBin *) dbin, output->decoder);
1860       output->decoder = NULL;
1861     }
1862   }
1863 }
1864
1865 static GstPadProbeReturn
1866 idle_reconfigure (GstPad * pad, GstPadProbeInfo * info, MultiQueueSlot * slot)
1867 {
1868   GstMessage *msg = NULL;
1869   DecodebinOutputStream *output;
1870
1871   SELECTION_LOCK (slot->dbin);
1872   output = get_output_for_slot (slot);
1873
1874   GST_DEBUG_OBJECT (pad, "output : %p", output);
1875
1876   if (output) {
1877     reconfigure_output_stream (output, slot);
1878     msg = is_selection_done (slot->dbin);
1879   }
1880   SELECTION_UNLOCK (slot->dbin);
1881   if (msg)
1882     gst_element_post_message ((GstElement *) slot->dbin, msg);
1883
1884   return GST_PAD_PROBE_REMOVE;
1885 }
1886
1887 static MultiQueueSlot *
1888 find_slot_for_stream_id (GstDecodebin3 * dbin, const gchar * sid)
1889 {
1890   GList *tmp;
1891
1892   for (tmp = dbin->slots; tmp; tmp = tmp->next) {
1893     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
1894     const gchar *stream_id;
1895     if (slot->active_stream) {
1896       stream_id = gst_stream_get_stream_id (slot->active_stream);
1897       if (!g_strcmp0 (sid, stream_id))
1898         return slot;
1899     }
1900     if (slot->pending_stream && slot->pending_stream != slot->active_stream) {
1901       stream_id = gst_stream_get_stream_id (slot->pending_stream);
1902       if (!g_strcmp0 (sid, stream_id))
1903         return slot;
1904     }
1905   }
1906
1907   return NULL;
1908 }
1909
1910 /* This function handles the reassignment of a slot. Call this from
1911  * the streaming thread of a slot. */
1912 static gboolean
1913 reassign_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot)
1914 {
1915   DecodebinOutputStream *output;
1916   MultiQueueSlot *target_slot = NULL;
1917   GList *tmp;
1918   const gchar *sid, *tsid;
1919
1920   SELECTION_LOCK (dbin);
1921   output = slot->output;
1922
1923   if (G_UNLIKELY (slot->active_stream == NULL)) {
1924     GST_DEBUG_OBJECT (slot->src_pad,
1925         "Called on inactive slot (active_stream == NULL)");
1926     SELECTION_UNLOCK (dbin);
1927     return FALSE;
1928   }
1929
1930   if (G_UNLIKELY (output == NULL)) {
1931     GST_DEBUG_OBJECT (slot->src_pad,
1932         "Slot doesn't have any output to be removed");
1933     SELECTION_UNLOCK (dbin);
1934     return FALSE;
1935   }
1936
1937   sid = gst_stream_get_stream_id (slot->active_stream);
1938   GST_DEBUG_OBJECT (slot->src_pad, "slot %s %p", sid, slot);
1939
1940   /* Recheck whether this stream is still in the list of streams to deactivate */
1941   if (stream_in_list (dbin->requested_selection, sid)) {
1942     /* Stream is in the list of requested streams, don't remove */
1943     SELECTION_UNLOCK (dbin);
1944     GST_DEBUG_OBJECT (slot->src_pad,
1945         "Stream '%s' doesn't need to be deactivated", sid);
1946     return FALSE;
1947   }
1948
1949   /* Unlink slot from output */
1950   /* FIXME : Handle flushing ? */
1951   /* FIXME : Handle outputs without decoders */
1952   GST_DEBUG_OBJECT (slot->src_pad, "Unlinking from decoder %p",
1953       output->decoder_sink);
1954   if (output->decoder_sink)
1955     gst_pad_unlink (slot->src_pad, output->decoder_sink);
1956   output->linked = FALSE;
1957   slot->output = NULL;
1958   output->slot = NULL;
1959   /* Remove sid from active selection */
1960   for (tmp = dbin->active_selection; tmp; tmp = tmp->next)
1961     if (!g_strcmp0 (sid, tmp->data)) {
1962       dbin->active_selection = g_list_delete_link (dbin->active_selection, tmp);
1963       break;
1964     }
1965
1966   /* Can we re-assign this output to a requested stream ? */
1967   GST_DEBUG_OBJECT (slot->src_pad, "Attempting to re-assing output stream");
1968   for (tmp = dbin->to_activate; tmp; tmp = tmp->next) {
1969     MultiQueueSlot *tslot = find_slot_for_stream_id (dbin, tmp->data);
1970     GST_LOG_OBJECT (tslot->src_pad, "Checking slot %p (output:%p , stream:%s)",
1971         tslot, tslot->output, gst_stream_get_stream_id (tslot->active_stream));
1972     if (tslot && tslot->type == output->type && tslot->output == NULL) {
1973       GST_DEBUG_OBJECT (tslot->src_pad, "Using as reassigned slot");
1974       target_slot = tslot;
1975       tsid = tmp->data;
1976       /* Pass target stream id to requested selection */
1977       dbin->requested_selection =
1978           g_list_append (dbin->requested_selection, tmp->data);
1979       dbin->to_activate = g_list_remove (dbin->to_activate, tmp->data);
1980       break;
1981     }
1982   }
1983
1984   if (target_slot) {
1985     GST_DEBUG_OBJECT (slot->src_pad, "Assigning output to slot %p '%s'",
1986         target_slot, tsid);
1987     target_slot->output = output;
1988     output->slot = target_slot;
1989     dbin->active_selection =
1990         g_list_append (dbin->active_selection, (gchar *) tsid);
1991     SELECTION_UNLOCK (dbin);
1992
1993     /* Wakeup the target slot so that it retries to send events/buffers
1994      * thereby triggering the output reconfiguration codepath */
1995     gst_pad_add_probe (target_slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
1996         (GstPadProbeCallback) idle_reconfigure, target_slot, NULL);
1997     /* gst_pad_send_event (target_slot->src_pad, gst_event_new_reconfigure ()); */
1998   } else {
1999     SELECTION_UNLOCK (dbin);
2000     /* FIXME : Remove output if no longer needed ? The tricky part is knowing
2001      * if it's really no longer needed or not */
2002     GST_FIXME_OBJECT (slot->src_pad, "Remove unused output stream ?");
2003   }
2004
2005   return TRUE;
2006 }
2007
2008 /* Idle probe called when a slot should be unassigned from its output stream.
2009  * This is needed to ensure nothing is flowing when unlinking the slot.
2010  *
2011  * Also, this method will search for a pending stream which could re-use
2012  * the output stream. */
2013 static GstPadProbeReturn
2014 slot_unassign_probe (GstPad * pad, GstPadProbeInfo * info,
2015     MultiQueueSlot * slot)
2016 {
2017   GstDecodebin3 *dbin = slot->dbin;
2018
2019   reassign_slot (dbin, slot);
2020
2021   return GST_PAD_PROBE_REMOVE;
2022 }
2023
2024 static gboolean
2025 handle_stream_switch (GstDecodebin3 * dbin, GList * select_streams,
2026     guint32 seqnum)
2027 {
2028   gboolean ret = TRUE;
2029   GList *tmp;
2030   /* List of slots to (de)activate. */
2031   GList *to_deactivate = NULL;
2032   GList *to_activate = NULL;
2033   /* List of unknown stream id, most likely means the event
2034    * should be sent upstream so that elements can expose the requested stream */
2035   GList *unknown = NULL;
2036   GList *to_reassign = NULL;
2037   GList *future_request_streams = NULL;
2038   GList *pending_streams = NULL;
2039   GList *slots_to_reassign = NULL;
2040
2041   SELECTION_LOCK (dbin);
2042   if (G_UNLIKELY (seqnum != dbin->select_streams_seqnum)) {
2043     GST_DEBUG_OBJECT (dbin, "New SELECT_STREAMS has arrived in the meantime");
2044     SELECTION_UNLOCK (dbin);
2045     return TRUE;
2046   }
2047   /* Remove pending select_streams */
2048   g_list_free (dbin->pending_select_streams);
2049   dbin->pending_select_streams = NULL;
2050
2051   /* COMPARE the requested streams to the active and requested streams
2052    * on multiqueue. */
2053
2054   /* First check the slots to activate and which ones are unknown */
2055   for (tmp = select_streams; tmp; tmp = tmp->next) {
2056     const gchar *sid = (const gchar *) tmp->data;
2057     MultiQueueSlot *slot;
2058     GST_DEBUG_OBJECT (dbin, "Checking stream '%s'", sid);
2059     slot = find_slot_for_stream_id (dbin, sid);
2060     /* Find the corresponding slot */
2061     if (slot == NULL) {
2062       if (stream_in_collection (dbin, (gchar *) sid)) {
2063         pending_streams = g_list_append (pending_streams, (gchar *) sid);
2064       } else {
2065         GST_DEBUG_OBJECT (dbin, "We don't have a slot for stream '%s'", sid);
2066         unknown = g_list_append (unknown, (gchar *) sid);
2067       }
2068     } else if (slot->output == NULL) {
2069       GST_DEBUG_OBJECT (dbin, "We need to activate slot %p for stream '%s')",
2070           slot, sid);
2071       to_activate = g_list_append (to_activate, slot);
2072     } else {
2073       GST_DEBUG_OBJECT (dbin,
2074           "Stream '%s' from slot %p is already active on output %p", sid, slot,
2075           slot->output);
2076       future_request_streams =
2077           g_list_append (future_request_streams, (gchar *) sid);
2078     }
2079   }
2080
2081   for (tmp = dbin->slots; tmp; tmp = tmp->next) {
2082     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
2083     /* For slots that have an output, check if it's part of the streams to
2084      * be active */
2085     if (slot->output) {
2086       gboolean slot_to_deactivate = TRUE;
2087
2088       if (slot->active_stream) {
2089         if (stream_in_list (select_streams,
2090                 gst_stream_get_stream_id (slot->active_stream)))
2091           slot_to_deactivate = FALSE;
2092       }
2093       if (slot_to_deactivate && slot->pending_stream
2094           && slot->pending_stream != slot->active_stream) {
2095         if (stream_in_list (select_streams,
2096                 gst_stream_get_stream_id (slot->pending_stream)))
2097           slot_to_deactivate = FALSE;
2098       }
2099       if (slot_to_deactivate) {
2100         GST_DEBUG_OBJECT (dbin,
2101             "Slot %p (%s) should be deactivated, no longer used", slot,
2102             gst_stream_get_stream_id (slot->active_stream));
2103         to_deactivate = g_list_append (to_deactivate, slot);
2104       }
2105     }
2106   }
2107
2108   if (to_deactivate != NULL) {
2109     GST_DEBUG_OBJECT (dbin, "Check if we can reassign slots");
2110     /* We need to compare what needs to be activated and deactivated in order
2111      * to determine whether there are outputs that can be transferred */
2112     /* Take the stream-id of the slots that are to be activated, for which there
2113      * is a slot of the same type that needs to be deactivated */
2114     tmp = to_deactivate;
2115     while (tmp) {
2116       MultiQueueSlot *slot_to_deactivate = (MultiQueueSlot *) tmp->data;
2117       gboolean removeit = FALSE;
2118       GList *tmp2, *next;
2119       GST_DEBUG_OBJECT (dbin,
2120           "Checking if slot to deactivate (%p) has a candidate slot to activate",
2121           slot_to_deactivate);
2122       for (tmp2 = to_activate; tmp2; tmp2 = tmp2->next) {
2123         MultiQueueSlot *slot_to_activate = (MultiQueueSlot *) tmp2->data;
2124         GST_DEBUG_OBJECT (dbin, "Comparing to slot %p", slot_to_activate);
2125         if (slot_to_activate->type == slot_to_deactivate->type) {
2126           GST_DEBUG_OBJECT (dbin, "Re-using");
2127           to_reassign = g_list_append (to_reassign, (gchar *)
2128               gst_stream_get_stream_id (slot_to_activate->active_stream));
2129           slots_to_reassign =
2130               g_list_append (slots_to_reassign, slot_to_deactivate);
2131           to_activate = g_list_remove (to_activate, slot_to_activate);
2132           removeit = TRUE;
2133           break;
2134         }
2135       }
2136       next = tmp->next;
2137       if (removeit)
2138         to_deactivate = g_list_delete_link (to_deactivate, tmp);
2139       tmp = next;
2140     }
2141   }
2142
2143   for (tmp = to_deactivate; tmp; tmp = tmp->next) {
2144     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
2145     GST_DEBUG_OBJECT (dbin,
2146         "Really need to deactivate slot %p, but no available alternative",
2147         slot);
2148   }
2149
2150   /* The only slots left to activate are the ones that won't be reassigned and
2151    * therefore really need to have a new output created */
2152   for (tmp = to_activate; tmp; tmp = tmp->next) {
2153     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
2154     if (slot->active_stream)
2155       future_request_streams =
2156           g_list_append (future_request_streams,
2157           (gchar *) gst_stream_get_stream_id (slot->active_stream));
2158     else if (slot->pending_stream)
2159       future_request_streams =
2160           g_list_append (future_request_streams,
2161           (gchar *) gst_stream_get_stream_id (slot->pending_stream));
2162     else
2163       GST_ERROR_OBJECT (dbin, "No stream for slot %p !!", slot);
2164   }
2165
2166   if (to_activate == NULL && pending_streams != NULL) {
2167     GST_DEBUG_OBJECT (dbin, "Stream switch requested for future collection");
2168     if (dbin->requested_selection)
2169       g_list_free (dbin->requested_selection);
2170     dbin->requested_selection = select_streams;
2171     g_list_free (to_deactivate);
2172     g_list_free (pending_streams);
2173     to_deactivate = NULL;
2174   } else {
2175     if (dbin->requested_selection)
2176       g_list_free (dbin->requested_selection);
2177     dbin->requested_selection = future_request_streams;
2178     dbin->requested_selection =
2179         g_list_concat (dbin->requested_selection, pending_streams);
2180     if (dbin->to_activate)
2181       g_list_free (dbin->to_activate);
2182     dbin->to_activate = to_reassign;
2183   }
2184
2185   dbin->selection_updated = TRUE;
2186   SELECTION_UNLOCK (dbin);
2187
2188   if (unknown)
2189     GST_FIXME_OBJECT (dbin, "Got request for an unknown stream");
2190
2191   /* For all streams to deactivate, add an idle probe where we will do
2192    * the unassignment and switch over */
2193   for (tmp = slots_to_reassign; tmp; tmp = tmp->next) {
2194     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
2195     gst_pad_add_probe (slot->src_pad, GST_PAD_PROBE_TYPE_IDLE,
2196         (GstPadProbeCallback) slot_unassign_probe, slot, NULL);
2197   }
2198
2199   return ret;
2200 }
2201
2202 static GstPadProbeReturn
2203 ghost_pad_event_probe (GstPad * pad, GstPadProbeInfo * info,
2204     DecodebinOutputStream * output)
2205 {
2206   GstPadProbeReturn ret = GST_PAD_PROBE_OK;
2207   GstDecodebin3 *dbin = output->dbin;
2208   GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
2209
2210   GST_DEBUG_OBJECT (pad, "Got event %p %s", event, GST_EVENT_TYPE_NAME (event));
2211
2212   switch (GST_EVENT_TYPE (event)) {
2213     case GST_EVENT_SELECT_STREAMS:
2214     {
2215       GstPad *peer;
2216       GList *streams = NULL;
2217       guint32 seqnum = gst_event_get_seqnum (event);
2218
2219       SELECTION_LOCK (dbin);
2220       if (seqnum == dbin->select_streams_seqnum) {
2221         SELECTION_UNLOCK (dbin);
2222         GST_DEBUG_OBJECT (pad,
2223             "Already handled/handling that SELECT_STREAMS event");
2224         break;
2225       }
2226       dbin->select_streams_seqnum = seqnum;
2227       if (dbin->pending_select_streams != NULL) {
2228         GST_LOG_OBJECT (dbin, "Replacing pending select streams");
2229         g_list_free (dbin->pending_select_streams);
2230         dbin->pending_select_streams = NULL;
2231       }
2232       gst_event_parse_select_streams (event, &streams);
2233       dbin->pending_select_streams = g_list_copy (streams);
2234       SELECTION_UNLOCK (dbin);
2235
2236       /* Send event upstream */
2237       if ((peer = gst_pad_get_peer (pad))) {
2238         gst_pad_send_event (peer, event);
2239         gst_object_unref (peer);
2240       }
2241       /* Finally handle the switch */
2242       if (streams)
2243         handle_stream_switch (dbin, streams, seqnum);
2244       ret = GST_PAD_PROBE_HANDLED;
2245     }
2246       break;
2247     default:
2248       break;
2249   }
2250
2251   return ret;
2252 }
2253
2254 static gboolean
2255 gst_decodebin3_send_event (GstElement * element, GstEvent * event)
2256 {
2257   GST_DEBUG_OBJECT (element, "event %s", GST_EVENT_TYPE_NAME (event));
2258   if (GST_EVENT_TYPE (event) == GST_EVENT_SELECT_STREAMS) {
2259     GstDecodebin3 *dbin = (GstDecodebin3 *) element;
2260     GList *streams = NULL;
2261     guint32 seqnum = gst_event_get_seqnum (event);
2262
2263     SELECTION_LOCK (dbin);
2264     if (seqnum == dbin->select_streams_seqnum) {
2265       SELECTION_UNLOCK (dbin);
2266       GST_DEBUG_OBJECT (dbin,
2267           "Already handled/handling that SELECT_STREAMS event");
2268       return TRUE;
2269     }
2270     dbin->select_streams_seqnum = seqnum;
2271     if (dbin->pending_select_streams != NULL) {
2272       GST_LOG_OBJECT (dbin, "Replacing pending select streams");
2273       g_list_free (dbin->pending_select_streams);
2274       dbin->pending_select_streams = NULL;
2275     }
2276     gst_event_parse_select_streams (event, &streams);
2277     dbin->pending_select_streams = g_list_copy (streams);
2278     SELECTION_UNLOCK (dbin);
2279
2280     /* FIXME : We don't have an upstream ?? */
2281 #if 0
2282     /* Send event upstream */
2283     if ((peer = gst_pad_get_peer (pad))) {
2284       gst_pad_send_event (peer, event);
2285       gst_object_unref (peer);
2286     }
2287 #endif
2288     /* Finally handle the switch */
2289     if (streams)
2290       handle_stream_switch (dbin, streams, seqnum);
2291
2292     return TRUE;
2293   }
2294   return GST_ELEMENT_CLASS (parent_class)->send_event (element, event);
2295 }
2296
2297
2298 static void
2299 free_multiqueue_slot (GstDecodebin3 * dbin, MultiQueueSlot * slot)
2300 {
2301   if (slot->probe_id)
2302     gst_pad_remove_probe (slot->src_pad, slot->probe_id);
2303   if (slot->input) {
2304     if (slot->input->srcpad)
2305       gst_pad_unlink (slot->input->srcpad, slot->sink_pad);
2306   }
2307
2308   gst_element_release_request_pad (dbin->multiqueue, slot->sink_pad);
2309   gst_object_replace ((GstObject **) & slot->sink_pad, NULL);
2310   gst_object_replace ((GstObject **) & slot->src_pad, NULL);
2311   g_free (slot);
2312 }
2313
2314 /* Create a DecodebinOutputStream for a given type
2315  * Note: It will be empty initially, it needs to be configured
2316  * afterwards */
2317 static DecodebinOutputStream *
2318 create_output_stream (GstDecodebin3 * dbin, GstStreamType type)
2319 {
2320   DecodebinOutputStream *res = g_new0 (DecodebinOutputStream, 1);
2321   gchar *pad_name;
2322   const gchar *prefix;
2323   GstStaticPadTemplate *templ;
2324   GstPadTemplate *ptmpl;
2325   guint32 *counter;
2326   GstPad *internal_pad;
2327
2328   GST_DEBUG_OBJECT (dbin, "Created new output stream %p for type %s",
2329       res, gst_stream_type_get_name (type));
2330
2331   res->type = type;
2332   res->dbin = dbin;
2333
2334   switch (type) {
2335     case GST_STREAM_TYPE_VIDEO:
2336       templ = &video_src_template;
2337       counter = &dbin->vpadcount;
2338       prefix = "video";
2339       break;
2340     case GST_STREAM_TYPE_AUDIO:
2341       templ = &audio_src_template;
2342       counter = &dbin->apadcount;
2343       prefix = "audio";
2344       break;
2345     case GST_STREAM_TYPE_TEXT:
2346       templ = &text_src_template;
2347       counter = &dbin->tpadcount;
2348       prefix = "text";
2349       break;
2350     default:
2351       templ = &src_template;
2352       counter = &dbin->opadcount;
2353       prefix = "src";
2354       break;
2355   }
2356
2357   pad_name = g_strdup_printf ("%s_%u", prefix, *counter);
2358   *counter += 1;
2359   ptmpl = gst_static_pad_template_get (templ);
2360   res->src_pad = gst_ghost_pad_new_no_target_from_template (pad_name, ptmpl);
2361   gst_object_unref (ptmpl);
2362   g_free (pad_name);
2363   gst_pad_set_active (res->src_pad, TRUE);
2364   /* Put an event probe on the internal proxy pad to detect upstream
2365    * events */
2366   internal_pad =
2367       (GstPad *) gst_proxy_pad_get_internal ((GstProxyPad *) res->src_pad);
2368   gst_pad_add_probe (internal_pad, GST_PAD_PROBE_TYPE_EVENT_UPSTREAM,
2369       (GstPadProbeCallback) ghost_pad_event_probe, res, NULL);
2370   gst_object_unref (internal_pad);
2371
2372   dbin->output_streams = g_list_append (dbin->output_streams, res);
2373
2374   return res;
2375 }
2376
2377 static void
2378 free_output_stream (GstDecodebin3 * dbin, DecodebinOutputStream * output)
2379 {
2380   if (output->slot) {
2381     if (output->decoder_sink && output->decoder)
2382       gst_pad_unlink (output->slot->src_pad, output->decoder_sink);
2383
2384     output->slot->output = NULL;
2385     output->slot = NULL;
2386   }
2387   gst_object_replace ((GstObject **) & output->decoder_sink, NULL);
2388   gst_ghost_pad_set_target ((GstGhostPad *) output->src_pad, NULL);
2389   gst_object_replace ((GstObject **) & output->decoder_src, NULL);
2390   if (output->src_exposed) {
2391     gst_element_remove_pad ((GstElement *) dbin, output->src_pad);
2392   }
2393   if (output->decoder) {
2394     gst_element_set_locked_state (output->decoder, TRUE);
2395     gst_element_set_state (output->decoder, GST_STATE_NULL);
2396     gst_bin_remove ((GstBin *) dbin, output->decoder);
2397   }
2398   g_free (output);
2399 }
2400
2401 static GstStateChangeReturn
2402 gst_decodebin3_change_state (GstElement * element, GstStateChange transition)
2403 {
2404   GstDecodebin3 *dbin = (GstDecodebin3 *) element;
2405   GstStateChangeReturn ret;
2406
2407   /* Upwards */
2408   switch (transition) {
2409     default:
2410       break;
2411   }
2412   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2413   if (ret == GST_STATE_CHANGE_FAILURE)
2414     goto beach;
2415
2416   switch (transition) {
2417     case GST_STATE_CHANGE_PAUSED_TO_READY:
2418     {
2419       GList *tmp;
2420
2421       /* Free output streams */
2422       for (tmp = dbin->output_streams; tmp; tmp = tmp->next) {
2423         DecodebinOutputStream *output = (DecodebinOutputStream *) tmp->data;
2424         free_output_stream (dbin, output);
2425       }
2426       g_list_free (dbin->output_streams);
2427       dbin->output_streams = NULL;
2428       /* Free multiqueue slots */
2429       for (tmp = dbin->slots; tmp; tmp = tmp->next) {
2430         MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
2431         free_multiqueue_slot (dbin, slot);
2432       }
2433       g_list_free (dbin->slots);
2434       dbin->slots = NULL;
2435       /* Free inputs */
2436     }
2437       break;
2438     default:
2439       break;
2440   }
2441 beach:
2442   return ret;
2443 }
2444
2445 gboolean
2446 gst_decodebin3_plugin_init (GstPlugin * plugin)
2447 {
2448   GST_DEBUG_CATEGORY_INIT (decodebin3_debug, "decodebin3", 0, "decoder bin");
2449
2450   return gst_element_register (plugin, "decodebin3", GST_RANK_NONE,
2451       GST_TYPE_DECODEBIN3);
2452 }