dashdemux/src/gstdashdemux.c

   1 /*
   2  * DASH demux plugin for GStreamer
   3  *
   4  * gstdashdemux.c
   5  *
   6  * Copyright (C) 2012 Orange
   7  *
   8  * Authors:
   9  *   David Corvoysier <david.corvoysier@orange.com>
  10  *   Hamid Zakari <hamid.zakari@gmail.com>
  11  *
  12  * This library is free software; you can redistribute it and/or
  13  * modify it under the terms of the GNU Library General Public
  14  * License as published by the Free Software Foundation; either
  15  * version 2.1 of the License, or (at your option) any later version.
  16  *
  17  * This library is distributed in the hope that it will be useful,
  18  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  19  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  20  * Library General Public License for more details.
  21  *
  22  * You should have received a copy of the GNU Library General Public
  23  * License along with this library (COPYING); if not, write to the
  24  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  25  * Boston, MA 02111-1307, USA.
  26  */
  27 /**
  28  * SECTION:element-dashdemux
  29  *
  30  * DASH demuxer element.
  31  * <title>Example launch line</title>
  32  * |[
  33  * gst-launch playbin2 uri="http://www-itec.uni-klu.ac.at/ftp/datasets/mmsys12/RedBullPlayStreets/redbull_4s/RedBullPlayStreets_4s_isoffmain_DIS_23009_1_v_2_1c2_2011_08_30.mpd"
  34  * ]|
  35  */
  36
  37 /* Implementation notes:
  38  *
  39  * The following section describes how dashdemux works internally.
  40  *
  41  * Introduction:
  42  *
  43  * dashdemux is a "fake" demux, as unlike traditional demux elements, it
  44  * doesn't split data streams contained in an enveloppe to expose them
  45  * to downstream decoding elements.
  46  *
  47  * Instead, it parses an XML file called a manifest to identify a set of
  48  * individual stream fragments it needs to fetch and expose to the actual
  49  * demux elements that will handle them (this behavior is sometimes
  50  * referred as the "demux after a demux" scenario).
  51  *
  52  * For a given section of content, several representations corresponding
  53  * to different bitrates may be available: dashdemux will select the most
  54  * appropriate representation based on local conditions (typically the
  55  * available bandwidth and the amount of buffering available, capped by
  56  * a maximum allowed bitrate).
  57  *
  58  * The representation selection algorithm can be configured using
  59  * specific properties: max bitrate, min/max buffering, bandwidth ratio.
  60  *
  61  *
  62  * General Design:
  63  *
  64  * dashdemux has a single sink pad that accepts the data corresponding
  65  * to the manifest, typically fetched from an HTTP or file source.
  66  *
  67  * dashdemux exposes the streams it recreates based on the fragments it
  68  * fetches through dedicated src pads corresponding to the caps of the
  69  * fragments container (ISOBMFF/MP4 or MPEG2TS).
  70  *
  71  * During playback, new representations will typically be exposed as a
  72  * new set of pads (see 'Switching between representations' below).
  73  *
  74  * Fragments downloading is performed using a dedicated task that fills
  75  * an internal queue. Another task is in charge of popping fragments
  76  * from the queue and pushing them downstream.
  77  *
  78  * Switching between representations:
  79  *
  80  * Decodebin supports scenarios allowing to seamlessly switch from one
  81  * stream to another inside the same "decoding chain".
  82  *
  83  * To achieve that, it combines the elements it autoplugged in chains
  84  *  and groups, allowing only one decoding group to be active at a given
  85  * time for a given chain.
  86  *
  87  * A chain can signal decodebin that it is complete by sending a
  88  * no-more-pads event, but even after that new pads can be added to
  89  * create new subgroups, providing that a new no-more-pads event is sent.
  90  *
  91  * We take advantage of that to dynamically create a new decoding group
  92  * in order to select a different representation during playback.
  93  *
  94  * Typically, assuming that each fragment contains both audio and video,
  95  * the following tree would be created:
  96  *
  97  * chain "DASH Demux"
  98  * |_ group "Representation set 1"
  99  * |   |_ chain "Qt Demux 0"
 100  * |       |_ group "Stream 0"
 101  * |           |_ chain "H264"
 102  * |           |_ chain "AAC"
 103  * |_ group "Representation set 2"
 104  *     |_ chain "Qt Demux 1"
 105  *         |_ group "Stream 1"
 106  *             |_ chain "H264"
 107  *             |_ chain "AAC"
 108  *
 109  * Or, if audio and video are contained in separate fragments:
 110  *
 111  * chain "DASH Demux"
 112  * |_ group "Representation set 1"
 113  * |   |_ chain "Qt Demux 0"
 114  * |   |   |_ group "Stream 0"
 115  * |   |       |_ chain "H264"
 116  * |   |_ chain "Qt Demux 1"
 117  * |       |_ group "Stream 1"
 118  * |           |_ chain "AAC"
 119  * |_ group "Representation set 2"
 120  *     |_ chain "Qt Demux 3"
 121  *     |   |_ group "Stream 2"
 122  *     |       |_ chain "H264"
 123  *     |_ chain "Qt Demux 4"
 124  *         |_ group "Stream 3"
 125  *             |_ chain "AAC"
 126  *
 127  * In both cases, when switching from Set 1 to Set 2 an EOS is sent on
 128  * each end pad corresponding to Rep 0, triggering the "drain" state to
 129  * propagate upstream.
 130  * Once both EOS have been processed, the "Set 1" group is completely
 131  * drained, and decodebin2 will switch to the "Set 2" group.
 132  *
 133  * Note: nothing can be pushed to the new decoding group before the
 134  * old one has been drained, which means that in order to be able to
 135  * adapt quickly to bandwidth changes, we will not be able to rely
 136  * on downstream buffering, and will instead manage an internal queue.
 137  *
 138  */
 139
 140 #ifdef HAVE_CONFIG_H
 141 #  include "config.h"
 142 #endif
 143
 144 /* FIXME 0.11: suppress warnings for deprecated API such as GStaticRecMutex
 145  * with newer GLib versions (>= 2.31.0) */
 146 #define GLIB_DISABLE_DEPRECATION_WARNINGS
 147
 148 #include <string.h>
 149 #include <inttypes.h>
 150 #include <gst/base/gsttypefindhelper.h>
 151 #include "gstdashdemux.h"
 152
 153 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
 154     GST_PAD_SRC,
 155     GST_PAD_SOMETIMES,
 156     GST_STATIC_CAPS_ANY);
 157
 158 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
 159     GST_PAD_SINK,
 160     GST_PAD_ALWAYS,
 161     GST_STATIC_CAPS ("application/dash+xml"));
 162
 163 GST_DEBUG_CATEGORY_STATIC (gst_dash_demux_debug);
 164 #define GST_CAT_DEFAULT gst_dash_demux_debug
 165
 166 enum
 167 {
 168   PROP_0,
 169
 170   PROP_MAX_BUFFERING_TIME,
 171   PROP_BANDWIDTH_USAGE,
 172   PROP_MAX_BITRATE,
 173   PROP_LAST
 174 };
 175
 176 /* Default values for properties */
 177 #define DEFAULT_MAX_BUFFERING_TIME       30     /* in seconds */
 178 #define DEFAULT_BANDWIDTH_USAGE         0.8     /* 0 to 1     */
 179 #define DEFAULT_MAX_BITRATE        24000000     /* in bit/s  */
 180
 181 #define DEFAULT_FAILED_COUNT 3
 182 #define DOWNLOAD_RATE_HISTORY_MAX 3
 183 #define DOWNLOAD_RATE_TIME_MAX 3 * GST_SECOND
 184
 185 /* Custom internal event to signal end of period */
 186 #define GST_EVENT_DASH_EOP GST_EVENT_MAKE_TYPE(81, GST_EVENT_TYPE_DOWNSTREAM | GST_EVENT_TYPE_SERIALIZED)
 187 static GstEvent *
 188 gst_event_new_dash_eop (void)
 189 {
 190   return gst_event_new_custom (GST_EVENT_DASH_EOP, NULL);
 191 }
 192
 193
 194 /* GObject */
 195 static void gst_dash_demux_set_property (GObject * object, guint prop_id,
 196     const GValue * value, GParamSpec * pspec);
 197 static void gst_dash_demux_get_property (GObject * object, guint prop_id,
 198     GValue * value, GParamSpec * pspec);
 199 static void gst_dash_demux_dispose (GObject * obj);
 200
 201 /* GstElement */
 202 static GstStateChangeReturn
 203 gst_dash_demux_change_state (GstElement * element, GstStateChange transition);
 204
 205 /* GstDashDemux */
 206 static GstFlowReturn gst_dash_demux_pad (GstPad * pad, GstBuffer * buf);
 207 static gboolean gst_dash_demux_sink_event (GstPad * pad, GstEvent * event);
 208 static gboolean gst_dash_demux_src_event (GstPad * pad, GstEvent * event);
 209 static gboolean gst_dash_demux_src_query (GstPad * pad, GstQuery * query);
 210 static void gst_dash_demux_stream_loop (GstDashDemux * demux);
 211 static void gst_dash_demux_download_loop (GstDashDemux * demux);
 212 static void gst_dash_demux_stop (GstDashDemux * demux);
 213 static void gst_dash_demux_resume_stream_task (GstDashDemux * demux);
 214 static void gst_dash_demux_resume_download_task (GstDashDemux * demux);
 215 static gboolean gst_dash_demux_setup_all_streams (GstDashDemux * demux);
 216 static gboolean gst_dash_demux_select_representations (GstDashDemux * demux);
 217 static GstCaps *gst_dash_demux_get_input_caps (GstDashDemux * demux, GstActiveStream * stream);
 218 static gboolean gst_dash_demux_get_next_fragment (GstDashDemux * demux, GstActiveStream **fragment_stream, GstClockTime *selected_ts);
 219
 220 static void gst_dash_demux_clear_streams(GstDashDemux * demux);
 221 static void gst_dash_demux_reset (GstDashDemux * demux, gboolean dispose);
 222 static GstClockTime gst_dash_demux_get_buffering_time (GstDashDemux * demux);
 223
 224 static void
 225 _do_init (GType type)
 226 {
 227   GST_DEBUG_CATEGORY_INIT (gst_dash_demux_debug, "dashdemux", 0,
 228       "dashdemux element");
 229 }
 230
 231 GST_BOILERPLATE_FULL (GstDashDemux, gst_dash_demux, GstElement,
 232     GST_TYPE_ELEMENT, _do_init);
 233
 234 static void
 235 gst_dash_demux_base_init (gpointer g_class)
 236 {
 237   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 238
 239   gst_element_class_add_static_pad_template (element_class, &srctemplate);
 240
 241   gst_element_class_add_static_pad_template (element_class, &sinktemplate);
 242
 243   gst_element_class_set_details_simple (element_class,
 244       "DASH Demuxer",
 245       "Codec/Demuxer",
 246       "Dynamic Adaptive Streaming over HTTP demuxer",
 247       "David Corvoysier <david.corvoysier@orange.com>\n\
 248                 Hamid Zakari <hamid.zakari@gmail.com>\n\
 249                 Gianluca Gennari <gennarone@gmail.com>");
 250 }
 251
 252 static void
 253 gst_dash_demux_dispose (GObject * obj)
 254 {
 255   GstDashDemux *demux = GST_DASH_DEMUX (obj);
 256
 257   if (demux->stream_task) {
 258     if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED) {
 259       GST_DEBUG_OBJECT (demux, "Leaving streaming task");
 260       gst_task_stop (demux->stream_task);
 261       gst_task_join (demux->stream_task);
 262     }
 263     gst_object_unref (demux->stream_task);
 264     g_static_rec_mutex_free (&demux->stream_lock);
 265     g_mutex_free(demux->stream_timed_lock);
 266     demux->stream_task = NULL;
 267   }
 268
 269   if (demux->download_task) {
 270     if (GST_TASK_STATE (demux->download_task) != GST_TASK_STOPPED) {
 271       GST_DEBUG_OBJECT (demux, "Leaving download task");
 272       gst_task_stop (demux->download_task);
 273       gst_task_join (demux->download_task);
 274     }
 275     gst_object_unref (demux->download_task);
 276     g_static_rec_mutex_free (&demux->download_lock);
 277     demux->download_task = NULL;
 278   }
 279
 280   g_cond_clear (&demux->download_cond);
 281   g_mutex_clear (&demux->download_mutex);
 282
 283   if (demux->downloader != NULL) {
 284     g_object_unref (demux->downloader);
 285     demux->downloader = NULL;
 286   }
 287
 288   gst_dash_demux_reset (demux, TRUE);
 289
 290   G_OBJECT_CLASS (parent_class)->dispose (obj);
 291 }
 292
 293 static void
 294 gst_dash_demux_class_init (GstDashDemuxClass * klass)
 295 {
 296   GObjectClass *gobject_class;
 297   GstElementClass *gstelement_class;
 298
 299   gobject_class = (GObjectClass *) klass;
 300   gstelement_class = (GstElementClass *) klass;
 301
 302   gobject_class->set_property = gst_dash_demux_set_property;
 303   gobject_class->get_property = gst_dash_demux_get_property;
 304   gobject_class->dispose = gst_dash_demux_dispose;
 305
 306   g_object_class_install_property (gobject_class, PROP_MAX_BUFFERING_TIME,
 307       g_param_spec_uint ("max-buffering-time", "Maximum buffering time",
 308           "Maximum number of seconds of buffer accumulated during playback",
 309           2, G_MAXUINT, DEFAULT_MAX_BUFFERING_TIME,
 310           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 311
 312   g_object_class_install_property (gobject_class, PROP_BANDWIDTH_USAGE,
 313       g_param_spec_float ("bandwidth-usage",
 314           "Bandwidth usage [0..1]",
 315           "Percentage of the available bandwidth to use when selecting representations",
 316           0, 1, DEFAULT_BANDWIDTH_USAGE,
 317           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 318
 319   g_object_class_install_property (gobject_class, PROP_MAX_BITRATE,
 320       g_param_spec_uint ("max-bitrate", "Max bitrate",
 321           "Max of bitrate supported by target decoder",
 322           1000, G_MAXUINT, DEFAULT_MAX_BITRATE,
 323           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 324
 325   gstelement_class->change_state =
 326       GST_DEBUG_FUNCPTR (gst_dash_demux_change_state);
 327 }
 328
 329 static gboolean
 330 _check_queue_full (GstDataQueue * q, guint visible, guint bytes, guint64 time,
 331     GstDashDemux *demux)
 332 {
 333   return time > demux->max_buffering_time;
 334 }
 335
 336 static void
 337 _data_queue_item_destroy (GstDataQueueItem * item)
 338 {
 339   gst_mini_object_unref (item->object);
 340   g_free (item);
 341 }
 342
 343 static void
 344 gst_dash_demux_stream_push_event (GstDashDemuxStream * stream,
 345                                   GstEvent *event)
 346 {
 347   GstDataQueueItem *item = g_new0 (GstDataQueueItem, 1);
 348
 349   item->object = GST_MINI_OBJECT_CAST (event);
 350   item->destroy = (GDestroyNotify) _data_queue_item_destroy;
 351
 352   gst_data_queue_push (stream->queue, item);
 353 }
 354
 355 static void
 356 gst_dash_demux_stream_push_data (GstDashDemuxStream * stream,
 357     GstBuffer * buffer)
 358 {
 359   GstDataQueueItem *item = g_new0 (GstDataQueueItem, 1);
 360
 361   item->object = GST_MINI_OBJECT_CAST (buffer);
 362   item->duration = GST_BUFFER_DURATION (buffer);
 363   item->visible = TRUE;
 364   item->size = GST_BUFFER_SIZE (buffer);
 365
 366   item->destroy = (GDestroyNotify) _data_queue_item_destroy;
 367
 368   gst_data_queue_push (stream->queue, item);
 369 }
 370
 371 static void
 372 gst_dash_demux_init (GstDashDemux * demux, GstDashDemuxClass * klass)
 373 {
 374   /* sink pad */
 375   demux->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 376   gst_pad_set_chain_function (demux->sinkpad,
 377       GST_DEBUG_FUNCPTR (gst_dash_demux_pad));
 378   gst_pad_set_event_function (demux->sinkpad,
 379       GST_DEBUG_FUNCPTR (gst_dash_demux_sink_event));
 380   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 381
 382   /* Downloader */
 383   demux->downloader = gst_uri_downloader_new ();
 384
 385   /* Properties */
 386   demux->max_buffering_time = DEFAULT_MAX_BUFFERING_TIME * GST_SECOND;
 387   demux->bandwidth_usage = DEFAULT_BANDWIDTH_USAGE;
 388   demux->max_bitrate = DEFAULT_MAX_BITRATE;
 389
 390   demux->max_video_width = 0;
 391   demux->max_video_height = 0;
 392
 393   /* Updates task */
 394   g_static_rec_mutex_init (&demux->download_lock);
 395   demux->download_task =
 396       gst_task_create ((GstTaskFunction) gst_dash_demux_download_loop, demux);
 397   gst_task_set_lock (demux->download_task, &demux->download_lock);
 398   g_cond_init (&demux->download_cond);
 399   g_mutex_init (&demux->download_mutex);
 400
 401   /* Streaming task */
 402   g_static_rec_mutex_init (&demux->stream_lock);
 403   demux->stream_task =
 404       gst_task_create ((GstTaskFunction) gst_dash_demux_stream_loop, demux);
 405   gst_task_set_lock (demux->stream_task, &demux->stream_lock);
 406   demux->stream_timed_lock = g_mutex_new ();
 407 }
 408
 409 static void
 410 gst_dash_demux_set_property (GObject * object, guint prop_id,
 411     const GValue * value, GParamSpec * pspec)
 412 {
 413   GstDashDemux *demux = GST_DASH_DEMUX (object);
 414
 415   switch (prop_id) {
 416     case PROP_MAX_BUFFERING_TIME:
 417       demux->max_buffering_time = g_value_get_uint (value) * GST_SECOND;
 418       break;
 419     case PROP_BANDWIDTH_USAGE:
 420       demux->bandwidth_usage = g_value_get_float (value);
 421       break;
 422     case PROP_MAX_BITRATE:
 423       demux->max_bitrate = g_value_get_uint (value);
 424       break;
 425     default:
 426       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 427       break;
 428   }
 429 }
 430
 431 static void
 432 gst_dash_demux_get_property (GObject * object, guint prop_id, GValue * value,
 433     GParamSpec * pspec)
 434 {
 435   GstDashDemux *demux = GST_DASH_DEMUX (object);
 436
 437   switch (prop_id) {
 438     case PROP_MAX_BUFFERING_TIME:
 439       g_value_set_uint (value, demux->max_buffering_time / GST_SECOND);
 440       break;
 441     case PROP_BANDWIDTH_USAGE:
 442       g_value_set_float (value, demux->bandwidth_usage);
 443       break;
 444     case PROP_MAX_BITRATE:
 445       g_value_set_uint (value, demux->max_bitrate);
 446       break;
 447     default:
 448       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 449       break;
 450   }
 451 }
 452
 453 static GstStateChangeReturn
 454 gst_dash_demux_change_state (GstElement * element, GstStateChange transition)
 455 {
 456   GstStateChangeReturn ret;
 457   GstDashDemux *demux = GST_DASH_DEMUX (element);
 458
 459   GST_DEBUG_OBJECT (demux, "changing state %s - %s",
 460       gst_element_state_get_name (GST_STATE_TRANSITION_CURRENT (transition)),
 461       gst_element_state_get_name (GST_STATE_TRANSITION_NEXT (transition)));
 462
 463   switch (transition) {
 464     case GST_STATE_CHANGE_READY_TO_PAUSED:
 465       gst_dash_demux_reset (demux, FALSE);
 466       break;
 467     default:
 468       break;
 469   }
 470
 471   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 472
 473   switch (transition) {
 474     case GST_STATE_CHANGE_PAUSED_TO_READY:
 475       demux->cancelled = TRUE;
 476       gst_dash_demux_stop (demux);
 477       gst_task_join (demux->stream_task);
 478       gst_task_join (demux->download_task);
 479       break;
 480     default:
 481       break;
 482   }
 483   return ret;
 484 }
 485
 486 void
 487 gst_dash_demux_flush_stream_queues (GstDashDemux * demux)
 488 {
 489   GSList *it;
 490   GstDashDemuxStream *stream;
 491   for(it = demux->streams; it; it=it->next)
 492   {
 493     stream = it->data;
 494     gst_data_queue_set_flushing(stream->queue, TRUE);
 495     gst_data_queue_flush(stream->queue);
 496   }
 497 }
 498
 499 static gboolean
 500 gst_dash_demux_src_event (GstPad * pad, GstEvent * event)
 501 {
 502   GstDashDemux *demux;
 503
 504   demux = GST_DASH_DEMUX (gst_pad_get_element_private (pad));
 505
 506   switch (event->type) {
 507     case GST_EVENT_SEEK:
 508     {
 509       gdouble rate;
 510       GstFormat format;
 511       GstSeekFlags flags;
 512       GstSeekType start_type, stop_type;
 513       gint64 start, stop;
 514       GList *list;
 515       GstClockTime current_pos = GST_CLOCK_TIME_NONE;
 516       GstClockTime target_pos;
 517       guint current_period;
 518       GstActiveStream *stream;
 519       GstStreamPeriod *period = NULL;
 520       guint nb_active_stream;
 521       guint stream_idx = 0;
 522       guint *seek_idx = NULL;   /*Seek positions on each stream*/
 523       gboolean end_of_mpd = FALSE;
 524
 525       if (gst_mpd_client_is_live (demux->client)) {
 526         GST_WARNING_OBJECT (demux, "Received seek event for live stream");
 527         return FALSE;
 528       }
 529
 530       gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
 531           &stop_type, &stop);
 532
 533       if (format != GST_FORMAT_TIME)
 534         return FALSE;
 535
 536       GST_DEBUG_OBJECT (demux,
 537           "seek event, rate: %f type: %d start: %" GST_TIME_FORMAT " stop: %"
 538           GST_TIME_FORMAT, rate, start_type, GST_TIME_ARGS (start),
 539           GST_TIME_ARGS (stop));
 540
 541       //GST_MPD_CLIENT_LOCK (demux->client);
 542
 543       /* select the requested Period in the Media Presentation */
 544       target_pos = (GstClockTime) start;
 545       current_period = 0;
 546       for (list = g_list_first (demux->client->periods); list;
 547           list = g_list_next (list)) {
 548         period = list->data;
 549         current_pos = period->start;
 550         current_period = period->number;
 551         if (current_pos <= target_pos
 552             && target_pos < current_pos + period->duration) {
 553           break;
 554         }
 555       }
 556       if(target_pos == current_pos + period->duration) {
 557         /*Seeking to the end of MPD*/
 558         end_of_mpd = TRUE;
 559         goto seeking;
 560       }
 561       if (list == NULL) {
 562         GST_WARNING_OBJECT (demux, "Could not find seeked Period");
 563         return FALSE;
 564       }
 565       if (current_period != gst_mpd_client_get_period_index (demux->client)) {
 566         GST_DEBUG_OBJECT (demux, "Seeking to Period %d", current_period);
 567         /* setup video, audio and subtitle streams, starting from the new Period */
 568         if (!gst_mpd_client_set_period_index (demux->client, current_period) ||
 569             !gst_dash_demux_setup_all_streams (demux))
 570           return FALSE;
 571       }
 572
 573       /*select the requested segments for all streams*/
 574       nb_active_stream = gst_mpdparser_get_nb_active_stream (demux->client);
 575       seek_idx = g_malloc0(sizeof(gint)*nb_active_stream);
 576       gint video_idx = gst_mpd_client_get_video_active_stream_id(demux->client);
 577       if(video_idx >= 0) {
 578         /*Seeking on video stream firstly.*/
 579         GstClockTime segment_start;
 580         segment_start = gst_mpd_client_stream_find_segment(demux->client, video_idx,
 581                                                          target_pos, &seek_idx[video_idx]);
 582         if(!GST_CLOCK_TIME_IS_VALID(segment_start))
 583           goto no_segment;
 584         target_pos = segment_start;
 585       }
 586       /*Seeking on non video streams*/
 587       for (stream_idx = 0; stream_idx < nb_active_stream; stream_idx++) {
 588         if (video_idx != stream_idx) {
 589           GstClockTime stream_start = gst_mpd_client_stream_find_segment(demux->client,
 590                                           stream_idx, target_pos, &seek_idx[stream_idx]);
 591           if(!GST_CLOCK_TIME_IS_VALID (stream_start)) {
 592             goto no_segment;
 593           }
 594         }
 595       }
 596
 597 seeking:
 598       /* We can actually perform the seek */
 599       nb_active_stream = gst_mpdparser_get_nb_active_stream (demux->client);
 600
 601       if (flags & GST_SEEK_FLAG_FLUSH) {
 602         GST_DEBUG_OBJECT (demux, "sending flush start");
 603         stream_idx = 0;
 604         while (stream_idx < nb_active_stream) {
 605           GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, stream_idx);
 606           dash_stream->need_header = TRUE;
 607           gst_pad_push_event (dash_stream->srcpad,
 608               gst_event_new_flush_start ());
 609           stream_idx++;
 610         }
 611       }
 612
 613       /* Stop the demux */
 614       demux->cancelled = TRUE;
 615       gst_dash_demux_stop (demux);
 616       GST_DEBUG_OBJECT (demux, "joining tasks");
 617       gst_task_join (demux->stream_task);
 618       gst_task_join (demux->download_task);
 619       GST_DEBUG_OBJECT (demux, "tasks was joined");
 620
 621       /* Wait for streaming to finish */
 622       g_static_rec_mutex_lock (&demux->stream_lock);
 623
 624       //GST_MPD_CLIENT_LOCK (demux->client);
 625       demux->end_of_period = end_of_mpd;
 626       //GST_MPD_CLIENT_UNLOCK (demux->client);
 627
 628
 629       for (stream_idx = 0; stream_idx < nb_active_stream; stream_idx++) {
 630         GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, stream_idx);
 631         GstCaps *caps = gst_pad_get_negotiated_caps (dash_stream->srcpad);
 632         if(caps) {
 633           gst_caps_replace (&dash_stream->input_caps, NULL);
 634           gst_caps_unref (caps);
 635         }
 636         if(!end_of_mpd) {
 637           GST_DEBUG_OBJECT (demux, "Seeking to sequence %d on stream %d", seek_idx[stream_idx], stream_idx);
 638           stream = gst_mpdparser_get_active_stream_by_index (demux->client, stream_idx);
 639           gst_mpd_client_set_segment_index(stream, seek_idx[stream_idx]);
 640         }
 641         gst_data_queue_set_flushing(dash_stream->queue, FALSE);
 642         dash_stream->start_time = target_pos;
 643         dash_stream->download_end_of_period = end_of_mpd;
 644         dash_stream->stream_end_of_period = end_of_mpd;
 645         dash_stream->stream_eos = end_of_mpd;
 646         dash_stream->need_segment = TRUE;
 647       }
 648       if(!end_of_mpd)
 649         g_free(seek_idx);
 650
 651       if (flags & GST_SEEK_FLAG_FLUSH) {
 652         GST_DEBUG_OBJECT (demux, "Sending flush stop on all pad");
 653
 654         for (stream_idx = 0; stream_idx < nb_active_stream; stream_idx++) {
 655           GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, stream_idx);
 656           gst_pad_push_event (dash_stream->srcpad,
 657               gst_event_new_flush_stop ());
 658         }
 659       }
 660
 661       /* Restart the demux */
 662       demux->cancelled = FALSE;
 663       gst_dash_demux_resume_download_task (demux);
 664       gst_dash_demux_resume_stream_task (demux);
 665       g_static_rec_mutex_unlock (&demux->stream_lock);
 666
 667       return TRUE;
 668 no_segment:
 669       {
 670         GST_WARNING_OBJECT (demux, "Could not find seeked fragment on stream %d", stream_idx);
 671         g_free(seek_idx);
 672         return FALSE;
 673       }
 674     }
 675     default:
 676       break;
 677   }
 678
 679   return gst_pad_event_default (pad, event);
 680 }
 681
 682 static gboolean
 683 gst_dash_demux_setup_mpdparser_streams (GstDashDemux * demux, GstMpdClient *client)
 684 {
 685   GList *listLang = NULL;
 686   guint i, nb_audio;
 687   gchar *lang;
 688
 689   GST_MPD_CLIENT_LOCK (client);
 690   /* clean old active stream list, if any */
 691   gst_active_streams_free (client);
 692
 693   if (!gst_mpd_client_setup_streaming (client, GST_STREAM_VIDEO, "")) {
 694     GST_INFO_OBJECT (demux, "No video adaptation set found");
 695   } else {
 696     gst_mpd_client_get_max_video_dimensions(client, &demux->max_video_width,
 697                                             &demux->max_video_height);
 698   }
 699
 700   nb_audio =
 701       gst_mpdparser_get_list_and_nb_of_audio_language (client,
 702       &listLang);
 703   if (nb_audio == 0)
 704     nb_audio = 1;
 705   GST_INFO_OBJECT (demux, "Number of language is=%d", nb_audio);
 706
 707   for (i = 0; i < nb_audio; i++) {
 708     lang = (gchar *) g_list_nth_data (listLang, i);
 709     if (gst_mpdparser_get_nb_adaptationSet (client) > 1)
 710       if (!gst_mpd_client_setup_streaming (client, GST_STREAM_AUDIO,
 711               lang))
 712         GST_INFO_OBJECT (demux, "No audio adaptation set found");
 713
 714     if (gst_mpdparser_get_nb_adaptationSet (client) > nb_audio)
 715       if (!gst_mpd_client_setup_streaming (client,
 716               GST_STREAM_APPLICATION, lang))
 717         GST_INFO_OBJECT (demux, "No application adaptation set found");
 718   }
 719   GST_MPD_CLIENT_UNLOCK (client);
 720   return TRUE;
 721 }
 722
 723 static gboolean
 724 gst_dash_demux_setup_all_streams (GstDashDemux * demux)
 725 {
 726   guint i;
 727   if( !gst_dash_demux_setup_mpdparser_streams(demux, demux->client))
 728     return FALSE;
 729
 730   GST_DEBUG_OBJECT (demux, "Creating dashdemux streams");
 731   gst_dash_demux_clear_streams(demux);
 732   for ( i =0; i < gst_mpdparser_get_nb_active_stream (demux->client); i++) {
 733     GstDashDemuxStream *dash_stream;
 734     GstCaps *caps;
 735     GstActiveStream *active_stream;
 736     dash_stream = g_new0(GstDashDemuxStream, 1);
 737     demux->streams = g_slist_append(demux->streams, dash_stream);
 738     dash_stream->idx = i;
 739     dash_stream->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction) _check_queue_full, demux);
 740     dash_stream->need_header = TRUE;
 741     dash_stream->need_segment = TRUE;
 742     dash_stream->start_time = GST_CLOCK_TIME_NONE;
 743     gst_download_rate_init (&dash_stream->dnl_rate);
 744     gst_download_rate_set_max_length (&dash_stream->dnl_rate,
 745                                       DOWNLOAD_RATE_HISTORY_MAX);
 746     gst_download_rate_set_aver_period (&dash_stream->dnl_rate,
 747                                        DOWNLOAD_RATE_TIME_MAX);
 748     /*Create stream pad*/
 749     active_stream = gst_mpdparser_get_active_stream_by_index(demux->client, i);
 750     caps = gst_dash_demux_get_input_caps(demux, active_stream);
 751     dash_stream->srcpad = gst_pad_new_from_static_template (&srctemplate, NULL);
 752     gst_pad_set_event_function (dash_stream->srcpad,
 753         GST_DEBUG_FUNCPTR (gst_dash_demux_src_event));
 754     gst_pad_set_query_function (dash_stream->srcpad,
 755         GST_DEBUG_FUNCPTR (gst_dash_demux_src_query));
 756     gst_pad_set_element_private (dash_stream->srcpad, demux);
 757     gst_pad_set_active (dash_stream->srcpad, TRUE);
 758     gst_pad_set_caps (dash_stream->srcpad, caps);
 759     gst_caps_unref(caps);
 760     gst_element_add_pad (GST_ELEMENT (demux), gst_object_ref (dash_stream->srcpad));
 761   }
 762   /* Send 'no-more-pads' to have decodebin create the new group */
 763   gst_element_no_more_pads (GST_ELEMENT (demux));
 764
 765   return TRUE;
 766 }
 767
 768 static gboolean
 769 gst_dash_demux_sink_event (GstPad * pad, GstEvent * event)
 770 {
 771   GstDashDemux *demux = GST_DASH_DEMUX (gst_pad_get_parent (pad));
 772
 773   switch (event->type) {
 774     case GST_EVENT_EOS:{
 775       gchar *manifest;
 776       GstQuery *query;
 777       gboolean res;
 778
 779       if (demux->manifest == NULL) {
 780         GST_WARNING_OBJECT (demux, "Received EOS without a manifest.");
 781         break;
 782       }
 783
 784       GST_DEBUG_OBJECT (demux, "Got EOS on the sink pad: manifest fetched");
 785
 786       if (demux->client)
 787         gst_mpd_client_free (demux->client);
 788       demux->client = gst_mpd_client_new ();
 789
 790       query = gst_query_new_uri ();
 791       res = gst_pad_peer_query (pad, query);
 792       if (res) {
 793         gst_query_parse_uri (query, &demux->client->mpd_uri);
 794         GST_DEBUG_OBJECT (demux, "Fetched MPD file at URI: %s",
 795             demux->client->mpd_uri);
 796       } else {
 797         GST_WARNING_OBJECT (demux, "MPD URI query failed.");
 798       }
 799       gst_query_unref (query);
 800
 801       manifest = (gchar *) GST_BUFFER_DATA (demux->manifest);
 802       if (manifest == NULL) {
 803         GST_WARNING_OBJECT (demux, "Error validating the manifest.");
 804       } else if (!gst_mpd_parse (demux->client, manifest,
 805               GST_BUFFER_SIZE (demux->manifest))) {
 806         /* In most cases, this will happen if we set a wrong url in the
 807          * source element and we have received the 404 HTML response instead of
 808          * the manifest */
 809         GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid manifest."),
 810             (NULL));
 811         return FALSE;
 812       }
 813
 814       gst_buffer_unref (demux->manifest);
 815       demux->manifest = NULL;
 816
 817       if (!gst_mpd_client_setup_media_presentation (demux->client)) {
 818         GST_ELEMENT_ERROR (demux, STREAM, DECODE,
 819             ("Incompatible manifest file."), (NULL));
 820         return FALSE;
 821       }
 822
 823       /* setup video, audio and subtitle streams, starting from first Period */
 824       if (!gst_mpd_client_set_period_index (demux->client, 0) ||
 825           !gst_dash_demux_setup_all_streams (demux))
 826         return FALSE;
 827
 828       /* start playing from the first segment */
 829       gst_mpd_client_set_segment_index_for_all_streams (demux->client, 0);
 830
 831       /* Send duration message */
 832       if (!gst_mpd_client_is_live (demux->client)) {
 833         GstClockTime duration =
 834             gst_mpd_client_get_media_presentation_duration (demux->client);
 835
 836         if (duration != GST_CLOCK_TIME_NONE) {
 837           GST_DEBUG_OBJECT (demux,
 838               "Sending duration message : %" GST_TIME_FORMAT,
 839               GST_TIME_ARGS (duration));
 840           gst_element_post_message (GST_ELEMENT (demux),
 841               gst_message_new_duration (GST_OBJECT (demux), GST_FORMAT_TIME,
 842                   duration));
 843         } else {
 844           GST_DEBUG_OBJECT (demux,
 845               "mediaPresentationDuration unknown, can not send the duration message");
 846         }
 847       }
 848       gst_dash_demux_resume_download_task (demux);
 849       gst_dash_demux_resume_stream_task (demux);
 850       gst_event_unref (event);
 851       return TRUE;
 852     }
 853     case GST_EVENT_NEWSEGMENT:
 854       /* Swallow newsegments, we'll push our own */
 855       gst_event_unref (event);
 856       return TRUE;
 857     default:
 858       break;
 859   }
 860
 861   return gst_pad_event_default (pad, event);
 862 }
 863
 864 static gboolean
 865 gst_dash_demux_src_query (GstPad * pad, GstQuery * query)
 866 {
 867   GstDashDemux *dashdemux;
 868   gboolean ret = FALSE;
 869
 870   if (query == NULL)
 871     return FALSE;
 872
 873   dashdemux = GST_DASH_DEMUX (gst_pad_get_element_private (pad));
 874
 875   switch (query->type) {
 876     case GST_QUERY_DURATION:{
 877       GstClockTime duration = -1;
 878       GstFormat fmt;
 879
 880       gst_query_parse_duration (query, &fmt, NULL);
 881       if (fmt == GST_FORMAT_TIME) {
 882         duration =
 883             gst_mpd_client_get_media_presentation_duration (dashdemux->client);
 884         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) {
 885           gst_query_set_duration (query, GST_FORMAT_TIME, duration);
 886           ret = TRUE;
 887         }
 888       }
 889       GST_DEBUG_OBJECT (dashdemux,
 890           "GST_QUERY_DURATION returns %s with duration %" GST_TIME_FORMAT,
 891           ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
 892       break;
 893     }
 894     case GST_QUERY_SEEKING:{
 895       GstFormat fmt;
 896       gint64 stop = -1;
 897
 898       gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL);
 899       GST_DEBUG_OBJECT (dashdemux, "Received GST_QUERY_SEEKING with format %d",
 900           fmt);
 901       if (fmt == GST_FORMAT_TIME) {
 902         GstClockTime duration;
 903
 904         duration =
 905             gst_mpd_client_get_media_presentation_duration (dashdemux->client);
 906         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0)
 907           stop = duration;
 908
 909         gst_query_set_seeking (query, fmt,
 910             !gst_mpd_client_is_live (dashdemux->client), 0, stop);
 911         ret = TRUE;
 912         GST_DEBUG_OBJECT (dashdemux, "GST_QUERY_SEEKING returning with stop : %"
 913             GST_TIME_FORMAT, GST_TIME_ARGS (stop));
 914       }
 915       break;
 916     }
 917     case GST_QUERY_URI: {
 918       /* forwarding uri */
 919       GST_DEBUG("URI query recevied in DASH demux.....");
 920         gboolean res;
 921         res = gst_pad_query_default (pad,query);
 922         if(res)
 923           GST_DEBUG("forwarding URI is done successfully!!...");
 924       ret = TRUE;
 925       break;
 926     }
 927     default:{
 928       // By default, do not forward queries upstream
 929       break;
 930     }
 931   }
 932
 933   return ret;
 934 }
 935
 936 static GstFlowReturn
 937 gst_dash_demux_pad (GstPad * pad, GstBuffer * buf)
 938 {
 939   GstDashDemux *demux = GST_DASH_DEMUX (gst_pad_get_parent (pad));
 940
 941   if (demux->manifest == NULL)
 942     demux->manifest = buf;
 943   else
 944     demux->manifest = gst_buffer_join (demux->manifest, buf);
 945
 946   gst_object_unref (demux);
 947
 948   return GST_FLOW_OK;
 949 }
 950
 951 static void
 952 gst_dash_demux_stop (GstDashDemux * demux)
 953 {
 954   gst_uri_downloader_cancel (demux->downloader);
 955   gst_dash_demux_flush_stream_queues (demux);
 956
 957   if (GST_TASK_STATE (demux->download_task) != GST_TASK_STOPPED) {
 958     GST_TASK_SIGNAL (demux->download_task);
 959     gst_task_stop (demux->download_task);
 960     g_mutex_lock (&demux->download_mutex);
 961     g_cond_signal (&demux->download_cond);
 962     g_mutex_unlock (&demux->download_mutex);
 963   }
 964   if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED) {
 965     GST_TASK_SIGNAL (demux->stream_task);
 966     gst_task_stop (demux->stream_task);
 967   }
 968 }
 969
 970 /* gst_dash_demux_stream_loop:
 971  *
 972  * Loop for the "stream' task that pushes fragments to the src pads.
 973  *
 974  * Startup:
 975  * The task is started as soon as we have received the manifest and
 976  * waits for the first fragment to be downloaded and pushed in the
 977  * queue. Once this fragment has been pushed, the task pauses itself
 978  * until actual playback begins.
 979  *
 980  * During playback:
 981  * The task pushes fragments downstream at regular intervals based on
 982  * the fragment duration. If it detects a queue underrun, it sends
 983  * a buffering event to tell the main application to pause.
 984  *
 985  * Teardown:
 986  * The task is stopped when we have reached the end of the manifest
 987  * and emptied our queue.
 988  *
 989  */
 990 static void
 991 gst_dash_demux_stream_loop (GstDashDemux * demux)
 992 {
 993   GstFlowReturn ret;
 994   GstActiveStream *active_stream;
 995   GstDashDemuxStream *selected_stream = NULL;
 996   GstClockTime min_ts = GST_CLOCK_TIME_NONE;
 997   guint i = 0;
 998   gboolean eos = TRUE;
 999   gboolean eop = TRUE;
1000
1001   for (i = 0; i < g_slist_length (demux->streams); i++) {
1002     GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, i);
1003     GstBuffer *buffer;
1004     GstDataQueueItem *item;
1005
1006     if (dash_stream->stream_eos) {
1007       GST_DEBUG_OBJECT (demux, "Stream %d is eos, skipping", dash_stream->idx);
1008       continue;
1009     }
1010
1011     if (dash_stream->stream_end_of_period) {
1012       GST_DEBUG_OBJECT (demux, "Stream %d is eop, skipping", dash_stream->idx);
1013       eos = FALSE;
1014       continue;
1015     }
1016     eos = FALSE;
1017     eop = FALSE;
1018
1019     if (!gst_data_queue_peek (dash_stream->queue, &item))
1020       goto flushing;
1021
1022     if(GST_IS_BUFFER(item->object)) {
1023       buffer = GST_BUFFER(item->object);
1024       if(GST_BUFFER_TIMESTAMP(buffer) < min_ts ||
1025          !GST_CLOCK_TIME_IS_VALID(min_ts)) {
1026         min_ts = GST_BUFFER_TIMESTAMP(buffer);
1027         selected_stream = dash_stream;
1028       } else if (!GST_CLOCK_TIME_IS_VALID (GST_BUFFER_TIMESTAMP (item->object))) {
1029         selected_stream = dash_stream;
1030         break;
1031       }
1032     } else {
1033       selected_stream = dash_stream;
1034       break;
1035     }
1036   }
1037
1038   if(selected_stream) {
1039     GstBuffer *buffer;
1040     GstDataQueueItem *item;
1041
1042     if (!gst_data_queue_pop (selected_stream->queue, &item))
1043       goto end;
1044     if ( GST_IS_BUFFER (item->object)) {
1045       buffer = GST_BUFFER(item->object);
1046       active_stream = gst_mpdparser_get_active_stream_by_index (demux->client, selected_stream->idx);
1047
1048       if (selected_stream->need_segment) {
1049         if(!GST_CLOCK_TIME_IS_VALID (selected_stream->start_time)) {
1050           if(GST_CLOCK_TIME_IS_VALID (GST_BUFFER_TIMESTAMP (buffer))){
1051             selected_stream->start_time = GST_BUFFER_TIMESTAMP (buffer);
1052           } else {
1053             selected_stream->start_time = 0;
1054           }
1055         }
1056         /* And send a newsegment */
1057         GST_DEBUG_OBJECT (demux, "Sending new-segment stream #%d. segment start:%"
1058             GST_TIME_FORMAT, selected_stream->idx, GST_TIME_ARGS (selected_stream->start_time));
1059         gst_pad_push_event (selected_stream->srcpad,
1060             gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME,
1061                 selected_stream->start_time, GST_CLOCK_TIME_NONE, selected_stream->start_time));
1062         selected_stream->need_segment = FALSE;
1063       }
1064
1065       GST_DEBUG_OBJECT (demux, "Pushing fragment #%llu (stream %d) ts=%"GST_TIME_FORMAT, GST_BUFFER_OFFSET (buffer),
1066                         selected_stream->idx, GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)));
1067       ret = gst_pad_push (selected_stream->srcpad, gst_buffer_ref(buffer) );
1068       item->destroy (item);
1069       if ((ret != GST_FLOW_OK) && (active_stream->mimeType == GST_STREAM_VIDEO))
1070         goto error_pushing;
1071     } else {
1072       GstEvent *event = GST_EVENT (item->object);
1073       if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
1074         selected_stream->stream_eos = TRUE;
1075         selected_stream->stream_end_of_period = TRUE;
1076       } else if (GST_EVENT_TYPE (event) == GST_EVENT_DASH_EOP) {
1077         selected_stream->stream_end_of_period = TRUE;
1078       }
1079
1080       if (GST_EVENT_TYPE (item->object) != GST_EVENT_DASH_EOP) {
1081         gst_pad_push_event (selected_stream->srcpad,
1082             gst_event_ref (GST_EVENT_CAST (item->object)));
1083       }
1084
1085       item->destroy (item);
1086     }
1087   } else {
1088     if (eos) {
1089       goto end_of_manifest;
1090     } else if (eop) {
1091       /*TODO Switch to next period*/
1092     }
1093   }
1094
1095 end:
1096   return;
1097
1098 flushing:
1099   {
1100     GST_INFO_OBJECT (demux, "Queue is flushing. Stopped streaming task");
1101     gst_task_stop (demux->stream_task);
1102     return;
1103   }
1104
1105 end_of_manifest:
1106   {
1107     GST_INFO_OBJECT (demux, "Reached end of manifest, sending EOS");
1108     guint i = 0;
1109     for (i = 0; i < gst_mpdparser_get_nb_active_stream (demux->client); i++) {
1110       GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, i);
1111       gst_pad_push_event (dash_stream->srcpad, gst_event_new_eos ());
1112     }
1113     GST_INFO_OBJECT (demux, "Stopped streaming task");
1114     gst_task_stop (demux->stream_task);
1115     return;
1116   }
1117
1118 error_pushing:
1119   {
1120     /* FIXME: handle error */
1121     GST_ERROR_OBJECT (demux,
1122         "Error pushing buffer: %s... terminating the demux",
1123         gst_flow_get_name (ret));
1124     gst_dash_demux_stop (demux);
1125     return;
1126   }
1127 }
1128
1129 static void
1130 gst_dash_demux_clear_streams(GstDashDemux * demux) {
1131   guint i = 0;
1132   gst_dash_demux_flush_stream_queues (demux);
1133   for (i = 0; i < g_slist_length(demux->streams); i++) {
1134     GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, i);
1135     gst_download_rate_deinit (&dash_stream->dnl_rate);
1136     if (dash_stream->input_caps) {
1137       gst_caps_unref (dash_stream->input_caps);
1138       dash_stream->input_caps = NULL;
1139     }
1140     if (dash_stream->srcpad) {
1141       gst_object_unref (dash_stream->srcpad);
1142       dash_stream->srcpad = NULL;
1143     }
1144     /*TODO consider unref stream->output_caps*/
1145     g_object_unref (dash_stream->queue);
1146   }
1147   if(demux->streams) {
1148     g_slist_free(demux->streams);
1149     demux->streams = NULL;
1150   }
1151 }
1152
1153 static void
1154 gst_dash_demux_reset (GstDashDemux * demux, gboolean dispose)
1155 {
1156   gint stream_idx;
1157   demux->end_of_period = FALSE;
1158   demux->cancelled = FALSE;
1159
1160   gst_dash_demux_clear_streams(demux);
1161
1162   if (demux->manifest) {
1163     gst_buffer_unref (demux->manifest);
1164     demux->manifest = NULL;
1165   }
1166   if (demux->client) {
1167     gst_mpd_client_free (demux->client);
1168     demux->client = NULL;
1169   }
1170   if (!dispose) {
1171     demux->client = gst_mpd_client_new ();
1172   }
1173
1174   demux->last_manifest_update = GST_CLOCK_TIME_NONE;
1175   for (stream_idx = 0; stream_idx < g_slist_length (demux->streams); stream_idx++) {
1176     GstDashDemuxStream *dash_stream = g_slist_nth_data (demux->streams, stream_idx);
1177     dash_stream->need_segment = TRUE;
1178   }
1179 }
1180
1181 static GstClockTime
1182 gst_dash_demux_get_buffering_time (GstDashDemux * demux)
1183 {
1184   GstClockTime buffer_time = 0;
1185   GSList *it;
1186   GstDashDemuxStream *stream;
1187   GstDataQueueSize queue_size;
1188
1189   for(it=demux->streams; it; it=it->next) {
1190     stream = it->data;
1191     gst_data_queue_get_level(stream->queue, &queue_size);
1192
1193     if (queue_size.time > 0) {
1194       buffer_time = queue_size.time;
1195       break;
1196     }
1197   }
1198
1199   return buffer_time;
1200 }
1201
1202 static gboolean
1203 gst_dash_demux_update_manifest(GstDashDemux *demux) {
1204   GstFragment *download;
1205   GstBuffer *buffer;
1206   GstClockTime duration, now = gst_util_get_timestamp();
1207   gint64 update_period = demux->client->mpd_node->minimumUpdatePeriod;
1208
1209   if (update_period == -1) {
1210     GST_DEBUG_OBJECT (demux, "minimumUpdatePeriod unspecified, will not update MPD");
1211     return TRUE;
1212   }
1213
1214   /* init reference time for manifest file updates */
1215   if (!GST_CLOCK_TIME_IS_VALID (demux->last_manifest_update))
1216     demux->last_manifest_update = now;
1217
1218   /* update the manifest file */
1219   if (now >= demux->last_manifest_update + update_period * GST_MSECOND) {
1220     GST_DEBUG_OBJECT (demux, "Updating manifest file from URL %s",
1221         demux->client->mpd_uri);
1222     download =
1223         gst_uri_downloader_fetch_uri (demux->downloader,
1224         demux->client->mpd_uri);
1225     if (download == NULL) {
1226       GST_WARNING_OBJECT (demux,
1227           "Failed to update the manifest file from URL %s",
1228           demux->client->mpd_uri);
1229     } else {
1230       GstMpdClient *new_client = NULL;
1231       guint period_idx;
1232       const gchar *period_id;
1233       GSList *iter;
1234
1235       buffer = gst_fragment_get_buffer(download);
1236       g_object_unref (download);
1237       /* parse the manifest file */
1238       if (buffer == NULL) {
1239         GST_WARNING_OBJECT (demux, "Error validating the manifest.");
1240         return TRUE;
1241       }
1242
1243       new_client = gst_mpd_client_new ();
1244       new_client->mpd_uri = g_strdup (demux->client->mpd_uri);
1245       if (!gst_mpd_parse (new_client,
1246               (gchar *) GST_BUFFER_DATA (buffer), GST_BUFFER_SIZE (buffer))) {
1247         /* In most cases, this will happen if we set a wrong url in the
1248          * source element and we have received the 404 HTML response instead of
1249          * the manifest */
1250         GST_WARNING_OBJECT (demux, "Error parsing the manifest.");
1251         gst_buffer_unref (buffer);
1252         return TRUE;
1253       }
1254
1255       gst_buffer_unref (buffer);
1256       GST_DEBUG_OBJECT (demux, "Updating manifest");
1257
1258       period_id = gst_mpd_client_get_period_id (demux->client);
1259       period_idx = gst_mpd_client_get_period_index (demux->client);
1260
1261       /* setup video, audio and subtitle streams, starting from current Period */
1262       if (!gst_mpd_client_setup_media_presentation (new_client)) {
1263         /* TODO */
1264       }
1265
1266       if (period_idx) {
1267         /*If more than one period exists.*/
1268         if (!gst_mpd_client_set_period_id (new_client, period_id)) {
1269           GST_DEBUG_OBJECT (demux,
1270               "Error setting up the updated manifest file");
1271           return FALSE;
1272         }
1273       } else {
1274         if (!gst_mpd_client_set_period_index (new_client, period_idx)) {
1275           GST_DEBUG_OBJECT (demux,
1276               "Error setting up the updated manifest file");
1277           return FALSE;
1278         }
1279       }
1280
1281       if (!gst_dash_demux_setup_mpdparser_streams (demux, new_client)) {
1282             GST_ERROR_OBJECT (demux, "Failed to setup streams on manifest "
1283                 "update");
1284             return FALSE;
1285       }
1286
1287       /* update the streams to play from the next segment */
1288       for (iter = demux->streams; iter; iter = g_slist_next (iter)) {
1289         GstDashDemuxStream *demux_stream = iter->data;
1290         GstActiveStream *new_stream;
1291         GstClockTime ts;
1292
1293         new_stream = gst_mpdparser_get_active_stream_by_index (new_client,
1294             demux_stream->idx);
1295
1296         if (!new_stream) {
1297           GST_DEBUG_OBJECT (demux,
1298               "Stream of index %d is missing from manifest update",
1299               demux_stream->idx);
1300           return FALSE;
1301         }
1302
1303         if (gst_mpd_client_get_next_fragment_timestamp (demux->client,
1304                 demux_stream->idx, &ts)) {
1305           gst_mpd_client_stream_seek (new_client, demux_stream->idx, ts);
1306
1307         } else
1308             if (gst_mpd_client_get_last_fragment_timestamp (demux->client,
1309                 demux_stream->idx, &ts)) {
1310           /* try to set to the old timestamp + 1 */
1311           gst_mpd_client_stream_seek (new_client, demux_stream->idx, ts+1);
1312         }
1313       }
1314
1315       /*Remember download failed count*/
1316       new_client->download_failed_count = demux->client->download_failed_count;
1317
1318       gst_mpd_client_free (demux->client);
1319       demux->client = new_client;
1320
1321       /* Send an updated duration message */
1322       duration =
1323           gst_mpd_client_get_media_presentation_duration (demux->client);
1324
1325       if (duration != GST_CLOCK_TIME_NONE) {
1326         GST_DEBUG_OBJECT (demux,
1327             "Sending duration message : %" GST_TIME_FORMAT,
1328             GST_TIME_ARGS (duration));
1329         gst_element_post_message (GST_ELEMENT (demux),
1330             gst_message_new_duration(GST_OBJECT (demux), GST_FORMAT_TIME, duration));
1331       } else {
1332         GST_DEBUG_OBJECT (demux,
1333             "mediaPresentationDuration unknown, can not send the duration message");
1334       }
1335       demux->last_manifest_update = gst_util_get_timestamp ();
1336       GST_DEBUG_OBJECT (demux, "Manifest file successfully updated");
1337     }
1338   }
1339   return TRUE;
1340 }
1341
1342 static void
1343 gst_dash_demux_download_wait (GstDashDemux * demux, GstClockTime time_diff)
1344 {
1345   gint64 end_time = g_get_monotonic_time () + time_diff / GST_USECOND;
1346
1347   GST_DEBUG_OBJECT (demux, "Download waiting for %" GST_TIME_FORMAT,
1348       GST_TIME_ARGS (time_diff));
1349   g_cond_wait_until (&demux->download_cond, &demux->download_mutex, end_time);
1350   GST_DEBUG_OBJECT (demux, "Download finished waiting");
1351 }
1352
1353 static void
1354 gst_dash_demux_check_live(GstDashDemux* demux, GstActiveStream *fragment_stream,
1355      GstClockTime fragment_ts)
1356 {
1357   gint64 time_diff;
1358   gint pos;
1359
1360   pos =
1361       gst_mpd_client_check_time_position (demux->client, fragment_stream,
1362       fragment_ts, &time_diff);
1363   GST_DEBUG_OBJECT (demux,
1364       "Checked position for fragment ts %" GST_TIME_FORMAT
1365       ", res: %d, diff: %" G_GINT64_FORMAT, GST_TIME_ARGS (fragment_ts),
1366       pos, time_diff);
1367
1368   time_diff *= GST_USECOND;
1369   if (pos < 0) {
1370     /* we're behind, try moving to the 'present' */
1371     GDateTime *now = g_date_time_new_now_utc ();
1372
1373     GST_DEBUG_OBJECT (demux,
1374         "Falling behind live stream, moving forward");
1375       gst_mpd_client_seek_to_time(demux->client, now);
1376     g_date_time_unref (now);
1377
1378   } else if (pos > 0) {
1379     /* we're ahead, wait a little */
1380
1381     GST_DEBUG_OBJECT (demux, "Waiting for next segment to be created");
1382     gst_dash_demux_download_wait (demux, time_diff);
1383   } else {
1384     demux->client->download_failed_count++;
1385   }
1386 }
1387
1388 /* gst_dash_demux_download_loop:
1389  *
1390  * Loop for the "download' task that fetches fragments based on the
1391  * selected representations.
1392  *
1393  * Startup:
1394  *
1395  * The task is started from the stream loop.
1396  *
1397  * During playback:
1398  *
1399  * It sequentially fetches fragments corresponding to the current
1400  * representations and pushes them into a queue.
1401  *
1402  * It tries to maintain the number of queued items within a predefined
1403  * range: if the queue is full, it will pause, checking every 100 ms if
1404  * it needs to restart downloading fragments.
1405  *
1406  * When a new set of fragments has been downloaded, it evaluates the
1407  * download time to check if we can or should switch to a different
1408  * set of representations.
1409  *
1410  * Teardown:
1411  *
1412  * The task will exit when it encounters an error or when the end of the
1413  * manifest has been reached.
1414  *
1415  */
1416 void
1417 gst_dash_demux_download_loop (GstDashDemux * demux)
1418 {
1419   GstActiveStream *fragment_stream = NULL;
1420   GstClockTime fragment_ts;
1421   if ( gst_mpd_client_is_live (demux->client) && demux->client->mpd_uri != NULL ) {
1422     if (!gst_dash_demux_update_manifest(demux))
1423       goto end_of_manifest;
1424   }
1425
1426   /* try to switch to another set of representations if needed */
1427   gst_dash_demux_select_representations (demux);
1428
1429   /* fetch the next fragment */
1430   while (!gst_dash_demux_get_next_fragment (demux, &fragment_stream, &fragment_ts)) {
1431     if (demux->end_of_period) {
1432       GST_INFO_OBJECT (demux, "Reached the end of the Period");
1433       /* setup video, audio and subtitle streams, starting from the next Period */
1434       if (!gst_mpd_client_set_period_index (demux->client,
1435               gst_mpd_client_get_period_index (demux->client) + 1)
1436           || !gst_dash_demux_setup_all_streams (demux)) {
1437         GST_INFO_OBJECT (demux, "Reached the end of the manifest file");
1438         gst_task_start (demux->stream_task);
1439         goto end_of_manifest;
1440       }
1441       /* start playing from the first segment of the new period */
1442       gst_mpd_client_set_segment_index_for_all_streams (demux->client, 0);
1443       demux->end_of_period = FALSE;
1444     } else if (!demux->cancelled) {
1445       gst_uri_downloader_reset (demux->downloader);
1446       if(gst_mpd_client_is_live (demux->client)) {
1447         gst_dash_demux_check_live (demux, fragment_stream, fragment_ts);
1448       } else {
1449         demux->client->download_failed_count++;
1450       }
1451
1452       if (demux->client->download_failed_count < DEFAULT_FAILED_COUNT) {
1453         GST_WARNING_OBJECT (demux, "Could not fetch the next fragment");
1454         goto quit;
1455       } else if (gst_mpd_client_set_next_baseURL_for_stream (demux->client)) {
1456         GST_INFO_OBJECT (demux, "Current baseURL is %s. Trying to select another",
1457                          gst_mpdparser_get_baseURL (demux->client));
1458         demux->client->download_failed_count = 0;
1459       } else {
1460         goto error_downloading;
1461       }
1462     } else {
1463       goto quit;
1464     }
1465   }
1466   GST_INFO_OBJECT (demux, "Internal buffering : %" PRIu64 " s",
1467       gst_dash_demux_get_buffering_time (demux) / GST_SECOND);
1468   demux->client->download_failed_count = 0;
1469
1470
1471 quit:
1472   {
1473     return;
1474   }
1475
1476 end_of_manifest:
1477   {
1478     GST_INFO_OBJECT (demux, "Stopped download task");
1479     gst_task_stop (demux->download_task);
1480     return;
1481   }
1482
1483 error_downloading:
1484   {
1485     GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
1486         ("Could not fetch the next fragment"), (NULL));
1487     gst_dash_demux_stop (demux);
1488     return;
1489   }
1490 }
1491
1492 static void
1493 gst_dash_demux_resume_stream_task (GstDashDemux * demux)
1494 {
1495   gst_task_start (demux->stream_task);
1496 }
1497
1498 static void
1499 gst_dash_demux_resume_download_task (GstDashDemux * demux)
1500 {
1501   gst_uri_downloader_reset(demux->downloader);
1502   gst_task_start (demux->download_task);
1503 }
1504
1505 /* gst_dash_demux_select_representations:
1506  *
1507  * Select the most appropriate media representations based on a target
1508  * bitrate.
1509  *
1510  * FIXME: all representations are selected against the same bitrate, but
1511  * they will share the same bandwidth. This only works today because the
1512  * audio representations bitrate usage is negligible as compared to the
1513  * video representation one.
1514  *
1515  * Returns TRUE if a new set of representations has been selected
1516  */
1517 static gboolean
1518 gst_dash_demux_select_representations (GstDashDemux * demux)
1519 {
1520   GstDashDemuxStream *stream = NULL;
1521   GstActiveStream *active_stream = NULL;
1522   GList *rep_list = NULL;
1523   GSList *iter;
1524   gint new_index;
1525   gboolean ret = FALSE;
1526
1527   GST_MPD_CLIENT_LOCK (demux->client);
1528   for (iter = demux->streams; iter; iter = g_slist_next (iter)) {
1529     guint64 bitrate;
1530
1531     stream = iter->data;
1532     active_stream =
1533       gst_mpdparser_get_active_stream_by_index (demux->client, stream->idx);
1534     if (!active_stream)
1535       return FALSE;
1536
1537     /* retrieve representation list */
1538     if (active_stream->cur_adapt_set)
1539       rep_list = active_stream->cur_adapt_set->Representations;
1540     if (!rep_list)
1541       return FALSE;
1542
1543     bitrate = gst_download_rate_get_current_rate (&stream->dnl_rate) *
1544       demux->bandwidth_usage;
1545
1546     GST_DEBUG_OBJECT (demux, "Trying to change bitrate to %" G_GUINT64_FORMAT, bitrate);
1547
1548     /* get representation index with current max_bandwidth */
1549     new_index =
1550         gst_mpdparser_get_rep_idx_with_max_bandwidth (rep_list, bitrate);
1551
1552     /* if no representation has the required bandwidth, take the lowest one */
1553     if (new_index == -1)
1554       new_index = 0;
1555
1556     if (new_index != active_stream->representation_idx) {
1557       GstRepresentationNode *rep = g_list_nth_data (rep_list, new_index);
1558       GST_INFO_OBJECT (demux, "Changing representation idx: %d %d %u",
1559                        stream->idx, new_index, rep->bandwidth);
1560       if (gst_mpd_client_setup_representation (demux->client, active_stream,
1561                                                rep)) {
1562         ret = TRUE;
1563         stream->need_header = TRUE;
1564         GST_INFO_OBJECT (demux, "Switching bitrate to %d",
1565             active_stream->cur_representation->bandwidth);
1566       } else {
1567         GST_WARNING_OBJECT (demux,
1568             "Can not switch representation, aborting...");
1569       }
1570     }
1571   }
1572   GST_MPD_CLIENT_UNLOCK (demux->client);
1573   return ret;
1574 }
1575
1576 static GstFragment *
1577 gst_dash_demux_get_next_header (GstDashDemux * demux, guint stream_idx)
1578 {
1579   const gchar *initializationURL;
1580   gchar *next_header_uri;
1581   GstFragment *fragment;
1582
1583   if (!gst_mpd_client_get_next_header (demux->client, &initializationURL,
1584           stream_idx))
1585     return NULL;
1586
1587   if (strncmp (initializationURL, "http://", 7) != 0) {
1588     next_header_uri =
1589         g_strconcat (gst_mpdparser_get_baseURL (demux->client),
1590         initializationURL, NULL);
1591   } else {
1592     next_header_uri = g_strdup (initializationURL);
1593   }
1594
1595   GST_INFO_OBJECT (demux, "Fetching header %s", next_header_uri);
1596
1597   fragment = gst_uri_downloader_fetch_uri (demux->downloader, next_header_uri);
1598   g_free (next_header_uri);
1599   g_free (initializationURL);
1600
1601   return fragment;
1602 }
1603
1604 static GstCaps *
1605 gst_dash_demux_get_video_input_caps (GstDashDemux * demux,
1606     GstActiveStream * stream)
1607 {
1608   guint width = 0, height = 0, bandwidth = 0;
1609   const gchar *mimeType = NULL;
1610   GstCaps *caps = NULL;
1611
1612   if (stream == NULL)
1613     return NULL;
1614 #ifdef DASHDEMUX_MODIFICATION
1615   /* caps need to inlcude resolution and bandwidth */
1616     width = gst_mpd_client_get_video_stream_width (stream);
1617     height = gst_mpd_client_get_video_stream_height (stream);
1618     bandwidth = gst_mpd_client_get_video_stream_bandwidth (stream);
1619 #else
1620   /* if bitstreamSwitching is true we dont need to swich pads on resolution change */
1621   if (!gst_mpd_client_get_bitstream_switching_flag (stream)) {
1622     width = gst_mpd_client_get_video_stream_width (stream);
1623     height = gst_mpd_client_get_video_stream_height (stream);
1624   }
1625 #endif
1626   mimeType = gst_mpd_client_get_stream_mimeType (stream);
1627   if (mimeType == NULL)
1628     return NULL;
1629
1630   caps = gst_caps_new_simple (mimeType, NULL);
1631   if (width > 0 && height > 0) {
1632     gst_caps_set_simple (caps, "width", G_TYPE_INT, width, "height",
1633         G_TYPE_INT, height, NULL);
1634   }
1635
1636   if (bandwidth > 0)
1637     gst_caps_set_simple (caps, "bandwidth", G_TYPE_INT, bandwidth, NULL);
1638
1639   gst_caps_set_simple (caps, "max-width", G_TYPE_INT, demux->max_video_width, "max-height",
1640       G_TYPE_INT, demux->max_video_height, NULL);
1641
1642 /*  add ContentProtection to caps  */
1643   if ( stream->cur_adapt_set->RepresentationBase->ContentProtection != NULL){
1644     GList *list;
1645     GstDescriptorType *ContentProtectionDesc;
1646
1647     list = g_list_first (stream->cur_adapt_set->RepresentationBase->ContentProtection);
1648     ContentProtectionDesc = (GstDescriptorType *) list->data;
1649     gchar *schemeIdUri = ContentProtectionDesc->schemeIdUri;
1650     gchar *msprPro = ContentProtectionDesc->msprPro;
1651
1652   if ( (schemeIdUri != NULL) && (msprPro != NULL) ){
1653     gst_caps_set_simple (caps, "contentprotection_scheme_iduri", G_TYPE_STRING, schemeIdUri, "mspr:pro", G_TYPE_STRING, msprPro, NULL);
1654   }
1655  }
1656   return caps;
1657 }
1658
1659 static GstCaps *
1660 gst_dash_demux_get_audio_input_caps (GstDashDemux * demux,
1661     GstActiveStream * stream)
1662 {
1663   guint rate = 0, channels = 0;
1664   const gchar *mimeType;
1665   GstCaps *caps = NULL;
1666
1667   if (stream == NULL)
1668     return NULL;
1669
1670   /* if bitstreamSwitching is true we dont need to swich pads on rate/channels change */
1671   if (!gst_mpd_client_get_bitstream_switching_flag (stream)) {
1672     channels = gst_mpd_client_get_audio_stream_num_channels (stream);
1673     rate = gst_mpd_client_get_audio_stream_rate (stream);
1674   }
1675   mimeType = gst_mpd_client_get_stream_mimeType (stream);
1676   if (mimeType == NULL)
1677     return NULL;
1678
1679   caps = gst_caps_new_simple (mimeType, NULL);
1680   if (rate > 0) {
1681     gst_caps_set_simple (caps, "rate", G_TYPE_INT, rate, NULL);
1682   }
1683   if (channels > 0) {
1684     gst_caps_set_simple (caps, "channels", G_TYPE_INT, channels, NULL);
1685   }
1686
1687 /*  add ContentProtection to caps  */
1688   if ( stream->cur_adapt_set->RepresentationBase->ContentProtection != NULL){
1689     GList *list;
1690     GstDescriptorType *ContentProtectionDesc;
1691
1692     list = g_list_first (stream->cur_adapt_set->RepresentationBase->ContentProtection);
1693     ContentProtectionDesc = (GstDescriptorType *) list->data;
1694     gchar *schemeIdUri = ContentProtectionDesc->schemeIdUri;
1695     gchar *msprPro = ContentProtectionDesc->msprPro;
1696
1697   if ( (schemeIdUri != NULL) && (msprPro != NULL) ){
1698     gst_caps_set_simple (caps, "contentprotection_scheme_iduri", G_TYPE_STRING, schemeIdUri, "mspr:pro", G_TYPE_STRING, msprPro, NULL);
1699   }
1700  }
1701   return caps;
1702 }
1703
1704 static GstCaps *
1705 gst_dash_demux_get_application_input_caps (GstDashDemux * demux,
1706     GstActiveStream * stream)
1707 {
1708   const gchar *mimeType;
1709   GstCaps *caps = NULL;
1710
1711   if (stream == NULL)
1712     return NULL;
1713
1714   mimeType = gst_mpd_client_get_stream_mimeType (stream);
1715   if (mimeType == NULL)
1716     return NULL;
1717
1718   caps = gst_caps_new_simple (mimeType, NULL);
1719
1720   return caps;
1721 }
1722
1723 static GstCaps *
1724 gst_dash_demux_get_input_caps (GstDashDemux * demux, GstActiveStream * stream)
1725 {
1726   GstCaps *caps;
1727   switch (stream->mimeType) {
1728     case GST_STREAM_VIDEO:
1729       caps = gst_dash_demux_get_video_input_caps (demux, stream);
1730       break;
1731     case GST_STREAM_AUDIO:
1732       caps = gst_dash_demux_get_audio_input_caps (demux, stream);
1733       break;
1734     case GST_STREAM_APPLICATION:
1735       caps = gst_dash_demux_get_application_input_caps (demux, stream);
1736       break;
1737     default:
1738       return GST_CAPS_NONE;
1739   }
1740   /*Need to signal downstream elements about dash*/
1741   gst_caps_set_simple(caps, "variant", G_TYPE_STRING, "dash-fragmented", NULL);
1742   return caps;
1743 }
1744
1745 /* gst_dash_demux_get_next_fragment_set:
1746  *
1747  * Get the next set of fragments for the current representations.
1748  *
1749  * This function uses the generic URI downloader API.
1750  *
1751  * Returns FALSE if an error occured while downloading fragments
1752  *
1753  */
1754 static gboolean
1755 gst_dash_demux_get_next_fragment (GstDashDemux * demux,GstActiveStream **fragment_stream,
1756      GstClockTime *selected_ts)
1757 {
1758   GstActiveStream *stream;
1759   GstDashDemuxStream *dash_stream;
1760   GstDashDemuxStream *selected_stream = NULL;
1761   GstFragment *download, *header;
1762   gchar *next_fragment_uri;
1763   GstClockTime duration;
1764   GstClockTime timestamp;
1765   GstClockTime min_timestamp = GST_CLOCK_TIME_NONE;
1766   gboolean discont;
1767   GTimeVal now;
1768   GTimeVal start;
1769   GstClockTime diff;
1770   guint64 size_buffer = 0;
1771   GstBuffer *buffer;
1772   guint stream_idx;
1773   gboolean end_of_period = TRUE;
1774
1775   /*Select stream with smallest progress*/
1776   for (stream_idx = 0; stream_idx < g_slist_length (demux->streams); stream_idx++) {
1777     dash_stream = g_slist_nth_data (demux->streams, stream_idx);
1778
1779     if (dash_stream->download_end_of_period)
1780       continue;
1781
1782     if (gst_mpd_client_get_next_fragment_timestamp (demux->client, stream_idx, &timestamp)) {
1783       if( timestamp < min_timestamp || !GST_CLOCK_TIME_IS_VALID(min_timestamp) ) {
1784         selected_stream = dash_stream;
1785         min_timestamp = timestamp;
1786       }
1787     } else {
1788       GstEvent *event = NULL;
1789
1790       GST_INFO_OBJECT (demux,
1791           "This Period doesn't contain more fragments for stream %u",
1792           dash_stream->idx);
1793
1794       /* check if this is live and we should wait for more data */
1795       if (gst_mpd_client_is_live (demux->client)
1796           && demux->client->mpd_node->minimumUpdatePeriod != -1) {
1797         end_of_period = FALSE;
1798         continue;
1799       }
1800
1801       if (gst_mpd_client_has_next_period (demux->client)) {
1802         event = gst_event_new_dash_eop ();
1803       } else {
1804         GST_DEBUG_OBJECT (demux,
1805             "No more fragments or periods for this stream, setting EOS");
1806         event = gst_event_new_eos ();
1807       }
1808       dash_stream->download_end_of_period = TRUE;
1809       gst_dash_demux_stream_push_event (dash_stream, event);
1810     }
1811   }
1812
1813   if (selected_ts)
1814     *selected_ts = min_timestamp;
1815   if (fragment_stream && selected_stream)
1816     *fragment_stream = gst_mpdparser_get_active_stream_by_index (demux->client, selected_stream->idx);
1817    /* Fetch next fragment from selected stream */
1818   if(selected_stream) {
1819
1820     if (!gst_mpd_client_get_next_fragment (demux->client,
1821             selected_stream->idx, &discont, &next_fragment_uri, &duration, &timestamp)) {
1822       GST_WARNING_OBJECT (demux, "Failed to download fragment for stream %d", selected_stream->idx);
1823     } else {
1824
1825       g_get_current_time (&start);
1826       GST_INFO_OBJECT (demux, "Fetching next fragment stream=%d ts=%"GST_TIME_FORMAT" url=%s",
1827                        selected_stream->idx, GST_TIME_ARGS(timestamp), next_fragment_uri);
1828
1829       stream = gst_mpdparser_get_active_stream_by_index (demux->client, selected_stream->idx);
1830
1831       end_of_period = FALSE;
1832
1833       download = gst_uri_downloader_fetch_uri (demux->downloader,
1834           next_fragment_uri);
1835       g_free (next_fragment_uri);
1836
1837       if (stream == NULL)
1838         return FALSE;
1839
1840       if (download == NULL) {
1841         guint segment_idx = gst_mpd_client_get_segment_index ( stream );
1842         if(segment_idx > 0) {
1843           /*Move to previous segment if download failed*/
1844           gst_mpd_client_set_segment_index (stream, segment_idx - 1);
1845         }
1846         return FALSE;
1847       }
1848
1849       download->start_time = timestamp;
1850       download->stop_time = timestamp + duration;
1851
1852       download->index = gst_mpd_client_get_segment_index (stream) - 1;
1853
1854       GstCaps *caps = gst_dash_demux_get_input_caps (demux, stream);
1855       buffer = gst_fragment_get_buffer (download);
1856       g_return_val_if_fail (buffer != NULL, FALSE);
1857
1858       if (selected_stream->need_header) {
1859         /* Store the new input caps for that stream */
1860         gst_caps_replace (&dash_stream->input_caps, caps);
1861         GST_INFO_OBJECT (demux, "Input source caps: %" GST_PTR_FORMAT,
1862             dash_stream->input_caps);
1863
1864         /* We need to fetch a new header */
1865         if ((header = gst_dash_demux_get_next_header (demux, selected_stream->idx)) == NULL) {
1866           GST_INFO_OBJECT (demux, "Unable to fetch header");
1867         } else {
1868           /* Replace fragment buffer with a new one including the header */
1869           GstBuffer *header_buffer = gst_fragment_get_buffer(header);
1870           buffer = gst_buffer_join(header_buffer, buffer);
1871           g_object_unref (header);
1872           selected_stream->need_header = FALSE;
1873         }
1874       } else {
1875         gst_caps_unref (caps);
1876       }
1877
1878       g_get_current_time (&now);
1879       g_object_unref (download);
1880
1881       gst_buffer_set_caps(buffer, dash_stream->input_caps);
1882       GST_BUFFER_TIMESTAMP(buffer) = timestamp;
1883       GST_BUFFER_DURATION(buffer) = duration;
1884       GST_BUFFER_OFFSET(buffer) = gst_mpd_client_get_segment_index (stream) - 1;
1885       size_buffer = GST_BUFFER_SIZE (buffer);
1886       /* Push fragment into the queue */
1887       gst_dash_demux_stream_push_data (selected_stream, buffer);
1888       diff = (GST_TIMEVAL_TO_TIME (now) - GST_TIMEVAL_TO_TIME (start));
1889       gst_download_rate_add_rate (&selected_stream->dnl_rate, size_buffer, diff, duration);
1890       GST_DEBUG_OBJECT (demux,
1891                         "Stream: %d Download rate = %" G_GUINT64_FORMAT " Kbits/s (%" G_GUINT64_FORMAT
1892                         " Ko in %.2f s)\n", selected_stream->idx,
1893                         gst_download_rate_get_current_rate (&selected_stream->dnl_rate) / 1000,
1894                         size_buffer / 1024,
1895                         ((double) diff / GST_SECOND));
1896     }
1897   }
1898
1899   demux->end_of_period = end_of_period;
1900
1901   return !end_of_period;
1902 }