diff --git a/ext/smoothstreaming/Makefile.am b/ext/smoothstreaming/Makefile.am index a683f754eb..06c6796bbe 100644 --- a/ext/smoothstreaming/Makefile.am +++ b/ext/smoothstreaming/Makefile.am @@ -7,6 +7,7 @@ libgstsmoothstreaming_la_CFLAGS = $(GST_PLUGINS_BAD_CFLAGS) \ libgstsmoothstreaming_la_LIBADD = \ $(top_builddir)/gst-libs/gst/codecparsers/libgstcodecparsers-$(GST_API_VERSION).la \ $(top_builddir)/gst-libs/gst/uridownloader/libgsturidownloader-$(GST_API_VERSION).la \ + $(top_builddir)/gst-libs/gst/adaptivedemux/libgstadaptivedemux-@GST_API_VERSION@.la \ $(GST_PLUGINS_BASE_LIBS) \ -lgsttag-$(GST_API_VERSION) \ $(GST_BASE_LIBS) $(GST_LIBS) $(ZLIB_LIBS) $(LIBXML2_LIBS) diff --git a/ext/smoothstreaming/gstmssdemux.c b/ext/smoothstreaming/gstmssdemux.c index f7a5265a78..24aa031bf3 100644 --- a/ext/smoothstreaming/gstmssdemux.c +++ b/ext/smoothstreaming/gstmssdemux.c @@ -81,8 +81,6 @@ GST_DEBUG_CATEGORY (mssdemux_debug); #define DEFAULT_MAX_QUEUE_SIZE_BUFFERS 0 #define DEFAULT_BITRATE_LIMIT 0.8 -#define MAX_DOWNLOAD_ERROR_COUNT 3 - enum { PROP_0, @@ -113,42 +111,39 @@ GST_STATIC_PAD_TEMPLATE ("audio_%02u", GST_STATIC_CAPS_ANY); #define gst_mss_demux_parent_class parent_class -G_DEFINE_TYPE (GstMssDemux, gst_mss_demux, GST_TYPE_BIN); +G_DEFINE_TYPE (GstMssDemux, gst_mss_demux, GST_TYPE_ADAPTIVE_DEMUX); static void gst_mss_demux_dispose (GObject * object); static void gst_mss_demux_set_property (GObject * object, guint prop_id, const GValue * value, GParamSpec * pspec); static void gst_mss_demux_get_property (GObject * object, guint prop_id, GValue * value, GParamSpec * pspec); -static GstStateChangeReturn gst_mss_demux_change_state (GstElement * element, - GstStateChange transition); -static void gst_mss_demux_handle_message (GstBin * bin, GstMessage * msg); -static GstFlowReturn gst_mss_demux_chain (GstPad * pad, GstObject * parent, + +static gboolean gst_mss_demux_is_live (GstAdaptiveDemux * demux); +static gboolean gst_mss_demux_process_manifest (GstAdaptiveDemux * demux, GstBuffer * buffer); -static GstFlowReturn gst_mss_demux_event (GstPad * pad, GstObject * parent, - GstEvent * event); -static gboolean gst_mss_demux_src_query (GstPad * pad, GstObject * parent, - GstQuery * query); - -static void gst_mss_demux_download_loop (GstMssDemuxStream * stream); -static GstFlowReturn gst_mss_demux_stream_push_event (GstMssDemuxStream * - stream, GstEvent * event); -static GstFlowReturn gst_mss_demux_combine_flows (GstMssDemux * mssdemux); - -static gboolean gst_mss_demux_process_manifest (GstMssDemux * mssdemux); -static void -gst_mss_demux_stop_tasks (GstMssDemux * mssdemux, gboolean immediate); +static GstClockTime gst_mss_demux_get_duration (GstAdaptiveDemux * demux); +static void gst_mss_demux_reset (GstAdaptiveDemux * demux); +static GstFlowReturn gst_mss_demux_stream_seek (GstAdaptiveDemuxStream * stream, + GstClockTime ts); +static GstFlowReturn +gst_mss_demux_stream_advance_fragment (GstAdaptiveDemuxStream * stream); +static gboolean gst_mss_demux_stream_select_bitrate (GstAdaptiveDemuxStream * + stream, guint64 bitrate); +static GstFlowReturn +gst_mss_demux_stream_update_fragment_info (GstAdaptiveDemuxStream * stream); +static gboolean gst_mss_demux_seek (GstAdaptiveDemux * demux, GstEvent * seek); static void gst_mss_demux_class_init (GstMssDemuxClass * klass) { GObjectClass *gobject_class; GstElementClass *gstelement_class; - GstBinClass *gstbin_class; + GstAdaptiveDemuxClass *gstadaptivedemux_class; gobject_class = (GObjectClass *) klass; gstelement_class = (GstElementClass *) klass; - gstbin_class = (GstBinClass *) klass; + gstadaptivedemux_class = (GstAdaptiveDemuxClass *) klass; gst_element_class_add_pad_template (gstelement_class, gst_static_pad_template_get (&gst_mss_demux_sink_template)); @@ -187,10 +182,18 @@ gst_mss_demux_class_init (GstMssDemuxClass * klass) 0, 1, DEFAULT_BITRATE_LIMIT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); - gstelement_class->change_state = - GST_DEBUG_FUNCPTR (gst_mss_demux_change_state); - - gstbin_class->handle_message = gst_mss_demux_handle_message; + gstadaptivedemux_class->process_manifest = gst_mss_demux_process_manifest; + gstadaptivedemux_class->is_live = gst_mss_demux_is_live; + gstadaptivedemux_class->get_duration = gst_mss_demux_get_duration; + gstadaptivedemux_class->reset = gst_mss_demux_reset; + gstadaptivedemux_class->seek = gst_mss_demux_seek; + gstadaptivedemux_class->stream_seek = gst_mss_demux_stream_seek; + gstadaptivedemux_class->stream_advance_fragment = + gst_mss_demux_stream_advance_fragment; + gstadaptivedemux_class->stream_select_bitrate = + gst_mss_demux_stream_select_bitrate; + gstadaptivedemux_class->stream_update_fragment_info = + gst_mss_demux_stream_update_fragment_info; GST_DEBUG_CATEGORY_INIT (mssdemux_debug, "mssdemux", 0, "mssdemux plugin"); } @@ -198,135 +201,33 @@ gst_mss_demux_class_init (GstMssDemuxClass * klass) static void gst_mss_demux_init (GstMssDemux * mssdemux) { - mssdemux->sinkpad = - gst_pad_new_from_static_template (&gst_mss_demux_sink_template, "sink"); - gst_pad_set_chain_function (mssdemux->sinkpad, - GST_DEBUG_FUNCPTR (gst_mss_demux_chain)); - gst_pad_set_event_function (mssdemux->sinkpad, - GST_DEBUG_FUNCPTR (gst_mss_demux_event)); - gst_element_add_pad (GST_ELEMENT_CAST (mssdemux), mssdemux->sinkpad); - mssdemux->data_queue_max_size = DEFAULT_MAX_QUEUE_SIZE_BUFFERS; mssdemux->bitrate_limit = DEFAULT_BITRATE_LIMIT; - mssdemux->have_group_id = FALSE; - mssdemux->group_id = G_MAXUINT; -} - -static GstMssDemuxStream * -gst_mss_demux_stream_new (GstMssDemux * mssdemux, - GstMssStream * manifeststream, GstPad * srcpad) -{ - GstMssDemuxStream *stream; - - stream = g_new0 (GstMssDemuxStream, 1); - - /* Downloading task */ - g_rec_mutex_init (&stream->download_lock); - stream->download_task = - gst_task_new ((GstTaskFunction) gst_mss_demux_download_loop, stream, - NULL); - gst_task_set_lock (stream->download_task, &stream->download_lock); - - stream->pad = srcpad; - stream->manifest_stream = manifeststream; - stream->parent = mssdemux; - - gst_segment_init (&stream->segment, GST_FORMAT_TIME); - g_cond_init (&stream->fragment_download_cond); - g_mutex_init (&stream->fragment_download_lock); - - return stream; + gst_adaptive_demux_set_stream_struct_size (GST_ADAPTIVE_DEMUX_CAST (mssdemux), + sizeof (GstMssDemuxStream)); } static void -gst_mss_demux_stream_free (GstMssDemuxStream * stream) +gst_mss_demux_reset (GstAdaptiveDemux * demux) { - if (stream->download_task) { - if (GST_TASK_STATE (stream->download_task) != GST_TASK_STOPPED) { - GST_DEBUG_OBJECT (stream->parent, "Leaving streaming task %s:%s", - GST_DEBUG_PAD_NAME (stream->pad)); - g_cond_signal (&stream->fragment_download_cond); - gst_task_stop (stream->download_task); - GST_LOG_OBJECT (stream->parent, "Waiting for task to finish"); - gst_task_join (stream->download_task); - GST_LOG_OBJECT (stream->parent, "Finished"); - } - gst_object_unref (stream->download_task); - g_rec_mutex_clear (&stream->download_lock); - stream->download_task = NULL; - } - - if (stream->pending_segment) { - gst_event_unref (stream->pending_segment); - stream->pending_segment = NULL; - } - - if (stream->src_srcpad) { - gst_object_unref (stream->src_srcpad); - stream->src_srcpad = NULL; - } - - if (stream->src) { - gst_element_set_state (stream->src, GST_STATE_NULL); - gst_bin_remove (GST_BIN_CAST (stream->parent), stream->src); - stream->src = NULL; - } - - g_cond_clear (&stream->fragment_download_cond); - g_mutex_clear (&stream->fragment_download_lock); - - if (stream->pad) { - gst_object_unref (stream->pad); - stream->pad = NULL; - } - if (stream->caps) - gst_caps_unref (stream->caps); - g_free (stream); -} - -static void -gst_mss_demux_reset (GstMssDemux * mssdemux) -{ - GSList *iter; - - gst_mss_demux_stop_tasks (mssdemux, TRUE); - - if (mssdemux->manifest_buffer) { - gst_buffer_unref (mssdemux->manifest_buffer); - mssdemux->manifest_buffer = NULL; - } - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - if (stream->pad) - gst_element_remove_pad (GST_ELEMENT_CAST (mssdemux), stream->pad); - gst_mss_demux_stream_free (stream); - } - g_slist_free (mssdemux->streams); - mssdemux->streams = NULL; + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); if (mssdemux->manifest) { gst_mss_manifest_free (mssdemux->manifest); mssdemux->manifest = NULL; } - - mssdemux->n_videos = mssdemux->n_audios = 0; g_free (mssdemux->base_url); mssdemux->base_url = NULL; - g_free (mssdemux->manifest_uri); - mssdemux->manifest_uri = NULL; - mssdemux->have_group_id = FALSE; - mssdemux->group_id = G_MAXUINT; + mssdemux->n_videos = mssdemux->n_audios = 0; + } static void gst_mss_demux_dispose (GObject * object) { - GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (object); - - gst_mss_demux_reset (mssdemux); + gst_mss_demux_reset (GST_ADAPTIVE_DEMUX_CAST (object)); G_OBJECT_CLASS (parent_class)->dispose (object); } @@ -380,390 +281,78 @@ gst_mss_demux_get_property (GObject * object, guint prop_id, GValue * value, } } -static GstStateChangeReturn -gst_mss_demux_change_state (GstElement * element, GstStateChange transition) +static gboolean +gst_mss_demux_is_live (GstAdaptiveDemux * demux) { - GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (element); - GstStateChangeReturn result = GST_STATE_CHANGE_FAILURE; + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); - switch (transition) { - case GST_STATE_CHANGE_PAUSED_TO_READY: - gst_mss_demux_reset (mssdemux); - break; - case GST_STATE_CHANGE_READY_TO_NULL: - break; - default: - break; - } + g_return_val_if_fail (mssdemux->manifest != NULL, FALSE); - result = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); - - return result; + return gst_mss_manifest_is_live (mssdemux->manifest); } -static void -gst_mss_demux_handle_message (GstBin * bin, GstMessage * msg) +static GstClockTime +gst_mss_demux_get_duration (GstAdaptiveDemux * demux) { - GstMssDemux *demux = GST_MSS_DEMUX_CAST (bin); + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); - switch (GST_MESSAGE_TYPE (msg)) { - case GST_MESSAGE_ERROR:{ - GSList *iter; - GstMssDemuxStream *stream; - GError *err = NULL; - gchar *debug = NULL; + g_return_val_if_fail (mssdemux->manifest != NULL, FALSE); - for (iter = demux->streams; iter; iter = g_slist_next (iter)) { - stream = iter->data; - if (GST_OBJECT_CAST (stream->src) == GST_MESSAGE_SRC (msg)) { - gst_message_parse_error (msg, &err, &debug); - - GST_WARNING_OBJECT (stream->pad, "Source posted error: %d:%d %s (%s)", - err->domain, err->code, err->message, debug); - - /* error, but ask to retry */ - g_mutex_lock (&stream->fragment_download_lock); - if (!stream->download_finished) { - stream->last_ret = GST_FLOW_CUSTOM_ERROR; - stream->download_finished = TRUE; - } - g_cond_signal (&stream->fragment_download_cond); - g_mutex_unlock (&stream->fragment_download_lock); - - g_error_free (err); - g_free (debug); - break; - } - } - - gst_message_unref (msg); - msg = NULL; - } - break; - default: - break; - } - - if (msg) - GST_BIN_CLASS (parent_class)->handle_message (bin, msg); + return gst_mss_manifest_get_gst_duration (mssdemux->manifest); } - static GstFlowReturn -gst_mss_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer) +gst_mss_demux_stream_update_fragment_info (GstAdaptiveDemuxStream * stream) { - GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (parent); - if (mssdemux->manifest_buffer == NULL) - mssdemux->manifest_buffer = buffer; - else - mssdemux->manifest_buffer = - gst_buffer_append (mssdemux->manifest_buffer, buffer); + GstMssDemuxStream *mssstream = (GstMssDemuxStream *) stream; + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (stream->demux); + GstFlowReturn ret; + gchar *path = NULL; - GST_INFO_OBJECT (mssdemux, "Received manifest buffer, total size is %i bytes", - (gint) gst_buffer_get_size (mssdemux->manifest_buffer)); + g_free (stream->fragment.uri); + stream->fragment.uri = NULL; + ret = gst_mss_stream_get_fragment_url (mssstream->manifest_stream, &path); + + if (ret == GST_FLOW_OK) { + stream->fragment.uri = g_strdup_printf ("%s/%s", mssdemux->base_url, path); + stream->fragment.timestamp = + gst_mss_stream_get_fragment_gst_timestamp (mssstream->manifest_stream); + stream->fragment.duration = + gst_mss_stream_get_fragment_gst_duration (mssstream->manifest_stream); + } + g_free (path); + + return ret; +} + +static GstFlowReturn +gst_mss_demux_stream_seek (GstAdaptiveDemuxStream * stream, GstClockTime ts) +{ + GstMssDemuxStream *mssstream = (GstMssDemuxStream *) stream; + + gst_mss_stream_seek (mssstream->manifest_stream, ts); return GST_FLOW_OK; } -static void -gst_mss_demux_start (GstMssDemux * mssdemux) +static GstFlowReturn +gst_mss_demux_stream_advance_fragment (GstAdaptiveDemuxStream * stream) { - GSList *iter; + GstMssDemuxStream *mssstream = (GstMssDemuxStream *) stream; - GST_INFO_OBJECT (mssdemux, "Starting streams' tasks"); - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - gst_task_start (stream->download_task); - } + if (stream->demux->segment.rate >= 0) + return gst_mss_stream_advance_fragment (mssstream->manifest_stream); + else + return gst_mss_stream_regress_fragment (mssstream->manifest_stream); } -static gboolean -gst_mss_demux_push_src_event (GstMssDemux * mssdemux, GstEvent * event) +static GstCaps * +create_mss_caps (GstMssDemuxStream * stream, GstCaps * caps) { - GSList *iter; - gboolean ret = TRUE; - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - gst_event_ref (event); - ret = ret & gst_pad_push_event (stream->pad, event); - } - gst_event_unref (event); - return ret; -} - -static gboolean -gst_mss_demux_event (GstPad * pad, GstObject * parent, GstEvent * event) -{ - GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (parent); - gboolean forward = TRUE; - gboolean ret = TRUE; - - switch (GST_EVENT_TYPE (event)) { - case GST_EVENT_FLUSH_STOP: - gst_mss_demux_reset (mssdemux); - break; - case GST_EVENT_EOS: - if (mssdemux->manifest_buffer == NULL) { - GST_WARNING_OBJECT (mssdemux, "Received EOS without a manifest."); - break; - } - GST_INFO_OBJECT (mssdemux, "Received EOS"); - - if (gst_mss_demux_process_manifest (mssdemux)) - gst_mss_demux_start (mssdemux); - forward = FALSE; - break; - default: - break; - } - - if (forward) { - ret = gst_pad_event_default (pad, parent, event); - } else { - gst_event_unref (event); - } - - return ret; -} - -static void -gst_mss_demux_stop_tasks (GstMssDemux * mssdemux, gboolean immediate) -{ - GSList *iter; - - GST_OBJECT_LOCK (mssdemux); - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - - gst_task_stop (stream->download_task); - stream->cancelled = TRUE; - if (immediate) { - 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_OBJECT_UNLOCK (mssdemux); - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - - gst_task_join (stream->download_task); - stream->download_error_count = 0; - } -} - -static void -gst_mss_demux_restart_tasks (GstMssDemux * mssdemux) -{ - GSList *iter; - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - stream->cancelled = FALSE; - gst_task_start (stream->download_task); - } -} - -static gboolean -gst_mss_demux_src_event (GstPad * pad, GstObject * parent, GstEvent * event) -{ - GstMssDemux *mssdemux; - - mssdemux = GST_MSS_DEMUX_CAST (parent); - - switch (event->type) { - case GST_EVENT_SEEK: - { - gdouble rate; - GstFormat format; - GstSeekFlags flags; - GstSeekType start_type, stop_type; - gint64 start, stop; - GstEvent *newsegment; - GSList *iter; - gboolean update; - - GST_INFO_OBJECT (mssdemux, "Received GST_EVENT_SEEK"); - - gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start, - &stop_type, &stop); - - if (format != GST_FORMAT_TIME) - goto not_supported; - - GST_DEBUG_OBJECT (mssdemux, - "seek event, rate: %f start: %" GST_TIME_FORMAT " stop: %" - GST_TIME_FORMAT, rate, GST_TIME_ARGS (start), GST_TIME_ARGS (stop)); - - if (flags & GST_SEEK_FLAG_FLUSH) { - GstEvent *flush = gst_event_new_flush_start (); - GST_DEBUG_OBJECT (mssdemux, "sending flush start"); - - gst_event_set_seqnum (flush, gst_event_get_seqnum (event)); - gst_mss_demux_push_src_event (mssdemux, flush); - } - - gst_mss_demux_stop_tasks (mssdemux, TRUE); - - gst_mss_manifest_seek (mssdemux->manifest, start); - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - - stream->eos = FALSE; - if (flags & GST_SEEK_FLAG_FLUSH) { - stream->last_ret = GST_FLOW_OK; - } - - gst_segment_do_seek (&stream->segment, rate, format, flags, - start_type, start, stop_type, stop, &update); - - newsegment = gst_event_new_segment (&stream->segment); - gst_event_set_seqnum (newsegment, gst_event_get_seqnum (event)); - if (stream->pending_segment) - gst_event_unref (stream->pending_segment); - stream->pending_segment = newsegment; - } - - if (flags & GST_SEEK_FLAG_FLUSH) { - GstEvent *flush = gst_event_new_flush_stop (TRUE); - GST_DEBUG_OBJECT (mssdemux, "sending flush stop"); - - gst_event_set_seqnum (flush, gst_event_get_seqnum (event)); - gst_mss_demux_push_src_event (mssdemux, flush); - } - - gst_mss_demux_restart_tasks (mssdemux); - - gst_event_unref (event); - return TRUE; - } - case GST_EVENT_RECONFIGURE:{ - GSList *iter; - - GST_OBJECT_LOCK (mssdemux); - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - - if (stream->pad == pad) { - - g_mutex_lock (&stream->fragment_download_lock); - if (stream->last_ret == GST_FLOW_NOT_LINKED) { - GST_DEBUG_OBJECT (stream->pad, "Received reconfigure"); - stream->restart_download = TRUE; - stream->last_ret = GST_FLOW_OK; - gst_task_start (stream->download_task); - } - g_mutex_unlock (&stream->fragment_download_lock); - GST_OBJECT_UNLOCK (mssdemux); - gst_event_unref (event); - return TRUE; - } - } - GST_OBJECT_UNLOCK (mssdemux); - } - 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; - } - default: - break; - } - - return gst_pad_event_default (pad, parent, event); - -not_supported: - gst_event_unref (event); - return FALSE; -} - -static gboolean -gst_mss_demux_src_query (GstPad * pad, GstObject * parent, GstQuery * query) -{ - GstMssDemux *mssdemux; - gboolean ret = FALSE; - - if (query == NULL) - return FALSE; - - mssdemux = GST_MSS_DEMUX (parent); - - switch (query->type) { - case GST_QUERY_DURATION:{ - GstClockTime duration = -1; - GstFormat fmt; - - gst_query_parse_duration (query, &fmt, NULL); - if (fmt == GST_FORMAT_TIME && mssdemux->manifest) { - /* TODO should we use the streams accumulated duration or the main manifest duration? */ - duration = gst_mss_manifest_get_gst_duration (mssdemux->manifest); - - if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) { - gst_query_set_duration (query, GST_FORMAT_TIME, duration); - ret = TRUE; - } - } - GST_INFO_OBJECT (mssdemux, "GST_QUERY_DURATION returns %s with duration %" - GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration)); - break; - } - case GST_QUERY_LATENCY:{ - gboolean live = FALSE; - - live = mssdemux->manifest - && gst_mss_manifest_is_live (mssdemux->manifest); - - gst_query_set_latency (query, live, 0, -1); - ret = TRUE; - break; - } - case GST_QUERY_SEEKING:{ - GstFormat fmt; - gint64 stop = -1; - - if (!mssdemux->manifest) { - return FALSE; /* no seeking without manifest */ - } else if (gst_mss_manifest_is_live (mssdemux->manifest)) { - return FALSE; /* no live seeking */ - } - - gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL); - GST_INFO_OBJECT (mssdemux, "Received GST_QUERY_SEEKING with format %d", - fmt); - if (fmt == GST_FORMAT_TIME) { - GstClockTime duration; - duration = gst_mss_manifest_get_gst_duration (mssdemux->manifest); - if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) - stop = duration; - gst_query_set_seeking (query, fmt, TRUE, 0, stop); - ret = TRUE; - GST_INFO_OBJECT (mssdemux, "GST_QUERY_SEEKING returning with stop : %" - GST_TIME_FORMAT, GST_TIME_ARGS (stop)); - } - break; - } - default: - /* Don't fordward 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; -} - -static void -_set_src_pad_functions (GstPad * pad) -{ - gst_pad_set_query_function (pad, GST_DEBUG_FUNCPTR (gst_mss_demux_src_query)); - gst_pad_set_event_function (pad, GST_DEBUG_FUNCPTR (gst_mss_demux_src_event)); + return gst_caps_new_simple ("video/quicktime", "variant", G_TYPE_STRING, + "mss-fragmented", "timescale", G_TYPE_UINT64, + gst_mss_stream_get_timescale (stream->manifest_stream), "media-caps", + GST_TYPE_CAPS, caps, NULL); } static GstPad * @@ -798,13 +387,13 @@ _create_pad (GstMssDemux * mssdemux, GstMssStream * manifeststream) return NULL; } - _set_src_pad_functions (srcpad); return srcpad; } -static void -gst_mss_demux_create_streams (GstMssDemux * mssdemux) +static gboolean +gst_mss_demux_setup_streams (GstAdaptiveDemux * demux) { + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); GSList *streams = gst_mss_manifest_get_streams (mssdemux->manifest); GSList *iter; @@ -813,13 +402,21 @@ gst_mss_demux_create_streams (GstMssDemux * mssdemux) GST_ELEMENT_ERROR (mssdemux, STREAM, DEMUX, (_("This file contains no playable streams.")), ("no streams found at the Manifest")); - return; + return FALSE; } + GST_INFO_OBJECT (mssdemux, "Changing max bitrate to %" G_GUINT64_FORMAT, + mssdemux->connection_speed); + gst_mss_manifest_change_bitrate (mssdemux->manifest, + mssdemux->connection_speed); + mssdemux->update_bitrates = FALSE; + for (iter = streams; iter; iter = g_slist_next (iter)) { GstPad *srcpad = NULL; GstMssDemuxStream *stream = NULL; GstMssStream *manifeststream = iter->data; + GstCaps *caps; + const gchar *lang; srcpad = _create_pad (mssdemux, manifeststream); @@ -827,195 +424,73 @@ gst_mss_demux_create_streams (GstMssDemux * mssdemux) continue; } - stream = gst_mss_demux_stream_new (mssdemux, manifeststream, srcpad); - gst_pad_set_element_private (srcpad, stream); + stream = (GstMssDemuxStream *) + gst_adaptive_demux_stream_new (GST_ADAPTIVE_DEMUX_CAST (mssdemux), + srcpad); + stream->manifest_stream = manifeststream; gst_mss_stream_set_active (manifeststream, TRUE); - mssdemux->streams = g_slist_append (mssdemux->streams, stream); - } - - /* select initial bitrates */ - GST_OBJECT_LOCK (mssdemux); - GST_INFO_OBJECT (mssdemux, "Changing max bitrate to %" G_GUINT64_FORMAT, - mssdemux->connection_speed); - gst_mss_manifest_change_bitrate (mssdemux->manifest, - mssdemux->connection_speed); - mssdemux->update_bitrates = FALSE; - GST_OBJECT_UNLOCK (mssdemux); -} - -static GstCaps * -create_mss_caps (GstMssDemuxStream * stream, GstCaps * caps) -{ - return gst_caps_new_simple ("video/quicktime", "variant", G_TYPE_STRING, - "mss-fragmented", "timescale", G_TYPE_UINT64, - gst_mss_stream_get_timescale (stream->manifest_stream), "media-caps", - GST_TYPE_CAPS, caps, NULL); -} - -static gboolean -gst_mss_demux_expose_stream (GstMssDemux * mssdemux, GstMssDemuxStream * stream) -{ - GstCaps *caps; - GstCaps *media_caps; - GstPad *pad = stream->pad; - - media_caps = gst_mss_stream_get_caps (stream->manifest_stream); - - if (media_caps) { - gchar *name = gst_pad_get_name (pad); - const gchar *lang; - GstEvent *event; - gchar *stream_id; - gst_pad_set_active (pad, TRUE); - - caps = create_mss_caps (stream, media_caps); - gst_caps_unref (media_caps); - - stream_id = - gst_pad_create_stream_id (pad, GST_ELEMENT_CAST (mssdemux), name); - - event = - gst_pad_get_sticky_event (mssdemux->sinkpad, GST_EVENT_STREAM_START, 0); - if (event) { - if (gst_event_parse_group_id (event, &mssdemux->group_id)) - mssdemux->have_group_id = TRUE; - else - mssdemux->have_group_id = FALSE; - gst_event_unref (event); - } else if (!mssdemux->have_group_id) { - mssdemux->have_group_id = TRUE; - mssdemux->group_id = gst_util_group_id_next (); - } - event = gst_event_new_stream_start (stream_id); - if (mssdemux->have_group_id) - gst_event_set_group_id (event, mssdemux->group_id); - - gst_pad_push_event (pad, event); - g_free (stream_id); - g_free (name); - - gst_pad_set_caps (pad, caps); - if (stream->caps) - gst_caps_unref (stream->caps); - stream->caps = caps; + caps = gst_mss_stream_get_caps (stream->manifest_stream); + gst_adaptive_demux_stream_set_caps (GST_ADAPTIVE_DEMUX_STREAM_CAST (stream), + create_mss_caps (stream, caps)); + gst_caps_unref (caps); lang = gst_mss_stream_get_lang (stream->manifest_stream); if (lang != NULL) { GstTagList *tags; tags = gst_tag_list_new (GST_TAG_LANGUAGE_CODE, lang, NULL); - gst_pad_push_event (stream->pad, gst_event_new_tag (tags)); + gst_adaptive_demux_stream_set_tags (GST_ADAPTIVE_DEMUX_STREAM_CAST + (stream), tags); } - - gst_pad_push_event (stream->pad, gst_event_new_segment (&stream->segment)); - - GST_INFO_OBJECT (mssdemux, "Adding srcpad %s:%s with caps %" GST_PTR_FORMAT, - GST_DEBUG_PAD_NAME (pad), caps); - gst_object_ref (pad); - - gst_element_add_pad (GST_ELEMENT_CAST (mssdemux), pad); - } else { - GST_WARNING_OBJECT (mssdemux, - "Couldn't get caps from manifest stream %p %s, not exposing it", stream, - GST_PAD_NAME (stream->pad)); - return FALSE; } + return TRUE; } static gboolean -gst_mss_demux_process_manifest (GstMssDemux * mssdemux) +gst_mss_demux_process_manifest (GstAdaptiveDemux * demux, GstBuffer * buf) { - GstQuery *query; - gboolean ret; - GSList *iter; - - g_return_val_if_fail (mssdemux->manifest_buffer != NULL, FALSE); - g_return_val_if_fail (mssdemux->manifest == NULL, FALSE); - - query = gst_query_new_uri (); - ret = gst_pad_peer_query (mssdemux->sinkpad, query); - if (ret) { - gchar *uri, *redirect_uri; - gboolean permanent; - gchar *baseurl_end; - - gst_query_parse_uri (query, &uri); - gst_query_parse_uri_redirection (query, &redirect_uri); - gst_query_parse_uri_redirection_permanent (query, &permanent); - - GST_INFO_OBJECT (mssdemux, "Upstream is using URI: %s (redirect: %s)", uri, - GST_STR_NULL (redirect_uri)); - - mssdemux->manifest_uri = g_strdup ((permanent - && redirect_uri) ? redirect_uri : uri); - mssdemux->base_url = g_strdup (redirect_uri ? redirect_uri : uri); - baseurl_end = g_strrstr (mssdemux->base_url, "/Manifest"); - if (baseurl_end == NULL) { - /* second try */ - baseurl_end = g_strrstr (mssdemux->base_url, "/manifest"); - } - - if (baseurl_end) { - /* set the new end of the string */ - baseurl_end[0] = '\0'; - } else { - GST_WARNING_OBJECT (mssdemux, "Stream's URI didn't end with /manifest"); - } - - g_free (uri); - g_free (redirect_uri); - } - gst_query_unref (query); + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); + gchar *baseurl_end; +#if 0 if (mssdemux->base_url == NULL) { GST_ELEMENT_ERROR (mssdemux, RESOURCE, NOT_FOUND, (_("Couldn't get the Manifest's URI")), ("need to get the manifest's URI from upstream elements")); return FALSE; } +#endif - GST_INFO_OBJECT (mssdemux, "Received manifest: %i bytes", - (gint) gst_buffer_get_size (mssdemux->manifest_buffer)); + mssdemux->base_url = + g_strdup (demux->manifest_base_uri ? demux->manifest_base_uri : demux-> + manifest_uri); + baseurl_end = g_strrstr (mssdemux->base_url, "/Manifest"); + if (baseurl_end == NULL) { + /* second try */ + baseurl_end = g_strrstr (mssdemux->base_url, "/manifest"); + } + if (baseurl_end) { + /* set the new end of the string */ + baseurl_end[0] = '\0'; + } else { + GST_WARNING_OBJECT (mssdemux, "Stream's URI didn't end with /manifest"); + } - mssdemux->manifest = gst_mss_manifest_new (mssdemux->manifest_buffer); + mssdemux->manifest = gst_mss_manifest_new (buf); if (!mssdemux->manifest) { GST_ELEMENT_ERROR (mssdemux, STREAM, FORMAT, ("Bad manifest file"), ("Xml manifest file couldn't be parsed")); return FALSE; } - - GST_INFO_OBJECT (mssdemux, "Live stream: %d", - gst_mss_manifest_is_live (mssdemux->manifest)); - - gst_mss_demux_create_streams (mssdemux); - for (iter = mssdemux->streams; iter;) { - GSList *current = iter; - GstMssDemuxStream *stream = iter->data; - iter = g_slist_next (iter); /* do it ourselves as we want it done in the beginning of the loop */ - if (!gst_mss_demux_expose_stream (mssdemux, stream)) { - gst_mss_demux_stream_free (stream); - mssdemux->streams = g_slist_delete_link (mssdemux->streams, current); - } - } - - if (!mssdemux->streams) { - /* no streams */ - GST_WARNING_OBJECT (mssdemux, "Couldn't identify the caps for any of the " - "streams found in the manifest"); - GST_ELEMENT_ERROR (mssdemux, STREAM, DEMUX, - (_("This file contains no playable streams.")), - ("No known stream formats found at the Manifest")); - return FALSE; - } - - gst_element_no_more_pads (GST_ELEMENT_CAST (mssdemux)); - return TRUE; + return gst_mss_demux_setup_streams (demux); } +#if 0 static void gst_mss_demux_reload_manifest (GstMssDemux * mssdemux) { + GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (mssdemux); GstUriDownloader *downloader; GstFragment *manifest_data; GstBuffer *manifest_buffer; @@ -1024,21 +499,22 @@ gst_mss_demux_reload_manifest (GstMssDemux * mssdemux) downloader = gst_uri_downloader_new (); manifest_data = - gst_uri_downloader_fetch_uri (downloader, mssdemux->manifest_uri, NULL, + gst_uri_downloader_fetch_uri (downloader, demux->manifest_uri, NULL, TRUE, TRUE, TRUE, NULL); - g_free (mssdemux->manifest_uri); - g_free (mssdemux->base_url); - mssdemux->manifest_uri = g_strdup ((manifest_data->redirect_permanent - && manifest_data-> - redirect_uri) ? manifest_data->redirect_uri : manifest_data->uri); - mssdemux->base_url = - g_strdup (manifest_data-> - redirect_uri ? manifest_data->redirect_uri : manifest_data->uri); - baseurl_end = g_strrstr (mssdemux->base_url, "/Manifest"); + /* FIXME not really nice to unref parent's data */ + g_free (demux->manifest_uri); + g_free (demux->manifest_base_uri); + demux->manifest_uri = g_strdup ((manifest_data->redirect_permanent + && manifest_data->redirect_uri) ? manifest_data-> + redirect_uri : manifest_data->uri); + demux->manifest_base_uri = + g_strdup (manifest_data->redirect_uri ? manifest_data-> + redirect_uri : manifest_data->uri); + baseurl_end = g_strrstr (demux->manifest_base_uri, "/Manifest"); if (baseurl_end == NULL) { /* second try */ - baseurl_end = g_strrstr (mssdemux->base_url, "/manifest"); + baseurl_end = g_strrstr (demux->manifest_base_uri, "/manifest"); } if (baseurl_end) { @@ -1052,644 +528,72 @@ gst_mss_demux_reload_manifest (GstMssDemux * mssdemux) g_object_unref (manifest_data); gst_mss_manifest_reload_fragments (mssdemux->manifest, manifest_buffer); +#if 0 gst_buffer_replace (&mssdemux->manifest_buffer, manifest_buffer); +#endif gst_buffer_unref (manifest_buffer); g_object_unref (downloader); } +#endif -static GstEvent * -gst_mss_demux_reconfigure_stream (GstMssDemuxStream * stream) +static gboolean +gst_mss_demux_stream_select_bitrate (GstAdaptiveDemuxStream * stream, + guint64 bitrate) { - GstEvent *capsevent = NULL; - GstMssDemux *mssdemux = stream->parent; - guint64 bitrate = 0; + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (stream->demux); + GstMssDemuxStream *mssstream = (GstMssDemuxStream *) stream; + gboolean ret = FALSE; - if (stream->download_total_time) - bitrate = - (stream->download_total_bytes * 8) / - ((double) stream->download_total_time / G_GUINT64_CONSTANT (1000000)); - - if (stream->current_download_rate != -1) - bitrate = (stream->current_download_rate + bitrate * 3) / 4; - if (bitrate > G_MAXINT) - bitrate = G_MAXINT; - stream->current_download_rate = bitrate; bitrate *= mssdemux->bitrate_limit; - if (mssdemux->connection_speed) { bitrate = MIN (mssdemux->connection_speed, bitrate); } GST_DEBUG_OBJECT (stream->pad, - "Current stream download bitrate %" G_GUINT64_FORMAT, bitrate); + "Using stream download bitrate %" G_GUINT64_FORMAT, bitrate); - if (gst_mss_stream_select_bitrate (stream->manifest_stream, bitrate)) { + if (gst_mss_stream_select_bitrate (mssstream->manifest_stream, bitrate)) { GstCaps *caps; - caps = gst_mss_stream_get_caps (stream->manifest_stream); + GstCaps *msscaps; + caps = gst_mss_stream_get_caps (mssstream->manifest_stream); GST_DEBUG_OBJECT (stream->pad, "Starting streams reconfiguration due to bitrate changes"); - if (stream->caps) - gst_caps_unref (stream->caps); - stream->caps = create_mss_caps (stream, caps); + msscaps = create_mss_caps (mssstream, caps); GST_DEBUG_OBJECT (stream->pad, "Stream changed bitrate to %" G_GUINT64_FORMAT " caps: %" GST_PTR_FORMAT, - gst_mss_stream_get_current_bitrate (stream->manifest_stream), caps); + gst_mss_stream_get_current_bitrate (mssstream->manifest_stream), caps); gst_caps_unref (caps); - capsevent = gst_event_new_caps (stream->caps); + gst_adaptive_demux_stream_set_caps (stream, msscaps); + ret = TRUE; GST_DEBUG_OBJECT (stream->pad, "Finished streams reconfiguration"); } - return capsevent; -} - -static GstFlowReturn -_src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer) -{ - GstPad *srcpad = (GstPad *) parent; - GstMssDemux *demux = (GstMssDemux *) GST_PAD_PARENT (srcpad); - GstMssDemuxStream *stream = gst_pad_get_element_private (srcpad); - GstFlowReturn ret; - gboolean discont = FALSE; - - if (stream->starting_fragment) { -#if 0 - 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; -#endif - stream->starting_fragment = FALSE; - - GST_BUFFER_PTS (buffer) = - gst_mss_stream_get_fragment_gst_timestamp (stream->manifest_stream); - - GST_LOG_OBJECT (stream->pad, "set fragment pts=%" GST_TIME_FORMAT, - GST_TIME_ARGS (GST_BUFFER_PTS (buffer))); - - if (stream->pending_segment) { - /* And send a newsegment */ - gst_pad_push_event (stream->pad, stream->pending_segment); - stream->pending_segment = NULL; - } - stream->segment.position = GST_BUFFER_PTS (buffer); - - } else { - GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE; - } - - 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); - } - - GST_BUFFER_DURATION (buffer) = GST_CLOCK_TIME_NONE; - GST_BUFFER_DTS (buffer) = GST_CLOCK_TIME_NONE; - - /* accumulate time and size to get this chunk */ - stream->download_total_time += - g_get_monotonic_time () - stream->download_start_time; - stream->download_total_bytes += gst_buffer_get_size (buffer); - - stream->have_data = TRUE; - - ret = gst_proxy_pad_chain_default (pad, parent, buffer); - stream->download_start_time = g_get_monotonic_time (); - GST_LOG_OBJECT (pad, "Chain res: %d %s", ret, gst_flow_get_name (ret)); - - if (ret != GST_FLOW_OK) { - if (ret < GST_FLOW_EOS) { - GST_ELEMENT_ERROR (demux, STREAM, FAILED, (NULL), - ("stream stopped, reason %s", gst_flow_get_name (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)); - } - - /* TODO properly stop tasks */ - /* gst_hls_demux_pause_tasks (demux); */ - g_mutex_lock (&stream->fragment_download_lock); - stream->download_finished = TRUE; - stream->last_ret = ret; - g_cond_signal (&stream->fragment_download_cond); - g_mutex_unlock (&stream->fragment_download_lock); - } - - /* avoid having the source handle the same error again */ - ret = GST_FLOW_OK; - return ret; } static gboolean -_src_event (GstPad * pad, GstObject * parent, GstEvent * event) +gst_mss_demux_seek (GstAdaptiveDemux * demux, GstEvent * seek) { - GstPad *srcpad = GST_PAD_CAST (parent); - GstMssDemuxStream *stream = gst_pad_get_element_private (srcpad); + gdouble rate; + GstFormat format; + GstSeekFlags flags; + GstSeekType start_type, stop_type; + gint64 start, stop; + GstMssDemux *mssdemux = GST_MSS_DEMUX_CAST (demux); - switch (GST_EVENT_TYPE (event)) { - case GST_EVENT_EOS: - 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); - break; - default: - break; - } + gst_event_parse_seek (seek, &rate, &format, &flags, &start_type, &start, + &stop_type, &stop); - gst_event_unref (event); + GST_DEBUG_OBJECT (mssdemux, + "seek event, rate: %f start: %" GST_TIME_FORMAT " stop: %" + GST_TIME_FORMAT, rate, GST_TIME_ARGS (start), GST_TIME_ARGS (stop)); + + gst_mss_manifest_seek (mssdemux->manifest, start); return TRUE; } - -static gboolean -_src_query (GstPad * pad, GstObject * parent, GstQuery * query) -{ - switch (GST_QUERY_TYPE (query)) { - case GST_QUERY_ALLOCATION: - return FALSE; - break; - default: - break; - } - - return gst_pad_query_default (pad, parent, query); -} - -static gboolean -_mss_demux_pad_remove_eos_sticky (GstPad * pad, GstEvent ** event, - gpointer udata) -{ - if (GST_EVENT_TYPE (*event) == GST_EVENT_EOS) { - gst_event_replace (event, NULL); - return FALSE; - } - return TRUE; -} - -static void -gst_mss_demux_stream_clear_eos_state (GstMssDemuxStream * stream) -{ - GstPad *internal_pad; - - internal_pad = - GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (stream->pad))); - gst_pad_sticky_events_foreach (internal_pad, - _mss_demux_pad_remove_eos_sticky, NULL); - GST_OBJECT_FLAG_UNSET (internal_pad, GST_PAD_FLAG_EOS); - - gst_object_unref (internal_pad); -} - -static gboolean -gst_mss_demux_stream_update_source (GstMssDemuxStream * stream, - const gchar * uri, const gchar * referer, gboolean refresh, - gboolean allow_cache) -{ - GstMssDemux *demux = stream->parent; - - if (!gst_uri_is_valid (uri)) { - GST_WARNING_OBJECT (stream->pad, "Invalid URI: %s", uri); - return FALSE; - } - - if (stream->src != NULL) { - gchar *old_protocol, *new_protocol; - gchar *old_uri; - - old_uri = gst_uri_handler_get_uri (GST_URI_HANDLER (stream->src)); - old_protocol = gst_uri_get_protocol (old_uri); - new_protocol = gst_uri_get_protocol (uri); - - if (!g_str_equal (old_protocol, new_protocol)) { - gst_object_unref (stream->src_srcpad); - gst_element_set_state (stream->src, GST_STATE_NULL); - gst_bin_remove (GST_BIN_CAST (demux), stream->src); - stream->src = NULL; - stream->src_srcpad = NULL; - 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->src), uri, &err)) { - GST_DEBUG_OBJECT (demux, "Failed to re-use old source element: %s", - err->message); - g_clear_error (&err); - gst_element_set_state (stream->src, GST_STATE_NULL); - gst_bin_remove (GST_BIN_CAST (demux), stream->src); - stream->src = NULL; - } - } - g_free (old_uri); - g_free (old_protocol); - g_free (new_protocol); - } - - if (stream->src == NULL) { - GObjectClass *gobject_class; - GstPad *internal_pad; - - stream->src = gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL); - if (stream->src == NULL) { - GST_ELEMENT_ERROR (demux, CORE, MISSING_PLUGIN, - ("Missing plugin to handle URI: '%s'", uri), (NULL)); - return FALSE; - } - - gobject_class = G_OBJECT_GET_CLASS (stream->src); - - if (g_object_class_find_property (gobject_class, "compress")) - g_object_set (stream->src, "compress", FALSE, NULL); - if (g_object_class_find_property (gobject_class, "keep-alive")) - g_object_set (stream->src, "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 (stream->src, "extra-headers", extra_headers, NULL); - - gst_structure_free (extra_headers); - } else { - g_object_set (stream->src, "extra-headers", NULL, NULL); - } - } - - 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"); - - gst_ghost_pad_set_target (GST_GHOST_PAD_CAST (stream->pad), - stream->src_srcpad); - - /* set up our internal pad to drop all events from - * the http src we don't care about. On the chain function - * we just push the buffer forward, but this way dash can get - * the flow return from downstream */ - internal_pad = - GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (stream->pad))); - gst_pad_set_chain_function (GST_PAD_CAST (internal_pad), _src_chain); - gst_pad_set_event_function (GST_PAD_CAST (internal_pad), _src_event); - /* need to set query otherwise deadlocks happen with allocation queries */ - gst_pad_set_query_function (GST_PAD_CAST (internal_pad), _src_query); - gst_object_unref (internal_pad); - } - return TRUE; -} - -static GstFlowReturn -gst_mss_demux_stream_download_uri (GstMssDemux * demux, - GstMssDemuxStream * stream, const gchar * uri, gint64 start, gint64 end) -{ - GstFlowReturn ret = GST_FLOW_OK; - GST_DEBUG_OBJECT (stream->pad, "Downloading uri: %s, range:%" G_GINT64_FORMAT - " - %" G_GINT64_FORMAT, uri, start, end); - - stream->download_finished = FALSE; - - if (!gst_mss_demux_stream_update_source (stream, uri, NULL, FALSE, TRUE)) { - g_mutex_lock (&stream->fragment_download_lock); - ret = stream->last_ret = GST_FLOW_ERROR; - g_mutex_unlock (&stream->fragment_download_lock); - return ret; - } - - if (gst_element_set_state (stream->src, - GST_STATE_READY) != GST_STATE_CHANGE_FAILURE) { - if (start != 0 || end != -1) { - if (!gst_element_send_event (stream->src, gst_event_new_seek (1.0, - GST_FORMAT_BYTES, (GstSeekFlags) GST_SEEK_FLAG_FLUSH, - GST_SEEK_TYPE_SET, start, GST_SEEK_TYPE_SET, end))) { - - /* looks like the source can't handle seeks in READY */ -/* - *err = g_error_new (GST_CORE_ERROR, GST_CORE_ERROR_NOT_IMPLEMENTED, - "Source element can't handle range requests"); -*/ - stream->last_ret = GST_FLOW_ERROR; - } - } - - g_mutex_lock (&stream->fragment_download_lock); - if (G_LIKELY (stream->last_ret == GST_FLOW_OK)) { - stream->download_start_time = g_get_monotonic_time (); - - g_mutex_unlock (&stream->fragment_download_lock); - gst_element_sync_state_with_parent (stream->src); - g_mutex_lock (&stream->fragment_download_lock); - - /* wait for the fragment to be completely downloaded */ - GST_DEBUG_OBJECT (stream->pad, - "Waiting for fragment download to finish: %s", uri); - - while (!stream->cancelled && !stream->download_finished) { - g_cond_wait (&stream->fragment_download_cond, - &stream->fragment_download_lock); - } - ret = stream->last_ret; - - GST_DEBUG_OBJECT (stream->pad, "Fragment download finished: %s", uri); - } - g_mutex_unlock (&stream->fragment_download_lock); - } else { - g_mutex_lock (&stream->fragment_download_lock); - ret = stream->last_ret = GST_FLOW_CUSTOM_ERROR; - g_mutex_unlock (&stream->fragment_download_lock); - } - - /* flush the proxypads so that the EOS state is reset */ - gst_pad_push_event (stream->src_srcpad, gst_event_new_flush_start ()); - gst_pad_push_event (stream->src_srcpad, gst_event_new_flush_stop (TRUE)); - gst_element_set_state (stream->src, GST_STATE_READY); - gst_mss_demux_stream_clear_eos_state (stream); - - return ret; -} - -static GstFlowReturn -gst_mss_demux_stream_download_fragment (GstMssDemuxStream * stream) -{ - GstMssDemux *mssdemux = stream->parent; - gchar *path = NULL; - gchar *url = NULL; - GstFlowReturn ret = GST_FLOW_OK; - - GST_DEBUG_OBJECT (stream->pad, "Getting url for stream"); - ret = gst_mss_stream_get_fragment_url (stream->manifest_stream, &path); - switch (ret) { - case GST_FLOW_OK: - break; /* all is good, let's go */ - case GST_FLOW_EOS: - g_free (path); - if (gst_mss_manifest_is_live (mssdemux->manifest)) { - gst_mss_demux_reload_manifest (mssdemux); - return GST_FLOW_OK; - } - return GST_FLOW_EOS; - case GST_FLOW_ERROR: - g_free (path); - goto error; - default: - break; - } - if (!path) { - goto no_url_error; - } - GST_DEBUG_OBJECT (mssdemux, "Got url path '%s' for stream %p", path, stream); - - url = g_strdup_printf ("%s/%s", mssdemux->base_url, path); - - GST_DEBUG_OBJECT (mssdemux, "Got url '%s' for stream %p", url, stream); - - g_mutex_lock (&stream->fragment_download_lock); - stream->last_ret = GST_FLOW_OK; - stream->starting_fragment = TRUE; - g_mutex_unlock (&stream->fragment_download_lock); - - ret = gst_mss_demux_stream_download_uri (mssdemux, stream, url, 0, -1); - - g_free (path); - g_free (url); - - if (ret != GST_FLOW_OK) { - GST_INFO_OBJECT (mssdemux, "No fragment downloaded (%d %s)", ret, - gst_flow_get_name (ret)); - /* TODO check if we are truly stoping */ - if (ret != GST_FLOW_ERROR && gst_mss_manifest_is_live (mssdemux->manifest)) { - /* looks like there is no way of knowing when a live stream has ended - * Have to assume we are falling behind and cause a manifest reload */ - return GST_FLOW_OK; - } - } - - return ret; - -no_url_error: - { - GST_ELEMENT_ERROR (mssdemux, STREAM, DEMUX, - (_("Failed to get fragment URL.")), - ("An error happened when getting fragment URL")); - gst_task_pause (stream->download_task); - return GST_FLOW_ERROR; - } -error: - { - GST_WARNING_OBJECT (mssdemux, "Error while pushing fragment"); - gst_task_pause (stream->download_task); - return GST_FLOW_ERROR; - } -} - -static void -gst_mss_demux_download_loop (GstMssDemuxStream * stream) -{ - GstMssDemux *mssdemux = stream->parent; - GstFlowReturn ret; - GstEvent *gap = NULL; - GstEvent *capsevent = NULL; - - GST_LOG_OBJECT (stream->pad, "download loop start"); - - GST_OBJECT_LOCK (mssdemux); - if (G_UNLIKELY (stream->restart_download)) { - GstClockTime cur, ts; - gint64 pos; - - GST_DEBUG_OBJECT (stream->pad, - "Activating stream due to reconfigure event"); - - cur = stream->segment.position; - - if (gst_pad_peer_query_position (stream->pad, GST_FORMAT_TIME, &pos)) { - ts = (GstClockTime) pos; - GST_DEBUG_OBJECT (mssdemux, "Downstream position: %" - GST_TIME_FORMAT, GST_TIME_ARGS (ts)); - } else { - gboolean have_pos = FALSE; - - /* query other pads as some faulty element in the pad's branch might - * reject position queries. This should be better than using the - * demux segment position that can be much ahead */ - for (GSList * iter = mssdemux->streams; iter != NULL; - iter = g_slist_next (iter)) { - GstMssDemuxStream *cur_stream = iter->data; - - have_pos = - gst_pad_peer_query_position (cur_stream->pad, GST_FORMAT_TIME, - &pos); - if (have_pos) { - ts = (GstClockTime) pos; - GST_DEBUG_OBJECT (stream->pad, "Downstream position: %" - GST_TIME_FORMAT, GST_TIME_ARGS (ts)); - break; - } - } - - if (!have_pos) { - ts = stream->segment.position; - GST_DEBUG_OBJECT (stream->pad, "Downstream position query failed, " - "failling back to looking at other pads"); - } - } - - /* we might have already pushed this data */ - ts = MAX (ts, cur); - - GST_DEBUG_OBJECT (stream->pad, "Restarting stream at " - "position %" GST_TIME_FORMAT, GST_TIME_ARGS (ts)); - - if (GST_CLOCK_TIME_IS_VALID (ts)) { - gst_mss_stream_seek (stream->manifest_stream, ts); - - if (cur < ts) { - gap = gst_event_new_gap (cur, ts - cur); - } - } - - stream->restart_download = FALSE; - } - capsevent = gst_mss_demux_reconfigure_stream (stream); - GST_OBJECT_UNLOCK (mssdemux); - - if (G_UNLIKELY (gap != NULL)) - gst_pad_push_event (stream->pad, gap); - if (G_UNLIKELY (capsevent != NULL)) - gst_pad_push_event (stream->pad, capsevent); - - ret = gst_mss_demux_stream_download_fragment (stream); - - GST_OBJECT_LOCK (mssdemux); - if (stream->cancelled) { - GST_OBJECT_UNLOCK (mssdemux); - stream->last_ret = GST_FLOW_FLUSHING; - goto cancelled; - } - - switch (ret) { - case GST_FLOW_OK: - stream->download_error_count = 0; - stream->segment.position += - gst_mss_stream_get_fragment_gst_duration (stream->manifest_stream); - gst_mss_stream_advance_fragment (stream->manifest_stream); - break; /* all is good, let's go */ - - case GST_FLOW_EOS: - GST_DEBUG_OBJECT (stream->pad, "EOS, stopping download loop"); - /* we push the EOS after releasing the object lock */ - gst_task_pause (stream->download_task); - break; - - case GST_FLOW_NOT_LINKED: - gst_task_pause (stream->download_task); - if (gst_mss_demux_combine_flows (mssdemux) == GST_FLOW_NOT_LINKED) { - GST_ELEMENT_ERROR (mssdemux, STREAM, FAILED, - (_("Internal data stream error.")), - ("stream stopped, reason %s", - gst_flow_get_name (GST_FLOW_NOT_LINKED))); - } - break; - - case GST_FLOW_FLUSHING:{ - GSList *iter; - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *other; - - other = iter->data; - gst_task_pause (other->download_task); - } - } - break; - - default: - if (ret <= GST_FLOW_ERROR) { - GST_WARNING_OBJECT (mssdemux, "Error while downloading fragment"); - if (++stream->download_error_count >= MAX_DOWNLOAD_ERROR_COUNT) { - goto download_error; - } - } - break; - } - GST_OBJECT_UNLOCK (mssdemux); - - if (G_UNLIKELY (ret == GST_FLOW_EOS)) { - gst_mss_demux_stream_push_event (stream, gst_event_new_eos ()); - } - -end: - GST_LOG_OBJECT (stream->pad, "download loop end"); - return; - -cancelled: - { - GST_DEBUG_OBJECT (stream->pad, "Stream has been cancelled"); - goto end; - } -download_error: - { - GST_OBJECT_UNLOCK (mssdemux); - GST_ELEMENT_ERROR (mssdemux, RESOURCE, NOT_FOUND, - (_("Couldn't download fragments")), - ("fragment downloading has failed too much consecutive times")); - goto end; - } -} - -static GstFlowReturn -gst_mss_demux_combine_flows (GstMssDemux * mssdemux) -{ - gboolean all_notlinked = TRUE; - GSList *iter; - - for (iter = mssdemux->streams; iter; iter = g_slist_next (iter)) { - GstMssDemuxStream *stream = iter->data; - - g_mutex_lock (&stream->fragment_download_lock); - if (stream->last_ret != GST_FLOW_NOT_LINKED) - all_notlinked = FALSE; - - if (stream->last_ret <= GST_FLOW_NOT_NEGOTIATED - || stream->last_ret == GST_FLOW_FLUSHING) { - g_mutex_unlock (&stream->fragment_download_lock); - return stream->last_ret; - } - g_mutex_unlock (&stream->fragment_download_lock); - } - if (all_notlinked) - return GST_FLOW_NOT_LINKED; - return GST_FLOW_OK; -} - -static gboolean -gst_mss_demux_stream_push_event (GstMssDemuxStream * stream, GstEvent * event) -{ - gboolean ret; - - if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) { - stream->eos = TRUE; - } - GST_DEBUG_OBJECT (stream->pad, "Pushing event %" GST_PTR_FORMAT, event); - ret = gst_pad_push_event (stream->pad, event); - return ret; -} diff --git a/ext/smoothstreaming/gstmssdemux.h b/ext/smoothstreaming/gstmssdemux.h index 4e243fc47a..fe8a49d344 100644 --- a/ext/smoothstreaming/gstmssdemux.h +++ b/ext/smoothstreaming/gstmssdemux.h @@ -24,6 +24,7 @@ #define __GST_MSSDEMUX_H__ #include +#include #include #include #include "gstmssmanifest.h" @@ -52,63 +53,20 @@ typedef struct _GstMssDemux GstMssDemux; typedef struct _GstMssDemuxClass GstMssDemuxClass; struct _GstMssDemuxStream { - GstPad *pad; - - GstCaps *caps; - - GstMssDemux *parent; + GstAdaptiveDemuxStream parent; GstMssStream *manifest_stream; - -#if 0 - GstUriDownloader *downloader; -#endif - - GstEvent *pending_segment; - - GstSegment segment; - - /* Downloading task */ - GstTask *download_task; - GRecMutex download_lock; - - GstFlowReturn last_ret; - gboolean eos; - gboolean have_data; - gboolean cancelled; - gboolean restart_download; - - guint download_error_count; - - /* download tooling */ - GstElement *src; - GstPad *src_srcpad; - GMutex fragment_download_lock; - GCond fragment_download_cond; - gboolean download_finished; - gboolean starting_fragment; - gint64 download_start_time; - gint64 download_total_time; - gint64 download_total_bytes; - gint current_download_rate; }; struct _GstMssDemux { - GstBin bin; + GstAdaptiveDemux bin; /* pads */ GstPad *sinkpad; - gboolean have_group_id; - guint group_id; - - GstBuffer *manifest_buffer; - GstMssManifest *manifest; gchar *base_url; - gchar *manifest_uri; - GSList *streams; guint n_videos; guint n_audios; @@ -121,7 +79,7 @@ struct _GstMssDemux { }; struct _GstMssDemuxClass { - GstBinClass parent_class; + GstAdaptiveDemuxClass parent_class; }; GType gst_mss_demux_get_type (void); diff --git a/ext/smoothstreaming/gstmssmanifest.c b/ext/smoothstreaming/gstmssmanifest.c index 9f2c681bfa..1f7ab1d3b2 100644 --- a/ext/smoothstreaming/gstmssmanifest.c +++ b/ext/smoothstreaming/gstmssmanifest.c @@ -878,6 +878,20 @@ gst_mss_stream_advance_fragment (GstMssStream * stream) return GST_FLOW_OK; } +GstFlowReturn +gst_mss_stream_regress_fragment (GstMssStream * stream) +{ + g_return_val_if_fail (stream->active, GST_FLOW_ERROR); + + if (stream->current_fragment == NULL) + return GST_FLOW_EOS; + + stream->current_fragment = g_list_previous (stream->current_fragment); + if (stream->current_fragment == NULL) + return GST_FLOW_EOS; + return GST_FLOW_OK; +} + const gchar * gst_mss_stream_type_name (GstMssStreamType streamtype) { diff --git a/ext/smoothstreaming/gstmssmanifest.h b/ext/smoothstreaming/gstmssmanifest.h index 9969f77fca..e4d4f8ed9f 100644 --- a/ext/smoothstreaming/gstmssmanifest.h +++ b/ext/smoothstreaming/gstmssmanifest.h @@ -60,6 +60,7 @@ GstFlowReturn gst_mss_stream_get_fragment_url (GstMssStream * stream, gchar ** u GstClockTime gst_mss_stream_get_fragment_gst_timestamp (GstMssStream * stream); GstClockTime gst_mss_stream_get_fragment_gst_duration (GstMssStream * stream); GstFlowReturn gst_mss_stream_advance_fragment (GstMssStream * stream); +GstFlowReturn gst_mss_stream_regress_fragment (GstMssStream * stream); void gst_mss_stream_seek (GstMssStream * stream, guint64 time); const gchar * gst_mss_stream_get_lang (GstMssStream * stream);