!!WIP tsdemux GAP supporttsdemux-gap

3 files changed, 113 insertions, 22 deletions

diff --git a/gst/mpegtsdemux/mpegtspacketizer.c b/gst/mpegtsdemux/mpegtspacketizer.c
index c258bb9f9..ba2321618 100644
--- a/gst/mpegtsdemux/mpegtspacketizer.c
+++ b/gst/mpegtsdemux/mpegtspacketizer.c
@@ -118,6 +118,19 @@ flush_observations (MpegTSPacketizer2 * packetizer)
   packetizer->lastobsid = 0;
 }
 
+GstClockTime
+mpegts_packetizer_get_current_time (MpegTSPacketizer2 * packetizer,
+    guint16 pcr_pid)
+{
+  MpegTSPCR *pcrtable = get_pcr_table (packetizer, pcr_pid);
+
+  if (pcrtable == NULL)
+    return GST_CLOCK_TIME_NONE;
+
+  return mpegts_packetizer_pts_to_ts (packetizer, pcrtable->last_pcrtime,
+      pcr_pid);
+}
+
 static inline MpegTSPacketizerStreamSubtable *
 find_subtable (GSList * subtables, guint8 table_id, guint16 subtable_extension)
 {
@@ -1849,6 +1862,7 @@ record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
 
   packetizer->nb_seen_offsets += 1;
 
+  pcrtable->last_pcrtime = PCRTIME_TO_GSTTIME (pcr);
   /* FIXME : Invert logic later (probability is higher that we have a
    * current estimator) */
 
diff --git a/gst/mpegtsdemux/mpegtspacketizer.h b/gst/mpegtsdemux/mpegtspacketizer.h
index a453bd90a..cb3202739 100644
--- a/gst/mpegtsdemux/mpegtspacketizer.h
+++ b/gst/mpegtsdemux/mpegtspacketizer.h
@@ -363,6 +363,9 @@ mpegts_packetizer_ts_to_offset (MpegTSPacketizer2 * packetizer,
 G_GNUC_INTERNAL GstClockTime
 mpegts_packetizer_pts_to_ts (MpegTSPacketizer2 * packetizer,
 			     GstClockTime pts, guint16 pcr_pid);
+G_GNUC_INTERNAL GstClockTime
+mpegts_packetizer_get_current_time (MpegTSPacketizer2 * packetizer,
+				    guint16 pcr_pid);
 G_GNUC_INTERNAL void
 mpegts_packetizer_set_current_pcr_offset (MpegTSPacketizer2 * packetizer,
 			  GstClockTime offset, guint16 pcr_pid);
diff --git a/gst/mpegtsdemux/tsdemux.c b/gst/mpegtsdemux/tsdemux.c
index ed78fd1e4..b4173ccba 100644
--- a/gst/mpegtsdemux/tsdemux.c
+++ b/gst/mpegtsdemux/tsdemux.c
@@ -142,6 +142,13 @@ struct _TSDemuxStream
   GstClockTime pts;
   GstClockTime dts;
 
+  /* Reference PTS used to detect gaps */
+  GstClockTime gap_ref_pts;
+  /* Number of outputted buffers */
+  guint32 nb_out_buffers;
+  /* Reference number of buffers for gaps */
+  guint32 gap_ref_buffers;
+
   /* Current PTS/DTS for this stream (in 90kHz unit) */
   guint64 raw_pts, raw_dts;
 
@@ -1159,8 +1166,11 @@ gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
     stream->need_newsegment = TRUE;
     stream->pts = GST_CLOCK_TIME_NONE;
     stream->dts = GST_CLOCK_TIME_NONE;
+    stream->gap_ref_pts = GST_CLOCK_TIME_NONE;
     stream->raw_pts = -1;
     stream->raw_dts = -1;
+    stream->nb_out_buffers = 0;
+    stream->gap_ref_buffers = 0;
     stream->pending_ts = TRUE;
     stream->first_dts = GST_CLOCK_TIME_NONE;
     stream->continuity_counter = CONTINUITY_UNSET;
@@ -1237,9 +1247,12 @@ gst_ts_demux_stream_flush (TSDemuxStream * stream)
   stream->need_newsegment = TRUE;
   stream->pts = GST_CLOCK_TIME_NONE;
   stream->dts = GST_CLOCK_TIME_NONE;
+  stream->gap_ref_pts = GST_CLOCK_TIME_NONE;
   stream->first_dts = GST_CLOCK_TIME_NONE;
   stream->raw_pts = -1;
   stream->raw_dts = -1;
+  stream->nb_out_buffers = 0;
+  stream->gap_ref_buffers = 0;
   if (stream->flow_return == GST_FLOW_FLUSHING) {
     stream->flow_return = GST_FLOW_OK;
   }
@@ -1264,6 +1277,7 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
 
   if (demux->requested_program_number == program->program_number ||
       (demux->requested_program_number == -1 && demux->program_number == -1)) {
+    GList *tmp;
 
     GST_LOG ("program %d started", program->program_number);
     demux->program_number = program->program_number;
@@ -1274,6 +1288,11 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
     demux->calculate_update_segment = !program->initial_program;
 
     /* FIXME : When do we emit no_more_pads ? */
+    for (tmp = program->stream_list; tmp; tmp = tmp->next) {
+      TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+      activate_pad_for_stream (demux, stream);
+    }
+
   }
 }
 
@@ -1710,32 +1729,37 @@ calculate_and_push_newsegment (GstTSDemux * demux, TSDemuxStream * stream)
   }
 
 push_new_segment:
-  if (demux->update_segment) {
-    GST_DEBUG_OBJECT (stream->pad, "Pushing update segment");
-    gst_event_ref (demux->update_segment);
-    gst_pad_push_event (stream->pad, demux->update_segment);
-  }
+  for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
+    stream = (TSDemuxStream *) tmp->data;
+    if (stream->pad == NULL)
+      continue;
+    if (demux->update_segment) {
+      GST_DEBUG_OBJECT (stream->pad, "Pushing update segment");
+      gst_event_ref (demux->update_segment);
+      gst_pad_push_event (stream->pad, demux->update_segment);
+    }
 
-  if (demux->segment_event) {
-    GST_DEBUG_OBJECT (stream->pad, "Pushing newsegment event");
-    gst_event_ref (demux->segment_event);
-    gst_pad_push_event (stream->pad, demux->segment_event);
-  }
+    if (demux->segment_event) {
+      GST_DEBUG_OBJECT (stream->pad, "Pushing newsegment event");
+      gst_event_ref (demux->segment_event);
+      gst_pad_push_event (stream->pad, demux->segment_event);
+    }
 
-  if (demux->global_tags) {
-    gst_pad_push_event (stream->pad,
-        gst_event_new_tag (gst_tag_list_ref (demux->global_tags)));
-  }
+    if (demux->global_tags) {
+      gst_pad_push_event (stream->pad,
+          gst_event_new_tag (gst_tag_list_ref (demux->global_tags)));
+    }
 
-  /* Push pending tags */
-  if (stream->taglist) {
-    GST_DEBUG_OBJECT (stream->pad, "Sending tags %" GST_PTR_FORMAT,
-        stream->taglist);
-    gst_pad_push_event (stream->pad, gst_event_new_tag (stream->taglist));
-    stream->taglist = NULL;
-  }
+    /* Push pending tags */
+    if (stream->taglist) {
+      GST_DEBUG_OBJECT (stream->pad, "Sending tags %" GST_PTR_FORMAT,
+          stream->taglist);
+      gst_pad_push_event (stream->pad, gst_event_new_tag (stream->taglist));
+      stream->taglist = NULL;
+    }
 
-  stream->need_newsegment = FALSE;
+    stream->need_newsegment = FALSE;
+  }
 }
 
 static GstFlowReturn
@@ -1803,6 +1827,7 @@ gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
           GST_TIME_ARGS (GST_BUFFER_DTS (pend->buffer)));
 
       res = gst_pad_push (stream->pad, pend->buffer);
+      stream->nb_out_buffers += 1;
       g_slice_free (PendingBuffer, pend);
     }
     g_list_free (stream->pending);
@@ -1822,10 +1847,59 @@ gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
       GST_TIME_ARGS (GST_BUFFER_DTS (buffer)));
 
   res = gst_pad_push (stream->pad, buffer);
+  stream->nb_out_buffers += 1;
   GST_DEBUG_OBJECT (stream->pad, "Returned %s", gst_flow_get_name (res));
   res = tsdemux_combine_flows (demux, stream, res);
   GST_DEBUG_OBJECT (stream->pad, "combined %s", gst_flow_get_name (res));
 
+  if (G_UNLIKELY (stream->gap_ref_pts == GST_CLOCK_TIME_NONE))
+    stream->gap_ref_pts = stream->pts;
+  else {
+    if (G_UNLIKELY (stream->pts > stream->gap_ref_pts + 2 * GST_SECOND)) {
+      GstClockTime curpcr =
+          mpegts_packetizer_get_current_time (MPEG_TS_BASE_PACKETIZER (demux),
+          demux->program->pcr_pid);
+      GList *tmp;
+      if (curpcr == GST_CLOCK_TIME_NONE || curpcr < 800 * GST_MSECOND)
+        goto beach;
+      curpcr -= 800 * GST_MSECOND;
+      GST_WARNING ("Recheck streams ? now : %" GST_TIME_FORMAT,
+          GST_TIME_ARGS (curpcr));
+      for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
+        TSDemuxStream *ps = (TSDemuxStream *) tmp->data;
+        GST_WARNING_OBJECT (ps->pad,
+            "0x%04x, PTS:%" GST_TIME_FORMAT " REFPTS:%" GST_TIME_FORMAT " Gap:%"
+            GST_TIME_FORMAT " nb_buffers: %d (ref:%d)",
+            ((MpegTSBaseStream *) ps)->pid, GST_TIME_ARGS (ps->pts),
+            GST_TIME_ARGS (ps->gap_ref_pts),
+            GST_TIME_ARGS (ps->pts - ps->gap_ref_pts), ps->nb_out_buffers,
+            ps->gap_ref_buffers);
+        if (ps->pad == NULL)
+          continue;
+        if (ps->nb_out_buffers == ps->gap_ref_buffers) {
+          GST_WARNING_OBJECT (ps->pad,
+              "No buffers pushed yet, we should push GAP event");
+          gst_pad_push_event (ps->pad, gst_event_new_gap (curpcr, 0));
+          ps->gap_ref_pts = curpcr;
+          /* } */
+          /* if (ps->pts == GST_CLOCK_TIME_NONE) { */
+          /*   GST_WARNING_OBJECT (ps->pad, */
+          /*       "No buffers seen yet, we should push GAP event"); */
+          /*   gst_pad_push_event (ps->pad, gst_event_new_gap (curpcr, 0)); */
+          /*   ps->gap_ref_pts = curpcr; */
+          /* } else if (ps->gap_ref_pts == ps->pts) { */
+          /*   GST_WARNING_OBJECT (ps->pad, */
+          /*       "No buffers pushed yet, we should push GAP event"); */
+          /*   gst_pad_push_event (ps->pad, gst_event_new_gap (curpcr, 0)); */
+          /*   ps->gap_ref_pts = curpcr; */
+        } else {
+          ps->gap_ref_pts = ps->pts;
+          ps->gap_ref_buffers = ps->nb_out_buffers;
+        }
+      }
+    }
+  }
+
 beach:
   /* Reset everything */
   GST_LOG ("Resetting to EMPTY, returning %s", gst_flow_get_name (res));