3 * Copyright (C) 2014 Samsung Electronics. All rights reserved.
4 * Author: Thiago Santos <thiagoss@osg.samsung.com>
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Library General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Library General Public License for more details.
16 * You should have received a copy of the GNU Library General Public
17 * License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
23 * SECTION:gstadaptivedemux
24 * @short_description: Base class for adaptive demuxers
26 * What is an adaptive demuxer?
27 * Adaptive demuxers are special demuxers in the sense that they don't
28 * actually demux data received from upstream but download the data
31 * Adaptive formats (HLS, DASH, MSS) are composed of a manifest file and
32 * a set of fragments. The manifest describes the available media and
33 * the sequence of fragments to use. Each fragment contains a small
34 * part of the media (typically only a few seconds). It is possible for
35 * the manifest to have the same media available in different configurations
36 * (bitrates for example) so that the client can select the one that
37 * best suits its scenario (network fluctuation, hardware requirements...).
38 * It is possible to switch from one representation of the media to another
39 * during playback. That's why it is called 'adaptive', because it can be
40 * adapted to the client's needs.
42 * Architectural overview:
43 * The manifest is received by the demuxer in its sink pad and, upon receiving
44 * EOS, it parses the manifest and exposes the streams available in it. For
45 * each stream a source element will be created and will download the list
46 * of fragments one by one. Once a fragment is finished downloading, the next
47 * URI is set to the source element and it starts fetching it and pushing
48 * through the stream's pad. This implies that each stream is independent from
49 * each other as it runs on a separate thread.
51 * After downloading each fragment, the download rate of it is calculated and
52 * the demuxer has a chance to switch to a different bitrate if needed. The
53 * switch can be done by simply pushing a new caps before the next fragment
54 * when codecs are the same, or by exposing a new pad group if it needs
58 * - Not linked streams: Streams that are not-linked have their download threads
59 * interrupted to save network bandwidth. When they are
60 * relinked a reconfigure event is received and the
61 * stream is restarted.
64 * While GstAdaptiveDemux is responsible for the workflow, it knows nothing
65 * about the intrinsics of the subclass formats, so the subclasses are
66 * responsible for maintaining the manifest data structures and stream
72 The following rules were observed while implementing MT safety in adaptive demux:
73 1. If a variable is accessed from multiple threads and at least one thread
74 writes to it, then all the accesses needs to be done from inside a critical section.
75 2. If thread A wants to join thread B then at the moment it calls gst_task_join
76 it must not hold any mutexes that thread B might take.
78 Adaptive demux API can be called from several threads. More, adaptive demux
79 starts some threads to monitor the download of fragments. In order to protect
80 accesses to shared variables (demux and streams) all the API functions that
81 can be run in different threads will need to get a mutex (manifest_lock)
82 when they start and release it when they end. Because some of those functions
83 can indirectly call other API functions (eg they can generate events or messages
84 that are processed in the same thread) the manifest_lock must be recursive.
86 The manifest_lock will serialize the public API making access to shared
87 variables safe. But some of these functions will try at some moment to join
88 threads created by adaptive demux, or to change the state of src elements
89 (which will block trying to join the src element streaming thread). Because
90 of rule 2, those functions will need to release the manifest_lock during the
91 call of gst_task_join. During this time they can be interrupted by other API calls.
92 For example, during the precessing of a seek event, gst_adaptive_demux_stop_tasks
93 is called and this will join all threads. In order to prevent interruptions
94 during such period, all the API functions will also use a second lock: api_lock.
95 This will be taken at the beginning of the function and released at the end,
96 but this time this lock will not be temporarily released during join.
97 This lock will be used only by API calls (not by gst_adaptive_demux_stream_download_loop
98 or gst_adaptive_demux_updates_loop or _src_chain or _src_event) so it is safe
99 to hold it while joining the threads or changing the src element state. The
100 api_lock will serialise all external requests to adaptive demux. In order to
101 avoid deadlocks, if a function needs to acquire both manifest and api locks,
102 the api_lock will be taken first and the manifest_lock second.
104 By using the api_lock a thread is protected against other API calls. But when
105 temporarily dropping the manifest_lock, it will be vulnerable to changes from
106 threads that use only the manifest_lock and not the api_lock. These threads run
107 one of the following functions: gst_adaptive_demux_stream_download_loop,
108 gst_adaptive_demux_updates_loop, _src_chain, _src_event. In order to guarantee
109 that all operations during an API call are not impacted by other writes, the
110 above mentioned functions must check a cancelled flag every time they reacquire
111 the manifest_lock. If the flag is set, they must exit immediately, without
112 performing any changes on the shared data. In this way, an API call (eg seek
113 request) can set the cancel flag before releasing the manifest_lock and be sure
114 that the demux object and its streams are not changed by anybody else.
121 #include "gstadaptivedemux.h"
122 #include "gst/gst-i18n-plugin.h"
123 #include <gst/base/gstadapter.h>
125 GST_DEBUG_CATEGORY (adaptivedemux_debug);
126 #define GST_CAT_DEFAULT adaptivedemux_debug
128 #define MAX_DOWNLOAD_ERROR_COUNT 3
129 #define DEFAULT_FAILED_COUNT 3
130 #define DEFAULT_CONNECTION_SPEED 0
131 #define DEFAULT_BITRATE_LIMIT 0.8f
132 #define SRC_QUEUE_MAX_BYTES 20 * 1024 * 1024 /* For safety. Large enough to hold a segment. */
133 #define NUM_LOOKBACK_FRAGMENTS 3
135 #define GST_MANIFEST_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->manifest_lock))
136 #define GST_MANIFEST_LOCK(d) G_STMT_START { \
137 GST_TRACE("Locking from thread %p", g_thread_self()); \
138 g_rec_mutex_lock (GST_MANIFEST_GET_LOCK (d)); \
139 GST_TRACE("Locked from thread %p", g_thread_self()); \
142 #define GST_MANIFEST_UNLOCK(d) G_STMT_START { \
143 GST_TRACE("Unlocking from thread %p", g_thread_self()); \
144 g_rec_mutex_unlock (GST_MANIFEST_GET_LOCK (d)); \
147 #define GST_API_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->api_lock))
148 #define GST_API_LOCK(d) g_mutex_lock (GST_API_GET_LOCK (d));
149 #define GST_API_UNLOCK(d) g_mutex_unlock (GST_API_GET_LOCK (d));
151 #define GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK(d) (&GST_ADAPTIVE_DEMUX_CAST(d)->priv->segment_lock)
152 #define GST_ADAPTIVE_DEMUX_SEGMENT_LOCK(d) g_mutex_lock (GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK (d))
153 #define GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK(d) g_mutex_unlock (GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK (d))
158 PROP_CONNECTION_SPEED,
163 /* Internal, so not using GST_FLOW_CUSTOM_SUCCESS_N */
164 #define GST_ADAPTIVE_DEMUX_FLOW_SWITCH (GST_FLOW_CUSTOM_SUCCESS_2 + 1)
166 struct _GstAdaptiveDemuxPrivate
168 GstAdapter *input_adapter; /* protected by manifest_lock */
169 gint have_manifest; /* MT safe */
171 GList *old_streams; /* protected by manifest_lock */
173 GstTask *updates_task; /* MT safe */
174 GRecMutex updates_lock;
175 GMutex updates_timed_lock;
176 GCond updates_timed_cond; /* protected by updates_timed_lock */
177 gboolean stop_updates_task; /* protected by updates_timed_lock */
179 /* used only from updates_task, no need to protect it */
180 gint update_failed_count;
182 guint32 segment_seqnum; /* protected by manifest_lock */
184 /* main lock used to protect adaptive demux and all its streams.
185 * It serializes the adaptive demux public API.
187 GRecMutex manifest_lock;
189 /* condition to wait for manifest updates on a live stream.
190 * In order to signal the manifest_cond, the caller needs to hold both
191 * manifest_lock and manifest_update_lock (taken in this order)
194 GMutex manifest_update_lock;
196 /* Lock and condition for prerolling streams before exposing */
199 gint preroll_pending;
203 /* Protects demux and stream segment information
204 * Needed because seeks can update segment information
205 * without needing to stop tasks when they just want to
206 * update the segment boundaries */
209 GstClockTime qos_earliest_time;
212 typedef struct _GstAdaptiveDemuxTimer
219 } GstAdaptiveDemuxTimer;
221 static GstBinClass *parent_class = NULL;
222 static gint private_offset = 0;
224 static void gst_adaptive_demux_class_init (GstAdaptiveDemuxClass * klass);
225 static void gst_adaptive_demux_init (GstAdaptiveDemux * dec,
226 GstAdaptiveDemuxClass * klass);
227 static void gst_adaptive_demux_finalize (GObject * object);
228 static GstStateChangeReturn gst_adaptive_demux_change_state (GstElement *
229 element, GstStateChange transition);
231 static void gst_adaptive_demux_handle_message (GstBin * bin, GstMessage * msg);
233 static gboolean gst_adaptive_demux_sink_event (GstPad * pad, GstObject * parent,
235 static GstFlowReturn gst_adaptive_demux_sink_chain (GstPad * pad,
236 GstObject * parent, GstBuffer * buffer);
237 static gboolean gst_adaptive_demux_src_query (GstPad * pad, GstObject * parent,
239 static gboolean gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent,
243 gst_adaptive_demux_push_src_event (GstAdaptiveDemux * demux, GstEvent * event);
245 static void gst_adaptive_demux_updates_loop (GstAdaptiveDemux * demux);
246 static void gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream *
248 static void gst_adaptive_demux_reset (GstAdaptiveDemux * demux);
249 static gboolean gst_adaptive_demux_prepare_streams (GstAdaptiveDemux * demux,
250 gboolean first_and_live);
251 static gboolean gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux);
252 static gboolean gst_adaptive_demux_is_live (GstAdaptiveDemux * demux);
253 static GstFlowReturn gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux,
254 GstAdaptiveDemuxStream * stream, gboolean forward, GstSeekFlags flags,
255 GstClockTime ts, GstClockTime * final_ts);
256 static gboolean gst_adaptive_demux_stream_has_next_fragment (GstAdaptiveDemux *
257 demux, GstAdaptiveDemuxStream * stream);
258 static gboolean gst_adaptive_demux_stream_select_bitrate (GstAdaptiveDemux *
259 demux, GstAdaptiveDemuxStream * stream, guint64 bitrate);
261 gst_adaptive_demux_stream_update_fragment_info (GstAdaptiveDemux * demux,
262 GstAdaptiveDemuxStream * stream);
264 gst_adaptive_demux_stream_get_fragment_waiting_time (GstAdaptiveDemux * demux,
265 GstAdaptiveDemuxStream * stream);
266 static GstFlowReturn gst_adaptive_demux_update_manifest (GstAdaptiveDemux *
269 gst_adaptive_demux_update_manifest_default (GstAdaptiveDemux * demux);
270 static gboolean gst_adaptive_demux_has_next_period (GstAdaptiveDemux * demux);
271 static void gst_adaptive_demux_advance_period (GstAdaptiveDemux * demux);
273 static void gst_adaptive_demux_stream_free (GstAdaptiveDemuxStream * stream);
275 gst_adaptive_demux_stream_push_event (GstAdaptiveDemuxStream * stream,
278 static void gst_adaptive_demux_stop_manifest_update_task (GstAdaptiveDemux *
280 static void gst_adaptive_demux_start_manifest_update_task (GstAdaptiveDemux *
283 static void gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux,
284 gboolean start_preroll_streams);
285 static void gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux,
286 gboolean stop_updates);
287 static GstFlowReturn gst_adaptive_demux_combine_flows (GstAdaptiveDemux *
290 gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream *
291 stream, GstFlowReturn ret, GError * err);
293 gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux,
294 GstAdaptiveDemuxStream * stream, GstBuffer * buffer);
296 gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux,
297 GstAdaptiveDemuxStream * stream);
299 gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux,
300 GstAdaptiveDemuxStream * stream, GstClockTime duration);
302 gst_adaptive_demux_wait_until (GstClock * clock, GCond * cond, GMutex * mutex,
303 GstClockTime end_time);
304 static gboolean gst_adaptive_demux_clock_callback (GstClock * clock,
305 GstClockTime time, GstClockID id, gpointer user_data);
307 gst_adaptive_demux_requires_periodical_playlist_update_default (GstAdaptiveDemux
310 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
311 * method to get to the padtemplates */
313 gst_adaptive_demux_get_type (void)
315 static gsize type = 0;
317 if (g_once_init_enter (&type)) {
319 static const GTypeInfo info = {
320 sizeof (GstAdaptiveDemuxClass),
323 (GClassInitFunc) gst_adaptive_demux_class_init,
326 sizeof (GstAdaptiveDemux),
328 (GInstanceInitFunc) gst_adaptive_demux_init,
331 _type = g_type_register_static (GST_TYPE_BIN,
332 "GstAdaptiveDemux", &info, G_TYPE_FLAG_ABSTRACT);
335 g_type_add_instance_private (_type, sizeof (GstAdaptiveDemuxPrivate));
337 g_once_init_leave (&type, _type);
342 static inline GstAdaptiveDemuxPrivate *
343 gst_adaptive_demux_get_instance_private (GstAdaptiveDemux * self)
345 return (G_STRUCT_MEMBER_P (self, private_offset));
349 gst_adaptive_demux_set_property (GObject * object, guint prop_id,
350 const GValue * value, GParamSpec * pspec)
352 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object);
354 GST_API_LOCK (demux);
355 GST_MANIFEST_LOCK (demux);
358 case PROP_CONNECTION_SPEED:
359 demux->connection_speed = g_value_get_uint (value) * 1000;
360 GST_DEBUG_OBJECT (demux, "Connection speed set to %u",
361 demux->connection_speed);
363 case PROP_BITRATE_LIMIT:
364 demux->bitrate_limit = g_value_get_float (value);
367 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
371 GST_MANIFEST_UNLOCK (demux);
372 GST_API_UNLOCK (demux);
376 gst_adaptive_demux_get_property (GObject * object, guint prop_id,
377 GValue * value, GParamSpec * pspec)
379 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object);
381 GST_MANIFEST_LOCK (demux);
384 case PROP_CONNECTION_SPEED:
385 g_value_set_uint (value, demux->connection_speed / 1000);
387 case PROP_BITRATE_LIMIT:
388 g_value_set_float (value, demux->bitrate_limit);
391 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
395 GST_MANIFEST_UNLOCK (demux);
399 gst_adaptive_demux_class_init (GstAdaptiveDemuxClass * klass)
401 GObjectClass *gobject_class;
402 GstElementClass *gstelement_class;
403 GstBinClass *gstbin_class;
405 gobject_class = G_OBJECT_CLASS (klass);
406 gstelement_class = GST_ELEMENT_CLASS (klass);
407 gstbin_class = GST_BIN_CLASS (klass);
409 GST_DEBUG_CATEGORY_INIT (adaptivedemux_debug, "adaptivedemux", 0,
410 "Base Adaptive Demux");
412 parent_class = g_type_class_peek_parent (klass);
414 if (private_offset != 0)
415 g_type_class_adjust_private_offset (klass, &private_offset);
417 gobject_class->set_property = gst_adaptive_demux_set_property;
418 gobject_class->get_property = gst_adaptive_demux_get_property;
419 gobject_class->finalize = gst_adaptive_demux_finalize;
421 g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED,
422 g_param_spec_uint ("connection-speed", "Connection Speed",
423 "Network connection speed in kbps (0 = calculate from downloaded"
424 " fragments)", 0, G_MAXUINT / 1000, DEFAULT_CONNECTION_SPEED,
425 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
427 /* FIXME 2.0: rename this property to bandwidth-usage or any better name */
428 g_object_class_install_property (gobject_class, PROP_BITRATE_LIMIT,
429 g_param_spec_float ("bitrate-limit",
430 "Bitrate limit in %",
431 "Limit of the available bitrate to use when switching to alternates.",
432 0, 1, DEFAULT_BITRATE_LIMIT,
433 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
435 gstelement_class->change_state = gst_adaptive_demux_change_state;
437 gstbin_class->handle_message = gst_adaptive_demux_handle_message;
439 klass->data_received = gst_adaptive_demux_stream_data_received_default;
440 klass->finish_fragment = gst_adaptive_demux_stream_finish_fragment_default;
441 klass->update_manifest = gst_adaptive_demux_update_manifest_default;
442 klass->requires_periodical_playlist_update =
443 gst_adaptive_demux_requires_periodical_playlist_update_default;
448 gst_adaptive_demux_init (GstAdaptiveDemux * demux,
449 GstAdaptiveDemuxClass * klass)
451 GstPadTemplate *pad_template;
452 GstClockType clock_type = GST_CLOCK_TYPE_OTHER;
453 GObjectClass *gobject_class;
455 GST_DEBUG_OBJECT (demux, "gst_adaptive_demux_init");
457 demux->priv = gst_adaptive_demux_get_instance_private (demux);
458 demux->priv->input_adapter = gst_adapter_new ();
459 demux->downloader = gst_uri_downloader_new ();
460 gst_uri_downloader_set_parent (demux->downloader, GST_ELEMENT_CAST (demux));
461 demux->stream_struct_size = sizeof (GstAdaptiveDemuxStream);
462 demux->priv->segment_seqnum = gst_util_seqnum_next ();
463 demux->have_group_id = FALSE;
464 demux->group_id = G_MAXUINT;
466 gst_segment_init (&demux->segment, GST_FORMAT_TIME);
468 gst_bin_set_suppressed_flags (GST_BIN_CAST (demux),
469 GST_ELEMENT_FLAG_SOURCE | GST_ELEMENT_FLAG_SINK);
471 demux->realtime_clock = gst_system_clock_obtain ();
472 g_assert (demux->realtime_clock != NULL);
473 gobject_class = G_OBJECT_GET_CLASS (demux->realtime_clock);
474 if (g_object_class_find_property (gobject_class, "clock-type")) {
475 g_object_get (demux->realtime_clock, "clock-type", &clock_type, NULL);
477 GST_WARNING_OBJECT (demux,
478 "System clock does not have clock-type property");
480 if (clock_type == GST_CLOCK_TYPE_REALTIME) {
481 demux->clock_offset = 0;
484 GstClockTime rtc_now;
486 utc_now = g_date_time_new_now_utc ();
487 rtc_now = gst_clock_get_time (demux->realtime_clock);
488 demux->clock_offset =
489 g_date_time_to_unix (utc_now) * G_TIME_SPAN_SECOND +
490 g_date_time_get_microsecond (utc_now) - GST_TIME_AS_USECONDS (rtc_now);
491 g_date_time_unref (utc_now);
493 g_rec_mutex_init (&demux->priv->updates_lock);
494 demux->priv->updates_task =
495 gst_task_new ((GstTaskFunction) gst_adaptive_demux_updates_loop,
497 gst_task_set_lock (demux->priv->updates_task, &demux->priv->updates_lock);
499 g_mutex_init (&demux->priv->updates_timed_lock);
500 g_cond_init (&demux->priv->updates_timed_cond);
502 g_cond_init (&demux->priv->manifest_cond);
503 g_mutex_init (&demux->priv->manifest_update_lock);
505 g_rec_mutex_init (&demux->priv->manifest_lock);
506 g_mutex_init (&demux->priv->api_lock);
507 g_mutex_init (&demux->priv->segment_lock);
509 g_cond_init (&demux->priv->preroll_cond);
510 g_mutex_init (&demux->priv->preroll_lock);
513 gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "sink");
514 g_return_if_fail (pad_template != NULL);
516 demux->sinkpad = gst_pad_new_from_template (pad_template, "sink");
517 gst_pad_set_event_function (demux->sinkpad,
518 GST_DEBUG_FUNCPTR (gst_adaptive_demux_sink_event));
519 gst_pad_set_chain_function (demux->sinkpad,
520 GST_DEBUG_FUNCPTR (gst_adaptive_demux_sink_chain));
523 demux->bitrate_limit = DEFAULT_BITRATE_LIMIT;
524 demux->connection_speed = DEFAULT_CONNECTION_SPEED;
526 gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
530 gst_adaptive_demux_finalize (GObject * object)
532 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (object);
533 GstAdaptiveDemuxPrivate *priv = demux->priv;
535 GST_DEBUG_OBJECT (object, "finalize");
537 g_object_unref (priv->input_adapter);
538 g_object_unref (demux->downloader);
540 g_mutex_clear (&priv->updates_timed_lock);
541 g_cond_clear (&priv->updates_timed_cond);
542 g_mutex_clear (&demux->priv->manifest_update_lock);
543 g_cond_clear (&demux->priv->manifest_cond);
544 g_object_unref (priv->updates_task);
545 g_rec_mutex_clear (&priv->updates_lock);
546 g_rec_mutex_clear (&demux->priv->manifest_lock);
547 g_mutex_clear (&demux->priv->api_lock);
548 g_mutex_clear (&demux->priv->segment_lock);
549 if (demux->realtime_clock) {
550 gst_object_unref (demux->realtime_clock);
551 demux->realtime_clock = NULL;
554 g_cond_clear (&demux->priv->preroll_cond);
555 g_mutex_clear (&demux->priv->preroll_lock);
557 G_OBJECT_CLASS (parent_class)->finalize (object);
560 static GstStateChangeReturn
561 gst_adaptive_demux_change_state (GstElement * element,
562 GstStateChange transition)
564 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (element);
565 GstStateChangeReturn result = GST_STATE_CHANGE_FAILURE;
567 switch (transition) {
568 case GST_STATE_CHANGE_PAUSED_TO_READY:
569 if (g_atomic_int_compare_and_exchange (&demux->running, TRUE, FALSE))
570 GST_DEBUG_OBJECT (demux, "demuxer has stopped running");
571 gst_uri_downloader_cancel (demux->downloader);
573 GST_API_LOCK (demux);
574 GST_MANIFEST_LOCK (demux);
575 gst_adaptive_demux_reset (demux);
576 GST_MANIFEST_UNLOCK (demux);
577 GST_API_UNLOCK (demux);
579 case GST_STATE_CHANGE_READY_TO_PAUSED:
580 GST_API_LOCK (demux);
581 GST_MANIFEST_LOCK (demux);
582 gst_adaptive_demux_reset (demux);
583 /* Clear "cancelled" flag in uridownloader since subclass might want to
584 * use uridownloader to fetch another manifest */
585 gst_uri_downloader_reset (demux->downloader);
586 if (g_atomic_int_get (&demux->priv->have_manifest))
587 gst_adaptive_demux_start_manifest_update_task (demux);
588 GST_MANIFEST_UNLOCK (demux);
589 GST_API_UNLOCK (demux);
590 if (g_atomic_int_compare_and_exchange (&demux->running, FALSE, TRUE))
591 GST_DEBUG_OBJECT (demux, "demuxer has started running");
597 /* this must be run without MANIFEST_LOCK taken.
598 * For PLAYING to PLAYING state changes, it will want to take a lock in
599 * src element and that lock is held while the streaming thread is running.
600 * The streaming thread will take the MANIFEST_LOCK, leading to a deadlock.
602 result = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
608 gst_adaptive_demux_sink_event (GstPad * pad, GstObject * parent,
611 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent);
614 switch (event->type) {
615 case GST_EVENT_FLUSH_STOP:{
616 GST_API_LOCK (demux);
617 GST_MANIFEST_LOCK (demux);
619 gst_adaptive_demux_reset (demux);
621 ret = gst_pad_event_default (pad, parent, event);
623 GST_MANIFEST_UNLOCK (demux);
624 GST_API_UNLOCK (demux);
629 GstAdaptiveDemuxClass *demux_class;
634 GstBuffer *manifest_buffer;
636 GST_API_LOCK (demux);
637 GST_MANIFEST_LOCK (demux);
639 demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
641 available = gst_adapter_available (demux->priv->input_adapter);
643 if (available == 0) {
644 GST_WARNING_OBJECT (demux, "Received EOS without a manifest.");
645 ret = gst_pad_event_default (pad, parent, event);
647 GST_MANIFEST_UNLOCK (demux);
648 GST_API_UNLOCK (demux);
653 GST_DEBUG_OBJECT (demux, "Got EOS on the sink pad: manifest fetched");
655 /* Need to get the URI to use it as a base to generate the fragment's
657 query = gst_query_new_uri ();
658 query_res = gst_pad_peer_query (pad, query);
660 gchar *uri, *redirect_uri;
663 gst_query_parse_uri (query, &uri);
664 gst_query_parse_uri_redirection (query, &redirect_uri);
665 gst_query_parse_uri_redirection_permanent (query, &permanent);
667 if (permanent && redirect_uri) {
668 demux->manifest_uri = redirect_uri;
669 demux->manifest_base_uri = NULL;
672 demux->manifest_uri = uri;
673 demux->manifest_base_uri = redirect_uri;
676 GST_DEBUG_OBJECT (demux, "Fetched manifest at URI: %s (base: %s)",
677 demux->manifest_uri, GST_STR_NULL (demux->manifest_base_uri));
679 GST_WARNING_OBJECT (demux, "Upstream URI query failed.");
681 gst_query_unref (query);
683 /* Let the subclass parse the manifest */
685 gst_adapter_take_buffer (demux->priv->input_adapter, available);
686 if (!demux_class->process_manifest (demux, manifest_buffer)) {
687 /* In most cases, this will happen if we set a wrong url in the
688 * source element and we have received the 404 HTML response instead of
690 GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid manifest."),
694 g_atomic_int_set (&demux->priv->have_manifest, TRUE);
696 gst_buffer_unref (manifest_buffer);
698 gst_element_post_message (GST_ELEMENT_CAST (demux),
699 gst_message_new_element (GST_OBJECT_CAST (demux),
700 gst_structure_new (GST_ADAPTIVE_DEMUX_STATISTICS_MESSAGE_NAME,
701 "manifest-uri", G_TYPE_STRING,
702 demux->manifest_uri, "uri", G_TYPE_STRING,
704 "manifest-download-start", GST_TYPE_CLOCK_TIME,
706 "manifest-download-stop", GST_TYPE_CLOCK_TIME,
707 gst_util_get_timestamp (), NULL)));
710 /* Send duration message */
711 if (!gst_adaptive_demux_is_live (demux)) {
712 GstClockTime duration = demux_class->get_duration (demux);
714 if (duration != GST_CLOCK_TIME_NONE) {
715 GST_DEBUG_OBJECT (demux,
716 "Sending duration message : %" GST_TIME_FORMAT,
717 GST_TIME_ARGS (duration));
718 gst_element_post_message (GST_ELEMENT (demux),
719 gst_message_new_duration_changed (GST_OBJECT (demux)));
721 GST_DEBUG_OBJECT (demux,
722 "media duration unknown, can not send the duration message");
726 if (demux->next_streams) {
727 gst_adaptive_demux_prepare_streams (demux,
728 gst_adaptive_demux_is_live (demux));
729 gst_adaptive_demux_start_tasks (demux, TRUE);
730 gst_adaptive_demux_start_manifest_update_task (demux);
733 GST_WARNING_OBJECT (demux, "No streams created from manifest");
734 GST_ELEMENT_ERROR (demux, STREAM, DEMUX,
735 (_("This file contains no playable streams.")),
736 ("No known stream formats found at the Manifest"));
741 GST_MANIFEST_UNLOCK (demux);
742 GST_API_UNLOCK (demux);
744 gst_event_unref (event);
747 case GST_EVENT_SEGMENT:
748 /* Swallow newsegments, we'll push our own */
749 gst_event_unref (event);
755 return gst_pad_event_default (pad, parent, event);
759 gst_adaptive_demux_sink_chain (GstPad * pad, GstObject * parent,
762 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent);
764 GST_MANIFEST_LOCK (demux);
766 gst_adapter_push (demux->priv->input_adapter, buffer);
768 GST_INFO_OBJECT (demux, "Received manifest buffer, total size is %i bytes",
769 (gint) gst_adapter_available (demux->priv->input_adapter));
771 GST_MANIFEST_UNLOCK (demux);
775 /* must be called with manifest_lock taken */
777 gst_adaptive_demux_reset (GstAdaptiveDemux * demux)
779 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
784 /* take ownership of old_streams before releasing the manifest_lock in
785 * gst_adaptive_demux_stop_tasks
787 old_streams = demux->priv->old_streams;
788 demux->priv->old_streams = NULL;
790 gst_adaptive_demux_stop_tasks (demux, TRUE);
793 klass->reset (demux);
795 eos = gst_event_new_eos ();
796 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
797 GstAdaptiveDemuxStream *stream = iter->data;
799 gst_pad_push_event (stream->pad, gst_event_ref (eos));
800 gst_pad_set_active (stream->pad, FALSE);
802 gst_element_remove_pad (GST_ELEMENT_CAST (demux), stream->pad);
804 gst_adaptive_demux_stream_free (stream);
806 gst_event_unref (eos);
807 g_list_free (demux->streams);
808 demux->streams = NULL;
809 if (demux->prepared_streams) {
810 g_list_free_full (demux->prepared_streams,
811 (GDestroyNotify) gst_adaptive_demux_stream_free);
812 demux->prepared_streams = NULL;
814 if (demux->next_streams) {
815 g_list_free_full (demux->next_streams,
816 (GDestroyNotify) gst_adaptive_demux_stream_free);
817 demux->next_streams = NULL;
821 g_list_free_full (old_streams,
822 (GDestroyNotify) gst_adaptive_demux_stream_free);
825 if (demux->priv->old_streams) {
826 g_list_free_full (demux->priv->old_streams,
827 (GDestroyNotify) gst_adaptive_demux_stream_free);
828 demux->priv->old_streams = NULL;
831 g_free (demux->manifest_uri);
832 g_free (demux->manifest_base_uri);
833 demux->manifest_uri = NULL;
834 demux->manifest_base_uri = NULL;
836 gst_adapter_clear (demux->priv->input_adapter);
837 g_atomic_int_set (&demux->priv->have_manifest, FALSE);
839 gst_segment_init (&demux->segment, GST_FORMAT_TIME);
841 demux->have_group_id = FALSE;
842 demux->group_id = G_MAXUINT;
843 demux->priv->segment_seqnum = gst_util_seqnum_next ();
847 gst_adaptive_demux_handle_message (GstBin * bin, GstMessage * msg)
849 GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (bin);
851 switch (GST_MESSAGE_TYPE (msg)) {
852 case GST_MESSAGE_ERROR:{
854 GstAdaptiveDemuxStream *stream = NULL;
857 gchar *new_error = NULL;
858 const GstStructure *details = NULL;
860 GST_MANIFEST_LOCK (demux);
862 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
863 GstAdaptiveDemuxStream *cur = iter->data;
864 if (gst_object_has_as_ancestor (GST_MESSAGE_SRC (msg),
865 GST_OBJECT_CAST (cur->src))) {
870 if (stream == NULL) {
871 for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) {
872 GstAdaptiveDemuxStream *cur = iter->data;
873 if (gst_object_has_as_ancestor (GST_MESSAGE_SRC (msg),
874 GST_OBJECT_CAST (cur->src))) {
879 if (stream == NULL) {
880 GST_WARNING_OBJECT (demux,
881 "Failed to locate stream for errored element");
886 gst_message_parse_error (msg, &err, &debug);
888 GST_WARNING_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
889 "Source posted error: %d:%d %s (%s)", err->domain, err->code,
890 err->message, debug);
893 new_error = g_strdup_printf ("%s: %s\n", err->message, debug);
895 g_free (err->message);
896 err->message = new_error;
899 gst_message_parse_error_details (msg, &details);
901 gst_structure_get_uint (details, "http-status-code",
902 &stream->last_status_code);
905 /* error, but ask to retry */
906 gst_adaptive_demux_stream_fragment_download_finish (stream,
907 GST_FLOW_CUSTOM_ERROR, err);
912 GST_MANIFEST_UNLOCK (demux);
914 gst_message_unref (msg);
923 GST_BIN_CLASS (parent_class)->handle_message (bin, msg);
927 gst_adaptive_demux_set_stream_struct_size (GstAdaptiveDemux * demux,
930 GST_API_LOCK (demux);
931 GST_MANIFEST_LOCK (demux);
932 demux->stream_struct_size = struct_size;
933 GST_MANIFEST_UNLOCK (demux);
934 GST_API_UNLOCK (demux);
937 /* must be called with manifest_lock taken */
939 gst_adaptive_demux_prepare_stream (GstAdaptiveDemux * demux,
940 GstAdaptiveDemuxStream * stream)
942 GstPad *pad = stream->pad;
943 gchar *name = gst_pad_get_name (pad);
947 gst_pad_set_active (pad, TRUE);
948 stream->need_header = TRUE;
950 stream_id = gst_pad_create_stream_id (pad, GST_ELEMENT_CAST (demux), name);
953 gst_pad_get_sticky_event (GST_ADAPTIVE_DEMUX_SINK_PAD (demux),
954 GST_EVENT_STREAM_START, 0);
956 if (gst_event_parse_group_id (event, &demux->group_id))
957 demux->have_group_id = TRUE;
959 demux->have_group_id = FALSE;
960 gst_event_unref (event);
961 } else if (!demux->have_group_id) {
962 demux->have_group_id = TRUE;
963 demux->group_id = gst_util_group_id_next ();
965 event = gst_event_new_stream_start (stream_id);
966 if (demux->have_group_id)
967 gst_event_set_group_id (event, demux->group_id);
969 gst_pad_push_event (pad, event);
973 GST_DEBUG_OBJECT (demux, "Preparing srcpad %s:%s", GST_DEBUG_PAD_NAME (pad));
975 stream->discont = TRUE;
981 gst_adaptive_demux_expose_stream (GstAdaptiveDemux * demux,
982 GstAdaptiveDemuxStream * stream)
985 GstPad *pad = stream->pad;
988 if (stream->pending_caps) {
989 gst_pad_set_caps (pad, stream->pending_caps);
990 caps = stream->pending_caps;
991 stream->pending_caps = NULL;
993 caps = gst_pad_get_current_caps (pad);
996 GST_DEBUG_OBJECT (demux, "Exposing srcpad %s:%s with caps %" GST_PTR_FORMAT,
997 GST_DEBUG_PAD_NAME (pad), caps);
999 gst_caps_unref (caps);
1001 gst_object_ref (pad);
1003 /* Don't hold the manifest lock while exposing a pad */
1004 GST_MANIFEST_UNLOCK (demux);
1005 ret = gst_element_add_pad (GST_ELEMENT_CAST (demux), pad);
1006 GST_MANIFEST_LOCK (demux);
1011 /* must be called with manifest_lock taken */
1013 gst_adaptive_demux_stream_get_presentation_offset (GstAdaptiveDemux * demux,
1014 GstAdaptiveDemuxStream * stream)
1016 GstAdaptiveDemuxClass *klass;
1018 klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1020 if (klass->get_presentation_offset == NULL)
1023 return klass->get_presentation_offset (demux, stream);
1026 /* must be called with manifest_lock taken */
1028 gst_adaptive_demux_get_period_start_time (GstAdaptiveDemux * demux)
1030 GstAdaptiveDemuxClass *klass;
1032 klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1034 if (klass->get_period_start_time == NULL)
1037 return klass->get_period_start_time (demux);
1040 /* must be called with manifest_lock taken */
1042 gst_adaptive_demux_prepare_streams (GstAdaptiveDemux * demux,
1043 gboolean first_and_live)
1046 GstClockTime period_start, min_pts = GST_CLOCK_TIME_NONE;
1048 g_return_val_if_fail (demux->next_streams != NULL, FALSE);
1049 if (demux->prepared_streams != NULL) {
1050 /* Old streams that were never exposed, due to a seek or so */
1051 GST_FIXME_OBJECT (demux,
1052 "Preparing new streams without cleaning up old ones!");
1056 demux->prepared_streams = demux->next_streams;
1057 demux->next_streams = NULL;
1059 if (!gst_adaptive_demux_is_running (demux)) {
1060 GST_DEBUG_OBJECT (demux, "Not exposing pads due to shutdown");
1064 for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) {
1065 GstAdaptiveDemuxStream *stream = iter->data;
1067 stream->do_block = TRUE;
1069 if (!gst_adaptive_demux_prepare_stream (demux,
1070 GST_ADAPTIVE_DEMUX_STREAM_CAST (stream))) {
1071 /* TODO act on error */
1072 GST_FIXME_OBJECT (stream->pad,
1073 "Do something on failure to expose stream");
1076 if (first_and_live) {
1077 /* TODO we only need the first timestamp, maybe create a simple function to
1078 * get the current PTS of a fragment ? */
1079 GST_DEBUG_OBJECT (demux, "Calling update_fragment_info");
1080 gst_adaptive_demux_stream_update_fragment_info (demux, stream);
1082 if (GST_CLOCK_TIME_IS_VALID (min_pts)) {
1083 min_pts = MIN (min_pts, stream->fragment.timestamp);
1085 min_pts = stream->fragment.timestamp;
1090 period_start = gst_adaptive_demux_get_period_start_time (demux);
1092 /* For live streams, the subclass is supposed to seek to the current
1093 * fragment and then tell us its timestamp in stream->fragment.timestamp.
1094 * We now also have to seek our demuxer segment to reflect this.
1096 * FIXME: This needs some refactoring at some point.
1098 if (first_and_live) {
1099 gst_segment_do_seek (&demux->segment, demux->segment.rate, GST_FORMAT_TIME,
1100 GST_SEEK_FLAG_FLUSH, GST_SEEK_TYPE_SET, min_pts + period_start,
1101 GST_SEEK_TYPE_NONE, -1, NULL);
1104 for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) {
1105 GstAdaptiveDemuxStream *stream = iter->data;
1106 GstClockTime offset;
1108 offset = gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
1109 stream->segment = demux->segment;
1111 /* The demuxer segment is just built from seek events, but for each stream
1112 * we have to adjust segments according to the current period and the
1113 * stream specific presentation time offset.
1115 * For each period, buffer timestamps start again from 0. Additionally the
1116 * buffer timestamps are shifted by the stream specific presentation time
1117 * offset, so the first buffer timestamp of a period is 0 + presentation
1118 * time offset. If the stream contains timestamps itself, this is also
1119 * supposed to be the presentation time stored inside the stream.
1121 * The stream time over periods is supposed to be continuous, that is the
1122 * buffer timestamp 0 + presentation time offset should map to the start
1123 * time of the current period.
1126 * The adjustment of the stream segments as such works the following.
1128 * If the demuxer segment start is bigger than the period start, this
1129 * means that we have to drop some media at the beginning of the current
1130 * period, e.g. because a seek into the middle of the period has
1131 * happened. The amount of media to drop is the difference between the
1132 * period start and the demuxer segment start, and as each period starts
1133 * again from 0, this difference is going to be the actual stream's
1134 * segment start. As all timestamps of the stream are shifted by the
1135 * presentation time offset, we will also have to move the segment start
1138 * Likewise, the demuxer segment stop value is adjusted in the same
1141 * Now the running time and stream time at the stream's segment start has
1142 * to be the one that is stored inside the demuxer's segment, which means
1143 * that segment.base and segment.time have to be copied over (done just
1147 * If the demuxer segment start is smaller than the period start time,
1148 * this means that the whole period is inside the segment. As each period
1149 * starts timestamps from 0, and additionally timestamps are shifted by
1150 * the presentation time offset, the stream's first timestamp (and as such
1151 * the stream's segment start) has to be the presentation time offset.
1152 * The stream time at the segment start is supposed to be the stream time
1153 * of the period start according to the demuxer segment, so the stream
1154 * segment's time would be set to that. The same goes for the stream
1155 * segment's base, which is supposed to be the running time of the period
1156 * start according to the demuxer's segment.
1158 * The same logic applies for negative rates with the segment stop and
1159 * the period stop time (which gets clamped).
1162 * For the first case where not the complete period is inside the segment,
1163 * the segment time and base as calculated by the second case would be
1166 GST_DEBUG_OBJECT (demux, "Using demux segment %" GST_SEGMENT_FORMAT,
1168 GST_DEBUG_OBJECT (demux,
1169 "period_start: %" GST_TIME_FORMAT " offset: %" GST_TIME_FORMAT,
1170 GST_TIME_ARGS (period_start), GST_TIME_ARGS (offset));
1171 /* note for readers:
1172 * Since stream->segment is initially a copy of demux->segment,
1173 * only the values that need updating are modified below. */
1174 if (first_and_live) {
1175 /* If first and live, demuxer did seek to the current position already */
1176 stream->segment.start = demux->segment.start - period_start + offset;
1177 if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop))
1178 stream->segment.stop = demux->segment.stop - period_start + offset;
1179 /* FIXME : Do we need to handle negative rates for this ? */
1180 stream->segment.position = stream->segment.start;
1181 } else if (demux->segment.start > period_start) {
1182 /* seek within a period */
1183 stream->segment.start = demux->segment.start - period_start + offset;
1184 if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop))
1185 stream->segment.stop = demux->segment.stop - period_start + offset;
1186 if (stream->segment.rate >= 0)
1187 stream->segment.position = offset;
1189 stream->segment.position = stream->segment.stop;
1191 stream->segment.start = offset;
1192 if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop))
1193 stream->segment.stop = demux->segment.stop - period_start + offset;
1194 if (stream->segment.rate >= 0)
1195 stream->segment.position = offset;
1197 stream->segment.position = stream->segment.stop;
1198 stream->segment.time =
1199 gst_segment_to_stream_time (&demux->segment, GST_FORMAT_TIME,
1201 stream->segment.base =
1202 gst_segment_to_running_time (&demux->segment, GST_FORMAT_TIME,
1206 stream->pending_segment = gst_event_new_segment (&stream->segment);
1207 gst_event_set_seqnum (stream->pending_segment, demux->priv->segment_seqnum);
1209 GST_DEBUG_OBJECT (demux,
1210 "Prepared segment %" GST_SEGMENT_FORMAT " for stream %p",
1211 &stream->segment, stream);
1213 demux->priv->qos_earliest_time = GST_CLOCK_TIME_NONE;
1219 gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux)
1224 g_return_val_if_fail (demux->prepared_streams != NULL, FALSE);
1226 old_streams = demux->streams;
1227 demux->streams = demux->prepared_streams;
1228 demux->prepared_streams = NULL;
1230 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
1231 GstAdaptiveDemuxStream *stream = iter->data;
1233 if (!gst_adaptive_demux_expose_stream (demux,
1234 GST_ADAPTIVE_DEMUX_STREAM_CAST (stream))) {
1235 /* TODO act on error */
1238 demux->priv->preroll_pending = 0;
1240 GST_MANIFEST_UNLOCK (demux);
1241 gst_element_no_more_pads (GST_ELEMENT_CAST (demux));
1242 GST_MANIFEST_LOCK (demux);
1245 GstEvent *eos = gst_event_new_eos ();
1247 /* before we put streams in the demux->priv->old_streams list,
1248 * we ask the download task to stop. In this way, it will no longer be
1249 * allowed to change the demux object.
1251 for (iter = old_streams; iter; iter = g_list_next (iter)) {
1252 GstAdaptiveDemuxStream *stream = iter->data;
1253 GstPad *pad = gst_object_ref (GST_PAD (stream->pad));
1255 GST_MANIFEST_UNLOCK (demux);
1257 GST_DEBUG_OBJECT (pad, "Pushing EOS");
1258 gst_pad_push_event (pad, gst_event_ref (eos));
1259 gst_pad_set_active (pad, FALSE);
1261 GST_LOG_OBJECT (pad, "Removing stream");
1262 gst_element_remove_pad (GST_ELEMENT (demux), pad);
1263 GST_MANIFEST_LOCK (demux);
1265 gst_object_unref (GST_OBJECT (pad));
1267 /* ask the download task to stop.
1268 * We will not join it now, because our thread can be one of these tasks.
1269 * We will do the joining later, from another stream download task or
1270 * from gst_adaptive_demux_stop_tasks.
1271 * We also cannot change the state of the stream->src element, because
1272 * that will wait on the streaming thread (which could be this thread)
1274 * Because we sent an EOS to the downstream element, the stream->src
1275 * element should detect this in its streaming task and stop.
1276 * Even if it doesn't do that, we will change its state later in
1277 * gst_adaptive_demux_stop_tasks.
1279 GST_LOG_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
1280 "Marking stream as cancelled");
1281 gst_task_stop (stream->download_task);
1282 g_mutex_lock (&stream->fragment_download_lock);
1283 stream->cancelled = TRUE;
1284 stream->replaced = TRUE;
1285 g_cond_signal (&stream->fragment_download_cond);
1286 g_mutex_unlock (&stream->fragment_download_lock);
1288 gst_event_unref (eos);
1290 /* The list should be freed from another thread as we can't properly
1291 * cleanup a GstTask from itself */
1292 demux->priv->old_streams =
1293 g_list_concat (demux->priv->old_streams, old_streams);
1296 /* Unblock after removing oldstreams */
1297 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
1298 GstAdaptiveDemuxStream *stream = iter->data;
1299 stream->do_block = FALSE;
1302 GST_DEBUG_OBJECT (demux, "All streams are exposed");
1307 /* must be called with manifest_lock taken */
1308 GstAdaptiveDemuxStream *
1309 gst_adaptive_demux_stream_new (GstAdaptiveDemux * demux, GstPad * pad)
1311 GstAdaptiveDemuxStream *stream;
1313 stream = g_malloc0 (demux->stream_struct_size);
1315 /* Downloading task */
1316 g_rec_mutex_init (&stream->download_lock);
1317 stream->download_task =
1318 gst_task_new ((GstTaskFunction) gst_adaptive_demux_stream_download_loop,
1320 gst_task_set_lock (stream->download_task, &stream->download_lock);
1323 stream->demux = demux;
1324 stream->fragment_bitrates =
1325 g_malloc0 (sizeof (guint64) * NUM_LOOKBACK_FRAGMENTS);
1326 gst_pad_set_element_private (pad, stream);
1327 stream->qos_earliest_time = GST_CLOCK_TIME_NONE;
1329 g_mutex_lock (&demux->priv->preroll_lock);
1330 stream->do_block = TRUE;
1331 demux->priv->preroll_pending++;
1332 g_mutex_unlock (&demux->priv->preroll_lock);
1334 gst_pad_set_query_function (pad,
1335 GST_DEBUG_FUNCPTR (gst_adaptive_demux_src_query));
1336 gst_pad_set_event_function (pad,
1337 GST_DEBUG_FUNCPTR (gst_adaptive_demux_src_event));
1339 gst_segment_init (&stream->segment, GST_FORMAT_TIME);
1340 g_cond_init (&stream->fragment_download_cond);
1341 g_mutex_init (&stream->fragment_download_lock);
1343 demux->next_streams = g_list_append (demux->next_streams, stream);
1348 GstAdaptiveDemuxStream *
1349 gst_adaptive_demux_find_stream_for_pad (GstAdaptiveDemux * demux, GstPad * pad)
1353 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
1354 GstAdaptiveDemuxStream *stream = iter->data;
1355 if (stream->pad == pad) {
1363 /* must be called with manifest_lock taken.
1364 * It will temporarily drop the manifest_lock in order to join the task.
1365 * It will join only the old_streams (the demux->streams are joined by
1366 * gst_adaptive_demux_stop_tasks before gst_adaptive_demux_stream_free is
1370 gst_adaptive_demux_stream_free (GstAdaptiveDemuxStream * stream)
1372 GstAdaptiveDemux *demux = stream->demux;
1373 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1375 if (klass->stream_free)
1376 klass->stream_free (stream);
1378 g_clear_error (&stream->last_error);
1379 if (stream->download_task) {
1380 if (GST_TASK_STATE (stream->download_task) != GST_TASK_STOPPED) {
1381 GST_DEBUG_OBJECT (demux, "Leaving streaming task %s:%s",
1382 GST_DEBUG_PAD_NAME (stream->pad));
1384 gst_task_stop (stream->download_task);
1386 g_mutex_lock (&stream->fragment_download_lock);
1387 stream->cancelled = TRUE;
1388 g_cond_signal (&stream->fragment_download_cond);
1389 g_mutex_unlock (&stream->fragment_download_lock);
1391 GST_LOG_OBJECT (demux, "Waiting for task to finish");
1393 /* temporarily drop the manifest lock to join the task */
1394 GST_MANIFEST_UNLOCK (demux);
1396 gst_task_join (stream->download_task);
1398 GST_MANIFEST_LOCK (demux);
1400 GST_LOG_OBJECT (demux, "Finished");
1401 gst_object_unref (stream->download_task);
1402 g_rec_mutex_clear (&stream->download_lock);
1403 stream->download_task = NULL;
1406 gst_adaptive_demux_stream_fragment_clear (&stream->fragment);
1408 if (stream->pending_segment) {
1409 gst_event_unref (stream->pending_segment);
1410 stream->pending_segment = NULL;
1413 if (stream->pending_events) {
1414 g_list_free_full (stream->pending_events, (GDestroyNotify) gst_event_unref);
1415 stream->pending_events = NULL;
1418 if (stream->internal_pad) {
1419 gst_object_unparent (GST_OBJECT_CAST (stream->internal_pad));
1422 if (stream->src_srcpad) {
1423 gst_object_unref (stream->src_srcpad);
1424 stream->src_srcpad = NULL;
1428 GstElement *src = stream->src;
1432 GST_MANIFEST_UNLOCK (demux);
1433 gst_element_set_locked_state (src, TRUE);
1434 gst_element_set_state (src, GST_STATE_NULL);
1435 gst_bin_remove (GST_BIN_CAST (demux), src);
1436 GST_MANIFEST_LOCK (demux);
1439 g_cond_clear (&stream->fragment_download_cond);
1440 g_mutex_clear (&stream->fragment_download_lock);
1441 g_free (stream->fragment_bitrates);
1444 gst_object_unref (stream->pad);
1447 if (stream->pending_caps)
1448 gst_caps_unref (stream->pending_caps);
1450 g_clear_pointer (&stream->pending_tags, gst_tag_list_unref);
1455 /* must be called with manifest_lock taken */
1457 gst_adaptive_demux_get_live_seek_range (GstAdaptiveDemux * demux,
1458 gint64 * range_start, gint64 * range_stop)
1460 GstAdaptiveDemuxClass *klass;
1462 klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1464 g_return_val_if_fail (klass->get_live_seek_range, FALSE);
1466 return klass->get_live_seek_range (demux, range_start, range_stop);
1469 /* must be called with manifest_lock taken */
1471 gst_adaptive_demux_stream_in_live_seek_range (GstAdaptiveDemux * demux,
1472 GstAdaptiveDemuxStream * stream)
1474 gint64 range_start, range_stop;
1475 if (gst_adaptive_demux_get_live_seek_range (demux, &range_start, &range_stop)) {
1476 GST_LOG_OBJECT (stream->pad,
1477 "stream position %" GST_TIME_FORMAT " live seek range %"
1478 GST_STIME_FORMAT " - %" GST_STIME_FORMAT,
1479 GST_TIME_ARGS (stream->segment.position), GST_STIME_ARGS (range_start),
1480 GST_STIME_ARGS (range_stop));
1481 return (stream->segment.position >= range_start
1482 && stream->segment.position <= range_stop);
1488 /* must be called with manifest_lock taken */
1490 gst_adaptive_demux_can_seek (GstAdaptiveDemux * demux)
1492 GstAdaptiveDemuxClass *klass;
1494 klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1495 if (gst_adaptive_demux_is_live (demux)) {
1496 return klass->get_live_seek_range != NULL;
1499 return klass->seek != NULL;
1503 gst_adaptive_demux_update_streams_segment (GstAdaptiveDemux * demux,
1504 GList * streams, gint64 period_start, GstSeekType start_type,
1505 GstSeekType stop_type)
1508 for (iter = streams; iter; iter = g_list_next (iter)) {
1509 GstAdaptiveDemuxStream *stream = iter->data;
1511 GstClockTime offset;
1513 /* See comments in gst_adaptive_demux_get_period_start_time() for
1514 * an explanation of the segment modifications */
1515 stream->segment = demux->segment;
1516 offset = gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
1517 stream->segment.start += offset - period_start;
1518 if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop))
1519 stream->segment.stop += offset - period_start;
1520 if (demux->segment.rate > 0 && start_type != GST_SEEK_TYPE_NONE)
1521 stream->segment.position = stream->segment.start;
1522 else if (demux->segment.rate < 0 && stop_type != GST_SEEK_TYPE_NONE)
1523 stream->segment.position = stream->segment.stop;
1524 seg_evt = gst_event_new_segment (&stream->segment);
1525 gst_event_set_seqnum (seg_evt, demux->priv->segment_seqnum);
1526 gst_event_replace (&stream->pending_segment, seg_evt);
1527 GST_DEBUG_OBJECT (stream->pad, "Pending segment now %" GST_PTR_FORMAT,
1528 stream->pending_segment);
1529 gst_event_unref (seg_evt);
1530 /* Make sure the first buffer after a seek has the discont flag */
1531 stream->discont = TRUE;
1533 demux->priv->qos_earliest_time = GST_CLOCK_TIME_NONE;
1536 #define IS_SNAP_SEEK(f) (f & (GST_SEEK_FLAG_SNAP_BEFORE | \
1537 GST_SEEK_FLAG_SNAP_AFTER | \
1538 GST_SEEK_FLAG_SNAP_NEAREST | \
1539 GST_SEEK_FLAG_TRICKMODE_KEY_UNITS | \
1540 GST_SEEK_FLAG_KEY_UNIT))
1541 #define REMOVE_SNAP_FLAGS(f) (f & ~(GST_SEEK_FLAG_SNAP_BEFORE | \
1542 GST_SEEK_FLAG_SNAP_AFTER | \
1543 GST_SEEK_FLAG_SNAP_NEAREST))
1546 gst_adaptive_demux_handle_seek_event (GstAdaptiveDemux * demux, GstPad * pad,
1549 GstAdaptiveDemuxClass *demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1553 GstSeekType start_type, stop_type;
1558 GstSegment oldsegment;
1559 GstAdaptiveDemuxStream *stream = NULL;
1561 GST_INFO_OBJECT (demux, "Received seek event");
1563 GST_API_LOCK (demux);
1564 GST_MANIFEST_LOCK (demux);
1566 gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
1569 if (format != GST_FORMAT_TIME) {
1570 GST_MANIFEST_UNLOCK (demux);
1571 GST_API_UNLOCK (demux);
1572 GST_WARNING_OBJECT (demux,
1573 "Adaptive demuxers only support TIME-based seeking");
1574 gst_event_unref (event);
1578 if (flags & GST_SEEK_FLAG_SEGMENT) {
1579 GST_FIXME_OBJECT (demux, "Handle segment seeks");
1580 GST_MANIFEST_UNLOCK (demux);
1581 GST_API_UNLOCK (demux);
1582 gst_event_unref (event);
1586 seqnum = gst_event_get_seqnum (event);
1588 if (flags & GST_SEEK_FLAG_INSTANT_RATE_CHANGE) {
1589 /* For instant rate seeks, reply directly and update
1590 * our segment so the new rate is reflected in any future
1594 /* instant rate change only supported if direction does not change. All
1595 * other requirements are already checked before creating the seek event
1596 * but let's double-check here to be sure */
1597 if ((demux->segment.rate > 0 && rate < 0) ||
1598 (demux->segment.rate < 0 && rate > 0) ||
1599 start_type != GST_SEEK_TYPE_NONE ||
1600 stop_type != GST_SEEK_TYPE_NONE || (flags & GST_SEEK_FLAG_FLUSH)) {
1601 GST_ERROR_OBJECT (demux,
1602 "Instant rate change seeks only supported in the "
1603 "same direction, without flushing and position change");
1604 GST_MANIFEST_UNLOCK (demux);
1605 GST_API_UNLOCK (demux);
1609 ev = gst_event_new_instant_rate_change (rate / demux->segment.rate,
1610 (GstSegmentFlags) flags);
1611 gst_event_set_seqnum (ev, seqnum);
1613 GST_MANIFEST_UNLOCK (demux);
1615 ret = gst_adaptive_demux_push_src_event (demux, ev);
1617 GST_API_UNLOCK (demux);
1618 gst_event_unref (event);
1623 if (!gst_adaptive_demux_can_seek (demux)) {
1624 GST_MANIFEST_UNLOCK (demux);
1625 GST_API_UNLOCK (demux);
1626 gst_event_unref (event);
1630 if (gst_adaptive_demux_is_live (demux)) {
1631 gint64 range_start, range_stop;
1632 gboolean changed = FALSE;
1633 gboolean start_valid = TRUE, stop_valid = TRUE;
1635 if (!gst_adaptive_demux_get_live_seek_range (demux, &range_start,
1637 GST_MANIFEST_UNLOCK (demux);
1638 GST_API_UNLOCK (demux);
1639 gst_event_unref (event);
1640 GST_WARNING_OBJECT (demux, "Failure getting the live seek ranges");
1644 GST_DEBUG_OBJECT (demux,
1645 "Live range is %" GST_STIME_FORMAT " %" GST_STIME_FORMAT,
1646 GST_STIME_ARGS (range_start), GST_STIME_ARGS (range_stop));
1648 /* Handle relative positioning for live streams (relative to the range_stop) */
1649 if (start_type == GST_SEEK_TYPE_END) {
1650 start = range_stop + start;
1651 start_type = GST_SEEK_TYPE_SET;
1654 if (stop_type == GST_SEEK_TYPE_END) {
1655 stop = range_stop + stop;
1656 stop_type = GST_SEEK_TYPE_SET;
1660 /* Adjust the requested start/stop position if it falls beyond the live
1662 * The only case where we don't adjust is for the starting point of
1663 * an accurate seek (start if forward and stop if backwards)
1665 if (start_type == GST_SEEK_TYPE_SET && start < range_start &&
1666 (rate < 0 || !(flags & GST_SEEK_FLAG_ACCURATE))) {
1667 GST_DEBUG_OBJECT (demux,
1668 "seek before live stream start, setting to range start: %"
1669 GST_TIME_FORMAT, GST_TIME_ARGS (range_start));
1670 start = range_start;
1673 /* truncate stop position also if set */
1674 if (stop_type == GST_SEEK_TYPE_SET && stop > range_stop &&
1675 (rate > 0 || !(flags & GST_SEEK_FLAG_ACCURATE))) {
1676 GST_DEBUG_OBJECT (demux,
1677 "seek ending after live start, adjusting to: %"
1678 GST_TIME_FORMAT, GST_TIME_ARGS (range_stop));
1683 if (start_type == GST_SEEK_TYPE_SET && GST_CLOCK_TIME_IS_VALID (start) &&
1684 (start < range_start || start > range_stop)) {
1685 GST_WARNING_OBJECT (demux,
1686 "Seek to invalid position start:%" GST_STIME_FORMAT
1687 " out of seekable range (%" GST_STIME_FORMAT " - %" GST_STIME_FORMAT
1688 ")", GST_STIME_ARGS (start), GST_STIME_ARGS (range_start),
1689 GST_STIME_ARGS (range_stop));
1690 start_valid = FALSE;
1692 if (stop_type == GST_SEEK_TYPE_SET && GST_CLOCK_TIME_IS_VALID (stop) &&
1693 (stop < range_start || stop > range_stop)) {
1694 GST_WARNING_OBJECT (demux,
1695 "Seek to invalid position stop:%" GST_STIME_FORMAT
1696 " out of seekable range (%" GST_STIME_FORMAT " - %" GST_STIME_FORMAT
1697 ")", GST_STIME_ARGS (stop), GST_STIME_ARGS (range_start),
1698 GST_STIME_ARGS (range_stop));
1702 /* If the seek position is still outside of the seekable range, refuse the seek */
1703 if (!start_valid || !stop_valid) {
1704 GST_MANIFEST_UNLOCK (demux);
1705 GST_API_UNLOCK (demux);
1706 gst_event_unref (event);
1710 /* Re-create seek event with changed/updated values */
1712 gst_event_unref (event);
1714 gst_event_new_seek (rate, format, flags,
1715 start_type, start, stop_type, stop);
1716 gst_event_set_seqnum (event, seqnum);
1720 GST_DEBUG_OBJECT (demux, "seek event, %" GST_PTR_FORMAT, event);
1722 /* have a backup in case seek fails */
1723 gst_segment_copy_into (&demux->segment, &oldsegment);
1725 if (flags & GST_SEEK_FLAG_FLUSH) {
1728 GST_DEBUG_OBJECT (demux, "sending flush start");
1729 fevent = gst_event_new_flush_start ();
1730 gst_event_set_seqnum (fevent, seqnum);
1731 GST_MANIFEST_UNLOCK (demux);
1732 gst_adaptive_demux_push_src_event (demux, fevent);
1733 GST_MANIFEST_LOCK (demux);
1735 gst_adaptive_demux_stop_tasks (demux, FALSE);
1736 } else if ((rate > 0 && start_type != GST_SEEK_TYPE_NONE) ||
1737 (rate < 0 && stop_type != GST_SEEK_TYPE_NONE)) {
1739 gst_adaptive_demux_stop_tasks (demux, FALSE);
1742 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
1745 * Handle snap seeks as follows:
1746 * 1) do the snap seeking on the stream that received
1748 * 2) use the final position on this stream to seek
1749 * on the other streams to the same position
1751 * We can't snap at all streams at the same time as
1752 * they might end in different positions, so just
1753 * use the one that received the event as the 'leading'
1754 * one to do the snap seek.
1756 if (IS_SNAP_SEEK (flags) && demux_class->stream_seek && (stream =
1757 gst_adaptive_demux_find_stream_for_pad (demux, pad))) {
1759 GstSeekFlags stream_seek_flags = flags;
1761 /* snap-seek on the stream that received the event and then
1762 * use the resulting position to seek on all streams */
1765 if (start_type != GST_SEEK_TYPE_NONE)
1768 ts = stream->segment.position;
1769 start_type = GST_SEEK_TYPE_SET;
1772 if (stop_type != GST_SEEK_TYPE_NONE)
1775 stop_type = GST_SEEK_TYPE_SET;
1776 ts = stream->segment.position;
1781 demux_class->stream_seek (stream, rate >= 0, stream_seek_flags, ts, &ts);
1784 /* replace event with a new one without snapping to seek on all streams */
1785 gst_event_unref (event);
1792 gst_event_new_seek (rate, format, REMOVE_SNAP_FLAGS (flags),
1793 start_type, start, stop_type, stop);
1794 GST_DEBUG_OBJECT (demux, "Adapted snap seek to %" GST_PTR_FORMAT, event);
1798 ret = gst_segment_do_seek (&demux->segment, rate, format, flags, start_type,
1799 start, stop_type, stop, &update);
1802 /* FIXME - this seems unatural, do_seek() is updating base when we
1803 * only want the start/stop position to change, maybe do_seek() needs
1805 if (!(flags & GST_SEEK_FLAG_FLUSH) && ((rate > 0
1806 && start_type == GST_SEEK_TYPE_NONE) || (rate < 0
1807 && stop_type == GST_SEEK_TYPE_NONE))) {
1808 demux->segment.base = oldsegment.base;
1811 GST_DEBUG_OBJECT (demux, "Calling subclass seek: %" GST_PTR_FORMAT, event);
1813 ret = demux_class->seek (demux, event);
1817 /* Is there anything else we can do if it fails? */
1818 gst_segment_copy_into (&oldsegment, &demux->segment);
1820 demux->priv->segment_seqnum = seqnum;
1822 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
1824 if (flags & GST_SEEK_FLAG_FLUSH) {
1827 GST_DEBUG_OBJECT (demux, "Sending flush stop on all pad");
1828 fevent = gst_event_new_flush_stop (TRUE);
1829 gst_event_set_seqnum (fevent, seqnum);
1830 gst_adaptive_demux_push_src_event (demux, fevent);
1833 if (demux->next_streams) {
1834 /* If the seek generated new streams, get them
1836 gst_adaptive_demux_prepare_streams (demux, FALSE);
1837 gst_adaptive_demux_start_tasks (demux, TRUE);
1839 GstClockTime period_start =
1840 gst_adaptive_demux_get_period_start_time (demux);
1842 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
1843 gst_adaptive_demux_update_streams_segment (demux, demux->streams,
1844 period_start, start_type, stop_type);
1845 gst_adaptive_demux_update_streams_segment (demux, demux->prepared_streams,
1846 period_start, start_type, stop_type);
1847 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
1849 /* Restart the demux */
1850 gst_adaptive_demux_start_tasks (demux, FALSE);
1853 GST_MANIFEST_UNLOCK (demux);
1854 GST_API_UNLOCK (demux);
1855 gst_event_unref (event);
1861 gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent,
1864 GstAdaptiveDemux *demux;
1866 demux = GST_ADAPTIVE_DEMUX_CAST (parent);
1868 /* FIXME handle events received on pads that are to be removed */
1870 switch (event->type) {
1871 case GST_EVENT_SEEK:
1873 guint32 seqnum = gst_event_get_seqnum (event);
1874 if (seqnum == demux->priv->segment_seqnum) {
1875 GST_LOG_OBJECT (pad,
1876 "Drop duplicated SEEK event seqnum %" G_GUINT32_FORMAT, seqnum);
1877 gst_event_unref (event);
1880 return gst_adaptive_demux_handle_seek_event (demux, pad, event);
1882 case GST_EVENT_RECONFIGURE:{
1883 GstAdaptiveDemuxStream *stream;
1885 GST_MANIFEST_LOCK (demux);
1886 stream = gst_adaptive_demux_find_stream_for_pad (demux, pad);
1889 if (!stream->cancelled && gst_adaptive_demux_is_running (demux) &&
1890 stream->last_ret == GST_FLOW_NOT_LINKED) {
1891 stream->last_ret = GST_FLOW_OK;
1892 stream->restart_download = TRUE;
1893 stream->need_header = TRUE;
1894 stream->discont = TRUE;
1895 GST_DEBUG_OBJECT (stream->pad, "Restarting download loop");
1896 gst_task_start (stream->download_task);
1898 gst_event_unref (event);
1899 GST_MANIFEST_UNLOCK (demux);
1902 GST_MANIFEST_UNLOCK (demux);
1905 case GST_EVENT_LATENCY:{
1906 /* Upstream and our internal source are irrelevant
1907 * for latency, and we should not fail here to
1908 * configure the latency */
1909 gst_event_unref (event);
1912 case GST_EVENT_QOS:{
1913 GstClockTimeDiff diff;
1914 GstClockTime timestamp;
1915 GstClockTime earliest_time;
1917 gst_event_parse_qos (event, NULL, NULL, &diff, ×tamp);
1918 /* Only take into account lateness if late */
1920 earliest_time = timestamp + 2 * diff;
1922 earliest_time = timestamp;
1924 GST_OBJECT_LOCK (demux);
1925 if (!GST_CLOCK_TIME_IS_VALID (demux->priv->qos_earliest_time) ||
1926 earliest_time > demux->priv->qos_earliest_time) {
1927 demux->priv->qos_earliest_time = earliest_time;
1928 GST_DEBUG_OBJECT (demux, "qos_earliest_time %" GST_TIME_FORMAT,
1929 GST_TIME_ARGS (demux->priv->qos_earliest_time));
1931 GST_OBJECT_UNLOCK (demux);
1938 return gst_pad_event_default (pad, parent, event);
1942 gst_adaptive_demux_src_query (GstPad * pad, GstObject * parent,
1945 GstAdaptiveDemux *demux;
1946 GstAdaptiveDemuxClass *demux_class;
1947 gboolean ret = FALSE;
1952 demux = GST_ADAPTIVE_DEMUX_CAST (parent);
1953 demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
1955 switch (query->type) {
1956 case GST_QUERY_DURATION:{
1957 GstClockTime duration = -1;
1960 gst_query_parse_duration (query, &fmt, NULL);
1962 if (gst_adaptive_demux_is_live (demux)) {
1963 /* We are able to answer this query: the duration is unknown */
1964 gst_query_set_duration (query, fmt, -1);
1969 if (fmt == GST_FORMAT_TIME
1970 && g_atomic_int_get (&demux->priv->have_manifest)) {
1971 duration = demux_class->get_duration (demux);
1973 if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) {
1974 gst_query_set_duration (query, GST_FORMAT_TIME, duration);
1979 GST_LOG_OBJECT (demux, "GST_QUERY_DURATION returns %s with duration %"
1980 GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
1983 case GST_QUERY_LATENCY:{
1984 gst_query_set_latency (query, FALSE, 0, -1);
1988 case GST_QUERY_SEEKING:{
1993 if (!g_atomic_int_get (&demux->priv->have_manifest)) {
1994 GST_INFO_OBJECT (demux,
1995 "Don't have manifest yet, can't answer seeking query");
1996 return FALSE; /* can't answer without manifest */
1999 GST_MANIFEST_LOCK (demux);
2001 gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL);
2002 GST_INFO_OBJECT (demux, "Received GST_QUERY_SEEKING with format %d", fmt);
2003 if (fmt == GST_FORMAT_TIME) {
2004 GstClockTime duration;
2005 gboolean can_seek = gst_adaptive_demux_can_seek (demux);
2009 if (gst_adaptive_demux_is_live (demux)) {
2010 ret = gst_adaptive_demux_get_live_seek_range (demux, &start, &stop);
2012 GST_MANIFEST_UNLOCK (demux);
2013 GST_INFO_OBJECT (demux, "can't answer seeking query");
2017 duration = demux_class->get_duration (demux);
2018 if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0)
2022 gst_query_set_seeking (query, fmt, can_seek, start, stop);
2023 GST_INFO_OBJECT (demux, "GST_QUERY_SEEKING returning with start : %"
2024 GST_TIME_FORMAT ", stop : %" GST_TIME_FORMAT,
2025 GST_TIME_ARGS (start), GST_TIME_ARGS (stop));
2027 GST_MANIFEST_UNLOCK (demux);
2032 GST_MANIFEST_LOCK (demux);
2034 /* TODO HLS can answer this differently it seems */
2035 if (demux->manifest_uri) {
2036 /* FIXME: (hls) Do we answer with the variant playlist, with the current
2037 * playlist or the the uri of the last downlowaded fragment? */
2038 gst_query_set_uri (query, demux->manifest_uri);
2042 GST_MANIFEST_UNLOCK (demux);
2045 /* Don't forward queries upstream because of the special nature of this
2046 * "demuxer", which relies on the upstream element only to be fed
2055 /* must be called with manifest_lock taken */
2057 gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux,
2058 gboolean start_preroll_streams)
2062 if (!gst_adaptive_demux_is_running (demux)) {
2063 GST_DEBUG_OBJECT (demux, "Not starting tasks due to shutdown");
2067 GST_INFO_OBJECT (demux, "Starting streams' tasks");
2069 iter = start_preroll_streams ? demux->prepared_streams : demux->streams;
2071 for (; iter; iter = g_list_next (iter)) {
2072 GstAdaptiveDemuxStream *stream = iter->data;
2074 if (!start_preroll_streams) {
2075 g_mutex_lock (&stream->fragment_download_lock);
2076 stream->cancelled = FALSE;
2077 stream->replaced = FALSE;
2078 g_mutex_unlock (&stream->fragment_download_lock);
2081 stream->last_ret = GST_FLOW_OK;
2082 gst_task_start (stream->download_task);
2086 /* must be called with manifest_lock taken */
2088 gst_adaptive_demux_stop_manifest_update_task (GstAdaptiveDemux * demux)
2090 gst_uri_downloader_cancel (demux->downloader);
2092 gst_task_stop (demux->priv->updates_task);
2094 g_mutex_lock (&demux->priv->updates_timed_lock);
2095 GST_DEBUG_OBJECT (demux, "requesting stop of the manifest update task");
2096 demux->priv->stop_updates_task = TRUE;
2097 g_cond_signal (&demux->priv->updates_timed_cond);
2098 g_mutex_unlock (&demux->priv->updates_timed_lock);
2101 /* must be called with manifest_lock taken */
2103 gst_adaptive_demux_start_manifest_update_task (GstAdaptiveDemux * demux)
2105 GstAdaptiveDemuxClass *demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
2107 if (gst_adaptive_demux_is_live (demux)) {
2108 gst_uri_downloader_reset (demux->downloader);
2109 g_mutex_lock (&demux->priv->updates_timed_lock);
2110 demux->priv->stop_updates_task = FALSE;
2111 g_mutex_unlock (&demux->priv->updates_timed_lock);
2112 /* Task to periodically update the manifest */
2113 if (demux_class->requires_periodical_playlist_update (demux)) {
2114 GST_DEBUG_OBJECT (demux, "requesting start of the manifest update task");
2115 gst_task_start (demux->priv->updates_task);
2120 /* must be called with manifest_lock taken
2121 * This function will temporarily release manifest_lock in order to join the
2123 * The api_lock will still protect it against other threads trying to modify
2124 * the demux element.
2127 gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux, gboolean stop_updates)
2131 GList *list_to_process;
2133 GST_LOG_OBJECT (demux, "Stopping tasks");
2136 gst_adaptive_demux_stop_manifest_update_task (demux);
2138 list_to_process = demux->streams;
2139 for (i = 0; i < 2; ++i) {
2140 for (iter = list_to_process; iter; iter = g_list_next (iter)) {
2141 GstAdaptiveDemuxStream *stream = iter->data;
2143 g_mutex_lock (&stream->fragment_download_lock);
2144 stream->cancelled = TRUE;
2145 gst_task_stop (stream->download_task);
2146 g_cond_signal (&stream->fragment_download_cond);
2147 g_mutex_unlock (&stream->fragment_download_lock);
2149 list_to_process = demux->prepared_streams;
2152 GST_MANIFEST_UNLOCK (demux);
2153 g_mutex_lock (&demux->priv->preroll_lock);
2154 g_cond_broadcast (&demux->priv->preroll_cond);
2155 g_mutex_unlock (&demux->priv->preroll_lock);
2156 GST_MANIFEST_LOCK (demux);
2158 g_mutex_lock (&demux->priv->manifest_update_lock);
2159 g_cond_broadcast (&demux->priv->manifest_cond);
2160 g_mutex_unlock (&demux->priv->manifest_update_lock);
2162 /* need to release manifest_lock before stopping the src element.
2163 * The streams were asked to cancel, so they will not make any writes to demux
2164 * object. Even if we temporarily release manifest_lock, the demux->streams
2165 * cannot change and iter cannot be invalidated.
2167 list_to_process = demux->streams;
2168 for (i = 0; i < 2; ++i) {
2169 for (iter = list_to_process; iter; iter = g_list_next (iter)) {
2170 GstAdaptiveDemuxStream *stream = iter->data;
2171 GstElement *src = stream->src;
2173 GST_MANIFEST_UNLOCK (demux);
2176 gst_element_set_locked_state (src, TRUE);
2177 gst_element_set_state (src, GST_STATE_READY);
2180 /* stream->download_task value never changes, so it is safe to read it
2181 * outside critical section
2183 gst_task_join (stream->download_task);
2185 GST_MANIFEST_LOCK (demux);
2187 list_to_process = demux->prepared_streams;
2190 GST_MANIFEST_UNLOCK (demux);
2192 gst_task_join (demux->priv->updates_task);
2194 GST_MANIFEST_LOCK (demux);
2196 list_to_process = demux->streams;
2197 for (i = 0; i < 2; ++i) {
2198 for (iter = list_to_process; iter; iter = g_list_next (iter)) {
2199 GstAdaptiveDemuxStream *stream = iter->data;
2201 stream->download_error_count = 0;
2202 stream->need_header = TRUE;
2204 list_to_process = demux->prepared_streams;
2206 demux->priv->qos_earliest_time = GST_CLOCK_TIME_NONE;
2209 /* must be called with manifest_lock taken */
2211 gst_adaptive_demux_push_src_event (GstAdaptiveDemux * demux, GstEvent * event)
2214 gboolean ret = TRUE;
2216 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
2217 GstAdaptiveDemuxStream *stream = iter->data;
2218 gst_event_ref (event);
2219 ret = ret & gst_pad_push_event (stream->pad, event);
2221 gst_event_unref (event);
2225 /* must be called with manifest_lock taken */
2227 gst_adaptive_demux_stream_set_caps (GstAdaptiveDemuxStream * stream,
2230 GST_DEBUG_OBJECT (stream->pad, "setting new caps for stream %" GST_PTR_FORMAT,
2232 gst_caps_replace (&stream->pending_caps, caps);
2233 gst_caps_unref (caps);
2236 /* must be called with manifest_lock taken */
2238 gst_adaptive_demux_stream_set_tags (GstAdaptiveDemuxStream * stream,
2241 GST_DEBUG_OBJECT (stream->pad, "setting new tags for stream %" GST_PTR_FORMAT,
2243 if (stream->pending_tags) {
2244 gst_tag_list_unref (stream->pending_tags);
2246 stream->pending_tags = tags;
2249 /* must be called with manifest_lock taken */
2251 gst_adaptive_demux_stream_queue_event (GstAdaptiveDemuxStream * stream,
2254 stream->pending_events = g_list_append (stream->pending_events, event);
2257 /* must be called with manifest_lock taken */
2259 _update_average_bitrate (GstAdaptiveDemux * demux,
2260 GstAdaptiveDemuxStream * stream, guint64 new_bitrate)
2262 gint index = stream->moving_index % NUM_LOOKBACK_FRAGMENTS;
2264 stream->moving_bitrate -= stream->fragment_bitrates[index];
2265 stream->fragment_bitrates[index] = new_bitrate;
2266 stream->moving_bitrate += new_bitrate;
2268 stream->moving_index += 1;
2270 if (stream->moving_index > NUM_LOOKBACK_FRAGMENTS)
2271 return stream->moving_bitrate / NUM_LOOKBACK_FRAGMENTS;
2272 return stream->moving_bitrate / stream->moving_index;
2275 /* must be called with manifest_lock taken */
2277 gst_adaptive_demux_stream_update_current_bitrate (GstAdaptiveDemux * demux,
2278 GstAdaptiveDemuxStream * stream)
2280 guint64 average_bitrate;
2281 guint64 fragment_bitrate;
2283 if (demux->connection_speed) {
2284 GST_LOG_OBJECT (demux, "Connection-speed is set to %u kbps, using it",
2285 demux->connection_speed / 1000);
2286 stream->current_download_rate = demux->connection_speed;
2287 return demux->connection_speed;
2290 fragment_bitrate = stream->last_bitrate;
2291 GST_DEBUG_OBJECT (demux, "Download bitrate is : %" G_GUINT64_FORMAT " bps",
2294 average_bitrate = _update_average_bitrate (demux, stream, fragment_bitrate);
2296 GST_INFO_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
2297 "last fragment bitrate was %" G_GUINT64_FORMAT, fragment_bitrate);
2298 GST_INFO_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
2299 "Last %u fragments average bitrate is %" G_GUINT64_FORMAT,
2300 NUM_LOOKBACK_FRAGMENTS, average_bitrate);
2302 /* Conservative approach, make sure we don't upgrade too fast */
2303 stream->current_download_rate = MIN (average_bitrate, fragment_bitrate);
2305 stream->current_download_rate *= demux->bitrate_limit;
2306 GST_DEBUG_OBJECT (demux, "Bitrate after bitrate limit (%0.2f): %"
2307 G_GUINT64_FORMAT, demux->bitrate_limit, stream->current_download_rate);
2310 /* Debugging code, modulate the bitrate every few fragments */
2312 static guint ctr = 0;
2314 GST_INFO_OBJECT (demux, "Halving reported bitrate for debugging");
2315 stream->current_download_rate /= 2;
2321 return stream->current_download_rate;
2324 /* must be called with manifest_lock taken */
2325 static GstFlowReturn
2326 gst_adaptive_demux_combine_flows (GstAdaptiveDemux * demux)
2328 gboolean all_notlinked = TRUE;
2329 gboolean all_eos = TRUE;
2332 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
2333 GstAdaptiveDemuxStream *stream = iter->data;
2335 if (stream->last_ret != GST_FLOW_NOT_LINKED) {
2336 all_notlinked = FALSE;
2337 if (stream->last_ret != GST_FLOW_EOS)
2341 if (stream->last_ret <= GST_FLOW_NOT_NEGOTIATED
2342 || stream->last_ret == GST_FLOW_FLUSHING) {
2343 return stream->last_ret;
2347 return GST_FLOW_NOT_LINKED;
2349 return GST_FLOW_EOS;
2353 /* Called with preroll_lock */
2355 gst_adaptive_demux_handle_preroll (GstAdaptiveDemux * demux,
2356 GstAdaptiveDemuxStream * stream)
2358 demux->priv->preroll_pending--;
2359 if (demux->priv->preroll_pending == 0) {
2360 /* That was the last one, time to release all streams
2361 * and expose them */
2362 GST_DEBUG_OBJECT (demux, "All streams prerolled. exposing");
2363 gst_adaptive_demux_expose_streams (demux);
2364 g_cond_broadcast (&demux->priv->preroll_cond);
2368 /* must be called with manifest_lock taken.
2369 * Temporarily releases manifest_lock
2372 gst_adaptive_demux_stream_push_buffer (GstAdaptiveDemuxStream * stream,
2375 GstAdaptiveDemux *demux = stream->demux;
2376 GstFlowReturn ret = GST_FLOW_OK;
2377 gboolean discont = FALSE;
2378 /* Pending events */
2379 GstEvent *pending_caps = NULL, *pending_segment = NULL, *pending_tags = NULL;
2380 GList *pending_events = NULL;
2383 * This is duplicating *exactly* the same thing as what is done at the beginning
2384 * of _src_chain if starting_fragment is TRUE */
2385 if (stream->first_fragment_buffer) {
2386 GstClockTime offset =
2387 gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
2388 GstClockTime period_start =
2389 gst_adaptive_demux_get_period_start_time (demux);
2391 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
2392 if (demux->segment.rate < 0)
2393 /* Set DISCONT flag for every first buffer in reverse playback mode
2394 * as each fragment for its own has to be reversed */
2397 GST_BUFFER_PTS (buffer) = stream->fragment.timestamp;
2398 if (GST_BUFFER_PTS_IS_VALID (buffer))
2399 GST_BUFFER_PTS (buffer) += offset;
2401 if (GST_BUFFER_PTS_IS_VALID (buffer)) {
2402 stream->segment.position = GST_BUFFER_PTS (buffer);
2404 /* Convert from position inside the stream's segment to the demuxer's
2405 * segment, they are not necessarily the same */
2406 if (stream->segment.position - offset + period_start >
2407 demux->segment.position)
2408 demux->segment.position =
2409 stream->segment.position - offset + period_start;
2411 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
2413 GST_LOG_OBJECT (stream->pad,
2414 "Going to push buffer with PTS %" GST_TIME_FORMAT,
2415 GST_TIME_ARGS (GST_BUFFER_PTS (buffer)));
2417 GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE;
2420 if (stream->discont) {
2422 stream->discont = FALSE;
2426 GST_DEBUG_OBJECT (stream->pad, "Marking fragment as discontinuous");
2427 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
2429 GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
2432 stream->first_fragment_buffer = FALSE;
2434 GST_BUFFER_DURATION (buffer) = GST_CLOCK_TIME_NONE;
2435 GST_BUFFER_DTS (buffer) = GST_CLOCK_TIME_NONE;
2436 if (G_UNLIKELY (stream->pending_caps)) {
2437 pending_caps = gst_event_new_caps (stream->pending_caps);
2438 gst_caps_unref (stream->pending_caps);
2439 stream->pending_caps = NULL;
2442 if (stream->do_block) {
2444 g_mutex_lock (&demux->priv->preroll_lock);
2446 /* If we are preroll state, set caps in here */
2448 gst_pad_push_event (stream->pad, pending_caps);
2449 pending_caps = NULL;
2452 gst_adaptive_demux_handle_preroll (demux, stream);
2453 GST_MANIFEST_UNLOCK (demux);
2455 while (stream->do_block && !stream->cancelled) {
2456 GST_LOG_OBJECT (demux, "Stream %p sleeping for preroll", stream);
2457 g_cond_wait (&demux->priv->preroll_cond, &demux->priv->preroll_lock);
2459 if (stream->cancelled) {
2460 GST_LOG_OBJECT (demux, "stream %p cancelled", stream);
2461 gst_buffer_unref (buffer);
2462 g_mutex_unlock (&demux->priv->preroll_lock);
2463 return GST_FLOW_FLUSHING;
2466 g_mutex_unlock (&demux->priv->preroll_lock);
2467 GST_MANIFEST_LOCK (demux);
2470 if (G_UNLIKELY (stream->pending_segment)) {
2471 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
2472 pending_segment = stream->pending_segment;
2473 stream->pending_segment = NULL;
2474 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
2476 if (G_UNLIKELY (stream->pending_tags || stream->bitrate_changed)) {
2477 GstTagList *tags = stream->pending_tags;
2479 stream->pending_tags = NULL;
2480 stream->bitrate_changed = 0;
2482 if (stream->fragment.bitrate != 0) {
2484 tags = gst_tag_list_make_writable (tags);
2486 tags = gst_tag_list_new_empty ();
2488 gst_tag_list_add (tags, GST_TAG_MERGE_KEEP,
2489 GST_TAG_NOMINAL_BITRATE, stream->fragment.bitrate, NULL);
2492 pending_tags = gst_event_new_tag (tags);
2494 if (G_UNLIKELY (stream->pending_events)) {
2495 pending_events = stream->pending_events;
2496 stream->pending_events = NULL;
2499 GST_MANIFEST_UNLOCK (demux);
2501 /* Do not push events or buffers holding the manifest lock */
2502 if (G_UNLIKELY (pending_caps)) {
2503 GST_DEBUG_OBJECT (stream->pad, "Setting pending caps: %" GST_PTR_FORMAT,
2505 gst_pad_push_event (stream->pad, pending_caps);
2507 if (G_UNLIKELY (pending_segment)) {
2508 GST_DEBUG_OBJECT (stream->pad, "Sending pending seg: %" GST_PTR_FORMAT,
2510 gst_pad_push_event (stream->pad, pending_segment);
2512 if (G_UNLIKELY (pending_tags)) {
2513 GST_DEBUG_OBJECT (stream->pad, "Sending pending tags: %" GST_PTR_FORMAT,
2515 gst_pad_push_event (stream->pad, pending_tags);
2517 while (pending_events != NULL) {
2518 GstEvent *event = pending_events->data;
2520 if (!gst_pad_push_event (stream->pad, event))
2521 GST_ERROR_OBJECT (stream->pad, "Failed to send pending event");
2523 pending_events = g_list_delete_link (pending_events, pending_events);
2526 /* Wait for preroll if blocking */
2527 GST_DEBUG_OBJECT (stream->pad,
2528 "About to push buffer of size %" G_GSIZE_FORMAT,
2529 gst_buffer_get_size (buffer));
2531 ret = gst_pad_push (stream->pad, buffer);
2533 GST_MANIFEST_LOCK (demux);
2535 g_mutex_lock (&stream->fragment_download_lock);
2536 if (G_UNLIKELY (stream->cancelled)) {
2537 GST_LOG_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
2538 "Stream was cancelled");
2539 ret = stream->last_ret = GST_FLOW_FLUSHING;
2540 g_mutex_unlock (&stream->fragment_download_lock);
2543 g_mutex_unlock (&stream->fragment_download_lock);
2545 GST_LOG_OBJECT (stream->pad, "Push result: %d %s", ret,
2546 gst_flow_get_name (ret));
2551 /* must be called with manifest_lock taken */
2552 static GstFlowReturn
2553 gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux,
2554 GstAdaptiveDemuxStream * stream)
2556 /* No need to advance, this isn't a real fragment */
2557 if (G_UNLIKELY (stream->downloading_header || stream->downloading_index))
2560 return gst_adaptive_demux_stream_advance_fragment (demux, stream,
2561 stream->fragment.duration);
2564 /* must be called with manifest_lock taken.
2565 * Can temporarily release manifest_lock
2567 static GstFlowReturn
2568 gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux,
2569 GstAdaptiveDemuxStream * stream, GstBuffer * buffer)
2571 return gst_adaptive_demux_stream_push_buffer (stream, buffer);
2575 gst_adaptive_demux_requires_periodical_playlist_update_default (GstAdaptiveDemux
2581 static GstFlowReturn
2582 _src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
2584 GstAdaptiveDemuxStream *stream;
2585 GstAdaptiveDemux *demux;
2586 GstAdaptiveDemuxClass *klass;
2587 GstFlowReturn ret = GST_FLOW_OK;
2589 demux = GST_ADAPTIVE_DEMUX_CAST (parent);
2590 stream = gst_pad_get_element_private (pad);
2591 klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
2593 GST_MANIFEST_LOCK (demux);
2595 /* do not make any changes if the stream is cancelled */
2596 g_mutex_lock (&stream->fragment_download_lock);
2597 if (G_UNLIKELY (stream->cancelled)) {
2598 g_mutex_unlock (&stream->fragment_download_lock);
2599 gst_buffer_unref (buffer);
2600 ret = stream->last_ret = GST_FLOW_FLUSHING;
2601 GST_MANIFEST_UNLOCK (demux);
2604 g_mutex_unlock (&stream->fragment_download_lock);
2606 /* starting_fragment is set to TRUE at the beginning of
2607 * _stream_download_fragment()
2608 * /!\ If there is a header/index being downloaded, then this will
2609 * be TRUE for the first one ... but FALSE for the remaining ones,
2610 * including the *actual* fragment ! */
2611 if (stream->starting_fragment) {
2612 GstClockTime offset =
2613 gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
2614 GstClockTime period_start =
2615 gst_adaptive_demux_get_period_start_time (demux);
2617 stream->starting_fragment = FALSE;
2618 if (klass->start_fragment) {
2619 if (!klass->start_fragment (demux, stream)) {
2620 ret = GST_FLOW_ERROR;
2625 GST_BUFFER_PTS (buffer) = stream->fragment.timestamp;
2626 if (GST_BUFFER_PTS_IS_VALID (buffer))
2627 GST_BUFFER_PTS (buffer) += offset;
2629 GST_LOG_OBJECT (stream->pad, "set fragment pts=%" GST_TIME_FORMAT,
2630 GST_TIME_ARGS (GST_BUFFER_PTS (buffer)));
2632 if (GST_BUFFER_PTS_IS_VALID (buffer)) {
2633 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
2634 stream->segment.position = GST_BUFFER_PTS (buffer);
2636 /* Convert from position inside the stream's segment to the demuxer's
2637 * segment, they are not necessarily the same */
2638 if (stream->segment.position - offset + period_start >
2639 demux->segment.position)
2640 demux->segment.position =
2641 stream->segment.position - offset + period_start;
2642 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
2646 GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE;
2649 /* downloading_first_buffer is set to TRUE in download_uri() just before
2650 * activating the source (i.e. requesting a given URI)
2652 * The difference with starting_fragment is that this will be called
2653 * for *all* first buffers (of index, and header, and fragment)
2655 * ... to then only do something useful (in this block) for actual
2657 if (stream->downloading_first_buffer) {
2658 gint64 chunk_size = 0;
2660 stream->downloading_first_buffer = FALSE;
2662 if (!stream->downloading_header && !stream->downloading_index) {
2663 /* If this is the first buffer of a fragment (not the headers or index)
2664 * and we don't have a birate from the sub-class, then see if we
2665 * can work it out from the fragment size and duration */
2666 if (stream->fragment.bitrate == 0 &&
2667 stream->fragment.duration != 0 &&
2668 gst_element_query_duration (stream->uri_handler, GST_FORMAT_BYTES,
2669 &chunk_size) && chunk_size != -1) {
2670 guint bitrate = MIN (G_MAXUINT, gst_util_uint64_scale (chunk_size,
2671 8 * GST_SECOND, stream->fragment.duration));
2672 GST_LOG_OBJECT (demux,
2673 "Fragment has size %" G_GINT64_FORMAT " duration %" GST_TIME_FORMAT
2674 " = bitrate %u", chunk_size,
2675 GST_TIME_ARGS (stream->fragment.duration), bitrate);
2676 stream->fragment.bitrate = bitrate;
2678 if (stream->fragment.bitrate) {
2679 stream->bitrate_changed = TRUE;
2681 GST_WARNING_OBJECT (demux, "Bitrate for fragment not available");
2686 stream->download_total_bytes += gst_buffer_get_size (buffer);
2688 GST_TRACE_OBJECT (stream->pad, "Received buffer of size %" G_GSIZE_FORMAT,
2689 gst_buffer_get_size (buffer));
2691 ret = klass->data_received (demux, stream, buffer);
2693 if (ret == GST_FLOW_FLUSHING) {
2694 /* do not make any changes if the stream is cancelled */
2695 g_mutex_lock (&stream->fragment_download_lock);
2696 if (G_UNLIKELY (stream->cancelled)) {
2697 g_mutex_unlock (&stream->fragment_download_lock);
2698 GST_MANIFEST_UNLOCK (demux);
2701 g_mutex_unlock (&stream->fragment_download_lock);
2704 if (ret != GST_FLOW_OK) {
2705 gboolean finished = FALSE;
2707 if (ret < GST_FLOW_EOS) {
2708 GST_ELEMENT_FLOW_ERROR (demux, ret);
2710 /* TODO push this on all pads */
2711 gst_pad_push_event (stream->pad, gst_event_new_eos ());
2713 GST_DEBUG_OBJECT (stream->pad, "stream stopped, reason %s",
2714 gst_flow_get_name (ret));
2717 if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) {
2718 ret = GST_FLOW_EOS; /* return EOS to make the source stop */
2719 } else if (ret == GST_ADAPTIVE_DEMUX_FLOW_END_OF_FRAGMENT) {
2720 /* Behaves like an EOS event from upstream */
2721 stream->fragment.finished = TRUE;
2722 ret = klass->finish_fragment (demux, stream);
2723 if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) {
2724 ret = GST_FLOW_EOS; /* return EOS to make the source stop */
2725 } else if (ret != GST_FLOW_OK) {
2731 gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL);
2738 GST_MANIFEST_UNLOCK (demux);
2743 /* must be called with manifest_lock taken */
2745 gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream *
2746 stream, GstFlowReturn ret, GError * err)
2748 GST_DEBUG_OBJECT (stream->pad, "Download finish: %d %s - err: %p", ret,
2749 gst_flow_get_name (ret), err);
2751 /* if we have an error, only replace last_ret if it was OK before to avoid
2752 * overwriting the first error we got */
2753 if (stream->last_ret == GST_FLOW_OK) {
2754 stream->last_ret = ret;
2756 g_clear_error (&stream->last_error);
2757 stream->last_error = g_error_copy (err);
2760 g_mutex_lock (&stream->fragment_download_lock);
2761 stream->download_finished = TRUE;
2762 g_cond_signal (&stream->fragment_download_cond);
2763 g_mutex_unlock (&stream->fragment_download_lock);
2766 static GstFlowReturn
2767 gst_adaptive_demux_eos_handling (GstAdaptiveDemuxStream * stream)
2769 GstFlowReturn ret = GST_FLOW_OK;
2770 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (stream->demux);
2772 if (!klass->need_another_chunk || stream->fragment.chunk_size == -1
2773 || !klass->need_another_chunk (stream)
2774 || stream->fragment.chunk_size == 0) {
2775 stream->fragment.finished = TRUE;
2777 /* Last chance to figure out a fallback nominal bitrate if neither baseclass
2778 nor the HTTP Content-Length implementation worked. */
2779 if (stream->fragment.bitrate == 0 && stream->fragment.duration != 0 &&
2780 stream->fragment_bytes_downloaded != 0 && !stream->downloading_index &&
2781 !stream->downloading_header) {
2782 guint bitrate = MIN (G_MAXUINT,
2783 gst_util_uint64_scale (stream->fragment_bytes_downloaded,
2784 8 * GST_SECOND, stream->fragment.duration));
2785 GST_LOG_OBJECT (stream->pad,
2786 "Fragment has size %" G_GUINT64_FORMAT " duration %" GST_TIME_FORMAT
2787 " = bitrate %u", stream->fragment_bytes_downloaded,
2788 GST_TIME_ARGS (stream->fragment.duration), bitrate);
2789 stream->fragment.bitrate = bitrate;
2790 stream->bitrate_changed = TRUE;
2792 ret = klass->finish_fragment (stream->demux, stream);
2794 gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL);
2800 _src_event (GstPad * pad, GstObject * parent, GstEvent * event)
2802 GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad);
2803 GstAdaptiveDemux *demux = stream->demux;
2805 switch (GST_EVENT_TYPE (event)) {
2806 case GST_EVENT_EOS:{
2807 GST_DEBUG_OBJECT (pad, "Saw EOS on src pad");
2808 GST_MANIFEST_LOCK (demux);
2810 gst_adaptive_demux_eos_handling (stream);
2813 * _eos_handling() calls fragment_download_finish() which does the
2814 * same thing as below.
2815 * Could this cause races ? */
2816 g_mutex_lock (&stream->fragment_download_lock);
2817 stream->download_finished = TRUE;
2818 g_cond_signal (&stream->fragment_download_cond);
2819 g_mutex_unlock (&stream->fragment_download_lock);
2821 GST_MANIFEST_UNLOCK (demux);
2828 gst_event_unref (event);
2834 _src_query (GstPad * pad, GstObject * parent, GstQuery * query)
2836 GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad);
2838 switch (GST_QUERY_TYPE (query)) {
2839 case GST_QUERY_ALLOCATION:
2846 return gst_pad_peer_query (stream->pad, query);
2849 static GstPadProbeReturn
2850 _uri_handler_probe (GstPad * pad, GstPadProbeInfo * info,
2851 GstAdaptiveDemuxStream * stream)
2853 GstPadProbeReturn ret = GST_PAD_PROBE_OK;
2855 if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_BUFFER) {
2856 GstBuffer *buf = GST_PAD_PROBE_INFO_BUFFER (info);
2857 if (stream->fragment_bytes_downloaded == 0) {
2858 stream->last_latency =
2859 gst_adaptive_demux_get_monotonic_time (stream->demux) -
2860 (stream->download_start_time * GST_USECOND);
2861 GST_DEBUG_OBJECT (pad,
2862 "FIRST BYTE since download_start %" GST_TIME_FORMAT,
2863 GST_TIME_ARGS (stream->last_latency));
2865 stream->fragment_bytes_downloaded += gst_buffer_get_size (buf);
2866 GST_LOG_OBJECT (pad,
2867 "Received buffer, size %" G_GSIZE_FORMAT " total %" G_GUINT64_FORMAT,
2868 gst_buffer_get_size (buf), stream->fragment_bytes_downloaded);
2869 } else if (GST_PAD_PROBE_INFO_TYPE (info) &
2870 GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
2871 GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info);
2872 GST_LOG_OBJECT (pad, "Received event %s %" GST_PTR_FORMAT,
2873 GST_EVENT_TYPE_NAME (ev), ev);
2874 switch (GST_EVENT_TYPE (ev)) {
2875 case GST_EVENT_SEGMENT:
2876 stream->fragment_bytes_downloaded = 0;
2880 stream->last_download_time =
2881 gst_adaptive_demux_get_monotonic_time (stream->demux) -
2882 (stream->download_start_time * GST_USECOND);
2883 stream->last_bitrate =
2884 gst_util_uint64_scale (stream->fragment_bytes_downloaded,
2885 8 * GST_SECOND, stream->last_download_time);
2886 GST_DEBUG_OBJECT (pad,
2887 "EOS since download_start %" GST_TIME_FORMAT " bitrate %"
2888 G_GUINT64_FORMAT " bps", GST_TIME_ARGS (stream->last_download_time),
2889 stream->last_bitrate);
2890 /* Calculate bitrate since URI request */
2901 /* must be called with manifest_lock taken.
2902 * Can temporarily release manifest_lock
2905 gst_adaptive_demux_stream_wait_manifest_update (GstAdaptiveDemux * demux,
2906 GstAdaptiveDemuxStream * stream)
2908 gboolean ret = TRUE;
2910 /* Wait until we're cancelled or there's something for
2911 * us to download in the playlist or the playlist
2912 * became non-live */
2914 GST_DEBUG_OBJECT (demux, "No fragment left but live playlist, wait a bit");
2916 /* get the manifest_update_lock while still holding the manifest_lock.
2917 * This will prevent other threads to signal the condition (they will need
2918 * both manifest_lock and manifest_update_lock in order to signal).
2919 * It cannot deadlock because all threads always get the manifest_lock first
2920 * and manifest_update_lock second.
2922 g_mutex_lock (&demux->priv->manifest_update_lock);
2924 GST_MANIFEST_UNLOCK (demux);
2926 g_cond_wait (&demux->priv->manifest_cond,
2927 &demux->priv->manifest_update_lock);
2928 g_mutex_unlock (&demux->priv->manifest_update_lock);
2930 GST_MANIFEST_LOCK (demux);
2932 /* check for cancelled every time we get the manifest_lock */
2933 g_mutex_lock (&stream->fragment_download_lock);
2934 if (G_UNLIKELY (stream->cancelled)) {
2936 stream->last_ret = GST_FLOW_FLUSHING;
2937 g_mutex_unlock (&stream->fragment_download_lock);
2940 g_mutex_unlock (&stream->fragment_download_lock);
2942 /* Got a new fragment or not live anymore? */
2943 if (gst_adaptive_demux_stream_update_fragment_info (demux, stream) ==
2945 GST_DEBUG_OBJECT (demux, "new fragment available, "
2946 "not waiting for manifest update");
2951 if (!gst_adaptive_demux_is_live (demux)) {
2952 GST_DEBUG_OBJECT (demux, "Not live anymore, "
2953 "not waiting for manifest update");
2958 GST_DEBUG_OBJECT (demux, "Retrying now");
2962 /* must be called with manifest_lock taken */
2964 gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream,
2965 const gchar * uri, const gchar * referer, gboolean refresh,
2966 gboolean allow_cache)
2968 GstAdaptiveDemux *demux = stream->demux;
2970 if (!gst_uri_is_valid (uri)) {
2971 GST_WARNING_OBJECT (stream->pad, "Invalid URI: %s", uri);
2975 /* Try to re-use existing source element */
2976 if (stream->src != NULL) {
2977 gchar *old_protocol, *new_protocol;
2980 old_uri = gst_uri_handler_get_uri (GST_URI_HANDLER (stream->uri_handler));
2981 old_protocol = gst_uri_get_protocol (old_uri);
2982 new_protocol = gst_uri_get_protocol (uri);
2984 if (!g_str_equal (old_protocol, new_protocol)) {
2985 GstElement *src = stream->src;
2988 gst_object_unref (stream->src_srcpad);
2989 stream->src_srcpad = NULL;
2990 GST_MANIFEST_UNLOCK (demux);
2991 gst_element_set_locked_state (src, TRUE);
2992 gst_element_set_state (src, GST_STATE_NULL);
2993 gst_bin_remove (GST_BIN_CAST (demux), src);
2994 GST_MANIFEST_LOCK (demux);
2995 GST_DEBUG_OBJECT (demux, "Can't re-use old source element");
2999 GST_DEBUG_OBJECT (demux, "Re-using old source element");
3000 if (!gst_uri_handler_set_uri (GST_URI_HANDLER (stream->uri_handler), uri,
3002 GstElement *src = stream->src;
3005 GST_DEBUG_OBJECT (demux, "Failed to re-use old source element: %s",
3006 err ? err->message : "Unknown error");
3007 g_clear_error (&err);
3008 gst_object_unref (stream->src_srcpad);
3009 stream->src_srcpad = NULL;
3010 GST_MANIFEST_UNLOCK (demux);
3011 gst_element_set_locked_state (src, TRUE);
3012 gst_element_set_state (src, GST_STATE_NULL);
3013 gst_bin_remove (GST_BIN_CAST (demux), src);
3014 GST_MANIFEST_LOCK (demux);
3018 g_free (old_protocol);
3019 g_free (new_protocol);
3022 if (stream->src == NULL) {
3023 GstPad *uri_handler_src;
3026 GstElement *uri_handler;
3028 GstPadLinkReturn pad_link_ret;
3029 GObjectClass *gobject_class;
3030 gchar *internal_name, *bin_name;
3032 /* Our src consists of a bin containing uri_handler -> queue . The
3033 * purpose of the queue is to allow the uri_handler to download an
3034 * entire fragment without blocking, so we can accurately measure the
3035 * download bitrate. */
3037 queue = gst_element_factory_make ("queue", NULL);
3041 g_object_set (queue, "max-size-bytes", (guint) SRC_QUEUE_MAX_BYTES, NULL);
3042 g_object_set (queue, "max-size-buffers", (guint) 0, NULL);
3043 g_object_set (queue, "max-size-time", (guint64) 0, NULL);
3045 uri_handler = gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
3046 if (uri_handler == NULL) {
3047 GST_ELEMENT_ERROR (demux, CORE, MISSING_PLUGIN,
3048 ("Missing plugin to handle URI: '%s'", uri), (NULL));
3049 gst_object_unref (queue);
3053 gobject_class = G_OBJECT_GET_CLASS (uri_handler);
3055 if (g_object_class_find_property (gobject_class, "compress"))
3056 g_object_set (uri_handler, "compress", FALSE, NULL);
3057 if (g_object_class_find_property (gobject_class, "keep-alive"))
3058 g_object_set (uri_handler, "keep-alive", TRUE, NULL);
3059 if (g_object_class_find_property (gobject_class, "extra-headers")) {
3060 if (referer || refresh || !allow_cache) {
3061 GstStructure *extra_headers = gst_structure_new_empty ("headers");
3064 gst_structure_set (extra_headers, "Referer", G_TYPE_STRING, referer,
3068 gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING,
3071 gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING,
3074 g_object_set (uri_handler, "extra-headers", extra_headers, NULL);
3076 gst_structure_free (extra_headers);
3078 g_object_set (uri_handler, "extra-headers", NULL, NULL);
3082 /* Source bin creation */
3083 bin_name = g_strdup_printf ("srcbin-%s", GST_PAD_NAME (stream->pad));
3084 stream->src = gst_bin_new (bin_name);
3086 if (stream->src == NULL) {
3087 gst_object_unref (queue);
3088 gst_object_unref (uri_handler);
3092 gst_bin_add (GST_BIN_CAST (stream->src), queue);
3093 gst_bin_add (GST_BIN_CAST (stream->src), uri_handler);
3095 uri_handler_src = gst_element_get_static_pad (uri_handler, "src");
3096 queue_sink = gst_element_get_static_pad (queue, "sink");
3099 gst_pad_link_full (uri_handler_src, queue_sink,
3100 GST_PAD_LINK_CHECK_NOTHING);
3101 if (GST_PAD_LINK_FAILED (pad_link_ret)) {
3102 GST_WARNING_OBJECT (demux,
3103 "Could not link pads %s:%s to %s:%s for reason %d",
3104 GST_DEBUG_PAD_NAME (uri_handler_src), GST_DEBUG_PAD_NAME (queue_sink),
3106 g_object_unref (queue_sink);
3107 g_object_unref (uri_handler_src);
3108 gst_object_unref (stream->src);
3113 /* Add a downstream event and data probe */
3114 gst_pad_add_probe (uri_handler_src, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM,
3115 (GstPadProbeCallback) _uri_handler_probe, stream, NULL);
3117 g_object_unref (queue_sink);
3118 g_object_unref (uri_handler_src);
3119 queue_src = gst_element_get_static_pad (queue, "src");
3120 stream->src_srcpad = gst_ghost_pad_new ("src", queue_src);
3121 g_object_unref (queue_src);
3122 gst_element_add_pad (stream->src, stream->src_srcpad);
3124 gst_element_set_locked_state (stream->src, TRUE);
3125 gst_bin_add (GST_BIN_CAST (demux), stream->src);
3126 stream->src_srcpad = gst_element_get_static_pad (stream->src, "src");
3128 /* set up our internal floating pad to drop all events from
3129 * the http src we don't care about. On the chain function
3130 * we just push the buffer forward */
3131 internal_name = g_strdup_printf ("internal-%s", GST_PAD_NAME (stream->pad));
3132 stream->internal_pad = gst_pad_new (internal_name, GST_PAD_SINK);
3133 g_free (internal_name);
3134 gst_object_set_parent (GST_OBJECT_CAST (stream->internal_pad),
3135 GST_OBJECT_CAST (demux));
3136 GST_OBJECT_FLAG_SET (stream->internal_pad, GST_PAD_FLAG_NEED_PARENT);
3137 gst_pad_set_element_private (stream->internal_pad, stream);
3138 gst_pad_set_active (stream->internal_pad, TRUE);
3139 gst_pad_set_chain_function (stream->internal_pad, _src_chain);
3140 gst_pad_set_event_function (stream->internal_pad, _src_event);
3141 gst_pad_set_query_function (stream->internal_pad, _src_query);
3143 if (gst_pad_link_full (stream->src_srcpad, stream->internal_pad,
3144 GST_PAD_LINK_CHECK_NOTHING) != GST_PAD_LINK_OK) {
3145 GST_ERROR_OBJECT (stream->pad, "Failed to link internal pad");
3149 stream->uri_handler = uri_handler;
3150 stream->queue = queue;
3152 stream->last_status_code = 200; /* default to OK */
3157 static GstPadProbeReturn
3158 gst_ad_stream_src_to_ready_cb (GstPad * pad, GstPadProbeInfo * info,
3161 GstAdaptiveDemuxStream *stream = user_data;
3163 /* The source's src pad is IDLE so now set the state to READY */
3164 g_mutex_lock (&stream->fragment_download_lock);
3165 stream->src_at_ready = TRUE;
3166 g_cond_signal (&stream->fragment_download_cond);
3167 g_mutex_unlock (&stream->fragment_download_lock);
3169 return GST_PAD_PROBE_REMOVE;
3172 #ifndef GST_DISABLE_GST_DEBUG
3174 uritype (GstAdaptiveDemuxStream * s)
3176 if (s->downloading_header)
3178 if (s->downloading_index)
3184 /* must be called with manifest_lock taken.
3185 * Can temporarily release manifest_lock
3187 * Will return when URI is fully downloaded (or aborted/errored)
3189 static GstFlowReturn
3190 gst_adaptive_demux_stream_download_uri (GstAdaptiveDemux * demux,
3191 GstAdaptiveDemuxStream * stream, const gchar * uri, gint64 start,
3192 gint64 end, guint * http_status)
3194 GstFlowReturn ret = GST_FLOW_OK;
3195 GST_DEBUG_OBJECT (stream->pad,
3196 "Downloading %s uri: %s, range:%" G_GINT64_FORMAT " - %" G_GINT64_FORMAT,
3197 uritype (stream), uri, start, end);
3200 *http_status = 200; /* default to ok if no further information */
3202 if (!gst_adaptive_demux_stream_update_source (stream, uri, NULL, FALSE, TRUE)) {
3203 ret = stream->last_ret = GST_FLOW_ERROR;
3207 gst_element_set_locked_state (stream->src, TRUE);
3209 GST_MANIFEST_UNLOCK (demux);
3210 if (gst_element_set_state (stream->src,
3211 GST_STATE_READY) != GST_STATE_CHANGE_FAILURE) {
3212 /* If ranges are specified, seek to it */
3213 if (start != 0 || end != -1) {
3214 /* HTTP ranges are inclusive, GStreamer segments are exclusive for the
3218 /* Send the seek event to the uri_handler, as the other pipeline elements
3219 * can't handle it when READY. */
3220 if (!gst_element_send_event (stream->uri_handler, gst_event_new_seek (1.0,
3221 GST_FORMAT_BYTES, (GstSeekFlags) GST_SEEK_FLAG_FLUSH,
3222 GST_SEEK_TYPE_SET, start, GST_SEEK_TYPE_SET, end))) {
3224 GST_MANIFEST_LOCK (demux);
3225 /* looks like the source can't handle seeks in READY */
3226 g_clear_error (&stream->last_error);
3227 stream->last_error = g_error_new (GST_CORE_ERROR,
3228 GST_CORE_ERROR_NOT_IMPLEMENTED,
3229 "Source element can't handle range requests");
3230 stream->last_ret = GST_FLOW_ERROR;
3232 GST_MANIFEST_LOCK (demux);
3235 GST_MANIFEST_LOCK (demux);
3238 if (G_LIKELY (stream->last_ret == GST_FLOW_OK)) {
3239 stream->download_start_time =
3240 GST_TIME_AS_USECONDS (gst_adaptive_demux_get_monotonic_time (demux));
3242 /* src element is in state READY. Before we start it, we reset
3245 g_mutex_lock (&stream->fragment_download_lock);
3246 stream->download_finished = FALSE;
3247 stream->downloading_first_buffer = TRUE;
3248 g_mutex_unlock (&stream->fragment_download_lock);
3250 GST_MANIFEST_UNLOCK (demux);
3252 if (!gst_element_sync_state_with_parent (stream->src)) {
3253 GST_WARNING_OBJECT (demux, "Could not sync state for src element");
3254 GST_MANIFEST_LOCK (demux);
3255 ret = stream->last_ret = GST_FLOW_ERROR;
3259 /* wait for the fragment to be completely downloaded */
3260 GST_DEBUG_OBJECT (stream->pad,
3261 "Waiting for %s download to finish: %s", uritype (stream), uri);
3263 g_mutex_lock (&stream->fragment_download_lock);
3264 stream->src_at_ready = FALSE;
3265 if (G_UNLIKELY (stream->cancelled)) {
3266 g_mutex_unlock (&stream->fragment_download_lock);
3267 GST_MANIFEST_LOCK (demux);
3268 ret = stream->last_ret = GST_FLOW_FLUSHING;
3271 /* download_finished is only set:
3272 * * in ::fragment_download_finish()
3273 * * if EOS is received on the _src pad
3275 while (!stream->cancelled && !stream->download_finished) {
3276 g_cond_wait (&stream->fragment_download_cond,
3277 &stream->fragment_download_lock);
3279 g_mutex_unlock (&stream->fragment_download_lock);
3281 GST_DEBUG_OBJECT (stream->pad,
3282 "Finished Waiting for %s download: %s", uritype (stream), uri);
3284 GST_MANIFEST_LOCK (demux);
3285 g_mutex_lock (&stream->fragment_download_lock);
3286 if (G_UNLIKELY (stream->cancelled)) {
3287 ret = stream->last_ret = GST_FLOW_FLUSHING;
3288 g_mutex_unlock (&stream->fragment_download_lock);
3291 g_mutex_unlock (&stream->fragment_download_lock);
3293 ret = stream->last_ret;
3295 GST_DEBUG_OBJECT (stream->pad, "%s download finished: %s %d %s",
3296 uritype (stream), uri, stream->last_ret,
3297 gst_flow_get_name (stream->last_ret));
3298 if (stream->last_ret != GST_FLOW_OK && http_status) {
3299 *http_status = stream->last_status_code;
3303 /* changing src element state might try to join the streaming thread, so
3304 * we must not hold the manifest lock.
3306 GST_MANIFEST_UNLOCK (demux);
3308 GST_MANIFEST_UNLOCK (demux);
3309 if (stream->last_ret == GST_FLOW_OK)
3310 stream->last_ret = GST_FLOW_CUSTOM_ERROR;
3311 ret = GST_FLOW_CUSTOM_ERROR;
3314 stream->src_at_ready = FALSE;
3316 gst_element_set_locked_state (stream->src, TRUE);
3317 gst_pad_add_probe (stream->src_srcpad, GST_PAD_PROBE_TYPE_IDLE,
3318 gst_ad_stream_src_to_ready_cb, stream, NULL);
3320 g_mutex_lock (&stream->fragment_download_lock);
3321 while (!stream->src_at_ready) {
3322 g_cond_wait (&stream->fragment_download_cond,
3323 &stream->fragment_download_lock);
3325 g_mutex_unlock (&stream->fragment_download_lock);
3327 gst_element_set_state (stream->src, GST_STATE_READY);
3329 /* Need to drop the fragment_download_lock to get the MANIFEST lock */
3330 GST_MANIFEST_LOCK (demux);
3331 g_mutex_lock (&stream->fragment_download_lock);
3332 if (G_UNLIKELY (stream->cancelled)) {
3333 ret = stream->last_ret = GST_FLOW_FLUSHING;
3334 g_mutex_unlock (&stream->fragment_download_lock);
3337 g_mutex_unlock (&stream->fragment_download_lock);
3339 /* deactivate and reactivate our ghostpad to make it fresh for a new
3341 gst_pad_set_active (stream->internal_pad, FALSE);
3342 gst_pad_set_active (stream->internal_pad, TRUE);
3347 /* must be called with manifest_lock taken.
3348 * Can temporarily release manifest_lock
3350 static GstFlowReturn
3351 gst_adaptive_demux_stream_download_header_fragment (GstAdaptiveDemuxStream *
3354 GstAdaptiveDemux *demux = stream->demux;
3355 GstFlowReturn ret = GST_FLOW_OK;
3357 if (stream->fragment.header_uri != NULL) {
3358 GST_DEBUG_OBJECT (demux, "Fetching header %s %" G_GINT64_FORMAT "-%"
3359 G_GINT64_FORMAT, stream->fragment.header_uri,
3360 stream->fragment.header_range_start, stream->fragment.header_range_end);
3362 stream->downloading_header = TRUE;
3363 ret = gst_adaptive_demux_stream_download_uri (demux, stream,
3364 stream->fragment.header_uri, stream->fragment.header_range_start,
3365 stream->fragment.header_range_end, NULL);
3366 stream->downloading_header = FALSE;
3369 /* check if we have an index */
3370 if (ret == GST_FLOW_OK) { /* TODO check for other valid types */
3372 if (stream->fragment.index_uri != NULL) {
3373 GST_DEBUG_OBJECT (demux,
3374 "Fetching index %s %" G_GINT64_FORMAT "-%" G_GINT64_FORMAT,
3375 stream->fragment.index_uri,
3376 stream->fragment.index_range_start, stream->fragment.index_range_end);
3377 stream->downloading_index = TRUE;
3378 ret = gst_adaptive_demux_stream_download_uri (demux, stream,
3379 stream->fragment.index_uri, stream->fragment.index_range_start,
3380 stream->fragment.index_range_end, NULL);
3381 stream->downloading_index = FALSE;
3388 /* must be called with manifest_lock taken.
3389 * Can temporarily release manifest_lock
3391 static GstFlowReturn
3392 gst_adaptive_demux_stream_download_fragment (GstAdaptiveDemuxStream * stream)
3394 GstAdaptiveDemux *demux = stream->demux;
3395 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
3398 gboolean retried_once = FALSE, live;
3400 guint last_status_code;
3403 /* THERE ARE THREE DIFFERENT VARIABLES FOR THE "BEGINNING" OF A FRAGMENT ! */
3404 stream->starting_fragment = TRUE;
3405 stream->last_ret = GST_FLOW_OK;
3406 stream->first_fragment_buffer = TRUE;
3408 GST_DEBUG_OBJECT (stream->pad, "Downloading %s%s%s",
3409 stream->fragment.uri ? "FRAGMENT " : "",
3410 stream->fragment.header_uri ? "HEADER " : "",
3411 stream->fragment.index_uri ? "INDEX" : "");
3413 if (stream->fragment.uri == NULL && stream->fragment.header_uri == NULL &&
3414 stream->fragment.index_uri == NULL)
3417 if (stream->need_header) {
3418 ret = gst_adaptive_demux_stream_download_header_fragment (stream);
3419 if (ret != GST_FLOW_OK) {
3422 stream->need_header = FALSE;
3427 url = stream->fragment.uri;
3428 GST_DEBUG_OBJECT (stream->pad, "Got url '%s' for stream %p", url, stream);
3432 stream->last_ret = GST_FLOW_OK;
3435 /* Download the actual fragment, either in fragments or in one go */
3436 if (klass->need_another_chunk && klass->need_another_chunk (stream)
3437 && stream->fragment.chunk_size != 0) {
3438 /* Handle chunk downloading */
3439 gint64 range_start, range_end, chunk_start, chunk_end;
3440 guint64 download_total_bytes;
3441 gint chunk_size = stream->fragment.chunk_size;
3443 range_start = chunk_start = stream->fragment.range_start;
3444 range_end = stream->fragment.range_end;
3445 /* HTTP ranges are inclusive for the end */
3446 if (chunk_size != -1)
3447 chunk_end = range_start + chunk_size - 1;
3449 chunk_end = range_end;
3451 if (range_end != -1)
3452 chunk_end = MIN (chunk_end, range_end);
3454 while (!stream->fragment.finished && (chunk_start <= range_end
3455 || range_end == -1)) {
3456 download_total_bytes = stream->download_total_bytes;
3459 gst_adaptive_demux_stream_download_uri (demux, stream, url,
3460 chunk_start, chunk_end, &http_status);
3462 GST_DEBUG_OBJECT (stream->pad,
3463 "Fragment chunk download result: %d (%d) %s", stream->last_ret,
3464 http_status, gst_flow_get_name (stream->last_ret));
3466 /* Don't retry for any chunks except the first. We would have sent
3467 * data downstream already otherwise and it's difficult to recover
3468 * from that in a meaningful way */
3469 if (chunk_start > range_start)
3470 retried_once = TRUE;
3472 /* FIXME: Check for 416 Range Not Satisfiable here and fall back to
3473 * downloading up to -1. We don't know the full duration.
3474 * Needs https://bugzilla.gnome.org/show_bug.cgi?id=756806 */
3475 if (ret != GST_FLOW_OK && chunk_end == -1) {
3477 } else if (ret != GST_FLOW_OK) {
3479 stream->last_ret = GST_FLOW_OK;
3483 if (chunk_end == -1)
3486 /* Short read, we're at the end now */
3487 if (stream->download_total_bytes - download_total_bytes <
3488 chunk_end + 1 - chunk_start)
3491 if (!klass->need_another_chunk (stream))
3494 /* HTTP ranges are inclusive for the end */
3495 chunk_start += chunk_size;
3496 chunk_size = stream->fragment.chunk_size;
3497 if (chunk_size != -1)
3498 chunk_end = chunk_start + chunk_size - 1;
3500 chunk_end = range_end;
3502 if (range_end != -1)
3503 chunk_end = MIN (chunk_end, range_end);
3507 gst_adaptive_demux_stream_download_uri (demux, stream, url,
3508 stream->fragment.range_start, stream->fragment.range_end, &http_status);
3509 GST_DEBUG_OBJECT (stream->pad, "Fragment download result: %d (%d) %s",
3510 stream->last_ret, http_status, gst_flow_get_name (stream->last_ret));
3512 if (ret == GST_FLOW_OK)
3515 g_mutex_lock (&stream->fragment_download_lock);
3516 if (G_UNLIKELY (stream->cancelled)) {
3517 g_mutex_unlock (&stream->fragment_download_lock);
3520 g_mutex_unlock (&stream->fragment_download_lock);
3522 /* TODO check if we are truly stopping */
3523 if (ret != GST_FLOW_CUSTOM_ERROR)
3526 last_status_code = stream->last_status_code;
3527 GST_WARNING_OBJECT (stream->pad, "Got custom error, status %u, dc %d",
3528 last_status_code, stream->download_error_count);
3530 live = gst_adaptive_demux_is_live (demux);
3531 if (!retried_once && ((last_status_code / 100 == 4 && live)
3532 || last_status_code / 100 == 5)) {
3534 /* if current position is before available start, switch to next */
3535 if (!gst_adaptive_demux_stream_has_next_fragment (demux, stream))
3539 gint64 range_start, range_stop;
3541 if (!gst_adaptive_demux_get_live_seek_range (demux, &range_start,
3545 if (demux->segment.position < range_start) {
3546 GST_DEBUG_OBJECT (stream->pad, "Retrying once with next segment");
3547 stream->last_ret = GST_FLOW_OK;
3548 ret = gst_adaptive_demux_eos_handling (stream);
3549 GST_DEBUG_OBJECT (stream->pad, "finish_fragment: %s",
3550 gst_flow_get_name (ret));
3551 GST_DEBUG_OBJECT (demux, "Calling update_fragment_info");
3552 ret = gst_adaptive_demux_stream_update_fragment_info (demux, stream);
3553 GST_DEBUG_OBJECT (stream->pad, "finish_fragment: %s",
3554 gst_flow_get_name (ret));
3555 if (ret == GST_FLOW_OK) {
3556 retried_once = TRUE;
3559 } else if (demux->segment.position > range_stop) {
3560 /* wait a bit to be in range, we don't have any locks at that point */
3562 gst_adaptive_demux_stream_get_fragment_waiting_time (demux, stream);
3563 if (wait_time > 0) {
3564 gint64 end_time = g_get_monotonic_time () + wait_time / GST_USECOND;
3566 GST_DEBUG_OBJECT (stream->pad,
3567 "Download waiting for %" GST_TIME_FORMAT,
3568 GST_TIME_ARGS (wait_time));
3570 GST_MANIFEST_UNLOCK (demux);
3571 g_mutex_lock (&stream->fragment_download_lock);
3572 if (G_UNLIKELY (stream->cancelled)) {
3573 g_mutex_unlock (&stream->fragment_download_lock);
3574 GST_MANIFEST_LOCK (demux);
3575 stream->last_ret = GST_FLOW_FLUSHING;
3579 g_cond_wait_until (&stream->fragment_download_cond,
3580 &stream->fragment_download_lock, end_time);
3581 if (G_UNLIKELY (stream->cancelled)) {
3582 g_mutex_unlock (&stream->fragment_download_lock);
3583 GST_MANIFEST_LOCK (demux);
3584 stream->last_ret = GST_FLOW_FLUSHING;
3587 } while (!stream->download_finished);
3588 g_mutex_unlock (&stream->fragment_download_lock);
3590 GST_MANIFEST_LOCK (demux);
3596 if (stream->download_error_count >= MAX_DOWNLOAD_ERROR_COUNT) {
3597 /* looks like there is no way of knowing when a live stream has ended
3598 * Have to assume we are falling behind and cause a manifest reload */
3599 GST_DEBUG_OBJECT (stream->pad, "Converting error of live stream to EOS");
3600 return GST_FLOW_EOS;
3602 } else if (!gst_adaptive_demux_stream_has_next_fragment (demux, stream)) {
3603 /* If this is the last fragment, consider failures EOS and not actual
3604 * errors. Due to rounding errors in the durations, the last fragment
3605 * might not actually exist */
3606 GST_DEBUG_OBJECT (stream->pad, "Converting error for last fragment to EOS");
3607 return GST_FLOW_EOS;
3609 /* retry once (same segment) for 5xx (server errors) */
3610 if (!retried_once) {
3611 retried_once = TRUE;
3612 /* wait a short time in case the server needs a bit to recover, we don't
3613 * care if we get woken up before end time. We can use sleep here since
3614 * we're already blocking and just want to wait some time. */
3615 g_usleep (100000); /* a tenth of a second */
3625 GST_ELEMENT_ERROR (demux, STREAM, DEMUX,
3626 (_("Failed to get fragment URL.")),
3627 ("An error happened when getting fragment URL"));
3628 gst_task_stop (stream->download_task);
3629 return GST_FLOW_ERROR;
3633 /* this function will take the manifest_lock and will keep it until the end.
3634 * It will release it temporarily only when going to sleep.
3635 * Every time it takes the manifest_lock, it will check for cancelled condition
3638 gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream * stream)
3640 GstAdaptiveDemux *demux = stream->demux;
3641 GstClockTime next_download = gst_adaptive_demux_get_monotonic_time (demux);
3645 GST_LOG_OBJECT (stream->pad, "download loop start");
3647 GST_MANIFEST_LOCK (demux);
3649 g_mutex_lock (&stream->fragment_download_lock);
3650 if (G_UNLIKELY (stream->cancelled)) {
3651 stream->last_ret = GST_FLOW_FLUSHING;
3652 g_mutex_unlock (&stream->fragment_download_lock);
3655 g_mutex_unlock (&stream->fragment_download_lock);
3657 /* Check if we're done with our segment */
3658 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
3659 if (demux->segment.rate > 0) {
3660 if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)
3661 && stream->segment.position >= stream->segment.stop) {
3662 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
3664 gst_task_stop (stream->download_task);
3665 goto end_of_manifest;
3668 if (GST_CLOCK_TIME_IS_VALID (demux->segment.start)
3669 && stream->segment.position <= stream->segment.start) {
3670 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
3672 gst_task_stop (stream->download_task);
3673 goto end_of_manifest;
3676 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
3678 /* Cleanup old streams if any */
3679 if (G_UNLIKELY (demux->priv->old_streams != NULL)) {
3680 GList *old_streams = demux->priv->old_streams;
3681 demux->priv->old_streams = NULL;
3683 GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams");
3684 g_list_free_full (old_streams,
3685 (GDestroyNotify) gst_adaptive_demux_stream_free);
3686 GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams (done)");
3688 /* gst_adaptive_demux_stream_free had temporarily released the manifest_lock.
3689 * Recheck the cancelled flag.
3691 g_mutex_lock (&stream->fragment_download_lock);
3692 if (G_UNLIKELY (stream->cancelled)) {
3693 stream->last_ret = GST_FLOW_FLUSHING;
3694 g_mutex_unlock (&stream->fragment_download_lock);
3697 g_mutex_unlock (&stream->fragment_download_lock);
3700 /* Restarting download, figure out new position
3701 * FIXME : Move this to a separate function ? */
3702 if (G_UNLIKELY (stream->restart_download)) {
3703 GstEvent *seg_event;
3704 GstClockTime cur, ts = 0;
3707 GST_DEBUG_OBJECT (stream->pad,
3708 "Activating stream due to reconfigure event");
3710 if (gst_pad_peer_query_position (stream->pad, GST_FORMAT_TIME, &pos)) {
3711 ts = (GstClockTime) pos;
3712 GST_DEBUG_OBJECT (demux, "Downstream position: %"
3713 GST_TIME_FORMAT, GST_TIME_ARGS (ts));
3715 /* query other pads as some faulty element in the pad's branch might
3716 * reject position queries. This should be better than using the
3717 * demux segment position that can be much ahead */
3720 for (iter = demux->streams; iter != NULL; iter = g_list_next (iter)) {
3721 GstAdaptiveDemuxStream *cur_stream =
3722 (GstAdaptiveDemuxStream *) iter->data;
3724 if (gst_pad_peer_query_position (cur_stream->pad, GST_FORMAT_TIME,
3726 ts = (GstClockTime) pos;
3727 GST_DEBUG_OBJECT (stream->pad, "Downstream position: %"
3728 GST_TIME_FORMAT, GST_TIME_ARGS (ts));
3734 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
3736 gst_segment_to_stream_time (&stream->segment, GST_FORMAT_TIME,
3737 stream->segment.position);
3739 /* we might have already pushed this data */
3742 GST_DEBUG_OBJECT (stream->pad, "Restarting stream at "
3743 "position %" GST_TIME_FORMAT, GST_TIME_ARGS (ts));
3745 if (GST_CLOCK_TIME_IS_VALID (ts)) {
3746 GstClockTime offset, period_start;
3749 gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
3750 period_start = gst_adaptive_demux_get_period_start_time (demux);
3752 /* TODO check return */
3753 gst_adaptive_demux_stream_seek (demux, stream, demux->segment.rate >= 0,
3756 stream->segment.position = ts - period_start + offset;
3759 /* The stream's segment is still correct except for
3760 * the position, so let's send a new one with the
3761 * updated position */
3762 seg_event = gst_event_new_segment (&stream->segment);
3763 gst_event_set_seqnum (seg_event, demux->priv->segment_seqnum);
3764 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
3766 GST_DEBUG_OBJECT (stream->pad, "Sending restart segment: %"
3767 GST_PTR_FORMAT, seg_event);
3768 gst_pad_push_event (stream->pad, seg_event);
3770 stream->discont = TRUE;
3771 stream->restart_download = FALSE;
3774 live = gst_adaptive_demux_is_live (demux);
3776 /* Get information about the fragment to download */
3777 GST_DEBUG_OBJECT (demux, "Calling update_fragment_info");
3778 ret = gst_adaptive_demux_stream_update_fragment_info (demux, stream);
3779 GST_DEBUG_OBJECT (stream->pad, "Fragment info update result: %d %s",
3780 ret, gst_flow_get_name (ret));
3781 if (ret == GST_FLOW_OK) {
3783 /* wait for live fragments to be available */
3786 gst_adaptive_demux_stream_get_fragment_waiting_time (demux, stream);
3787 if (wait_time > 0) {
3788 GstClockTime end_time =
3789 gst_adaptive_demux_get_monotonic_time (demux) + wait_time;
3791 GST_DEBUG_OBJECT (stream->pad, "Download waiting for %" GST_TIME_FORMAT,
3792 GST_TIME_ARGS (wait_time));
3794 GST_MANIFEST_UNLOCK (demux);
3796 g_mutex_lock (&stream->fragment_download_lock);
3797 if (G_UNLIKELY (stream->cancelled)) {
3798 g_mutex_unlock (&stream->fragment_download_lock);
3799 GST_MANIFEST_LOCK (demux);
3800 stream->last_ret = GST_FLOW_FLUSHING;
3803 gst_adaptive_demux_wait_until (demux->realtime_clock,
3804 &stream->fragment_download_cond, &stream->fragment_download_lock,
3806 g_mutex_unlock (&stream->fragment_download_lock);
3808 GST_DEBUG_OBJECT (stream->pad, "Download finished waiting");
3810 GST_MANIFEST_LOCK (demux);
3812 g_mutex_lock (&stream->fragment_download_lock);
3813 if (G_UNLIKELY (stream->cancelled)) {
3814 stream->last_ret = GST_FLOW_FLUSHING;
3815 g_mutex_unlock (&stream->fragment_download_lock);
3818 g_mutex_unlock (&stream->fragment_download_lock);
3822 stream->last_ret = GST_FLOW_OK;
3824 next_download = gst_adaptive_demux_get_monotonic_time (demux);
3825 ret = gst_adaptive_demux_stream_download_fragment (stream);
3827 if (ret == GST_FLOW_FLUSHING) {
3828 g_mutex_lock (&stream->fragment_download_lock);
3829 if (G_UNLIKELY (stream->cancelled)) {
3830 stream->last_ret = GST_FLOW_FLUSHING;
3831 g_mutex_unlock (&stream->fragment_download_lock);
3834 g_mutex_unlock (&stream->fragment_download_lock);
3838 stream->last_ret = ret;
3843 break; /* all is good, let's go */
3845 GST_DEBUG_OBJECT (stream->pad, "EOS, checking to stop download loop");
3847 /* we push the EOS after releasing the object lock */
3848 if (gst_adaptive_demux_is_live (demux)
3849 && (demux->segment.rate == 1.0
3850 || gst_adaptive_demux_stream_in_live_seek_range (demux,
3852 GstAdaptiveDemuxClass *demux_class =
3853 GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
3855 /* this might be a fragment download error, refresh the manifest, just in case */
3856 if (!demux_class->requires_periodical_playlist_update (demux)) {
3857 ret = gst_adaptive_demux_update_manifest (demux);
3859 /* Wait only if we can ensure current manifest has been expired.
3860 * The meaning "we have next period" *WITH* EOS is that, current
3861 * period has been ended but we can continue to the next period */
3862 } else if (!gst_adaptive_demux_has_next_period (demux) &&
3863 gst_adaptive_demux_stream_wait_manifest_update (demux, stream)) {
3866 gst_task_stop (stream->download_task);
3867 if (stream->replaced) {
3871 gst_task_stop (stream->download_task);
3874 if (gst_adaptive_demux_combine_flows (demux) == GST_FLOW_EOS) {
3875 if (gst_adaptive_demux_has_next_period (demux)) {
3876 GST_DEBUG_OBJECT (stream->pad,
3877 "Next period available, not sending EOS");
3878 gst_adaptive_demux_advance_period (demux);
3884 case GST_FLOW_NOT_LINKED:
3887 gst_task_stop (stream->download_task);
3889 ret = gst_adaptive_demux_combine_flows (demux);
3890 if (ret == GST_FLOW_NOT_LINKED) {
3891 GST_ELEMENT_FLOW_ERROR (demux, ret);
3896 case GST_FLOW_FLUSHING:{
3899 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
3900 GstAdaptiveDemuxStream *other;
3903 gst_task_stop (other->download_task);
3909 if (ret <= GST_FLOW_ERROR) {
3910 gboolean is_live = gst_adaptive_demux_is_live (demux);
3911 GST_WARNING_OBJECT (demux, "Error while downloading fragment");
3912 if (++stream->download_error_count > MAX_DOWNLOAD_ERROR_COUNT) {
3913 goto download_error;
3916 g_clear_error (&stream->last_error);
3918 /* First try to update the playlist for non-live playlists
3919 * in case the URIs have changed in the meantime. But only
3920 * try it the first time, after that we're going to wait a
3921 * a bit to not flood the server */
3922 if (stream->download_error_count == 1 && !is_live) {
3923 /* TODO hlsdemux had more options to this function (boolean and err) */
3925 if (gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {
3926 /* Retry immediately, the playlist actually has changed */
3927 GST_DEBUG_OBJECT (demux, "Updated the playlist");
3932 /* Wait half the fragment duration before retrying */
3933 next_download += stream->fragment.duration / 2;
3935 GST_MANIFEST_UNLOCK (demux);
3937 g_mutex_lock (&stream->fragment_download_lock);
3938 if (G_UNLIKELY (stream->cancelled)) {
3939 g_mutex_unlock (&stream->fragment_download_lock);
3940 GST_MANIFEST_LOCK (demux);
3941 stream->last_ret = GST_FLOW_FLUSHING;
3944 gst_adaptive_demux_wait_until (demux->realtime_clock,
3945 &stream->fragment_download_cond, &stream->fragment_download_lock,
3947 g_mutex_unlock (&stream->fragment_download_lock);
3949 GST_DEBUG_OBJECT (demux, "Retrying now");
3951 GST_MANIFEST_LOCK (demux);
3953 g_mutex_lock (&stream->fragment_download_lock);
3954 if (G_UNLIKELY (stream->cancelled)) {
3955 stream->last_ret = GST_FLOW_FLUSHING;
3956 g_mutex_unlock (&stream->fragment_download_lock);
3959 g_mutex_unlock (&stream->fragment_download_lock);
3961 /* Refetch the playlist now after we waited */
3963 && gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {
3964 GST_DEBUG_OBJECT (demux, "Updated the playlist");
3972 if (G_UNLIKELY (ret == GST_FLOW_EOS)) {
3973 if (GST_OBJECT_PARENT (stream->pad) != NULL) {
3974 if (demux->next_streams == NULL && demux->prepared_streams == NULL) {
3975 GST_DEBUG_OBJECT (stream->src, "Pushing EOS on pad");
3976 gst_adaptive_demux_stream_push_event (stream, gst_event_new_eos ());
3978 GST_DEBUG_OBJECT (stream->src,
3979 "Stream is EOS, but we're switching fragments. Not sending.");
3982 GST_ERROR_OBJECT (demux, "Can't push EOS on non-exposed pad");
3983 goto download_error;
3988 GST_MANIFEST_UNLOCK (demux);
3989 GST_LOG_OBJECT (stream->pad, "download loop end");
3994 GST_DEBUG_OBJECT (stream->pad, "Stream has been cancelled");
4001 if (stream->last_error) {
4002 gchar *debug = g_strdup_printf ("Error on stream %s:%s",
4003 GST_DEBUG_PAD_NAME (stream->pad));
4005 gst_message_new_error (GST_OBJECT_CAST (demux), stream->last_error,
4007 GST_ERROR_OBJECT (stream->pad, "Download error: %s",
4008 stream->last_error->message);
4012 g_error_new (GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_NOT_FOUND,
4013 _("Couldn't download fragments"));
4015 gst_message_new_error (GST_OBJECT_CAST (demux), err,
4016 "Fragment downloading has failed consecutive times");
4018 GST_ERROR_OBJECT (stream->pad,
4019 "Download error: Couldn't download fragments, too many failures");
4022 gst_task_stop (stream->download_task);
4024 GstElement *src = stream->src;
4027 GST_MANIFEST_UNLOCK (demux);
4028 gst_element_set_locked_state (src, TRUE);
4029 gst_element_set_state (src, GST_STATE_NULL);
4030 gst_bin_remove (GST_BIN_CAST (demux), src);
4031 GST_MANIFEST_LOCK (demux);
4034 gst_element_post_message (GST_ELEMENT_CAST (demux), msg);
4041 gst_adaptive_demux_updates_loop (GstAdaptiveDemux * demux)
4043 GstClockTime next_update;
4044 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4046 /* Loop for updating of the playlist. This periodically checks if
4047 * the playlist is updated and does so, then signals the streaming
4048 * thread in case it can continue downloading now. */
4050 /* block until the next scheduled update or the signal to quit this thread */
4051 GST_DEBUG_OBJECT (demux, "Started updates task");
4053 GST_MANIFEST_LOCK (demux);
4056 gst_adaptive_demux_get_monotonic_time (demux) +
4057 klass->get_manifest_update_interval (demux) * GST_USECOND;
4059 /* Updating playlist only needed for live playlists */
4060 while (gst_adaptive_demux_is_live (demux)) {
4061 GstFlowReturn ret = GST_FLOW_OK;
4063 /* Wait here until we should do the next update or we're cancelled */
4064 GST_DEBUG_OBJECT (demux, "Wait for next playlist update");
4066 GST_MANIFEST_UNLOCK (demux);
4068 g_mutex_lock (&demux->priv->updates_timed_lock);
4069 if (demux->priv->stop_updates_task) {
4070 g_mutex_unlock (&demux->priv->updates_timed_lock);
4073 gst_adaptive_demux_wait_until (demux->realtime_clock,
4074 &demux->priv->updates_timed_cond,
4075 &demux->priv->updates_timed_lock, next_update);
4076 g_mutex_unlock (&demux->priv->updates_timed_lock);
4078 g_mutex_lock (&demux->priv->updates_timed_lock);
4079 if (demux->priv->stop_updates_task) {
4080 g_mutex_unlock (&demux->priv->updates_timed_lock);
4083 g_mutex_unlock (&demux->priv->updates_timed_lock);
4085 GST_MANIFEST_LOCK (demux);
4087 GST_DEBUG_OBJECT (demux, "Updating playlist");
4089 ret = gst_adaptive_demux_update_manifest (demux);
4091 if (ret == GST_FLOW_EOS) {
4092 } else if (ret != GST_FLOW_OK) {
4093 /* update_failed_count is used only here, no need to protect it */
4094 demux->priv->update_failed_count++;
4095 if (demux->priv->update_failed_count <= DEFAULT_FAILED_COUNT) {
4096 GST_WARNING_OBJECT (demux, "Could not update the playlist, flow: %s",
4097 gst_flow_get_name (ret));
4098 next_update = gst_adaptive_demux_get_monotonic_time (demux)
4099 + klass->get_manifest_update_interval (demux) * GST_USECOND;
4101 GST_ELEMENT_ERROR (demux, STREAM, FAILED,
4102 (_("Internal data stream error.")), ("Could not update playlist"));
4103 GST_DEBUG_OBJECT (demux, "Stopped updates task because of error");
4104 gst_task_stop (demux->priv->updates_task);
4105 GST_MANIFEST_UNLOCK (demux);
4109 GST_DEBUG_OBJECT (demux, "Updated playlist successfully");
4110 demux->priv->update_failed_count = 0;
4112 gst_adaptive_demux_get_monotonic_time (demux) +
4113 klass->get_manifest_update_interval (demux) * GST_USECOND;
4115 /* Wake up download tasks */
4116 g_mutex_lock (&demux->priv->manifest_update_lock);
4117 g_cond_broadcast (&demux->priv->manifest_cond);
4118 g_mutex_unlock (&demux->priv->manifest_update_lock);
4122 GST_MANIFEST_UNLOCK (demux);
4126 GST_DEBUG_OBJECT (demux, "Stop updates task request detected.");
4135 /* must be called with manifest_lock taken */
4137 gst_adaptive_demux_stream_push_event (GstAdaptiveDemuxStream * stream,
4142 GstAdaptiveDemux *demux = stream->demux;
4144 if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
4148 pad = gst_object_ref (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream));
4150 /* Can't push events holding the manifest lock */
4151 GST_MANIFEST_UNLOCK (demux);
4153 GST_DEBUG_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream),
4154 "Pushing event %" GST_PTR_FORMAT, event);
4156 ret = gst_pad_push_event (pad, event);
4158 gst_object_unref (pad);
4160 GST_MANIFEST_LOCK (demux);
4165 /* must be called with manifest_lock taken */
4167 gst_adaptive_demux_is_live (GstAdaptiveDemux * demux)
4169 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4172 return klass->is_live (demux);
4176 /* must be called with manifest_lock taken */
4177 static GstFlowReturn
4178 gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux,
4179 GstAdaptiveDemuxStream * stream, gboolean forward, GstSeekFlags flags,
4180 GstClockTime ts, GstClockTime * final_ts)
4182 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4184 if (klass->stream_seek)
4185 return klass->stream_seek (stream, forward, flags, ts, final_ts);
4186 return GST_FLOW_ERROR;
4189 /* must be called with manifest_lock taken */
4191 gst_adaptive_demux_stream_has_next_fragment (GstAdaptiveDemux * demux,
4192 GstAdaptiveDemuxStream * stream)
4194 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4195 gboolean ret = TRUE;
4197 if (klass->stream_has_next_fragment)
4198 ret = klass->stream_has_next_fragment (stream);
4203 /* must be called with manifest_lock taken */
4205 * the ::finish_fragment() handlers when an *actual* fragment is done
4208 gst_adaptive_demux_stream_advance_fragment (GstAdaptiveDemux * demux,
4209 GstAdaptiveDemuxStream * stream, GstClockTime duration)
4213 if (stream->last_ret == GST_FLOW_OK) {
4215 gst_adaptive_demux_stream_advance_fragment_unlocked (demux, stream,
4218 ret = stream->last_ret;
4223 /* must be called with manifest_lock taken */
4225 gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux,
4226 GstAdaptiveDemuxStream * stream, GstClockTime duration)
4228 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4231 g_return_val_if_fail (klass->stream_advance_fragment != NULL, GST_FLOW_ERROR);
4233 GST_LOG_OBJECT (stream->pad,
4234 "timestamp %" GST_TIME_FORMAT " duration:%" GST_TIME_FORMAT,
4235 GST_TIME_ARGS (stream->fragment.timestamp), GST_TIME_ARGS (duration));
4237 stream->download_error_count = 0;
4238 g_clear_error (&stream->last_error);
4240 /* FIXME - url has no indication of byte ranges for subsegments */
4241 /* FIXME : All those time statistics are biased, since they are calculated
4242 * *AFTER* the queue2, which might be blocking. They should ideally be
4243 * calculated *before* queue2 in the uri_handler_probe */
4244 gst_element_post_message (GST_ELEMENT_CAST (demux),
4245 gst_message_new_element (GST_OBJECT_CAST (demux),
4246 gst_structure_new (GST_ADAPTIVE_DEMUX_STATISTICS_MESSAGE_NAME,
4247 "manifest-uri", G_TYPE_STRING,
4248 demux->manifest_uri, "uri", G_TYPE_STRING,
4249 stream->fragment.uri, "fragment-start-time",
4250 GST_TYPE_CLOCK_TIME, stream->download_start_time,
4251 "fragment-stop-time", GST_TYPE_CLOCK_TIME,
4252 gst_util_get_timestamp (), "fragment-size", G_TYPE_UINT64,
4253 stream->download_total_bytes, "fragment-download-time",
4254 GST_TYPE_CLOCK_TIME, stream->last_download_time, NULL)));
4256 /* Don't update to the end of the segment if in reverse playback */
4257 GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);
4258 if (GST_CLOCK_TIME_IS_VALID (duration) && demux->segment.rate > 0) {
4259 GstClockTime offset =
4260 gst_adaptive_demux_stream_get_presentation_offset (demux, stream);
4261 GstClockTime period_start =
4262 gst_adaptive_demux_get_period_start_time (demux);
4264 stream->segment.position += duration;
4266 /* Convert from position inside the stream's segment to the demuxer's
4267 * segment, they are not necessarily the same */
4268 if (stream->segment.position - offset + period_start >
4269 demux->segment.position)
4270 demux->segment.position =
4271 stream->segment.position - offset + period_start;
4273 GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);
4275 /* When advancing with a non 1.0 rate on live streams, we need to check
4276 * the live seeking range again to make sure we can still advance to
4278 if (demux->segment.rate != 1.0 && gst_adaptive_demux_is_live (demux)) {
4279 if (!gst_adaptive_demux_stream_in_live_seek_range (demux, stream))
4282 ret = klass->stream_advance_fragment (stream);
4283 } else if (gst_adaptive_demux_is_live (demux)
4284 || gst_adaptive_demux_stream_has_next_fragment (demux, stream)) {
4285 ret = klass->stream_advance_fragment (stream);
4290 stream->download_start_time =
4291 GST_TIME_AS_USECONDS (gst_adaptive_demux_get_monotonic_time (demux));
4293 if (ret == GST_FLOW_OK) {
4294 if (gst_adaptive_demux_stream_select_bitrate (demux, stream,
4295 gst_adaptive_demux_stream_update_current_bitrate (demux, stream))) {
4296 stream->need_header = TRUE;
4297 ret = (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH;
4300 /* the subclass might want to switch pads */
4301 if (G_UNLIKELY (demux->next_streams)) {
4303 gboolean can_expose = TRUE;
4305 gst_task_stop (stream->download_task);
4309 for (iter = demux->streams; iter; iter = g_list_next (iter)) {
4310 /* Only expose if all streams are now cancelled or finished downloading */
4311 GstAdaptiveDemuxStream *other = iter->data;
4312 if (other != stream) {
4313 g_mutex_lock (&other->fragment_download_lock);
4314 can_expose &= (other->cancelled == TRUE
4315 || other->download_finished == TRUE);
4316 g_mutex_unlock (&other->fragment_download_lock);
4321 GST_DEBUG_OBJECT (demux, "Subclass wants new pads "
4322 "to do bitrate switching");
4323 gst_adaptive_demux_prepare_streams (demux, FALSE);
4324 gst_adaptive_demux_start_tasks (demux, TRUE);
4326 GST_LOG_OBJECT (demux, "Not switching yet - ongoing downloads");
4334 /* must be called with manifest_lock taken */
4336 gst_adaptive_demux_stream_select_bitrate (GstAdaptiveDemux *
4337 demux, GstAdaptiveDemuxStream * stream, guint64 bitrate)
4339 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4341 if (klass->stream_select_bitrate)
4342 return klass->stream_select_bitrate (stream, bitrate);
4346 /* must be called with manifest_lock taken */
4347 static GstFlowReturn
4348 gst_adaptive_demux_stream_update_fragment_info (GstAdaptiveDemux * demux,
4349 GstAdaptiveDemuxStream * stream)
4351 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4354 g_return_val_if_fail (klass->stream_update_fragment_info != NULL,
4357 /* Make sure the sub-class will update bitrate, or else
4359 stream->fragment.bitrate = 0;
4360 stream->fragment.finished = FALSE;
4362 GST_LOG_OBJECT (stream->pad, "position %" GST_TIME_FORMAT,
4363 GST_TIME_ARGS (stream->segment.position));
4365 ret = klass->stream_update_fragment_info (stream);
4367 GST_LOG_OBJECT (stream->pad, "ret:%s uri:%s", gst_flow_get_name (ret),
4368 stream->fragment.uri);
4369 if (ret == GST_FLOW_OK) {
4370 GST_LOG_OBJECT (stream->pad,
4371 "timestamp %" GST_TIME_FORMAT " duration:%" GST_TIME_FORMAT,
4372 GST_TIME_ARGS (stream->fragment.timestamp),
4373 GST_TIME_ARGS (stream->fragment.duration));
4374 GST_LOG_OBJECT (stream->pad,
4375 "range start:%" G_GINT64_FORMAT " end:%" G_GINT64_FORMAT,
4376 stream->fragment.range_start, stream->fragment.range_end);
4382 /* must be called with manifest_lock taken */
4384 gst_adaptive_demux_stream_get_fragment_waiting_time (GstAdaptiveDemux *
4385 demux, GstAdaptiveDemuxStream * stream)
4387 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4389 if (klass->stream_get_fragment_waiting_time)
4390 return klass->stream_get_fragment_waiting_time (stream);
4394 /* must be called with manifest_lock taken */
4395 static GstFlowReturn
4396 gst_adaptive_demux_update_manifest_default (GstAdaptiveDemux * demux)
4398 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4399 GstFragment *download;
4402 GError *error = NULL;
4404 download = gst_uri_downloader_fetch_uri (demux->downloader,
4405 demux->manifest_uri, NULL, TRUE, TRUE, TRUE, &error);
4407 g_free (demux->manifest_uri);
4408 g_free (demux->manifest_base_uri);
4409 if (download->redirect_permanent && download->redirect_uri) {
4410 demux->manifest_uri = g_strdup (download->redirect_uri);
4411 demux->manifest_base_uri = NULL;
4413 demux->manifest_uri = g_strdup (download->uri);
4414 demux->manifest_base_uri = g_strdup (download->redirect_uri);
4417 buffer = gst_fragment_get_buffer (download);
4418 g_object_unref (download);
4419 ret = klass->update_manifest_data (demux, buffer);
4420 gst_buffer_unref (buffer);
4421 /* FIXME: Should the manifest uri vars be reverted to original
4422 * values if updating fails? */
4424 GST_WARNING_OBJECT (demux, "Failed to download manifest: %s",
4426 ret = GST_FLOW_NOT_LINKED;
4428 g_clear_error (&error);
4433 /* must be called with manifest_lock taken */
4434 static GstFlowReturn
4435 gst_adaptive_demux_update_manifest (GstAdaptiveDemux * demux)
4437 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4440 ret = klass->update_manifest (demux);
4442 if (ret == GST_FLOW_OK) {
4443 GstClockTime duration;
4444 /* Send an updated duration message */
4445 duration = klass->get_duration (demux);
4446 if (duration != GST_CLOCK_TIME_NONE) {
4447 GST_DEBUG_OBJECT (demux,
4448 "Sending duration message : %" GST_TIME_FORMAT,
4449 GST_TIME_ARGS (duration));
4450 gst_element_post_message (GST_ELEMENT (demux),
4451 gst_message_new_duration_changed (GST_OBJECT (demux)));
4453 GST_DEBUG_OBJECT (demux,
4454 "Duration unknown, can not send the duration message");
4457 /* If a manifest changes it's liveness or periodic updateness, we need
4458 * to start/stop the manifest update task appropriately */
4459 /* Keep this condition in sync with the one in
4460 * gst_adaptive_demux_start_manifest_update_task()
4462 if (gst_adaptive_demux_is_live (demux) &&
4463 klass->requires_periodical_playlist_update (demux)) {
4464 gst_adaptive_demux_start_manifest_update_task (demux);
4466 gst_adaptive_demux_stop_manifest_update_task (demux);
4474 gst_adaptive_demux_stream_fragment_clear (GstAdaptiveDemuxStreamFragment * f)
4481 g_free (f->header_uri);
4482 f->header_uri = NULL;
4483 f->header_range_start = 0;
4484 f->header_range_end = -1;
4486 g_free (f->index_uri);
4487 f->index_uri = NULL;
4488 f->index_range_start = 0;
4489 f->index_range_end = -1;
4491 f->finished = FALSE;
4494 /* must be called with manifest_lock taken */
4496 gst_adaptive_demux_has_next_period (GstAdaptiveDemux * demux)
4498 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4499 gboolean ret = FALSE;
4501 if (klass->has_next_period)
4502 ret = klass->has_next_period (demux);
4503 GST_DEBUG_OBJECT (demux, "Has next period: %d", ret);
4507 /* must be called with manifest_lock taken */
4509 gst_adaptive_demux_advance_period (GstAdaptiveDemux * demux)
4511 GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
4513 g_return_if_fail (klass->advance_period != NULL);
4515 GST_DEBUG_OBJECT (demux, "Advancing to next period");
4516 klass->advance_period (demux);
4517 gst_adaptive_demux_prepare_streams (demux, FALSE);
4518 gst_adaptive_demux_start_tasks (demux, TRUE);
4522 * gst_adaptive_demux_get_monotonic_time:
4523 * Returns: a monotonically increasing time, using the system realtime clock
4526 gst_adaptive_demux_get_monotonic_time (GstAdaptiveDemux * demux)
4528 g_return_val_if_fail (demux != NULL, GST_CLOCK_TIME_NONE);
4529 return gst_clock_get_time (demux->realtime_clock);
4533 * gst_adaptive_demux_get_client_now_utc:
4534 * @demux: #GstAdaptiveDemux
4535 * Returns: the client's estimate of UTC
4537 * Used to find the client's estimate of UTC, using the system realtime clock.
4540 gst_adaptive_demux_get_client_now_utc (GstAdaptiveDemux * demux)
4542 GstClockTime rtc_now;
4543 GDateTime *unix_datetime;
4544 GDateTime *result_datetime;
4545 gint64 utc_now_in_us;
4547 rtc_now = gst_clock_get_time (demux->realtime_clock);
4548 utc_now_in_us = demux->clock_offset + GST_TIME_AS_USECONDS (rtc_now);
4550 g_date_time_new_from_unix_utc (utc_now_in_us / G_TIME_SPAN_SECOND);
4552 g_date_time_add (unix_datetime, utc_now_in_us % G_TIME_SPAN_SECOND);
4553 g_date_time_unref (unix_datetime);
4554 return result_datetime;
4558 * gst_adaptive_demux_is_running
4559 * @demux: #GstAdaptiveDemux
4560 * Returns: whether the demuxer is processing data
4562 * Returns FALSE if shutdown has started (transitioning down from
4563 * PAUSED), otherwise TRUE.
4566 gst_adaptive_demux_is_running (GstAdaptiveDemux * demux)
4568 return g_atomic_int_get (&demux->running);
4571 static GstAdaptiveDemuxTimer *
4572 gst_adaptive_demux_timer_new (GCond * cond, GMutex * mutex)
4574 GstAdaptiveDemuxTimer *timer;
4576 timer = g_slice_new (GstAdaptiveDemuxTimer);
4577 timer->fired = FALSE;
4579 timer->mutex = mutex;
4580 g_atomic_int_set (&timer->ref_count, 1);
4584 static GstAdaptiveDemuxTimer *
4585 gst_adaptive_demux_timer_ref (GstAdaptiveDemuxTimer * timer)
4587 g_return_val_if_fail (timer != NULL, NULL);
4588 g_atomic_int_inc (&timer->ref_count);
4593 gst_adaptive_demux_timer_unref (GstAdaptiveDemuxTimer * timer)
4595 g_return_if_fail (timer != NULL);
4596 if (g_atomic_int_dec_and_test (&timer->ref_count)) {
4597 g_slice_free (GstAdaptiveDemuxTimer, timer);
4601 /* gst_adaptive_demux_wait_until:
4602 * A replacement for g_cond_wait_until that uses the clock rather
4603 * than system time to control the duration of the sleep. Typically
4604 * clock is actually a #GstSystemClock, in which case this function
4605 * behaves exactly like g_cond_wait_until. Inside unit tests,
4606 * the clock is typically a #GstTestClock, which allows tests to run
4608 * This function must be called with mutex held.
4611 gst_adaptive_demux_wait_until (GstClock * clock, GCond * cond, GMutex * mutex,
4612 GstClockTime end_time)
4614 GstAdaptiveDemuxTimer *timer;
4618 if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (end_time))) {
4619 /* for an invalid time, gst_clock_id_wait_async will try to call
4620 * gst_adaptive_demux_clock_callback from the current thread.
4621 * It still holds the mutex while doing that, so it will deadlock.
4622 * g_cond_wait_until would return immediately with false, so we'll do the same.
4626 timer = gst_adaptive_demux_timer_new (cond, mutex);
4627 timer->clock_id = gst_clock_new_single_shot_id (clock, end_time);
4629 gst_clock_id_wait_async (timer->clock_id,
4630 gst_adaptive_demux_clock_callback, gst_adaptive_demux_timer_ref (timer),
4631 (GDestroyNotify) gst_adaptive_demux_timer_unref);
4632 /* clock does not support asynchronously wait. Assert and return */
4633 if (res == GST_CLOCK_UNSUPPORTED) {
4634 gst_clock_id_unref (timer->clock_id);
4635 gst_adaptive_demux_timer_unref (timer);
4636 g_return_val_if_reached (TRUE);
4638 g_assert (!timer->fired);
4639 /* the gst_adaptive_demux_clock_callback() will signal the
4640 * cond when the clock's single shot timer fires, or the cond will be
4641 * signalled by another thread that wants to cause this wait to finish
4642 * early (e.g. to terminate the waiting thread).
4643 * There is no need for a while loop here, because that logic is
4644 * implemented by the function calling gst_adaptive_demux_wait_until() */
4645 g_cond_wait (cond, mutex);
4646 fired = timer->fired;
4648 gst_clock_id_unschedule (timer->clock_id);
4649 gst_clock_id_unref (timer->clock_id);
4650 gst_adaptive_demux_timer_unref (timer);
4655 gst_adaptive_demux_clock_callback (GstClock * clock,
4656 GstClockTime time, GstClockID id, gpointer user_data)
4658 GstAdaptiveDemuxTimer *timer = (GstAdaptiveDemuxTimer *) user_data;
4659 g_return_val_if_fail (timer != NULL, FALSE);
4660 g_mutex_lock (timer->mutex);
4661 timer->fired = TRUE;
4662 g_cond_signal (timer->cond);
4663 g_mutex_unlock (timer->mutex);
4668 * gst_adaptive_demux_get_qos_earliest_time:
4670 * Returns: The QOS earliest time
4675 gst_adaptive_demux_get_qos_earliest_time (GstAdaptiveDemux * demux)
4677 GstClockTime earliest;
4679 GST_OBJECT_LOCK (demux);
4680 earliest = demux->priv->qos_earliest_time;
4681 GST_OBJECT_UNLOCK (demux);