adaptivedemux: handle snap seeks

Adaptive demuxers need to start downloading from specific positions
(fragments) for every stream, this means that all streams can snap-seek
to a different position when requested. Snap seeking in this case will
be done in 2 steps:

1) do the snap seeking on the pad that received the seek event and
   get the final position

2) use this position to do a regular seek on the other streams to
   make sure they all start from the same position

More arguments were added to the stream_seek function, allowing better control
of how seeking is done. Knowing the flags and the playback direction allows
subclasses to handle snap-seeking.
And also adds a new return parameter to inform of the final
selected seeking position that is used to align the other streams.

https://bugzilla.gnome.org/show_bug.cgi?id=759158

ext/dash/gstdashdemux.c

@@ -233,19 +233,20 @@ static gboolean gst_dash_demux_seek (GstAdaptiveDemux * demux, GstEvent * seek);
 static GstFlowReturn
 gst_dash_demux_stream_update_fragment_info (GstAdaptiveDemuxStream * stream);
 static GstFlowReturn gst_dash_demux_stream_seek (GstAdaptiveDemuxStream *
-    stream, GstClockTime ts);
-static gboolean
-gst_dash_demux_stream_has_next_fragment (GstAdaptiveDemuxStream * stream);
+    stream, gboolean forward, GstSeekFlags flags, GstClockTime ts,
+    GstClockTime * final_ts);
+static gboolean gst_dash_demux_stream_has_next_fragment (GstAdaptiveDemuxStream
+    * stream);
 static GstFlowReturn
 gst_dash_demux_stream_advance_fragment (GstAdaptiveDemuxStream * stream);
 static gboolean
 gst_dash_demux_stream_advance_subfragment (GstAdaptiveDemuxStream * stream);
 static gboolean gst_dash_demux_stream_select_bitrate (GstAdaptiveDemuxStream *
     stream, guint64 bitrate);
-static gint64
-gst_dash_demux_get_manifest_update_interval (GstAdaptiveDemux * demux);
-static GstFlowReturn
-gst_dash_demux_update_manifest_data (GstAdaptiveDemux * demux, GstBuffer * buf);
+static gint64 gst_dash_demux_get_manifest_update_interval (GstAdaptiveDemux *
+    demux);
+static GstFlowReturn gst_dash_demux_update_manifest_data (GstAdaptiveDemux *
+    demux, GstBuffer * buf);
 static gint64
 gst_dash_demux_stream_get_fragment_waiting_time (GstAdaptiveDemuxStream *
     stream);
@@ -1066,7 +1067,7 @@ gst_dash_demux_index_entry_search (GstSidxBoxEntry * entry, GstClockTime * ts,
 
 static void
 gst_dash_demux_stream_sidx_seek (GstDashDemuxStream * dashstream,
-    GstClockTime ts)
+    GstSeekFlags flags, GstClockTime ts, GstClockTime * final_ts)
 {
   GstSidxBox *sidx = SIDX (dashstream);
   GstSidxBoxEntry *entry;
@@ -1088,24 +1089,34 @@ gst_dash_demux_stream_sidx_seek (GstDashDemuxStream * dashstream,
 
   sidx->entry_index = idx;
   dashstream->sidx_index = idx;
+
+  if (final_ts) {
+    if (idx == sidx->entries_count)
+      *final_ts = sidx->entries[idx].pts + sidx->entries[idx].duration;
+    else
+      *final_ts = sidx->entries[idx].pts;
+  }
 }
 
 static GstFlowReturn
-gst_dash_demux_stream_seek (GstAdaptiveDemuxStream * stream, GstClockTime ts)
+gst_dash_demux_stream_seek (GstAdaptiveDemuxStream * stream, gboolean forward,
+    GstSeekFlags flags, GstClockTime ts, GstClockTime * final_ts)
 {
   GstDashDemuxStream *dashstream = (GstDashDemuxStream *) stream;
   GstDashDemux *dashdemux = GST_DASH_DEMUX_CAST (stream->demux);
 
   if (gst_mpd_client_has_isoff_ondemand_profile (dashdemux->client)) {
     if (dashstream->sidx_parser.status == GST_ISOFF_SIDX_PARSER_FINISHED) {
-      gst_dash_demux_stream_sidx_seek (dashstream, ts);
+      gst_dash_demux_stream_sidx_seek (dashstream, flags, ts, final_ts);
     } else {
       /* no index yet, seek when we have it */
+      /* FIXME - the final_ts won't be correct here */
       dashstream->pending_seek_ts = ts;
     }
   }
 
-  gst_mpd_client_stream_seek (dashdemux->client, dashstream->active_stream, ts);
+  gst_mpd_client_stream_seek (dashdemux->client, dashstream->active_stream,
+      forward, flags, ts, final_ts);
   return GST_FLOW_OK;
 }
 
@@ -1340,7 +1351,7 @@ gst_dash_demux_seek (GstAdaptiveDemux * demux, GstEvent * seek)
       gst_isoff_sidx_parser_clear (&dashstream->sidx_parser);
       gst_isoff_sidx_parser_init (&dashstream->sidx_parser);
     }
-    gst_dash_demux_stream_seek (iter->data, target_pos);
+    gst_dash_demux_stream_seek (iter->data, rate >= 0, 0, target_pos, NULL);
   }
   return TRUE;
 }
@@ -1438,7 +1449,8 @@ gst_dash_demux_update_manifest_data (GstAdaptiveDemux * demux,
             GST_TIME_FORMAT, GST_TIME_ARGS (ts),
             GST_TIME_ARGS (ts + (10 * GST_USECOND)));
         ts += 10 * GST_USECOND;
-        gst_mpd_client_stream_seek (new_client, new_stream, ts);
+        gst_mpd_client_stream_seek (new_client, new_stream,
+            demux->segment.rate >= 0, 0, ts, NULL);
       }
 
       demux_stream->active_stream = new_stream;
@@ -1598,8 +1610,9 @@ gst_dash_demux_data_received (GstAdaptiveDemux * demux,
         /* when finished, prepare for real data streaming */
         if (dash_stream->sidx_parser.status == GST_ISOFF_SIDX_PARSER_FINISHED) {
           if (GST_CLOCK_TIME_IS_VALID (dash_stream->pending_seek_ts)) {
-            gst_dash_demux_stream_sidx_seek (dash_stream,
-                dash_stream->pending_seek_ts);
+            /* FIXME, preserve seek flags */
+            gst_dash_demux_stream_sidx_seek (dash_stream, 0,
+                dash_stream->pending_seek_ts, NULL);
             dash_stream->pending_seek_ts = GST_CLOCK_TIME_NONE;
           } else {
             SIDX (dash_stream)->entry_index = dash_stream->sidx_index;

ext/dash/gstmpdparser.c

@@ -4652,7 +4652,8 @@ gst_mpd_client_setup_streaming (GstMpdClient * client,
 
 gboolean
 gst_mpd_client_stream_seek (GstMpdClient * client, GstActiveStream * stream,
-    GstClockTime ts)
+    gboolean forward, GstSeekFlags flags, GstClockTime ts,
+    GstClockTime * final_ts)
 {
   gint index = 0;
   gint repeat_index = 0;
@@ -4692,6 +4693,9 @@ gst_mpd_client_stream_seek (GstMpdClient * client, GstActiveStream * stream,
       GST_DEBUG ("Seek to after last segment");
       return FALSE;
     }
+
+    if (final_ts)
+      *final_ts = selectedChunk->start + selectedChunk->duration * repeat_index;
   } else {
     GstClockTime duration =
         gst_mpd_client_get_segment_duration (client, stream, NULL);
@@ -4716,6 +4720,8 @@ gst_mpd_client_stream_seek (GstMpdClient * client, GstActiveStream * stream,
       GST_DEBUG ("Seek to after last segment");
       return FALSE;
     }
+    if (final_ts)
+      *final_ts = index * duration;
   }
 
   stream->segment_repeat_index = repeat_index;
@@ -5816,7 +5822,9 @@ gst_mpd_client_seek_to_time (GstMpdClient * client, GDateTime * time)
 
   ts = ts_microseconds * GST_USECOND;
   for (stream = client->active_streams; stream; stream = g_list_next (stream)) {
-    ret = ret & gst_mpd_client_stream_seek (client, stream->data, ts);
+    ret =
+        ret & gst_mpd_client_stream_seek (client, stream->data, TRUE, 0, ts,
+        NULL);
   }
   return ret;
 }

ext/dash/gstmpdparser.h

@@ -544,7 +544,7 @@ gboolean gst_mpd_client_get_next_fragment (GstMpdClient *client, guint indexStre
 gboolean gst_mpd_client_get_next_header (GstMpdClient *client, gchar **uri, guint stream_idx, gint64 * range_start, gint64 * range_end);
 gboolean gst_mpd_client_get_next_header_index (GstMpdClient *client, gchar **uri, guint stream_idx, gint64 * range_start, gint64 * range_end);
 gboolean gst_mpd_client_is_live (GstMpdClient * client);
-gboolean gst_mpd_client_stream_seek (GstMpdClient * client, GstActiveStream * stream, GstClockTime ts);
+gboolean gst_mpd_client_stream_seek (GstMpdClient * client, GstActiveStream * stream, gboolean forward, GstSeekFlags flags, GstClockTime ts, GstClockTime * final_ts);
 gboolean gst_mpd_client_seek_to_time (GstMpdClient * client, GDateTime * time);
 GstClockTime gst_mpd_parser_get_stream_presentation_offset (GstMpdClient *client, guint stream_idx);
 gchar** gst_mpd_client_get_utc_timing_sources (GstMpdClient *client, guint methods, GstMPDUTCTimingType *selected_method);

ext/smoothstreaming/gstmssdemux.c

@@ -122,9 +122,10 @@ static gboolean gst_mss_demux_process_manifest (GstAdaptiveDemux * demux,
 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 gboolean
-gst_mss_demux_stream_has_next_fragment (GstAdaptiveDemuxStream * stream);
+    gboolean forward, GstSeekFlags flags, GstClockTime ts,
+    GstClockTime * final_ts);
+static gboolean gst_mss_demux_stream_has_next_fragment (GstAdaptiveDemuxStream *
+    stream);
 static GstFlowReturn
 gst_mss_demux_stream_advance_fragment (GstAdaptiveDemuxStream * stream);
 static gboolean gst_mss_demux_stream_select_bitrate (GstAdaptiveDemuxStream *
@@ -304,11 +305,12 @@ gst_mss_demux_stream_update_fragment_info (GstAdaptiveDemuxStream * stream)
 }
 
 static GstFlowReturn
-gst_mss_demux_stream_seek (GstAdaptiveDemuxStream * stream, GstClockTime ts)
+gst_mss_demux_stream_seek (GstAdaptiveDemuxStream * stream, gboolean forward,
+    GstSeekFlags flags, GstClockTime ts, GstClockTime * final_ts)
 {
   GstMssDemuxStream *mssstream = (GstMssDemuxStream *) stream;
 
-  gst_mss_stream_seek (mssstream->manifest_stream, ts);
+  gst_mss_stream_seek (mssstream->manifest_stream, flags, ts, final_ts);
   return GST_FLOW_OK;
 }
 

ext/smoothstreaming/gstmssmanifest.c

@@ -1097,7 +1097,7 @@ gst_mss_manifest_seek (GstMssManifest * manifest, guint64 time)
   GSList *iter;
 
   for (iter = manifest->streams; iter; iter = g_slist_next (iter)) {
-    gst_mss_stream_seek (iter->data, time);
+    gst_mss_stream_seek (iter->data, 0, time, NULL);
   }
 }
 
@@ -1107,7 +1107,8 @@ gst_mss_manifest_seek (GstMssManifest * manifest, guint64 time)
  * @time: time in nanoseconds
  */
 void
-gst_mss_stream_seek (GstMssStream * stream, guint64 time)
+gst_mss_stream_seek (GstMssStream * stream, GstSeekFlags flags, guint64 time,
+    guint64 * final_time)
 {
   GList *iter;
   guint64 timescale;
@@ -1148,6 +1149,19 @@ gst_mss_stream_seek (GstMssStream * stream, guint64 time)
   GST_DEBUG ("Stream %s seeked to fragment time %" G_GUINT64_FORMAT
       " repetition %u", stream->url, fragment->time,
       stream->fragment_repetition_index);
+  if (final_time) {
+    if (fragment)
+      *final_time =
+          fragment->time +
+          stream->fragment_repetition_index * fragment->duration;
+    else {
+      /* always stops on the last one */
+      GstMssStreamFragment *last_fragment = iter->data;
+      *final_time =
+          last_fragment->time +
+          last_fragment->repetitions * last_fragment->duration;
+    }
+  }
 }
 
 guint64
@@ -1200,7 +1214,7 @@ gst_mss_stream_reload_fragments (GstMssStream * stream, xmlNodePtr streamIndex)
     g_list_free_full (stream->fragments, g_free);
     stream->fragments = g_list_reverse (builder.fragments);
     stream->current_fragment = stream->fragments;
-    gst_mss_stream_seek (stream, current_gst_time);
+    gst_mss_stream_seek (stream, 0, current_gst_time, NULL);
   }
 }
 

ext/smoothstreaming/gstmssmanifest.h

@@ -67,7 +67,7 @@ GstClockTime gst_mss_stream_get_fragment_gst_duration (GstMssStream * stream);
 gboolean gst_mss_stream_has_next_fragment (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);
+void gst_mss_stream_seek (GstMssStream * stream, GstSeekFlags flags, guint64 time, guint64 * final_time);
 const gchar * gst_mss_stream_get_lang (GstMssStream * stream);
 
 const gchar * gst_mss_stream_type_name (GstMssStreamType streamtype);

gst-libs/gst/adaptivedemux/gstadaptivedemux.c

@@ -230,7 +230,8 @@ static gboolean gst_adaptive_demux_expose_streams (GstAdaptiveDemux * demux,
     gboolean first_and_live);
 static gboolean gst_adaptive_demux_is_live (GstAdaptiveDemux * demux);
 static GstFlowReturn gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux,
-    GstAdaptiveDemuxStream * stream, GstClockTime ts);
+    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 *
@@ -1205,6 +1206,13 @@ gst_adaptive_demux_can_seek (GstAdaptiveDemux * demux)
   return klass->seek != NULL;
 }
 
+#define IS_SNAP_SEEK(f) (f & (GST_SEEK_FLAG_SNAP_BEFORE | \
+                              GST_SEEK_FLAG_SNAP_AFTER | \
+                              GST_SEEK_FLAG_SNAP_NEAREST))
+#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_src_event (GstPad * pad, GstObject * parent,
     GstEvent * event)
@@ -1215,6 +1223,8 @@ gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent,
   demux = GST_ADAPTIVE_DEMUX_CAST (parent);
   demux_class = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
 
+  /* FIXME handle events received on pads that are to be removed */
+
   switch (event->type) {
     case GST_EVENT_SEEK:
     {
@@ -1270,10 +1280,7 @@ gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent,
 
       seqnum = gst_event_get_seqnum (event);
 
-      GST_DEBUG_OBJECT (demux,
-          "seek event, rate: %f type: %d start: %" GST_TIME_FORMAT " stop: %"
-          GST_TIME_FORMAT, rate, start_type, GST_TIME_ARGS (start),
-          GST_TIME_ARGS (stop));
+      GST_DEBUG_OBJECT (demux, "seek event, %" GST_PTR_FORMAT, event);
 
       /* have a backup in case seek fails */
       gst_segment_copy_into (&demux->segment, &oldsegment);
@@ -1309,6 +1316,52 @@ gst_adaptive_demux_src_event (GstPad * pad, GstObject * parent,
       GST_DEBUG_OBJECT (demux, "Seeking to segment %" GST_SEGMENT_FORMAT,
           &demux->segment);
 
+      /*
+       * 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) {
+        GstAdaptiveDemuxStream *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 && start_type != GST_SEEK_TYPE_NONE) {
+          ts = start;
+        } else if (rate < 0 && stop_type != GST_SEEK_TYPE_NONE) {
+          ts = stop;
+        }
+
+        ret =
+            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_type != GST_SEEK_TYPE_NONE) {
+          start = ts;
+        } else if (rate < 0 && stop_type != GST_SEEK_TYPE_NONE) {
+          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);
+      }
+      GST_DEBUG_OBJECT (demux, "Calling subclass seek: %" GST_PTR_FORMAT,
+          event);
       ret = demux_class->seek (demux, event);
 
       if (!ret) {
@@ -2686,7 +2739,8 @@ gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream * stream)
       period_start = gst_adaptive_demux_get_period_start_time (demux);
 
       /* TODO check return */
-      gst_adaptive_demux_stream_seek (demux, stream, ts);
+      gst_adaptive_demux_stream_seek (demux, stream, demux->segment.rate >= 0,
+          0, ts, &ts);
 
       segment.position = ts - period_start + offset;
     }
@@ -3050,12 +3104,13 @@ gst_adaptive_demux_is_live (GstAdaptiveDemux * demux)
 /* must be called with manifest_lock taken */
 static GstFlowReturn
 gst_adaptive_demux_stream_seek (GstAdaptiveDemux * demux,
-    GstAdaptiveDemuxStream * stream, GstClockTime ts)
+    GstAdaptiveDemuxStream * stream, gboolean forward, GstSeekFlags flags,
+    GstClockTime ts, GstClockTime * final_ts)
 {
   GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
 
   if (klass->stream_seek)
-    return klass->stream_seek (stream, ts);
+    return klass->stream_seek (stream, forward, flags, ts, final_ts);
   return GST_FLOW_ERROR;
 }
 

gst-libs/gst/adaptivedemux/gstadaptivedemux.h

@@ -310,7 +310,7 @@ struct _GstAdaptiveDemuxClass
   void          (*advance_period)  (GstAdaptiveDemux * demux);
 
   void          (*stream_free)     (GstAdaptiveDemuxStream * stream);
-  GstFlowReturn (*stream_seek)     (GstAdaptiveDemuxStream * stream, GstClockTime ts);
+  GstFlowReturn (*stream_seek)     (GstAdaptiveDemuxStream * stream, gboolean forward, GstSeekFlags flags, GstClockTime target_ts, GstClockTime * final_ts);
   gboolean      (*stream_has_next_fragment)  (GstAdaptiveDemuxStream * stream);
   GstFlowReturn (*stream_advance_fragment) (GstAdaptiveDemuxStream * stream);
   /**