| /* GStreamer |
| * |
| * Copyright (C) 2014 Samsung Electronics. All rights reserved. |
| * Author: Thiago Santos <thiagoss@osg.samsung.com> |
| * |
| * This library is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU Library General Public |
| * License as published by the Free Software Foundation; either |
| * version 2 of the License, or (at your option) any later version. |
| * |
| * This library is distributed in the hope that it will be useful, |
| * but WITHOUT ANY WARRANTY; without even the implied warranty of |
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| * Library General Public License for more details. |
| * |
| * You should have received a copy of the GNU Library General Public |
| * License along with this library; if not, write to the |
| * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, |
| * Boston, MA 02110-1301, USA. |
| */ |
| |
| /** |
| * SECTION:gstadaptivedemux |
| * @short_description: Base class for adaptive demuxers |
| * @see_also: |
| * |
| * What is an adaptive demuxer? |
| * Adaptive demuxers are special demuxers in the sense that they don't |
| * actually demux data received from upstream but download the data |
| * themselves. |
| * |
| * Adaptive formats (HLS, DASH, MSS) are composed of a manifest file and |
| * a set of fragments. The manifest describes the available media and |
| * the sequence of fragments to use. Each fragment contains a small |
| * part of the media (typically only a few seconds). It is possible for |
| * the manifest to have the same media available in different configurations |
| * (bitrates for example) so that the client can select the one that |
| * best suits its scenario (network fluctuation, hardware requirements...). |
| * It is possible to switch from one representation of the media to another |
| * during playback. That's why it is called 'adaptive', because it can be |
| * adapted to the client's needs. |
| * |
| * Architectural overview: |
| * The manifest is received by the demuxer in its sink pad and, upon receiving |
| * EOS, it parses the manifest and exposes the streams available in it. For |
| * each stream a source element will be created and will download the list |
| * of fragments one by one. Once a fragment is finished downloading, the next |
| * URI is set to the source element and it starts fetching it and pushing |
| * through the stream's pad. This implies that each stream is independent from |
| * each other as it runs on a separate thread. |
| * |
| * After downloading each fragment, the download rate of it is calculated and |
| * the demuxer has a chance to switch to a different bitrate if needed. The |
| * switch can be done by simply pushing a new caps before the next fragment |
| * when codecs are the same, or by exposing a new pad group if it needs |
| * a codec change. |
| * |
| * Extra features: |
| * - Not linked streams: Streams that are not-linked have their download threads |
| * interrupted to save network bandwidth. When they are |
| * relinked a reconfigure event is received and the |
| * stream is restarted. |
| * |
| * Subclasses: |
| * While GstAdaptiveDemux is responsible for the workflow, it knows nothing |
| * about the intrinsics of the subclass formats, so the subclasses are |
| * resposible for maintaining the manifest data structures and stream |
| * information. |
| */ |
| |
| /* |
| MT safety. |
| The following rules were observed while implementing MT safety in adaptive demux: |
| 1. If a variable is accessed from multiple threads and at least one thread |
| writes to it, then all the accesses needs to be done from inside a critical section. |
| 2. If thread A wants to join thread B then at the moment it calls gst_task_join |
| it must not hold any mutexes that thread B might take. |
| |
| Adaptive demux API can be called from several threads. More, adaptive demux |
| starts some threads to monitor the download of fragments. In order to protect |
| accesses to shared variables (demux and streams) all the API functions that |
| can be run in different threads will need to get a mutex (manifest_lock) |
| when they start and release it when they end. Because some of those functions |
| can indirectly call other API functions (eg they can generate events or messages |
| that are processed in the same thread) the manifest_lock must be recursive. |
| |
| The manifest_lock will serialize the public API making access to shared |
| variables safe. But some of these functions will try at some moment to join |
| threads created by adaptive demux, or to change the state of src elements |
| (which will block trying to join the src element streaming thread). Because |
| of rule 2, those functions will need to release the manifest_lock during the |
| call of gst_task_join. During this time they can be interrupted by other API calls. |
| For example, during the precessing of a seek event, gst_adaptive_demux_stop_tasks |
| is called and this will join all threads. In order to prevent interruptions |
| during such period, all the API functions will also use a second lock: api_lock. |
| This will be taken at the beginning of the function and released at the end, |
| but this time this lock will not be temporarily released during join. |
| This lock will be used only by API calls (not by gst_adaptive_demux_stream_download_loop |
| or gst_adaptive_demux_updates_loop or _src_chain or _src_event) so it is safe |
| to hold it while joining the threads or changing the src element state. The |
| api_lock will serialise all external requests to adaptive demux. In order to |
| avoid deadlocks, if a function needs to acquire both manifest and api locks, |
| the api_lock will be taken first and the manifest_lock second. |
| |
| By using the api_lock a thread is protected against other API calls. But when |
| temporarily dropping the manifest_lock, it will be vulnerable to changes from |
| threads that use only the manifest_lock and not the api_lock. These threads run |
| one of the following functions: gst_adaptive_demux_stream_download_loop, |
| gst_adaptive_demux_updates_loop, _src_chain, _src_event. In order to guarantee |
| that all operations during an API call are not impacted by other writes, the |
| above mentioned functions must check a cancelled flag every time they reacquire |
| the manifest_lock. If the flag is set, they must exit immediately, without |
| performing any changes on the shared data. In this way, an API call (eg seek |
| request) can set the cancel flag before releasing the manifest_lock and be sure |
| that the demux object and its streams are not changed by anybody else. |
| */ |
| |
| #ifdef HAVE_CONFIG_H |
| #include "config.h" |
| #endif |
| |
| #include "gstadaptivedemux.h" |
| #include "gst/gst-i18n-plugin.h" |
| #include <gst/base/gstadapter.h> |
| |
| GST_DEBUG_CATEGORY (adaptivedemux_debug); |
| #define GST_CAT_DEFAULT adaptivedemux_debug |
| |
| #define GST_ADAPTIVE_DEMUX_GET_PRIVATE(obj) \ |
| (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_ADAPTIVE_DEMUX, \ |
| GstAdaptiveDemuxPrivate)) |
| |
| #define MAX_DOWNLOAD_ERROR_COUNT 3 |
| #define DEFAULT_FAILED_COUNT 3 |
| #define DEFAULT_CONNECTION_SPEED 0 |
| #define DEFAULT_BITRATE_LIMIT 0.8f |
| #define SRC_QUEUE_MAX_BYTES 20 * 1024 * 1024 /* For safety. Large enough to hold a segment. */ |
| #define NUM_LOOKBACK_FRAGMENTS 3 |
| |
| #define GST_MANIFEST_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->manifest_lock)) |
| #define GST_MANIFEST_LOCK(d) G_STMT_START { \ |
| GST_TRACE("Locking from thread %p", g_thread_self()); \ |
| g_rec_mutex_lock (GST_MANIFEST_GET_LOCK (d)); \ |
| GST_TRACE("Locked from thread %p", g_thread_self()); \ |
| } G_STMT_END |
| |
| #define GST_MANIFEST_UNLOCK(d) G_STMT_START { \ |
| GST_TRACE("Unlocking from thread %p", g_thread_self()); \ |
| g_rec_mutex_unlock (GST_MANIFEST_GET_LOCK (d)); \ |
| } G_STMT_END |
| |
| #define GST_API_GET_LOCK(d) (&(GST_ADAPTIVE_DEMUX_CAST(d)->priv->api_lock)) |
| #define GST_API_LOCK(d) g_mutex_lock (GST_API_GET_LOCK (d)); |
| #define GST_API_UNLOCK(d) g_mutex_unlock (GST_API_GET_LOCK (d)); |
| |
| #define GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK(d) (&GST_ADAPTIVE_DEMUX_CAST(d)->priv->segment_lock) |
| #define GST_ADAPTIVE_DEMUX_SEGMENT_LOCK(d) g_mutex_lock (GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK (d)) |
| #define GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK(d) g_mutex_unlock (GST_ADAPTIVE_DEMUX_SEGMENT_GET_LOCK (d)) |
| |
| enum |
| { |
| PROP_0, |
| PROP_CONNECTION_SPEED, |
| PROP_BITRATE_LIMIT, |
| PROP_LAST |
| }; |
| |
| /* Internal, so not using GST_FLOW_CUSTOM_SUCCESS_N */ |
| #define GST_ADAPTIVE_DEMUX_FLOW_SWITCH (GST_FLOW_CUSTOM_SUCCESS_2 + 1) |
| |
| struct _GstAdaptiveDemuxPrivate |
| { |
| GstAdapter *input_adapter; /* protected by manifest_lock */ |
| gboolean have_manifest; /* protected by manifest_lock */ |
| |
| GList *old_streams; /* protected by manifest_lock */ |
| |
| GstTask *updates_task; /* MT safe */ |
| GRecMutex updates_lock; |
| GMutex updates_timed_lock; |
| GCond updates_timed_cond; /* protected by updates_timed_lock */ |
| gboolean stop_updates_task; /* protected by updates_timed_lock */ |
| |
| /* used only from updates_task, no need to protect it */ |
| gint update_failed_count; |
| |
| guint32 segment_seqnum; /* protected by manifest_lock */ |
| |
| /* main lock used to protect adaptive demux and all its streams. |
| * It serializes the adaptive demux public API. |
| */ |
| GRecMutex manifest_lock; |
| |
| /* condition to wait for manifest updates on a live stream. |
| * In order to signal the manifest_cond, the caller needs to hold both |
| * manifest_lock and manifest_update_lock (taken in this order) |
| */ |
| GCond manifest_cond; |
| GMutex manifest_update_lock; |
| |
| /* Lock and condition for prerolling streams before exposing */ |
| GMutex preroll_lock; |
| GCond preroll_cond; |
| gint preroll_pending; |
| |
| GMutex api_lock; |
| |
| /* Protects demux and stream segment information |
| * Needed because seeks can update segment information |
| * without needing to stop tasks when they just want to |
| * update the segment boundaries */ |
| GMutex segment_lock; |
| }; |
| |
| typedef struct _GstAdaptiveDemuxTimer |
| { |
| volatile gint ref_count; |
| GCond *cond; |
| GMutex *mutex; |
| GstClockID clock_id; |
| gboolean fired; |
| } GstAdaptiveDemuxTimer; |
| |
| static GstBinClass *parent_class = NULL; |
| static void gst_adaptive_demux_class_init (GstAdaptiveDemuxClass * klass); |
| static void gst_adaptive_demux_init (GstAdaptiveDemux * dec, |
| GstAdaptiveDemuxClass * klass); |
| static void gst_adaptive_demux_finalize (GObject * object); |
| static GstStateChangeReturn gst_adaptive_demux_change_state (GstElement * |
| element, GstStateChange transition); |
| |
| static void gst_adaptive_demux_handle_message (GstBin * bin, GstMessage * msg); |
| |
| static gboolean gst_adaptive_demux_sink_event (GstPad * pad, GstObject * parent, |
| GstEvent * event); |
| static GstFlowReturn gst_adaptive_demux_sink_chain (GstPad * pad, |
| GstObject * parent, GstBuffer * buffer); |
| static gboolean gst_adaptive_demux_src_query (GstPad * pad, GstObject * parent, |
| GstQuery * query); |
| static gboolean gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent, |
| GstEvent * event); |
| |
| static gboolean |
| gst_adaptive_demux_push_src_event (GstAdaptiveDemux * demux, GstEvent * event); |
| |
| static void gst_adaptive_demux_updates_loop (GstAdaptiveDemux * demux); |
| static void gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream * |
| stream); |
| static void gst_adaptive_demux_reset (GstAdaptiveDemux * demux); |
| static gboolean gst_adaptive_demux_prepare_streams (GstAdaptiveDemux * demux, |
| gboolean first_and_live); |
| static gboolean gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux); |
| static gboolean gst_adaptive_demux_is_live (GstAdaptiveDemux * demux); |
| static GstFlowReturn gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream, gboolean forward, GstSeekFlags flags, |
| GstClockTime ts, GstClockTime * final_ts); |
| static gboolean gst_adaptive_demux_stream_has_next_fragment (GstAdaptiveDemux * |
| demux, GstAdaptiveDemuxStream * stream); |
| static gboolean gst_adaptive_demux_stream_select_bitrate (GstAdaptiveDemux * |
| demux, GstAdaptiveDemuxStream * stream, guint64 bitrate); |
| static GstFlowReturn |
| gst_adaptive_demux_stream_update_fragment_info (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream); |
| static gint64 |
| gst_adaptive_demux_stream_get_fragment_waiting_time (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream); |
| static GstFlowReturn gst_adaptive_demux_update_manifest (GstAdaptiveDemux * |
| demux); |
| static GstFlowReturn |
| gst_adaptive_demux_update_manifest_default (GstAdaptiveDemux * demux); |
| static gboolean gst_adaptive_demux_has_next_period (GstAdaptiveDemux * demux); |
| static void gst_adaptive_demux_advance_period (GstAdaptiveDemux * demux); |
| |
| static void gst_adaptive_demux_stream_free (GstAdaptiveDemuxStream * stream); |
| static GstFlowReturn |
| gst_adaptive_demux_stream_push_event (GstAdaptiveDemuxStream * stream, |
| GstEvent * event); |
| |
| static void gst_adaptive_demux_stop_manifest_update_task (GstAdaptiveDemux * |
| demux); |
| static void gst_adaptive_demux_start_manifest_update_task (GstAdaptiveDemux * |
| demux); |
| |
| static void gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux, |
| gboolean start_preroll_streams); |
| static void gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux, |
| gboolean stop_updates); |
| static GstFlowReturn gst_adaptive_demux_combine_flows (GstAdaptiveDemux * |
| demux); |
| static void |
| gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream * |
| stream, GstFlowReturn ret, GError * err); |
| static GstFlowReturn |
| gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream, GstBuffer * buffer); |
| static GstFlowReturn |
| gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream); |
| static GstFlowReturn |
| gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream, GstClockTime duration); |
| static gboolean |
| gst_adaptive_demux_wait_until (GstClock * clock, GCond * cond, GMutex * mutex, |
| GstClockTime end_time); |
| static gboolean gst_adaptive_demux_clock_callback (GstClock * clock, |
| GstClockTime time, GstClockID id, gpointer user_data); |
| static gboolean |
| gst_adaptive_demux_requires_periodical_playlist_update_default (GstAdaptiveDemux |
| * demux); |
| |
| /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init |
| * method to get to the padtemplates */ |
| GType |
| gst_adaptive_demux_get_type (void) |
| { |
| static volatile gsize type = 0; |
| |
| if (g_once_init_enter (&type)) { |
| GType _type; |
| static const GTypeInfo info = { |
| sizeof (GstAdaptiveDemuxClass), |
| NULL, |
| NULL, |
| (GClassInitFunc) gst_adaptive_demux_class_init, |
| NULL, |
| NULL, |
| sizeof (GstAdaptiveDemux), |
| 0, |
| (GInstanceInitFunc) gst_adaptive_demux_init, |
| }; |
| |
| _type = g_type_register_static (GST_TYPE_BIN, |
| "GstAdaptiveDemux", &info, G_TYPE_FLAG_ABSTRACT); |
| g_once_init_leave (&type, _type); |
| } |
| return type; |
| } |
| |
| static void |
| gst_adaptive_demux_set_property (GObject * object, guint prop_id, |
| const GValue * value, GParamSpec * pspec) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object); |
| |
| GST_API_LOCK (demux); |
| GST_MANIFEST_LOCK (demux); |
| |
| switch (prop_id) { |
| case PROP_CONNECTION_SPEED: |
| demux->connection_speed = g_value_get_uint (value) * 1000; |
| GST_DEBUG_OBJECT (demux, "Connection speed set to %u", |
| demux->connection_speed); |
| break; |
| case PROP_BITRATE_LIMIT: |
| demux->bitrate_limit = g_value_get_float (value); |
| break; |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| break; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| } |
| |
| static void |
| gst_adaptive_demux_get_property (GObject * object, guint prop_id, |
| GValue * value, GParamSpec * pspec) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX (object); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| switch (prop_id) { |
| case PROP_CONNECTION_SPEED: |
| g_value_set_uint (value, demux->connection_speed / 1000); |
| break; |
| case PROP_BITRATE_LIMIT: |
| g_value_set_float (value, demux->bitrate_limit); |
| break; |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| break; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| } |
| |
| static void |
| gst_adaptive_demux_class_init (GstAdaptiveDemuxClass * klass) |
| { |
| GObjectClass *gobject_class; |
| GstElementClass *gstelement_class; |
| GstBinClass *gstbin_class; |
| |
| gobject_class = G_OBJECT_CLASS (klass); |
| gstelement_class = GST_ELEMENT_CLASS (klass); |
| gstbin_class = GST_BIN_CLASS (klass); |
| |
| GST_DEBUG_CATEGORY_INIT (adaptivedemux_debug, "adaptivedemux", 0, |
| "Base Adaptive Demux"); |
| |
| parent_class = g_type_class_peek_parent (klass); |
| g_type_class_add_private (klass, sizeof (GstAdaptiveDemuxPrivate)); |
| |
| gobject_class->set_property = gst_adaptive_demux_set_property; |
| gobject_class->get_property = gst_adaptive_demux_get_property; |
| gobject_class->finalize = gst_adaptive_demux_finalize; |
| |
| g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED, |
| g_param_spec_uint ("connection-speed", "Connection Speed", |
| "Network connection speed in kbps (0 = calculate from downloaded" |
| " fragments)", 0, G_MAXUINT / 1000, DEFAULT_CONNECTION_SPEED, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| /* FIXME 2.0: rename this property to bandwidth-usage or any better name */ |
| g_object_class_install_property (gobject_class, PROP_BITRATE_LIMIT, |
| g_param_spec_float ("bitrate-limit", |
| "Bitrate limit in %", |
| "Limit of the available bitrate to use when switching to alternates.", |
| 0, 1, DEFAULT_BITRATE_LIMIT, |
| G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| |
| gstelement_class->change_state = gst_adaptive_demux_change_state; |
| |
| gstbin_class->handle_message = gst_adaptive_demux_handle_message; |
| |
| klass->data_received = gst_adaptive_demux_stream_data_received_default; |
| klass->finish_fragment = gst_adaptive_demux_stream_finish_fragment_default; |
| klass->update_manifest = gst_adaptive_demux_update_manifest_default; |
| klass->requires_periodical_playlist_update = |
| gst_adaptive_demux_requires_periodical_playlist_update_default; |
| |
| } |
| |
| static void |
| gst_adaptive_demux_init (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxClass * klass) |
| { |
| GstPadTemplate *pad_template; |
| GstClockType clock_type = GST_CLOCK_TYPE_OTHER; |
| GObjectClass *gobject_class; |
| |
| GST_DEBUG_OBJECT (demux, "gst_adaptive_demux_init"); |
| |
| demux->priv = GST_ADAPTIVE_DEMUX_GET_PRIVATE (demux); |
| demux->priv->input_adapter = gst_adapter_new (); |
| demux->downloader = gst_uri_downloader_new (); |
| gst_uri_downloader_set_parent (demux->downloader, GST_ELEMENT_CAST (demux)); |
| demux->stream_struct_size = sizeof (GstAdaptiveDemuxStream); |
| demux->priv->segment_seqnum = gst_util_seqnum_next (); |
| demux->have_group_id = FALSE; |
| demux->group_id = G_MAXUINT; |
| |
| gst_segment_init (&demux->segment, GST_FORMAT_TIME); |
| |
| gst_bin_set_suppressed_flags (GST_BIN_CAST (demux), |
| GST_ELEMENT_FLAG_SOURCE | GST_ELEMENT_FLAG_SINK); |
| |
| demux->realtime_clock = gst_system_clock_obtain (); |
| g_assert (demux->realtime_clock != NULL); |
| gobject_class = G_OBJECT_GET_CLASS (demux->realtime_clock); |
| if (g_object_class_find_property (gobject_class, "clock-type")) { |
| g_object_get (demux->realtime_clock, "clock-type", &clock_type, NULL); |
| } else { |
| GST_WARNING_OBJECT (demux, |
| "System clock does not have clock-type property"); |
| } |
| if (clock_type == GST_CLOCK_TYPE_REALTIME) { |
| demux->clock_offset = 0; |
| } else { |
| GDateTime *utc_now; |
| GstClockTime rtc_now; |
| GTimeVal gtv; |
| |
| utc_now = g_date_time_new_now_utc (); |
| rtc_now = gst_clock_get_time (demux->realtime_clock); |
| g_date_time_to_timeval (utc_now, >v); |
| demux->clock_offset = |
| gtv.tv_sec * G_TIME_SPAN_SECOND + gtv.tv_usec - |
| GST_TIME_AS_USECONDS (rtc_now); |
| g_date_time_unref (utc_now); |
| } |
| g_rec_mutex_init (&demux->priv->updates_lock); |
| demux->priv->updates_task = |
| gst_task_new ((GstTaskFunction) gst_adaptive_demux_updates_loop, |
| demux, NULL); |
| gst_task_set_lock (demux->priv->updates_task, &demux->priv->updates_lock); |
| |
| g_mutex_init (&demux->priv->updates_timed_lock); |
| g_cond_init (&demux->priv->updates_timed_cond); |
| |
| g_cond_init (&demux->priv->manifest_cond); |
| g_mutex_init (&demux->priv->manifest_update_lock); |
| |
| g_rec_mutex_init (&demux->priv->manifest_lock); |
| g_mutex_init (&demux->priv->api_lock); |
| g_mutex_init (&demux->priv->segment_lock); |
| |
| g_cond_init (&demux->priv->preroll_cond); |
| g_mutex_init (&demux->priv->preroll_lock); |
| |
| pad_template = |
| gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "sink"); |
| g_return_if_fail (pad_template != NULL); |
| |
| demux->sinkpad = gst_pad_new_from_template (pad_template, "sink"); |
| gst_pad_set_event_function (demux->sinkpad, |
| GST_DEBUG_FUNCPTR (gst_adaptive_demux_sink_event)); |
| gst_pad_set_chain_function (demux->sinkpad, |
| GST_DEBUG_FUNCPTR (gst_adaptive_demux_sink_chain)); |
| |
| /* Properties */ |
| demux->bitrate_limit = DEFAULT_BITRATE_LIMIT; |
| demux->connection_speed = DEFAULT_CONNECTION_SPEED; |
| |
| gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad); |
| } |
| |
| static void |
| gst_adaptive_demux_finalize (GObject * object) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (object); |
| GstAdaptiveDemuxPrivate *priv = demux->priv; |
| |
| GST_DEBUG_OBJECT (object, "finalize"); |
| |
| g_object_unref (priv->input_adapter); |
| g_object_unref (demux->downloader); |
| |
| g_mutex_clear (&priv->updates_timed_lock); |
| g_cond_clear (&priv->updates_timed_cond); |
| g_mutex_clear (&demux->priv->manifest_update_lock); |
| g_cond_clear (&demux->priv->manifest_cond); |
| g_object_unref (priv->updates_task); |
| g_rec_mutex_clear (&priv->updates_lock); |
| g_rec_mutex_clear (&demux->priv->manifest_lock); |
| g_mutex_clear (&demux->priv->api_lock); |
| g_mutex_clear (&demux->priv->segment_lock); |
| if (demux->realtime_clock) { |
| gst_object_unref (demux->realtime_clock); |
| demux->realtime_clock = NULL; |
| } |
| |
| g_cond_clear (&demux->priv->preroll_cond); |
| g_mutex_clear (&demux->priv->preroll_lock); |
| |
| G_OBJECT_CLASS (parent_class)->finalize (object); |
| } |
| |
| static GstStateChangeReturn |
| gst_adaptive_demux_change_state (GstElement * element, |
| GstStateChange transition) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (element); |
| GstStateChangeReturn result = GST_STATE_CHANGE_FAILURE; |
| |
| GST_API_LOCK (demux); |
| |
| switch (transition) { |
| case GST_STATE_CHANGE_PAUSED_TO_READY: |
| GST_MANIFEST_LOCK (demux); |
| demux->running = FALSE; |
| gst_adaptive_demux_reset (demux); |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| case GST_STATE_CHANGE_READY_TO_PAUSED: |
| GST_MANIFEST_LOCK (demux); |
| gst_adaptive_demux_reset (demux); |
| /* Clear "cancelled" flag in uridownloader since subclass might want to |
| * use uridownloader to fetch another manifest */ |
| gst_uri_downloader_reset (demux->downloader); |
| if (demux->priv->have_manifest) |
| gst_adaptive_demux_start_manifest_update_task (demux); |
| demux->running = TRUE; |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| default: |
| break; |
| } |
| |
| /* this must be run without MANIFEST_LOCK taken. |
| * For PLAYING to PLAYING state changes, it will want to take a lock in |
| * src element and that lock is held while the streaming thread is running. |
| * The streaming thread will take the MANIFEST_LOCK, leading to a deadlock. |
| */ |
| result = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); |
| |
| GST_API_UNLOCK (demux); |
| return result; |
| } |
| |
| static gboolean |
| gst_adaptive_demux_sink_event (GstPad * pad, GstObject * parent, |
| GstEvent * event) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent); |
| gboolean ret; |
| |
| switch (event->type) { |
| case GST_EVENT_FLUSH_STOP:{ |
| GST_API_LOCK (demux); |
| GST_MANIFEST_LOCK (demux); |
| |
| gst_adaptive_demux_reset (demux); |
| |
| ret = gst_pad_event_default (pad, parent, event); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| |
| return ret; |
| } |
| case GST_EVENT_EOS:{ |
| GstAdaptiveDemuxClass *demux_class; |
| GstQuery *query; |
| gboolean query_res; |
| gboolean ret = TRUE; |
| gsize available; |
| GstBuffer *manifest_buffer; |
| |
| GST_API_LOCK (demux); |
| GST_MANIFEST_LOCK (demux); |
| |
| demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| available = gst_adapter_available (demux->priv->input_adapter); |
| |
| if (available == 0) { |
| GST_WARNING_OBJECT (demux, "Received EOS without a manifest."); |
| ret = gst_pad_event_default (pad, parent, event); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| |
| return ret; |
| } |
| |
| GST_DEBUG_OBJECT (demux, "Got EOS on the sink pad: manifest fetched"); |
| |
| /* Need to get the URI to use it as a base to generate the fragment's |
| * uris */ |
| query = gst_query_new_uri (); |
| query_res = gst_pad_peer_query (pad, query); |
| if (query_res) { |
| gchar *uri, *redirect_uri; |
| gboolean permanent; |
| |
| gst_query_parse_uri (query, &uri); |
| gst_query_parse_uri_redirection (query, &redirect_uri); |
| gst_query_parse_uri_redirection_permanent (query, &permanent); |
| |
| if (permanent && redirect_uri) { |
| demux->manifest_uri = redirect_uri; |
| demux->manifest_base_uri = NULL; |
| g_free (uri); |
| } else { |
| demux->manifest_uri = uri; |
| demux->manifest_base_uri = redirect_uri; |
| } |
| |
| GST_DEBUG_OBJECT (demux, "Fetched manifest at URI: %s (base: %s)", |
| demux->manifest_uri, GST_STR_NULL (demux->manifest_base_uri)); |
| } else { |
| GST_WARNING_OBJECT (demux, "Upstream URI query failed."); |
| } |
| gst_query_unref (query); |
| |
| /* Let the subclass parse the manifest */ |
| manifest_buffer = |
| gst_adapter_take_buffer (demux->priv->input_adapter, available); |
| if (!demux_class->process_manifest (demux, manifest_buffer)) { |
| /* In most cases, this will happen if we set a wrong url in the |
| * source element and we have received the 404 HTML response instead of |
| * the manifest */ |
| GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid manifest."), |
| (NULL)); |
| ret = FALSE; |
| } else { |
| demux->priv->have_manifest = TRUE; |
| } |
| gst_buffer_unref (manifest_buffer); |
| |
| gst_element_post_message (GST_ELEMENT_CAST (demux), |
| gst_message_new_element (GST_OBJECT_CAST (demux), |
| gst_structure_new (GST_ADAPTIVE_DEMUX_STATISTICS_MESSAGE_NAME, |
| "manifest-uri", G_TYPE_STRING, |
| demux->manifest_uri, "uri", G_TYPE_STRING, |
| demux->manifest_uri, |
| "manifest-download-start", GST_TYPE_CLOCK_TIME, |
| GST_CLOCK_TIME_NONE, |
| "manifest-download-stop", GST_TYPE_CLOCK_TIME, |
| gst_util_get_timestamp (), NULL))); |
| |
| if (ret) { |
| /* Send duration message */ |
| if (!gst_adaptive_demux_is_live (demux)) { |
| GstClockTime duration = demux_class->get_duration (demux); |
| |
| if (duration != GST_CLOCK_TIME_NONE) { |
| GST_DEBUG_OBJECT (demux, |
| "Sending duration message : %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (duration)); |
| gst_element_post_message (GST_ELEMENT (demux), |
| gst_message_new_duration_changed (GST_OBJECT (demux))); |
| } else { |
| GST_DEBUG_OBJECT (demux, |
| "media duration unknown, can not send the duration message"); |
| } |
| } |
| |
| if (demux->next_streams) { |
| gst_adaptive_demux_prepare_streams (demux, |
| gst_adaptive_demux_is_live (demux)); |
| gst_adaptive_demux_start_tasks (demux, TRUE); |
| gst_adaptive_demux_start_manifest_update_task (demux); |
| } else { |
| /* no streams */ |
| GST_WARNING_OBJECT (demux, "No streams created from manifest"); |
| GST_ELEMENT_ERROR (demux, STREAM, DEMUX, |
| (_("This file contains no playable streams.")), |
| ("No known stream formats found at the Manifest")); |
| ret = FALSE; |
| } |
| |
| } |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| |
| gst_event_unref (event); |
| return ret; |
| } |
| case GST_EVENT_SEGMENT: |
| /* Swallow newsegments, we'll push our own */ |
| gst_event_unref (event); |
| return TRUE; |
| default: |
| break; |
| } |
| |
| return gst_pad_event_default (pad, parent, event); |
| } |
| |
| static GstFlowReturn |
| gst_adaptive_demux_sink_chain (GstPad * pad, GstObject * parent, |
| GstBuffer * buffer) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| gst_adapter_push (demux->priv->input_adapter, buffer); |
| |
| GST_INFO_OBJECT (demux, "Received manifest buffer, total size is %i bytes", |
| (gint) gst_adapter_available (demux->priv->input_adapter)); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| return GST_FLOW_OK; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static void |
| gst_adaptive_demux_reset (GstAdaptiveDemux * demux) |
| { |
| GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| GList *iter; |
| GList *old_streams; |
| GstEvent *eos; |
| |
| /* take ownership of old_streams before releasing the manifest_lock in |
| * gst_adaptive_demux_stop_tasks |
| */ |
| old_streams = demux->priv->old_streams; |
| demux->priv->old_streams = NULL; |
| |
| gst_adaptive_demux_stop_tasks (demux, TRUE); |
| |
| if (klass->reset) |
| klass->reset (demux); |
| |
| eos = gst_event_new_eos (); |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| if (stream->pad) { |
| gst_pad_push_event (stream->pad, gst_event_ref (eos)); |
| gst_pad_set_active (stream->pad, FALSE); |
| |
| gst_element_remove_pad (GST_ELEMENT_CAST (demux), stream->pad); |
| } |
| gst_adaptive_demux_stream_free (stream); |
| } |
| gst_event_unref (eos); |
| g_list_free (demux->streams); |
| demux->streams = NULL; |
| if (demux->prepared_streams) { |
| g_list_free_full (demux->prepared_streams, |
| (GDestroyNotify) gst_adaptive_demux_stream_free); |
| demux->prepared_streams = NULL; |
| } |
| if (demux->next_streams) { |
| g_list_free_full (demux->next_streams, |
| (GDestroyNotify) gst_adaptive_demux_stream_free); |
| demux->next_streams = NULL; |
| } |
| |
| if (old_streams) { |
| g_list_free_full (old_streams, |
| (GDestroyNotify) gst_adaptive_demux_stream_free); |
| } |
| |
| if (demux->priv->old_streams) { |
| g_list_free_full (demux->priv->old_streams, |
| (GDestroyNotify) gst_adaptive_demux_stream_free); |
| demux->priv->old_streams = NULL; |
| } |
| |
| g_free (demux->manifest_uri); |
| g_free (demux->manifest_base_uri); |
| demux->manifest_uri = NULL; |
| demux->manifest_base_uri = NULL; |
| |
| gst_adapter_clear (demux->priv->input_adapter); |
| demux->priv->have_manifest = FALSE; |
| |
| gst_segment_init (&demux->segment, GST_FORMAT_TIME); |
| |
| demux->have_group_id = FALSE; |
| demux->group_id = G_MAXUINT; |
| demux->priv->segment_seqnum = gst_util_seqnum_next (); |
| } |
| |
| static void |
| gst_adaptive_demux_handle_message (GstBin * bin, GstMessage * msg) |
| { |
| GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (bin); |
| |
| switch (GST_MESSAGE_TYPE (msg)) { |
| case GST_MESSAGE_ERROR:{ |
| GList *iter; |
| GstAdaptiveDemuxStream *stream = NULL; |
| GError *err = NULL; |
| gchar *debug = NULL; |
| gchar *new_error = NULL; |
| const GstStructure *details = NULL; |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *cur = iter->data; |
| if (gst_object_has_as_ancestor (GST_MESSAGE_SRC (msg), |
| GST_OBJECT_CAST (cur->src))) { |
| stream = cur; |
| break; |
| } |
| } |
| if (stream == NULL) { |
| for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *cur = iter->data; |
| if (gst_object_has_as_ancestor (GST_MESSAGE_SRC (msg), |
| GST_OBJECT_CAST (cur->src))) { |
| stream = cur; |
| break; |
| } |
| } |
| if (stream == NULL) { |
| GST_WARNING_OBJECT (demux, |
| "Failed to locate stream for errored element"); |
| break; |
| } |
| } |
| |
| gst_message_parse_error (msg, &err, &debug); |
| |
| GST_WARNING_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (stream), |
| "Source posted error: %d:%d %s (%s)", err->domain, err->code, |
| err->message, debug); |
| |
| if (debug) |
| new_error = g_strdup_printf ("%s: %s\n", err->message, debug); |
| if (new_error) { |
| g_free (err->message); |
| err->message = new_error; |
| } |
| |
| gst_message_parse_error_details (msg, &details); |
| if (details) { |
| gst_structure_get_uint (details, "http-status-code", |
| &stream->last_status_code); |
| } |
| |
| /* error, but ask to retry */ |
| gst_adaptive_demux_stream_fragment_download_finish (stream, |
| GST_FLOW_CUSTOM_ERROR, err); |
| |
| g_error_free (err); |
| g_free (debug); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| gst_message_unref (msg); |
| msg = NULL; |
| } |
| break; |
| default: |
| break; |
| } |
| |
| if (msg) |
| GST_BIN_CLASS (parent_class)->handle_message (bin, msg); |
| } |
| |
| void |
| gst_adaptive_demux_set_stream_struct_size (GstAdaptiveDemux * demux, |
| gsize struct_size) |
| { |
| GST_API_LOCK (demux); |
| GST_MANIFEST_LOCK (demux); |
| demux->stream_struct_size = struct_size; |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_prepare_stream (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| GstPad *pad = stream->pad; |
| gchar *name = gst_pad_get_name (pad); |
| GstEvent *event; |
| gchar *stream_id; |
| |
| gst_pad_set_active (pad, TRUE); |
| stream->need_header = TRUE; |
| |
| stream_id = gst_pad_create_stream_id (pad, GST_ELEMENT_CAST (demux), name); |
| |
| event = |
| gst_pad_get_sticky_event (GST_ADAPTIVE_DEMUX_SINK_PAD (demux), |
| GST_EVENT_STREAM_START, 0); |
| if (event) { |
| if (gst_event_parse_group_id (event, &demux->group_id)) |
| demux->have_group_id = TRUE; |
| else |
| demux->have_group_id = FALSE; |
| gst_event_unref (event); |
| } else if (!demux->have_group_id) { |
| demux->have_group_id = TRUE; |
| demux->group_id = gst_util_group_id_next (); |
| } |
| event = gst_event_new_stream_start (stream_id); |
| if (demux->have_group_id) |
| gst_event_set_group_id (event, demux->group_id); |
| |
| gst_pad_push_event (pad, event); |
| g_free (stream_id); |
| g_free (name); |
| |
| GST_DEBUG_OBJECT (demux, "Preparing srcpad %s:%s", GST_DEBUG_PAD_NAME (pad)); |
| |
| stream->discont = TRUE; |
| |
| return TRUE; |
| } |
| |
| static gboolean |
| gst_adaptive_demux_expose_stream (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| gboolean ret; |
| GstPad *pad = stream->pad; |
| GstCaps *caps; |
| |
| if (stream->pending_caps) { |
| gst_pad_set_caps (pad, stream->pending_caps); |
| caps = stream->pending_caps; |
| stream->pending_caps = NULL; |
| } else { |
| caps = gst_pad_get_current_caps (pad); |
| } |
| |
| GST_DEBUG_OBJECT (demux, "Exposing srcpad %s:%s with caps %" GST_PTR_FORMAT, |
| GST_DEBUG_PAD_NAME (pad), caps); |
| if (caps) |
| gst_caps_unref (caps); |
| |
| gst_object_ref (pad); |
| |
| /* Don't hold the manifest lock while exposing a pad */ |
| GST_MANIFEST_UNLOCK (demux); |
| ret = gst_element_add_pad (GST_ELEMENT_CAST (demux), pad); |
| GST_MANIFEST_LOCK (demux); |
| |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static GstClockTime |
| gst_adaptive_demux_stream_get_presentation_offset (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| GstAdaptiveDemuxClass *klass; |
| |
| klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| if (klass->get_presentation_offset == NULL) |
| return 0; |
| |
| return klass->get_presentation_offset (demux, stream); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static GstClockTime |
| gst_adaptive_demux_get_period_start_time (GstAdaptiveDemux * demux) |
| { |
| GstAdaptiveDemuxClass *klass; |
| |
| klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| if (klass->get_period_start_time == NULL) |
| return 0; |
| |
| return klass->get_period_start_time (demux); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_prepare_streams (GstAdaptiveDemux * demux, |
| gboolean first_and_live) |
| { |
| GList *iter; |
| GstClockTime period_start, min_pts = GST_CLOCK_TIME_NONE; |
| |
| g_return_val_if_fail (demux->next_streams != NULL, FALSE); |
| if (demux->prepared_streams != NULL) { |
| /* Old streams that were never exposed, due to a seek or so */ |
| GST_FIXME_OBJECT (demux, |
| "Preparing new streams without cleaning up old ones!"); |
| return FALSE; |
| } |
| |
| demux->prepared_streams = demux->next_streams; |
| demux->next_streams = NULL; |
| |
| if (!demux->running) { |
| GST_DEBUG_OBJECT (demux, "Not exposing pads due to shutdown"); |
| return TRUE; |
| } |
| |
| for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| stream->do_block = TRUE; |
| |
| if (!gst_adaptive_demux_prepare_stream (demux, |
| GST_ADAPTIVE_DEMUX_STREAM_CAST (stream))) { |
| /* TODO act on error */ |
| GST_FIXME_OBJECT (stream->pad, |
| "Do something on failure to expose stream"); |
| } |
| |
| if (first_and_live) { |
| /* TODO we only need the first timestamp, maybe create a simple function to |
| * get the current PTS of a fragment ? */ |
| GST_DEBUG_OBJECT (demux, "Calling update_fragment_info"); |
| gst_adaptive_demux_stream_update_fragment_info (demux, stream); |
| |
| if (GST_CLOCK_TIME_IS_VALID (min_pts)) { |
| min_pts = MIN (min_pts, stream->fragment.timestamp); |
| } else { |
| min_pts = stream->fragment.timestamp; |
| } |
| } |
| } |
| |
| period_start = gst_adaptive_demux_get_period_start_time (demux); |
| |
| /* For live streams, the subclass is supposed to seek to the current |
| * fragment and then tell us its timestamp in stream->fragment.timestamp. |
| * We now also have to seek our demuxer segment to reflect this. |
| * |
| * FIXME: This needs some refactoring at some point. |
| */ |
| if (first_and_live) { |
| gst_segment_do_seek (&demux->segment, demux->segment.rate, GST_FORMAT_TIME, |
| GST_SEEK_FLAG_FLUSH, GST_SEEK_TYPE_SET, min_pts + period_start, |
| GST_SEEK_TYPE_NONE, -1, NULL); |
| } |
| |
| for (iter = demux->prepared_streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| GstClockTime offset; |
| |
| offset = gst_adaptive_demux_stream_get_presentation_offset (demux, stream); |
| stream->segment = demux->segment; |
| |
| /* The demuxer segment is just built from seek events, but for each stream |
| * we have to adjust segments according to the current period and the |
| * stream specific presentation time offset. |
| * |
| * For each period, buffer timestamps start again from 0. Additionally the |
| * buffer timestamps are shifted by the stream specific presentation time |
| * offset, so the first buffer timestamp of a period is 0 + presentation |
| * time offset. If the stream contains timestamps itself, this is also |
| * supposed to be the presentation time stored inside the stream. |
| * |
| * The stream time over periods is supposed to be continuous, that is the |
| * buffer timestamp 0 + presentation time offset should map to the start |
| * time of the current period. |
| * |
| * |
| * The adjustment of the stream segments as such works the following. |
| * |
| * If the demuxer segment start is bigger than the period start, this |
| * means that we have to drop some media at the beginning of the current |
| * period, e.g. because a seek into the middle of the period has |
| * happened. The amount of media to drop is the difference between the |
| * period start and the demuxer segment start, and as each period starts |
| * again from 0, this difference is going to be the actual stream's |
| * segment start. As all timestamps of the stream are shifted by the |
| * presentation time offset, we will also have to move the segment start |
| * by that offset. |
| * |
| * Likewise, the demuxer segment stop value is adjusted in the same |
| * fashion. |
| * |
| * Now the running time and stream time at the stream's segment start has |
| * to be the one that is stored inside the demuxer's segment, which means |
| * that segment.base and segment.time have to be copied over (done just |
| * above) |
| * |
| * |
| * If the demuxer segment start is smaller than the period start time, |
| * this means that the whole period is inside the segment. As each period |
| * starts timestamps from 0, and additionally timestamps are shifted by |
| * the presentation time offset, the stream's first timestamp (and as such |
| * the stream's segment start) has to be the presentation time offset. |
| * The stream time at the segment start is supposed to be the stream time |
| * of the period start according to the demuxer segment, so the stream |
| * segment's time would be set to that. The same goes for the stream |
| * segment's base, which is supposed to be the running time of the period |
| * start according to the demuxer's segment. |
| * |
| * The same logic applies for negative rates with the segment stop and |
| * the period stop time (which gets clamped). |
| * |
| * |
| * For the first case where not the complete period is inside the segment, |
| * the segment time and base as calculated by the second case would be |
| * equivalent. |
| */ |
| GST_DEBUG_OBJECT (demux, "Using demux segment %" GST_SEGMENT_FORMAT, |
| &demux->segment); |
| GST_DEBUG_OBJECT (demux, |
| "period_start: %" GST_TIME_FORMAT " offset: %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (period_start), GST_TIME_ARGS (offset)); |
| /* note for readers: |
| * Since stream->segment is initially a copy of demux->segment, |
| * only the values that need updating are modified below. */ |
| if (first_and_live) { |
| /* If first and live, demuxer did seek to the current position already */ |
| stream->segment.start = demux->segment.start - period_start + offset; |
| if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)) |
| stream->segment.stop = demux->segment.stop - period_start + offset; |
| /* FIXME : Do we need to handle negative rates for this ? */ |
| stream->segment.position = stream->segment.start; |
| } else if (demux->segment.start > period_start) { |
| /* seek within a period */ |
| stream->segment.start = demux->segment.start - period_start + offset; |
| if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)) |
| stream->segment.stop = demux->segment.stop - period_start + offset; |
| if (stream->segment.rate >= 0) |
| stream->segment.position = offset; |
| else |
| stream->segment.position = stream->segment.stop; |
| } else { |
| stream->segment.start = offset; |
| if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)) |
| stream->segment.stop = demux->segment.stop - period_start + offset; |
| if (stream->segment.rate >= 0) |
| stream->segment.position = offset; |
| else |
| stream->segment.position = stream->segment.stop; |
| stream->segment.time = |
| gst_segment_to_stream_time (&demux->segment, GST_FORMAT_TIME, |
| period_start); |
| stream->segment.base = |
| gst_segment_to_running_time (&demux->segment, GST_FORMAT_TIME, |
| period_start); |
| } |
| |
| stream->pending_segment = gst_event_new_segment (&stream->segment); |
| gst_event_set_seqnum (stream->pending_segment, demux->priv->segment_seqnum); |
| stream->qos_earliest_time = GST_CLOCK_TIME_NONE; |
| |
| GST_DEBUG_OBJECT (demux, |
| "Prepared segment %" GST_SEGMENT_FORMAT " for stream %p", |
| &stream->segment, stream); |
| } |
| |
| return TRUE; |
| } |
| |
| static gboolean |
| gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux) |
| { |
| GList *iter; |
| GList *old_streams; |
| |
| g_return_val_if_fail (demux->prepared_streams != NULL, FALSE); |
| |
| old_streams = demux->streams; |
| demux->streams = demux->prepared_streams; |
| demux->prepared_streams = NULL; |
| |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| if (!gst_adaptive_demux_expose_stream (demux, |
| GST_ADAPTIVE_DEMUX_STREAM_CAST (stream))) { |
| /* TODO act on error */ |
| } |
| } |
| demux->priv->preroll_pending = 0; |
| |
| GST_MANIFEST_UNLOCK (demux); |
| gst_element_no_more_pads (GST_ELEMENT_CAST (demux)); |
| GST_MANIFEST_LOCK (demux); |
| |
| if (old_streams) { |
| GstEvent *eos = gst_event_new_eos (); |
| |
| /* before we put streams in the demux->priv->old_streams list, |
| * we ask the download task to stop. In this way, it will no longer be |
| * allowed to change the demux object. |
| */ |
| for (iter = old_streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| GstPad *pad = gst_object_ref (GST_PAD (stream->pad)); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| GST_DEBUG_OBJECT (pad, "Pushing EOS"); |
| gst_pad_push_event (pad, gst_event_ref (eos)); |
| gst_pad_set_active (pad, FALSE); |
| |
| GST_LOG_OBJECT (pad, "Removing stream"); |
| gst_element_remove_pad (GST_ELEMENT (demux), pad); |
| GST_MANIFEST_LOCK (demux); |
| |
| gst_object_unref (GST_OBJECT (pad)); |
| |
| /* ask the download task to stop. |
| * We will not join it now, because our thread can be one of these tasks. |
| * We will do the joining later, from another stream download task or |
| * from gst_adaptive_demux_stop_tasks. |
| * We also cannot change the state of the stream->src element, because |
| * that will wait on the streaming thread (which could be this thread) |
| * to stop first. |
| * Because we sent an EOS to the downstream element, the stream->src |
| * element should detect this in its streaming task and stop. |
| * Even if it doesn't do that, we will change its state later in |
| * gst_adaptive_demux_stop_tasks. |
| */ |
| GST_LOG_OBJECT (stream, "Marking stream as cancelled"); |
| gst_task_stop (stream->download_task); |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->cancelled = TRUE; |
| stream->replaced = TRUE; |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| gst_event_unref (eos); |
| |
| /* The list should be freed from another thread as we can't properly |
| * cleanup a GstTask from itself */ |
| demux->priv->old_streams = |
| g_list_concat (demux->priv->old_streams, old_streams); |
| } |
| |
| /* Unblock after removing oldstreams */ |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| stream->do_block = FALSE; |
| } |
| |
| GST_DEBUG_OBJECT (demux, "All streams are exposed"); |
| |
| return TRUE; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| GstAdaptiveDemuxStream * |
| gst_adaptive_demux_stream_new (GstAdaptiveDemux * demux, GstPad * pad) |
| { |
| GstAdaptiveDemuxStream *stream; |
| |
| stream = g_malloc0 (demux->stream_struct_size); |
| |
| /* Downloading task */ |
| g_rec_mutex_init (&stream->download_lock); |
| stream->download_task = |
| gst_task_new ((GstTaskFunction) gst_adaptive_demux_stream_download_loop, |
| stream, NULL); |
| gst_task_set_lock (stream->download_task, &stream->download_lock); |
| |
| stream->pad = pad; |
| stream->demux = demux; |
| stream->fragment_bitrates = |
| g_malloc0 (sizeof (guint64) * NUM_LOOKBACK_FRAGMENTS); |
| gst_pad_set_element_private (pad, stream); |
| stream->qos_earliest_time = GST_CLOCK_TIME_NONE; |
| |
| g_mutex_lock (&demux->priv->preroll_lock); |
| stream->do_block = TRUE; |
| demux->priv->preroll_pending++; |
| g_mutex_unlock (&demux->priv->preroll_lock); |
| |
| gst_pad_set_query_function (pad, |
| GST_DEBUG_FUNCPTR (gst_adaptive_demux_src_query)); |
| gst_pad_set_event_function (pad, |
| GST_DEBUG_FUNCPTR (gst_adaptive_demux_src_event)); |
| |
| gst_segment_init (&stream->segment, GST_FORMAT_TIME); |
| g_cond_init (&stream->fragment_download_cond); |
| g_mutex_init (&stream->fragment_download_lock); |
| |
| demux->next_streams = g_list_append (demux->next_streams, stream); |
| |
| return stream; |
| } |
| |
| GstAdaptiveDemuxStream * |
| gst_adaptive_demux_find_stream_for_pad (GstAdaptiveDemux * demux, GstPad * pad) |
| { |
| GList *iter; |
| |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| if (stream->pad == pad) { |
| return stream; |
| } |
| } |
| |
| return NULL; |
| } |
| |
| /* must be called with manifest_lock taken. |
| * It will temporarily drop the manifest_lock in order to join the task. |
| * It will join only the old_streams (the demux->streams are joined by |
| * gst_adaptive_demux_stop_tasks before gst_adaptive_demux_stream_free is |
| * called) |
| */ |
| static void |
| gst_adaptive_demux_stream_free (GstAdaptiveDemuxStream * stream) |
| { |
| GstAdaptiveDemux *demux = stream->demux; |
| GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| if (klass->stream_free) |
| klass->stream_free (stream); |
| |
| g_clear_error (&stream->last_error); |
| if (stream->download_task) { |
| if (GST_TASK_STATE (stream->download_task) != GST_TASK_STOPPED) { |
| GST_DEBUG_OBJECT (demux, "Leaving streaming task %s:%s", |
| GST_DEBUG_PAD_NAME (stream->pad)); |
| |
| gst_task_stop (stream->download_task); |
| |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->cancelled = TRUE; |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| GST_LOG_OBJECT (demux, "Waiting for task to finish"); |
| |
| /* temporarily drop the manifest lock to join the task */ |
| GST_MANIFEST_UNLOCK (demux); |
| |
| gst_task_join (stream->download_task); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| GST_LOG_OBJECT (demux, "Finished"); |
| gst_object_unref (stream->download_task); |
| g_rec_mutex_clear (&stream->download_lock); |
| stream->download_task = NULL; |
| } |
| |
| gst_adaptive_demux_stream_fragment_clear (&stream->fragment); |
| |
| if (stream->pending_segment) { |
| gst_event_unref (stream->pending_segment); |
| stream->pending_segment = NULL; |
| } |
| |
| if (stream->pending_events) { |
| g_list_free_full (stream->pending_events, (GDestroyNotify) gst_event_unref); |
| stream->pending_events = NULL; |
| } |
| |
| if (stream->internal_pad) { |
| gst_object_unparent (GST_OBJECT_CAST (stream->internal_pad)); |
| } |
| |
| if (stream->src_srcpad) { |
| gst_object_unref (stream->src_srcpad); |
| stream->src_srcpad = NULL; |
| } |
| |
| if (stream->src) { |
| GstElement *src = stream->src; |
| |
| stream->src = NULL; |
| |
| GST_MANIFEST_UNLOCK (demux); |
| gst_element_set_locked_state (src, TRUE); |
| gst_element_set_state (src, GST_STATE_NULL); |
| gst_bin_remove (GST_BIN_CAST (demux), src); |
| GST_MANIFEST_LOCK (demux); |
| } |
| |
| g_cond_clear (&stream->fragment_download_cond); |
| g_mutex_clear (&stream->fragment_download_lock); |
| g_free (stream->fragment_bitrates); |
| |
| if (stream->pad) { |
| gst_object_unref (stream->pad); |
| stream->pad = NULL; |
| } |
| if (stream->pending_caps) |
| gst_caps_unref (stream->pending_caps); |
| |
| g_clear_pointer (&stream->pending_tags, gst_tag_list_unref); |
| |
| g_free (stream); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_get_live_seek_range (GstAdaptiveDemux * demux, |
| gint64 * range_start, gint64 * range_stop) |
| { |
| GstAdaptiveDemuxClass *klass; |
| |
| klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| g_return_val_if_fail (klass->get_live_seek_range, FALSE); |
| |
| return klass->get_live_seek_range (demux, range_start, range_stop); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_stream_in_live_seek_range (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| gint64 range_start, range_stop; |
| if (gst_adaptive_demux_get_live_seek_range (demux, &range_start, &range_stop)) { |
| GST_LOG_OBJECT (stream->pad, |
| "stream position %" GST_TIME_FORMAT " live seek range %" |
| GST_STIME_FORMAT " - %" GST_STIME_FORMAT, |
| GST_TIME_ARGS (stream->segment.position), GST_STIME_ARGS (range_start), |
| GST_STIME_ARGS (range_stop)); |
| return (stream->segment.position >= range_start |
| && stream->segment.position <= range_stop); |
| } |
| |
| return FALSE; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_can_seek (GstAdaptiveDemux * demux) |
| { |
| GstAdaptiveDemuxClass *klass; |
| |
| klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| if (gst_adaptive_demux_is_live (demux)) { |
| return klass->get_live_seek_range != NULL; |
| } |
| |
| return klass->seek != NULL; |
| } |
| |
| static void |
| gst_adaptive_demux_update_streams_segment (GstAdaptiveDemux * demux, |
| GList * streams, gint64 period_start, GstSeekType start_type, |
| GstSeekType stop_type) |
| { |
| GList *iter; |
| for (iter = streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| GstEvent *seg_evt; |
| GstClockTime offset; |
| |
| /* See comments in gst_adaptive_demux_get_period_start_time() for |
| * an explanation of the segment modifications */ |
| stream->segment = demux->segment; |
| offset = gst_adaptive_demux_stream_get_presentation_offset (demux, stream); |
| stream->segment.start += offset - period_start; |
| if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)) |
| stream->segment.stop += offset - period_start; |
| if (demux->segment.rate > 0 && start_type != GST_SEEK_TYPE_NONE) |
| stream->segment.position = stream->segment.start; |
| else if (demux->segment.rate < 0 && stop_type != GST_SEEK_TYPE_NONE) |
| stream->segment.position = stream->segment.stop; |
| seg_evt = gst_event_new_segment (&stream->segment); |
| gst_event_set_seqnum (seg_evt, demux->priv->segment_seqnum); |
| gst_event_replace (&stream->pending_segment, seg_evt); |
| GST_DEBUG_OBJECT (stream->pad, "Pending segment now %" GST_PTR_FORMAT, |
| stream->pending_segment); |
| gst_event_unref (seg_evt); |
| /* Make sure the first buffer after a seek has the discont flag */ |
| stream->discont = TRUE; |
| stream->qos_earliest_time = GST_CLOCK_TIME_NONE; |
| } |
| } |
| |
| #define IS_SNAP_SEEK(f) (f & (GST_SEEK_FLAG_SNAP_BEFORE | \ |
| GST_SEEK_FLAG_SNAP_AFTER | \ |
| GST_SEEK_FLAG_SNAP_NEAREST | \ |
| GST_SEEK_FLAG_TRICKMODE_KEY_UNITS | \ |
| GST_SEEK_FLAG_KEY_UNIT)) |
| #define REMOVE_SNAP_FLAGS(f) (f & ~(GST_SEEK_FLAG_SNAP_BEFORE | \ |
| GST_SEEK_FLAG_SNAP_AFTER | \ |
| GST_SEEK_FLAG_SNAP_NEAREST)) |
| |
| static gboolean |
| gst_adaptive_demux_handle_seek_event (GstAdaptiveDemux * demux, GstPad * pad, |
| GstEvent * event) |
| { |
| GstAdaptiveDemuxClass *demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| gdouble rate; |
| GstFormat format; |
| GstSeekFlags flags; |
| GstSeekType start_type, stop_type; |
| gint64 start, stop; |
| guint32 seqnum; |
| gboolean update; |
| gboolean ret; |
| GstSegment oldsegment; |
| GstAdaptiveDemuxStream *stream = NULL; |
| |
| GST_INFO_OBJECT (demux, "Received seek event"); |
| |
| GST_API_LOCK (demux); |
| GST_MANIFEST_LOCK (demux); |
| |
| if (!gst_adaptive_demux_can_seek (demux)) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| gst_event_unref (event); |
| return FALSE; |
| } |
| |
| gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start, |
| &stop_type, &stop); |
| |
| if (format != GST_FORMAT_TIME) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| GST_WARNING_OBJECT (demux, |
| "Adaptive demuxers only support TIME-based seeking"); |
| gst_event_unref (event); |
| return FALSE; |
| } |
| |
| if (flags & GST_SEEK_FLAG_SEGMENT) { |
| GST_FIXME_OBJECT (demux, "Handle segment seeks"); |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| gst_event_unref (event); |
| return FALSE; |
| } |
| |
| seqnum = gst_event_get_seqnum (event); |
| |
| if (gst_adaptive_demux_is_live (demux)) { |
| gint64 range_start, range_stop; |
| gboolean changed = FALSE; |
| gboolean start_valid = TRUE, stop_valid = TRUE; |
| |
| if (!gst_adaptive_demux_get_live_seek_range (demux, &range_start, |
| &range_stop)) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| gst_event_unref (event); |
| GST_WARNING_OBJECT (demux, "Failure getting the live seek ranges"); |
| return FALSE; |
| } |
| |
| GST_DEBUG_OBJECT (demux, |
| "Live range is %" GST_STIME_FORMAT " %" GST_STIME_FORMAT, |
| GST_STIME_ARGS (range_start), GST_STIME_ARGS (range_stop)); |
| |
| /* Handle relative positioning for live streams (relative to the range_stop) */ |
| if (start_type == GST_SEEK_TYPE_END) { |
| start = range_stop + start; |
| start_type = GST_SEEK_TYPE_SET; |
| changed = TRUE; |
| } |
| if (stop_type == GST_SEEK_TYPE_END) { |
| stop = range_stop + stop; |
| stop_type = GST_SEEK_TYPE_SET; |
| changed = TRUE; |
| } |
| |
| /* Adjust the requested start/stop position if it falls beyond the live |
| * seek range. |
| * The only case where we don't adjust is for the starting point of |
| * an accurate seek (start if forward and stop if backwards) |
| */ |
| if (start_type == GST_SEEK_TYPE_SET && start < range_start && |
| (rate < 0 || !(flags & GST_SEEK_FLAG_ACCURATE))) { |
| GST_DEBUG_OBJECT (demux, |
| "seek before live stream start, setting to range start: %" |
| GST_TIME_FORMAT, GST_TIME_ARGS (range_start)); |
| start = range_start; |
| changed = TRUE; |
| } |
| /* truncate stop position also if set */ |
| if (stop_type == GST_SEEK_TYPE_SET && stop > range_stop && |
| (rate > 0 || !(flags & GST_SEEK_FLAG_ACCURATE))) { |
| GST_DEBUG_OBJECT (demux, |
| "seek ending after live start, adjusting to: %" |
| GST_TIME_FORMAT, GST_TIME_ARGS (range_stop)); |
| stop = range_stop; |
| changed = TRUE; |
| } |
| |
| if (start_type == GST_SEEK_TYPE_SET && GST_CLOCK_TIME_IS_VALID (start) && |
| (start < range_start || start > range_stop)) { |
| GST_WARNING_OBJECT (demux, |
| "Seek to invalid position start:%" GST_STIME_FORMAT |
| " out of seekable range (%" GST_STIME_FORMAT " - %" GST_STIME_FORMAT |
| ")", GST_STIME_ARGS (start), GST_STIME_ARGS (range_start), |
| GST_STIME_ARGS (range_stop)); |
| start_valid = FALSE; |
| } |
| if (stop_type == GST_SEEK_TYPE_SET && GST_CLOCK_TIME_IS_VALID (stop) && |
| (stop < range_start || stop > range_stop)) { |
| GST_WARNING_OBJECT (demux, |
| "Seek to invalid position stop:%" GST_STIME_FORMAT |
| " out of seekable range (%" GST_STIME_FORMAT " - %" GST_STIME_FORMAT |
| ")", GST_STIME_ARGS (stop), GST_STIME_ARGS (range_start), |
| GST_STIME_ARGS (range_stop)); |
| stop_valid = FALSE; |
| } |
| |
| /* If the seek position is still outside of the seekable range, refuse the seek */ |
| if (!start_valid || !stop_valid) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| gst_event_unref (event); |
| return FALSE; |
| } |
| |
| /* Re-create seek event with changed/updated values */ |
| if (changed) { |
| gst_event_unref (event); |
| event = |
| gst_event_new_seek (rate, format, flags, |
| start_type, start, stop_type, stop); |
| gst_event_set_seqnum (event, seqnum); |
| } |
| } |
| |
| GST_DEBUG_OBJECT (demux, "seek event, %" GST_PTR_FORMAT, event); |
| |
| /* have a backup in case seek fails */ |
| gst_segment_copy_into (&demux->segment, &oldsegment); |
| |
| if (flags & GST_SEEK_FLAG_FLUSH) { |
| GstEvent *fevent; |
| |
| GST_DEBUG_OBJECT (demux, "sending flush start"); |
| fevent = gst_event_new_flush_start (); |
| gst_event_set_seqnum (fevent, seqnum); |
| GST_MANIFEST_UNLOCK (demux); |
| gst_adaptive_demux_push_src_event (demux, fevent); |
| GST_MANIFEST_LOCK (demux); |
| |
| gst_adaptive_demux_stop_tasks (demux, FALSE); |
| } else if ((rate > 0 && start_type != GST_SEEK_TYPE_NONE) || |
| (rate < 0 && stop_type != GST_SEEK_TYPE_NONE)) { |
| |
| gst_adaptive_demux_stop_tasks (demux, FALSE); |
| } |
| |
| GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux); |
| |
| /* |
| * Handle snap seeks as follows: |
| * 1) do the snap seeking on the stream that received |
| * the event |
| * 2) use the final position on this stream to seek |
| * on the other streams to the same position |
| * |
| * We can't snap at all streams at the same time as |
| * they might end in different positions, so just |
| * use the one that received the event as the 'leading' |
| * one to do the snap seek. |
| */ |
| if (IS_SNAP_SEEK (flags) && demux_class->stream_seek && (stream = |
| gst_adaptive_demux_find_stream_for_pad (demux, pad))) { |
| GstClockTime ts; |
| GstSeekFlags stream_seek_flags = flags; |
| |
| /* snap-seek on the stream that received the event and then |
| * use the resulting position to seek on all streams */ |
| |
| if (rate >= 0) { |
| if (start_type != GST_SEEK_TYPE_NONE) |
| ts = start; |
| else { |
| ts = stream->segment.position; |
| start_type = GST_SEEK_TYPE_SET; |
| } |
| } else { |
| if (stop_type != GST_SEEK_TYPE_NONE) |
| ts = stop; |
| else { |
| stop_type = GST_SEEK_TYPE_SET; |
| ts = stream->segment.position; |
| } |
| } |
| |
| if (stream) { |
| demux_class->stream_seek (stream, rate >= 0, stream_seek_flags, ts, &ts); |
| } |
| |
| /* replace event with a new one without snaping to seek on all streams */ |
| gst_event_unref (event); |
| if (rate >= 0) { |
| start = ts; |
| } else { |
| stop = ts; |
| } |
| event = |
| gst_event_new_seek (rate, format, REMOVE_SNAP_FLAGS (flags), |
| start_type, start, stop_type, stop); |
| GST_DEBUG_OBJECT (demux, "Adapted snap seek to %" GST_PTR_FORMAT, event); |
| } |
| stream = NULL; |
| |
| gst_segment_do_seek (&demux->segment, rate, format, flags, start_type, |
| start, stop_type, stop, &update); |
| |
| /* FIXME - this seems unatural, do_seek() is updating base when we |
| * only want the start/stop position to change, maybe do_seek() needs |
| * some fixing? */ |
| if (!(flags & GST_SEEK_FLAG_FLUSH) && ((rate > 0 |
| && start_type == GST_SEEK_TYPE_NONE) || (rate < 0 |
| && stop_type == GST_SEEK_TYPE_NONE))) { |
| demux->segment.base = oldsegment.base; |
| } |
| |
| GST_DEBUG_OBJECT (demux, "Calling subclass seek: %" GST_PTR_FORMAT, event); |
| |
| ret = demux_class->seek (demux, event); |
| |
| if (!ret) { |
| /* Is there anything else we can do if it fails? */ |
| gst_segment_copy_into (&oldsegment, &demux->segment); |
| } else { |
| demux->priv->segment_seqnum = seqnum; |
| } |
| GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux); |
| |
| if (flags & GST_SEEK_FLAG_FLUSH) { |
| GstEvent *fevent; |
| |
| GST_DEBUG_OBJECT (demux, "Sending flush stop on all pad"); |
| fevent = gst_event_new_flush_stop (TRUE); |
| gst_event_set_seqnum (fevent, seqnum); |
| gst_adaptive_demux_push_src_event (demux, fevent); |
| } |
| |
| if (demux->next_streams) { |
| /* If the seek generated new streams, get them |
| * to preroll */ |
| gst_adaptive_demux_prepare_streams (demux, FALSE); |
| gst_adaptive_demux_start_tasks (demux, TRUE); |
| } else { |
| GstClockTime period_start = |
| gst_adaptive_demux_get_period_start_time (demux); |
| |
| GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux); |
| gst_adaptive_demux_update_streams_segment (demux, demux->streams, |
| period_start, start_type, stop_type); |
| gst_adaptive_demux_update_streams_segment (demux, demux->prepared_streams, |
| period_start, start_type, stop_type); |
| GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux); |
| |
| /* Restart the demux */ |
| gst_adaptive_demux_start_tasks (demux, FALSE); |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| GST_API_UNLOCK (demux); |
| gst_event_unref (event); |
| |
| return ret; |
| } |
| |
| static gboolean |
| gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent, |
| GstEvent * event) |
| { |
| GstAdaptiveDemux *demux; |
| |
| demux = GST_ADAPTIVE_DEMUX_CAST (parent); |
| |
| /* FIXME handle events received on pads that are to be removed */ |
| |
| switch (event->type) { |
| case GST_EVENT_SEEK: |
| { |
| guint32 seqnum = gst_event_get_seqnum (event); |
| if (seqnum == demux->priv->segment_seqnum) { |
| GST_LOG_OBJECT (pad, |
| "Drop duplicated SEEK event seqnum %" G_GUINT32_FORMAT, seqnum); |
| gst_event_unref (event); |
| return TRUE; |
| } |
| return gst_adaptive_demux_handle_seek_event (demux, pad, event); |
| } |
| case GST_EVENT_RECONFIGURE:{ |
| GstAdaptiveDemuxStream *stream; |
| |
| GST_MANIFEST_LOCK (demux); |
| stream = gst_adaptive_demux_find_stream_for_pad (demux, pad); |
| |
| if (stream) { |
| if (!stream->cancelled && demux->running && |
| stream->last_ret == GST_FLOW_NOT_LINKED) { |
| stream->last_ret = GST_FLOW_OK; |
| stream->restart_download = TRUE; |
| stream->need_header = TRUE; |
| stream->discont = TRUE; |
| GST_DEBUG_OBJECT (stream->pad, "Restarting download loop"); |
| gst_task_start (stream->download_task); |
| } |
| gst_event_unref (event); |
| GST_MANIFEST_UNLOCK (demux); |
| return TRUE; |
| } |
| GST_MANIFEST_UNLOCK (demux); |
| } |
| break; |
| case GST_EVENT_LATENCY:{ |
| /* Upstream and our internal source are irrelevant |
| * for latency, and we should not fail here to |
| * configure the latency */ |
| gst_event_unref (event); |
| return TRUE; |
| } |
| case GST_EVENT_QOS:{ |
| GstAdaptiveDemuxStream *stream; |
| |
| GST_MANIFEST_LOCK (demux); |
| stream = gst_adaptive_demux_find_stream_for_pad (demux, pad); |
| |
| if (stream) { |
| GstClockTimeDiff diff; |
| GstClockTime timestamp; |
| |
| gst_event_parse_qos (event, NULL, NULL, &diff, ×tamp); |
| /* Only take into account lateness if late */ |
| if (diff > 0) |
| stream->qos_earliest_time = timestamp + 2 * diff; |
| else |
| stream->qos_earliest_time = timestamp; |
| GST_DEBUG_OBJECT (stream->pad, "qos_earliest_time %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (stream->qos_earliest_time)); |
| } |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| } |
| default: |
| break; |
| } |
| |
| return gst_pad_event_default (pad, parent, event); |
| } |
| |
| static gboolean |
| gst_adaptive_demux_src_query (GstPad * pad, GstObject * parent, |
| GstQuery * query) |
| { |
| GstAdaptiveDemux *demux; |
| GstAdaptiveDemuxClass *demux_class; |
| gboolean ret = FALSE; |
| |
| if (query == NULL) |
| return FALSE; |
| |
| demux = GST_ADAPTIVE_DEMUX_CAST (parent); |
| demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| switch (query->type) { |
| case GST_QUERY_DURATION:{ |
| GstClockTime duration = -1; |
| GstFormat fmt; |
| |
| gst_query_parse_duration (query, &fmt, NULL); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| if (fmt == GST_FORMAT_TIME && demux->priv->have_manifest) { |
| duration = demux_class->get_duration (demux); |
| |
| if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) { |
| gst_query_set_duration (query, GST_FORMAT_TIME, duration); |
| ret = TRUE; |
| } |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| GST_LOG_OBJECT (demux, "GST_QUERY_DURATION returns %s with duration %" |
| GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration)); |
| break; |
| } |
| case GST_QUERY_LATENCY:{ |
| gst_query_set_latency (query, FALSE, 0, -1); |
| ret = TRUE; |
| break; |
| } |
| case GST_QUERY_SEEKING:{ |
| GstFormat fmt; |
| gint64 stop = -1; |
| gint64 start = 0; |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| if (!demux->priv->have_manifest) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_INFO_OBJECT (demux, |
| "Don't have manifest yet, can't answer seeking query"); |
| return FALSE; /* can't answer without manifest */ |
| } |
| |
| gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL); |
| GST_INFO_OBJECT (demux, "Received GST_QUERY_SEEKING with format %d", fmt); |
| if (fmt == GST_FORMAT_TIME) { |
| GstClockTime duration; |
| gboolean can_seek = gst_adaptive_demux_can_seek (demux); |
| |
| ret = TRUE; |
| if (can_seek) { |
| if (gst_adaptive_demux_is_live (demux)) { |
| ret = gst_adaptive_demux_get_live_seek_range (demux, &start, &stop); |
| if (!ret) { |
| GST_MANIFEST_UNLOCK (demux); |
| GST_INFO_OBJECT (demux, "can't answer seeking query"); |
| return FALSE; |
| } |
| } else { |
| duration = demux_class->get_duration (demux); |
| if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) |
| stop = duration; |
| } |
| } |
| gst_query_set_seeking (query, fmt, can_seek, start, stop); |
| GST_INFO_OBJECT (demux, "GST_QUERY_SEEKING returning with start : %" |
| GST_TIME_FORMAT ", stop : %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (start), GST_TIME_ARGS (stop)); |
| } |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| } |
| case GST_QUERY_URI: |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| /* TODO HLS can answer this differently it seems */ |
| if (demux->manifest_uri) { |
| /* FIXME: (hls) Do we answer with the variant playlist, with the current |
| * playlist or the the uri of the last downlowaded fragment? */ |
| gst_query_set_uri (query, demux->manifest_uri); |
| ret = TRUE; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| default: |
| /* Don't forward queries upstream because of the special nature of this |
| * "demuxer", which relies on the upstream element only to be fed |
| * the Manifest |
| */ |
| break; |
| } |
| |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static void |
| gst_adaptive_demux_start_tasks (GstAdaptiveDemux * demux, |
| gboolean start_preroll_streams) |
| { |
| GList *iter; |
| |
| if (!demux->running) { |
| GST_DEBUG_OBJECT (demux, "Not starting tasks due to shutdown"); |
| return; |
| } |
| |
| GST_INFO_OBJECT (demux, "Starting streams' tasks"); |
| |
| iter = start_preroll_streams ? demux->prepared_streams : demux->streams; |
| |
| for (; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| if (!start_preroll_streams) { |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->cancelled = FALSE; |
| stream->replaced = FALSE; |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| |
| stream->last_ret = GST_FLOW_OK; |
| gst_task_start (stream->download_task); |
| } |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static void |
| gst_adaptive_demux_stop_manifest_update_task (GstAdaptiveDemux * demux) |
| { |
| gst_uri_downloader_cancel (demux->downloader); |
| |
| gst_task_stop (demux->priv->updates_task); |
| |
| g_mutex_lock (&demux->priv->updates_timed_lock); |
| GST_DEBUG_OBJECT (demux, "requesting stop of the manifest update task"); |
| demux->priv->stop_updates_task = TRUE; |
| g_cond_signal (&demux->priv->updates_timed_cond); |
| g_mutex_unlock (&demux->priv->updates_timed_lock); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static void |
| gst_adaptive_demux_start_manifest_update_task (GstAdaptiveDemux * demux) |
| { |
| GstAdaptiveDemuxClass *demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| if (gst_adaptive_demux_is_live (demux)) { |
| gst_uri_downloader_reset (demux->downloader); |
| g_mutex_lock (&demux->priv->updates_timed_lock); |
| demux->priv->stop_updates_task = FALSE; |
| g_mutex_unlock (&demux->priv->updates_timed_lock); |
| /* Task to periodically update the manifest */ |
| if (demux_class->requires_periodical_playlist_update (demux)) { |
| GST_DEBUG_OBJECT (demux, "requesting start of the manifest update task"); |
| gst_task_start (demux->priv->updates_task); |
| } |
| } |
| } |
| |
| /* must be called with manifest_lock taken |
| * This function will temporarily release manifest_lock in order to join the |
| * download threads. |
| * The api_lock will still protect it against other threads trying to modify |
| * the demux element. |
| */ |
| static void |
| gst_adaptive_demux_stop_tasks (GstAdaptiveDemux * demux, gboolean stop_updates) |
| { |
| int i; |
| GList *iter; |
| GList *list_to_process; |
| |
| GST_LOG_OBJECT (demux, "Stopping tasks"); |
| |
| if (stop_updates) |
| gst_adaptive_demux_stop_manifest_update_task (demux); |
| |
| list_to_process = demux->streams; |
| for (i = 0; i < 2; ++i) { |
| for (iter = list_to_process; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->cancelled = TRUE; |
| gst_task_stop (stream->download_task); |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| list_to_process = demux->prepared_streams; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| g_mutex_lock (&demux->priv->preroll_lock); |
| g_cond_broadcast (&demux->priv->preroll_cond); |
| g_mutex_unlock (&demux->priv->preroll_lock); |
| GST_MANIFEST_LOCK (demux); |
| |
| g_mutex_lock (&demux->priv->manifest_update_lock); |
| g_cond_broadcast (&demux->priv->manifest_cond); |
| g_mutex_unlock (&demux->priv->manifest_update_lock); |
| |
| /* need to release manifest_lock before stopping the src element. |
| * The streams were asked to cancel, so they will not make any writes to demux |
| * object. Even if we temporarily release manifest_lock, the demux->streams |
| * cannot change and iter cannot be invalidated. |
| */ |
| list_to_process = demux->streams; |
| for (i = 0; i < 2; ++i) { |
| for (iter = list_to_process; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| GstElement *src = stream->src; |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| if (src) { |
| gst_element_set_locked_state (src, TRUE); |
| gst_element_set_state (src, GST_STATE_READY); |
| } |
| |
| /* stream->download_task value never changes, so it is safe to read it |
| * outside critical section |
| */ |
| gst_task_join (stream->download_task); |
| |
| GST_MANIFEST_LOCK (demux); |
| } |
| list_to_process = demux->prepared_streams; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| if (stop_updates) |
| gst_task_join (demux->priv->updates_task); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| list_to_process = demux->streams; |
| for (i = 0; i < 2; ++i) { |
| for (iter = list_to_process; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| stream->download_error_count = 0; |
| stream->need_header = TRUE; |
| stream->qos_earliest_time = GST_CLOCK_TIME_NONE; |
| } |
| list_to_process = demux->prepared_streams; |
| } |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_push_src_event (GstAdaptiveDemux * demux, GstEvent * event) |
| { |
| GList *iter; |
| gboolean ret = TRUE; |
| |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| gst_event_ref (event); |
| ret = ret & gst_pad_push_event (stream->pad, event); |
| } |
| gst_event_unref (event); |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| void |
| gst_adaptive_demux_stream_set_caps (GstAdaptiveDemuxStream * stream, |
| GstCaps * caps) |
| { |
| GST_DEBUG_OBJECT (stream->pad, "setting new caps for stream %" GST_PTR_FORMAT, |
| caps); |
| gst_caps_replace (&stream->pending_caps, caps); |
| gst_caps_unref (caps); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| void |
| gst_adaptive_demux_stream_set_tags (GstAdaptiveDemuxStream * stream, |
| GstTagList * tags) |
| { |
| GST_DEBUG_OBJECT (stream->pad, "setting new tags for stream %" GST_PTR_FORMAT, |
| tags); |
| if (stream->pending_tags) { |
| gst_tag_list_unref (stream->pending_tags); |
| } |
| stream->pending_tags = tags; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| void |
| gst_adaptive_demux_stream_queue_event (GstAdaptiveDemuxStream * stream, |
| GstEvent * event) |
| { |
| stream->pending_events = g_list_append (stream->pending_events, event); |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static guint64 |
| _update_average_bitrate (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream, guint64 new_bitrate) |
| { |
| gint index = stream->moving_index % NUM_LOOKBACK_FRAGMENTS; |
| |
| stream->moving_bitrate -= stream->fragment_bitrates[index]; |
| stream->fragment_bitrates[index] = new_bitrate; |
| stream->moving_bitrate += new_bitrate; |
| |
| stream->moving_index += 1; |
| |
| if (stream->moving_index > NUM_LOOKBACK_FRAGMENTS) |
| return stream->moving_bitrate / NUM_LOOKBACK_FRAGMENTS; |
| return stream->moving_bitrate / stream->moving_index; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static guint64 |
| gst_adaptive_demux_stream_update_current_bitrate (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| guint64 average_bitrate; |
| guint64 fragment_bitrate; |
| |
| if (demux->connection_speed) { |
| GST_LOG_OBJECT (demux, "Connection-speed is set to %u kbps, using it", |
| demux->connection_speed / 1000); |
| return demux->connection_speed; |
| } |
| |
| fragment_bitrate = stream->last_bitrate; |
| GST_DEBUG_OBJECT (demux, "Download bitrate is : %" G_GUINT64_FORMAT " bps", |
| fragment_bitrate); |
| |
| average_bitrate = _update_average_bitrate (demux, stream, fragment_bitrate); |
| |
| GST_INFO_OBJECT (stream, "last fragment bitrate was %" G_GUINT64_FORMAT, |
| fragment_bitrate); |
| GST_INFO_OBJECT (stream, |
| "Last %u fragments average bitrate is %" G_GUINT64_FORMAT, |
| NUM_LOOKBACK_FRAGMENTS, average_bitrate); |
| |
| /* Conservative approach, make sure we don't upgrade too fast */ |
| stream->current_download_rate = MIN (average_bitrate, fragment_bitrate); |
| |
| stream->current_download_rate *= demux->bitrate_limit; |
| GST_DEBUG_OBJECT (demux, "Bitrate after bitrate limit (%0.2f): %" |
| G_GUINT64_FORMAT, demux->bitrate_limit, stream->current_download_rate); |
| |
| #if 0 |
| /* Debugging code, modulate the bitrate every few fragments */ |
| { |
| static guint ctr = 0; |
| if (ctr % 3 == 0) { |
| GST_INFO_OBJECT (demux, "Halving reported bitrate for debugging"); |
| stream->current_download_rate /= 2; |
| } |
| ctr++; |
| } |
| #endif |
| |
| return stream->current_download_rate; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static GstFlowReturn |
| gst_adaptive_demux_combine_flows (GstAdaptiveDemux * demux) |
| { |
| gboolean all_notlinked = TRUE; |
| gboolean all_eos = TRUE; |
| GList *iter; |
| |
| for (iter = demux->streams; iter; iter = g_list_next (iter)) { |
| GstAdaptiveDemuxStream *stream = iter->data; |
| |
| if (stream->last_ret != GST_FLOW_NOT_LINKED) { |
| all_notlinked = FALSE; |
| if (stream->last_ret != GST_FLOW_EOS) |
| all_eos = FALSE; |
| } |
| |
| if (stream->last_ret <= GST_FLOW_NOT_NEGOTIATED |
| || stream->last_ret == GST_FLOW_FLUSHING) { |
| return stream->last_ret; |
| } |
| } |
| if (all_notlinked) |
| return GST_FLOW_NOT_LINKED; |
| else if (all_eos) |
| return GST_FLOW_EOS; |
| return GST_FLOW_OK; |
| } |
| |
| /* Called with preroll_lock */ |
| static void |
| gst_adaptive_demux_handle_preroll (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| demux->priv->preroll_pending--; |
| if (demux->priv->preroll_pending == 0) { |
| /* That was the last one, time to release all streams |
| * and expose them */ |
| GST_DEBUG_OBJECT (demux, "All streams prerolled. exposing"); |
| gst_adaptive_demux_expose_streams (demux); |
| g_cond_broadcast (&demux->priv->preroll_cond); |
| } |
| } |
| |
| /* must be called with manifest_lock taken. |
| * Temporarily releases manifest_lock |
| */ |
| GstFlowReturn |
| gst_adaptive_demux_stream_push_buffer (GstAdaptiveDemuxStream * stream, |
| GstBuffer * buffer) |
| { |
| GstAdaptiveDemux *demux = stream->demux; |
| GstFlowReturn ret = GST_FLOW_OK; |
| gboolean discont = FALSE; |
| /* Pending events */ |
| GstEvent *pending_caps = NULL, *pending_segment = NULL, *pending_tags = NULL; |
| GList *pending_events = NULL; |
| |
| /* FIXME : |
| * This is duplicating *exactly* the same thing as what is done at the beginning |
| * of _src_chain if starting_fragment is TRUE */ |
| if (stream->first_fragment_buffer) { |
| GstClockTime offset = |
| gst_adaptive_demux_stream_get_presentation_offset (demux, stream); |
| GstClockTime period_start = |
| gst_adaptive_demux_get_period_start_time (demux); |
| |
| GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux); |
| if (demux->segment.rate < 0) |
| /* Set DISCONT flag for every first buffer in reverse playback mode |
| * as each fragment for its own has to be reversed */ |
| discont = TRUE; |
| |
| GST_BUFFER_PTS (buffer) = stream->fragment.timestamp; |
| if (GST_BUFFER_PTS_IS_VALID (buffer)) |
| GST_BUFFER_PTS (buffer) += offset; |
| |
| if (GST_BUFFER_PTS_IS_VALID (buffer)) { |
| stream->segment.position = GST_BUFFER_PTS (buffer); |
| |
| /* Convert from position inside the stream's segment to the demuxer's |
| * segment, they are not necessarily the same */ |
| if (stream->segment.position - offset + period_start > |
| demux->segment.position) |
| demux->segment.position = |
| stream->segment.position - offset + period_start; |
| } |
| GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux); |
| |
| GST_LOG_OBJECT (stream->pad, |
| "Going to push buffer with PTS %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (GST_BUFFER_PTS (buffer))); |
| } else { |
| GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE; |
| } |
| |
| if (stream->discont) { |
| discont = TRUE; |
| stream->discont = FALSE; |
| } |
| |
| if (discont) { |
| GST_DEBUG_OBJECT (stream->pad, "Marking fragment as discontinuous"); |
| GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT); |
| } else { |
| GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT); |
| } |
| |
| stream->first_fragment_buffer = FALSE; |
| |
| GST_BUFFER_DURATION (buffer) = GST_CLOCK_TIME_NONE; |
| GST_BUFFER_DTS (buffer) = GST_CLOCK_TIME_NONE; |
| if (G_UNLIKELY (stream->pending_caps)) { |
| pending_caps = gst_event_new_caps (stream->pending_caps); |
| gst_caps_unref (stream->pending_caps); |
| stream->pending_caps = NULL; |
| } |
| |
| if (stream->do_block) { |
| |
| g_mutex_lock (&demux->priv->preroll_lock); |
| |
| /* If we are preroll state, set caps in here */ |
| if (pending_caps) { |
| gst_pad_push_event (stream->pad, pending_caps); |
| pending_caps = NULL; |
| } |
| |
| gst_adaptive_demux_handle_preroll (demux, stream); |
| GST_MANIFEST_UNLOCK (demux); |
| |
| while (stream->do_block && !stream->cancelled) { |
| GST_LOG_OBJECT (demux, "Stream %p sleeping for preroll", stream); |
| g_cond_wait (&demux->priv->preroll_cond, &demux->priv->preroll_lock); |
| } |
| if (stream->cancelled) { |
| GST_LOG_OBJECT (demux, "stream %p cancelled", stream); |
| gst_buffer_unref (buffer); |
| g_mutex_unlock (&demux->priv->preroll_lock); |
| return GST_FLOW_FLUSHING; |
| } |
| |
| g_mutex_unlock (&demux->priv->preroll_lock); |
| GST_MANIFEST_LOCK (demux); |
| } |
| |
| if (G_UNLIKELY (stream->pending_segment)) { |
| GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux); |
| pending_segment = stream->pending_segment; |
| stream->pending_segment = NULL; |
| GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux); |
| } |
| if (G_UNLIKELY (stream->pending_tags || stream->bitrate_changed)) { |
| GstTagList *tags = stream->pending_tags; |
| |
| stream->pending_tags = NULL; |
| stream->bitrate_changed = 0; |
| |
| if (stream->fragment.bitrate != 0) { |
| if (tags) |
| tags = gst_tag_list_make_writable (tags); |
| else |
| tags = gst_tag_list_new_empty (); |
| |
| gst_tag_list_add (tags, GST_TAG_MERGE_KEEP, |
| GST_TAG_NOMINAL_BITRATE, stream->fragment.bitrate, NULL); |
| } |
| if (tags) |
| pending_tags = gst_event_new_tag (tags); |
| } |
| if (G_UNLIKELY (stream->pending_events)) { |
| pending_events = stream->pending_events; |
| stream->pending_events = NULL; |
| } |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| /* Do not push events or buffers holding the manifest lock */ |
| if (G_UNLIKELY (pending_caps)) { |
| GST_DEBUG_OBJECT (stream->pad, "Setting pending caps: %" GST_PTR_FORMAT, |
| pending_caps); |
| gst_pad_push_event (stream->pad, pending_caps); |
| } |
| if (G_UNLIKELY (pending_segment)) { |
| GST_DEBUG_OBJECT (stream->pad, "Sending pending seg: %" GST_PTR_FORMAT, |
| pending_segment); |
| gst_pad_push_event (stream->pad, pending_segment); |
| } |
| if (G_UNLIKELY (pending_tags)) { |
| GST_DEBUG_OBJECT (stream->pad, "Sending pending tags: %" GST_PTR_FORMAT, |
| pending_tags); |
| gst_pad_push_event (stream->pad, pending_tags); |
| } |
| while (pending_events != NULL) { |
| GstEvent *event = pending_events->data; |
| |
| if (!gst_pad_push_event (stream->pad, event)) |
| GST_ERROR_OBJECT (stream->pad, "Failed to send pending event"); |
| |
| pending_events = g_list_delete_link (pending_events, pending_events); |
| } |
| |
| /* Wait for preroll if blocking */ |
| GST_DEBUG_OBJECT (stream->pad, |
| "About to push buffer of size %" G_GSIZE_FORMAT, |
| gst_buffer_get_size (buffer)); |
| |
| ret = gst_pad_push (stream->pad, buffer); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| g_mutex_lock (&stream->fragment_download_lock); |
| if (G_UNLIKELY (stream->cancelled)) { |
| GST_LOG_OBJECT (stream, "Stream was cancelled"); |
| ret = stream->last_ret = GST_FLOW_FLUSHING; |
| g_mutex_unlock (&stream->fragment_download_lock); |
| return ret; |
| } |
| g_mutex_unlock (&stream->fragment_download_lock); |
| |
| GST_LOG_OBJECT (stream->pad, "Push result: %d %s", ret, |
| gst_flow_get_name (ret)); |
| |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static GstFlowReturn |
| gst_adaptive_demux_stream_finish_fragment_default (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| /* No need to advance, this isn't a real fragment */ |
| if (G_UNLIKELY (stream->downloading_header || stream->downloading_index)) |
| return GST_FLOW_OK; |
| |
| return gst_adaptive_demux_stream_advance_fragment (demux, stream, |
| stream->fragment.duration); |
| } |
| |
| /* must be called with manifest_lock taken. |
| * Can temporarily release manifest_lock |
| */ |
| static GstFlowReturn |
| gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream, GstBuffer * buffer) |
| { |
| return gst_adaptive_demux_stream_push_buffer (stream, buffer); |
| } |
| |
| static gboolean |
| gst_adaptive_demux_requires_periodical_playlist_update_default (GstAdaptiveDemux |
| * demux) |
| { |
| return TRUE; |
| } |
| |
| static GstFlowReturn |
| _src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer) |
| { |
| GstAdaptiveDemuxStream *stream; |
| GstAdaptiveDemux *demux; |
| GstAdaptiveDemuxClass *klass; |
| GstFlowReturn ret = GST_FLOW_OK; |
| |
| demux = GST_ADAPTIVE_DEMUX_CAST (parent); |
| stream = gst_pad_get_element_private (pad); |
| klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| /* do not make any changes if the stream is cancelled */ |
| g_mutex_lock (&stream->fragment_download_lock); |
| if (G_UNLIKELY (stream->cancelled)) { |
| g_mutex_unlock (&stream->fragment_download_lock); |
| gst_buffer_unref (buffer); |
| ret = stream->last_ret = GST_FLOW_FLUSHING; |
| GST_MANIFEST_UNLOCK (demux); |
| return ret; |
| } |
| g_mutex_unlock (&stream->fragment_download_lock); |
| |
| /* starting_fragment is set to TRUE at the beginning of |
| * _stream_download_fragment() |
| * /!\ If there is a header/index being downloaded, then this will |
| * be TRUE for the first one ... but FALSE for the remaining ones, |
| * including the *actual* fragment ! */ |
| if (stream->starting_fragment) { |
| GstClockTime offset = |
| gst_adaptive_demux_stream_get_presentation_offset (demux, stream); |
| GstClockTime period_start = |
| gst_adaptive_demux_get_period_start_time (demux); |
| |
| stream->starting_fragment = FALSE; |
| if (klass->start_fragment) { |
| if (!klass->start_fragment (demux, stream)) { |
| ret = GST_FLOW_ERROR; |
| goto error; |
| } |
| } |
| |
| GST_BUFFER_PTS (buffer) = stream->fragment.timestamp; |
| if (GST_BUFFER_PTS_IS_VALID (buffer)) |
| GST_BUFFER_PTS (buffer) += offset; |
| |
| GST_LOG_OBJECT (stream->pad, "set fragment pts=%" GST_TIME_FORMAT, |
| GST_TIME_ARGS (GST_BUFFER_PTS (buffer))); |
| |
| if (GST_BUFFER_PTS_IS_VALID (buffer)) { |
| GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux); |
| stream->segment.position = GST_BUFFER_PTS (buffer); |
| |
| /* Convert from position inside the stream's segment to the demuxer's |
| * segment, they are not necessarily the same */ |
| if (stream->segment.position - offset + period_start > |
| demux->segment.position) |
| demux->segment.position = |
| stream->segment.position - offset + period_start; |
| GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux); |
| } |
| |
| } else { |
| GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE; |
| } |
| |
| /* downloading_first_buffer is set to TRUE in download_uri() just before |
| * activating the source (i.e. requesting a given URI) |
| * |
| * The difference with starting_fragment is that this will be called |
| * for *all* first buffers (of index, and header, and fragment) |
| * |
| * ... to then only do something useful (in this block) for actual |
| * fragments... */ |
| if (stream->downloading_first_buffer) { |
| gint64 chunk_size = 0; |
| |
| stream->downloading_first_buffer = FALSE; |
| |
| if (!stream->downloading_header && !stream->downloading_index) { |
| /* If this is the first buffer of a fragment (not the headers or index) |
| * and we don't have a birate from the sub-class, then see if we |
| * can work it out from the fragment size and duration */ |
| if (stream->fragment.bitrate == 0 && |
| stream->fragment.duration != 0 && |
| gst_element_query_duration (stream->uri_handler, GST_FORMAT_BYTES, |
| &chunk_size)) { |
| guint bitrate = MIN (G_MAXUINT, gst_util_uint64_scale (chunk_size, |
| 8 * GST_SECOND, stream->fragment.duration)); |
| GST_LOG_OBJECT (demux, |
| "Fragment has size %" G_GUINT64_FORMAT " duration %" GST_TIME_FORMAT |
| " = bitrate %u", chunk_size, |
| GST_TIME_ARGS (stream->fragment.duration), bitrate); |
| stream->fragment.bitrate = bitrate; |
| } |
| if (stream->fragment.bitrate) { |
| stream->bitrate_changed = TRUE; |
| } else { |
| GST_WARNING_OBJECT (demux, "Bitrate for fragment not available"); |
| } |
| } |
| } |
| |
| stream->download_total_bytes += gst_buffer_get_size (buffer); |
| |
| GST_TRACE_OBJECT (stream->pad, "Received buffer of size %" G_GSIZE_FORMAT, |
| gst_buffer_get_size (buffer)); |
| |
| ret = klass->data_received (demux, stream, buffer); |
| |
| if (ret == GST_FLOW_FLUSHING) { |
| /* do not make any changes if the stream is cancelled */ |
| g_mutex_lock (&stream->fragment_download_lock); |
| if (G_UNLIKELY (stream->cancelled)) { |
| g_mutex_unlock (&stream->fragment_download_lock); |
| GST_MANIFEST_UNLOCK (demux); |
| return ret; |
| } |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| |
| if (ret != GST_FLOW_OK) { |
| gboolean finished = FALSE; |
| |
| if (ret < GST_FLOW_EOS) { |
| GST_ELEMENT_FLOW_ERROR (demux, ret); |
| |
| /* TODO push this on all pads */ |
| gst_pad_push_event (stream->pad, gst_event_new_eos ()); |
| } else { |
| GST_DEBUG_OBJECT (stream->pad, "stream stopped, reason %s", |
| gst_flow_get_name (ret)); |
| } |
| |
| if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) { |
| ret = GST_FLOW_EOS; /* return EOS to make the source stop */ |
| } else if (ret == GST_ADAPTIVE_DEMUX_FLOW_END_OF_FRAGMENT) { |
| /* Behaves like an EOS event from upstream */ |
| stream->fragment.finished = TRUE; |
| ret = klass->finish_fragment (demux, stream); |
| if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) { |
| ret = GST_FLOW_EOS; /* return EOS to make the source stop */ |
| } else if (ret != GST_FLOW_OK) { |
| goto error; |
| } |
| finished = TRUE; |
| } |
| |
| gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL); |
| if (finished) |
| ret = GST_FLOW_EOS; |
| } |
| |
| error: |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static void |
| gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream * |
| stream, GstFlowReturn ret, GError * err) |
| { |
| GST_DEBUG_OBJECT (stream->pad, "Download finish: %d %s - err: %p", ret, |
| gst_flow_get_name (ret), err); |
| |
| /* if we have an error, only replace last_ret if it was OK before to avoid |
| * overwriting the first error we got */ |
| if (stream->last_ret == GST_FLOW_OK) { |
| stream->last_ret = ret; |
| if (err) { |
| g_clear_error (&stream->last_error); |
| stream->last_error = g_error_copy (err); |
| } |
| } |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->download_finished = TRUE; |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| } |
| |
| static GstFlowReturn |
| gst_adaptive_demux_eos_handling (GstAdaptiveDemuxStream * stream) |
| { |
| GstFlowReturn ret = GST_FLOW_OK; |
| GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (stream->demux); |
| |
| if (!klass->need_another_chunk || stream->fragment.chunk_size == -1 |
| || !klass->need_another_chunk (stream) |
| || stream->fragment.chunk_size == 0) { |
| stream->fragment.finished = TRUE; |
| ret = klass->finish_fragment (stream->demux, stream); |
| } |
| gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL); |
| |
| return ret; |
| } |
| |
| static gboolean |
| _src_event (GstPad * pad, GstObject * parent, GstEvent * event) |
| { |
| GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad); |
| GstAdaptiveDemux *demux = stream->demux; |
| |
| switch (GST_EVENT_TYPE (event)) { |
| case GST_EVENT_EOS:{ |
| GST_DEBUG_OBJECT (pad, "Saw EOS on src pad"); |
| GST_MANIFEST_LOCK (demux); |
| |
| gst_adaptive_demux_eos_handling (stream); |
| |
| /* FIXME ? |
| * _eos_handling() calls fragment_download_finish() which does the |
| * same thing as below. |
| * Could this cause races ? */ |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->download_finished = TRUE; |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| break; |
| } |
| default: |
| break; |
| } |
| |
| gst_event_unref (event); |
| |
| return TRUE; |
| } |
| |
| static gboolean |
| _src_query (GstPad * pad, GstObject * parent, GstQuery * query) |
| { |
| GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad); |
| |
| switch (GST_QUERY_TYPE (query)) { |
| case GST_QUERY_ALLOCATION: |
| return FALSE; |
| break; |
| default: |
| break; |
| } |
| |
| return gst_pad_peer_query (stream->pad, query); |
| } |
| |
| static GstPadProbeReturn |
| _uri_handler_probe (GstPad * pad, GstPadProbeInfo * info, |
| GstAdaptiveDemuxStream * stream) |
| { |
| GstPadProbeReturn ret = GST_PAD_PROBE_OK; |
| |
| if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_BUFFER) { |
| GstBuffer *buf = GST_PAD_PROBE_INFO_BUFFER (info); |
| if (stream->fragment_bytes_downloaded == 0) { |
| stream->last_latency = |
| gst_adaptive_demux_get_monotonic_time (stream->demux) - |
| (stream->download_start_time * GST_USECOND); |
| GST_DEBUG_OBJECT (pad, |
| "FIRST BYTE since download_start %" GST_TIME_FORMAT, |
| GST_TIME_ARGS (stream->last_latency)); |
| } |
| stream->fragment_bytes_downloaded += gst_buffer_get_size (buf); |
| GST_LOG_OBJECT (pad, |
| "Received buffer, size %" G_GSIZE_FORMAT " total %" G_GUINT64_FORMAT, |
| gst_buffer_get_size (buf), stream->fragment_bytes_downloaded); |
| } else if (GST_PAD_PROBE_INFO_TYPE (info) & |
| GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) { |
| GstEvent *ev = GST_PAD_PROBE_INFO_EVENT (info); |
| GST_LOG_OBJECT (pad, "Received event %s %" GST_PTR_FORMAT, |
| GST_EVENT_TYPE_NAME (ev), ev); |
| switch (GST_EVENT_TYPE (ev)) { |
| case GST_EVENT_SEGMENT: |
| stream->fragment_bytes_downloaded = 0; |
| break; |
| case GST_EVENT_EOS: |
| { |
| stream->last_download_time = |
| gst_adaptive_demux_get_monotonic_time (stream->demux) - |
| (stream->download_start_time * GST_USECOND); |
| stream->last_bitrate = |
| gst_util_uint64_scale (stream->fragment_bytes_downloaded, |
| 8 * GST_SECOND, stream->last_download_time); |
| GST_DEBUG_OBJECT (pad, |
| "EOS since download_start %" GST_TIME_FORMAT " bitrate %" |
| G_GUINT64_FORMAT " bps", GST_TIME_ARGS (stream->last_download_time), |
| stream->last_bitrate); |
| /* Calculate bitrate since URI request */ |
| } |
| break; |
| default: |
| break; |
| } |
| } |
| |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken. |
| * Can temporarily release manifest_lock |
| */ |
| static gboolean |
| gst_adaptive_demux_stream_wait_manifest_update (GstAdaptiveDemux * demux, |
| GstAdaptiveDemuxStream * stream) |
| { |
| gboolean ret = TRUE; |
| |
| /* Wait until we're cancelled or there's something for |
| * us to download in the playlist or the playlist |
| * became non-live */ |
| while (TRUE) { |
| GST_DEBUG_OBJECT (demux, "No fragment left but live playlist, wait a bit"); |
| |
| /* get the manifest_update_lock while still holding the manifest_lock. |
| * This will prevent other threads to signal the condition (they will need |
| * both manifest_lock and manifest_update_lock in order to signal). |
| * It cannot deadlock because all threads always get the manifest_lock first |
| * and manifest_update_lock second. |
| */ |
| g_mutex_lock (&demux->priv->manifest_update_lock); |
| |
| GST_MANIFEST_UNLOCK (demux); |
| |
| g_cond_wait (&demux->priv->manifest_cond, |
| &demux->priv->manifest_update_lock); |
| g_mutex_unlock (&demux->priv->manifest_update_lock); |
| |
| GST_MANIFEST_LOCK (demux); |
| |
| /* check for cancelled every time we get the manifest_lock */ |
| g_mutex_lock (&stream->fragment_download_lock); |
| if (G_UNLIKELY (stream->cancelled)) { |
| ret = FALSE; |
| stream->last_ret = GST_FLOW_FLUSHING; |
| g_mutex_unlock (&stream->fragment_download_lock); |
| break; |
| } |
| g_mutex_unlock (&stream->fragment_download_lock); |
| |
| /* Got a new fragment or not live anymore? */ |
| if (gst_adaptive_demux_stream_update_fragment_info (demux, stream) == |
| GST_FLOW_OK) { |
| GST_DEBUG_OBJECT (demux, "new fragment available, " |
| "not waiting for manifest update"); |
| ret = TRUE; |
| break; |
| } |
| |
| if (!gst_adaptive_demux_is_live (demux)) { |
| GST_DEBUG_OBJECT (demux, "Not live anymore, " |
| "not waiting for manifest update"); |
| ret = FALSE; |
| break; |
| } |
| } |
| GST_DEBUG_OBJECT (demux, "Retrying now"); |
| return ret; |
| } |
| |
| /* must be called with manifest_lock taken */ |
| static gboolean |
| gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream, |
| const gchar * uri, const gchar * referer, gboolean refresh, |
| gboolean allow_cache) |
| { |
| GstAdaptiveDemux *demux = stream->demux; |
| |
| if (!gst_uri_is_valid (uri)) { |
| GST_WARNING_OBJECT (stream->pad, "Invalid URI: %s", uri); |
| return FALSE; |
| } |
| |
| /* Try to re-use existing source element */ |
| if (stream->src != NULL) { |
| gchar *old_protocol, *new_protocol; |
| gchar *old_uri; |
| |
| old_uri = gst_uri_handler_get_uri (GST_URI_HANDLER (stream->uri_handler)); |
| old_protocol = gst_uri_get_protocol (old_uri); |
| new_protocol = gst_uri_get_protocol (uri); |
| |
| if (!g_str_equal (old_protocol, new_protocol)) { |
| GstElement *src = stream->src; |
| |
| stream->src = NULL; |
| gst_object_unref (stream->src_srcpad); |
| stream->src_srcpad = NULL; |
| GST_MANIFEST_UNLOCK (demux); |
| gst_element_set_locked_state (src, TRUE); |
| gst_element_set_state (src, GST_STATE_NULL); |
| gst_bin_remove (GST_BIN_CAST (demux), src); |
| GST_MANIFEST_LOCK (demux); |
| GST_DEBUG_OBJECT (demux, "Can't re-use old source element"); |
| } else { |
| GError *err = NULL; |
| |
| GST_DEBUG_OBJECT (demux, "Re-using old source element"); |
| if (!gst_uri_handler_set_uri (GST_URI_HANDLER (stream->uri_handler), uri, |
| &err)) { |
| GstElement *src = stream->src; |
| |
| stream->src = NULL; |
| GST_DEBUG_OBJECT (demux, "Failed to re-use old source element: %s", |
| err ? err->message : "Unknown error"); |
| g_clear_error (&err); |
| gst_object_unref (stream->src_srcpad); |
| stream->src_srcpad = NULL; |
| GST_MANIFEST_UNLOCK (demux); |
| gst_element_set_locked_state (src, TRUE); |
| gst_element_set_state (src, GST_STATE_NULL); |
| gst_bin_remove (GST_BIN_CAST (demux), src); |
| GST_MANIFEST_LOCK (demux); |
| } |
| } |
| g_free (old_uri); |
| g_free (old_protocol); |
| g_free (new_protocol); |
| } |
| |
| if (stream->src == NULL) { |
| GstPad *uri_handler_src; |
| GstPad *queue_sink; |
| GstPad *queue_src; |
| GstElement *uri_handler; |
| GstElement *queue; |
| GstPadLinkReturn pad_link_ret; |
| GObjectClass *gobject_class; |
| gchar *internal_name, *bin_name; |
| |
| /* Our src consists of a bin containing uri_handler -> queue . The |
| * purpose of the queue is to allow the uri_handler to download an |
| * entire fragment without blocking, so we can accurately measure the |
| * download bitrate. */ |
| |
| queue = gst_element_factory_make ("queue", NULL); |
| if (queue == NULL) |
| return FALSE; |
| |
| g_object_set (queue, "max-size-bytes", (guint) SRC_QUEUE_MAX_BYTES, NULL); |
| g_object_set (queue, "max-size-buffers", (guint) 0, NULL); |
| g_object_set (queue, "max-size-time", (guint64) 0, NULL); |
| |
| uri_handler = gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL); |
| if (uri_handler == NULL) { |
| GST_ELEMENT_ERROR (demux, CORE, MISSING_PLUGIN, |
| ("Missing plugin to handle URI: '%s'", uri), (NULL)); |
| gst_object_unref (queue); |
| return FALSE; |
| } |
| |
| gobject_class = G_OBJECT_GET_CLASS (uri_handler); |
| |
| if (g_object_class_find_property (gobject_class, "compress")) |
| g_object_set (uri_handler, "compress", FALSE, NULL); |
| if (g_object_class_find_property (gobject_class, "keep-alive")) |
| g_object_set (uri_handler, "keep-alive", TRUE, NULL); |
| if (g_object_class_find_property (gobject_class, "extra-headers")) { |
| if (referer || refresh || !allow_cache) { |
| GstStructure *extra_headers = gst_structure_new_empty ("headers"); |
| |
| if (referer) |
| gst_structure_set (extra_headers, "Referer", G_TYPE_STRING, referer, |
| NULL); |
| |
| if (!allow_cache) |
| gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING, |
| "no-cache", NULL); |
| else if (refresh) |
| gst_structure_set (extra_headers, "Cache-Control", G_TYPE_STRING, |
| "max-age=0", NULL); |
| |
| g_object_set (uri_handler, "extra-headers", extra_headers, NULL); |
| |
| gst_structure_free (extra_headers); |
| } else { |
| g_object_set (uri_handler, "extra-headers", NULL, NULL); |
| } |
| } |
| |
| /* Source bin creation */ |
| bin_name = g_strdup_printf ("srcbin-%s", GST_PAD_NAME (stream->pad)); |
| stream->src = gst_bin_new (bin_name); |
| g_free (bin_name); |
| if (stream->src == NULL) { |
| gst_object_unref (queue); |
| gst_object_unref (uri_handler); |
| return FALSE; |
| } |
| |
| gst_bin_add (GST_BIN_CAST (stream->src), queue); |
| gst_bin_add (GST_BIN_CAST (stream->src), uri_handler); |
| |
| uri_handler_src = gst_element_get_static_pad (uri_handler, "src"); |
| queue_sink = gst_element_get_static_pad (queue, "sink"); |
| |
| pad_link_ret = |
| gst_pad_link_full (uri_handler_src, queue_sink, |
| GST_PAD_LINK_CHECK_NOTHING); |
| if (GST_PAD_LINK_FAILED (pad_link_ret)) { |
| GST_WARNING_OBJECT (demux, |
| "Could not link pads %s:%s to %s:%s for reason %d", |
| GST_DEBUG_PAD_NAME (uri_handler_src), GST_DEBUG_PAD_NAME (queue_sink), |
| pad_link_ret); |
| g_object_unref (queue_sink); |
| g_object_unref (uri_handler_src); |
| gst_object_unref (stream->src); |
| stream->src = NULL; |
| return FALSE; |
| } |
| |
| /* Add a downstream event and data probe */ |
| gst_pad_add_probe (uri_handler_src, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM, |
| (GstPadProbeCallback) _uri_handler_probe, stream, NULL); |
| |
| g_object_unref (queue_sink); |
| g_object_unref (uri_handler_src); |
| queue_src = gst_element_get_static_pad (queue, "src"); |
| stream->src_srcpad = gst_ghost_pad_new ("src", queue_src); |
| g_object_unref (queue_src); |
| gst_element_add_pad (stream->src, stream->src_srcpad); |
| |
| gst_element_set_locked_state (stream->src, TRUE); |
| gst_bin_add (GST_BIN_CAST (demux), stream->src); |
| stream->src_srcpad = gst_element_get_static_pad (stream->src, "src"); |
| |
| /* set up our internal floating pad to drop all events from |
| * the http src we don't care about. On the chain function |
| * we just push the buffer forward */ |
| internal_name = g_strdup_printf ("internal-%s", GST_PAD_NAME (stream->pad)); |
| stream->internal_pad = gst_pad_new (internal_name, GST_PAD_SINK); |
| g_free (internal_name); |
| gst_object_set_parent (GST_OBJECT_CAST (stream->internal_pad), |
| GST_OBJECT_CAST (demux)); |
| GST_OBJECT_FLAG_SET (stream->internal_pad, GST_PAD_FLAG_NEED_PARENT); |
| gst_pad_set_element_private (stream->internal_pad, stream); |
| gst_pad_set_active (stream->internal_pad, TRUE); |
| gst_pad_set_chain_function (stream->internal_pad, _src_chain); |
| gst_pad_set_event_function (stream->internal_pad, _src_event); |
| gst_pad_set_query_function (stream->internal_pad, _src_query); |
| |
| if (gst_pad_link_full (stream->src_srcpad, stream->internal_pad, |
| GST_PAD_LINK_CHECK_NOTHING) != GST_PAD_LINK_OK) { |
| GST_ERROR_OBJECT (stream->pad, "Failed to link internal pad"); |
| return FALSE; |
| } |
| |
| stream->uri_handler = uri_handler; |
| stream->queue = queue; |
| |
| stream->last_status_code = 200; /* default to OK */ |
| } |
| return TRUE; |
| } |
| |
| static GstPadProbeReturn |
| gst_ad_stream_src_to_ready_cb (GstPad * pad, GstPadProbeInfo * info, |
| gpointer user_data) |
| { |
| GstAdaptiveDemuxStream *stream = user_data; |
| |
| /* The source's src pad is IDLE so now set the state to READY */ |
| g_mutex_lock (&stream->fragment_download_lock); |
| stream->src_at_ready = TRUE; |
| g_cond_signal (&stream->fragment_download_cond); |
| g_mutex_unlock (&stream->fragment_download_lock); |
| |
| return GST_PAD_PROBE_REMOVE; |
| } |
| |
| #ifndef GST_DISABLE_GST_DEBUG |
| static const char * |
| uritype (GstAdaptiveDemuxStream * s) |
| { |
| if (s->downloading_header) |
| return "header"; |
| if (s->downloading_index) |
| return "index"; |
| return "fragment"; |
| } |
| #endif |
| |
| /* must be called with manifest_lock taken. |
|