mpegtsdemux: Switch to using GstStream/GstStreamCollection

For each MpegTSBaseStream, we have a GstStream object which subclasses can extend with information. For each program a GstStreamCollection is created with all GstStream from each stream.
2025-02-17 03:35:21 +00:00 · 2015-07-29 15:52:05 +02:00 · 2015-07-29 15:52:05 +02:00 · bcc06d679a
commit bcc06d679a
parent dc703b50b3
3 changed files with 67 additions and 14 deletions
--- a/gst/mpegtsdemux/mpegtsbase.c
+++ b/gst/mpegtsdemux/mpegtsbase.c
@ -333,6 +333,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);
@ -344,6 +345,12 @@ mpegts_base_new_program (MpegTSBase * base,
  program->streams = g_new0 (MpegTSBaseStream *, 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;
 }

@ -397,6 +404,16 @@ mpegts_base_steal_program (MpegTSBase * base, gint program_number)
  return program;
 }

+static void
+mpegts_base_free_stream (MpegTSBaseStream * stream)
+{
+  if (stream->stream_object)
+    gst_object_unref (stream->stream_object);
+  if (stream->stream_id)
+    g_free (stream->stream_id);
+  g_free (stream);
+}
+
 static void
 mpegts_base_free_program (MpegTSBaseProgram * program)
 {
@ -407,8 +424,10 @@ mpegts_base_free_program (MpegTSBaseProgram * program)
    program->pmt = NULL;
  }

+  /* FIXME FIXME FIXME FREE STREAM OBJECT ! */
  for (tmp = program->stream_list; tmp; tmp = tmp->next)
-    g_free (tmp->data);
+    mpegts_base_free_stream ((MpegTSBaseStream *) tmp->data);
+
  if (program->stream_list)
    g_list_free (program->stream_list);

@ -416,6 +435,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);
 }
@ -465,6 +486,9 @@ mpegts_base_program_add_stream (MpegTSBase * base,

  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 !");
@ -472,9 +496,15 @@ mpegts_base_program_add_stream (MpegTSBase * base,
  }

  bstream = g_malloc0 (base->stream_size);
+  bstream->stream_id =
+      g_strdup_printf ("%s/%08x",
+      gst_stream_collection_get_upstream_id (program->collection), pid);
  bstream->pid = pid;
  bstream->stream_type = stream_type;
  bstream->stream = stream;
+  /* We don't yet know the stream type, subclasses will fill that */
+  bstream->stream_object = gst_stream_new (bstream->stream_id, NULL,
+      GST_STREAM_TYPE_UNKNOWN, GST_STREAM_FLAG_NONE);
  if (stream) {
    bstream->registration_id =
        get_registration_from_descriptors (stream->descriptors);
@ -482,12 +512,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->stream_object));
+

  return bstream;
 }
@ -514,7 +546,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);
+  mpegts_base_free_stream (stream);
  program->streams[pid] = NULL;
 }

--- a/gst/mpegtsdemux/mpegtsbase.h
+++ b/gst/mpegtsdemux/mpegtsbase.h
@ -64,6 +64,8 @@ struct _MpegTSBaseStream
  guint32             registration_id;

  GstMpegtsPMTStream *stream;
+  GstStream          *stream_object;
+  gchar              *stream_id;
 };

 struct _MpegTSBaseProgram
@ -82,6 +84,8 @@ struct _MpegTSBaseProgram
  GList              *stream_list;
  gint                patcount;

+  GstStreamCollection *collection;
+
  /* Pending Tags for the program */
  GstTagList *tags;
  guint event_id;
@ -175,7 +179,7 @@ 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, MpegTSBaseStream *stream, MpegTSBaseProgram *program);
  /* stream_removed is called whenever a stream is no longer referenced */
  void (*stream_removed) (MpegTSBase *base, MpegTSBaseStream *stream);

--- a/gst/mpegtsdemux/tsdemux.c
+++ b/gst/mpegtsdemux/tsdemux.c
@ -293,7 +293,7 @@ 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
+static gboolean
 gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * stream,
    MpegTSBaseProgram * program);
 static void
@ -1549,7 +1549,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;
  }
@ -1561,11 +1561,15 @@ done:
      name =
          g_strdup_printf ("audio_%01x_%04x", demux->program_generation,
          bstream->pid);
+      gst_stream_set_stream_type (bstream->stream_object,
+          GST_STREAM_TYPE_AUDIO);
    } else if (is_video) {
      template = gst_static_pad_template_get (&video_template);
      name =
          g_strdup_printf ("video_%01x_%04x", demux->program_generation,
          bstream->pid);
+      gst_stream_set_stream_type (bstream->stream_object,
+          GST_STREAM_TYPE_VIDEO);
    } else if (is_private) {
      template = gst_static_pad_template_get (&private_template);
      name =
@ -1576,6 +1580,7 @@ done:
      name =
          g_strdup_printf ("subpicture_%01x_%04x", demux->program_generation,
          bstream->pid);
+      gst_stream_set_stream_type (bstream->stream_object, GST_STREAM_TYPE_TEXT);
    } else
      g_assert_not_reached ();

@ -1583,16 +1588,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 (bstream->stream_object);

    event = gst_pad_get_sticky_event (base->sinkpad, GST_EVENT_STREAM_START, 0);
    if (event) {
@ -1606,19 +1609,24 @@ done:
      demux->group_id = gst_util_group_id_next ();
    }
    event = gst_event_new_stream_start (stream_id);
+    gst_event_set_stream (event, bstream->stream_object);
    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 (bstream->stream_object,
+          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 (bstream->stream_object, 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 (bstream->stream_object, stream->taglist);
    gst_pad_set_query_function (pad, gst_ts_demux_srcpad_query);
    gst_pad_set_event_function (pad, gst_ts_demux_srcpad_event);
  }
@ -1632,7 +1640,7 @@ done:
  return pad;
 }

-static void
+static gboolean
 gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
    MpegTSBaseProgram * program)
 {
@ -1675,6 +1683,8 @@ gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
    stream->pending_ts = program->pcr_pid < 0x1fff;
    stream->continuity_counter = CONTINUITY_UNSET;
  }
+
+  return (stream->pad != NULL);
 }

 static void
@ -1837,6 +1847,11 @@ gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
    /* Increment the program_generation counter */
    demux->program_generation = (demux->program_generation + 1) & 0xf;

+    /* 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) {
@ -2506,7 +2521,9 @@ gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream,
    MpegTSBaseProgram * target_program)
 {
  GstFlowReturn res = GST_FLOW_OK;
+#ifndef GST_DISABLE_GST_DEBUG
  MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
+#endif
  GstBuffer *buffer = NULL;
  GstBufferList *buffer_list = NULL;