gst/playback/gstdecodebin3-parse.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 #if 0
  24 /* Not needed for now - we're including gstdecodebin3-parse.c into gstdecodebin3.c */
  25 #ifdef HAVE_CONFIG_H
  26 #include "config.h"
  27 #endif
  28
  29 #include <glib.h>
  30 #include <glib-object.h>
  31 #include <glib/gprintf.h>
  32 #include <gst/gst.h>
  33 #include <gst/pbutils/pbutils.h>
  34
  35 #include "gstplayback.h"
  36 #endif
  37
  38 #define CUSTOM_EOS_QUARK _custom_eos_quark_get ()
  39 #define CUSTOM_EOS_QUARK_DATA "custom-eos"
  40 static GQuark
  41 _custom_eos_quark_get (void)
  42 {
  43   static gsize g_quark;
  44
  45   if (g_once_init_enter (&g_quark)) {
  46     gsize quark = (gsize) g_quark_from_static_string ("decodebin3-custom-eos");
  47     g_once_init_leave (&g_quark, quark);
  48   }
  49   return g_quark;
  50 }
  51
  52 /* Streams that come from demuxers (input/upstream) */
  53 /* FIXME : All this is hardcoded. Switch to tree of chains */
  54 struct _DecodebinInputStream
  55 {
  56   GstDecodebin3 *dbin;
  57   GstStream *pending_stream;    /* Extra ref */
  58   GstStream *active_stream;
  59
  60   DecodebinInput *input;
  61
  62   GstPad *srcpad;               /* From demuxer */
  63
  64   /* id of the pad event probe */
  65   gulong output_event_probe_id;
  66
  67   /* id of the buffer blocking probe on the input (demuxer src) pad */
  68   gulong input_buffer_probe_id;
  69
  70   /* Whether we saw an EOS on input. This should be treated accordingly
  71    * when the stream is no longer used */
  72   gboolean saw_eos;
  73 };
  74
  75 static void parsebin_pad_added_cb (GstElement * demux, GstPad * pad,
  76     DecodebinInput * input);
  77 static void parsebin_pad_removed_cb (GstElement * demux, GstPad * pad,
  78     DecodebinInput * input);
  79
  80 /* WITH SELECTION_LOCK TAKEN! */
  81 static gboolean
  82 pending_pads_are_eos (DecodebinInput * input)
  83 {
  84   GList *tmp;
  85
  86   for (tmp = input->pending_pads; tmp; tmp = tmp->next) {
  87     PendingPad *ppad = (PendingPad *) tmp->data;
  88     if (ppad->saw_eos == FALSE)
  89       return FALSE;
  90   }
  91
  92   return TRUE;
  93 }
  94
  95 /* WITH SELECTION_LOCK TAKEN! */
  96 static gboolean
  97 all_inputs_are_eos (GstDecodebin3 * dbin)
  98 {
  99   GList *tmp;
 100   /* First check input streams */
 101   for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
 102     DecodebinInputStream *input = (DecodebinInputStream *) tmp->data;
 103     if (input->saw_eos == FALSE)
 104       return FALSE;
 105   }
 106
 107   /* Check pending pads */
 108   if (!pending_pads_are_eos (dbin->main_input))
 109     return FALSE;
 110   for (tmp = dbin->other_inputs; tmp; tmp = tmp->next)
 111     if (!pending_pads_are_eos ((DecodebinInput *) tmp->data))
 112       return FALSE;
 113
 114   GST_DEBUG_OBJECT (dbin, "All streams are EOS");
 115   return TRUE;
 116 }
 117
 118 /* WITH SELECTION_LOCK TAKEN! */
 119 static void
 120 check_all_streams_for_eos (GstDecodebin3 * dbin)
 121 {
 122   GList *tmp;
 123
 124   if (!all_inputs_are_eos (dbin))
 125     return;
 126
 127   /* We know all streams are EOS, properly clean up everything */
 128   for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
 129     DecodebinInputStream *input = (DecodebinInputStream *) tmp->data;
 130     GstPad *peer = gst_pad_get_peer (input->srcpad);
 131
 132     /* Send EOS and then remove elements */
 133     if (peer) {
 134       gst_pad_send_event (peer, gst_event_new_eos ());
 135       gst_object_unref (peer);
 136     }
 137     GST_FIXME_OBJECT (input->srcpad, "Remove input stream");
 138   }
 139 }
 140
 141 /* Get the intersection of parser caps and available (sorted) decoders */
 142 static GstCaps *
 143 get_parser_caps_filter (GstDecodebin3 * dbin, GstCaps * caps)
 144 {
 145   GList *tmp;
 146   GstCaps *filter_caps = gst_caps_new_empty ();
 147
 148   g_mutex_lock (&dbin->factories_lock);
 149   gst_decode_bin_update_factories_list (dbin);
 150   for (tmp = dbin->decoder_factories; tmp; tmp = tmp->next) {
 151     GstElementFactory *factory = (GstElementFactory *) tmp->data;
 152     GstCaps *tcaps, *intersection;
 153     const GList *tmps;
 154
 155     GST_LOG ("Trying factory %s",
 156         gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
 157     for (tmps = gst_element_factory_get_static_pad_templates (factory); tmps;
 158         tmps = tmps->next) {
 159       GstStaticPadTemplate *st = (GstStaticPadTemplate *) tmps->data;
 160       if (st->direction != GST_PAD_SINK || st->presence != GST_PAD_ALWAYS)
 161         continue;
 162       tcaps = gst_static_pad_template_get_caps (st);
 163       intersection =
 164           gst_caps_intersect_full (tcaps, caps, GST_CAPS_INTERSECT_FIRST);
 165       filter_caps = gst_caps_merge (filter_caps, intersection);
 166       gst_caps_unref (tcaps);
 167     }
 168   }
 169   g_mutex_unlock (&dbin->factories_lock);
 170   GST_DEBUG_OBJECT (dbin, "Got filter caps %" GST_PTR_FORMAT, filter_caps);
 171   return filter_caps;
 172 }
 173
 174 static gboolean
 175 check_parser_caps_filter (GstDecodebin3 * dbin, GstCaps * caps)
 176 {
 177   GList *tmp;
 178   gboolean res = FALSE;
 179
 180   g_mutex_lock (&dbin->factories_lock);
 181   gst_decode_bin_update_factories_list (dbin);
 182   for (tmp = dbin->decoder_factories; tmp; tmp = tmp->next) {
 183     GstElementFactory *factory = (GstElementFactory *) tmp->data;
 184     GstCaps *tcaps;
 185     const GList *tmps;
 186
 187     GST_LOG ("Trying factory %s",
 188         gst_plugin_feature_get_name (GST_PLUGIN_FEATURE (factory)));
 189     for (tmps = gst_element_factory_get_static_pad_templates (factory); tmps;
 190         tmps = tmps->next) {
 191       GstStaticPadTemplate *st = (GstStaticPadTemplate *) tmps->data;
 192       if (st->direction != GST_PAD_SINK || st->presence != GST_PAD_ALWAYS)
 193         continue;
 194       tcaps = gst_static_pad_template_get_caps (st);
 195       if (gst_caps_can_intersect (tcaps, caps)) {
 196         res = TRUE;
 197         gst_caps_unref (tcaps);
 198         goto beach;
 199       }
 200       gst_caps_unref (tcaps);
 201     }
 202   }
 203 beach:
 204   g_mutex_unlock (&dbin->factories_lock);
 205   GST_DEBUG_OBJECT (dbin, "Can intersect : %d", res);
 206   return res;
 207 }
 208
 209 /* Probe on the output of a parser chain (the last
 210  * src pad) */
 211 static GstPadProbeReturn
 212 parse_chain_output_probe (GstPad * pad, GstPadProbeInfo * info,
 213     DecodebinInputStream * input)
 214 {
 215   GstPadProbeReturn ret = GST_PAD_PROBE_OK;
 216
 217   if (GST_IS_EVENT (GST_PAD_PROBE_INFO_DATA (info))) {
 218     GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
 219
 220     GST_DEBUG_OBJECT (pad, "Got event %s", GST_EVENT_TYPE_NAME (ev));
 221     switch (GST_EVENT_TYPE (ev)) {
 222       case GST_EVENT_STREAM_START:
 223       {
 224         GstStream *stream = NULL;
 225         guint group_id = GST_GROUP_ID_INVALID;
 226
 227         if (!gst_event_parse_group_id (ev, &group_id)) {
 228           GST_FIXME_OBJECT (pad,
 229               "Consider implementing group-id handling on stream-start event");
 230           group_id = gst_util_group_id_next ();
 231         }
 232
 233         GST_DEBUG_OBJECT (pad, "Got stream-start, group_id:%d, input %p",
 234             group_id, input->input);
 235         if (set_input_group_id (input->input, &group_id)) {
 236           ev = gst_event_make_writable (ev);
 237           gst_event_set_group_id (ev, group_id);
 238           GST_PAD_PROBE_INFO_DATA (info) = ev;
 239         }
 240         input->saw_eos = FALSE;
 241
 242         gst_event_parse_stream (ev, &stream);
 243         /* FIXME : Would we ever end up with a stream already set on the input ?? */
 244         if (stream) {
 245           if (input->active_stream != stream) {
 246             MultiQueueSlot *slot;
 247             if (input->active_stream)
 248               gst_object_unref (input->active_stream);
 249             input->active_stream = stream;
 250             /* We have the beginning of a stream, get a multiqueue slot and link to it */
 251             SELECTION_LOCK (input->dbin);
 252             slot = get_slot_for_input (input->dbin, input);
 253             link_input_to_slot (input, slot);
 254             SELECTION_UNLOCK (input->dbin);
 255           } else
 256             gst_object_unref (stream);
 257         }
 258       }
 259         break;
 260       case GST_EVENT_CAPS:
 261       {
 262         GstCaps *caps = NULL;
 263         gst_event_parse_caps (ev, &caps);
 264         GST_DEBUG_OBJECT (pad, "caps %" GST_PTR_FORMAT, caps);
 265         if (caps && input->active_stream)
 266           gst_stream_set_caps (input->active_stream, caps);
 267       }
 268         break;
 269       case GST_EVENT_EOS:
 270         input->saw_eos = TRUE;
 271         if (all_inputs_are_eos (input->dbin)) {
 272           GST_DEBUG_OBJECT (pad, "real input pad, marking as EOS");
 273           SELECTION_LOCK (input->dbin);
 274           check_all_streams_for_eos (input->dbin);
 275           SELECTION_UNLOCK (input->dbin);
 276         } else {
 277           GstPad *peer = gst_pad_get_peer (input->srcpad);
 278           if (peer) {
 279             /* Send custom-eos event to multiqueue slot */
 280             GstEvent *event;
 281
 282             GST_DEBUG_OBJECT (pad,
 283                 "Got EOS end of input stream, post custom-eos");
 284             event = gst_event_new_eos ();
 285             gst_event_set_seqnum (event, gst_event_get_seqnum (ev));
 286             gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (event),
 287                 CUSTOM_EOS_QUARK, (gchar *) CUSTOM_EOS_QUARK_DATA, NULL);
 288             gst_pad_send_event (peer, event);
 289             gst_object_unref (peer);
 290           } else {
 291             GST_FIXME_OBJECT (pad, "No peer, what should we do ?");
 292           }
 293         }
 294         ret = GST_PAD_PROBE_DROP;
 295         break;
 296       case GST_EVENT_FLUSH_STOP:
 297         GST_DEBUG_OBJECT (pad, "Clear saw_eos flag");
 298         input->saw_eos = FALSE;
 299       default:
 300         break;
 301     }
 302   } else if (GST_IS_QUERY (GST_PAD_PROBE_INFO_DATA (info))) {
 303     GstQuery *q = GST_PAD_PROBE_INFO_QUERY (info);
 304     GST_DEBUG_OBJECT (pad, "Seeing query %s", GST_QUERY_TYPE_NAME (q));
 305     /* If we have a parser, we want to reply to the caps query */
 306     /* FIXME: Set a flag when the input stream is created for
 307      * streams where we shouldn't reply to these queries */
 308     if (GST_QUERY_TYPE (q) == GST_QUERY_CAPS
 309         && (info->type & GST_PAD_PROBE_TYPE_PULL)) {
 310       GstCaps *filter = NULL;
 311       GstCaps *allowed;
 312       gst_query_parse_caps (q, &filter);
 313       allowed = get_parser_caps_filter (input->dbin, filter);
 314       GST_DEBUG_OBJECT (pad,
 315           "Intercepting caps query, setting %" GST_PTR_FORMAT, allowed);
 316       gst_query_set_caps_result (q, allowed);
 317       gst_caps_unref (allowed);
 318       ret = GST_PAD_PROBE_HANDLED;
 319     } else if (GST_QUERY_TYPE (q) == GST_QUERY_ACCEPT_CAPS) {
 320       GstCaps *prop = NULL;
 321       gst_query_parse_accept_caps (q, &prop);
 322       /* Fast check against target caps */
 323       if (gst_caps_can_intersect (prop, input->dbin->caps))
 324         gst_query_set_accept_caps_result (q, TRUE);
 325       else {
 326         gboolean accepted = check_parser_caps_filter (input->dbin, prop);
 327         /* check against caps filter */
 328         gst_query_set_accept_caps_result (q, accepted);
 329         GST_DEBUG_OBJECT (pad, "ACCEPT_CAPS query, returning %d", accepted);
 330       }
 331       ret = GST_PAD_PROBE_HANDLED;
 332     }
 333   }
 334
 335   return ret;
 336 }
 337
 338 static DecodebinInputStream *
 339 create_input_stream (GstDecodebin3 * dbin, GstStream * stream, GstPad * pad,
 340     DecodebinInput * input)
 341 {
 342   DecodebinInputStream *res = g_new0 (DecodebinInputStream, 1);
 343
 344   GST_DEBUG_OBJECT (pad, "Creating input stream for stream %p %s (input:%p)",
 345       stream, gst_stream_get_stream_id (stream), input);
 346
 347   res->dbin = dbin;
 348   res->input = input;
 349   res->pending_stream = gst_object_ref (stream);
 350   res->srcpad = pad;
 351
 352   /* Put probe on output source pad (for detecting EOS/STREAM_START/FLUSH) */
 353   res->output_event_probe_id =
 354       gst_pad_add_probe (pad,
 355       GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM
 356       | GST_PAD_PROBE_TYPE_EVENT_FLUSH,
 357       (GstPadProbeCallback) parse_chain_output_probe, res, NULL);
 358
 359   /* Add to list of current input streams */
 360   SELECTION_LOCK (dbin);
 361   dbin->input_streams = g_list_append (dbin->input_streams, res);
 362   SELECTION_UNLOCK (dbin);
 363   GST_DEBUG_OBJECT (pad, "Done creating input stream");
 364
 365   return res;
 366 }
 367
 368 /* WITH SELECTION_LOCK TAKEN! */
 369 static void
 370 remove_input_stream (GstDecodebin3 * dbin, DecodebinInputStream * stream)
 371 {
 372   MultiQueueSlot *slot;
 373
 374   GST_DEBUG_OBJECT (dbin, "Removing input stream %p (%s)", stream,
 375       stream->active_stream ? gst_stream_get_stream_id (stream->active_stream) :
 376       "<NONE>");
 377
 378   /* Unlink from slot */
 379   if (stream->srcpad) {
 380     GstPad *peer;
 381     peer = gst_pad_get_peer (stream->srcpad);
 382     if (peer) {
 383       gst_pad_unlink (stream->srcpad, peer);
 384       gst_object_unref (peer);
 385     }
 386   }
 387
 388   slot = get_slot_for_input (dbin, stream);
 389   if (slot) {
 390     slot->pending_stream = NULL;
 391     slot->input = NULL;
 392     GST_DEBUG_OBJECT (dbin, "slot %p cleared", slot);
 393   }
 394
 395   if (stream->active_stream)
 396     gst_object_unref (stream->active_stream);
 397   if (stream->pending_stream)
 398     gst_object_unref (stream->pending_stream);
 399
 400   dbin->input_streams = g_list_remove (dbin->input_streams, stream);
 401
 402   g_free (stream);
 403 }
 404
 405
 406 /* FIXME : HACK, REMOVE, USE INPUT CHAINS */
 407 static GstPadProbeReturn
 408 parsebin_buffer_probe (GstPad * pad, GstPadProbeInfo * info,
 409     DecodebinInput * input)
 410 {
 411   GstDecodebin3 *dbin = input->dbin;
 412   GList *tmp, *unused_slot = NULL;
 413
 414   GST_DEBUG_OBJECT (pad, "Got a buffer ! UNBLOCK !");
 415
 416   /* Any data out the demuxer means it's not creating pads
 417    * any more right now */
 418
 419   /* 1. Re-use existing streams if/when possible */
 420   GST_FIXME_OBJECT (dbin, "Re-use existing input streams if/when possible");
 421
 422   /* 2. Remove unused streams (push EOS) */
 423   GST_DEBUG_OBJECT (dbin, "Removing unused streams");
 424   SELECTION_LOCK (dbin);
 425   tmp = dbin->input_streams;
 426   while (tmp != NULL) {
 427     DecodebinInputStream *input_stream = (DecodebinInputStream *) tmp->data;
 428     GList *next = tmp->next;
 429
 430     if (input_stream->input != input) {
 431       tmp = next;
 432       continue;
 433     }
 434
 435     GST_DEBUG_OBJECT (dbin, "Checking input stream %p", input_stream);
 436     if (input_stream->input_buffer_probe_id) {
 437       GST_DEBUG_OBJECT (dbin,
 438           "Removing pad block on input %p pad %" GST_PTR_FORMAT, input_stream,
 439           input_stream->srcpad);
 440       gst_pad_remove_probe (input_stream->srcpad,
 441           input_stream->input_buffer_probe_id);
 442     }
 443     input_stream->input_buffer_probe_id = 0;
 444
 445     if (input_stream->saw_eos) {
 446       remove_input_stream (dbin, input_stream);
 447       tmp = dbin->input_streams;
 448     } else
 449       tmp = next;
 450   }
 451   SELECTION_UNLOCK (dbin);
 452
 453   GST_DEBUG_OBJECT (dbin, "Creating new streams (if needed)");
 454   /* 3. Create new streams */
 455   for (tmp = input->pending_pads; tmp; tmp = tmp->next) {
 456     GstStream *stream;
 457     PendingPad *ppad = (PendingPad *) tmp->data;
 458
 459     stream = gst_pad_get_stream (ppad->pad);
 460     if (stream == NULL) {
 461       GST_ERROR_OBJECT (dbin, "No stream for pad ????");
 462     } else {
 463       MultiQueueSlot *slot;
 464       DecodebinInputStream *input_stream;
 465       /* The remaining pads in pending_pads are the ones that require a new
 466        * input stream */
 467       input_stream = create_input_stream (dbin, stream, ppad->pad, ppad->input);
 468       /* See if we can link it straight away */
 469       input_stream->active_stream = stream;
 470
 471       SELECTION_LOCK (dbin);
 472       slot = get_slot_for_input (dbin, input_stream);
 473       link_input_to_slot (input_stream, slot);
 474       SELECTION_UNLOCK (dbin);
 475
 476       /* Remove the buffer and event probe */
 477       gst_pad_remove_probe (ppad->pad, ppad->buffer_probe);
 478       gst_pad_remove_probe (ppad->pad, ppad->event_probe);
 479       g_free (ppad);
 480     }
 481   }
 482
 483   g_list_free (input->pending_pads);
 484   input->pending_pads = NULL;
 485
 486   /* Weed out unused multiqueue slots */
 487   SELECTION_LOCK (dbin);
 488   for (tmp = dbin->slots; tmp; tmp = tmp->next) {
 489     MultiQueueSlot *slot = (MultiQueueSlot *) tmp->data;
 490     GST_LOG_OBJECT (dbin, "Slot %d input:%p", slot->id, slot->input);
 491     if (slot->input == NULL) {
 492       unused_slot =
 493           g_list_append (unused_slot, gst_object_ref (slot->sink_pad));
 494     }
 495   }
 496   SELECTION_UNLOCK (dbin);
 497
 498   for (tmp = unused_slot; tmp; tmp = tmp->next) {
 499     GstPad *sink_pad = (GstPad *) tmp->data;
 500     GST_DEBUG_OBJECT (sink_pad, "Sending EOS to unused slot");
 501     gst_pad_send_event (sink_pad, gst_event_new_eos ());
 502   }
 503
 504   if (unused_slot)
 505     g_list_free_full (unused_slot, (GDestroyNotify) gst_object_unref);
 506
 507   return GST_PAD_PROBE_OK;
 508 }
 509
 510 static GstPadProbeReturn
 511 parsebin_pending_event_probe (GstPad * pad, GstPadProbeInfo * info,
 512     PendingPad * ppad)
 513 {
 514   GstDecodebin3 *dbin = ppad->dbin;
 515   /* We drop all events by default */
 516   GstPadProbeReturn ret = GST_PAD_PROBE_DROP;
 517   GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
 518
 519   GST_DEBUG_OBJECT (pad, "Got event %p %s", ev, GST_EVENT_TYPE_NAME (ev));
 520   switch (GST_EVENT_TYPE (ev)) {
 521     case GST_EVENT_EOS:
 522     {
 523       GST_DEBUG_OBJECT (pad, "Pending pad marked as EOS, removing");
 524       ppad->input->pending_pads =
 525           g_list_remove (ppad->input->pending_pads, ppad);
 526       gst_pad_remove_probe (ppad->pad, ppad->buffer_probe);
 527       gst_pad_remove_probe (ppad->pad, ppad->event_probe);
 528       g_free (ppad);
 529
 530       check_all_streams_for_eos (dbin);
 531     }
 532       break;
 533     default:
 534       break;
 535   }
 536
 537   return ret;
 538 }
 539
 540 static void
 541 parsebin_pad_added_cb (GstElement * demux, GstPad * pad, DecodebinInput * input)
 542 {
 543   GstDecodebin3 *dbin = input->dbin;
 544   PendingPad *ppad;
 545   GList *tmp;
 546
 547   GST_DEBUG_OBJECT (dbin, "New pad %s:%s (input:%p)", GST_DEBUG_PAD_NAME (pad),
 548       input);
 549
 550   ppad = g_new0 (PendingPad, 1);
 551   ppad->dbin = dbin;
 552   ppad->input = input;
 553   ppad->pad = pad;
 554
 555   ppad->event_probe =
 556       gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
 557       (GstPadProbeCallback) parsebin_pending_event_probe, ppad, NULL);
 558   ppad->buffer_probe =
 559       gst_pad_add_probe (pad,
 560       GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_BUFFER,
 561       (GstPadProbeCallback) parsebin_buffer_probe, input, NULL);
 562
 563   input->pending_pads = g_list_append (input->pending_pads, ppad);
 564
 565   /* Check if all existing input streams have a buffer probe set */
 566   for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
 567     DecodebinInputStream *input_stream = (DecodebinInputStream *) tmp->data;
 568     if (input_stream->input == input &&
 569         input_stream->input_buffer_probe_id == 0) {
 570       GST_DEBUG_OBJECT (input_stream->srcpad, "Adding blocking buffer probe");
 571       input_stream->input_buffer_probe_id =
 572           gst_pad_add_probe (input_stream->srcpad,
 573           GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_BUFFER,
 574           (GstPadProbeCallback) parsebin_buffer_probe, input_stream->input,
 575           NULL);
 576     }
 577   }
 578 }
 579
 580 static void
 581 parsebin_pad_removed_cb (GstElement * demux, GstPad * pad, DecodebinInput * inp)
 582 {
 583   GstDecodebin3 *dbin = inp->dbin;
 584   DecodebinInputStream *input = NULL;
 585   GList *tmp;
 586   GST_DEBUG_OBJECT (pad, "removed");
 587
 588   for (tmp = dbin->input_streams; tmp; tmp = tmp->next) {
 589     DecodebinInputStream *cand = (DecodebinInputStream *) tmp->data;
 590     if (cand->srcpad == pad)
 591       input = cand;
 592   }
 593   /* If there are no pending pads, this means we will definitely not need this
 594    * stream anymore */
 595   if (input) {
 596     GST_DEBUG_OBJECT (pad, "stream %p", input);
 597     if (inp->pending_pads == NULL) {
 598       MultiQueueSlot *slot;
 599
 600       GST_DEBUG_OBJECT (pad, "Remove input stream %p", input);
 601
 602       SELECTION_LOCK (dbin);
 603       slot = get_slot_for_input (dbin, input);
 604
 605       remove_input_stream (dbin, input);
 606       if (slot && g_list_find (dbin->slots, slot) && slot->is_drained) {
 607         /* if slot is still there and already drained, remove it in here */
 608         if (slot->output) {
 609           DecodebinOutputStream *output = slot->output;
 610           GST_DEBUG_OBJECT (pad,
 611               "Multiqueue was drained, Remove output stream");
 612
 613           dbin->output_streams = g_list_remove (dbin->output_streams, output);
 614           free_output_stream (dbin, output);
 615         }
 616         GST_DEBUG_OBJECT (pad, "No pending pad, Remove multiqueue slot");
 617         if (slot->probe_id)
 618           gst_pad_remove_probe (slot->src_pad, slot->probe_id);
 619         slot->probe_id = 0;
 620         dbin->slots = g_list_remove (dbin->slots, slot);
 621         free_multiqueue_slot_async (dbin, slot);
 622       }
 623       SELECTION_UNLOCK (dbin);
 624     } else {
 625       input->srcpad = NULL;
 626       if (input->input_buffer_probe_id)
 627         gst_pad_remove_probe (pad, input->input_buffer_probe_id);
 628       input->input_buffer_probe_id = 0;
 629     }
 630   }
 631 }