mpegtsdemux: Initial switch to using GstStream/GstStreamCollection

* MpegTSBaseStream is now MpegTSStream, a subclass of GstStream
* GstStream are created for each elementary stream with the proper
  properties set on them
* GstStreamCollection is created for each program
* tsdemux will post GstStreamCollection on bus before activating
  the program
* tsdemux: Stream objects are set on stream-start events for valid
  streams
* tsdemux: Stream-collection events are sent on pads

4 files changed, 196 insertions, 71 deletions

diff --git a/gst/mpegtsdemux/mpegtsbase.c b/gst/mpegtsdemux/mpegtsbase.c
index c0610f12a..4902acd82 100644
--- a/gst/mpegtsdemux/mpegtsbase.c
+++ b/gst/mpegtsdemux/mpegtsbase.c
@@ -41,6 +41,18 @@
 #include "mpegtsbase.h"
 #include "gstmpegdesc.h"
 
+/* TODO STREAMS
+ *
+ * 1) Make MpegTSStream a subclass of GstStream
+ *  => Refcounting ! Will solve quite a bit of issues with streams
+ *   being referenced in plenty of different locations
+ *
+ *  The problem is that caps information and nature of stream
+ *  is determined at the tsdemux level.
+ *    => Only demuxer needs to provide extra info
+ *
+ */
+
 #define RUNNING_STATUS_RUNNING 4
 
 GST_DEBUG_CATEGORY_STATIC (mpegts_base_debug);
@@ -92,6 +104,35 @@ static gboolean mpegts_base_parse_atsc_mgt (MpegTSBase * base,
 static gboolean remove_each_program (gpointer key, MpegTSBaseProgram * program,
     MpegTSBase * base);
 
+G_DEFINE_TYPE (MpegTSStream, mpegts_stream, GST_TYPE_STREAM);
+
+static void
+mpegts_stream_finalize (GObject * object)
+{
+  MpegTSStream *stream = (MpegTSStream *) object;
+
+  if (stream->demux_info && stream->notify) {
+    stream->notify (stream->demux_info);
+    stream->demux_info = NULL;
+  }
+  if (G_OBJECT_CLASS (mpegts_stream_parent_class)->finalize)
+    G_OBJECT_CLASS (mpegts_stream_parent_class)->finalize (object);
+}
+
+static void
+mpegts_stream_class_init (MpegTSStreamClass * klass)
+{
+  GObjectClass *gobjectclass = (GObjectClass *) klass;
+
+  gobjectclass->finalize = mpegts_stream_finalize;
+}
+
+static void
+mpegts_stream_init (MpegTSStream * stream)
+{
+  stream->demux_info = NULL;
+}
+
 static void
 _extra_init (void)
 {
@@ -235,7 +276,6 @@ mpegts_base_init (MpegTSBase * base)
   base->is_pes = g_new0 (guint8, 1024);
   base->known_psi = g_new0 (guint8, 1024);
   base->program_size = sizeof (MpegTSBaseProgram);
-  base->stream_size = sizeof (MpegTSBaseStream);
 
   base->push_data = TRUE;
   base->push_section = TRUE;
@@ -279,7 +319,7 @@ mpegts_base_finalize (GObject * object)
  * otherwise returns a descriptor that needs to *
  * be freed */
 const GstMpegtsDescriptor *
-mpegts_get_descriptor_from_stream (MpegTSBaseStream * stream, guint8 tag)
+mpegts_get_descriptor_from_stream (MpegTSStream * stream, guint8 tag)
 {
   GstMpegtsPMTStream *pmt = stream->stream;
 
@@ -334,6 +374,7 @@ mpegts_base_new_program (MpegTSBase * base,
     gint program_number, guint16 pmt_pid)
 {
   MpegTSBaseProgram *program;
+  gchar *upstream_id, *stream_id;
 
   GST_DEBUG_OBJECT (base, "program_number : %d, pmt_pid : %d",
       program_number, pmt_pid);
@@ -342,9 +383,15 @@ mpegts_base_new_program (MpegTSBase * base,
   program->program_number = program_number;
   program->pmt_pid = pmt_pid;
   program->pcr_pid = G_MAXUINT16;
-  program->streams = g_new0 (MpegTSBaseStream *, 0x2000);
+  program->streams = g_new0 (MpegTSStream *, 0x2000);
   program->patcount = 0;
 
+  upstream_id = gst_pad_get_stream_id (base->sinkpad);
+  stream_id = g_strdup_printf ("%s:%d", upstream_id, program_number);
+  program->collection = gst_stream_collection_new (stream_id);
+  g_free (stream_id);
+  g_free (upstream_id);
+
   return program;
 }
 
@@ -409,7 +456,7 @@ mpegts_base_free_program (MpegTSBaseProgram * program)
   }
 
   for (tmp = program->stream_list; tmp; tmp = tmp->next)
-    g_free (tmp->data);
+    gst_object_unref (tmp->data);
   if (program->stream_list)
     g_list_free (program->stream_list);
 
@@ -417,6 +464,8 @@ mpegts_base_free_program (MpegTSBaseProgram * program)
 
   if (program->tags)
     gst_tag_list_unref (program->tags);
+  if (program->collection)
+    gst_object_unref (program->collection);
 
   g_free (program);
 }
@@ -456,25 +505,34 @@ get_registration_from_descriptors (GPtrArray * descriptors)
   return 0;
 }
 
-static MpegTSBaseStream *
+static MpegTSStream *
 mpegts_base_program_add_stream (MpegTSBase * base,
     MpegTSBaseProgram * program, guint16 pid, guint8 stream_type,
     GstMpegtsPMTStream * stream)
 {
   MpegTSBaseClass *klass = GST_MPEGTS_BASE_GET_CLASS (base);
-  MpegTSBaseStream *bstream;
+  MpegTSStream *bstream;
+  gchar *stream_id;
 
   GST_DEBUG ("pid:0x%04x, stream_type:0x%03x", pid, stream_type);
 
+  /* FIXME : PID information/nature might change through time.
+   * We therefore *do* want to be able to replace an existing stream
+   * with updated information */
   if (G_UNLIKELY (program->streams[pid])) {
     if (stream_type != 0xff)
       GST_WARNING ("Stream already present !");
     return NULL;
   }
 
-  bstream = g_malloc0 (base->stream_size);
+  stream_id =
+      g_strdup_printf ("%s/%08x",
+      gst_stream_collection_get_upstream_id (program->collection), pid);
+  bstream = g_object_new (MPEG_TYPE_TS_STREAM, "stream-id", stream_id, NULL);
+  g_free (stream_id);
   bstream->pid = pid;
   bstream->stream_type = stream_type;
+  /* FIXME : What's the life cycle of GstMpegtsPMTStream ?? */
   bstream->stream = stream;
   if (stream) {
     bstream->registration_id =
@@ -483,12 +541,14 @@ mpegts_base_program_add_stream (MpegTSBase * base,
         bstream->pid, SAFE_FOURCC_ARGS (bstream->registration_id));
   }
 
-
   program->streams[pid] = bstream;
   program->stream_list = g_list_append (program->stream_list, bstream);
 
   if (klass->stream_added)
-    klass->stream_added (base, bstream, program);
+    if (klass->stream_added (base, bstream, program))
+      gst_stream_collection_add_stream (program->collection,
+          (GstStream *) gst_object_ref (bstream));
+
 
   return bstream;
 }
@@ -498,7 +558,7 @@ mpegts_base_program_remove_stream (MpegTSBase * base,
     MpegTSBaseProgram * program, guint16 pid)
 {
   MpegTSBaseClass *klass;
-  MpegTSBaseStream *stream = program->streams[pid];
+  MpegTSStream *stream = program->streams[pid];
 
   GST_DEBUG ("pid:0x%04x", pid);
 
@@ -515,7 +575,7 @@ mpegts_base_program_remove_stream (MpegTSBase * base,
     klass->stream_removed (base, stream);
 
   program->stream_list = g_list_remove_all (program->stream_list, stream);
-  g_free (stream);
+  gst_object_unref (stream);
   program->streams[pid] = NULL;
 }
 
@@ -525,7 +585,7 @@ mpegts_base_is_same_program (MpegTSBase * base, MpegTSBaseProgram * oldprogram,
     guint16 new_pmt_pid, const GstMpegtsPMT * new_pmt)
 {
   guint i, nbstreams;
-  MpegTSBaseStream *oldstream;
+  MpegTSStream *oldstream;
   gboolean sawpcrpid = FALSE;
 
   if (oldprogram->pmt_pid != new_pmt_pid) {
diff --git a/gst/mpegtsdemux/mpegtsbase.h b/gst/mpegtsdemux/mpegtsbase.h
index fc0c6512c..4e102eaa0 100644
--- a/gst/mpegtsdemux/mpegtsbase.h
+++ b/gst/mpegtsdemux/mpegtsbase.h
@@ -52,11 +52,18 @@ G_BEGIN_DECLS
 
 typedef struct _MpegTSBase MpegTSBase;
 typedef struct _MpegTSBaseClass MpegTSBaseClass;
-typedef struct _MpegTSBaseStream MpegTSBaseStream;
+
+#define MPEG_TYPE_TS_STREAM (mpegts_stream_get_type())
+
+typedef struct _MpegTSStream MpegTSStream;
+typedef struct _MpegTSStreamClass MpegTSStreamClass;
+
 typedef struct _MpegTSBaseProgram MpegTSBaseProgram;
 
-struct _MpegTSBaseStream
+struct _MpegTSStream
 {
+  GstStream           gst_stream;
+  
   guint16             pid;
   guint8              stream_type;
 
@@ -64,6 +71,16 @@ struct _MpegTSBaseStream
   guint32             registration_id;
 
   GstMpegtsPMTStream *stream;
+
+  /* Field for demuxer */
+  /* FIXME : Explicit object ? Need a GDestroyNotify */
+  gpointer           *demux_info;
+  GDestroyNotify      notify;
+};
+
+struct _MpegTSStreamClass
+{
+  GstStreamClass      klass;
 };
 
 struct _MpegTSBaseProgram
@@ -78,10 +95,12 @@ struct _MpegTSBaseProgram
   GstMpegtsSection   *section;
   const GstMpegtsPMT *pmt;
 
-  MpegTSBaseStream  **streams;
+  MpegTSStream  **streams;
   GList              *stream_list;
   gint                patcount;
 
+  GstStreamCollection *collection;
+
   /* Pending Tags for the program */
   GstTagList *tags;
   guint event_id;
@@ -134,10 +153,6 @@ struct _MpegTSBase {
    * by subclasses if they have their own MpegTSBaseProgram subclasses. */
   gsize program_size;
 
-  /* size of the MpegTSBaseStream structure, can be overridden
-   * by subclasses if they have their own MpegTSBaseStream subclasses */
-  gsize stream_size;
-
   /* Whether we saw a PAT yet */
   gboolean seen_pat;
 
@@ -180,9 +195,9 @@ struct _MpegTSBaseClass {
   gboolean (*can_remove_program) (MpegTSBase *base, MpegTSBaseProgram *program);
 
   /* stream_added is called whenever a new stream has been identified */
-  void (*stream_added) (MpegTSBase *base, MpegTSBaseStream *stream, MpegTSBaseProgram *program);
+  gboolean (*stream_added) (MpegTSBase *base, MpegTSStream *stream, MpegTSBaseProgram *program);
   /* stream_removed is called whenever a stream is no longer referenced */
-  void (*stream_removed) (MpegTSBase *base, MpegTSBaseStream *stream);
+  void (*stream_removed) (MpegTSBase *base, MpegTSStream *stream);
 
   /* find_timestamps is called to find PCR */
   GstFlowReturn (*find_timestamps) (MpegTSBase * base, guint64 initoff, guint64 *offset);
@@ -218,7 +233,7 @@ G_GNUC_INTERNAL GType mpegts_base_get_type(void);
 G_GNUC_INTERNAL MpegTSBaseProgram *mpegts_base_get_program (MpegTSBase * base, gint program_number);
 G_GNUC_INTERNAL MpegTSBaseProgram *mpegts_base_add_program (MpegTSBase * base, gint program_number, guint16 pmt_pid);
 
-G_GNUC_INTERNAL const GstMpegtsDescriptor *mpegts_get_descriptor_from_stream (MpegTSBaseStream * stream, guint8 tag);
+G_GNUC_INTERNAL const GstMpegtsDescriptor *mpegts_get_descriptor_from_stream (MpegTSStream * stream, guint8 tag);
 G_GNUC_INTERNAL const GstMpegtsDescriptor *mpegts_get_descriptor_from_program (MpegTSBaseProgram * program, guint8 tag);
 
 G_GNUC_INTERNAL gboolean
@@ -230,4 +245,8 @@ G_GNUC_INTERNAL void mpegts_base_deactivate_and_free_program (MpegTSBase *base,
 
 G_END_DECLS
 
+G_GNUC_INTERNAL GType mpegts_stream_get_type(void);
+
+
+
 #endif /* GST_MPEG_TS_BASE_H */
diff --git a/gst/mpegtsdemux/mpegtsparse.c b/gst/mpegtsdemux/mpegtsparse.c
index 6e4faf508..d7e5896e6 100644
--- a/gst/mpegtsdemux/mpegtsparse.c
+++ b/gst/mpegtsdemux/mpegtsparse.c
@@ -552,7 +552,7 @@ mpegts_parse_tspad_push (MpegTSParse2 * parse, MpegTSParsePad * tspad,
     MpegTSPacketizerPacket * packet)
 {
   GstFlowReturn ret = GST_FLOW_OK;
-  MpegTSBaseStream **pad_pids = NULL;
+  MpegTSStream **pad_pids = NULL;
 
   if (tspad->program_number != -1) {
     if (tspad->program) {
diff --git a/gst/mpegtsdemux/tsdemux.c b/gst/mpegtsdemux/tsdemux.c
index b54a7d371..1a584c9bf 100644
--- a/gst/mpegtsdemux/tsdemux.c
+++ b/gst/mpegtsdemux/tsdemux.c
@@ -135,7 +135,7 @@ struct _TSDemuxH264ParsingInfos
 
 struct _TSDemuxStream
 {
-  MpegTSBaseStream stream;
+  MpegTSStream *base;
 
   GstPad *pad;
 
@@ -291,11 +291,11 @@ gst_ts_demux_push (MpegTSBase * base, MpegTSPacketizerPacket * packet,
     GstMpegtsSection * section);
 static void gst_ts_demux_flush (MpegTSBase * base, gboolean hard);
 static GstFlowReturn gst_ts_demux_drain (MpegTSBase * base);
-static void
-gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * stream,
+static gboolean
+gst_ts_demux_stream_added (MpegTSBase * base, MpegTSStream * stream,
     MpegTSBaseProgram * program);
 static void
-gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSBaseStream * stream);
+gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSStream * stream);
 static GstFlowReturn gst_ts_demux_do_seek (MpegTSBase * base, GstEvent * event);
 static void gst_ts_demux_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec);
@@ -418,7 +418,6 @@ gst_ts_demux_init (GstTSDemux * demux)
 {
   MpegTSBase *base = (MpegTSBase *) demux;
 
-  base->stream_size = sizeof (TSDemuxStream);
   base->parse_private_sections = TRUE;
   /* We are not interested in sections (all handled by mpegtsbase) */
   base->push_section = FALSE;
@@ -785,7 +784,7 @@ gst_ts_demux_adjust_seek_offset_for_keyframe (TSDemuxStream * stream,
   if (!stream->scan_function)
     return TRUE;
 
-  scan_pid = ((MpegTSBaseStream *) stream)->pid;
+  scan_pid = stream->base->pid;
 
   if (scan_pid != -1) {
     return stream->scan_function (stream, data, size, size);
@@ -839,7 +838,8 @@ gst_ts_demux_do_seek (MpegTSBase * base, GstEvent * event)
     }
   } else {
     for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-      TSDemuxStream *stream = tmp->data;
+      MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+      TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
 
       stream->need_newsegment = TRUE;
     }
@@ -870,7 +870,8 @@ gst_ts_demux_do_seek (MpegTSBase * base, GstEvent * event)
   }
 
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *stream = tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
 
     if (flags & GST_SEEK_FLAG_ACCURATE)
       stream->needs_keyframe = TRUE;
@@ -959,7 +960,8 @@ push_event (MpegTSBase * base, GstEvent * event)
   }
 
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
     if (stream->pad) {
       /* If we are pushing out EOS, flush out pending data first */
       if (GST_EVENT_TYPE (event) == GST_EVENT_EOS &&
@@ -998,7 +1000,7 @@ add_iso639_language_to_tags (TSDemuxStream * stream, gchar * lang_code)
 static void
 gst_ts_demux_create_tags (TSDemuxStream * stream)
 {
-  MpegTSBaseStream *bstream = (MpegTSBaseStream *) stream;
+  MpegTSStream *bstream = stream->base;
   const GstMpegtsDescriptor *desc = NULL;
   int i, nb;
 
@@ -1042,11 +1044,11 @@ gst_ts_demux_create_tags (TSDemuxStream * stream)
 }
 
 static GstPad *
-create_pad_for_stream (MpegTSBase * base, MpegTSBaseStream * bstream,
+create_pad_for_stream (MpegTSBase * base, MpegTSStream * bstream,
     MpegTSBaseProgram * program)
 {
   GstTSDemux *demux = GST_TS_DEMUX (base);
-  TSDemuxStream *stream = (TSDemuxStream *) bstream;
+  TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
   gchar *name = NULL;
   GstCaps *caps = NULL;
   GstPadTemplate *template = NULL;
@@ -1517,7 +1519,7 @@ create_pad_for_stream (MpegTSBase * base, MpegTSBaseStream * bstream,
       caps = gst_caps_new_empty_simple ("video/x-cavs");
       break;
     default:
-      GST_WARNING ("Non-media stream (stream_type:0x%x). Not creating pad",
+      GST_DEBUG ("Non-media stream (stream_type:0x%x). Not creating pad",
           bstream->stream_type);
       break;
   }
@@ -1527,15 +1529,18 @@ done:
     if (is_audio) {
       template = gst_static_pad_template_get (&audio_template);
       name = g_strdup_printf ("audio_%04x", bstream->pid);
+      gst_stream_set_stream_type ((GstStream *) bstream, GST_STREAM_TYPE_AUDIO);
     } else if (is_video) {
       template = gst_static_pad_template_get (&video_template);
       name = g_strdup_printf ("video_%04x", bstream->pid);
+      gst_stream_set_stream_type ((GstStream *) bstream, GST_STREAM_TYPE_VIDEO);
     } else if (is_private) {
       template = gst_static_pad_template_get (&private_template);
       name = g_strdup_printf ("private_%04x", bstream->pid);
     } else if (is_subpicture) {
       template = gst_static_pad_template_get (&subpicture_template);
       name = g_strdup_printf ("subpicture_%04x", bstream->pid);
+      gst_stream_set_stream_type ((GstStream *) bstream, GST_STREAM_TYPE_TEXT);
     } else
       g_assert_not_reached ();
 
@@ -1543,16 +1548,14 @@ done:
 
   if (template && name && caps) {
     GstEvent *event;
-    gchar *stream_id;
+    const gchar *stream_id;
 
     GST_LOG ("stream:%p creating pad with name %s and caps %" GST_PTR_FORMAT,
         stream, name, caps);
     pad = gst_pad_new_from_template (template, name);
     gst_pad_set_active (pad, TRUE);
     gst_pad_use_fixed_caps (pad);
-    stream_id =
-        gst_pad_create_stream_id_printf (pad, GST_ELEMENT_CAST (base), "%08x",
-        bstream->pid);
+    stream_id = gst_stream_get_stream_id (GST_STREAM (bstream));
 
     event = gst_pad_get_sticky_event (base->sinkpad, GST_EVENT_STREAM_START, 0);
     if (event) {
@@ -1566,19 +1569,24 @@ done:
       demux->group_id = gst_util_group_id_next ();
     }
     event = gst_event_new_stream_start (stream_id);
+    gst_event_set_stream (event, GST_STREAM (bstream));
     if (demux->have_group_id)
       gst_event_set_group_id (event, demux->group_id);
-    if (sparse)
+    if (sparse) {
       gst_event_set_stream_flags (event, GST_STREAM_FLAG_SPARSE);
+      gst_stream_set_stream_flags ((GstStream *) bstream,
+          GST_STREAM_FLAG_SPARSE);
+    }
     stream->sparse = sparse;
 
     gst_pad_push_event (pad, event);
-    g_free (stream_id);
     gst_pad_set_caps (pad, caps);
+    gst_stream_set_caps ((GstStream *) bstream, caps);
     if (!stream->taglist)
       stream->taglist = gst_tag_list_new_empty ();
     gst_pb_utils_add_codec_description_to_tag_list (stream->taglist, NULL,
         caps);
+    gst_stream_set_tags ((GstStream *) bstream, stream->taglist);
     gst_pad_set_query_function (pad, gst_ts_demux_srcpad_query);
     gst_pad_set_event_function (pad, gst_ts_demux_srcpad_event);
   }
@@ -1593,13 +1601,25 @@ done:
 }
 
 static void
-gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
+ts_demux_stream_notify_cb (TSDemuxStream * stream)
+{
+  g_free (stream);
+}
+
+static gboolean
+gst_ts_demux_stream_added (MpegTSBase * base, MpegTSStream * bstream,
     MpegTSBaseProgram * program)
 {
   GstTSDemux *demux = (GstTSDemux *) base;
-  TSDemuxStream *stream = (TSDemuxStream *) bstream;
+  TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
+
+  if (stream == NULL) {
+    stream = g_new0 (TSDemuxStream, 1);
+
+    stream->base = bstream;
+    bstream->demux_info = (gpointer) stream;
+    bstream->notify = (GDestroyNotify) ts_demux_stream_notify_cb;
 
-  if (!stream->pad) {
     /* Create the pad */
     if (bstream->stream_type != 0xff) {
       stream->pad = create_pad_for_stream (base, bstream, program);
@@ -1632,6 +1652,8 @@ gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
     stream->gap_ref_pts = GST_CLOCK_TIME_NONE;
     stream->continuity_counter = CONTINUITY_UNSET;
   }
+
+  return (stream->pad != NULL);
 }
 
 static void
@@ -1648,9 +1670,9 @@ tsdemux_h264_parsing_info_clear (TSDemuxH264ParsingInfos * h264infos)
 }
 
 static void
-gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSBaseStream * bstream)
+gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSStream * bstream)
 {
-  TSDemuxStream *stream = (TSDemuxStream *) bstream;
+  TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
 
   if (stream->pad) {
     gst_flow_combiner_remove_pad (GST_TS_DEMUX_CAST (base)->flowcombiner,
@@ -1695,11 +1717,13 @@ activate_pad_for_stream (GstTSDemux * tsdemux, TSDemuxStream * stream)
     gst_element_add_pad ((GstElement *) tsdemux, stream->pad);
     stream->active = TRUE;
     GST_DEBUG_OBJECT (stream->pad, "done adding pad");
-  } else if (((MpegTSBaseStream *) stream)->stream_type != 0xff) {
+    /* Send collection event */
+    gst_pad_push_event (stream->pad,
+        gst_event_new_stream_collection (tsdemux->program->collection));
+  } else if (stream->base->stream_type != 0xff) {
     GST_DEBUG_OBJECT (tsdemux,
         "stream %p (pid 0x%04x, type:0x%02x) has no pad", stream,
-        ((MpegTSBaseStream *) stream)->pid,
-        ((MpegTSBaseStream *) stream)->stream_type);
+        stream->base->pid, stream->base->stream_type);
   }
 }
 
@@ -1752,8 +1776,11 @@ gst_ts_demux_flush_streams (GstTSDemux * demux, gboolean hard)
   if (!demux->program)
     return;
 
-  for (walk = demux->program->stream_list; walk; walk = g_list_next (walk))
-    gst_ts_demux_stream_flush (walk->data, demux, hard);
+  for (walk = demux->program->stream_list; walk; walk = g_list_next (walk)) {
+    MpegTSStream *bstream = (MpegTSStream *) walk->data;
+    gst_ts_demux_stream_flush ((TSDemuxStream *) bstream->demux_info, demux,
+        hard);
+  }
 }
 
 static gboolean
@@ -1791,6 +1818,11 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
     demux->program_number = program->program_number;
     demux->program = program;
 
+    /* Emit collection message */
+    gst_element_post_message ((GstElement *) base,
+        gst_message_new_stream_collection ((GstObject *) base,
+            program->collection));
+
     /* If this is not the initial program, we need to calculate
      * a new segment */
     if (demux->segment_event) {
@@ -1803,7 +1835,8 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
       GList *tmp;
       GST_DEBUG_OBJECT (demux, "Draining previous program");
       for (tmp = demux->previous_program->stream_list; tmp; tmp = tmp->next) {
-        TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+        MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+        TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
         if (stream->pad)
           gst_ts_demux_push_pending_data (demux, stream);
       }
@@ -1811,7 +1844,8 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
 
     /* Add all streams, then fire no-more-pads */
     for (tmp = program->stream_list; tmp; tmp = tmp->next) {
-      TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+      MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+      TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
       activate_pad_for_stream (demux, stream);
     }
 
@@ -1827,7 +1861,8 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
      * a GAP event *is* considering data, and we want to ensure the (potential)
      * old pads are all removed before we push any data on the new ones */
     for (tmp = program->stream_list; tmp; tmp = tmp->next) {
-      TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+      MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+      TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
       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
@@ -1857,7 +1892,7 @@ static inline void
 gst_ts_demux_record_pts (GstTSDemux * demux, TSDemuxStream * stream,
     guint64 pts, guint64 offset)
 {
-  MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
+  MpegTSStream *bs = stream->base;
 
   stream->raw_pts = pts;
   if (pts == -1) {
@@ -1891,7 +1926,7 @@ static inline void
 gst_ts_demux_record_dts (GstTSDemux * demux, TSDemuxStream * stream,
     guint64 dts, guint64 offset)
 {
-  MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
+  MpegTSStream *bs = stream->base;
 
   stream->raw_dts = dts;
   if (dts == -1) {
@@ -1932,7 +1967,8 @@ check_pending_buffers (GstTSDemux * demux)
 
   /* 1. Go over all streams */
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *tmpstream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *tmpstream = (TSDemuxStream *) bstream->demux_info;
     /* 1.1 check if at least one stream got a valid DTS */
     if ((tmpstream->raw_dts != -1 && tmpstream->dts != GST_CLOCK_TIME_NONE) ||
         (tmpstream->raw_pts != -1 && tmpstream->pts != GST_CLOCK_TIME_NONE)) {
@@ -1947,7 +1983,8 @@ check_pending_buffers (GstTSDemux * demux)
 
   /* 3. Go over all streams that have current/pending data */
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *tmpstream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *tmpstream = (TSDemuxStream *) bstream->demux_info;
     PendingBuffer *pend;
     guint64 firstval, lastval, ts;
 
@@ -1997,7 +2034,8 @@ check_pending_buffers (GstTSDemux * demux)
 
   /* 4. Go over all streams */
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
 
     stream->pending_ts = FALSE;
     /* 4.1 Set pending_ts for FALSE */
@@ -2060,7 +2098,7 @@ gst_ts_demux_parse_pes_header (GstTSDemux * demux, TSDemuxStream * stream,
     goto discont;
   if (G_UNLIKELY (parseres == PES_PARSING_BAD)) {
     GST_WARNING ("Error parsing PES header. pid: 0x%x stream_type: 0x%x",
-        stream->stream.pid, stream->stream.stream_type);
+        stream->base->pid, stream->base->stream_type);
     goto discont;
   }
 
@@ -2136,7 +2174,7 @@ gst_ts_demux_queue_data (GstTSDemux * demux, TSDemuxStream * stream,
   guint size;
   guint8 cc = FLAGS_CONTINUITY_COUNTER (packet->scram_afc_cc);
 
-  GST_LOG ("pid: 0x%04x state:%d", stream->stream.pid, stream->state);
+  GST_LOG ("pid: 0x%04x state:%d", stream->base->pid, stream->state);
 
   size = packet->data_end - packet->payload;
   data = packet->payload;
@@ -2220,7 +2258,8 @@ calculate_and_push_newsegment (GstTSDemux * demux, TSDemuxStream * stream)
 
   /* Calculate the 'new_start' value, used for newsegment */
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *pstream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *pstream = (TSDemuxStream *) bstream->demux_info;
 
     if (GST_CLOCK_TIME_IS_VALID (pstream->first_pts)) {
       if (!GST_CLOCK_TIME_IS_VALID (lowest_pts)
@@ -2269,7 +2308,8 @@ calculate_and_push_newsegment (GstTSDemux * demux, TSDemuxStream * stream)
 
 push_new_segment:
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    stream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    stream = (TSDemuxStream *) bstream->demux_info;
     if (stream->pad == NULL)
       continue;
 
@@ -2326,11 +2366,12 @@ gst_ts_demux_check_and_sync_streams (GstTSDemux * demux, GstClockTime time)
    * and but don't push a gap event
    */
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *ps = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *ps = (TSDemuxStream *) bstream->demux_info;
     GST_DEBUG_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),
+        bstream->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);
@@ -2454,7 +2495,9 @@ static GstFlowReturn
 gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
 {
   GstFlowReturn res = GST_FLOW_OK;
-  MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
+#ifndef GST_DISABLE_GST_DEBUG
+  MpegTSStream *bs = stream->base;
+#endif
   GstBuffer *buffer = NULL;
   GstBufferList *buffer_list = NULL;
 
@@ -2755,7 +2798,8 @@ gst_ts_demux_drain (MpegTSBase * base)
     return res;
 
   for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
-    TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
+    MpegTSStream *bstream = (MpegTSStream *) tmp->data;
+    TSDemuxStream *stream = (TSDemuxStream *) bstream->demux_info;
     if (stream->pad) {
       res = gst_ts_demux_push_pending_data (demux, stream);
       if (G_UNLIKELY (res != GST_FLOW_OK))
@@ -2771,14 +2815,16 @@ gst_ts_demux_push (MpegTSBase * base, MpegTSPacketizerPacket * packet,
     GstMpegtsSection * section)
 {
   GstTSDemux *demux = GST_TS_DEMUX_CAST (base);
-  TSDemuxStream *stream = NULL;
+  MpegTSStream *stream = NULL;
   GstFlowReturn res = GST_FLOW_OK;
 
   if (G_LIKELY (demux->program)) {
-    stream = (TSDemuxStream *) demux->program->streams[packet->pid];
+    stream = demux->program->streams[packet->pid];
 
     if (stream) {
-      res = gst_ts_demux_handle_packet (demux, stream, packet, section);
+      res =
+          gst_ts_demux_handle_packet (demux,
+          (TSDemuxStream *) stream->demux_info, packet, section);
     }
   }
   return res;