tests/check/elements/multifdsink.c: add two more tests, one doing streamheader

Original commit message from CVS:
* tests/check/elements/multifdsink.c: (wait_bytes_served),
(GST_START_TEST), (fail_unless_read), (multifdsink_suite):
add two more tests, one doing streamheader

ChangeLog

@@ -1,3 +1,9 @@
+2006-05-10  Thomas Vander Stichele  <thomas at apestaart dot org>
+
+	* tests/check/elements/multifdsink.c: (wait_bytes_served),
+	(GST_START_TEST), (fail_unless_read), (multifdsink_suite):
+	  add two more tests, one doing streamheader
+
 2006-05-10  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* gst/tcp/gstmultifdsink.c: (gst_multi_fd_sink_stop):

tests/check/elements/multifdsink.c

@@ -19,6 +19,10 @@
  */
 
 #include <unistd.h>
+#include <sys/ioctl.h>
+#ifdef HAVE_FIONREAD_IN_SYS_FILIO
+#include <sys/filio.h>
+#endif
 
 #include <gst/check/gstcheck.h>
 
@@ -51,6 +55,16 @@ cleanup_multifdsink (GstElement * multifdsink)
   gst_check_teardown_element (multifdsink);
 }
 
+static void
+wait_bytes_served (GstElement * sink, guint64 bytes)
+{
+  guint64 bytes_served = 0;
+
+  while (bytes_served != bytes) {
+    g_object_get (sink, "bytes-served", &bytes_served, NULL);
+  }
+}
+
 GST_START_TEST (test_no_clients)
 {
   GstElement *sink;
@@ -70,6 +84,161 @@ GST_START_TEST (test_no_clients)
 
 GST_END_TEST;
 
+GST_START_TEST (test_add_client)
+{
+  GstElement *sink;
+  GstBuffer *buffer;
+  int pfd[2];
+  gchar data[4];
+  guint64 bytes_served;
+
+  sink = setup_multifdsink ();
+
+  fail_if (pipe (pfd) == -1);
+
+  ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
+
+  /* add the client */
+  g_signal_emit_by_name (sink, "add", pfd[1]);
+
+  buffer = gst_buffer_new_and_alloc (4);
+  memcpy (GST_BUFFER_DATA (buffer), "dead", 4);
+  fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
+
+  fail_if (read (pfd[0], data, 4) < 4);
+  fail_unless (strncmp (data, "dead", 4) == 0);
+  wait_bytes_served (sink, 4);
+
+  GST_DEBUG ("cleaning up multifdsink");
+  ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
+  cleanup_multifdsink (sink);
+}
+
+GST_END_TEST;
+
+static void
+fail_unless_read (const char *msg, int fd, int size, const char *ref)
+{
+  char data[size];
+  int nbytes;
+
+  GST_DEBUG ("%s: reading %d bytes", msg, size);
+  nbytes = read (fd, data, size);
+  GST_DEBUG ("%s: read %d bytes", msg, nbytes);
+  fail_if (nbytes < size);
+  fail_unless (memcmp (data, ref, size) == 0, "data read differs from '%s'",
+      ref);
+}
+
+/* this test:
+ * - adds a first client
+ * - sets streamheader caps on the pad
+ * - pushes the IN_CAPS buffers
+ * - pushes a buffer
+ * - verifies that the client received all the data correctly
+ * - adds a second client
+ * - verifies that this second client receives the streamheader caps too
+ */
+GST_START_TEST (test_streamheader)
+{
+  GstElement *sink;
+  GstBuffer *hbuf1, *hbuf2, *buf;
+  GstCaps *caps;
+  GstStructure *structure;
+  int pfd1[2], pfd2[2];
+  guint8 data[12];
+  GValue array = { 0 };
+  GValue value = { 0 };
+  guint64 bytes_served;
+  int avail;
+
+  sink = setup_multifdsink ();
+
+  fail_if (pipe (pfd1) == -1);
+  fail_if (pipe (pfd2) == -1);
+
+  ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
+
+  /* add the first client */
+  g_signal_emit_by_name (sink, "add", pfd1[1]);
+
+  /* create caps with streamheader, set the caps, and push the IN_CAPS
+   * buffers */
+  hbuf1 = gst_buffer_new_and_alloc (4);
+  GST_BUFFER_FLAG_SET (hbuf1, GST_BUFFER_FLAG_IN_CAPS);
+  memcpy (GST_BUFFER_DATA (hbuf1), "babe", 4);
+  hbuf2 = gst_buffer_new_and_alloc (8);
+  GST_BUFFER_FLAG_SET (hbuf2, GST_BUFFER_FLAG_IN_CAPS);
+  memcpy (GST_BUFFER_DATA (hbuf2), "deadbeef", 8);
+  /* we want to keep them around for the tests */
+  gst_buffer_ref (hbuf1);
+  gst_buffer_ref (hbuf2);
+
+  g_value_init (&array, GST_TYPE_ARRAY);
+
+  g_value_init (&value, GST_TYPE_BUFFER);
+  gst_value_set_buffer (&value, hbuf1);
+  gst_value_array_append_value (&array, &value);
+  g_value_unset (&value);
+
+  g_value_init (&value, GST_TYPE_BUFFER);
+  gst_value_set_buffer (&value, hbuf2);
+  gst_value_array_append_value (&array, &value);
+  g_value_unset (&value);
+
+  caps = gst_caps_from_string ("application/x-gst-check");
+  structure = gst_caps_get_structure (caps, 0);
+
+  gst_structure_set_value (structure, "streamheader", &array);
+  g_value_unset (&array);
+
+  fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
+  fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
+
+  /* verify this hasn't triggered a write yet */
+  /* FIXME: possibly racy, since if it would write, we may not get it
+   * immediately ? */
+  fail_if (ioctl (pfd1[0], FIONREAD, &avail) < 0);
+  fail_if (avail > 0);
+
+  /* push a non-IN_CAPS buffer, this should trigger the client receiving the
+   * first three buffers */
+  buf = gst_buffer_new_and_alloc (4);
+  memcpy (GST_BUFFER_DATA (buf), "f00d", 4);
+  gst_pad_push (mysrcpad, buf);
+
+  fail_unless_read ("first client", pfd1[0], 4, "babe");
+  fail_unless_read ("first client", pfd1[0], 8, "deadbeef");
+  fail_unless_read ("first client", pfd1[0], 4, "f00d");
+  wait_bytes_served (sink, 16);
+
+  /* now add the second client */
+  g_signal_emit_by_name (sink, "add", pfd2[1]);
+  fail_if (ioctl (pfd2[0], FIONREAD, &avail) < 0);
+  fail_if (avail > 0);
+
+  /* now push another buffer, which will trigger streamheader for second
+   * client */
+  buf = gst_buffer_new_and_alloc (4);
+  memcpy (GST_BUFFER_DATA (buf), "deaf", 4);
+  gst_pad_push (mysrcpad, buf);
+
+  fail_unless_read ("first client", pfd1[0], 4, "deaf");
+
+  fail_unless_read ("second client", pfd2[0], 4, "babe");
+  fail_unless_read ("second client", pfd2[0], 8, "deadbeef");
+  /* we missed the f00d buffer */
+  fail_unless_read ("second client", pfd2[0], 4, "deaf");
+  wait_bytes_served (sink, 36);
+
+  GST_DEBUG ("cleaning up multifdsink");
+  ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
+  cleanup_multifdsink (sink);
+}
+
+GST_END_TEST;
+
+
 Suite *
 multifdsink_suite (void)
 {
@@ -78,6 +247,8 @@ multifdsink_suite (void)
 
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_no_clients);
+  tcase_add_test (tc_chain, test_add_client);
+  tcase_add_test (tc_chain, test_streamheader);
 
   return s;
 }