mpegtsdemux: Implement efficient program updates

If the parent bin can handle it, only add/remove the new/gone stream
instead of re-adding/re-moving everything

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

gst/mpegtsdemux/mpegtsbase.c

@@ -210,6 +210,11 @@ mpegts_base_reset (MpegTSBase * base)
   g_hash_table_foreach_remove (base->programs, (GHRFunc) remove_each_program,
       base);
 
+  base->streams_aware = GST_OBJECT_PARENT (base)
+      && GST_OBJECT_FLAG_IS_SET (GST_OBJECT_PARENT (base),
+      GST_BIN_FLAG_STREAMS_AWARE);
+  GST_DEBUG_OBJECT (base, "Streams aware : %d", base->streams_aware);
+
   if (klass->reset)
     klass->reset (base);
 }
@@ -590,6 +595,90 @@ mpegts_base_program_remove_stream (MpegTSBase * base,
   program->streams[pid] = NULL;
 }
 
+/* Check if pmtstream is already present in the program */
+static inline gboolean
+_stream_in_pmt (const GstMpegtsPMT * pmt, MpegTSBaseStream * stream)
+{
+  guint i, nbstreams = pmt->streams->len;
+
+  for (i = 0; i < nbstreams; i++) {
+    GstMpegtsPMTStream *pmt_stream = g_ptr_array_index (pmt->streams, i);
+
+    if (pmt_stream->pid == stream->pid &&
+        pmt_stream->stream_type == stream->stream_type)
+      return TRUE;
+  }
+
+  return FALSE;
+}
+
+static inline gboolean
+_pmt_stream_in_program (MpegTSBaseProgram * program,
+    GstMpegtsPMTStream * stream)
+{
+  MpegTSBaseStream *old_stream = program->streams[stream->pid];
+  if (!old_stream)
+    return FALSE;
+  return old_stream->stream_type == stream->stream_type;
+}
+
+static gboolean
+mpegts_base_update_program (MpegTSBase * base, MpegTSBaseProgram * program,
+    GstMpegtsSection * section, const GstMpegtsPMT * pmt)
+{
+  MpegTSBaseClass *klass = GST_MPEGTS_BASE_GET_CLASS (base);
+  const gchar *stream_id =
+      gst_stream_collection_get_upstream_id (program->collection);
+  GstStreamCollection *collection;
+  GList *tmp, *toremove;
+  guint i, nbstreams;
+
+  /* Create new collection */
+  collection = gst_stream_collection_new (stream_id);
+  gst_object_unref (program->collection);
+  program->collection = collection;
+
+  /* Replace section and pmt with the new one */
+  gst_mpegts_section_unref (program->section);
+  program->section = gst_mpegts_section_ref (section);
+  program->pmt = pmt;
+
+  /* Copy over gststream that still exist into the collection */
+  for (tmp = program->stream_list; tmp; tmp = tmp->next) {
+    MpegTSBaseStream *stream = (MpegTSBaseStream *) tmp->data;
+    if (_stream_in_pmt (pmt, stream)) {
+      gst_stream_collection_add_stream (program->collection,
+          gst_object_ref (stream->stream_object));
+    }
+  }
+
+  /* Add new streams (will also create and add gststream to the collection) */
+  nbstreams = pmt->streams->len;
+  for (i = 0; i < nbstreams; i++) {
+    GstMpegtsPMTStream *stream = g_ptr_array_index (pmt->streams, i);
+    if (!_pmt_stream_in_program (program, stream))
+      mpegts_base_program_add_stream (base, program, stream->pid,
+          stream->stream_type, stream);
+  }
+
+  /* Call subclass update */
+  if (klass->update_program)
+    klass->update_program (base, program);
+
+  /* Remove streams no longer present */
+  toremove = NULL;
+  for (tmp = program->stream_list; tmp; tmp = tmp->next) {
+    MpegTSBaseStream *stream = (MpegTSBaseStream *) tmp->data;
+    if (!_stream_in_pmt (pmt, stream))
+      toremove = g_list_prepend (toremove, stream);
+  }
+  for (tmp = toremove; tmp; tmp = tmp->next) {
+    MpegTSBaseStream *stream = (MpegTSBaseStream *) tmp->data;
+    mpegts_base_program_remove_stream (base, program, stream->pid);
+  }
+  return TRUE;
+}
+
 
 static gboolean
 _stream_is_private_section (GstMpegtsPMTStream * stream)
@@ -675,6 +764,68 @@ mpegts_base_is_same_program (MpegTSBase * base, MpegTSBaseProgram * oldprogram,
   return TRUE;
 }
 
+/* Return TRUE if program is an update
+ *
+ * A program is equal if:
+ * * The program number is the same (will be if it enters this function)
+ * * AND The PMT PID is equal to the old one
+ * * AND It contains at least one stream from the previous program
+ *
+ * Changes that are acceptable are therefore:
+ * * New streams appearing
+ * * Old streams going away
+ * * PCR PID changing
+ *
+ * Unclear changes:
+ * * PMT PID being changed ?
+ * * Properties of elementary stream being changed ? (new tags ? metadata ?)
+ */
+static gboolean
+mpegts_base_is_program_update (MpegTSBase * base,
+    MpegTSBaseProgram * oldprogram, guint16 new_pmt_pid,
+    const GstMpegtsPMT * new_pmt)
+{
+  guint i, nbstreams;
+  MpegTSBaseStream *oldstream;
+
+  if (oldprogram->pmt_pid != new_pmt_pid) {
+    /* FIXME/CHECK: Can a program be updated by just changing its PID
+     * in the PAT ? */
+    GST_DEBUG ("Different pmt_pid (new:0x%04x, old:0x%04x)", new_pmt_pid,
+        oldprogram->pmt_pid);
+    return FALSE;
+  }
+
+  /* Check if at least one stream from the previous program is still present
+   * in the new program */
+
+  /* Check the streams */
+  nbstreams = new_pmt->streams->len;
+  for (i = 0; i < nbstreams; ++i) {
+    GstMpegtsPMTStream *stream = g_ptr_array_index (new_pmt->streams, i);
+
+    oldstream = oldprogram->streams[stream->pid];
+    if (!oldstream) {
+      GST_DEBUG ("New stream 0x%04x not present in old program", stream->pid);
+    } else if (oldstream->stream_type != stream->stream_type) {
+      GST_DEBUG
+          ("New stream 0x%04x has a different stream type (new:%d, old:%d)",
+          stream->pid, stream->stream_type, oldstream->stream_type);
+    } else if (!_stream_is_private_section (stream)) {
+      /* FIXME : We should actually be checking a bit deeper,
+       * especially for private streams (where the differentiation is
+       * done at the registration level) */
+      GST_DEBUG
+          ("Stream 0x%04x is identical (stream_type %d) ! Program is an update",
+          stream->pid, stream->stream_type);
+      return TRUE;
+    }
+  }
+
+  GST_DEBUG ("Program is not an update of the previous one");
+  return FALSE;
+}
+
 static void
 mpegts_base_deactivate_program (MpegTSBase * base, MpegTSBaseProgram * program)
 {
@@ -929,6 +1080,14 @@ mpegts_base_apply_pmt (MpegTSBase * base, GstMpegtsSection * section)
   if (G_UNLIKELY (old_program == NULL))
     goto no_program;
 
+  if (base->streams_aware
+      && mpegts_base_is_program_update (base, old_program, section->pid, pmt)) {
+    GST_FIXME ("We are streams_aware and new program is an update");
+    /* The program is an update, and we can add/remove pads dynamically */
+    mpegts_base_update_program (base, old_program, section, pmt);
+    goto beach;
+  }
+
   if (G_UNLIKELY (mpegts_base_is_same_program (base, old_program, section->pid,
               pmt)))
     goto same_program;
@@ -956,14 +1115,18 @@ mpegts_base_apply_pmt (MpegTSBase * base, GstMpegtsSection * section)
     g_hash_table_insert (base->programs,
         GINT_TO_POINTER (program_number), program);
     initial_program = FALSE;
-  } else
+  } else {
+    GST_DEBUG ("Program update, re-using same program");
     program = old_program;
+  }
 
   /* activate program */
   /* Ownership of pmt_info is given to the program */
   mpegts_base_activate_program (base, program, section->pid, section, pmt,
       initial_program);
 
+beach:
+  GST_DEBUG ("Done activating program");
   return TRUE;
 
 no_program:

gst/mpegtsdemux/mpegtsbase.h

@@ -157,6 +157,10 @@ struct _MpegTSBase {
   /* Whether to push data and/or sections to subclasses */
   gboolean push_data;
   gboolean push_section;
+
+  /* Whether the parent bin is streams-aware, meaning we can
+   * add/remove streams at any point in time */
+  gboolean streams_aware;
 };
 
 struct _MpegTSBaseClass {
@@ -173,6 +177,7 @@ struct _MpegTSBaseClass {
   void (*program_started) (MpegTSBase *base, MpegTSBaseProgram *program);
   /* program_stopped gets called when pat no longer has program's pmt */
   void (*program_stopped) (MpegTSBase *base, MpegTSBaseProgram *program);
+  void (*update_program) (MpegTSBase *base, MpegTSBaseProgram *program);
   /* Whether mpegtbase can deactivate/free a program or whether the subclass will do it
    * If the subclass responds TRUE, it should call mpegts_base_deactivate_and_free_program()
    * when it wants to remove it */

gst/mpegtsdemux/tsdemux.c

@@ -281,6 +281,8 @@ enum
 
 /* mpegtsbase methods */
 static void
+gst_ts_demux_update_program (MpegTSBase * base, MpegTSBaseProgram * program);
+static void
 gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program);
 static void
 gst_ts_demux_program_stopped (MpegTSBase * base, MpegTSBaseProgram * program);
@@ -385,6 +387,7 @@ gst_ts_demux_class_init (GstTSDemuxClass * klass)
   ts_class->push_event = GST_DEBUG_FUNCPTR (push_event);
   ts_class->program_started = GST_DEBUG_FUNCPTR (gst_ts_demux_program_started);
   ts_class->program_stopped = GST_DEBUG_FUNCPTR (gst_ts_demux_program_stopped);
+  ts_class->update_program = GST_DEBUG_FUNCPTR (gst_ts_demux_update_program);
   ts_class->can_remove_program = gst_ts_demux_can_remove_program;
   ts_class->stream_added = gst_ts_demux_stream_added;
   ts_class->stream_removed = gst_ts_demux_stream_removed;
@@ -1826,6 +1829,34 @@ gst_ts_demux_can_remove_program (MpegTSBase * base, MpegTSBaseProgram * program)
   return TRUE;
 }
 
+static void
+gst_ts_demux_update_program (MpegTSBase * base, MpegTSBaseProgram * program)
+{
+  GstTSDemux *demux = GST_TS_DEMUX (base);
+  GList *tmp;
+
+  GST_DEBUG ("Updating program %d", program->program_number);
+  /* Emit collection message */
+  gst_element_post_message ((GstElement *) base,
+      gst_message_new_stream_collection ((GstObject *) base,
+          program->collection));
+
+  /* Add all streams, then fire no-more-pads */
+  for (tmp = program->stream_list; tmp; tmp = tmp->next) {
+    TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+    if (!stream->pad) {
+      activate_pad_for_stream (demux, stream);
+      if (stream->sparse) {
+        /* force sending of pending sticky events which have been stored on the
+         * pad already and which otherwise would only be sent on the first buffer
+         * or serialized event (which means very late in case of subtitle streams),
+         * and playsink waits for stream-start or another serialized event */
+        GST_DEBUG_OBJECT (stream->pad, "sparse stream, pushing GAP event");
+        gst_pad_push_event (stream->pad, gst_event_new_gap (0, 0));
+      }
+    }
+  }
+}
 
 static void
 gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)