gst/hls/gsthlsdemux.c

   1 /* GStreamer
   2  * Copyright (C) 2010 Marc-Andre Lureau <marcandre.lureau@gmail.com>
   3  * Copyright (C) 2010 Andoni Morales Alastruey <ylatuya@gmail.com>
   4  * Copyright (C) 2011, Hewlett-Packard Development Company, L.P.
   5  *  Author: Youness Alaoui <youness.alaoui@collabora.co.uk>, Collabora Ltd.
   6  *  Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>, Collabora Ltd.
   7  *
   8  * Gsthlsdemux.c:
   9  *
  10  * This library is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Library General Public
  12  * License as published by the Free Software Foundation; either
  13  * version 2 of the License, or (at your option) any later version.
  14  *
  15  * This library is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18  * Library General Public License for more details.
  19  *
  20  * You should have received a copy of the GNU Library General Public
  21  * License along with this library; if not, write to the
  22  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  23  * Boston, MA 02111-1307, USA.
  24  */
  25 /**
  26  * SECTION:element-hlsdemux
  27  *
  28  * HTTP Live Streaming demuxer element.
  29  *
  30  * <refsect2>
  31  * <title>Example launch line</title>
  32  * |[
  33  * gst-launch souphttpsrc location=http://devimages.apple.com/iphone/samples/bipbop/gear4/prog_index.m3u8 ! hlsdemux ! decodebin2 ! ffmpegcolorspace ! videoscale ! autovideosink
  34  * ]|
  35  * </refsect2>
  36  *
  37  * Last reviewed on 2010-10-07
  38  */
  39
  40 #ifdef HAVE_CONFIG_H
  41 #  include "config.h"
  42 #endif
  43
  44 /* FIXME 0.11: suppress warnings for deprecated API such as GStaticRecMutex
  45  * with newer GLib versions (>= 2.31.0) */
  46 #define GLIB_DISABLE_DEPRECATION_WARNINGS
  47
  48 #include <string.h>
  49 #include <gst/base/gsttypefindhelper.h>
  50 #include <gst/glib-compat-private.h>
  51 #include "gsthlsdemux.h"
  52
  53 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",
  54     GST_PAD_SRC,
  55     GST_PAD_SOMETIMES,
  56     GST_STATIC_CAPS_ANY);
  57
  58 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  59     GST_PAD_SINK,
  60     GST_PAD_ALWAYS,
  61     GST_STATIC_CAPS ("application/x-hls"));
  62
  63 GST_DEBUG_CATEGORY_STATIC (gst_hls_demux_debug);
  64 #define GST_CAT_DEFAULT gst_hls_demux_debug
  65
  66 enum
  67 {
  68   PROP_0,
  69
  70   PROP_FRAGMENTS_CACHE,
  71   PROP_BITRATE_LIMIT,
  72   PROP_CONNECTION_SPEED,
  73   PROP_LAST
  74 };
  75
  76 static const float update_interval_factor[] = { 1, 0.5, 1.5, 3 };
  77
  78 #define DEFAULT_FRAGMENTS_CACHE 3
  79 #define DEFAULT_FAILED_COUNT 3
  80 #define DEFAULT_BITRATE_LIMIT 0.8
  81 #define DEFAULT_CONNECTION_SPEED    0
  82 #ifdef GST_EXT_HLS_MODIFICATION
  83 #define DEFAULT_FIRST_FRAGMENTS_CACHE 2
  84 #endif
  85
  86 /* GObject */
  87 static void gst_hls_demux_set_property (GObject * object, guint prop_id,
  88     const GValue * value, GParamSpec * pspec);
  89 static void gst_hls_demux_get_property (GObject * object, guint prop_id,
  90     GValue * value, GParamSpec * pspec);
  91 static void gst_hls_demux_dispose (GObject * obj);
  92
  93 /* GstElement */
  94 static GstStateChangeReturn
  95 gst_hls_demux_change_state (GstElement * element, GstStateChange transition);
  96
  97 /* GstHLSDemux */
  98 static GstFlowReturn gst_hls_demux_chain (GstPad * pad, GstBuffer * buf);
  99 static gboolean gst_hls_demux_sink_event (GstPad * pad, GstEvent * event);
 100 static gboolean gst_hls_demux_src_event (GstPad * pad, GstEvent * event);
 101 static gboolean gst_hls_demux_src_query (GstPad * pad, GstQuery * query);
 102 static void gst_hls_demux_stream_loop (GstHLSDemux * demux);
 103 static void gst_hls_demux_updates_loop (GstHLSDemux * demux);
 104 static void gst_hls_demux_stop (GstHLSDemux * demux);
 105 static gboolean gst_hls_demux_cache_fragments (GstHLSDemux * demux);
 106 static gboolean gst_hls_demux_schedule (GstHLSDemux * demux);
 107 static gboolean gst_hls_demux_switch_playlist (GstHLSDemux * demux);
 108 static gboolean gst_hls_demux_get_next_fragment (GstHLSDemux * demux,
 109     gboolean caching);
 110 static gboolean gst_hls_demux_update_playlist (GstHLSDemux * demux,
 111     gboolean update);
 112 static void gst_hls_demux_reset (GstHLSDemux * demux, gboolean dispose);
 113 static gboolean gst_hls_demux_set_location (GstHLSDemux * demux,
 114     const gchar * uri);
 115 static gchar *gst_hls_src_buf_to_utf8_playlist (GstBuffer * buf);
 116
 117 static void
 118 _do_init (GType type)
 119 {
 120   GST_DEBUG_CATEGORY_INIT (gst_hls_demux_debug, "hlsdemux", 0,
 121       "hlsdemux element");
 122 }
 123
 124 GST_BOILERPLATE_FULL (GstHLSDemux, gst_hls_demux, GstElement,
 125     GST_TYPE_ELEMENT, _do_init);
 126
 127 static void
 128 gst_hls_demux_base_init (gpointer g_class)
 129 {
 130   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 131
 132   gst_element_class_add_static_pad_template (element_class, &srctemplate);
 133
 134   gst_element_class_add_static_pad_template (element_class, &sinktemplate);
 135
 136   gst_element_class_set_details_simple (element_class,
 137       "HLS Demuxer",
 138       "Demuxer/URIList",
 139       "HTTP Live Streaming demuxer",
 140       "Marc-Andre Lureau <marcandre.lureau@gmail.com>\n"
 141       "Andoni Morales Alastruey <ylatuya@gmail.com>");
 142 }
 143
 144 static void
 145 gst_hls_demux_dispose (GObject * obj)
 146 {
 147   GstHLSDemux *demux = GST_HLS_DEMUX (obj);
 148
 149   if (demux->stream_task) {
 150     if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED) {
 151       GST_DEBUG_OBJECT (demux, "Leaving streaming task");
 152       gst_task_stop (demux->stream_task);
 153       gst_task_join (demux->stream_task);
 154     }
 155     gst_object_unref (demux->stream_task);
 156     g_static_rec_mutex_free (&demux->stream_lock);
 157     demux->stream_task = NULL;
 158   }
 159
 160   if (demux->updates_task) {
 161     if (GST_TASK_STATE (demux->updates_task) != GST_TASK_STOPPED) {
 162       GST_DEBUG_OBJECT (demux, "Leaving updates task");
 163       gst_task_stop (demux->updates_task);
 164       gst_task_join (demux->updates_task);
 165     }
 166     gst_object_unref (demux->updates_task);
 167     g_mutex_free (demux->updates_timed_lock);
 168     g_static_rec_mutex_free (&demux->updates_lock);
 169     demux->updates_task = NULL;
 170   }
 171
 172   if (demux->downloader != NULL) {
 173     g_object_unref (demux->downloader);
 174     demux->downloader = NULL;
 175   }
 176
 177   gst_hls_demux_reset (demux, TRUE);
 178
 179   g_queue_free (demux->queue);
 180
 181   G_OBJECT_CLASS (parent_class)->dispose (obj);
 182 }
 183
 184 static void
 185 gst_hls_demux_class_init (GstHLSDemuxClass * klass)
 186 {
 187   GObjectClass *gobject_class;
 188   GstElementClass *gstelement_class;
 189
 190   gobject_class = (GObjectClass *) klass;
 191   gstelement_class = (GstElementClass *) klass;
 192
 193   gobject_class->set_property = gst_hls_demux_set_property;
 194   gobject_class->get_property = gst_hls_demux_get_property;
 195   gobject_class->dispose = gst_hls_demux_dispose;
 196
 197   g_object_class_install_property (gobject_class, PROP_FRAGMENTS_CACHE,
 198       g_param_spec_uint ("fragments-cache", "Fragments cache",
 199           "Number of fragments needed to be cached to start playing",
 200           2, G_MAXUINT, DEFAULT_FRAGMENTS_CACHE,
 201           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 202
 203   g_object_class_install_property (gobject_class, PROP_BITRATE_LIMIT,
 204       g_param_spec_float ("bitrate-limit",
 205           "Bitrate limit in %",
 206           "Limit of the available bitrate to use when switching to alternates.",
 207           0, 1, DEFAULT_BITRATE_LIMIT,
 208           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 209
 210   g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED,
 211       g_param_spec_uint ("connection-speed", "Connection Speed",
 212           "Network connection speed in kbps (0 = unknown)",
 213           0, G_MAXUINT / 1000, DEFAULT_CONNECTION_SPEED,
 214           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 215
 216   gstelement_class->change_state =
 217       GST_DEBUG_FUNCPTR (gst_hls_demux_change_state);
 218 }
 219
 220 static void
 221 gst_hls_demux_init (GstHLSDemux * demux, GstHLSDemuxClass * klass)
 222 {
 223   /* sink pad */
 224   demux->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 225   gst_pad_set_chain_function (demux->sinkpad,
 226       GST_DEBUG_FUNCPTR (gst_hls_demux_chain));
 227   gst_pad_set_event_function (demux->sinkpad,
 228       GST_DEBUG_FUNCPTR (gst_hls_demux_sink_event));
 229   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 230
 231   /* Downloader */
 232   demux->downloader = gst_uri_downloader_new ();
 233
 234   demux->do_typefind = TRUE;
 235
 236   /* Properties */
 237   demux->fragments_cache = DEFAULT_FRAGMENTS_CACHE;
 238   demux->bitrate_limit = DEFAULT_BITRATE_LIMIT;
 239   demux->connection_speed = DEFAULT_CONNECTION_SPEED;
 240
 241   demux->queue = g_queue_new ();
 242
 243   /* Updates task */
 244   g_static_rec_mutex_init (&demux->updates_lock);
 245   demux->updates_task =
 246       gst_task_create ((GstTaskFunction) gst_hls_demux_updates_loop, demux);
 247   gst_task_set_lock (demux->updates_task, &demux->updates_lock);
 248   demux->updates_timed_lock = g_mutex_new ();
 249
 250   /* Streaming task */
 251   g_static_rec_mutex_init (&demux->stream_lock);
 252   demux->stream_task =
 253       gst_task_create ((GstTaskFunction) gst_hls_demux_stream_loop, demux);
 254   gst_task_set_lock (demux->stream_task, &demux->stream_lock);
 255 }
 256
 257 static void
 258 gst_hls_demux_set_property (GObject * object, guint prop_id,
 259     const GValue * value, GParamSpec * pspec)
 260 {
 261   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 262
 263   switch (prop_id) {
 264     case PROP_FRAGMENTS_CACHE:
 265       demux->fragments_cache = g_value_get_uint (value);
 266       break;
 267     case PROP_BITRATE_LIMIT:
 268       demux->bitrate_limit = g_value_get_float (value);
 269       break;
 270     case PROP_CONNECTION_SPEED:
 271       demux->connection_speed = g_value_get_uint (value) * 1000;
 272       break;
 273     default:
 274       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 275       break;
 276   }
 277 }
 278
 279 static void
 280 gst_hls_demux_get_property (GObject * object, guint prop_id, GValue * value,
 281     GParamSpec * pspec)
 282 {
 283   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 284
 285   switch (prop_id) {
 286     case PROP_FRAGMENTS_CACHE:
 287       g_value_set_uint (value, demux->fragments_cache);
 288       break;
 289     case PROP_BITRATE_LIMIT:
 290       g_value_set_float (value, demux->bitrate_limit);
 291       break;
 292     case PROP_CONNECTION_SPEED:
 293       g_value_set_uint (value, demux->connection_speed / 1000);
 294       break;
 295     default:
 296       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 297       break;
 298   }
 299 }
 300
 301 static GstStateChangeReturn
 302 gst_hls_demux_change_state (GstElement * element, GstStateChange transition)
 303 {
 304   GstStateChangeReturn ret;
 305   GstHLSDemux *demux = GST_HLS_DEMUX (element);
 306
 307   switch (transition) {
 308     case GST_STATE_CHANGE_READY_TO_PAUSED:
 309       gst_hls_demux_reset (demux, FALSE);
 310       break;
 311     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
 312       /* Start the streaming loop in paused only if we already received
 313          the main playlist. It might have been stopped if we were in PAUSED
 314          state and we filled our queue with enough cached fragments
 315        */
 316       if (gst_m3u8_client_get_uri (demux->client)[0] != '\0')
 317         gst_task_start (demux->updates_task);
 318       break;
 319     default:
 320       break;
 321   }
 322
 323   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 324
 325   switch (transition) {
 326     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
 327       gst_task_stop (demux->updates_task);
 328       break;
 329     case GST_STATE_CHANGE_PAUSED_TO_READY:
 330       demux->cancelled = TRUE;
 331       gst_hls_demux_stop (demux);
 332       gst_task_join (demux->stream_task);
 333       gst_hls_demux_reset (demux, FALSE);
 334       break;
 335     default:
 336       break;
 337   }
 338   return ret;
 339 }
 340
 341 static gboolean
 342 gst_hls_demux_src_event (GstPad * pad, GstEvent * event)
 343 {
 344   GstHLSDemux *demux;
 345
 346   demux = GST_HLS_DEMUX (gst_pad_get_element_private (pad));
 347
 348   switch (event->type) {
 349     case GST_EVENT_SEEK:
 350     {
 351       gdouble rate;
 352       GstFormat format;
 353       GstSeekFlags flags;
 354       GstSeekType start_type, stop_type;
 355       gint64 start, stop;
 356       GList *walk;
 357       GstClockTime position, current_pos, target_pos;
 358       gint current_sequence;
 359       GstM3U8MediaFile *file;
 360
 361       GST_INFO_OBJECT (demux, "Received GST_EVENT_SEEK");
 362
 363       if (gst_m3u8_client_is_live (demux->client)) {
 364         GST_WARNING_OBJECT (demux, "Received seek event for live stream");
 365         return FALSE;
 366       }
 367
 368       gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
 369           &stop_type, &stop);
 370
 371       if (format != GST_FORMAT_TIME)
 372         return FALSE;
 373
 374       GST_DEBUG_OBJECT (demux, "seek event, rate: %f start: %" GST_TIME_FORMAT
 375           " stop: %" GST_TIME_FORMAT, rate, GST_TIME_ARGS (start),
 376           GST_TIME_ARGS (stop));
 377
 378       GST_M3U8_CLIENT_LOCK (demux->client);
 379       file = GST_M3U8_MEDIA_FILE (demux->client->current->files->data);
 380       current_sequence = file->sequence;
 381       current_pos = 0;
 382       target_pos = (GstClockTime) start;
 383       for (walk = demux->client->current->files; walk; walk = walk->next) {
 384         file = walk->data;
 385
 386         current_sequence = file->sequence;
 387         if (current_pos <= target_pos
 388             && target_pos < current_pos + file->duration) {
 389           break;
 390         }
 391         current_pos += file->duration;
 392       }
 393       GST_M3U8_CLIENT_UNLOCK (demux->client);
 394
 395       if (walk == NULL) {
 396         GST_WARNING_OBJECT (demux, "Could not find seeked fragment");
 397         return FALSE;
 398       }
 399
 400       if (flags & GST_SEEK_FLAG_FLUSH) {
 401         GST_DEBUG_OBJECT (demux, "sending flush start");
 402         gst_pad_push_event (demux->srcpad, gst_event_new_flush_start ());
 403       }
 404
 405       demux->cancelled = TRUE;
 406       gst_task_pause (demux->stream_task);
 407       gst_uri_downloader_cancel (demux->downloader);
 408       gst_task_stop (demux->updates_task);
 409       gst_task_pause (demux->stream_task);
 410
 411       /* wait for streaming to finish */
 412       g_static_rec_mutex_lock (&demux->stream_lock);
 413
 414       demux->need_cache = TRUE;
 415       while (!g_queue_is_empty (demux->queue)) {
 416         GstBufferList *buf_list = g_queue_pop_head (demux->queue);
 417         gst_buffer_list_unref (buf_list);
 418       }
 419       g_queue_clear (demux->queue);
 420
 421       GST_M3U8_CLIENT_LOCK (demux->client);
 422       GST_DEBUG_OBJECT (demux, "seeking to sequence %d", current_sequence);
 423       demux->client->sequence = current_sequence;
 424       gst_m3u8_client_get_current_position (demux->client, &position);
 425       demux->position_shift = start - position;
 426       demux->need_segment = TRUE;
 427       GST_M3U8_CLIENT_UNLOCK (demux->client);
 428
 429
 430       if (flags & GST_SEEK_FLAG_FLUSH) {
 431         GST_DEBUG_OBJECT (demux, "sending flush stop");
 432         gst_pad_push_event (demux->srcpad, gst_event_new_flush_stop ());
 433       }
 434
 435       demux->cancelled = FALSE;
 436       gst_task_start (demux->stream_task);
 437       g_static_rec_mutex_unlock (&demux->stream_lock);
 438
 439       return TRUE;
 440     }
 441     default:
 442       break;
 443   }
 444
 445   return gst_pad_event_default (pad, event);
 446 }
 447
 448 static gboolean
 449 gst_hls_demux_sink_event (GstPad * pad, GstEvent * event)
 450 {
 451   GstHLSDemux *demux;
 452   GstQuery *query;
 453   gboolean ret;
 454   gchar *uri;
 455
 456   demux = GST_HLS_DEMUX (gst_pad_get_parent (pad));
 457
 458   switch (event->type) {
 459     case GST_EVENT_EOS:{
 460       gchar *playlist = NULL;
 461 #ifdef GST_EXT_HLS_MODIFICATION
 462       GstPad *peer = gst_pad_get_peer (demux->sinkpad);
 463       GObject *src = G_OBJECT (gst_pad_get_parent (peer));
 464
 465       g_object_get (src, "cookies", &demux->cookies, NULL);
 466 #endif
 467
 468       if (demux->playlist == NULL) {
 469         GST_WARNING_OBJECT (demux, "Received EOS without a playlist.");
 470         break;
 471       }
 472
 473       GST_DEBUG_OBJECT (demux,
 474           "Got EOS on the sink pad: main playlist fetched");
 475
 476       query = gst_query_new_uri ();
 477       ret = gst_pad_peer_query (demux->sinkpad, query);
 478       if (ret) {
 479         gst_query_parse_uri (query, &uri);
 480         gst_hls_demux_set_location (demux, uri);
 481         g_free (uri);
 482       }
 483       gst_query_unref (query);
 484
 485       playlist = gst_hls_src_buf_to_utf8_playlist (demux->playlist);
 486       demux->playlist = NULL;
 487       if (playlist == NULL) {
 488         GST_WARNING_OBJECT (demux, "Error validating first playlist.");
 489       } else if (!gst_m3u8_client_update (demux->client, playlist)) {
 490         /* In most cases, this will happen if we set a wrong url in the
 491          * source element and we have received the 404 HTML response instead of
 492          * the playlist */
 493         GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid playlist."),
 494             (NULL));
 495         gst_object_unref (demux);
 496         return FALSE;
 497       }
 498
 499       if (!ret && gst_m3u8_client_is_live (demux->client)) {
 500         GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 501             ("Failed querying the playlist uri, "
 502                 "required for live sources."), (NULL));
 503         gst_object_unref (demux);
 504         return FALSE;
 505       }
 506
 507       gst_task_start (demux->stream_task);
 508       gst_event_unref (event);
 509       gst_object_unref (demux);
 510       return TRUE;
 511     }
 512     case GST_EVENT_NEWSEGMENT:
 513       /* Swallow newsegments, we'll push our own */
 514       gst_event_unref (event);
 515       gst_object_unref (demux);
 516       return TRUE;
 517     default:
 518       break;
 519   }
 520
 521   gst_object_unref (demux);
 522
 523   return gst_pad_event_default (pad, event);
 524 }
 525
 526 static gboolean
 527 gst_hls_demux_src_query (GstPad * pad, GstQuery * query)
 528 {
 529   GstHLSDemux *hlsdemux;
 530   gboolean ret = FALSE;
 531
 532   if (query == NULL)
 533     return FALSE;
 534
 535   hlsdemux = GST_HLS_DEMUX (gst_pad_get_element_private (pad));
 536
 537   switch (query->type) {
 538     case GST_QUERY_DURATION:{
 539       GstClockTime duration = -1;
 540       GstFormat fmt;
 541
 542       gst_query_parse_duration (query, &fmt, NULL);
 543       if (fmt == GST_FORMAT_TIME) {
 544         duration = gst_m3u8_client_get_duration (hlsdemux->client);
 545         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) {
 546           gst_query_set_duration (query, GST_FORMAT_TIME, duration);
 547           ret = TRUE;
 548         }
 549       }
 550       GST_INFO_OBJECT (hlsdemux, "GST_QUERY_DURATION returns %s with duration %"
 551           GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
 552       break;
 553     }
 554     case GST_QUERY_URI:
 555       if (hlsdemux->client) {
 556         /* FIXME: Do we answer with the variant playlist, with the current
 557          * playlist or the the uri of the least downlowaded fragment? */
 558         gst_query_set_uri (query, gst_m3u8_client_get_uri (hlsdemux->client));
 559         ret = TRUE;
 560       }
 561       break;
 562     case GST_QUERY_SEEKING:{
 563       GstFormat fmt;
 564       gint64 stop = -1;
 565
 566       gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL);
 567       GST_INFO_OBJECT (hlsdemux, "Received GST_QUERY_SEEKING with format %d",
 568           fmt);
 569       if (fmt == GST_FORMAT_TIME) {
 570         GstClockTime duration;
 571
 572         duration = gst_m3u8_client_get_duration (hlsdemux->client);
 573         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0)
 574           stop = duration;
 575
 576         gst_query_set_seeking (query, fmt,
 577             !gst_m3u8_client_is_live (hlsdemux->client), 0, stop);
 578         ret = TRUE;
 579         GST_INFO_OBJECT (hlsdemux, "GST_QUERY_SEEKING returning with stop : %"
 580             GST_TIME_FORMAT, GST_TIME_ARGS (stop));
 581       }
 582       break;
 583     }
 584     default:
 585       /* Don't fordward queries upstream because of the special nature of this
 586        * "demuxer", which relies on the upstream element only to be fed with the
 587        * first playlist */
 588       break;
 589   }
 590
 591   return ret;
 592 }
 593
 594 static GstFlowReturn
 595 gst_hls_demux_chain (GstPad * pad, GstBuffer * buf)
 596 {
 597   GstHLSDemux *demux = GST_HLS_DEMUX (gst_pad_get_parent (pad));
 598
 599   if (demux->playlist == NULL)
 600     demux->playlist = buf;
 601   else
 602     demux->playlist = gst_buffer_join (demux->playlist, buf);
 603
 604   gst_object_unref (demux);
 605
 606   return GST_FLOW_OK;
 607 }
 608
 609 static void
 610 gst_hls_demux_stop (GstHLSDemux * demux)
 611 {
 612   gst_uri_downloader_cancel (demux->downloader);
 613
 614   if (GST_TASK_STATE (demux->updates_task) != GST_TASK_STOPPED) {
 615     demux->stop_stream_task = TRUE;
 616     gst_task_stop (demux->updates_task);
 617     GST_TASK_SIGNAL (demux->updates_task);
 618   }
 619
 620   if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED)
 621     gst_task_stop (demux->stream_task);
 622 }
 623
 624 static void
 625 switch_pads (GstHLSDemux * demux, GstCaps * newcaps)
 626 {
 627   GstPad *oldpad = demux->srcpad;
 628
 629   GST_DEBUG ("Switching pads (oldpad:%p) with caps: %" GST_PTR_FORMAT, oldpad,
 630       newcaps);
 631
 632   /* FIXME: This is a workaround for a bug in playsink.
 633    * If we're switching from an audio-only or video-only fragment
 634    * to an audio-video segment, the new sink doesn't know about
 635    * the current running time and audio/video will go out of sync.
 636    *
 637    * This should be fixed in playsink by distributing the
 638    * current running time to newly created sinks and is
 639    * fixed in 0.11 with the new segments.
 640    */
 641   if (demux->srcpad)
 642     gst_pad_push_event (demux->srcpad, gst_event_new_flush_stop ());
 643
 644   /* First create and activate new pad */
 645   demux->srcpad = gst_pad_new_from_static_template (&srctemplate, NULL);
 646   gst_pad_set_event_function (demux->srcpad,
 647       GST_DEBUG_FUNCPTR (gst_hls_demux_src_event));
 648   gst_pad_set_query_function (demux->srcpad,
 649       GST_DEBUG_FUNCPTR (gst_hls_demux_src_query));
 650   gst_pad_set_element_private (demux->srcpad, demux);
 651   gst_pad_set_active (demux->srcpad, TRUE);
 652   gst_pad_set_caps (demux->srcpad, newcaps);
 653   gst_element_add_pad (GST_ELEMENT (demux), demux->srcpad);
 654
 655   gst_element_no_more_pads (GST_ELEMENT (demux));
 656
 657   if (oldpad) {
 658     /* Push out EOS */
 659     gst_pad_push_event (oldpad, gst_event_new_eos ());
 660     gst_pad_set_active (oldpad, FALSE);
 661     gst_element_remove_pad (GST_ELEMENT (demux), oldpad);
 662   }
 663 }
 664
 665 static void
 666 gst_hls_demux_stream_loop (GstHLSDemux * demux)
 667 {
 668   GstFragment *fragment;
 669   GstBufferList *buffer_list;
 670   GstBuffer *buf;
 671   GstFlowReturn ret;
 672
 673   /* Loop for the source pad task. The task is started when we have
 674    * received the main playlist from the source element. It tries first to
 675    * cache the first fragments and then it waits until it has more data in the
 676    * queue. This task is woken up when we push a new fragment to the queue or
 677    * when we reached the end of the playlist  */
 678
 679   if (G_UNLIKELY (demux->need_cache)) {
 680     if (!gst_hls_demux_cache_fragments (demux))
 681       goto cache_error;
 682
 683     /* we can start now the updates thread (only if on playing) */
 684     if (GST_STATE (demux) == GST_STATE_PLAYING)
 685       gst_task_start (demux->updates_task);
 686     GST_INFO_OBJECT (demux, "First fragments cached successfully");
 687   }
 688
 689   if (g_queue_is_empty (demux->queue)) {
 690     if (demux->end_of_playlist)
 691       goto end_of_playlist;
 692
 693     goto pause_task;
 694   }
 695
 696   fragment = g_queue_pop_head (demux->queue);
 697   buffer_list = gst_fragment_get_buffer_list (fragment);
 698   /* Work with the first buffer of the list */
 699   buf = gst_buffer_list_get (buffer_list, 0, 0);
 700
 701   /* Figure out if we need to create/switch pads */
 702   if (G_UNLIKELY (!demux->srcpad
 703 #ifdef GST_EXT_HLS_MODIFICATION
 704           || ((GST_BUFFER_CAPS (buf))
 705               && (!gst_caps_is_equal_fixed (GST_BUFFER_CAPS (buf),
 706               GST_PAD_CAPS (demux->srcpad))))
 707 #else
 708           || !gst_caps_is_equal_fixed (GST_BUFFER_CAPS (buf),
 709               GST_PAD_CAPS (demux->srcpad))
 710 #endif
 711           || demux->need_segment)) {
 712     switch_pads (demux, GST_BUFFER_CAPS (buf));
 713     demux->need_segment = TRUE;
 714   }
 715   g_object_unref (fragment);
 716
 717   if (demux->need_segment) {
 718     GstClockTime start = GST_BUFFER_TIMESTAMP (buf);
 719
 720     start += demux->position_shift;
 721     /* And send a newsegment */
 722     GST_DEBUG_OBJECT (demux, "Sending new-segment. segment start:%"
 723         GST_TIME_FORMAT, GST_TIME_ARGS (start));
 724     gst_pad_push_event (demux->srcpad,
 725         gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME,
 726             start, GST_CLOCK_TIME_NONE, start));
 727     demux->need_segment = FALSE;
 728     demux->position_shift = 0;
 729   }
 730
 731   ret = gst_pad_push_list (demux->srcpad, buffer_list);
 732   if (ret != GST_FLOW_OK)
 733     goto error_pushing;
 734
 735   return;
 736
 737 end_of_playlist:
 738   {
 739     GST_DEBUG_OBJECT (demux, "Reached end of playlist, sending EOS");
 740     gst_pad_push_event (demux->srcpad, gst_event_new_eos ());
 741     gst_hls_demux_stop (demux);
 742     return;
 743   }
 744
 745 cache_error:
 746   {
 747     gst_task_pause (demux->stream_task);
 748     if (!demux->cancelled) {
 749       GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 750           ("Could not cache the first fragments"), (NULL));
 751       gst_hls_demux_stop (demux);
 752     }
 753     return;
 754   }
 755
 756 error_pushing:
 757   {
 758     /* FIXME: handle error */
 759     GST_DEBUG_OBJECT (demux, "Error pushing buffer: %s... stopping task",
 760         gst_flow_get_name (ret));
 761     gst_hls_demux_stop (demux);
 762     return;
 763   }
 764
 765 pause_task:
 766   {
 767     gst_task_pause (demux->stream_task);
 768     return;
 769   }
 770 }
 771
 772 static void
 773 gst_hls_demux_reset (GstHLSDemux * demux, gboolean dispose)
 774 {
 775   demux->need_cache = TRUE;
 776   demux->end_of_playlist = FALSE;
 777   demux->cancelled = FALSE;
 778   demux->do_typefind = TRUE;
 779
 780   if (demux->input_caps) {
 781     gst_caps_unref (demux->input_caps);
 782     demux->input_caps = NULL;
 783   }
 784
 785   if (demux->playlist) {
 786     gst_buffer_unref (demux->playlist);
 787     demux->playlist = NULL;
 788   }
 789
 790   if (demux->client) {
 791     gst_m3u8_client_free (demux->client);
 792     demux->client = NULL;
 793   }
 794
 795   if (!dispose) {
 796     demux->client = gst_m3u8_client_new ("");
 797   }
 798
 799   while (!g_queue_is_empty (demux->queue)) {
 800     GstFragment *fragment = g_queue_pop_head (demux->queue);
 801     g_object_unref (fragment);
 802   }
 803   g_queue_clear (demux->queue);
 804
 805   demux->position_shift = 0;
 806   demux->need_segment = TRUE;
 807 }
 808
 809 static gboolean
 810 gst_hls_demux_set_location (GstHLSDemux * demux, const gchar * uri)
 811 {
 812   if (demux->client)
 813     gst_m3u8_client_free (demux->client);
 814   demux->client = gst_m3u8_client_new (uri);
 815   GST_INFO_OBJECT (demux, "Changed location: %s", uri);
 816   return TRUE;
 817 }
 818
 819 void
 820 gst_hls_demux_updates_loop (GstHLSDemux * demux)
 821 {
 822   /* Loop for the updates. It's started when the first fragments are cached and
 823    * schedules the next update of the playlist (for lives sources) and the next
 824    * update of fragments. When a new fragment is downloaded, it compares the
 825    * download time with the next scheduled update to check if we can or should
 826    * switch to a different bitrate */
 827
 828   g_mutex_lock (demux->updates_timed_lock);
 829   GST_DEBUG_OBJECT (demux, "Started updates task");
 830   while (TRUE) {
 831     /* schedule the next update */
 832     gst_hls_demux_schedule (demux);
 833
 834     /*  block until the next scheduled update or the signal to quit this thread */
 835     if (g_cond_timed_wait (GST_TASK_GET_COND (demux->updates_task),
 836             demux->updates_timed_lock, &demux->next_update)) {
 837       goto quit;
 838     }
 839     /* update the playlist for live sources */
 840     if (gst_m3u8_client_is_live (demux->client)) {
 841       if (!gst_hls_demux_update_playlist (demux, TRUE)) {
 842         demux->client->update_failed_count++;
 843         if (demux->client->update_failed_count < DEFAULT_FAILED_COUNT) {
 844           GST_WARNING_OBJECT (demux, "Could not update the playlist");
 845           continue;
 846         } else {
 847           GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 848               ("Could not update the playlist"), (NULL));
 849           goto quit;
 850         }
 851       }
 852     }
 853
 854     /* if it's a live source and the playlist couldn't be updated, there aren't
 855      * more fragments in the playlist, so we just wait for the next schedulled
 856      * update */
 857     if (gst_m3u8_client_is_live (demux->client) &&
 858         demux->client->update_failed_count > 0) {
 859       GST_WARNING_OBJECT (demux,
 860           "The playlist hasn't been updated, failed count is %d",
 861           demux->client->update_failed_count);
 862       continue;
 863     }
 864
 865     /* fetch the next fragment */
 866 #ifdef GST_EXT_HLS_MODIFICATION
 867     while (g_queue_get_length(demux->queue) < demux->fragments_cache) {
 868       GST_INFO_OBJECT (demux, "demux->queue (%d) < demux->fragments_cache (%d). fetch next fragment",
 869         g_queue_get_length(demux->queue), demux->fragments_cache);
 870       if (!gst_hls_demux_get_next_fragment (demux, FALSE)) {
 871         if (!demux->end_of_playlist && !demux->cancelled) {
 872           demux->client->update_failed_count++;
 873           if (demux->client->update_failed_count < DEFAULT_FAILED_COUNT) {
 874             GST_WARNING_OBJECT (demux, "Could not fetch the next fragment (%d)", demux->client->update_failed_count);
 875             break;
 876           } else {
 877             GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 878                 ("Could not fetch the next fragment"), (NULL));
 879             goto quit;
 880           }
 881         }
 882         GST_INFO_OBJECT (demux, "Got end_of_playlist = %d, cancelled = %d", demux->end_of_playlist, demux->cancelled);
 883         break;
 884       } else {
 885         demux->client->update_failed_count = 0;
 886
 887         /* try to switch to another bitrate if needed */
 888         gst_hls_demux_switch_playlist (demux);
 889       }
 890     }
 891 #else
 892     if (g_queue_is_empty (demux->queue)) {
 893       if (!gst_hls_demux_get_next_fragment (demux, FALSE)) {
 894         if (!demux->end_of_playlist && !demux->cancelled) {
 895           demux->client->update_failed_count++;
 896           if (demux->client->update_failed_count < DEFAULT_FAILED_COUNT) {
 897             GST_WARNING_OBJECT (demux, "Could not fetch the next fragment");
 898             continue;
 899           } else {
 900             GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 901                 ("Could not fetch the next fragment"), (NULL));
 902             goto quit;
 903           }
 904         }
 905       } else {
 906         demux->client->update_failed_count = 0;
 907
 908         /* try to switch to another bitrate if needed */
 909         gst_hls_demux_switch_playlist (demux);
 910       }
 911     }
 912 #endif
 913   }
 914
 915 quit:
 916   {
 917     GST_DEBUG_OBJECT (demux, "Stopped updates task");
 918     gst_hls_demux_stop (demux);
 919     g_mutex_unlock (demux->updates_timed_lock);
 920   }
 921 }
 922
 923 static gboolean
 924 gst_hls_demux_cache_fragments (GstHLSDemux * demux)
 925 {
 926   gint i;
 927
 928   /* If this playlist is a variant playlist, select the first one
 929    * and update it */
 930   if (gst_m3u8_client_has_variant_playlist (demux->client)) {
 931     GstM3U8 *child = NULL;
 932
 933     if (demux->connection_speed == 0) {
 934
 935       GST_M3U8_CLIENT_LOCK (demux->client);
 936       child = demux->client->main->current_variant->data;
 937       GST_M3U8_CLIENT_UNLOCK (demux->client);
 938     } else {
 939       GList *tmp = gst_m3u8_client_get_playlist_for_bitrate (demux->client,
 940           demux->connection_speed);
 941
 942       child = GST_M3U8 (tmp->data);
 943     }
 944
 945     gst_m3u8_client_set_current (demux->client, child);
 946     if (!gst_hls_demux_update_playlist (demux, FALSE)) {
 947       GST_ERROR_OBJECT (demux, "Could not fetch the child playlist %s",
 948           child->uri);
 949       return FALSE;
 950     }
 951   }
 952
 953   if (!gst_m3u8_client_is_live (demux->client)) {
 954     GstClockTime duration = gst_m3u8_client_get_duration (demux->client);
 955
 956     GST_DEBUG_OBJECT (demux, "Sending duration message : %" GST_TIME_FORMAT,
 957         GST_TIME_ARGS (duration));
 958     if (duration != GST_CLOCK_TIME_NONE)
 959       gst_element_post_message (GST_ELEMENT (demux),
 960           gst_message_new_duration (GST_OBJECT (demux),
 961               GST_FORMAT_TIME, duration));
 962   }
 963
 964   /* Cache the first fragments */
 965 #ifdef GST_EXT_HLS_MODIFICATION
 966   for (i = 0; i < DEFAULT_FIRST_FRAGMENTS_CACHE; i++) {
 967     gst_element_post_message (GST_ELEMENT (demux),
 968         gst_message_new_buffering (GST_OBJECT (demux),
 969             100 * i / DEFAULT_FIRST_FRAGMENTS_CACHE));
 970 #else
 971   for (i = 0; i < demux->fragments_cache; i++) {
 972     gst_element_post_message (GST_ELEMENT (demux),
 973         gst_message_new_buffering (GST_OBJECT (demux),
 974             100 * i / demux->fragments_cache));
 975 #endif
 976     g_get_current_time (&demux->next_update);
 977     if (!gst_hls_demux_get_next_fragment (demux, TRUE)) {
 978       if (demux->end_of_playlist)
 979         break;
 980       if (!demux->cancelled)
 981         GST_ERROR_OBJECT (demux, "Error caching the first fragments");
 982       return FALSE;
 983     }
 984     /* make sure we stop caching fragments if something cancelled it */
 985     if (demux->cancelled)
 986       return FALSE;
 987
 988     gst_hls_demux_switch_playlist (demux);
 989   }
 990   gst_element_post_message (GST_ELEMENT (demux),
 991       gst_message_new_buffering (GST_OBJECT (demux), 100));
 992
 993   g_get_current_time (&demux->next_update);
 994
 995   demux->need_cache = FALSE;
 996   return TRUE;
 997
 998 }
 999
1000 static gchar *
1001 gst_hls_src_buf_to_utf8_playlist (GstBuffer * buf)
1002 {
1003   gint size;
1004   gchar *data;
1005   gchar *playlist;
1006
1007   data = (gchar *) GST_BUFFER_DATA (buf);
1008   size = GST_BUFFER_SIZE (buf);
1009
1010   if (!g_utf8_validate (data, size, NULL))
1011     goto validate_error;
1012
1013   /* alloc size + 1 to end with a null character */
1014   playlist = g_malloc0 (size + 1);
1015   memcpy (playlist, data, size + 1);
1016
1017   gst_buffer_unref (buf);
1018   return playlist;
1019
1020 validate_error:
1021   gst_buffer_unref (buf);
1022   return NULL;
1023 }
1024
1025 static gboolean
1026 gst_hls_demux_update_playlist (GstHLSDemux * demux, gboolean update)
1027 {
1028   GstFragment *download;
1029   GstBufferListIterator *it;
1030   GstBuffer *buf;
1031   gchar *playlist;
1032   gboolean updated = FALSE;
1033
1034   const gchar *uri = gst_m3u8_client_get_current_uri (demux->client);
1035
1036 #ifdef GST_EXT_HLS_MODIFICATION
1037   download = gst_uri_downloader_fetch_uri (demux->downloader, uri,
1038       &demux->cookies);
1039 #else
1040   download = gst_uri_downloader_fetch_uri (demux->downloader, uri);
1041 #endif
1042
1043   if (download == NULL)
1044     return FALSE;
1045
1046   /* Merge all the buffers in the list to build a unique buffer with the
1047    * playlist */
1048   it = gst_buffer_list_iterate (gst_fragment_get_buffer_list (download));
1049   gst_buffer_list_iterator_next_group (it);
1050   buf = gst_buffer_list_iterator_merge_group (it);
1051
1052   playlist = gst_hls_src_buf_to_utf8_playlist (buf);
1053   gst_buffer_list_iterator_free (it);
1054   g_object_unref (download);
1055
1056   if (playlist == NULL) {
1057     GST_WARNING_OBJECT (demux, "Couldn't not validate playlist encoding");
1058     return FALSE;
1059   }
1060
1061   updated = gst_m3u8_client_update (demux->client, playlist);
1062
1063   /*  If it's a live source, do not let the sequence number go beyond
1064    * three fragments before the end of the list */
1065   if (updated && update == FALSE && demux->client->current &&
1066       gst_m3u8_client_is_live (demux->client)) {
1067     guint last_sequence;
1068
1069     GST_M3U8_CLIENT_LOCK (demux->client);
1070     last_sequence =
1071         GST_M3U8_MEDIA_FILE (g_list_last (demux->client->current->
1072             files)->data)->sequence;
1073
1074     if (demux->client->sequence >= last_sequence - 3) {
1075       GST_DEBUG_OBJECT (demux, "Sequence is beyond playlist. Moving back to %d",
1076           last_sequence - 3);
1077       demux->need_segment = TRUE;
1078       demux->client->sequence = last_sequence - 3;
1079     }
1080     GST_M3U8_CLIENT_UNLOCK (demux->client);
1081   }
1082
1083   return updated;
1084 }
1085
1086 static gboolean
1087 gst_hls_demux_change_playlist (GstHLSDemux * demux, guint max_bitrate)
1088 {
1089   GList *previous_variant, *current_variant;
1090   gint old_bandwidth, new_bandwidth;
1091
1092   /* If user specifies a connection speed never use a playlist with a bandwidth
1093    * superior than it */
1094   if (demux->connection_speed != 0 && max_bitrate > demux->connection_speed)
1095     max_bitrate = demux->connection_speed;
1096
1097   previous_variant = demux->client->main->current_variant;
1098   current_variant = gst_m3u8_client_get_playlist_for_bitrate (demux->client,
1099       max_bitrate);
1100
1101 retry_failover_protection:
1102   old_bandwidth = GST_M3U8 (previous_variant->data)->bandwidth;
1103   new_bandwidth = GST_M3U8 (current_variant->data)->bandwidth;
1104
1105   /* Don't do anything else if the playlist is the same */
1106   if (new_bandwidth == old_bandwidth) {
1107     return TRUE;
1108   }
1109
1110   demux->client->main->current_variant = current_variant;
1111   GST_M3U8_CLIENT_UNLOCK (demux->client);
1112
1113   gst_m3u8_client_set_current (demux->client, current_variant->data);
1114
1115   GST_INFO_OBJECT (demux, "Client was on %dbps, max allowed is %dbps, switching"
1116       " to bitrate %dbps", old_bandwidth, max_bitrate, new_bandwidth);
1117
1118   if (gst_hls_demux_update_playlist (demux, FALSE)) {
1119     GstStructure *s;
1120
1121     s = gst_structure_new ("playlist",
1122         "uri", G_TYPE_STRING, gst_m3u8_client_get_current_uri (demux->client),
1123         "bitrate", G_TYPE_INT, new_bandwidth, NULL);
1124     gst_element_post_message (GST_ELEMENT_CAST (demux),
1125         gst_message_new_element (GST_OBJECT_CAST (demux), s));
1126   } else {
1127     GList *failover = NULL;
1128
1129     GST_INFO_OBJECT (demux, "Unable to update playlist. Switching back");
1130     GST_M3U8_CLIENT_LOCK (demux->client);
1131
1132     failover = g_list_previous (current_variant);
1133     if (failover && new_bandwidth == GST_M3U8 (failover->data)->bandwidth) {
1134       current_variant = failover;
1135       goto retry_failover_protection;
1136     }
1137
1138     demux->client->main->current_variant = previous_variant;
1139     GST_M3U8_CLIENT_UNLOCK (demux->client);
1140     gst_m3u8_client_set_current (demux->client, previous_variant->data);
1141     /*  Try a lower bitrate (or stop if we just tried the lowest) */
1142     if (new_bandwidth ==
1143         GST_M3U8 (g_list_first (demux->client->main->lists)->data)->bandwidth)
1144       return FALSE;
1145     else
1146       return gst_hls_demux_change_playlist (demux, new_bandwidth - 1);
1147   }
1148
1149   /* Force typefinding since we might have changed media type */
1150   demux->do_typefind = TRUE;
1151
1152   return TRUE;
1153 }
1154
1155 static gboolean
1156 gst_hls_demux_schedule (GstHLSDemux * demux)
1157 {
1158   gfloat update_factor;
1159   gint count;
1160
1161   /* As defined in §6.3.4. Reloading the Playlist file:
1162    * "If the client reloads a Playlist file and finds that it has not
1163    * changed then it MUST wait for a period of time before retrying.  The
1164    * minimum delay is a multiple of the target duration.  This multiple is
1165    * 0.5 for the first attempt, 1.5 for the second, and 3.0 thereafter."
1166    */
1167   count = demux->client->update_failed_count;
1168   if (count < 3)
1169     update_factor = update_interval_factor[count];
1170   else
1171     update_factor = update_interval_factor[3];
1172
1173   /* schedule the next update using the target duration field of the
1174    * playlist */
1175   g_time_val_add (&demux->next_update,
1176       gst_m3u8_client_get_target_duration (demux->client)
1177       / GST_SECOND * G_USEC_PER_SEC * update_factor);
1178   GST_DEBUG_OBJECT (demux, "Next update scheduled at %s",
1179       g_time_val_to_iso8601 (&demux->next_update));
1180
1181   return TRUE;
1182 }
1183
1184 static gboolean
1185 gst_hls_demux_switch_playlist (GstHLSDemux * demux)
1186 {
1187   GTimeVal now;
1188   GstClockTime diff;
1189   gsize size;
1190   gint bitrate;
1191   GstFragment *fragment = g_queue_peek_tail (demux->queue);
1192
1193   GST_M3U8_CLIENT_LOCK (demux->client);
1194   if (!demux->client->main->lists) {
1195     GST_M3U8_CLIENT_UNLOCK (demux->client);
1196     return TRUE;
1197   }
1198   GST_M3U8_CLIENT_UNLOCK (demux->client);
1199
1200   /* compare the time when the fragment was downloaded with the time when it was
1201    * scheduled */
1202   g_get_current_time (&now);
1203   diff = (GST_TIMEVAL_TO_TIME (now) - GST_TIMEVAL_TO_TIME (demux->next_update));
1204   size = gst_fragment_get_total_size (fragment);
1205   bitrate = (size * 8) / ((double) diff / GST_SECOND);
1206
1207   GST_DEBUG ("Downloaded %d bytes in %" GST_TIME_FORMAT ". Bitrate is : %d",
1208       size, GST_TIME_ARGS (diff), bitrate);
1209
1210   return gst_hls_demux_change_playlist (demux, bitrate * demux->bitrate_limit);
1211 }
1212
1213 static gboolean
1214 gst_hls_demux_decrypt_buffer_list (GstHLSDemux * demux,
1215     GstBufferList * buffer_list)
1216 {
1217   GstBufferListIterator *it;
1218   gint remain_size;
1219   GstBuffer *buf, *remained_buf, *in_buf, *out_buf;
1220   gint in_size, out_size;
1221
1222   remained_buf = NULL;
1223   it = gst_buffer_list_iterate (buffer_list);
1224   gst_buffer_list_iterator_next_group (it);
1225   while (buf = gst_buffer_list_iterator_next (it)) {
1226     if (remained_buf) {
1227       in_buf = gst_buffer_merge (remained_buf, buf);
1228       gst_buffer_unref (remained_buf);
1229       remained_buf = NULL;
1230     } else {
1231       in_buf = buf;
1232     }
1233
1234     in_size = GST_BUFFER_SIZE (in_buf);
1235     remain_size = in_size % GST_HLS_DEMUX_AES_BLOCK_SIZE;
1236
1237     if (remain_size) {
1238       remained_buf = gst_buffer_new_and_alloc (remain_size);
1239       memcpy (GST_BUFFER_DATA (remained_buf),
1240           GST_BUFFER_DATA (in_buf) + in_size - remain_size, remain_size);
1241     }
1242
1243     out_size = in_size - remain_size + GST_HLS_DEMUX_AES_BLOCK_SIZE;
1244     out_buf = gst_buffer_new_and_alloc (out_size);
1245
1246     if (!gst_m3u8_client_decrypt_update (demux->client, GST_BUFFER_DATA (out_buf),
1247         &out_size, GST_BUFFER_DATA (in_buf), in_size - remain_size)) {
1248       gst_buffer_unref (in_buf);
1249       gst_buffer_unref (out_buf);
1250       gst_buffer_list_iterator_free (it);
1251       return FALSE;
1252     }
1253
1254     GST_BUFFER_SIZE (out_buf) = out_size;
1255     GST_BUFFER_TIMESTAMP (out_buf) = GST_CLOCK_TIME_NONE;
1256     GST_BUFFER_DURATION (out_buf) = GST_CLOCK_TIME_NONE;
1257     GST_BUFFER_FLAGS (out_buf) = GST_BUFFER_FLAGS (in_buf);
1258
1259     gst_buffer_list_iterator_take (it, out_buf);
1260     if (in_buf != buf)
1261       gst_buffer_unref (in_buf);
1262   }
1263   if (remained_buf) {
1264     GST_WARNING_OBJECT (demux, "remained buffer should be empty!");
1265     gst_buffer_unref (remained_buf);
1266   }
1267   gst_buffer_list_iterator_free (it);
1268
1269   return TRUE;
1270 }
1271
1272 static gboolean
1273 gst_hls_demux_get_next_fragment (GstHLSDemux * demux, gboolean caching)
1274 {
1275   GstFragment *download;
1276   const gchar *next_fragment_uri;
1277   GstM3U8Key *next_fragment_key;
1278   GstClockTime duration;
1279   GstClockTime timestamp;
1280   GstBufferList *buffer_list;
1281   GstBufferListIterator *it;
1282   GstBuffer *buf;
1283   gboolean discont;
1284
1285   if (!gst_m3u8_client_get_next_fragment (demux->client, &discont,
1286           &next_fragment_uri, &duration, &timestamp, &next_fragment_key)) {
1287     GST_INFO_OBJECT (demux, "This playlist doesn't contain more fragments");
1288     demux->end_of_playlist = TRUE;
1289     gst_task_start (demux->stream_task);
1290     return FALSE;
1291   }
1292
1293   if (next_fragment_key && !next_fragment_key->data) {
1294     GST_INFO_OBJECT (demux, "Fetching next fragment key %s",
1295         next_fragment_key->uri);
1296
1297 #ifdef GST_EXT_HLS_MODIFICATION
1298     download = gst_uri_downloader_fetch_uri (demux->downloader,
1299         next_fragment_key->uri, &demux->cookies);
1300 #else
1301     download = gst_uri_downloader_fetch_uri (demux->downloader,
1302         next_fragment_key->uri);
1303 #endif
1304
1305     if (download == NULL)
1306       goto error;
1307
1308     buffer_list = gst_fragment_get_buffer_list (download);
1309     it = gst_buffer_list_iterate (buffer_list);
1310     gst_buffer_list_iterator_next_group (it);
1311     buf = gst_buffer_list_iterator_merge_group (it);
1312     next_fragment_key->data = GST_BUFFER_DATA (buf);
1313     gst_buffer_list_iterator_free (it);
1314     gst_buffer_list_unref (buffer_list);
1315
1316     gst_m3u8_client_decrypt_init (demux->client, next_fragment_key);
1317   }
1318
1319   GST_INFO_OBJECT (demux, "Fetching next fragment %s", next_fragment_uri);
1320
1321 #ifdef GST_EXT_HLS_MODIFICATION
1322   download = gst_uri_downloader_fetch_uri (demux->downloader,
1323       next_fragment_uri, &demux->cookies);
1324 #else
1325   download = gst_uri_downloader_fetch_uri (demux->downloader,
1326       next_fragment_uri);
1327 #endif
1328
1329   if (download == NULL)
1330     goto error;
1331
1332   buffer_list = gst_fragment_get_buffer_list (download);
1333
1334   if (next_fragment_key && next_fragment_key->data) {
1335     if (!gst_hls_demux_decrypt_buffer_list (demux, buffer_list))
1336       goto error;
1337   }
1338
1339   buf = gst_buffer_list_get (buffer_list, 0, 0);
1340   GST_BUFFER_DURATION (buf) = duration;
1341   GST_BUFFER_TIMESTAMP (buf) = timestamp;
1342
1343   /* We actually need to do this every time we switch bitrate */
1344   if (G_UNLIKELY (demux->do_typefind)) {
1345     GstCaps *caps = gst_type_find_helper_for_buffer (NULL, buf, NULL);
1346
1347     if (!demux->input_caps || !gst_caps_is_equal (caps, demux->input_caps)) {
1348       gst_caps_replace (&demux->input_caps, caps);
1349       /* gst_pad_set_caps (demux->srcpad, demux->input_caps); */
1350       GST_INFO_OBJECT (demux, "Input source caps: %" GST_PTR_FORMAT,
1351           demux->input_caps);
1352       demux->do_typefind = FALSE;
1353     }
1354     gst_caps_unref (caps);
1355   }
1356   gst_buffer_set_caps (buf, demux->input_caps);
1357
1358   if (discont) {
1359     GST_DEBUG_OBJECT (demux, "Marking fragment as discontinuous");
1360     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
1361   }
1362
1363   g_queue_push_tail (demux->queue, download);
1364   gst_buffer_list_unref (buffer_list);
1365   if (!caching) {
1366     GST_TASK_SIGNAL (demux->updates_task);
1367     gst_task_start (demux->stream_task);
1368   }
1369   return TRUE;
1370
1371 error:
1372   {
1373     gst_hls_demux_stop (demux);
1374     return FALSE;
1375   }
1376 }