gst/hls/gsthlsdemux.c

   1 /* GStreamer
   2  * Copyright (C) 2010 Marc-Andre Lureau <marcandre.lureau@gmail.com>
   3  * Copyright (C) 2010 Andoni Morales Alastruey <ylatuya@gmail.com>
   4  *
   5  * Gsthlsdemux.c:
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22 /**
  23  * SECTION:element-hlsdemux
  24  *
  25  * HTTP Live Streaming source element.
  26  *
  27  * <refsect2>
  28  * <title>Example launch line</title>
  29  * |[
  30  * gst-launch hlsdemux location=http://devimages.apple.com/iphone/samples/bipbop/gear1/prog_index.m3u8 ! decodebin2 ! xvimagesink
  31  * ]|
  32  * </refsect2>
  33  *
  34  * Last reviewed on 2010-10-07
  35  */
  36
  37 #ifdef HAVE_CONFIG_H
  38 #  include "config.h"
  39 #endif
  40
  41
  42 #include <string.h>
  43 #include <gst/base/gsttypefindhelper.h>
  44 #include "gsthlsdemux.h"
  45
  46 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  47     GST_PAD_SRC,
  48     GST_PAD_ALWAYS,
  49     GST_STATIC_CAPS_ANY);
  50
  51 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  52     GST_PAD_SINK,
  53     GST_PAD_ALWAYS,
  54     GST_STATIC_CAPS ("playlist/m3u8"));
  55
  56 static GstStaticPadTemplate fetchertemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  57     GST_PAD_SINK,
  58     GST_PAD_ALWAYS,
  59     GST_STATIC_CAPS_ANY);
  60
  61 GST_DEBUG_CATEGORY_STATIC (gst_hls_demux_debug);
  62 #define GST_CAT_DEFAULT gst_hls_demux_debug
  63
  64 enum
  65 {
  66   PROP_0,
  67
  68   PROP_FRAGMENTS_CACHE,
  69   PROP_BITRATE_SWITCH_TOLERANCE,
  70   PROP_LAST
  71 };
  72
  73 static const float update_interval_factor[] = { 1, 0.5, 1.5, 3 };
  74
  75 #define DEFAULT_FRAGMENTS_CACHE 3
  76 #define DEFAULT_FAILED_COUNT 3
  77 #define DEFAULT_BITRATE_SWITCH_TOLERANCE 0.4
  78
  79 /* GObject */
  80 static void gst_hls_demux_set_property (GObject * object, guint prop_id,
  81     const GValue * value, GParamSpec * pspec);
  82 static void gst_hls_demux_get_property (GObject * object, guint prop_id,
  83     GValue * value, GParamSpec * pspec);
  84 static void gst_hls_demux_dispose (GObject * obj);
  85
  86 /* GstElement */
  87 static GstStateChangeReturn
  88 gst_hls_demux_change_state (GstElement * element, GstStateChange transition);
  89
  90 /* GstHLSDemux */
  91 static GstBusSyncReply gst_hls_demux_fetcher_bus_handler (GstBus * bus,
  92     GstMessage * message, gpointer data);
  93 static GstFlowReturn gst_hls_demux_chain (GstPad * pad, GstBuffer * buf);
  94 static gboolean gst_hls_demux_sink_event (GstPad * pad, GstEvent * event);
  95 static GstFlowReturn gst_hls_demux_fetcher_chain (GstPad * pad, GstBuffer * buf);
  96 static gboolean gst_hls_demux_fetcher_sink_event (GstPad * pad, GstEvent * event);
  97 static void gst_hls_demux_loop (GstHLSDemux * demux);
  98 static void gst_hls_demux_stop (GstHLSDemux * demux);
  99 static void gst_hls_demux_stop_fetcher (GstHLSDemux * demux, gboolean cancelled);
 100 static gboolean gst_hls_demux_start_update (GstHLSDemux * demux);
 101 static gboolean gst_hls_demux_cache_fragments (GstHLSDemux * demux);
 102 static gboolean gst_hls_demux_schedule (GstHLSDemux * demux);
 103 static gboolean gst_hls_demux_switch_playlist (GstHLSDemux * demux);
 104 static gboolean gst_hls_demux_get_next_fragment (GstHLSDemux * demux, gboolean retry);
 105 static gboolean gst_hls_demux_update_playlist (GstHLSDemux * demux, gboolean retry);
 106 static void gst_hls_demux_reset (GstHLSDemux * demux, gboolean dispose);
 107 static gboolean gst_hls_demux_set_location (GstHLSDemux * demux, const gchar * uri);
 108
 109 static void
 110 _do_init (GType type)
 111 {
 112   GST_DEBUG_CATEGORY_INIT (gst_hls_demux_debug, "hlsdemux", 0, "hlsdemux element");
 113 }
 114
 115 GST_BOILERPLATE_FULL (GstHLSDemux, gst_hls_demux, GstElement,
 116     GST_TYPE_ELEMENT, _do_init);
 117
 118 static void
 119 gst_hls_demux_base_init (gpointer g_class)
 120 {
 121   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 122
 123   gst_element_class_add_pad_template (element_class,
 124       gst_static_pad_template_get (&srctemplate));
 125
 126   gst_element_class_add_pad_template (element_class,
 127       gst_static_pad_template_get (&sinktemplate));
 128
 129   gst_element_class_set_details_simple (element_class,
 130       "HLS Source",
 131       "Demuxer/URIList",
 132       "HTTP Live Streaming source",
 133       "Marc-Andre Lureau <marcandre.lureau@gmail.com>\n"
 134       "Andoni Morales Alastruey <ylatuya@gmail.com>");
 135 }
 136
 137 static void
 138 gst_hls_demux_dispose (GObject * obj)
 139 {
 140   GstHLSDemux *demux = GST_HLS_DEMUX (obj);
 141
 142   g_cond_free (demux->fetcher_cond);
 143   g_mutex_free (demux->fetcher_lock);
 144
 145   g_cond_free (demux->thread_cond);
 146   g_mutex_free (demux->thread_lock);
 147
 148   if (GST_TASK_STATE (demux->task) != GST_TASK_STOPPED) {
 149     gst_task_stop (demux->task);
 150     gst_task_join (demux->task);
 151   }
 152   gst_object_unref (demux->task);
 153   g_static_rec_mutex_free (&demux->task_lock);
 154
 155   gst_object_unref (demux->fetcher_bus);
 156   gst_object_unref (demux->fetcherpad);
 157
 158   gst_hls_demux_reset (demux, TRUE);
 159
 160   G_OBJECT_CLASS (parent_class)->dispose (obj);
 161 }
 162
 163 static void
 164 gst_hls_demux_class_init (GstHLSDemuxClass * klass)
 165 {
 166   GObjectClass *gobject_class;
 167   GstElementClass *gstelement_class;
 168
 169   gobject_class = G_OBJECT_CLASS (klass);
 170   gstelement_class = (GstElementClass *) klass;
 171
 172   parent_class = g_type_class_peek_parent (klass);
 173
 174   gobject_class->set_property = gst_hls_demux_set_property;
 175   gobject_class->get_property = gst_hls_demux_get_property;
 176   gobject_class->dispose = gst_hls_demux_dispose;
 177
 178   g_object_class_install_property (gobject_class, PROP_FRAGMENTS_CACHE,
 179       g_param_spec_uint ("fragments-cache", "Fragments cache",
 180           "Number of fragments needed to be cached to start playing",
 181           2, G_MAXUINT, DEFAULT_FRAGMENTS_CACHE, G_PARAM_READWRITE));
 182
 183   g_object_class_install_property (gobject_class, PROP_BITRATE_SWITCH_TOLERANCE,
 184       g_param_spec_float ("bitrate-switch-tolerance",
 185           "Bitrate switch tolerance",
 186           "Tolerance with respect of the fragment duration to switch to "
 187           "a different bitrate if the client is too slow/fast.",
 188           0, 1, DEFAULT_BITRATE_SWITCH_TOLERANCE, G_PARAM_READWRITE));
 189
 190   gstelement_class->change_state = gst_hls_demux_change_state;
 191 }
 192
 193 static void
 194 gst_hls_demux_init (GstHLSDemux * demux, GstHLSDemuxClass * klass)
 195 {
 196   /* sink pad */
 197   demux->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 198   gst_pad_set_chain_function (demux->sinkpad,
 199       GST_DEBUG_FUNCPTR (gst_hls_demux_chain));
 200   gst_pad_set_event_function (demux->sinkpad,
 201       GST_DEBUG_FUNCPTR (gst_hls_demux_sink_event));
 202   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 203
 204   /* demux pad */
 205   demux->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
 206   gst_element_add_pad (GST_ELEMENT (demux), demux->srcpad);
 207
 208   /* fetcher pad */
 209   demux->fetcherpad = gst_pad_new_from_static_template (&fetchertemplate, "sink");
 210   gst_pad_set_chain_function (demux->fetcherpad,
 211       GST_DEBUG_FUNCPTR (gst_hls_demux_fetcher_chain));
 212   gst_pad_set_event_function (demux->fetcherpad,
 213       GST_DEBUG_FUNCPTR (gst_hls_demux_fetcher_sink_event));
 214   gst_pad_set_element_private (demux->fetcherpad, demux);
 215   gst_pad_activate_push (demux->fetcherpad, TRUE);
 216
 217   /* Properties */
 218   demux->fragments_cache = DEFAULT_FRAGMENTS_CACHE;
 219   demux->bitrate_switch_tol = DEFAULT_BITRATE_SWITCH_TOLERANCE;
 220
 221   demux->fetcher_bus = gst_bus_new ();
 222   gst_bus_set_sync_handler (demux->fetcher_bus, gst_hls_demux_fetcher_bus_handler,
 223       demux);
 224   demux->thread_cond = g_cond_new ();
 225   demux->thread_lock = g_mutex_new ();
 226   demux->fetcher_cond = g_cond_new ();
 227   demux->fetcher_lock = g_mutex_new ();
 228   demux->queue = g_queue_new ();
 229   g_static_rec_mutex_init (&demux->task_lock);
 230   demux->task = gst_task_create ((GstTaskFunction) gst_hls_demux_loop, demux);
 231   gst_task_set_lock (demux->task, &demux->task_lock);
 232 }
 233
 234 static void
 235 gst_hls_demux_set_property (GObject * object, guint prop_id, const GValue * value,
 236     GParamSpec * pspec)
 237 {
 238   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 239
 240   switch (prop_id) {
 241     case PROP_FRAGMENTS_CACHE:
 242       demux->fragments_cache = g_value_get_uint (value);
 243       break;
 244     case PROP_BITRATE_SWITCH_TOLERANCE:
 245       demux->bitrate_switch_tol = g_value_get_float (value);
 246       break;
 247     default:
 248       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 249       break;
 250   }
 251 }
 252
 253 static void
 254 gst_hls_demux_get_property (GObject * object, guint prop_id, GValue * value,
 255     GParamSpec * pspec)
 256 {
 257   GstHLSDemux *demux = GST_HLS_DEMUX (object);
 258
 259   switch (prop_id) {
 260     case PROP_FRAGMENTS_CACHE:
 261       g_value_set_uint (value, demux->fragments_cache);
 262       break;
 263     case PROP_BITRATE_SWITCH_TOLERANCE:
 264       g_value_set_float (value, demux->bitrate_switch_tol);
 265       break;
 266     default:
 267       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 268       break;
 269   }
 270 }
 271
 272 static GstStateChangeReturn
 273 gst_hls_demux_change_state (GstElement * element, GstStateChange transition)
 274 {
 275   GstStateChangeReturn ret;
 276   GstHLSDemux *demux = GST_HLS_DEMUX (element);
 277
 278   switch (transition) {
 279     case GST_STATE_CHANGE_NULL_TO_READY:
 280       gst_hls_demux_reset (demux, FALSE);
 281       break;
 282     default:
 283       break;
 284   }
 285
 286   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 287
 288   switch (transition) {
 289     case GST_STATE_CHANGE_PAUSED_TO_READY:
 290       demux->cancelled = TRUE;
 291       gst_hls_demux_stop_fetcher (demux, TRUE);
 292       break;
 293     default:
 294       break;
 295   }
 296   return ret;
 297 }
 298
 299 static gboolean
 300 gst_hls_demux_sink_event (GstPad * pad, GstEvent * event)
 301 {
 302   GstHLSDemux *demux = GST_HLS_DEMUX (gst_pad_get_parent (pad));
 303   GstQuery *query;
 304   gboolean ret;
 305   gchar *uri;
 306
 307
 308   switch (event->type) {
 309     case GST_EVENT_EOS:{
 310       gchar *playlist;
 311
 312       if (demux->playlist == NULL) {
 313         GST_WARNING_OBJECT (demux, "Received EOS without a playlist.");
 314         break;
 315       }
 316
 317       GST_DEBUG_OBJECT (demux, "Got EOS on the sink pad: main playlist fetched");
 318
 319       query = gst_query_new_uri ();
 320       ret = gst_pad_peer_query (demux->sinkpad, query);
 321       if (ret) {
 322         gst_query_parse_uri (query, &uri);
 323         gst_hls_demux_set_location (demux, uri);
 324         g_free (uri);
 325       }
 326
 327       playlist = g_strndup ((gchar *) GST_BUFFER_DATA (demux->playlist),
 328           GST_BUFFER_SIZE (demux->playlist));
 329       gst_buffer_unref (demux->playlist);
 330       if (!gst_m3u8_client_update (demux->client, playlist)) {
 331         /* In most cases, this will happen if we set a wrong url in the
 332          * source element and we have received the 404 HTML response instead of
 333          * the playlist */
 334         GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid playlist."), NULL);
 335         return FALSE;
 336       }
 337
 338       if (!ret && gst_m3u8_client_is_live (demux->client)) {
 339         GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 340             ("Failed querying the playlist uri, "
 341                 "required for live sources."), NULL);
 342         return FALSE;
 343       }
 344
 345       gst_task_start (demux->task);
 346       gst_event_unref (event);
 347       return TRUE;
 348     }
 349     default:
 350       break;
 351   }
 352
 353   return gst_pad_event_default (pad, event);
 354 }
 355
 356 static gboolean
 357 gst_hls_demux_fetcher_sink_event (GstPad * pad, GstEvent * event)
 358 {
 359   GstHLSDemux *demux = GST_HLS_DEMUX (gst_pad_get_element_private (pad));
 360
 361   switch (event->type) {
 362     case GST_EVENT_EOS:{
 363       GST_DEBUG_OBJECT (demux, "Got EOS on the fetcher pad");
 364       /* signal we have fetched the URI */
 365       if (!demux->cancelled)
 366         g_cond_signal (demux->fetcher_cond);
 367     }
 368     default:
 369       break;
 370   }
 371
 372   gst_event_unref (event);
 373   return FALSE;
 374 }
 375
 376 static GstFlowReturn
 377 gst_hls_demux_chain (GstPad * pad, GstBuffer * buf)
 378 {
 379   GstHLSDemux *demux = GST_HLS_DEMUX (gst_pad_get_parent (pad));
 380
 381   if (demux->playlist == NULL)
 382     demux->playlist = buf;
 383   else
 384     demux->playlist = gst_buffer_join (demux->playlist, buf);
 385
 386   gst_object_unref (demux);
 387
 388   return GST_FLOW_OK;
 389 }
 390
 391 static GstFlowReturn
 392 gst_hls_demux_fetcher_chain (GstPad * pad, GstBuffer * buf)
 393 {
 394   GstHLSDemux *demux = GST_HLS_DEMUX (gst_pad_get_element_private (pad));
 395
 396   /* The source element can be an http source element. In case we get a 404,
 397    * the html response will be sent downstream and demux->downloaded_uri
 398    * will not be null, which might make us think that the request proceed
 399    * successfully. But it will also post an error message in the bus that
 400    * is handled synchronously and that will set demux->fetcher_error to TRUE,
 401    * which is used to discard this buffer with the html response. */
 402   if (demux->fetcher_error) {
 403     goto done;
 404   }
 405
 406   GST_LOG_OBJECT (demux, "The uri fetcher received a new buffer of size %lld",
 407       GST_BUFFER_SIZE (buf));
 408   if (demux->downloaded_uri == NULL)
 409     demux->downloaded_uri = buf;
 410   else
 411     demux->downloaded_uri = gst_buffer_join (demux->downloaded_uri, buf);
 412
 413 done:
 414   {
 415     return GST_FLOW_OK;
 416   }
 417 }
 418
 419 static void
 420 gst_hls_demux_stop_fetcher (GstHLSDemux * demux, gboolean cancelled)
 421 {
 422   GstPad *pad;
 423
 424   /* When the fetcher is stopped while it's downloading, we will get an EOS that
 425    * unblocks the fetcher thread and tries to stop it again from that thread.
 426    * Here we check if the fetcher as already been stopped before continuing */
 427   if (demux->fetcher == NULL || demux->stopping_fetcher)
 428     return;
 429
 430   GST_DEBUG_OBJECT (demux, "Stopping fetcher.");
 431   demux->stopping_fetcher = TRUE;
 432   /* set the element state to NULL */
 433   gst_element_set_state (demux->fetcher, GST_STATE_NULL);
 434   gst_element_get_state (demux->fetcher, NULL, NULL, GST_CLOCK_TIME_NONE);
 435   /* unlink it from the internal pad */
 436   pad = gst_pad_get_peer (demux->fetcherpad);
 437   if (pad) {
 438     gst_pad_unlink (pad, demux->fetcherpad);
 439     gst_object_unref (pad);
 440   }
 441   /* and finally unref it */
 442   gst_object_unref (demux->fetcher);
 443   demux->fetcher = NULL;
 444
 445   /* if we stopped it to cancell a download, free the cached buffer */
 446   if (cancelled && demux->downloaded_uri != NULL) {
 447     gst_buffer_unref (demux->downloaded_uri);
 448     demux->downloaded_uri = NULL;
 449     /* signal the fetcher thread that the download has finished/cancelled */
 450     g_cond_signal (demux->fetcher_cond);
 451   }
 452 }
 453
 454 static void
 455 gst_hls_demux_stop (GstHLSDemux * demux)
 456 {
 457   gst_hls_demux_stop_fetcher (demux, TRUE);
 458   if (GST_TASK_STATE (demux->task) != GST_TASK_STOPPED)
 459     gst_task_stop (demux->task);
 460   g_cond_signal (demux->thread_cond);
 461 }
 462
 463 static void
 464 gst_hls_demux_loop (GstHLSDemux * demux)
 465 {
 466   GstBuffer *buf;
 467   GstFlowReturn ret;
 468
 469   /* Loop for the source pad task. The task is started when we have
 470    * received the main playlist from the source element. It tries first to
 471    * cache the first fragments and then it waits until it has more data in the
 472    * queue. This task is woken up when we push a new fragment to the queue or
 473    * when we reached the end of the playlist  */
 474
 475   if (G_UNLIKELY (demux->need_cache)) {
 476     if (!gst_hls_demux_cache_fragments (demux))
 477       goto cache_error;
 478
 479     /* we can start now the updates thread */
 480     gst_hls_demux_start_update (demux);
 481     GST_INFO_OBJECT (demux, "First fragments cached successfully");
 482   }
 483
 484   if (g_queue_is_empty (demux->queue)) {
 485     if (demux->end_of_playlist) {
 486       goto end_of_playlist;
 487     }
 488     GST_TASK_WAIT (demux->task);
 489   }
 490
 491   buf = g_queue_pop_head (demux->queue);
 492   ret = gst_pad_push (demux->srcpad, buf);
 493   if (ret != GST_FLOW_OK)
 494     goto error;
 495
 496   return;
 497
 498 end_of_playlist:
 499   {
 500     GST_DEBUG_OBJECT (demux, "Reached end of playlist, sending EOS");
 501     gst_pad_push_event (demux->srcpad, gst_event_new_eos ());
 502     gst_hls_demux_stop (demux);
 503     return;
 504   }
 505
 506 cache_error:
 507   {
 508     GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
 509         ("Could not cache the first fragments"), NULL);
 510     gst_hls_demux_stop (demux);
 511     return;
 512   }
 513
 514 error:
 515   {
 516     /* FIXME: handle error */
 517     gst_hls_demux_stop (demux);
 518     return;
 519   }
 520 }
 521
 522 static GstBusSyncReply
 523 gst_hls_demux_fetcher_bus_handler (GstBus * bus,
 524     GstMessage * message, gpointer data)
 525 {
 526   GstHLSDemux *demux = GST_HLS_DEMUX (data);
 527
 528   if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR) {
 529     demux->fetcher_error = TRUE;
 530     g_cond_signal (demux->fetcher_cond);
 531   }
 532
 533   gst_message_unref (message);
 534   return GST_BUS_DROP;
 535 }
 536
 537 static gboolean
 538 gst_hls_demux_make_fetcher (GstHLSDemux * demux, const gchar * uri)
 539 {
 540   GstPad *pad;
 541
 542   if (!gst_uri_is_valid (uri))
 543     return FALSE;
 544
 545   GST_DEBUG_OBJECT (demux, "Creating fetcher for the URI:%s", uri);
 546   demux->fetcher = gst_element_make_from_uri (GST_URI_SRC, uri, NULL);
 547   if (!demux->fetcher)
 548     return FALSE;
 549
 550   demux->fetcher_error = FALSE;
 551   demux->stopping_fetcher = FALSE;
 552   gst_element_set_bus (GST_ELEMENT (demux->fetcher), demux->fetcher_bus);
 553
 554   g_object_set (G_OBJECT (demux->fetcher), "location", uri, NULL);
 555   pad = gst_element_get_static_pad (demux->fetcher, "src");
 556   if (pad) {
 557     gst_pad_link (pad, demux->fetcherpad);
 558     gst_object_unref (pad);
 559   }
 560   return TRUE;
 561 }
 562
 563 static void
 564 gst_hls_demux_reset (GstHLSDemux * demux, gboolean dispose)
 565 {
 566   demux->need_cache = TRUE;
 567   demux->thread_return = FALSE;
 568   demux->accumulated_delay = 0;
 569   demux->end_of_playlist = FALSE;
 570   demux->cancelled = FALSE;
 571
 572   if (demux->input_caps) {
 573     gst_caps_unref (demux->input_caps);
 574     demux->input_caps = NULL;
 575   }
 576
 577   if (demux->playlist) {
 578     gst_buffer_unref (demux->playlist);
 579     demux->playlist = NULL;
 580   }
 581
 582   if (demux->downloaded_uri) {
 583     gst_buffer_unref (demux->downloaded_uri);
 584     demux->downloaded_uri = NULL;
 585   }
 586
 587   if (demux->client)
 588     gst_m3u8_client_free (demux->client);
 589
 590   if (!dispose) {
 591     demux->client = gst_m3u8_client_new ("");
 592   }
 593
 594   while (!g_queue_is_empty (demux->queue)) {
 595     GstBuffer *buf = g_queue_pop_head (demux->queue);
 596     gst_buffer_unref (buf);
 597   }
 598   g_queue_clear (demux->queue);
 599 }
 600
 601 static gboolean
 602 gst_hls_demux_set_location (GstHLSDemux * demux, const gchar * uri)
 603 {
 604   if (demux->client)
 605     gst_m3u8_client_free (demux->client);
 606   demux->client = gst_m3u8_client_new (uri);
 607   GST_INFO_OBJECT (demux, "Changed location: %s", uri);
 608   return TRUE;
 609 }
 610
 611 static gboolean
 612 gst_hls_demux_update_thread (GstHLSDemux * demux)
 613 {
 614   /* Loop for the updates. It's started when the first fragments are cached and
 615    * schedules the next update of the playlist (for lives sources) and the next
 616    * update of fragments. When a new fragment is downloaded, it compares the
 617    * download time with the next scheduled update to check if we can or should
 618    * switch to a different bitrate */
 619
 620   g_mutex_lock (demux->thread_lock);
 621   while (TRUE) {
 622     /* block until the next scheduled update or the signal to quit this thread */
 623     if (g_cond_timed_wait (demux->thread_cond, demux->thread_lock,
 624             &demux->next_update)) {
 625       goto quit;
 626     }
 627
 628     /* update the playlist for live sources */
 629     if (gst_m3u8_client_is_live (demux->client)) {
 630       if (!gst_hls_demux_update_playlist (demux, TRUE)) {
 631         GST_ERROR_OBJECT (demux, "Could not update the playlist");
 632         goto quit;
 633       }
 634     }
 635
 636     /* schedule the next update */
 637     gst_hls_demux_schedule (demux);
 638
 639     /* if it's a live source and the playlist couldn't be updated, there aren't
 640      * more fragments in the playlist, so we just wait for the next schedulled
 641      * update */
 642     if (gst_m3u8_client_is_live (demux->client) &&
 643         demux->client->update_failed_count > 0) {
 644       GST_WARNING_OBJECT (demux,
 645           "The playlist hasn't been updated, failed count is %d",
 646           demux->client->update_failed_count);
 647       continue;
 648     }
 649
 650     /* fetch the next fragment */
 651     if (!gst_hls_demux_get_next_fragment (demux, TRUE)) {
 652       if (!demux->end_of_playlist && !demux->cancelled)
 653         GST_ERROR_OBJECT (demux, "Could not fetch the next fragment");
 654       goto quit;
 655     }
 656
 657     /* try to switch to another bitrate if needed */
 658     gst_hls_demux_switch_playlist (demux);
 659   }
 660
 661 quit:
 662   {
 663     g_mutex_unlock (demux->thread_lock);
 664     return TRUE;
 665   }
 666 }
 667
 668 static gboolean
 669 gst_hls_demux_start_update (GstHLSDemux * demux)
 670 {
 671   GError *error;
 672
 673   /* creates a new thread for the updates */
 674   demux->updates_thread = g_thread_create (
 675       (GThreadFunc) gst_hls_demux_update_thread, demux, TRUE, &error);
 676   return (error != NULL);
 677 }
 678
 679 static gboolean
 680 gst_hls_demux_cache_fragments (GstHLSDemux * demux)
 681 {
 682   gint i;
 683
 684   /* Start parsing the main playlist */
 685   gst_m3u8_client_set_current (demux->client, demux->client->main);
 686
 687   if (gst_m3u8_client_is_live (demux->client)) {
 688     if (!gst_hls_demux_update_playlist (demux, FALSE)) {
 689       GST_ERROR_OBJECT (demux, "Could not fetch the main playlist %s",
 690           demux->client->main->uri);
 691       return FALSE;
 692     }
 693   }
 694
 695   /* If this playlist is a variant playlist, select the first one
 696    * and update it */
 697   if (gst_m3u8_client_has_variant_playlist (demux->client)) {
 698     GstM3U8 *child = demux->client->main->lists->data;
 699     gst_m3u8_client_set_current (demux->client, child);
 700     if (!gst_hls_demux_update_playlist (demux, FALSE)) {
 701       GST_ERROR_OBJECT (demux, "Could not fetch the child playlist %s",
 702           child->uri);
 703       return FALSE;
 704     }
 705   }
 706
 707   /* If it's a live source, set the sequence number to the end of the list
 708    * and substract the 'fragmets_cache' to start from the last fragment*/
 709   if (gst_m3u8_client_is_live (demux->client)) {
 710     demux->client->sequence += g_list_length (demux->client->current->files);
 711     if (demux->client->sequence >= demux->fragments_cache)
 712       demux->client->sequence -= demux->fragments_cache;
 713     else
 714       demux->client->sequence = 0;
 715   }
 716
 717   /* Cache the first fragments */
 718   for (i = 0; i < demux->fragments_cache - 1; i++) {
 719     if (!gst_hls_demux_get_next_fragment (demux, FALSE)) {
 720       if (!demux->cancelled)
 721         GST_ERROR_OBJECT (demux, "Error caching the first fragments");
 722       return FALSE;
 723     }
 724     /* make sure we stop caching fragments if something cancelled it */
 725     if (demux->cancelled)
 726       return FALSE;
 727   }
 728
 729   g_get_current_time (&demux->next_update);
 730
 731   demux->need_cache = FALSE;
 732   return TRUE;
 733 }
 734
 735 static gboolean
 736 gst_hls_demux_fetch_location (GstHLSDemux * demux, const gchar * uri)
 737 {
 738   GstStateChangeReturn ret;
 739   gboolean bret = FALSE;
 740
 741   g_mutex_lock (demux->fetcher_lock);
 742
 743   if (!gst_hls_demux_make_fetcher (demux, uri)) {
 744     goto uri_error;
 745   }
 746
 747   ret = gst_element_set_state (demux->fetcher, GST_STATE_PLAYING);
 748   if (ret == GST_STATE_CHANGE_FAILURE)
 749     goto state_change_error;
 750
 751   /* wait until we have fetched the uri */
 752   GST_DEBUG_OBJECT (demux, "Waiting to fetch the URI");
 753   g_cond_wait (demux->fetcher_cond, demux->fetcher_lock);
 754
 755   gst_hls_demux_stop_fetcher (demux, FALSE);
 756
 757   if (demux->downloaded_uri != NULL) {
 758     GST_INFO_OBJECT (demux, "URI fetched successfully");
 759     bret = TRUE;
 760   }
 761   goto quit;
 762
 763 uri_error:
 764   {
 765     GST_ELEMENT_ERROR (demux, RESOURCE, OPEN_READ,
 766         ("Could not create an element to fetch the given URI."), ("URI: \"%s\"",
 767             uri));
 768     bret = FALSE;
 769     goto quit;
 770   }
 771
 772 state_change_error:
 773   {
 774     GST_ELEMENT_ERROR (demux, CORE, STATE_CHANGE,
 775         ("Error changing state of the fetcher element."), NULL);
 776     bret = FALSE;
 777     goto quit;
 778   }
 779
 780 quit:
 781   {
 782     g_mutex_unlock (demux->fetcher_lock);
 783     return bret;
 784   }
 785 }
 786
 787 static gboolean
 788 gst_hls_demux_update_playlist (GstHLSDemux * demux, gboolean retry)
 789 {
 790   gchar *playlist;
 791
 792   GST_INFO_OBJECT (demux, "Updating the playlist %s", demux->client->current->uri);
 793   if (!gst_hls_demux_fetch_location (demux, demux->client->current->uri))
 794     return FALSE;
 795
 796   playlist = g_strndup ((gchar *) GST_BUFFER_DATA (demux->downloaded_uri),
 797       GST_BUFFER_SIZE (demux->downloaded_uri));
 798   gst_m3u8_client_update (demux->client, playlist);
 799   gst_buffer_unref (demux->downloaded_uri);
 800   demux->downloaded_uri = NULL;
 801   return TRUE;
 802 }
 803
 804 static gboolean
 805 gst_hls_demux_change_playlist (GstHLSDemux * demux, gboolean is_fast)
 806 {
 807   if (is_fast) {
 808     if (!demux->client->main->lists->next)
 809       return TRUE;
 810     demux->client->main->lists = g_list_next (demux->client->main->lists);
 811   } else {
 812     if (!demux->client->main->lists->prev)
 813       return TRUE;
 814     demux->client->main->lists = g_list_previous (demux->client->main->lists);
 815   }
 816
 817   gst_m3u8_client_set_current (demux->client, demux->client->main->lists->data);
 818   gst_hls_demux_update_playlist (demux, TRUE);
 819   GST_INFO_OBJECT (demux, "Client is %s, switching to bitrate %d",
 820       is_fast ? "fast" : "slow", demux->client->current->bandwidth);
 821
 822   return TRUE;
 823 }
 824
 825 static gboolean
 826 gst_hls_demux_schedule (GstHLSDemux * demux)
 827 {
 828   gfloat update_factor;
 829   gint count;
 830
 831   /* As defined in §6.3.4. Reloading the Playlist file:
 832    * "If the client reloads a Playlist file and finds that it has not
 833    * changed then it MUST wait for a period of time before retrying.  The
 834    * minimum delay is a multiple of the target duration.  This multiple is
 835    * 0.5 for the first attempt, 1.5 for the second, and 3.0 thereafter."
 836    */
 837   count = demux->client->update_failed_count;
 838   if (count < 3)
 839     update_factor = update_interval_factor[count];
 840   else
 841     update_factor = update_interval_factor[3];
 842
 843   /* schedule the next update using the target duration field of the
 844    * playlist */
 845   g_time_val_add (&demux->next_update,
 846       demux->client->current->targetduration * update_factor * 1000000);
 847   GST_DEBUG_OBJECT (demux, "Next update scheduled at %s",
 848       g_time_val_to_iso8601 (&demux->next_update));
 849
 850   return TRUE;
 851 }
 852
 853 static gboolean
 854 gst_hls_demux_switch_playlist (GstHLSDemux * demux)
 855 {
 856   GTimeVal now;
 857   gint64 diff, limit;
 858
 859   g_get_current_time (&now);
 860   if (!demux->client->main->lists)
 861     return TRUE;
 862
 863   /* compare the time when the fragment was downloaded with the time when it was
 864    * scheduled */
 865   diff = (GST_TIMEVAL_TO_TIME (demux->next_update) - GST_TIMEVAL_TO_TIME (now));
 866   limit = demux->client->current->targetduration * GST_SECOND *
 867       demux->bitrate_switch_tol;
 868
 869   /* if we are on time switch to a higher bitrate */
 870   if (diff > limit) {
 871     gst_hls_demux_change_playlist (demux, TRUE);
 872   } else if (diff < 0) {
 873     /* if the client is too slow wait until it has accumulated a certain delay to
 874      * switch to a lower bitrate */
 875     demux->accumulated_delay -= diff;
 876     if (demux->accumulated_delay >= limit) {
 877       gst_hls_demux_change_playlist (demux, FALSE);
 878     } else if (demux->accumulated_delay < 0) {
 879       demux->accumulated_delay = 0;
 880     }
 881   }
 882   return TRUE;
 883 }
 884
 885 static gboolean
 886 gst_hls_demux_get_next_fragment (GstHLSDemux * demux, gboolean retry)
 887 {
 888   GstBuffer *buf;
 889   const gchar *next_fragment_uri;
 890   gboolean discont;
 891
 892   next_fragment_uri = gst_m3u8_client_get_next_fragment (demux->client, &discont);
 893
 894   if (!next_fragment_uri) {
 895     GST_INFO_OBJECT (demux, "This playlist doesn't contain more fragments");
 896     demux->end_of_playlist = TRUE;
 897     GST_TASK_SIGNAL (demux->task);
 898     return FALSE;
 899   }
 900
 901   GST_INFO_OBJECT (demux, "Fetching next fragment %s", next_fragment_uri);
 902
 903   if (!gst_hls_demux_fetch_location (demux, next_fragment_uri))
 904     return FALSE;
 905
 906   buf = demux->downloaded_uri;
 907
 908   if (G_UNLIKELY (demux->input_caps == NULL)) {
 909     demux->input_caps = gst_type_find_helper_for_buffer (NULL, buf, NULL);
 910     if (demux->input_caps) {
 911       gst_pad_set_caps (demux->srcpad, demux->input_caps);
 912       GST_INFO_OBJECT (demux, "Input source caps: %" GST_PTR_FORMAT,
 913           demux->input_caps);
 914     }
 915   }
 916
 917   if (discont) {
 918     GST_DEBUG_OBJECT (demux, "Marking fragment as discontinuous");
 919     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
 920   }
 921
 922   g_queue_push_tail (demux->queue, buf);
 923   GST_TASK_SIGNAL (demux->task);
 924   demux->downloaded_uri = NULL;
 925   return TRUE;
 926 }