mpegtsmux: allow attaching PCR to non-PES streams

There is an existing PMT mapping between PCR_%s and an mpegtsmux sink
pad name, where %s equals the program number that the PCR corresponds
to. We re-purpose this functionality to also support a mapping between
PCR_%s and an arbitrary PID. If this mapping is set, then the header PCR
PID is set to this value, and PCR is attached to the stream with this
PID.

Note: the current implementation also attaches PCR to the video stream,
so this may be inefficient.

Co-authored-by: Jordan Yelloz <jordan.yelloz@collabora.com>
Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/5726>

subprojects/gst-plugins-bad/gst/mpegtsmux/gstbasetsmux.c

@@ -876,6 +876,32 @@ gst_base_ts_mux_create_stream (GstBaseTsMux * mux, GstBaseTsMuxPad * ts_pad)
   return ret;
 }
 
+static guint16
+get_pmt_pcr_pid (GstBaseTsMux * mux, const gchar * prop_name)
+{
+  if (mux->prog_map == NULL)
+    return 0;
+  gint pcr_pid = 0;
+  if (!gst_structure_get (mux->prog_map, prop_name, G_TYPE_INT, &pcr_pid, NULL))
+    return 0;
+  if (pcr_pid < 1 || pcr_pid > G_MAXUINT16)
+    return 0;
+  return (guint16) pcr_pid;
+}
+
+static gchar *
+get_pmt_pcr_sink (GstBaseTsMux * mux, const gchar * prop_name)
+{
+  if (mux->prog_map == NULL)
+    return 0;
+  gchar *pcr_sink = NULL;
+  if (!gst_structure_get (mux->prog_map, prop_name, G_TYPE_STRING, &pcr_sink,
+          NULL)) {
+    return NULL;
+  }
+  return pcr_sink;
+}
+
 /* Must be called with mux->lock held */
 static GstFlowReturn
 gst_base_ts_mux_create_pad_stream (GstBaseTsMux * mux, GstPad * pad)
@@ -945,6 +971,7 @@ gst_base_ts_mux_create_pad_stream (GstBaseTsMux * mux, GstPad * pad)
     if (ret != GST_FLOW_OK)
       goto no_stream;
   }
+  ts_pad->stream->program = ts_pad->prog;
 
   if (ts_pad->prog->pcr_stream == NULL) {
     /* Take the first stream of the program for the PCR */
@@ -957,17 +984,23 @@ gst_base_ts_mux_create_pad_stream (GstBaseTsMux * mux, GstPad * pad)
 
   /* Check for user-specified PCR PID */
   prop_name = g_strdup_printf ("PCR_%d", ts_pad->prog->pgm_number);
-  if (mux->prog_map && gst_structure_has_field (mux->prog_map, prop_name)) {
-    const gchar *sink_name =
-        gst_structure_get_string (mux->prog_map, prop_name);
-
-    if (!g_strcmp0 (name, sink_name)) {
-      GST_DEBUG_OBJECT (mux, "User specified stream (pid=%d) as PCR for "
-          "program (prog_id = %d)", ts_pad->pid, ts_pad->prog->pgm_number);
-      tsmux_program_set_pcr_stream (ts_pad->prog, ts_pad->stream);
-    }
+  guint16 pcr_pid = get_pmt_pcr_pid (mux, prop_name);
+  if (pcr_pid) {
+    GST_DEBUG_OBJECT (mux, "User specified PID %d as PCR for "
+        "program (prog_id = %d)", pcr_pid, ts_pad->prog->pgm_number);
+    tsmux_program_set_pcr_pid (ts_pad->prog, pcr_pid);
+    goto have_pcr_pid;
   }
-  g_free (prop_name);
+  gchar *pcr_sink_name = get_pmt_pcr_sink (mux, prop_name);
+  if (!g_strcmp0 (GST_PAD_NAME (pad), pcr_sink_name)) {
+    GST_DEBUG_OBJECT (mux, "User specified stream (pid=%d) as PCR for "
+        "program (prog_id = %d)", ts_pad->pid, ts_pad->prog->pgm_number);
+    tsmux_program_set_pcr_stream (ts_pad->prog, ts_pad->stream);
+  }
+  g_clear_pointer (&pcr_sink_name, g_free);
+
+have_pcr_pid:
+  g_clear_pointer (&prop_name, g_free);
 
   return ret;
 
@@ -1383,7 +1416,7 @@ gst_base_ts_mux_aggregate_buffer (GstBaseTsMux * mux,
     }
   }
 
-  if (G_UNLIKELY (prog->pcr_stream == NULL)) {
+  if (!prog->pcr_pid && G_UNLIKELY (prog->pcr_stream == NULL)) {
     /* Take the first data stream for the PCR */
     GST_DEBUG_OBJECT (best,
         "Use stream (pid=%d) from pad as PCR for program (prog_id = %d)",

subprojects/gst-plugins-bad/gst/mpegtsmux/tsmux/tsmux.c

@@ -440,6 +440,7 @@ tsmux_program_new (TsMux * mux, gint prog_id)
   program->pmt_interval = TSMUX_DEFAULT_PMT_INTERVAL;
 
   program->next_pmt_pcr = -1;
+  program->next_pcr = -1;
 
   if (prog_id == 0) {
     program->pgm_number = mux->next_pgm_no++;
@@ -457,6 +458,7 @@ tsmux_program_new (TsMux * mux, gint prog_id)
 
   program->pmt_pid = mux->next_pmt_pid++;
   program->pcr_stream = NULL;
+  program->pcr_pid = 0;
 
   /* SCTE35 is disabled by default */
   program->scte35_pid = 0;
@@ -675,6 +677,7 @@ tsmux_program_set_pcr_stream (TsMuxProgram * program, TsMuxStream * stream)
   if (program->pcr_stream == stream)
     return;
 
+  program->pcr_pid = 0;
   if (program->pcr_stream != NULL)
     tsmux_stream_pcr_unref (program->pcr_stream);
   if (stream)
@@ -684,6 +687,23 @@ tsmux_program_set_pcr_stream (TsMuxProgram * program, TsMuxStream * stream)
   program->pmt_changed = TRUE;
 }
 
+/**
+ * tsmux_program_set_pcr_pid:
+ * @program: a #TsMuxProgram
+ * @pid: a PID
+ *
+ * Set @pid as the PCR PID for @program, overwriting the previously
+ * configured PCR PID. When pid == 0, program will have no PCR PID configured.
+ */
+void
+tsmux_program_set_pcr_pid (TsMuxProgram * program, guint16 pid)
+{
+  g_return_if_fail (program != NULL);
+
+  program->pcr_pid = pid;
+  program->pmt_changed = TRUE;
+}
+
 /**
  * tsmux_get_new_pid:
  * @mux: a #TsMux
@@ -1552,6 +1572,25 @@ done:
   return ret;
 }
 
+static gint64
+write_new_prog_pcr (TsMux * mux, TsMuxProgram * prog, gint64 cur_pcr)
+{
+  if (prog->next_pcr == -1 || cur_pcr > prog->next_pcr) {
+    prog->pi.flags |=
+        TSMUX_PACKET_FLAG_ADAPTATION | TSMUX_PACKET_FLAG_WRITE_PCR;
+    prog->pi.pcr = cur_pcr;
+
+    if (prog->next_pcr == -1)
+      prog->next_pcr = cur_pcr + mux->pcr_interval * 300;
+    else
+      prog->next_pcr += mux->pcr_interval * 300;
+  } else {
+    cur_pcr = -1;
+  }
+
+  return cur_pcr;
+}
+
 /**
  * tsmux_write_stream_packet:
  * @mux: a #TsMux
@@ -1574,7 +1613,7 @@ tsmux_write_stream_packet (TsMux * mux, TsMuxStream * stream)
   g_return_val_if_fail (mux != NULL, FALSE);
   g_return_val_if_fail (stream != NULL, FALSE);
 
-  if (tsmux_stream_is_pcr (stream)) {
+  if (tsmux_stream_is_pcr (stream) || stream->program->pcr_pid) {
     gint64 cur_ts = CLOCK_BASE;
     if (tsmux_stream_get_dts (stream) != G_MININT64)
       cur_ts += tsmux_stream_get_dts (stream);
@@ -1590,6 +1629,29 @@ tsmux_write_stream_packet (TsMux * mux, TsMuxStream * stream)
     new_pcr =
         write_new_pcr (mux, stream, get_current_pcr (mux, cur_ts),
         get_next_pcr (mux, cur_ts));
+
+    if (stream->program->pcr_pid) {
+      /* this should only enter block when time to send a PCR packet */
+      new_pcr = write_new_prog_pcr (mux, stream->program, get_current_pcr (mux,
+              cur_ts));
+      if (new_pcr != -1) {
+        if (!tsmux_get_buffer (mux, &buf))
+          return FALSE;
+
+        if (!gst_buffer_map (buf, &map, GST_MAP_WRITE))
+          goto fail_unmapped;
+
+        if (!tsmux_write_ts_header (mux, map.data, &stream->program->pi, 0,
+                NULL, NULL))
+          goto fail;
+
+        gst_buffer_unmap (buf, &map);
+        stream->program->pi.pid = stream->program->pcr_pid;
+        stream->program->pi.flags &= TSMUX_PACKET_FLAG_PES_FULL_HEADER;
+        if (!tsmux_packet_out (mux, buf, new_pcr))
+          return FALSE;
+      }
+    }
   }
 
   pi->packet_start_unit_indicator = tsmux_stream_at_pes_start (stream);
@@ -1635,6 +1697,9 @@ fail:
     }
     return FALSE;
   }
+fail_unmapped:
+  gst_clear_buffer (&buf);
+  return FALSE;
 }
 
 /**
@@ -1667,7 +1732,10 @@ tsmux_program_free (TsMuxProgram * program)
 void
 tsmux_program_set_pmt_pid (TsMuxProgram * program, guint16 pmt_pid)
 {
+  g_return_if_fail (program != NULL);
+
   program->pmt_pid = pmt_pid;
+  program->pmt_changed = TRUE;
 }
 
 static gint
@@ -1759,8 +1827,10 @@ tsmux_write_pmt (TsMux * mux, TsMuxProgram * program)
 
     pmt = gst_mpegts_pmt_new ();
 
-    if (program->pcr_stream == NULL)
+    if ((program->pcr_stream == NULL) && (program->pcr_pid == 0))
       pmt->pcr_pid = 0x1FFF;
+    else if (program->pcr_pid != 0)
+      pmt->pcr_pid = program->pcr_pid;
     else
       pmt->pcr_pid = tsmux_stream_get_pid (program->pcr_stream);
 

subprojects/gst-plugins-bad/gst/mpegtsmux/tsmux/tsmux.h

@@ -97,6 +97,8 @@ struct TsMuxProgram {
   gboolean wrote_si;
 
   TsMuxSection pmt;
+  TsMuxPacketInfo pi;
+
   /* PMT version */
   guint8   pmt_version;
   /* trigger for writing PMT */
@@ -107,6 +109,7 @@ struct TsMuxProgram {
 
   /* Next PMT position, 27 MHz */
   gint64   next_pmt_pcr;
+  gint64   next_pcr;
 
   /* program ID for the PAT */
   guint16 pgm_number;
@@ -123,6 +126,7 @@ struct TsMuxProgram {
 
   /* stream which carries the PCR */
   TsMuxStream *pcr_stream;
+  guint16 pcr_pid;
 
   /* programs TsMuxStream's */
   GPtrArray *streams;
@@ -229,6 +233,7 @@ gboolean        tsmux_remove_stream             (TsMux *mux, guint16 pid, TsMuxP
 
 void 		tsmux_program_add_stream 	(TsMuxProgram *program, TsMuxStream *stream);
 void 		tsmux_program_set_pcr_stream 	(TsMuxProgram *program, TsMuxStream *stream);
+void           tsmux_program_set_pcr_pid       (TsMuxProgram *program, guint16 pid);
 void    tsmux_set_pcr_interval (TsMux * mux, guint freq);
 
 /* writing stuff */

subprojects/gst-plugins-bad/gst/mpegtsmux/tsmux/tsmuxstream.h

@@ -151,6 +151,9 @@ struct TsMuxStream {
   guint8 id;
   /* extended stream id (13818-1 Amdt 2) */
   guint8 id_extended;
+
+  struct TsMuxProgram *program;
+
   /* requested index in the PMT */
   gint pmt_index;