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  * Copyright (C) 2014 Sebastian Dröge <sebastian@centricular.com>
   8  *
   9  * Gsthlsdemux.c:
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Library General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Library General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Library General Public
  22  * License along with this library; if not, write to the
  23  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  24  * Boston, MA 02110-1301, USA.
  25  */
  26 /**
  27  * SECTION:element-hlsdemux
  28  *
  29  * HTTP Live Streaming demuxer element.
  30  *
  31  * <refsect2>
  32  * <title>Example launch line</title>
  33  * |[
  34  * gst-launch souphttpsrc location=http://devimages.apple.com/iphone/samples/bipbop/gear4/prog_index.m3u8 ! hlsdemux ! decodebin2 ! videoconvert ! videoscale ! autovideosink
  35  * ]|
  36  * </refsect2>
  37  */
  38
  39 #ifdef HAVE_CONFIG_H
  40 #  include "config.h"
  41 #endif
  42
  43 #include <string.h>
  44 #include <gst/base/gsttypefindhelper.h>
  45 #include "gsthlsdemux.h"
  46
  47 #define GST_ELEMENT_ERROR_FROM_ERROR(el, msg, err) G_STMT_START { \
  48   gchar *__dbg = g_strdup_printf ("%s: %s", msg, err->message);         \
  49   GST_WARNING_OBJECT (el, "error: %s", __dbg);                          \
  50   gst_element_message_full (GST_ELEMENT(el), GST_MESSAGE_ERROR,         \
  51     err->domain, err->code,                                             \
  52     NULL, __dbg, __FILE__, GST_FUNCTION, __LINE__);                     \
  53   g_clear_error (&err); \
  54 } G_STMT_END
  55
  56 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src_%u",
  57     GST_PAD_SRC,
  58     GST_PAD_SOMETIMES,
  59     GST_STATIC_CAPS_ANY);
  60
  61 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  62     GST_PAD_SINK,
  63     GST_PAD_ALWAYS,
  64     GST_STATIC_CAPS ("application/x-hls"));
  65
  66 GST_DEBUG_CATEGORY_STATIC (gst_hls_demux_debug);
  67 #define GST_CAT_DEFAULT gst_hls_demux_debug
  68
  69 enum
  70 {
  71   PROP_0,
  72
  73   PROP_FRAGMENTS_CACHE,
  74   PROP_BITRATE_LIMIT,
  75   PROP_CONNECTION_SPEED,
  76   PROP_LAST
  77 };
  78
  79 #define DEFAULT_FRAGMENTS_CACHE 1
  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 static void gst_hls_demux_handle_message (GstBin * bin, GstMessage * msg);
  96
  97 /* GstHLSDemux */
  98 static GstFlowReturn gst_hls_demux_chain (GstPad * pad, GstObject * parent,
  99     GstBuffer * buf);
 100 static gboolean gst_hls_demux_sink_event (GstPad * pad, GstObject * parent,
 101     GstEvent * event);
 102 static gboolean gst_hls_demux_src_event (GstPad * pad, GstObject * parent,
 103     GstEvent * event);
 104 static gboolean gst_hls_demux_src_query (GstPad * pad, GstObject * parent,
 105     GstQuery * query);
 106 static void gst_hls_demux_stream_loop (GstHLSDemux * demux);
 107 static void gst_hls_demux_updates_loop (GstHLSDemux * demux);
 108 static void gst_hls_demux_stop (GstHLSDemux * demux);
 109 static void gst_hls_demux_pause_tasks (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 * end_of_playlist, GError ** err);
 113 static gboolean gst_hls_demux_update_playlist (GstHLSDemux * demux,
 114     gboolean update, GError ** err);
 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, const gchar * base_uri);
 118 static gchar *gst_hls_src_buf_to_utf8_playlist (GstBuffer * buf);
 119
 120 static gboolean gst_hls_demux_change_playlist (GstHLSDemux * demux,
 121     guint max_bitrate);
 122 static GstBuffer *gst_hls_demux_decrypt_fragment (GstHLSDemux * demux,
 123     GstBuffer * encrypted_buffer, GError ** err);
 124 static gboolean
 125 gst_hls_demux_decrypt_start (GstHLSDemux * demux, const guint8 * key_data,
 126     const guint8 * iv_data);
 127 static void gst_hls_demux_decrypt_end (GstHLSDemux * demux);
 128
 129 #define gst_hls_demux_parent_class parent_class
 130 G_DEFINE_TYPE (GstHLSDemux, gst_hls_demux, GST_TYPE_BIN);
 131
 132 #define STATISTICS_MESSAGE_NAME "adaptive-streaming-statistics"
 133
 134 static void
 135 gst_hls_demux_dispose (GObject * obj)
 136 {
 137   GstHLSDemux *demux = GST_HLS_DEMUX (obj);
 138
 139   if (demux->stream_task) {
 140     gst_object_unref (demux->stream_task);
 141     g_rec_mutex_clear (&demux->stream_lock);
 142     demux->stream_task = NULL;
 143   }
 144
 145   if (demux->updates_task) {
 146     gst_object_unref (demux->updates_task);
 147     g_rec_mutex_clear (&demux->updates_lock);
 148     demux->updates_task = NULL;
 149   }
 150
 151   if (demux->downloader != NULL) {
 152     g_object_unref (demux->downloader);
 153     demux->downloader = NULL;
 154   }
 155
 156   gst_hls_demux_reset (demux, TRUE);
 157
 158   if (demux->src_srcpad) {
 159     gst_object_unref (demux->src_srcpad);
 160     demux->src_srcpad = NULL;
 161   }
 162
 163   g_mutex_clear (&demux->download_lock);
 164   g_cond_clear (&demux->download_cond);
 165   g_mutex_clear (&demux->updates_timed_lock);
 166   g_cond_clear (&demux->updates_timed_cond);
 167   g_mutex_clear (&demux->fragment_download_lock);
 168   g_cond_clear (&demux->fragment_download_cond);
 169
 170   G_OBJECT_CLASS (parent_class)->dispose (obj);
 171 }
 172
 173 static void
 174 gst_hls_demux_class_init (GstHLSDemuxClass * klass)
 175 {
 176   GObjectClass *gobject_class;
 177   GstElementClass *element_class;
 178   GstBinClass *bin_class;
 179
 180   gobject_class = (GObjectClass *) klass;
 181   element_class = (GstElementClass *) klass;
 182   bin_class = (GstBinClass *) klass;
 183
 184   gobject_class->set_property = gst_hls_demux_set_property;
 185   gobject_class->get_property = gst_hls_demux_get_property;
 186   gobject_class->dispose = gst_hls_demux_dispose;
 187
 188   g_object_class_install_property (gobject_class, PROP_FRAGMENTS_CACHE,
 189       g_param_spec_uint ("fragments-cache", "Fragments cache",
 190           "Number of fragments needed to be cached to start playing "
 191           "(DEPRECATED: Has no effect since 1.3.1)",
 192           1, G_MAXUINT, DEFAULT_FRAGMENTS_CACHE,
 193           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 194
 195   g_object_class_install_property (gobject_class, PROP_BITRATE_LIMIT,
 196       g_param_spec_float ("bitrate-limit",
 197           "Bitrate limit in %",
 198           "Limit of the available bitrate to use when switching to alternates.",
 199           0, 1, DEFAULT_BITRATE_LIMIT,
 200           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 201
 202   g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED,
 203       g_param_spec_uint ("connection-speed", "Connection Speed",
 204           "Network connection speed in kbps (0 = unknown)",
 205           0, G_MAXUINT / 1000, DEFAULT_CONNECTION_SPEED,
 206           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 207
 208   element_class->change_state = GST_DEBUG_FUNCPTR (gst_hls_demux_change_state);
 209
 210   gst_element_class_add_pad_template (element_class,
 211       gst_static_pad_template_get (&srctemplate));
 212
 213   gst_element_class_add_pad_template (element_class,
 214       gst_static_pad_template_get (&sinktemplate));
 215
 216   gst_element_class_set_static_metadata (element_class,
 217       "HLS Demuxer",
 218       "Codec/Demuxer/Adaptive",
 219       "HTTP Live Streaming demuxer",
 220       "Marc-Andre Lureau <marcandre.lureau@gmail.com>\n"
 221       "Andoni Morales Alastruey <ylatuya@gmail.com>");
 222
 223   bin_class->handle_message = gst_hls_demux_handle_message;
 224
 225   GST_DEBUG_CATEGORY_INIT (gst_hls_demux_debug, "hlsdemux", 0,
 226       "hlsdemux element");
 227 }
 228
 229 static void
 230 gst_hls_demux_init (GstHLSDemux * demux)
 231 {
 232   /* sink pad */
 233   demux->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 234   gst_pad_set_chain_function (demux->sinkpad,
 235       GST_DEBUG_FUNCPTR (gst_hls_demux_chain));
 236   gst_pad_set_event_function (demux->sinkpad,
 237       GST_DEBUG_FUNCPTR (gst_hls_demux_sink_event));
 238   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 239
 240   /* Downloader */
 241   demux->downloader = gst_uri_downloader_new ();
 242
 243   demux->do_typefind = TRUE;
 244
 245   /* Properties */
 246   demux->bitrate_limit = DEFAULT_BITRATE_LIMIT;
 247   demux->connection_speed = DEFAULT_CONNECTION_SPEED;
 248
 249   g_mutex_init (&demux->download_lock);
 250   g_cond_init (&demux->download_cond);
 251   g_mutex_init (&demux->updates_timed_lock);
 252   g_cond_init (&demux->updates_timed_cond);
 253   g_mutex_init (&demux->fragment_download_lock);
 254   g_cond_init (&demux->fragment_download_cond);
 255
 256   /* Updates task */
 257   g_rec_mutex_init (&demux->updates_lock);
 258   demux->updates_task =
 259       gst_task_new ((GstTaskFunction) gst_hls_demux_updates_loop, demux, NULL);
 260   gst_task_set_lock (demux->updates_task, &demux->updates_lock);
 261
 262   /* Streaming task */
 263   g_rec_mutex_init (&demux->stream_lock);
 264   demux->stream_task =
 265       gst_task_new ((GstTaskFunction) gst_hls_demux_stream_loop, demux, NULL);
 266   gst_task_set_lock (demux->stream_task, &demux->stream_lock);
 267
 268   demux->have_group_id = FALSE;
 269   demux->group_id = G_MAXUINT;
 270 }
 271
 272 static void
 273 gst_hls_demux_set_property (GObject * object, guint prop_id,
 274     const GValue * value, GParamSpec * pspec)
 275 {
 276   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 277
 278   switch (prop_id) {
 279     case PROP_FRAGMENTS_CACHE:
 280       break;
 281     case PROP_BITRATE_LIMIT:
 282       demux->bitrate_limit = g_value_get_float (value);
 283       break;
 284     case PROP_CONNECTION_SPEED:
 285       demux->connection_speed = g_value_get_uint (value) * 1000;
 286       break;
 287     default:
 288       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 289       break;
 290   }
 291 }
 292
 293 static void
 294 gst_hls_demux_get_property (GObject * object, guint prop_id, GValue * value,
 295     GParamSpec * pspec)
 296 {
 297   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 298
 299   switch (prop_id) {
 300     case PROP_FRAGMENTS_CACHE:
 301       g_value_set_uint (value, 1);
 302       break;
 303     case PROP_BITRATE_LIMIT:
 304       g_value_set_float (value, demux->bitrate_limit);
 305       break;
 306     case PROP_CONNECTION_SPEED:
 307       g_value_set_uint (value, demux->connection_speed / 1000);
 308       break;
 309     default:
 310       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 311       break;
 312   }
 313 }
 314
 315 static GstStateChangeReturn
 316 gst_hls_demux_change_state (GstElement * element, GstStateChange transition)
 317 {
 318   GstStateChangeReturn ret;
 319   GstHLSDemux *demux = GST_HLS_DEMUX (element);
 320
 321   switch (transition) {
 322     case GST_STATE_CHANGE_READY_TO_PAUSED:
 323       gst_hls_demux_reset (demux, FALSE);
 324       gst_uri_downloader_reset (demux->downloader);
 325       break;
 326     case GST_STATE_CHANGE_NULL_TO_READY:
 327       demux->adapter = gst_adapter_new ();
 328       break;
 329     default:
 330       break;
 331   }
 332
 333   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 334
 335   switch (transition) {
 336     case GST_STATE_CHANGE_PAUSED_TO_READY:
 337       gst_hls_demux_stop (demux);
 338       gst_task_join (demux->updates_task);
 339       gst_task_join (demux->stream_task);
 340       gst_hls_demux_reset (demux, FALSE);
 341       break;
 342     case GST_STATE_CHANGE_READY_TO_NULL:
 343       gst_object_unref (demux->adapter);
 344       demux->adapter = NULL;
 345       break;
 346     default:
 347       break;
 348   }
 349   return ret;
 350 }
 351
 352 static void
 353 gst_hls_demux_handle_message (GstBin * bin, GstMessage * msg)
 354 {
 355   GstHLSDemux *demux = GST_HLS_DEMUX_CAST (bin);
 356
 357   switch (GST_MESSAGE_TYPE (msg)) {
 358     case GST_MESSAGE_ERROR:{
 359       GError *err = NULL;
 360       gchar *debug = NULL;
 361       gchar *new_error = NULL;
 362
 363       gst_message_parse_error (msg, &err, &debug);
 364
 365       GST_WARNING_OBJECT (demux, "Source posted error: %d:%d %s (%s)",
 366           err->domain, err->code, err->message, debug);
 367
 368       if (debug)
 369         new_error = g_strdup_printf ("%s: %s\n", err->message, debug);
 370       if (new_error) {
 371         g_free (err->message);
 372         err->message = new_error;
 373       }
 374
 375       /* error, but ask to retry */
 376       g_mutex_lock (&demux->fragment_download_lock);
 377       demux->last_ret = GST_FLOW_CUSTOM_ERROR;
 378       g_clear_error (&demux->last_error);
 379       demux->last_error = g_error_copy (err);
 380       demux->download_finished = TRUE;
 381       g_cond_signal (&demux->fragment_download_cond);
 382       g_mutex_unlock (&demux->fragment_download_lock);
 383
 384       g_error_free (err);
 385       g_free (debug);
 386       gst_message_unref (msg);
 387       msg = NULL;
 388     }
 389       break;
 390     default:
 391       break;
 392   }
 393
 394   if (msg)
 395     GST_BIN_CLASS (parent_class)->handle_message (bin, msg);
 396 }
 397
 398 static gboolean
 399 gst_hls_demux_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 400 {
 401   GstHLSDemux *demux;
 402
 403   demux = GST_HLS_DEMUX (parent);
 404
 405   switch (event->type) {
 406     case GST_EVENT_SEEK:
 407     {
 408       gdouble rate;
 409       GstFormat format;
 410       GstSeekFlags flags;
 411       GstSeekType start_type, stop_type;
 412       gint64 start, stop;
 413       GList *walk;
 414       GstClockTime current_pos, target_pos;
 415       gint64 current_sequence;
 416       GstM3U8MediaFile *file;
 417
 418       GST_INFO_OBJECT (demux, "Received GST_EVENT_SEEK");
 419
 420       if (gst_m3u8_client_is_live (demux->client)) {
 421         GST_WARNING_OBJECT (demux, "Received seek event for live stream");
 422         gst_event_unref (event);
 423         return FALSE;
 424       }
 425
 426       gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
 427           &stop_type, &stop);
 428
 429       if (format != GST_FORMAT_TIME) {
 430         gst_event_unref (event);
 431         return FALSE;
 432       }
 433
 434       GST_DEBUG_OBJECT (demux, "seek event, rate: %f start: %" GST_TIME_FORMAT
 435           " stop: %" GST_TIME_FORMAT, rate, GST_TIME_ARGS (start),
 436           GST_TIME_ARGS (stop));
 437
 438       if (flags & GST_SEEK_FLAG_FLUSH) {
 439         GST_DEBUG_OBJECT (demux, "sending flush start");
 440         gst_pad_push_event (demux->srcpad, gst_event_new_flush_start ());
 441       }
 442
 443       gst_hls_demux_pause_tasks (demux);
 444
 445       /* wait for streaming to finish */
 446       g_rec_mutex_lock (&demux->updates_lock);
 447       g_rec_mutex_unlock (&demux->updates_lock);
 448
 449       g_rec_mutex_lock (&demux->stream_lock);
 450
 451       /* properly cleanup pending decryption status */
 452       if (flags & GST_SEEK_FLAG_FLUSH) {
 453         if (demux->adapter)
 454           gst_adapter_clear (demux->adapter);
 455         if (demux->pending_buffer)
 456           gst_buffer_unref (demux->pending_buffer);
 457         demux->pending_buffer = NULL;
 458         gst_hls_demux_decrypt_end (demux);
 459       }
 460
 461       /* Use I-frame variants for trick modes */
 462       if (demux->client->main->iframe_lists &&
 463           rate < -1.0 && demux->segment.rate >= -1.0
 464           && demux->segment.rate <= 1.0) {
 465         GError *err = NULL;
 466
 467         GST_M3U8_CLIENT_LOCK (demux->client);
 468         /* Switch to I-frame variant */
 469         demux->client->main->current_variant =
 470             demux->client->main->iframe_lists;
 471         GST_M3U8_CLIENT_UNLOCK (demux->client);
 472         gst_m3u8_client_set_current (demux->client,
 473             demux->client->main->iframe_lists->data);
 474         gst_uri_downloader_reset (demux->downloader);
 475         if (!gst_hls_demux_update_playlist (demux, FALSE, &err)) {
 476           g_rec_mutex_unlock (&demux->stream_lock);
 477           GST_ELEMENT_ERROR_FROM_ERROR (demux, "Could not switch playlist",
 478               err);
 479           gst_event_unref (event);
 480           return FALSE;
 481         }
 482         demux->discont = TRUE;
 483         demux->new_playlist = TRUE;
 484         demux->do_typefind = TRUE;
 485
 486         gst_hls_demux_change_playlist (demux,
 487             demux->current_download_rate * demux->bitrate_limit / ABS (rate));
 488       } else if (rate > -1.0 && rate <= 1.0 && (demux->segment.rate < -1.0
 489               || demux->segment.rate > 1.0)) {
 490         GError *err = NULL;
 491
 492         GST_M3U8_CLIENT_LOCK (demux->client);
 493         /* Switch to normal variant */
 494         demux->client->main->current_variant = demux->client->main->lists;
 495         GST_M3U8_CLIENT_UNLOCK (demux->client);
 496         gst_m3u8_client_set_current (demux->client,
 497             demux->client->main->lists->data);
 498
 499         gst_uri_downloader_reset (demux->downloader);
 500
 501         if (!gst_hls_demux_update_playlist (demux, FALSE, &err)) {
 502           g_rec_mutex_unlock (&demux->stream_lock);
 503
 504           GST_ELEMENT_ERROR_FROM_ERROR (demux, "Could not switch playlist",
 505               err);
 506           gst_event_unref (event);
 507           return FALSE;
 508         }
 509         demux->discont = TRUE;
 510         demux->new_playlist = TRUE;
 511         demux->do_typefind = TRUE;
 512
 513         gst_hls_demux_change_playlist (demux,
 514             demux->current_download_rate * demux->bitrate_limit);
 515       }
 516
 517       GST_M3U8_CLIENT_LOCK (demux->client);
 518       file = GST_M3U8_MEDIA_FILE (demux->client->current->files->data);
 519       current_sequence = file->sequence;
 520       current_pos = 0;
 521       target_pos = rate > 0 ? start : stop;
 522       /* FIXME: Here we need proper discont handling */
 523       for (walk = demux->client->current->files; walk; walk = walk->next) {
 524         file = walk->data;
 525
 526         current_sequence = file->sequence;
 527         if (current_pos <= target_pos
 528             && target_pos < current_pos + file->duration) {
 529           break;
 530         }
 531         current_pos += file->duration;
 532       }
 533       GST_M3U8_CLIENT_UNLOCK (demux->client);
 534
 535       if (walk == NULL) {
 536         GST_DEBUG_OBJECT (demux, "seeking further than track duration");
 537         current_sequence++;
 538       }
 539
 540       GST_M3U8_CLIENT_LOCK (demux->client);
 541       GST_DEBUG_OBJECT (demux, "seeking to sequence %u",
 542           (guint) current_sequence);
 543       demux->client->sequence = current_sequence;
 544       demux->client->sequence_position = current_pos;
 545       GST_M3U8_CLIENT_UNLOCK (demux->client);
 546
 547       gst_segment_do_seek (&demux->segment, rate, format, flags, start_type,
 548           start, stop_type, stop, NULL);
 549       demux->need_segment = TRUE;
 550
 551       if (flags & GST_SEEK_FLAG_FLUSH) {
 552         GST_DEBUG_OBJECT (demux, "sending flush stop");
 553         gst_pad_push_event (demux->srcpad, gst_event_new_flush_stop (TRUE));
 554       }
 555
 556       demux->stop_updates_task = FALSE;
 557       gst_uri_downloader_reset (demux->downloader);
 558       demux->stop_stream_task = FALSE;
 559
 560       gst_task_start (demux->updates_task);
 561       g_rec_mutex_unlock (&demux->stream_lock);
 562
 563       gst_event_unref (event);
 564       return TRUE;
 565     }
 566     case GST_EVENT_LATENCY:{
 567       /* Upstream and our internal source are irrelevant
 568        * for latency, and we should not fail here to
 569        * configure the latency */
 570       gst_event_unref (event);
 571       return TRUE;
 572     }
 573     default:
 574       break;
 575   }
 576
 577   return gst_pad_event_default (pad, parent, event);
 578 }
 579
 580 static gboolean
 581 gst_hls_demux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 582 {
 583   GstHLSDemux *demux;
 584   GstQuery *query;
 585   gboolean ret;
 586
 587   demux = GST_HLS_DEMUX (parent);
 588
 589   switch (event->type) {
 590     case GST_EVENT_EOS:{
 591       gchar *uri, *playlist = NULL;
 592
 593       if (demux->playlist == NULL) {
 594         GST_WARNING_OBJECT (demux, "Received EOS without a playlist.");
 595         break;
 596       }
 597
 598       GST_DEBUG_OBJECT (demux,
 599           "Got EOS on the sink pad: main playlist fetched");
 600
 601       query = gst_query_new_uri ();
 602       ret = gst_pad_peer_query (demux->sinkpad, query);
 603       if (ret) {
 604         gboolean permanent;
 605         gchar *redirect_uri;
 606
 607         gst_query_parse_uri (query, &uri);
 608         gst_query_parse_uri_redirection (query, &redirect_uri);
 609         gst_query_parse_uri_redirection_permanent (query, &permanent);
 610
 611         if (permanent && redirect_uri) {
 612           gst_hls_demux_set_location (demux, redirect_uri, NULL);
 613         } else {
 614           gst_hls_demux_set_location (demux, uri, redirect_uri);
 615         }
 616         g_free (uri);
 617         g_free (redirect_uri);
 618       }
 619       gst_query_unref (query);
 620
 621       uri = gst_m3u8_client_get_uri (demux->client);
 622       gst_element_post_message (GST_ELEMENT_CAST (demux),
 623           gst_message_new_element (GST_OBJECT_CAST (demux),
 624               gst_structure_new (STATISTICS_MESSAGE_NAME,
 625                   "manifest-uri", G_TYPE_STRING, uri,
 626                   "uri", G_TYPE_STRING, uri,
 627                   "manifest-download-start", GST_TYPE_CLOCK_TIME,
 628                   GST_CLOCK_TIME_NONE,
 629                   "manifest-download-stop", GST_TYPE_CLOCK_TIME,
 630                   gst_util_get_timestamp (), NULL)));
 631       g_free (uri);
 632
 633       playlist = gst_hls_src_buf_to_utf8_playlist (demux->playlist);
 634       demux->playlist = NULL;
 635       if (playlist == NULL) {
 636         GST_WARNING_OBJECT (demux, "Error validating first playlist.");
 637       } else if (!gst_m3u8_client_update (demux->client, playlist)) {
 638         /* In most cases, this will happen if we set a wrong url in the
 639          * source element and we have received the 404 HTML response instead of
 640          * the playlist */
 641         GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid playlist."),
 642             (NULL));
 643         return FALSE;
 644       }
 645
 646       if (!ret && gst_m3u8_client_is_live (demux->client)) {
 647         GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 648             ("Failed querying the playlist uri, "
 649                 "required for live sources."), (NULL));
 650         return FALSE;
 651       }
 652
 653       gst_task_start (demux->updates_task);
 654       gst_event_unref (event);
 655       return TRUE;
 656     }
 657     case GST_EVENT_SEGMENT:
 658       /* Swallow newsegments, we'll push our own */
 659       gst_event_unref (event);
 660       return TRUE;
 661     default:
 662       break;
 663   }
 664
 665   return gst_pad_event_default (pad, parent, event);
 666 }
 667
 668 static gboolean
 669 gst_hls_demux_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
 670 {
 671   GstHLSDemux *hlsdemux;
 672   gboolean ret = FALSE;
 673
 674   if (query == NULL)
 675     return FALSE;
 676
 677   hlsdemux = GST_HLS_DEMUX (parent);
 678
 679   switch (query->type) {
 680     case GST_QUERY_DURATION:{
 681       GstClockTime duration = -1;
 682       GstFormat fmt;
 683
 684       gst_query_parse_duration (query, &fmt, NULL);
 685       if (fmt == GST_FORMAT_TIME) {
 686         duration = gst_m3u8_client_get_duration (hlsdemux->client);
 687         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) {
 688           gst_query_set_duration (query, GST_FORMAT_TIME, duration);
 689           ret = TRUE;
 690         }
 691       }
 692       GST_INFO_OBJECT (hlsdemux, "GST_QUERY_DURATION returns %s with duration %"
 693           GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
 694       break;
 695     }
 696     case GST_QUERY_URI:
 697       if (hlsdemux->client) {
 698         gchar *uri;
 699
 700         /* FIXME: Do we answer with the variant playlist, with the current
 701          * playlist or the the uri of the least downlowaded fragment? */
 702         uri = gst_m3u8_client_get_uri (hlsdemux->client);
 703         gst_query_set_uri (query, uri);
 704         g_free (uri);
 705
 706         ret = TRUE;
 707       }
 708       break;
 709     case GST_QUERY_SEEKING:{
 710       GstFormat fmt;
 711       gint64 stop = -1;
 712
 713       gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL);
 714       GST_INFO_OBJECT (hlsdemux, "Received GST_QUERY_SEEKING with format %d",
 715           fmt);
 716       if (fmt == GST_FORMAT_TIME) {
 717         GstClockTime duration;
 718
 719         duration = gst_m3u8_client_get_duration (hlsdemux->client);
 720         if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0)
 721           stop = duration;
 722
 723         gst_query_set_seeking (query, fmt,
 724             !gst_m3u8_client_is_live (hlsdemux->client), 0, stop);
 725         ret = TRUE;
 726         GST_INFO_OBJECT (hlsdemux, "GST_QUERY_SEEKING returning with stop : %"
 727             GST_TIME_FORMAT, GST_TIME_ARGS (stop));
 728       }
 729       break;
 730     }
 731     default:
 732       /* Don't fordward queries upstream because of the special nature of this
 733        * "demuxer", which relies on the upstream element only to be fed with the
 734        * first playlist */
 735       break;
 736   }
 737
 738   return ret;
 739 }
 740
 741 static GstFlowReturn
 742 gst_hls_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 743 {
 744   GstHLSDemux *demux = GST_HLS_DEMUX (parent);
 745
 746   if (demux->playlist == NULL)
 747     demux->playlist = buf;
 748   else
 749     demux->playlist = gst_buffer_append (demux->playlist, buf);
 750
 751   return GST_FLOW_OK;
 752 }
 753
 754 static void
 755 gst_hls_demux_pause_tasks (GstHLSDemux * demux)
 756 {
 757   if (GST_TASK_STATE (demux->updates_task) != GST_TASK_STOPPED) {
 758     g_mutex_lock (&demux->updates_timed_lock);
 759     demux->stop_updates_task = TRUE;
 760     g_cond_signal (&demux->updates_timed_cond);
 761     g_mutex_unlock (&demux->updates_timed_lock);
 762     gst_uri_downloader_cancel (demux->downloader);
 763     gst_task_pause (demux->updates_task);
 764   }
 765
 766   if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED) {
 767     g_mutex_lock (&demux->download_lock);
 768     demux->stop_stream_task = TRUE;
 769     g_cond_signal (&demux->download_cond);
 770     g_mutex_unlock (&demux->download_lock);
 771     g_mutex_lock (&demux->fragment_download_lock);
 772     demux->download_finished = TRUE;
 773     g_cond_signal (&demux->fragment_download_cond);
 774     g_mutex_unlock (&demux->fragment_download_lock);
 775     gst_task_pause (demux->stream_task);
 776   }
 777 }
 778
 779 static void
 780 gst_hls_demux_stop (GstHLSDemux * demux)
 781 {
 782   if (GST_TASK_STATE (demux->updates_task) != GST_TASK_STOPPED) {
 783     g_mutex_lock (&demux->updates_timed_lock);
 784     demux->stop_updates_task = TRUE;
 785     g_cond_signal (&demux->updates_timed_cond);
 786     g_mutex_unlock (&demux->updates_timed_lock);
 787     gst_uri_downloader_cancel (demux->downloader);
 788     gst_task_stop (demux->updates_task);
 789     g_rec_mutex_lock (&demux->updates_lock);
 790     g_rec_mutex_unlock (&demux->updates_lock);
 791   }
 792
 793   if (GST_TASK_STATE (demux->stream_task) != GST_TASK_STOPPED) {
 794     g_mutex_lock (&demux->download_lock);
 795     demux->stop_stream_task = TRUE;
 796     g_cond_signal (&demux->download_cond);
 797     g_mutex_unlock (&demux->download_lock);
 798     g_mutex_lock (&demux->fragment_download_lock);
 799     demux->download_finished = TRUE;
 800     g_cond_signal (&demux->fragment_download_cond);
 801     g_mutex_unlock (&demux->fragment_download_lock);
 802     gst_task_stop (demux->stream_task);
 803     g_rec_mutex_lock (&demux->stream_lock);
 804     g_rec_mutex_unlock (&demux->stream_lock);
 805   }
 806 }
 807
 808 static GstFlowReturn
 809 _src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 810 {
 811   GstPad *srcpad = (GstPad *) parent;
 812   GstHLSDemux *demux = (GstHLSDemux *) GST_PAD_PARENT (srcpad);
 813   GstFlowReturn ret;
 814   GstCaps *caps;
 815
 816   /* Is it encrypted? */
 817   if (demux->current_key) {
 818     GError *err = NULL;
 819     GstBuffer *tmp_buffer;
 820     gsize available;
 821
 822     /* restart the decrypting lib for a new fragment */
 823     if (demux->reset_crypto) {
 824       GstFragment *key_fragment;
 825       GstBuffer *key_buffer;
 826       GstMapInfo key_info;
 827
 828       /* new key? */
 829       if (demux->key_url && strcmp (demux->key_url, demux->current_key) == 0) {
 830         key_fragment = g_object_ref (demux->key_fragment);
 831       } else {
 832         g_free (demux->key_url);
 833         demux->key_url = NULL;
 834
 835         if (demux->key_fragment)
 836           g_object_unref (demux->key_fragment);
 837         demux->key_fragment = NULL;
 838
 839         GST_INFO_OBJECT (demux, "Fetching key %s", demux->current_key);
 840         key_fragment =
 841             gst_uri_downloader_fetch_uri (demux->downloader,
 842             demux->current_key, demux->client->main ?
 843             demux->client->main->uri : NULL, FALSE, FALSE,
 844             demux->client->current ? demux->client->current->allowcache : TRUE,
 845             &err);
 846         if (key_fragment == NULL)
 847           goto key_failed;
 848         demux->key_url = g_strdup (demux->current_key);
 849         demux->key_fragment = g_object_ref (key_fragment);
 850       }
 851
 852       key_buffer = gst_fragment_get_buffer (key_fragment);
 853       gst_buffer_map (key_buffer, &key_info, GST_MAP_READ);
 854
 855       gst_hls_demux_decrypt_start (demux, key_info.data, demux->current_iv);
 856
 857       gst_buffer_unmap (key_buffer, &key_info);
 858       gst_buffer_unref (key_buffer);
 859       g_object_unref (key_fragment);
 860
 861       demux->reset_crypto = FALSE;
 862     }
 863
 864     gst_adapter_push (demux->adapter, buffer);
 865
 866     /* must be a multiple of 16 */
 867     available = gst_adapter_available (demux->adapter) & (~0xF);
 868
 869     if (available == 0) {
 870       return GST_FLOW_OK;
 871     }
 872
 873     buffer = gst_adapter_take_buffer (demux->adapter, available);
 874     buffer = gst_hls_demux_decrypt_fragment (demux, buffer, &err);
 875     if (buffer == NULL) {
 876       GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Failed to decrypt buffer"),
 877           ("decryption failed %s", err->message));
 878       g_error_free (err);
 879
 880       g_mutex_lock (&demux->fragment_download_lock);
 881       demux->last_ret = GST_FLOW_ERROR;
 882       demux->download_finished = TRUE;
 883       g_cond_signal (&demux->fragment_download_cond);
 884       g_mutex_unlock (&demux->fragment_download_lock);
 885
 886       return GST_FLOW_ERROR;
 887     }
 888
 889     tmp_buffer = demux->pending_buffer;
 890     demux->pending_buffer = buffer;
 891     buffer = tmp_buffer;
 892   }
 893
 894   if (!buffer) {
 895     return GST_FLOW_OK;
 896   }
 897
 898   if (demux->starting_fragment) {
 899     GST_LOG_OBJECT (demux, "set buffer pts=%" GST_TIME_FORMAT,
 900         GST_TIME_ARGS (demux->current_timestamp));
 901     GST_BUFFER_PTS (buffer) = demux->current_timestamp;
 902
 903     if (demux->segment.rate < 0)
 904       /* Set DISCONT flag for every first buffer in reverse playback mode
 905        * as each fragment for its own has to be reversed */
 906       demux->discont = TRUE;
 907     demux->starting_fragment = FALSE;
 908     demux->segment.position = GST_BUFFER_PTS (buffer);
 909   } else {
 910     GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE;
 911   }
 912
 913   GST_BUFFER_DURATION (buffer) = GST_CLOCK_TIME_NONE;
 914   GST_BUFFER_DTS (buffer) = GST_CLOCK_TIME_NONE;
 915
 916   /* We actually need to do this every time we switch bitrate */
 917   if (G_UNLIKELY (demux->do_typefind)) {
 918     caps = gst_type_find_helper_for_buffer (NULL, buffer, NULL);
 919     if (G_UNLIKELY (!caps)) {
 920       GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND,
 921           ("Could not determine type of stream"), (NULL));
 922       gst_buffer_unref (buffer);
 923       g_mutex_lock (&demux->fragment_download_lock);
 924       demux->last_ret = GST_FLOW_NOT_NEGOTIATED;
 925       demux->download_finished = TRUE;
 926       g_cond_signal (&demux->fragment_download_cond);
 927       g_mutex_unlock (&demux->fragment_download_lock);
 928
 929       return GST_FLOW_NOT_NEGOTIATED;
 930     }
 931
 932     if (!demux->input_caps || !gst_caps_is_equal (caps, demux->input_caps)) {
 933       gst_caps_replace (&demux->input_caps, caps);
 934       GST_INFO_OBJECT (demux, "Input source caps: %" GST_PTR_FORMAT,
 935           demux->input_caps);
 936     }
 937     gst_pad_set_caps (srcpad, caps);
 938     demux->do_typefind = FALSE;
 939     gst_caps_unref (caps);
 940   }
 941
 942   if (demux->discont) {
 943     GST_DEBUG_OBJECT (demux, "Marking fragment as discontinuous");
 944     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
 945     demux->discont = FALSE;
 946   } else {
 947     GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
 948   }
 949
 950   demux->starting_fragment = FALSE;
 951
 952   if (demux->need_segment) {
 953     /* And send a newsegment */
 954     GST_DEBUG_OBJECT (demux, "Sending segment event: %"
 955         GST_SEGMENT_FORMAT, &demux->segment);
 956     gst_pad_push_event (demux->srcpad, gst_event_new_segment (&demux->segment));
 957     demux->need_segment = FALSE;
 958   }
 959
 960   /* accumulate time and size to get this chunk */
 961   demux->download_total_time +=
 962       g_get_monotonic_time () - demux->download_start_time;
 963   demux->download_total_bytes += gst_buffer_get_size (buffer);
 964
 965   ret = gst_proxy_pad_chain_default (pad, parent, buffer);
 966   demux->download_start_time = g_get_monotonic_time ();
 967
 968   if (ret != GST_FLOW_OK) {
 969     if (ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS) {
 970       GST_ELEMENT_ERROR (demux, STREAM, FAILED, (NULL),
 971           ("stream stopped, reason %s", gst_flow_get_name (ret)));
 972       gst_pad_push_event (demux->srcpad, gst_event_new_eos ());
 973     } else {
 974       GST_DEBUG_OBJECT (demux, "stream stopped, reason %s",
 975           gst_flow_get_name (ret));
 976     }
 977     gst_hls_demux_pause_tasks (demux);
 978
 979     g_mutex_lock (&demux->fragment_download_lock);
 980     demux->last_ret = ret;
 981     demux->download_finished = TRUE;
 982     g_cond_signal (&demux->fragment_download_cond);
 983     g_mutex_unlock (&demux->fragment_download_lock);
 984   }
 985
 986   /* avoid having the source handle the same error again */
 987   ret = GST_FLOW_OK;
 988
 989   return ret;
 990
 991 key_failed:
 992   /* TODO Raise this error to the user */
 993   GST_WARNING_OBJECT (demux, "Failed to decrypt data");
 994   g_mutex_lock (&demux->fragment_download_lock);
 995   demux->last_ret = GST_FLOW_ERROR;
 996   demux->download_finished = TRUE;
 997   g_cond_signal (&demux->fragment_download_cond);
 998   g_mutex_unlock (&demux->fragment_download_lock);
 999   return GST_FLOW_ERROR;
1000 }
1001
1002 static gboolean
1003 _src_event (GstPad * pad, GstObject * parent, GstEvent * event)
1004 {
1005   GstPad *srcpad = GST_PAD_CAST (parent);
1006   GstHLSDemux *demux = (GstHLSDemux *) GST_PAD_PARENT (srcpad);;
1007
1008   switch (GST_EVENT_TYPE (event)) {
1009     case GST_EVENT_EOS:{
1010       GstFlowReturn ret;
1011       if (demux->current_key)
1012         gst_hls_demux_decrypt_end (demux);
1013
1014       g_mutex_lock (&demux->fragment_download_lock);
1015       ret = demux->last_ret;
1016       g_mutex_unlock (&demux->fragment_download_lock);
1017
1018       /* ideally this should be empty, but this eos might have been
1019        * caused by an error on the source element */
1020       GST_DEBUG_OBJECT (demux, "Data still on the adapter when EOS was received"
1021           ": %" G_GSIZE_FORMAT, gst_adapter_available (demux->adapter));
1022       gst_adapter_clear (demux->adapter);
1023
1024       /* pending buffer is only used for encrypted streams */
1025       if (ret == GST_FLOW_OK) {
1026         if (demux->pending_buffer) {
1027           GstMapInfo info;
1028           gsize unpadded_size;
1029
1030           /* Handle pkcs7 unpadding here */
1031           gst_buffer_map (demux->pending_buffer, &info, GST_MAP_READ);
1032           unpadded_size = info.size - info.data[info.size - 1];
1033           gst_buffer_unmap (demux->pending_buffer, &info);
1034
1035           gst_buffer_resize (demux->pending_buffer, 0, unpadded_size);
1036
1037           demux->download_total_time +=
1038               g_get_monotonic_time () - demux->download_start_time;
1039           demux->download_total_bytes +=
1040               gst_buffer_get_size (demux->pending_buffer);
1041           ret = gst_pad_push (demux->srcpad, demux->pending_buffer);
1042
1043           demux->pending_buffer = NULL;
1044         }
1045       } else {
1046         if (demux->pending_buffer)
1047           gst_buffer_unref (demux->pending_buffer);
1048         demux->pending_buffer = NULL;
1049       }
1050
1051       GST_DEBUG_OBJECT (demux, "Fragment download finished");
1052
1053       g_mutex_lock (&demux->fragment_download_lock);
1054       demux->last_ret = ret;
1055       demux->download_finished = TRUE;
1056       g_cond_signal (&demux->fragment_download_cond);
1057       g_mutex_unlock (&demux->fragment_download_lock);
1058       break;
1059     }
1060     default:
1061       break;
1062   }
1063
1064   gst_event_unref (event);
1065
1066   return TRUE;
1067 }
1068
1069 static gboolean
1070 _src_query (GstPad * pad, GstObject * parent, GstQuery * query)
1071 {
1072   switch (GST_QUERY_TYPE (query)) {
1073     case GST_QUERY_ALLOCATION:
1074       return FALSE;
1075       break;
1076     default:
1077       break;
1078   }
1079
1080   return gst_pad_query_default (pad, parent, query);
1081 }
1082
1083 static gboolean
1084 _hls_demux_pad_remove_eos_sticky (GstPad * pad, GstEvent ** event,
1085     gpointer udata)
1086 {
1087   if (GST_EVENT_TYPE (*event) == GST_EVENT_EOS) {
1088     gst_event_replace (event, NULL);
1089     return FALSE;
1090   }
1091   return TRUE;
1092 }
1093
1094 static void
1095 gst_hls_demux_stream_clear_eos_state (GstHLSDemux * demux)
1096 {
1097   GstPad *internal_pad;
1098
1099   internal_pad =
1100       GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (demux->srcpad)));
1101   gst_pad_sticky_events_foreach (internal_pad,
1102       _hls_demux_pad_remove_eos_sticky, NULL);
1103   GST_OBJECT_FLAG_UNSET (internal_pad, GST_PAD_FLAG_EOS);
1104
1105   gst_object_unref (internal_pad);
1106 }
1107
1108 static void
1109 switch_pads (GstHLSDemux * demux)
1110 {
1111   GstPad *oldpad = demux->srcpad;
1112   GstEvent *event;
1113   gchar *stream_id;
1114   gchar *name;
1115   GstPadTemplate *tmpl;
1116   GstProxyPad *internal_pad;
1117
1118   GST_DEBUG_OBJECT (demux, "Switching pad (oldpad:%p)", oldpad);
1119
1120   if (oldpad) {
1121     gst_ghost_pad_set_target (GST_GHOST_PAD_CAST (oldpad), NULL);
1122   }
1123
1124   /* First create and activate new pad */
1125   name = g_strdup_printf ("src_%u", demux->srcpad_counter++);
1126   tmpl = gst_static_pad_template_get (&srctemplate);
1127   demux->srcpad =
1128       gst_ghost_pad_new_from_template (name, demux->src_srcpad, tmpl);
1129   gst_object_unref (tmpl);
1130   g_free (name);
1131
1132   /* set up our internal pad to drop all events from
1133    * the http src we don't care about. On the chain function
1134    * we just push the buffer forward, but this way hls can get
1135    * the flow return from downstream */
1136   internal_pad = gst_proxy_pad_get_internal (GST_PROXY_PAD (demux->srcpad));
1137   gst_pad_set_chain_function (GST_PAD_CAST (internal_pad), _src_chain);
1138   gst_pad_set_event_function (GST_PAD_CAST (internal_pad), _src_event);
1139   /* need to set query otherwise deadlocks happen with allocation queries */
1140   gst_pad_set_query_function (GST_PAD_CAST (internal_pad), _src_query);
1141   gst_object_unref (internal_pad);
1142
1143   gst_pad_set_event_function (demux->srcpad,
1144       GST_DEBUG_FUNCPTR (gst_hls_demux_src_event));
1145   gst_pad_set_query_function (demux->srcpad,
1146       GST_DEBUG_FUNCPTR (gst_hls_demux_src_query));
1147   gst_pad_use_fixed_caps (demux->srcpad);
1148   gst_pad_set_active (demux->srcpad, TRUE);
1149
1150   stream_id =
1151       gst_pad_create_stream_id (demux->srcpad, GST_ELEMENT_CAST (demux), NULL);
1152
1153   event = gst_pad_get_sticky_event (demux->sinkpad, GST_EVENT_STREAM_START, 0);
1154   if (event) {
1155     if (gst_event_parse_group_id (event, &demux->group_id))
1156       demux->have_group_id = TRUE;
1157     else
1158       demux->have_group_id = FALSE;
1159     gst_event_unref (event);
1160   } else if (!demux->have_group_id) {
1161     demux->have_group_id = TRUE;
1162     demux->group_id = gst_util_group_id_next ();
1163   }
1164   event = gst_event_new_stream_start (stream_id);
1165   if (demux->have_group_id)
1166     gst_event_set_group_id (event, demux->group_id);
1167
1168   gst_pad_push_event (demux->srcpad, event);
1169   g_free (stream_id);
1170
1171   gst_element_add_pad (GST_ELEMENT (demux), demux->srcpad);
1172
1173   gst_element_no_more_pads (GST_ELEMENT (demux));
1174
1175   demux->new_playlist = FALSE;
1176
1177   if (oldpad) {
1178     /* Push out EOS */
1179     gst_pad_push_event (oldpad, gst_event_new_eos ());
1180     gst_pad_set_active (oldpad, FALSE);
1181     gst_element_remove_pad (GST_ELEMENT (demux), oldpad);
1182   }
1183 }
1184
1185 static void
1186 gst_hls_demux_configure_src_pad (GstHLSDemux * demux)
1187 {
1188   if (G_UNLIKELY (!demux->srcpad || demux->new_playlist)) {
1189     switch_pads (demux);
1190     demux->need_segment = TRUE;
1191   }
1192 }
1193
1194 static void
1195 gst_hls_demux_stream_loop (GstHLSDemux * demux)
1196 {
1197   gboolean end_of_playlist;
1198   GError *err = NULL;
1199
1200   /* This task will download fragments as fast as possible, sends
1201    * SEGMENT and CAPS events and switches pads if necessary.
1202    * If downloading a fragment fails we try again up to 3 times
1203    * after waiting a bit. If we're at the end of the playlist
1204    * we wait for the playlist to update before getting the next
1205    * fragment.
1206    */
1207   GST_DEBUG_OBJECT (demux, "Enter task");
1208
1209   if (demux->stop_stream_task)
1210     goto pause_task;
1211
1212   /* Check if we're done with our segment */
1213   if (demux->segment.rate > 0) {
1214     if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)
1215         && demux->segment.position >= demux->segment.stop)
1216       goto end_of_playlist;
1217   } else {
1218     if (GST_CLOCK_TIME_IS_VALID (demux->segment.start)
1219         && demux->segment.position < demux->segment.start)
1220       goto end_of_playlist;
1221   }
1222
1223   demux->next_download = g_get_monotonic_time ();
1224   if (!gst_hls_demux_get_next_fragment (demux, &end_of_playlist, &err)) {
1225     if (demux->stop_stream_task) {
1226       g_clear_error (&err);
1227       goto pause_task;
1228     }
1229
1230     if (end_of_playlist) {
1231       if (!gst_m3u8_client_is_live (demux->client)) {
1232         GST_DEBUG_OBJECT (demux, "End of playlist");
1233         demux->end_of_playlist = TRUE;
1234         goto end_of_playlist;
1235       } else {
1236         g_mutex_lock (&demux->download_lock);
1237
1238         /* Wait until we're cancelled or there's something for
1239          * us to download in the playlist or the playlist
1240          * became non-live */
1241         while (TRUE) {
1242           if (demux->stop_stream_task) {
1243             g_mutex_unlock (&demux->download_lock);
1244             goto pause_task;
1245           }
1246
1247           /* Got a new fragment or not live anymore? */
1248           if (gst_m3u8_client_get_next_fragment (demux->client, NULL, NULL,
1249                   NULL, NULL, NULL, NULL, NULL, NULL, demux->segment.rate > 0)
1250               || !gst_m3u8_client_is_live (demux->client))
1251             break;
1252
1253           GST_DEBUG_OBJECT (demux,
1254               "No fragment left but live playlist, wait a bit");
1255           g_cond_wait (&demux->download_cond, &demux->download_lock);
1256         }
1257         g_mutex_unlock (&demux->download_lock);
1258         GST_DEBUG_OBJECT (demux, "Retrying now");
1259         return;
1260       }
1261     } else {
1262       demux->download_failed_count++;
1263       if (demux->download_failed_count <= DEFAULT_FAILED_COUNT) {
1264         GST_WARNING_OBJECT (demux, "Could not fetch the next fragment");
1265         g_clear_error (&err);
1266
1267         /* First try to update the playlist for non-live playlists
1268          * in case the URIs have changed in the meantime. But only
1269          * try it the first time, after that we're going to wait a
1270          * a bit to not flood the server */
1271         if (demux->download_failed_count == 1
1272             && !gst_m3u8_client_is_live (demux->client)
1273             && gst_hls_demux_update_playlist (demux, FALSE, &err)) {
1274           /* Retry immediately, the playlist actually has changed */
1275           GST_DEBUG_OBJECT (demux, "Updated the playlist");
1276           return;
1277         } else {
1278           /* Wait half the fragment duration before retrying */
1279           demux->next_download +=
1280               gst_util_uint64_scale
1281               (gst_m3u8_client_get_current_fragment_duration (demux->client),
1282               G_USEC_PER_SEC, 2 * GST_SECOND);
1283         }
1284
1285         g_clear_error (&err);
1286
1287         g_mutex_lock (&demux->download_lock);
1288         if (demux->stop_stream_task) {
1289           g_mutex_unlock (&demux->download_lock);
1290           goto pause_task;
1291         }
1292         g_cond_wait_until (&demux->download_cond, &demux->download_lock,
1293             demux->next_download);
1294         g_mutex_unlock (&demux->download_lock);
1295         GST_DEBUG_OBJECT (demux, "Retrying now");
1296
1297         /* Refetch the playlist now after we waited */
1298         if (!gst_m3u8_client_is_live (demux->client)
1299             && gst_hls_demux_update_playlist (demux, FALSE, &err)) {
1300           GST_DEBUG_OBJECT (demux, "Updated the playlist");
1301         }
1302         return;
1303       } else {
1304         GST_ELEMENT_ERROR_FROM_ERROR (demux,
1305             "Could not fetch the next fragment", err);
1306         goto pause_task;
1307       }
1308     }
1309   } else {
1310     demux->download_failed_count = 0;
1311     gst_m3u8_client_advance_fragment (demux->client, demux->segment.rate > 0);
1312
1313     if (demux->stop_updates_task) {
1314       goto pause_task;
1315     }
1316   }
1317
1318   if (demux->stop_updates_task) {
1319     goto pause_task;
1320   }
1321
1322   /* try to switch to another bitrate if needed */
1323   /* FIXME: Currently several issues have be found when letting bitrate adaptation
1324    * happen using trick modes (such as 'All streams finished without buffers') and
1325    * the adaptive algorithm does not properly behave. */
1326   if (demux->segment.rate == 1.0)
1327     gst_hls_demux_switch_playlist (demux);
1328   demux->download_total_bytes = 0;
1329   demux->download_total_time = 0;
1330
1331   GST_DEBUG_OBJECT (demux, "Finished pushing fragment");
1332
1333   return;
1334
1335 end_of_playlist:
1336   {
1337     GST_DEBUG_OBJECT (demux, "Reached end of playlist, sending EOS");
1338
1339     gst_hls_demux_configure_src_pad (demux);
1340
1341     gst_pad_push_event (demux->srcpad, gst_event_new_eos ());
1342     gst_hls_demux_pause_tasks (demux);
1343     return;
1344   }
1345
1346 pause_task:
1347   {
1348     GST_DEBUG_OBJECT (demux, "Pause task");
1349     /* Pausing a stopped task will start it */
1350     gst_hls_demux_pause_tasks (demux);
1351     return;
1352   }
1353 }
1354
1355 static void
1356 gst_hls_demux_reset (GstHLSDemux * demux, gboolean dispose)
1357 {
1358   demux->end_of_playlist = FALSE;
1359   demux->stop_updates_task = FALSE;
1360   demux->do_typefind = TRUE;
1361
1362   demux->download_failed_count = 0;
1363
1364   g_free (demux->key_url);
1365   demux->key_url = NULL;
1366
1367   if (demux->key_fragment)
1368     g_object_unref (demux->key_fragment);
1369   demux->key_fragment = NULL;
1370
1371   if (demux->input_caps) {
1372     gst_caps_unref (demux->input_caps);
1373     demux->input_caps = NULL;
1374   }
1375
1376   if (demux->playlist) {
1377     gst_buffer_unref (demux->playlist);
1378     demux->playlist = NULL;
1379   }
1380
1381   if (demux->client) {
1382     gst_m3u8_client_free (demux->client);
1383     demux->client = NULL;
1384   }
1385
1386   if (!dispose) {
1387     demux->client = gst_m3u8_client_new ("", NULL);
1388   }
1389
1390   gst_segment_init (&demux->segment, GST_FORMAT_TIME);
1391   demux->need_segment = TRUE;
1392   demux->discont = TRUE;
1393
1394   demux->have_group_id = FALSE;
1395   demux->group_id = G_MAXUINT;
1396
1397   demux->srcpad_counter = 0;
1398   if (demux->srcpad) {
1399     gst_element_remove_pad (GST_ELEMENT_CAST (demux), demux->srcpad);
1400     demux->srcpad = NULL;
1401   }
1402
1403   if (demux->src) {
1404     gst_element_set_state (demux->src, GST_STATE_NULL);
1405   }
1406
1407   g_clear_error (&demux->last_error);
1408
1409   if (demux->adapter)
1410     gst_adapter_clear (demux->adapter);
1411   if (demux->pending_buffer)
1412     gst_buffer_unref (demux->pending_buffer);
1413   demux->pending_buffer = NULL;
1414   if (demux->current_key) {
1415     g_free (demux->current_key);
1416     demux->current_key = NULL;
1417   }
1418   if (demux->current_iv) {
1419     g_free (demux->current_iv);
1420     demux->current_iv = NULL;
1421   }
1422   gst_hls_demux_decrypt_end (demux);
1423
1424   demux->current_download_rate = -1;
1425 }
1426
1427 static gboolean
1428 gst_hls_demux_set_location (GstHLSDemux * demux, const gchar * uri,
1429     const gchar * base_uri)
1430 {
1431   if (demux->client)
1432     gst_m3u8_client_free (demux->client);
1433   demux->client = gst_m3u8_client_new (uri, base_uri);
1434   GST_INFO_OBJECT (demux, "Changed location: %s (base uri: %s)", uri,
1435       GST_STR_NULL (base_uri));
1436   return TRUE;
1437 }
1438
1439 void
1440 gst_hls_demux_updates_loop (GstHLSDemux * demux)
1441 {
1442   /* Loop for updating of the playlist. This periodically checks if
1443    * the playlist is updated and does so, then signals the streaming
1444    * thread in case it can continue downloading now.
1445    * For non-live playlists this thread is not doing much else than
1446    * setting up the initial playlist and then stopping. */
1447
1448   /* block until the next scheduled update or the signal to quit this thread */
1449   GST_DEBUG_OBJECT (demux, "Started updates task");
1450
1451   /* If this playlist is a variant playlist, select the first one
1452    * and update it */
1453   if (gst_m3u8_client_has_variant_playlist (demux->client)) {
1454     GstM3U8 *child = NULL;
1455     GError *err = NULL;
1456
1457     if (demux->connection_speed == 0) {
1458       GST_M3U8_CLIENT_LOCK (demux->client);
1459       child = demux->client->main->current_variant->data;
1460       GST_M3U8_CLIENT_UNLOCK (demux->client);
1461     } else {
1462       GList *tmp = gst_m3u8_client_get_playlist_for_bitrate (demux->client,
1463           demux->connection_speed);
1464
1465       child = GST_M3U8 (tmp->data);
1466     }
1467
1468     gst_m3u8_client_set_current (demux->client, child);
1469     if (!gst_hls_demux_update_playlist (demux, FALSE, &err)) {
1470       GST_ELEMENT_ERROR_FROM_ERROR (demux, "Could not fetch the child playlist",
1471           err);
1472       goto error;
1473     }
1474   }
1475
1476   if (!gst_m3u8_client_is_live (demux->client)) {
1477     GstClockTime duration = gst_m3u8_client_get_duration (demux->client);
1478
1479     GST_DEBUG_OBJECT (demux, "Sending duration message : %" GST_TIME_FORMAT,
1480         GST_TIME_ARGS (duration));
1481     if (duration != GST_CLOCK_TIME_NONE)
1482       gst_element_post_message (GST_ELEMENT (demux),
1483           gst_message_new_duration_changed (GST_OBJECT (demux)));
1484   }
1485
1486   /* Now start stream task */
1487   gst_task_start (demux->stream_task);
1488
1489   demux->next_update =
1490       g_get_monotonic_time () +
1491       gst_util_uint64_scale (gst_m3u8_client_get_target_duration
1492       (demux->client), G_USEC_PER_SEC, GST_SECOND);
1493
1494   /* Updating playlist only needed for live playlists */
1495   while (gst_m3u8_client_is_live (demux->client)) {
1496     GError *err = NULL;
1497
1498     /* Wait here until we should do the next update or we're cancelled */
1499     GST_DEBUG_OBJECT (demux, "Wait for next playlist update");
1500     g_mutex_lock (&demux->updates_timed_lock);
1501     if (demux->stop_updates_task) {
1502       g_mutex_unlock (&demux->updates_timed_lock);
1503       goto quit;
1504     }
1505     g_cond_wait_until (&demux->updates_timed_cond, &demux->updates_timed_lock,
1506         demux->next_update);
1507     if (demux->stop_updates_task) {
1508       g_mutex_unlock (&demux->updates_timed_lock);
1509       goto quit;
1510     }
1511     g_mutex_unlock (&demux->updates_timed_lock);
1512
1513     GST_DEBUG_OBJECT (demux, "Updating playlist");
1514     if (!gst_hls_demux_update_playlist (demux, TRUE, &err)) {
1515       if (demux->stop_updates_task)
1516         goto quit;
1517       demux->client->update_failed_count++;
1518       if (demux->client->update_failed_count <= DEFAULT_FAILED_COUNT) {
1519         GST_WARNING_OBJECT (demux, "Could not update the playlist");
1520         demux->next_update =
1521             g_get_monotonic_time () +
1522             gst_util_uint64_scale (gst_m3u8_client_get_target_duration
1523             (demux->client), G_USEC_PER_SEC, 2 * GST_SECOND);
1524       } else {
1525         GST_ELEMENT_ERROR_FROM_ERROR (demux, "Could not update playlist", err);
1526         goto error;
1527       }
1528     } else {
1529       GST_DEBUG_OBJECT (demux, "Updated playlist successfully");
1530       demux->next_update =
1531           g_get_monotonic_time () +
1532           gst_util_uint64_scale (gst_m3u8_client_get_target_duration
1533           (demux->client), G_USEC_PER_SEC, GST_SECOND);
1534       /* Wake up download task */
1535       g_mutex_lock (&demux->download_lock);
1536       g_cond_signal (&demux->download_cond);
1537       g_mutex_unlock (&demux->download_lock);
1538     }
1539   }
1540
1541 quit:
1542   {
1543     GST_DEBUG_OBJECT (demux, "Stopped updates task");
1544     gst_task_pause (demux->updates_task);
1545     return;
1546   }
1547
1548 error:
1549   {
1550     GST_DEBUG_OBJECT (demux, "Stopped updates task because of error");
1551     gst_hls_demux_pause_tasks (demux);
1552   }
1553 }
1554
1555 static gchar *
1556 gst_hls_src_buf_to_utf8_playlist (GstBuffer * buf)
1557 {
1558   GstMapInfo info;
1559   gchar *playlist;
1560
1561   if (!gst_buffer_map (buf, &info, GST_MAP_READ))
1562     goto map_error;
1563
1564   if (!g_utf8_validate ((gchar *) info.data, info.size, NULL))
1565     goto validate_error;
1566
1567   /* alloc size + 1 to end with a null character */
1568   playlist = g_malloc0 (info.size + 1);
1569   memcpy (playlist, info.data, info.size);
1570
1571   gst_buffer_unmap (buf, &info);
1572   gst_buffer_unref (buf);
1573   return playlist;
1574
1575 validate_error:
1576   gst_buffer_unmap (buf, &info);
1577 map_error:
1578   gst_buffer_unref (buf);
1579   return NULL;
1580 }
1581
1582 static gboolean
1583 gst_hls_demux_update_playlist (GstHLSDemux * demux, gboolean update,
1584     GError ** err)
1585 {
1586   GstFragment *download;
1587   GstBuffer *buf;
1588   gchar *playlist;
1589   gboolean main_checked = FALSE, updated = FALSE;
1590   gchar *uri, *main_uri;
1591
1592 retry:
1593   uri = gst_m3u8_client_get_current_uri (demux->client);
1594   main_uri = gst_m3u8_client_get_uri (demux->client);
1595   download =
1596       gst_uri_downloader_fetch_uri (demux->downloader, uri,
1597       main_uri, TRUE, TRUE, TRUE, err);
1598   g_free (main_uri);
1599   if (download == NULL) {
1600     if (update && !main_checked
1601         && gst_m3u8_client_has_variant_playlist (demux->client)
1602         && gst_m3u8_client_has_main (demux->client)) {
1603       GError *err2 = NULL;
1604       main_uri = gst_m3u8_client_get_uri (demux->client);
1605       GST_INFO_OBJECT (demux,
1606           "Updating playlist %s failed, attempt to refresh variant playlist %s",
1607           uri, main_uri);
1608       download =
1609           gst_uri_downloader_fetch_uri (demux->downloader,
1610           main_uri, NULL, TRUE, TRUE, TRUE, &err2);
1611       g_free (main_uri);
1612       g_clear_error (&err2);
1613       if (download != NULL) {
1614         gchar *base_uri;
1615
1616         buf = gst_fragment_get_buffer (download);
1617         playlist = gst_hls_src_buf_to_utf8_playlist (buf);
1618
1619         if (playlist == NULL) {
1620           GST_WARNING_OBJECT (demux,
1621               "Failed to validate variant playlist encoding");
1622           return FALSE;
1623         }
1624
1625         g_free (uri);
1626         if (download->redirect_permanent && download->redirect_uri) {
1627           uri = download->redirect_uri;
1628           base_uri = NULL;
1629         } else {
1630           uri = download->uri;
1631           base_uri = download->redirect_uri;
1632         }
1633
1634         if (!gst_m3u8_client_update_variant_playlist (demux->client, playlist,
1635                 uri, base_uri)) {
1636           GST_WARNING_OBJECT (demux, "Failed to update the variant playlist");
1637           return FALSE;
1638         }
1639
1640         g_object_unref (download);
1641
1642         g_clear_error (err);
1643         main_checked = TRUE;
1644         goto retry;
1645       } else {
1646         g_free (uri);
1647         return FALSE;
1648       }
1649     } else {
1650       g_free (uri);
1651       return FALSE;
1652     }
1653   }
1654
1655   g_free (uri);
1656
1657   /* Set the base URI of the playlist to the redirect target if any */
1658   GST_M3U8_CLIENT_LOCK (demux->client);
1659   g_free (demux->client->current->uri);
1660   g_free (demux->client->current->base_uri);
1661   if (download->redirect_permanent && download->redirect_uri) {
1662     demux->client->current->uri = g_strdup (download->redirect_uri);
1663     demux->client->current->base_uri = NULL;
1664   } else {
1665     demux->client->current->uri = g_strdup (download->uri);
1666     demux->client->current->base_uri = g_strdup (download->redirect_uri);
1667   }
1668   GST_M3U8_CLIENT_UNLOCK (demux->client);
1669
1670   buf = gst_fragment_get_buffer (download);
1671   playlist = gst_hls_src_buf_to_utf8_playlist (buf);
1672
1673   if (playlist == NULL) {
1674     GST_WARNING_OBJECT (demux, "Couldn't validate playlist encoding");
1675     g_set_error (err, GST_STREAM_ERROR, GST_STREAM_ERROR_WRONG_TYPE,
1676         "Couldn't validate playlist encoding");
1677     g_object_unref (download);
1678     return FALSE;
1679   }
1680
1681   updated = gst_m3u8_client_update (demux->client, playlist);
1682   if (!updated) {
1683     GST_WARNING_OBJECT (demux, "Couldn't update playlist");
1684     g_set_error (err, GST_STREAM_ERROR, GST_STREAM_ERROR_FAILED,
1685         "Couldn't update playlist");
1686     g_object_unref (download);
1687     return FALSE;
1688   }
1689
1690   uri = gst_m3u8_client_get_current_uri(demux->client);
1691   main_uri = gst_m3u8_client_get_uri (demux->client);
1692   gst_element_post_message (GST_ELEMENT_CAST (demux),
1693       gst_message_new_element (GST_OBJECT_CAST (demux),
1694           gst_structure_new (STATISTICS_MESSAGE_NAME,
1695               "manifest-uri", G_TYPE_STRING, main_uri,
1696               "uri", G_TYPE_STRING, uri,
1697               "manifest-download-start", GST_TYPE_CLOCK_TIME,
1698               download->download_start_time,
1699               "manifest-download-stop", GST_TYPE_CLOCK_TIME,
1700               download->download_stop_time, NULL)));
1701   g_free (main_uri);
1702   g_free (uri);
1703
1704   g_object_unref (download);
1705
1706   GST_M3U8_CLIENT_LOCK (demux->client);
1707
1708   /* If it's a live source, do not let the sequence number go beyond
1709    * three fragments before the end of the list */
1710   if (update == FALSE && demux->client->current &&
1711       GST_M3U8_CLIENT_IS_LIVE (demux->client)) {
1712     gint64 last_sequence;
1713
1714     last_sequence =
1715         GST_M3U8_MEDIA_FILE (g_list_last (demux->client->current->
1716             files)->data)->sequence;
1717
1718     if (demux->client->sequence >= last_sequence - 3) {
1719       GST_DEBUG_OBJECT (demux, "Sequence is beyond playlist. Moving back to %u",
1720           (guint) (last_sequence - 3));
1721       demux->need_segment = TRUE;
1722       demux->client->sequence = last_sequence - 3;
1723     }
1724   } else if (demux->client->current && !GST_M3U8_CLIENT_IS_LIVE (demux->client)) {
1725     GstClockTime current_pos, target_pos;
1726     guint sequence = 0;
1727     GList *walk;
1728
1729     /* Sequence numbers are not guaranteed to be the same in different
1730      * playlists, so get the correct fragment here based on the current
1731      * position
1732      */
1733     current_pos = 0;
1734     target_pos = demux->segment.position;
1735     for (walk = demux->client->current->files; walk; walk = walk->next) {
1736       GstM3U8MediaFile *file = walk->data;
1737
1738       sequence = file->sequence;
1739       if (current_pos <= target_pos
1740           && target_pos < current_pos + file->duration) {
1741         break;
1742       }
1743       current_pos += file->duration;
1744     }
1745     /* End of playlist */
1746     if (!walk)
1747       sequence++;
1748     demux->client->sequence = sequence;
1749     demux->client->sequence_position = current_pos;
1750   }
1751
1752   GST_M3U8_CLIENT_UNLOCK (demux->client);
1753
1754   return updated;
1755 }
1756
1757 static gboolean
1758 gst_hls_demux_change_playlist (GstHLSDemux * demux, guint max_bitrate)
1759 {
1760   GList *previous_variant, *current_variant;
1761   gint old_bandwidth, new_bandwidth;
1762
1763   /* If user specifies a connection speed never use a playlist with a bandwidth
1764    * superior than it */
1765   if (demux->connection_speed != 0 && max_bitrate > demux->connection_speed)
1766     max_bitrate = demux->connection_speed;
1767
1768   previous_variant = demux->client->main->current_variant;
1769   current_variant = gst_m3u8_client_get_playlist_for_bitrate (demux->client,
1770       max_bitrate);
1771
1772   GST_M3U8_CLIENT_LOCK (demux->client);
1773
1774 retry_failover_protection:
1775   old_bandwidth = GST_M3U8 (previous_variant->data)->bandwidth;
1776   new_bandwidth = GST_M3U8 (current_variant->data)->bandwidth;
1777
1778   /* Don't do anything else if the playlist is the same */
1779   if (new_bandwidth == old_bandwidth) {
1780     GST_M3U8_CLIENT_UNLOCK (demux->client);
1781     return TRUE;
1782   }
1783
1784   demux->client->main->current_variant = current_variant;
1785   GST_M3U8_CLIENT_UNLOCK (demux->client);
1786
1787   gst_m3u8_client_set_current (demux->client, current_variant->data);
1788
1789   GST_INFO_OBJECT (demux, "Client was on %dbps, max allowed is %dbps, switching"
1790       " to bitrate %dbps", old_bandwidth, max_bitrate, new_bandwidth);
1791   demux->discont = TRUE;
1792   demux->new_playlist = TRUE;
1793
1794   if (gst_hls_demux_update_playlist (demux, FALSE, NULL)) {
1795     gchar *uri, *main_uri;
1796     uri = gst_m3u8_client_get_current_uri (demux->client);
1797     main_uri = gst_m3u8_client_get_uri (demux->client);
1798     gst_element_post_message (GST_ELEMENT_CAST (demux),
1799         gst_message_new_element (GST_OBJECT_CAST (demux),
1800             gst_structure_new (STATISTICS_MESSAGE_NAME,
1801                 "manifest-uri", G_TYPE_STRING, main_uri,
1802                 "uri", G_TYPE_STRING, uri,
1803                 "bitrate", G_TYPE_INT, new_bandwidth, NULL)));
1804     g_free (uri);
1805     g_free (main_uri);
1806   } else {
1807     GList *failover = NULL;
1808
1809     GST_INFO_OBJECT (demux, "Unable to update playlist. Switching back");
1810     GST_M3U8_CLIENT_LOCK (demux->client);
1811
1812     failover = g_list_previous (current_variant);
1813     if (failover && new_bandwidth == GST_M3U8 (failover->data)->bandwidth) {
1814       current_variant = failover;
1815       goto retry_failover_protection;
1816     }
1817
1818     demux->client->main->current_variant = previous_variant;
1819     GST_M3U8_CLIENT_UNLOCK (demux->client);
1820     gst_m3u8_client_set_current (demux->client, previous_variant->data);
1821     /*  Try a lower bitrate (or stop if we just tried the lowest) */
1822     if (GST_M3U8 (previous_variant->data)->iframe && new_bandwidth ==
1823         GST_M3U8 (g_list_first (demux->client->main->iframe_lists)->data)->
1824         bandwidth)
1825       return FALSE;
1826     else if (!GST_M3U8 (previous_variant->data)->iframe && new_bandwidth ==
1827         GST_M3U8 (g_list_first (demux->client->main->lists)->data)->bandwidth)
1828       return FALSE;
1829     else
1830       return gst_hls_demux_change_playlist (demux, new_bandwidth - 1);
1831   }
1832
1833   /* Force typefinding since we might have changed media type */
1834   demux->do_typefind = TRUE;
1835
1836   return TRUE;
1837 }
1838
1839 static gboolean
1840 gst_hls_demux_switch_playlist (GstHLSDemux * demux)
1841 {
1842   gint64 bitrate;
1843
1844   /* compare the time when the fragment was downloaded with the time when it was
1845    * scheduled */
1846   bitrate =
1847       (demux->download_total_bytes * 8) / ((double) demux->download_total_time /
1848       G_GUINT64_CONSTANT (1000000));
1849
1850   GST_DEBUG_OBJECT (demux,
1851       "Downloaded %u bytes in %" GST_TIME_FORMAT ". Bitrate is : %d",
1852       (guint) demux->download_total_bytes,
1853       GST_TIME_ARGS (demux->download_total_time * GST_USECOND), (gint) bitrate);
1854
1855   /* Take old rate into account too */
1856   if (demux->current_download_rate != -1)
1857     bitrate = (demux->current_download_rate + bitrate * 3) / 4;
1858   if (bitrate > G_MAXINT)
1859     bitrate = G_MAXINT;
1860   demux->current_download_rate = bitrate;
1861
1862   GST_DEBUG_OBJECT (demux, "Using current download rate: %d", (gint) bitrate);
1863
1864   GST_M3U8_CLIENT_LOCK (demux->client);
1865   if (!demux->client->main->lists) {
1866     GST_M3U8_CLIENT_UNLOCK (demux->client);
1867     return TRUE;
1868   }
1869   GST_M3U8_CLIENT_UNLOCK (demux->client);
1870
1871   return gst_hls_demux_change_playlist (demux, bitrate * demux->bitrate_limit);
1872 }
1873
1874 #if defined(HAVE_OPENSSL)
1875 static gboolean
1876 gst_hls_demux_decrypt_start (GstHLSDemux * demux, const guint8 * key_data,
1877     const guint8 * iv_data)
1878 {
1879   EVP_CIPHER_CTX_init (&demux->aes_ctx);
1880   if (!EVP_DecryptInit_ex (&demux->aes_ctx, EVP_aes_128_cbc (), NULL, key_data,
1881           iv_data))
1882     return FALSE;
1883   EVP_CIPHER_CTX_set_padding (&demux->aes_ctx, 0);
1884   return TRUE;
1885 }
1886
1887 static gboolean
1888 decrypt_fragment (GstHLSDemux * demux, gsize length,
1889     const guint8 * encrypted_data, guint8 * decrypted_data)
1890 {
1891   int len, flen = 0;
1892
1893   if (G_UNLIKELY (length > G_MAXINT || length % 16 != 0))
1894     return FALSE;
1895
1896   len = (int) length;
1897   if (!EVP_DecryptUpdate (&demux->aes_ctx, decrypted_data, &len, encrypted_data,
1898           len))
1899     return FALSE;
1900   EVP_DecryptFinal_ex (&demux->aes_ctx, decrypted_data + len, &flen);
1901   g_return_val_if_fail (len + flen == length, FALSE);
1902   return TRUE;
1903 }
1904
1905 static void
1906 gst_hls_demux_decrypt_end (GstHLSDemux * demux)
1907 {
1908   EVP_CIPHER_CTX_cleanup (&demux->aes_ctx);
1909 }
1910
1911 #elif defined(HAVE_NETTLE)
1912 static gboolean
1913 gst_hls_demux_decrypt_start (GstHLSDemux * demux, const guint8 * key_data,
1914     const guint8 * iv_data)
1915 {
1916   aes_set_decrypt_key (&demux->aes_ctx.ctx, 16, key_data);
1917   CBC_SET_IV (&demux->aes_ctx, iv_data);
1918
1919   return TRUE;
1920 }
1921
1922 static gboolean
1923 decrypt_fragment (GstHLSDemux * demux, gsize length,
1924     const guint8 * encrypted_data, guint8 * decrypted_data)
1925 {
1926   if (length % 16 != 0)
1927     return FALSE;
1928
1929   CBC_DECRYPT (&demux->aes_ctx, aes_decrypt, length, decrypted_data,
1930       encrypted_data);
1931
1932   return TRUE;
1933 }
1934
1935 static void
1936 gst_hls_demux_decrypt_end (GstHLSDemux * demux)
1937 {
1938   /* NOP */
1939 }
1940
1941 #else
1942 static gboolean
1943 gst_hls_demux_decrypt_start (GstHLSDemux * demux, const guint8 * key_data,
1944     const guint8 * iv_data)
1945 {
1946   gcry_error_t err = 0;
1947   gboolean ret = FALSE;
1948
1949   err =
1950       gcry_cipher_open (&demux->aes_ctx, GCRY_CIPHER_AES128,
1951       GCRY_CIPHER_MODE_CBC, 0);
1952   if (err)
1953     goto out;
1954   err = gcry_cipher_setkey (demux->aes_ctx, key_data, 16);
1955   if (err)
1956     goto out;
1957   err = gcry_cipher_setiv (demux->aes_ctx, iv_data, 16);
1958   if (!err)
1959     ret = TRUE;
1960
1961 out:
1962   if (!ret)
1963     if (demux->aes_ctx)
1964       gcry_cipher_close (demux->aes_ctx);
1965
1966   return ret;
1967 }
1968
1969 static gboolean
1970 decrypt_fragment (GstHLSDemux * demux, gsize length,
1971     const guint8 * encrypted_data, guint8 * decrypted_data)
1972 {
1973   gcry_error_t err = 0;
1974
1975   err = gcry_cipher_decrypt (demux->aes_ctx, decrypted_data, length,
1976       encrypted_data, length);
1977
1978   return err == 0;
1979 }
1980
1981 static void
1982 gst_hls_demux_decrypt_end (GstHLSDemux * demux)
1983 {
1984   if (demux->aes_ctx) {
1985     gcry_cipher_close (demux->aes_ctx);
1986     demux->aes_ctx = NULL;
1987   }
1988 }
1989 #endif
1990
1991 static GstBuffer *
1992 gst_hls_demux_decrypt_fragment (GstHLSDemux * demux,
1993     GstBuffer * encrypted_buffer, GError ** err)
1994 {
1995   GstBuffer *decrypted_buffer = NULL;
1996   GstMapInfo encrypted_info, decrypted_info;
1997
1998   decrypted_buffer =
1999       gst_buffer_new_allocate (NULL, gst_buffer_get_size (encrypted_buffer),
2000       NULL);
2001
2002   gst_buffer_map (encrypted_buffer, &encrypted_info, GST_MAP_READ);
2003   gst_buffer_map (decrypted_buffer, &decrypted_info, GST_MAP_WRITE);
2004
2005   if (!decrypt_fragment (demux, encrypted_info.size,
2006           encrypted_info.data, decrypted_info.data))
2007     goto decrypt_error;
2008
2009
2010   gst_buffer_unmap (decrypted_buffer, &decrypted_info);
2011   gst_buffer_unmap (encrypted_buffer, &encrypted_info);
2012
2013   gst_buffer_unref (encrypted_buffer);
2014
2015   return decrypted_buffer;
2016
2017 decrypt_error:
2018   GST_ERROR_OBJECT (demux, "Failed to decrypt fragment");
2019   g_set_error (err, GST_STREAM_ERROR, GST_STREAM_ERROR_DECRYPT,
2020       "Failed to decrypt fragment");
2021
2022   gst_buffer_unmap (decrypted_buffer, &decrypted_info);
2023   gst_buffer_unmap (encrypted_buffer, &encrypted_info);
2024
2025   gst_buffer_unref (encrypted_buffer);
2026   gst_buffer_unref (decrypted_buffer);
2027
2028   return NULL;
2029 }
2030
2031 static gboolean
2032 gst_hls_demux_update_source (GstHLSDemux * demux, const gchar * uri,
2033     const gchar * referer, gboolean refresh, gboolean allow_cache)
2034 {
2035   if (!gst_uri_is_valid (uri))
2036     return FALSE;
2037
2038   if (demux->src != NULL) {
2039     gchar *old_protocol, *new_protocol;
2040     gchar *old_uri;
2041
2042     old_uri = gst_uri_handler_get_uri (GST_URI_HANDLER (demux->src));
2043     old_protocol = gst_uri_get_protocol (old_uri);
2044     new_protocol = gst_uri_get_protocol (uri);
2045
2046     if (!g_str_equal (old_protocol, new_protocol)) {
2047       gst_object_unref (demux->src_srcpad);
2048       gst_element_set_state (demux->src, GST_STATE_NULL);
2049       gst_bin_remove (GST_BIN_CAST (demux), demux->src);
2050       demux->src = NULL;
2051       demux->src_srcpad = NULL;
2052       GST_DEBUG_OBJECT (demux, "Can't re-use old source element");
2053     } else {
2054       GError *err = NULL;
2055
2056       GST_DEBUG_OBJECT (demux, "Re-using old source element");
2057       if (!gst_uri_handler_set_uri (GST_URI_HANDLER (demux->src), uri, &err)) {
2058         GST_DEBUG_OBJECT (demux, "Failed to re-use old source element: %s",
2059             err->message);
2060         g_clear_error (&err);
2061         gst_element_set_state (demux->src, GST_STATE_NULL);
2062         gst_bin_remove (GST_BIN_CAST (demux), demux->src);
2063         demux->src = NULL;
2064       }
2065     }
2066     g_free (old_uri);
2067     g_free (old_protocol);
2068     g_free (new_protocol);
2069   }
2070
2071   if (demux->src == NULL) {
2072     GObjectClass *gobject_class;
2073
2074     demux->src = gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
2075     if (demux->src == NULL) {
2076       GST_WARNING_OBJECT (demux, "No element to handle uri: %s", uri);
2077       return FALSE;
2078     }
2079
2080     gobject_class = G_OBJECT_GET_CLASS (demux->src);
2081
2082     if (g_object_class_find_property (gobject_class, "compress"))
2083       g_object_set (demux->src, "compress", FALSE, NULL);
2084     if (g_object_class_find_property (gobject_class, "keep-alive"))
2085       g_object_set (demux->src, "keep-alive", TRUE, NULL);
2086     if (g_object_class_find_property (gobject_class, "extra-headers")) {
2087       if (referer || refresh || !allow_cache) {
2088         GstStructure *extra_headers = gst_structure_new_empty ("headers");
2089
2090         if (referer)
2091           gst_structure_set (extra_headers, "Referer", G_TYPE_STRING, referer,
2092               NULL);
2093
2094         if (!allow_cache)
2095           gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING,
2096               "no-cache", NULL);
2097         else if (refresh)
2098           gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING,
2099               "max-age=0", NULL);
2100
2101         g_object_set (demux->src, "extra-headers", extra_headers, NULL);
2102
2103         gst_structure_free (extra_headers);
2104       } else {
2105         g_object_set (demux->src, "extra-headers", NULL, NULL);
2106       }
2107     }
2108
2109     gst_element_set_locked_state (demux->src, TRUE);
2110     gst_bin_add (GST_BIN_CAST (demux), demux->src);
2111     demux->src_srcpad = gst_element_get_static_pad (demux->src, "src");
2112   }
2113   return TRUE;
2114 }
2115
2116 static gboolean
2117 gst_hls_demux_get_next_fragment (GstHLSDemux * demux,
2118     gboolean * end_of_playlist, GError ** err)
2119 {
2120   gchar *uri, *next_fragment_uri;
2121   GstClockTime duration;
2122   GstClockTime timestamp;
2123   gboolean discont;
2124   gint64 range_start, range_end;
2125   gchar *key = NULL;
2126   guint8 *iv = NULL;
2127   GstClockTime download_start_time = 0;
2128   GstFlowReturn ret = GST_FLOW_OK;
2129
2130   *end_of_playlist = FALSE;
2131   demux->download_finished = FALSE;
2132   if (!gst_m3u8_client_get_next_fragment (demux->client, &discont,
2133           &next_fragment_uri, &duration, &timestamp, &range_start, &range_end,
2134           &key, &iv, demux->segment.rate > 0)) {
2135     GST_INFO_OBJECT (demux, "This playlist doesn't contain more fragments");
2136     *end_of_playlist = TRUE;
2137     return FALSE;
2138   }
2139
2140   GST_DEBUG_OBJECT (demux,
2141       "Fetching next fragment %s %" GST_TIME_FORMAT "(range=%" G_GINT64_FORMAT
2142       "-%" G_GINT64_FORMAT ")", next_fragment_uri, GST_TIME_ARGS (timestamp),
2143       range_start, range_end);
2144
2145   /* set up our source for download */
2146   demux->current_timestamp = timestamp;
2147   demux->current_duration = duration;
2148   demux->starting_fragment = TRUE;
2149   demux->reset_crypto = TRUE;
2150   if (demux->current_key)
2151     g_free (demux->current_key);
2152   demux->current_key = key;
2153   if (demux->current_iv)
2154     g_free (demux->current_iv);
2155   demux->current_iv = iv;
2156
2157   /* Reset last flow return */
2158   g_mutex_lock (&demux->fragment_download_lock);
2159   demux->last_ret = GST_FLOW_OK;
2160   g_clear_error (&demux->last_error);
2161   g_mutex_unlock (&demux->fragment_download_lock);
2162
2163   GST_M3U8_CLIENT_LOCK (demux->client);
2164   if (!gst_hls_demux_update_source (demux, next_fragment_uri,
2165           demux->client->main ? demux->client->main->uri : NULL,
2166           FALSE,
2167           demux->client->current ? demux->client->current->allowcache : TRUE)) {
2168     GST_M3U8_CLIENT_UNLOCK (demux->client);
2169     *err =
2170         g_error_new (GST_CORE_ERROR, GST_CORE_ERROR_MISSING_PLUGIN,
2171         "Missing plugin to handle URI: '%s'", next_fragment_uri);
2172     g_free (next_fragment_uri);
2173     return FALSE;
2174   }
2175   GST_M3U8_CLIENT_UNLOCK (demux->client);
2176
2177   gst_hls_demux_configure_src_pad (demux);
2178
2179   if (gst_element_set_state (demux->src,
2180           GST_STATE_READY) != GST_STATE_CHANGE_FAILURE) {
2181     if (range_start != 0 || range_end != -1) {
2182       if (!gst_element_send_event (demux->src, gst_event_new_seek (1.0,
2183                   GST_FORMAT_BYTES, (GstSeekFlags) GST_SEEK_FLAG_FLUSH,
2184                   GST_SEEK_TYPE_SET, range_start, GST_SEEK_TYPE_SET,
2185                   range_end))) {
2186
2187         /* looks like the source can't handle seeks in READY */
2188         g_mutex_lock (&demux->fragment_download_lock);
2189         *err = g_error_new (GST_CORE_ERROR, GST_CORE_ERROR_NOT_IMPLEMENTED,
2190             "Source element can't handle range requests");
2191         demux->last_ret = GST_FLOW_ERROR;
2192         g_mutex_unlock (&demux->fragment_download_lock);
2193         g_free (next_fragment_uri);
2194         return FALSE;
2195       }
2196     }
2197
2198     g_mutex_lock (&demux->fragment_download_lock);
2199     if (G_LIKELY (demux->last_ret == GST_FLOW_OK)) {
2200       demux->download_start_time = g_get_monotonic_time ();
2201       download_start_time = gst_util_get_timestamp ();
2202
2203       g_mutex_unlock (&demux->fragment_download_lock);
2204       gst_element_sync_state_with_parent (demux->src);
2205
2206       g_mutex_lock (&demux->fragment_download_lock);
2207       /* wait for the fragment to be completely downloaded */
2208       GST_DEBUG_OBJECT (demux, "Waiting for fragment download to finish: %s",
2209           next_fragment_uri);
2210       while (!demux->stop_stream_task && !demux->download_finished) {
2211         g_cond_wait (&demux->fragment_download_cond,
2212             &demux->fragment_download_lock);
2213       }
2214
2215       ret = demux->last_ret;
2216     }
2217     g_mutex_unlock (&demux->fragment_download_lock);
2218   } else {
2219     g_mutex_lock (&demux->fragment_download_lock);
2220     ret = demux->last_ret = GST_FLOW_CUSTOM_ERROR;
2221     g_mutex_unlock (&demux->fragment_download_lock);
2222   }
2223
2224   /* flush the proxypads so that the EOS state is reset */
2225   gst_pad_push_event (demux->src_srcpad, gst_event_new_flush_start ());
2226   gst_pad_push_event (demux->src_srcpad, gst_event_new_flush_stop (TRUE));
2227
2228   if (ret != GST_FLOW_OK) {
2229     gst_element_set_state (demux->src, GST_STATE_NULL);
2230     if (*err == NULL) {
2231       g_mutex_lock (&demux->fragment_download_lock);
2232       if (demux->last_error) {
2233         *err = demux->last_error;
2234         demux->last_error = NULL;
2235       } else {
2236         *err = g_error_new (GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_FAILED,
2237             "Failed to download fragment");
2238       }
2239       g_mutex_unlock (&demux->fragment_download_lock);
2240     }
2241   } else {
2242     gst_element_set_state (demux->src, GST_STATE_READY);
2243     if (demux->segment.rate > 0)
2244       demux->segment.position += demux->current_duration;
2245
2246     uri = gst_m3u8_client_get_uri (demux->client);
2247     gst_element_post_message (GST_ELEMENT_CAST (demux),
2248         gst_message_new_element (GST_OBJECT_CAST (demux),
2249             gst_structure_new (STATISTICS_MESSAGE_NAME,
2250                 "manifest-uri", G_TYPE_STRING, uri,
2251                 "uri", G_TYPE_STRING, next_fragment_uri,
2252                 "fragment-start-time", GST_TYPE_CLOCK_TIME, download_start_time,
2253                 "fragment-stop-time", GST_TYPE_CLOCK_TIME,
2254                 gst_util_get_timestamp (), "fragment-size", G_TYPE_UINT64,
2255                 demux->download_total_bytes, "fragment-download-time",
2256                 GST_TYPE_CLOCK_TIME, demux->download_total_time * GST_USECOND,
2257                 NULL)));
2258     g_free (uri);
2259   }
2260
2261   /* clear the ghostpad eos state */
2262   gst_hls_demux_stream_clear_eos_state (demux);
2263
2264   g_free (next_fragment_uri);
2265
2266   if (ret != GST_FLOW_OK)
2267     return FALSE;
2268
2269   return TRUE;
2270 }