srtpdec: Add support for buffer list

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

diff --git a/ext/srtp/gstsrtpdec.c b/ext/srtp/gstsrtpdec.c
index cde941887454f9c52fc6f5b4b9aba318202f8bf7..4b1781a4f6bbea3e6a72d6509ca7a3c0002d06cf 100644 (file)
--- a/ext/srtp/gstsrtpdec.c
+++ b/ext/srtp/gstsrtpdec.c
@@ -142,6 +142,22 @@ enum
   PROP_0
 };
 
+typedef struct _ValidateBufferItData
+{
+  GstSrtpDecSsrcStream **stream;
+  GstSrtpDec *filter;
+  guint32 *ssrc;
+  gboolean *is_rtcp;
+} ValidateBufferItData;
+
+typedef struct _DecodeBufferItData
+{
+  GstSrtpDec *filter;
+  GstPad *pad;
+  guint32 ssrc;
+  gboolean is_rtcp;
+} DecodeBufferItData;
+
 /* the capabilities of the inputs and outputs.
  *
  * describe the real formats here.
@@ -202,6 +218,11 @@ static GstFlowReturn gst_srtp_dec_chain_rtp (GstPad * pad,
 static GstFlowReturn gst_srtp_dec_chain_rtcp (GstPad * pad,
     GstObject * parent, GstBuffer * buf);
 
+static GstFlowReturn gst_srtp_dec_chain_list_rtp (GstPad * pad,
+    GstObject * parent, GstBufferList * buf_list);
+static GstFlowReturn gst_srtp_dec_chain_list_rtcp (GstPad * pad,
+    GstObject * parent, GstBufferList * buf_list);
+
 static GstStateChangeReturn gst_srtp_dec_change_state (GstElement * element,
     GstStateChange transition);
 
@@ -347,6 +368,8 @@ gst_srtp_dec_init (GstSrtpDec * filter)
       GST_DEBUG_FUNCPTR (gst_srtp_dec_iterate_internal_links_rtp));
   gst_pad_set_chain_function (filter->rtp_sinkpad,
       GST_DEBUG_FUNCPTR (gst_srtp_dec_chain_rtp));
+  gst_pad_set_chain_list_function (filter->rtp_sinkpad,
+      GST_DEBUG_FUNCPTR (gst_srtp_dec_chain_list_rtp));
 
   filter->rtp_srcpad =
       gst_pad_new_from_static_template (&rtp_src_template, "rtp_src");
@@ -370,6 +393,8 @@ gst_srtp_dec_init (GstSrtpDec * filter)
       GST_DEBUG_FUNCPTR (gst_srtp_dec_iterate_internal_links_rtcp));
   gst_pad_set_chain_function (filter->rtcp_sinkpad,
       GST_DEBUG_FUNCPTR (gst_srtp_dec_chain_rtcp));
+  gst_pad_set_chain_list_function (filter->rtcp_sinkpad,
+      GST_DEBUG_FUNCPTR (gst_srtp_dec_chain_list_rtcp));
 
   filter->rtcp_srcpad =
       gst_pad_new_from_static_template (&rtcp_src_template, "rtcp_src");
@@ -1121,7 +1146,6 @@ gst_srtp_dec_chain (GstPad * pad, GstObject * parent, GstBuffer * buf,
   GST_OBJECT_LOCK (filter);
 
   /* Check if this stream exists, if not create a new stream */
-
   if (!(stream = validate_buffer (filter, buf, &ssrc, &is_rtcp))) {
     GST_OBJECT_UNLOCK (filter);
     GST_WARNING_OBJECT (filter, "Invalid buffer, dropping");
@@ -1169,6 +1193,107 @@ drop_buffer:
   return ret;
 }
 
+static gboolean
+validate_buffer_it (GstBuffer ** buffer, guint index, gpointer user_data)
+{
+  ValidateBufferItData *data = user_data;
+
+  if (!(*data->stream =
+          validate_buffer (data->filter, *buffer, data->ssrc, data->is_rtcp))) {
+    GST_WARNING_OBJECT (data->filter, "Invalid buffer, dropping");
+    gst_buffer_replace (buffer, NULL);
+  }
+
+  return TRUE;
+}
+
+static gboolean
+decode_buffer_it (GstBuffer ** buffer, guint index, gpointer user_data)
+{
+  DecodeBufferItData *data = user_data;
+
+  if (!gst_srtp_dec_decode_buffer (data->filter, data->pad, *buffer,
+          data->is_rtcp, data->ssrc)) {
+    GST_WARNING_OBJECT (data->filter, "Error decoding buffer, dropping");
+    gst_buffer_replace (buffer, NULL);
+  }
+
+  return TRUE;
+}
+
+static GstFlowReturn
+gst_srtp_dec_chain_list (GstPad * pad, GstObject * parent,
+    GstBufferList * buf_list, gboolean is_rtcp)
+{
+  GstSrtpDec *filter = GST_SRTP_DEC (parent);
+  GstPad *otherpad;
+  GstSrtpDecSsrcStream *stream = NULL;
+  GstFlowReturn ret = GST_FLOW_OK;
+  guint32 ssrc = 0;
+  ValidateBufferItData validate_data;
+  DecodeBufferItData decode_data;
+
+  validate_data.stream = &stream;
+  validate_data.filter = filter;
+  validate_data.ssrc = &ssrc;
+  validate_data.is_rtcp = &is_rtcp;
+
+  decode_data.filter = filter;
+  decode_data.pad = pad;
+
+  GST_OBJECT_LOCK (filter);
+
+  /* Check if this stream exists, if not create a new stream */
+  gst_buffer_list_foreach (buf_list, validate_buffer_it, &validate_data);
+
+  if (!gst_buffer_list_length (buf_list)) {
+    GST_OBJECT_LOCK (filter);
+    gst_buffer_list_unref (buf_list);
+    return GST_FLOW_OK;
+  }
+
+  if (!STREAM_HAS_CRYPTO (stream)) {
+    GST_OBJECT_UNLOCK (filter);
+    goto push_out;
+  }
+
+  decode_data.ssrc = ssrc;
+  decode_data.is_rtcp = is_rtcp;
+
+  gst_buffer_list_foreach (buf_list, decode_buffer_it, &decode_data);
+
+  GST_OBJECT_UNLOCK (filter);
+
+  if (!gst_buffer_list_length (buf_list)) {
+    gst_buffer_list_unref (buf_list);
+    return GST_FLOW_OK;
+  }
+
+  /* If all is well, we may have reached soft limit */
+  if (gst_srtp_get_soft_limit_reached ())
+    request_key_with_signal (filter, ssrc, SIGNAL_SOFT_LIMIT);
+
+push_out:
+  /* Push buffer list to source pad */
+  if (is_rtcp) {
+    otherpad = filter->rtcp_srcpad;
+    if (!filter->rtcp_has_segment)
+      gst_srtp_dec_push_early_events (filter, filter->rtcp_srcpad,
+          filter->rtp_srcpad, TRUE);
+  } else {
+    otherpad = filter->rtp_srcpad;
+    if (!filter->rtp_has_segment)
+      gst_srtp_dec_push_early_events (filter, filter->rtp_srcpad,
+          filter->rtcp_srcpad, FALSE);
+  }
+
+  GST_LOG_OBJECT (pad, "Pushing buffer chain of %d",
+      gst_buffer_list_length (buf_list));
+  ret = gst_pad_push_list (otherpad, buf_list);
+
+  return ret;
+}
+
 static GstFlowReturn
 gst_srtp_dec_chain_rtp (GstPad * pad, GstObject * parent, GstBuffer * buf)
 {
@@ -1181,6 +1306,21 @@ gst_srtp_dec_chain_rtcp (GstPad * pad, GstObject * parent, GstBuffer * buf)
   return gst_srtp_dec_chain (pad, parent, buf, TRUE);
 }
 
+static GstFlowReturn
+gst_srtp_dec_chain_list_rtp (GstPad * pad, GstObject * parent,
+    GstBufferList * buf_list)
+{
+  return gst_srtp_dec_chain_list (pad, parent, buf_list, FALSE);
+}
+
+static GstFlowReturn
+gst_srtp_dec_chain_list_rtcp (GstPad * pad, GstObject * parent,
+    GstBufferList * buf_list)
+{
+  return gst_srtp_dec_chain_list (pad, parent, buf_list, TRUE);
+}
+
+
 static GstStateChangeReturn
 gst_srtp_dec_change_state (GstElement * element, GstStateChange transition)
 {