1 files changed, 124 insertions, 20 deletions

diff --git a/gst/mpegpsmux/mpegpsmux.c b/gst/mpegpsmux/mpegpsmux.c
index 48c0828fb..3fff4be51 100644
--- a/gst/mpegpsmux/mpegpsmux.c
+++ b/gst/mpegpsmux/mpegpsmux.c
@@ -54,9 +54,11 @@ GST_DEBUG_CATEGORY (mpegpsmux_debug);
 
 enum
 {
-  ARG_0
+  PROP_AGGREGATE_GOPS = 1
 };
 
+#define DEFAULT_AGGREGATE_GOPS FALSE
+
 static GstStaticPadTemplate mpegpsmux_sink_factory =
     GST_STATIC_PAD_TEMPLATE ("sink_%u",
     GST_PAD_SINK,
@@ -94,7 +96,7 @@ static gboolean new_packet_cb (guint8 * data, guint len, void *user_data);
 static void release_buffer_cb (guint8 * data, void *user_data);
 
 static gboolean mpegpsdemux_prepare_srcpad (MpegPsMux * mux);
-static GstFlowReturn mpegpsmux_collected (GstCollectPads * pads,
+static GstFlowReturn mpegpsmux_collected (GstCollectPads2 * pads,
     MpegPsMux * mux);
 static GstPad *mpegpsmux_request_new_pad (GstElement * element,
     GstPadTemplate * templ, const gchar * name);
@@ -135,6 +137,10 @@ mpegpsmux_class_init (MpegPsMuxClass * klass)
   gstelement_class->release_pad = mpegpsmux_release_pad;
   gstelement_class->change_state = mpegpsmux_change_state;
 
+  g_object_class_install_property (gobject_class, PROP_AGGREGATE_GOPS,
+      g_param_spec_boolean ("aggregate-gops", "Aggregate GOPs",
+          "Whether to aggregate GOPs and push them out as buffer lists",
+          DEFAULT_AGGREGATE_GOPS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 }
 
 static void
@@ -145,9 +151,9 @@ mpegpsmux_init (MpegPsMux * mux, MpegPsMuxClass * g_class)
   gst_pad_use_fixed_caps (mux->srcpad);
   gst_element_add_pad (GST_ELEMENT (mux), mux->srcpad);
 
-  mux->collect = gst_collect_pads_new ();
-  gst_collect_pads_set_function (mux->collect,
-      (GstCollectPadsFunction) GST_DEBUG_FUNCPTR (mpegpsmux_collected), mux);
+  mux->collect = gst_collect_pads2_new ();
+  gst_collect_pads2_set_function (mux->collect,
+      (GstCollectPads2Function) GST_DEBUG_FUNCPTR (mpegpsmux_collected), mux);
 
   mux->psmux = psmux_new ();
   psmux_set_write_func (mux->psmux, new_packet_cb, mux);
@@ -171,6 +177,11 @@ mpegpsmux_dispose (GObject * object)
     mux->psmux = NULL;
   }
 
+  if (mux->gop_list != NULL) {
+    gst_buffer_list_unref (mux->gop_list);
+    mux->gop_list = NULL;
+  }
+
   GST_CALL_PARENT (G_OBJECT_CLASS, dispose, (object));
 }
 
@@ -178,9 +189,12 @@ static void
 gst_mpegpsmux_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec)
 {
-/*  MpegPsMux *mux = GST_MPEG_PSMUX (object); */
+  MpegPsMux *mux = GST_MPEG_PSMUX (object);
 
   switch (prop_id) {
+    case PROP_AGGREGATE_GOPS:
+      mux->aggregate_gops = g_value_get_boolean (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -191,9 +205,12 @@ static void
 gst_mpegpsmux_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec)
 {
-/*  MpegPsMux *mux = GST_MPEG_PSMUX (object); */
+  MpegPsMux *mux = GST_MPEG_PSMUX (object);
 
   switch (prop_id) {
+    case PROP_AGGREGATE_GOPS:
+      g_value_set_boolean (value, mux->aggregate_gops);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -217,6 +234,7 @@ mpegpsmux_create_stream (MpegPsMux * mux, MpegPsPadData * ps_data, GstPad * pad)
   GstFlowReturn ret = GST_FLOW_ERROR;
   GstCaps *caps = gst_pad_get_negotiated_caps (pad);
   GstStructure *s;
+  gboolean is_video = FALSE;
 
   if (caps == NULL) {
     GST_DEBUG_OBJECT (pad, "Sink pad caps were not set before pushing");
@@ -229,6 +247,7 @@ mpegpsmux_create_stream (MpegPsMux * mux, MpegPsPadData * ps_data, GstPad * pad)
   if (gst_structure_has_name (s, "video/x-dirac")) {
     GST_DEBUG_OBJECT (pad, "Creating Dirac stream");
     ps_data->stream = psmux_create_stream (mux->psmux, PSMUX_ST_VIDEO_DIRAC);
+    is_video = TRUE;
   } else if (gst_structure_has_name (s, "audio/x-ac3")) {
     GST_DEBUG_OBJECT (pad, "Creating AC3 stream");
     ps_data->stream = psmux_create_stream (mux->psmux, PSMUX_ST_PS_AUDIO_AC3);
@@ -252,6 +271,7 @@ mpegpsmux_create_stream (MpegPsMux * mux, MpegPsPadData * ps_data, GstPad * pad)
       ps_data->codec_data = NULL;
     }
     ps_data->stream = psmux_create_stream (mux->psmux, PSMUX_ST_VIDEO_H264);
+    is_video = TRUE;
   } else if (gst_structure_has_name (s, "audio/mpeg")) {
     gint mpegversion;
     if (!gst_structure_get_int (s, "mpegversion", &mpegversion)) {
@@ -312,6 +332,7 @@ mpegpsmux_create_stream (MpegPsMux * mux, MpegPsPadData * ps_data, GstPad * pad)
       GST_DEBUG_OBJECT (pad, "Creating MPEG Video, version 4 stream");
       ps_data->stream = psmux_create_stream (mux->psmux, PSMUX_ST_VIDEO_MPEG4);
     }
+    is_video = TRUE;
   }
 
   if (ps_data->stream != NULL) {
@@ -327,6 +348,11 @@ mpegpsmux_create_stream (MpegPsMux * mux, MpegPsPadData * ps_data, GstPad * pad)
     psmux_stream_set_buffer_release_func (ps_data->stream, release_buffer_cb);
 
     ret = GST_FLOW_OK;
+
+    if (is_video && mux->video_stream_id == 0) {
+      mux->video_stream_id = ps_data->stream_id;
+      GST_INFO_OBJECT (mux, "video pad stream_id 0x%02x", mux->video_stream_id);
+    }
   }
 
 beach:
@@ -343,7 +369,7 @@ mpegpsmux_create_streams (MpegPsMux * mux)
 
   /* Create the streams */
   while (walk) {
-    GstCollectData *c_data = (GstCollectData *) walk->data;
+    GstCollectData2 *c_data = (GstCollectData2 *) walk->data;
     MpegPsPadData *ps_data = (MpegPsPadData *) walk->data;
 
     walk = g_slist_next (walk);
@@ -368,11 +394,11 @@ mpegpsmux_choose_best_stream (MpegPsMux * mux)
   /* Choose from which stream to mux with */
 
   MpegPsPadData *best = NULL;
-  GstCollectData *c_best = NULL;
+  GstCollectData2 *c_best = NULL;
   GSList *walk;
 
   for (walk = mux->collect->data; walk != NULL; walk = g_slist_next (walk)) {
-    GstCollectData *c_data = (GstCollectData *) walk->data;
+    GstCollectData2 *c_data = (GstCollectData2 *) walk->data;
     MpegPsPadData *ps_data = (MpegPsPadData *) walk->data;
 
     if (ps_data->eos == FALSE) {
@@ -380,7 +406,7 @@ mpegpsmux_choose_best_stream (MpegPsMux * mux)
         GstBuffer *buf;
 
         ps_data->queued_buf = buf =
-            gst_collect_pads_peek (mux->collect, c_data);
+            gst_collect_pads2_peek (mux->collect, c_data);
 
         if (buf != NULL) {
           if (ps_data->prepare_func) {
@@ -441,19 +467,34 @@ mpegpsmux_choose_best_stream (MpegPsMux * mux)
     }
   }
   if (c_best) {
-    gst_buffer_unref (gst_collect_pads_pop (mux->collect, c_best));
+    gst_buffer_unref (gst_collect_pads2_pop (mux->collect, c_best));
   }
 
   return best;
 }
 
 static GstFlowReturn
-mpegpsmux_collected (GstCollectPads * pads, MpegPsMux * mux)
+mpegpsmux_push_gop_list (MpegPsMux * mux)
+{
+  GstFlowReturn flow;
+
+  g_assert (mux->gop_list != NULL);
+
+  GST_DEBUG_OBJECT (mux, "Sending pending GOP of %u buffers",
+      gst_buffer_list_n_groups (mux->gop_list));
+  flow = gst_pad_push_list (mux->srcpad, mux->gop_list);
+  mux->gop_list = NULL;
+  return flow;
+}
+
+static GstFlowReturn
+mpegpsmux_collected (GstCollectPads2 * pads, MpegPsMux * mux)
 {
   /* main muxing function */
 
   GstFlowReturn ret = GST_FLOW_OK;
   MpegPsPadData *best = NULL;
+  gboolean keyunit;
 
   GST_DEBUG_OBJECT (mux, "Pads collected");
 
@@ -496,9 +537,20 @@ mpegpsmux_collected (GstCollectPads * pads, MpegPsMux * mux)
           G_GINT64_FORMAT, GST_TIME_ARGS (best->cur_ts), pts);
     }
 
+    /* start of new GOP? */
+    keyunit = !GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
+
+    if (keyunit && best->stream_id == mux->video_stream_id
+        && mux->gop_list != NULL) {
+      ret = mpegpsmux_push_gop_list (mux);
+      if (ret != GST_FLOW_OK)
+        goto done;
+    }
+
     /* give the buffer to libpsmux for processing */
     psmux_stream_add_data (best->stream, GST_BUFFER_DATA (buf),
-        GST_BUFFER_SIZE (buf), buf, pts, -1);
+        GST_BUFFER_SIZE (buf), buf, pts, -1, keyunit);
+
     best->queued_buf = NULL;
 
     /* write the data from libpsmux to stream */
@@ -513,12 +565,17 @@ mpegpsmux_collected (GstCollectPads * pads, MpegPsMux * mux)
   } else {
     /* FIXME: Drain all remaining streams */
     /* At EOS */
+    if (mux->gop_list != NULL)
+      mpegpsmux_push_gop_list (mux);
+
     if (psmux_write_end_code (mux->psmux)) {
       GST_WARNING_OBJECT (mux, "Writing MPEG PS Program end code failed.");
     }
     gst_pad_push_event (mux->srcpad, gst_event_new_eos ());
   }
 
+done:
+
   return ret;
 new_seg_fail:
   return GST_FLOW_ERROR;
@@ -538,8 +595,8 @@ mpegpsmux_request_new_pad (GstElement * element,
 
   pad = gst_pad_new_from_template (templ, name);
 
-  pad_data = (MpegPsPadData *) gst_collect_pads_add_pad (mux->collect, pad,
-      sizeof (MpegPsPadData), NULL);
+  pad_data = (MpegPsPadData *) gst_collect_pads2_add_pad (mux->collect, pad,
+      sizeof (MpegPsPadData));
   if (pad_data == NULL)
     goto pad_failure;
 
@@ -555,7 +612,7 @@ mpegpsmux_request_new_pad (GstElement * element,
 could_not_add:
   GST_ELEMENT_ERROR (element, STREAM, FAILED,
       ("Internal data stream error."), ("Could not add pad to element"));
-  gst_collect_pads_remove_pad (mux->collect, pad);
+  gst_collect_pads2_remove_pad (mux->collect, pad);
   gst_object_unref (pad);
   return NULL;
 pad_failure:
@@ -586,9 +643,31 @@ mpegpsmux_release_pad (GstElement * element, GstPad * pad)
       pad_data->codec_data = NULL;
     }
   }
+  if (pad_data->stream_id == mux->video_stream_id)
+    mux->video_stream_id = 0;
   GST_OBJECT_UNLOCK (pad);
 
-  gst_collect_pads_remove_pad (mux->collect, pad);
+  gst_collect_pads2_remove_pad (mux->collect, pad);
+}
+
+static void
+add_buffer_to_goplist (MpegPsMux * mux, GstBuffer * buf)
+{
+  GstBufferListIterator *it;
+
+  if (mux->gop_list == NULL)
+    mux->gop_list = gst_buffer_list_new ();
+
+  it = gst_buffer_list_iterate (mux->gop_list);
+
+  /* move iterator to end */
+  while (gst_buffer_list_iterator_next_group (it)) {
+    /* .. */
+  }
+
+  gst_buffer_list_iterator_add_group (it);
+  gst_buffer_list_iterator_add (it, buf);
+  gst_buffer_list_iterator_free (it);
 }
 
 static gboolean
@@ -611,7 +690,14 @@ new_packet_cb (guint8 * data, guint len, void *user_data)
 
   memcpy (GST_BUFFER_DATA (buf), data, len);
   GST_BUFFER_TIMESTAMP (buf) = mux->last_ts;
+
+  if (mux->aggregate_gops) {
+    add_buffer_to_goplist (mux, buf);
+    return TRUE;
+  }
+
   ret = gst_pad_push (mux->srcpad, buf);
+
   if (G_UNLIKELY (ret != GST_FLOW_OK)) {
     mux->last_flow_ret = ret;
     return FALSE;
@@ -623,6 +709,9 @@ new_packet_cb (guint8 * data, guint len, void *user_data)
 static gboolean
 mpegpsdemux_prepare_srcpad (MpegPsMux * mux)
 {
+  GValue val = { 0, };
+  GList *headers, *l;
+
   /* prepare the source pad for output */
 
   GstEvent *new_seg =
@@ -634,6 +723,21 @@ mpegpsdemux_prepare_srcpad (MpegPsMux * mux)
 
 /*      gst_static_pad_template_get_caps (&mpegpsmux_src_factory); */
 
+  headers = psmux_get_stream_headers (mux->psmux);
+  g_value_init (&val, GST_TYPE_ARRAY);
+  for (l = headers; l != NULL; l = l->next) {
+    GValue buf_val = { 0, };
+
+    g_value_init (&buf_val, GST_TYPE_BUFFER);
+    gst_value_take_buffer (&buf_val, GST_BUFFER (l->data));
+    l->data = NULL;
+    gst_value_array_append_value (&val, &buf_val);
+    g_value_unset (&buf_val);
+  }
+  gst_caps_set_value (caps, "streamheader", &val);
+  g_value_unset (&val);
+  g_list_free (headers);
+
   /* Set caps on src pad from our template and push new segment */
   gst_pad_set_caps (mux->srcpad, caps);
 
@@ -657,12 +761,12 @@ mpegpsmux_change_state (GstElement * element, GstStateChange transition)
     case GST_STATE_CHANGE_NULL_TO_READY:
       break;
     case GST_STATE_CHANGE_READY_TO_PAUSED:
-      gst_collect_pads_start (mux->collect);
+      gst_collect_pads2_start (mux->collect);
       break;
     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
       break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:
-      gst_collect_pads_stop (mux->collect);
+      gst_collect_pads2_stop (mux->collect);
       break;
     case GST_STATE_CHANGE_READY_TO_NULL:
       break;