multiudpsink: add support for buffer lists

Add support for BufferList and add a unit test.

Fixes #585842

gst/udp/gstmultiudpsink.c

@@ -117,6 +117,8 @@ static void gst_multiudpsink_finalize (GObject * object);
 
 static GstFlowReturn gst_multiudpsink_render (GstBaseSink * sink,
     GstBuffer * buffer);
+static GstFlowReturn gst_multiudpsink_render_list (GstBaseSink * bsink,
+    GstBufferList * list);
 static GstStateChangeReturn gst_multiudpsink_change_state (GstElement *
     element, GstStateChange transition);
 
@@ -318,6 +320,7 @@ gst_multiudpsink_class_init (GstMultiUDPSinkClass * klass)
   gstelement_class->change_state = gst_multiudpsink_change_state;
 
   gstbasesink_class->render = gst_multiudpsink_render;
+  gstbasesink_class->render_list = gst_multiudpsink_render_list;
   klass->add = gst_multiudpsink_add;
   klass->remove = gst_multiudpsink_remove;
   klass->clear = gst_multiudpsink_clear;
@@ -427,6 +430,93 @@ gst_multiudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
   return GST_FLOW_OK;
 }
 
+static GstFlowReturn
+gst_multiudpsink_render_list (GstBaseSink * bsink, GstBufferList * list)
+{
+  GstMultiUDPSink *sink;
+  GList *clients;
+  gint ret, size = 0, num = 0, no_clients = 0;
+  struct iovec *iov;
+  struct msghdr msg = { 0 };
+
+  GstBufferListIterator *it;
+  guint gsize;
+  GstBuffer *buf;
+
+  sink = GST_MULTIUDPSINK (bsink);
+
+  g_return_val_if_fail (list != NULL, GST_FLOW_ERROR);
+  g_return_val_if_fail ((it = gst_buffer_list_iterate (list)) != NULL,
+      GST_FLOW_ERROR);
+
+  while (gst_buffer_list_iterator_next_group (it)) {
+    msg.msg_iovlen = 0;
+    size = 0;
+
+    if ((gsize = gst_buffer_list_iterator_n_buffers (it)) == 0) {
+      goto invalid_list;
+    }
+
+    iov = (struct iovec *) g_malloc (gsize * sizeof (struct iovec));
+    msg.msg_iov = iov;
+
+    while ((buf = gst_buffer_list_iterator_next (it))) {
+      msg.msg_iov[msg.msg_iovlen].iov_len = GST_BUFFER_SIZE (buf);
+      msg.msg_iov[msg.msg_iovlen].iov_base = GST_BUFFER_DATA (buf);
+      msg.msg_iovlen++;
+      size += GST_BUFFER_SIZE (buf);
+    }
+
+    sink->bytes_to_serve += size;
+
+    /* grab lock while iterating and sending to clients, this should be
+     * fast as UDP never blocks */
+    g_mutex_lock (sink->client_lock);
+    GST_LOG_OBJECT (bsink, "about to send %d bytes", size);
+
+    for (clients = sink->clients; clients; clients = g_list_next (clients)) {
+      GstUDPClient *client;
+
+      client = (GstUDPClient *) clients->data;
+      no_clients++;
+      GST_LOG_OBJECT (sink, "sending %d bytes to client %p", size, client);
+
+      while (TRUE) {
+        msg.msg_name = (void *) &client->theiraddr;
+        msg.msg_namelen = sizeof (client->theiraddr);
+        ret = sendmsg (*client->sock, &msg, 0);
+
+        if (ret < 0) {
+          if (errno != EINTR && errno != EAGAIN) {
+            break;
+          }
+        } else {
+          num++;
+          client->bytes_sent += ret;
+          client->packets_sent++;
+          sink->bytes_served += ret;
+          break;
+        }
+      }
+    }
+    g_mutex_unlock (sink->client_lock);
+
+    g_free (iov);
+    msg.msg_iov = NULL;
+
+    GST_LOG_OBJECT (sink, "sent %d bytes to %d (of %d) clients", size, num,
+        no_clients);
+  }
+
+  gst_buffer_list_iterator_free (it);
+
+  return GST_FLOW_OK;
+
+invalid_list:
+  gst_buffer_list_iterator_free (it);
+  return GST_FLOW_ERROR;
+}
+
 static void
 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
     const gchar * string)

tests/check/Makefile.am

@@ -111,6 +111,7 @@ check_PROGRAMS = \
 	elements/rgvolume \
 	elements/rtp-payloading \
 	elements/spectrum \
+	elements/udpsink \
 	elements/videocrop \
 	elements/videofilter \
 	elements/y4menc \

tests/check/elements/udpsink.c

@@ -0,0 +1,199 @@
+/* GStreamer RTP payloader unit tests
+ * Copyright (C) 2009 Axis Communications <dev-gstreamer@axis.com>
+ * @author Ognyan Tonchev <ognyan@axis.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+#include <gst/check/gstcheck.h>
+#include <gst/base/gstbasesink.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
+    GST_PAD_SRC,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
+#define RTP_HEADER_SIZE 12
+#define RTP_PAYLOAD_SIZE 1024
+
+/*
+ * Number of bytes received in the render function when using buffer lists
+ */
+static guint render_list_bytes_received;
+
+/*
+ * Render function for testing udpsink with buffer lists
+ */
+static GstFlowReturn
+udpsink_render (GstBaseSink * sink, GstBufferList * list)
+{
+  GstBufferListIterator *it;
+
+  fail_if (!list);
+
+  /*
+   * Count the size of the rtp header and the payload in the buffer list.
+   */
+
+  it = gst_buffer_list_iterate (list);
+
+  /* Loop through all groups */
+  while (gst_buffer_list_iterator_next_group (it)) {
+    GstBuffer *buf;
+    /* Loop through all buffers in the current group */
+    while ((buf = gst_buffer_list_iterator_next (it))) {
+      render_list_bytes_received += GST_BUFFER_SIZE (buf);
+    }
+  }
+
+  gst_buffer_list_iterator_free (it);
+
+  return GST_FLOW_OK;
+}
+
+static void
+_set_render_function (GstElement * bsink)
+{
+  GstBaseSinkClass *bsclass;
+  bsclass = GST_BASE_SINK_GET_CLASS ((GstBaseSink *) bsink);
+  /* Add callback function for the buffer list tests */
+  bsclass->render_list = udpsink_render;
+}
+
+static GstBufferList *
+_create_buffer_list (guint * data_size)
+{
+  GstBufferList *list;
+  GstBufferListIterator *it;
+  GstBuffer *rtp_buffer;
+  GstBuffer *data_buffer;
+
+  list = gst_buffer_list_new ();
+  it = gst_buffer_list_iterate (list);
+
+  /*** First group, i.e. first packet. **/
+
+  /* Create the RTP header buffer */
+  rtp_buffer = gst_buffer_new ();
+  GST_BUFFER_MALLOCDATA (rtp_buffer) = g_malloc (RTP_HEADER_SIZE);
+  GST_BUFFER_DATA (rtp_buffer) = GST_BUFFER_MALLOCDATA (rtp_buffer);
+  GST_BUFFER_SIZE (rtp_buffer) = RTP_HEADER_SIZE;
+  memset (GST_BUFFER_DATA (rtp_buffer), 0, RTP_HEADER_SIZE);
+
+  /* Create the buffer that holds the payload */
+  data_buffer = gst_buffer_new ();
+  GST_BUFFER_MALLOCDATA (data_buffer) = g_malloc (RTP_PAYLOAD_SIZE);
+  GST_BUFFER_DATA (data_buffer) = GST_BUFFER_MALLOCDATA (data_buffer);
+  GST_BUFFER_SIZE (data_buffer) = RTP_PAYLOAD_SIZE;
+  memset (GST_BUFFER_DATA (data_buffer), 0, RTP_PAYLOAD_SIZE);
+
+  /* Create a new group to hold the rtp header and the payload */
+  gst_buffer_list_iterator_add_group (it);
+  gst_buffer_list_iterator_add (it, rtp_buffer);
+  gst_buffer_list_iterator_add (it, data_buffer);
+
+  /***  Second group, i.e. second packet. ***/
+
+  /* Create the RTP header buffer */
+  rtp_buffer = gst_buffer_new ();
+  GST_BUFFER_MALLOCDATA (rtp_buffer) = g_malloc (RTP_HEADER_SIZE);
+  GST_BUFFER_DATA (rtp_buffer) = GST_BUFFER_MALLOCDATA (rtp_buffer);
+  GST_BUFFER_SIZE (rtp_buffer) = RTP_HEADER_SIZE;
+  memset (GST_BUFFER_DATA (rtp_buffer), 0, RTP_HEADER_SIZE);
+
+  /* Create the buffer that holds the payload */
+  data_buffer = gst_buffer_new ();
+  GST_BUFFER_MALLOCDATA (data_buffer) = g_malloc (RTP_PAYLOAD_SIZE);
+  GST_BUFFER_DATA (data_buffer) = GST_BUFFER_MALLOCDATA (data_buffer);
+  GST_BUFFER_SIZE (data_buffer) = RTP_PAYLOAD_SIZE;
+  memset (GST_BUFFER_DATA (data_buffer), 0, RTP_PAYLOAD_SIZE);
+
+  /* Create a new group to hold the rtp header and the payload */
+  gst_buffer_list_iterator_add_group (it);
+  gst_buffer_list_iterator_add (it, rtp_buffer);
+  gst_buffer_list_iterator_add (it, data_buffer);
+
+  /* Calculate the size of the data */
+  *data_size = 2 * RTP_HEADER_SIZE + 2 * RTP_PAYLOAD_SIZE;
+
+  gst_buffer_list_iterator_free (it);
+
+  return list;
+}
+
+static void
+udpsink_test (gboolean use_buffer_lists)
+{
+  GstElement *udpsink;
+  GstPad *srcpad;
+  GstBufferList *list;
+  guint data_size;
+
+  list = _create_buffer_list (&data_size);
+
+  udpsink = gst_check_setup_element ("udpsink");
+  if (use_buffer_lists)
+    _set_render_function (udpsink);
+
+  srcpad = gst_check_setup_src_pad_by_name (udpsink, &srctemplate, "sink");
+
+  gst_element_set_state (udpsink, GST_STATE_PLAYING);
+
+  gst_pad_push_list (srcpad, list);
+
+  gst_check_teardown_pad_by_name (udpsink, "sink");
+  gst_check_teardown_element (udpsink);
+
+  if (use_buffer_lists)
+    fail_if (data_size != render_list_bytes_received);
+}
+
+GST_START_TEST (test_udpsink)
+{
+  udpsink_test (FALSE);
+}
+
+GST_END_TEST;
+GST_START_TEST (test_udpsink_bufferlist)
+{
+  udpsink_test (TRUE);
+}
+
+GST_END_TEST;
+
+/*
+ * Creates the test suite.
+ *
+ * Returns: pointer to the test suite.
+ */
+static Suite *
+udpsink_suite ()
+{
+  Suite *s = suite_create ("udpsink_test");
+
+  TCase *tc_chain = tcase_create ("linear");
+
+  /* Set timeout to 60 seconds. */
+  tcase_set_timeout (tc_chain, 60);
+
+  suite_add_tcase (s, tc_chain);
+  tcase_add_test (tc_chain, test_udpsink);
+  tcase_add_test (tc_chain, test_udpsink_bufferlist);
+  return s;
+}
+
+GST_CHECK_MAIN (udpsink)