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