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