tests/check/elements/multifdsink.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) 2006 Thomas Vander Stichele <thomas at apestaart dot org>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  */
  20
  21 #include <unistd.h>
  22 #include <sys/ioctl.h>
  23 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
  24 #include <sys/filio.h>
  25 #endif
  26
  27 #include <gst/check/gstcheck.h>
  28
  29 /* FIXME: remove this header once formats are refactored */
  30 #include "gst/tcp/gstmultifdsink.h"
  31
  32 static GstPad *mysrcpad;
  33
  34 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  35     GST_PAD_SRC,
  36     GST_PAD_ALWAYS,
  37     GST_STATIC_CAPS ("application/x-gst-check")
  38     );
  39
  40 static GstElement *
  41 setup_multifdsink (void)
  42 {
  43   GstElement *multifdsink;
  44
  45   GST_DEBUG ("setup_multifdsink");
  46   multifdsink = gst_check_setup_element ("multifdsink");
  47   mysrcpad = gst_check_setup_src_pad (multifdsink, &srctemplate);
  48   GST_PAD_UNSET_FLUSHING (mysrcpad);
  49
  50   return multifdsink;
  51 }
  52
  53 static void
  54 cleanup_multifdsink (GstElement * multifdsink)
  55 {
  56   GST_DEBUG ("cleanup_multifdsink");
  57
  58   gst_check_teardown_src_pad (multifdsink);
  59   gst_check_teardown_element (multifdsink);
  60 }
  61
  62 static void
  63 wait_bytes_served (GstElement * sink, guint64 bytes)
  64 {
  65   guint64 bytes_served = 0;
  66
  67   while (bytes_served != bytes) {
  68     g_object_get (sink, "bytes-served", &bytes_served, NULL);
  69   }
  70 }
  71
  72 /* FIXME: possibly racy, since if it would write, we may not get it
  73  * immediately ? */
  74 #define fail_if_can_read(msg,fd) \
  75 G_STMT_START { \
  76   long avail; \
  77 \
  78   fail_if (ioctl (fd, FIONREAD, &avail) < 0, "%s: could not ioctl", msg); \
  79   fail_if (avail > 0, "%s: has bytes available to read"); \
  80 } G_STMT_END;
  81
  82
  83 GST_START_TEST (test_no_clients)
  84 {
  85   GstElement *sink;
  86   GstBuffer *buffer;
  87   GstCaps *caps;
  88
  89   sink = setup_multifdsink ();
  90
  91   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
  92
  93   caps = gst_caps_from_string ("application/x-gst-check");
  94   buffer = gst_buffer_new_and_alloc (4);
  95   gst_pad_set_caps (mysrcpad, caps);
  96   gst_caps_unref (caps);
  97   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
  98
  99   GST_DEBUG ("cleaning up multifdsink");
 100   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 101   cleanup_multifdsink (sink);
 102 }
 103
 104 GST_END_TEST;
 105
 106 GST_START_TEST (test_add_client)
 107 {
 108   GstElement *sink;
 109   GstBuffer *buffer;
 110   GstCaps *caps;
 111   int pfd[2];
 112   gchar data[4];
 113
 114   sink = setup_multifdsink ();
 115
 116   fail_if (pipe (pfd) == -1);
 117
 118   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 119
 120   /* add the client */
 121   g_signal_emit_by_name (sink, "add", pfd[1]);
 122
 123   caps = gst_caps_from_string ("application/x-gst-check");
 124   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 125   GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
 126   buffer = gst_buffer_new_and_alloc (4);
 127   gst_pad_set_caps (mysrcpad, caps);
 128   ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
 129   gst_buffer_fill (buffer, 0, "dead", 4);
 130   fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 131
 132   GST_DEBUG ("reading");
 133   fail_if (read (pfd[0], data, 4) < 4);
 134   fail_unless (strncmp (data, "dead", 4) == 0);
 135   wait_bytes_served (sink, 4);
 136
 137   GST_DEBUG ("cleaning up multifdsink");
 138   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 139   cleanup_multifdsink (sink);
 140
 141   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 142   gst_caps_unref (caps);
 143 }
 144
 145 GST_END_TEST;
 146
 147 #define fail_unless_read(msg,fd,size,ref) \
 148 G_STMT_START { \
 149   char data[size + 1]; \
 150   int nbytes; \
 151 \
 152   GST_DEBUG ("%s: reading %d bytes", msg, size); \
 153   nbytes = read (fd, data, size); \
 154   data[size] = 0; \
 155   GST_DEBUG ("%s: read %d bytes", msg, nbytes); \
 156   fail_if (nbytes < size); \
 157   fail_unless (memcmp (data, ref, size) == 0, \
 158       "data read '%s' differs from '%s'", data, ref); \
 159 } G_STMT_END;
 160
 161 /* from the given two data buffers, create two streamheader buffers and
 162  * some caps that match it, and store them in the given pointers
 163  * returns  one ref to each of the buffers and the caps */
 164 static void
 165 gst_multifdsink_create_streamheader (const gchar * data1,
 166     const gchar * data2, GstBuffer ** hbuf1, GstBuffer ** hbuf2,
 167     GstCaps ** caps)
 168 {
 169   GstBuffer *buf;
 170   GValue array = { 0 };
 171   GValue value = { 0 };
 172   GstStructure *structure;
 173   guint size1 = strlen (data1);
 174   guint size2 = strlen (data2);
 175
 176   fail_if (hbuf1 == NULL);
 177   fail_if (hbuf2 == NULL);
 178   fail_if (caps == NULL);
 179
 180   /* create caps with streamheader, set the caps, and push the IN_CAPS
 181    * buffers */
 182   *hbuf1 = gst_buffer_new_and_alloc (size1);
 183   GST_BUFFER_FLAG_SET (*hbuf1, GST_BUFFER_FLAG_IN_CAPS);
 184   gst_buffer_fill (*hbuf1, 0, data1, size1);
 185   *hbuf2 = gst_buffer_new_and_alloc (size2);
 186   GST_BUFFER_FLAG_SET (*hbuf2, GST_BUFFER_FLAG_IN_CAPS);
 187   gst_buffer_fill (*hbuf2, 0, data2, size2);
 188
 189   g_value_init (&array, GST_TYPE_ARRAY);
 190
 191   g_value_init (&value, GST_TYPE_BUFFER);
 192   /* we take a copy, set it on the array (which refs it), then unref our copy */
 193   buf = gst_buffer_copy (*hbuf1);
 194   gst_value_set_buffer (&value, buf);
 195   ASSERT_BUFFER_REFCOUNT (buf, "copied buffer", 2);
 196   gst_buffer_unref (buf);
 197   gst_value_array_append_value (&array, &value);
 198   g_value_unset (&value);
 199
 200   g_value_init (&value, GST_TYPE_BUFFER);
 201   buf = gst_buffer_copy (*hbuf2);
 202   gst_value_set_buffer (&value, buf);
 203   ASSERT_BUFFER_REFCOUNT (buf, "copied buffer", 2);
 204   gst_buffer_unref (buf);
 205   gst_value_array_append_value (&array, &value);
 206   g_value_unset (&value);
 207
 208   *caps = gst_caps_from_string ("application/x-gst-check");
 209   structure = gst_caps_get_structure (*caps, 0);
 210
 211   gst_structure_set_value (structure, "streamheader", &array);
 212   g_value_unset (&array);
 213   ASSERT_CAPS_REFCOUNT (*caps, "streamheader caps", 1);
 214
 215   /* we want to keep them around for the tests */
 216   gst_buffer_ref (*hbuf1);
 217   gst_buffer_ref (*hbuf2);
 218
 219   GST_DEBUG ("created streamheader caps %p %" GST_PTR_FORMAT, *caps, *caps);
 220 }
 221
 222
 223 /* this test:
 224  * - adds a first client
 225  * - sets streamheader caps on the pad
 226  * - pushes the IN_CAPS buffers
 227  * - pushes a buffer
 228  * - verifies that the client received all the data correctly, and did not
 229  *   get multiple copies of the streamheader
 230  * - adds a second client
 231  * - verifies that this second client receives the streamheader caps too, plus
 232  * - the new buffer
 233  */
 234 GST_START_TEST (test_streamheader)
 235 {
 236   GstElement *sink;
 237   GstBuffer *hbuf1, *hbuf2, *buf;
 238   GstCaps *caps;
 239   int pfd1[2], pfd2[2];
 240
 241   sink = setup_multifdsink ();
 242
 243   fail_if (pipe (pfd1) == -1);
 244   fail_if (pipe (pfd2) == -1);
 245
 246   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 247
 248   /* add the first client */
 249   g_signal_emit_by_name (sink, "add", pfd1[1]);
 250
 251   /* create caps with streamheader, set the caps, and push the IN_CAPS
 252    * buffers */
 253   gst_multifdsink_create_streamheader ("babe", "deadbeef", &hbuf1, &hbuf2,
 254       &caps);
 255   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 256   fail_unless (gst_pad_set_caps (mysrcpad, caps));
 257   /* one is ours, two from set_caps */
 258   ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
 259
 260   fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
 261   fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
 262
 263   //FIXME:
 264   //fail_if_can_read ("first client", pfd1[0]);
 265
 266   /* push a non-IN_CAPS buffer, this should trigger the client receiving the
 267    * first three buffers */
 268   buf = gst_buffer_new_and_alloc (4);
 269   gst_buffer_fill (buf, 0, "f00d", 4);
 270   gst_pad_push (mysrcpad, buf);
 271
 272   fail_unless_read ("first client", pfd1[0], 4, "babe");
 273   fail_unless_read ("first client", pfd1[0], 8, "deadbeef");
 274   fail_unless_read ("first client", pfd1[0], 4, "f00d");
 275   wait_bytes_served (sink, 16);
 276
 277   /* now add the second client */
 278   g_signal_emit_by_name (sink, "add", pfd2[1]);
 279   //FIXME:
 280   //fail_if_can_read ("second client", pfd2[0]);
 281
 282   /* now push another buffer, which will trigger streamheader for second
 283    * client */
 284   buf = gst_buffer_new_and_alloc (4);
 285   gst_buffer_fill (buf, 0, "deaf", 4);
 286   gst_pad_push (mysrcpad, buf);
 287
 288   fail_unless_read ("first client", pfd1[0], 4, "deaf");
 289
 290   fail_unless_read ("second client", pfd2[0], 4, "babe");
 291   fail_unless_read ("second client", pfd2[0], 8, "deadbeef");
 292   /* we missed the f00d buffer */
 293   fail_unless_read ("second client", pfd2[0], 4, "deaf");
 294   wait_bytes_served (sink, 36);
 295
 296   GST_DEBUG ("cleaning up multifdsink");
 297
 298   g_signal_emit_by_name (sink, "remove", pfd1[1]);
 299   g_signal_emit_by_name (sink, "remove", pfd2[1]);
 300
 301   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 302   cleanup_multifdsink (sink);
 303
 304   ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 1);
 305   ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 1);
 306   gst_buffer_unref (hbuf1);
 307   gst_buffer_unref (hbuf2);
 308
 309   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 310   gst_caps_unref (caps);
 311 }
 312
 313 GST_END_TEST;
 314
 315 /* this tests changing of streamheaders
 316  * - set streamheader caps on the pad
 317  * - pushes the IN_CAPS buffers
 318  * - pushes a buffer
 319  * - add a first client
 320  * - verifies that this first client receives the first streamheader caps,
 321  *   plus a new buffer
 322  * - change streamheader caps
 323  * - verify that the first client receives the new streamheader buffers as well
 324  */
 325 GST_START_TEST (test_change_streamheader)
 326 {
 327   GstElement *sink;
 328   GstBuffer *hbuf1, *hbuf2, *buf;
 329   GstCaps *caps;
 330   int pfd1[2], pfd2[2];
 331
 332   sink = setup_multifdsink ();
 333
 334   fail_if (pipe (pfd1) == -1);
 335   fail_if (pipe (pfd2) == -1);
 336
 337   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 338
 339   /* create caps with streamheader, set the caps, and push the IN_CAPS
 340    * buffers */
 341   gst_multifdsink_create_streamheader ("first", "header", &hbuf1, &hbuf2,
 342       &caps);
 343   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 344   fail_unless (gst_pad_set_caps (mysrcpad, caps));
 345   /* one is ours, two from set_caps */
 346   ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
 347
 348   /* one to hold for the test and one to give away */
 349   ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
 350   ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
 351
 352   fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
 353   fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
 354
 355   /* add the first client */
 356   g_signal_emit_by_name (sink, "add", pfd1[1]);
 357
 358   /* verify this hasn't triggered a write yet */
 359   /* FIXME: possibly racy, since if it would write, we may not get it
 360    * immediately ? */
 361   //fail_if_can_read ("first client, no buffer", pfd1[0]);
 362
 363   /* now push a buffer and read */
 364   buf = gst_buffer_new_and_alloc (4);
 365   gst_buffer_fill (buf, 0, "f00d", 4);
 366   gst_pad_push (mysrcpad, buf);
 367
 368   fail_unless_read ("change: first client", pfd1[0], 5, "first");
 369   fail_unless_read ("change: first client", pfd1[0], 6, "header");
 370   fail_unless_read ("change: first client", pfd1[0], 4, "f00d");
 371   //wait_bytes_served (sink, 16);
 372
 373   /* now add the second client */
 374   g_signal_emit_by_name (sink, "add", pfd2[1]);
 375   //fail_if_can_read ("second client, no buffer", pfd2[0]);
 376
 377   /* change the streamheader */
 378
 379   /* before we change, multifdsink still has a list of the old streamheaders */
 380   ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
 381   ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
 382   gst_buffer_unref (hbuf1);
 383   gst_buffer_unref (hbuf2);
 384
 385   /* drop our ref to the previous caps */
 386   gst_caps_unref (caps);
 387
 388   gst_multifdsink_create_streamheader ("second", "header", &hbuf1, &hbuf2,
 389       &caps);
 390   fail_unless (gst_pad_set_caps (mysrcpad, caps));
 391   /* one to hold for the test and one to give away */
 392   ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
 393   ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
 394
 395   fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
 396   fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
 397
 398   /* verify neither client has new data available to read */
 399   //fail_if_can_read ("first client, changed streamheader", pfd1[0]);
 400   //fail_if_can_read ("second client, changed streamheader", pfd2[0]);
 401
 402   /* now push another buffer, which will trigger streamheader for second
 403    * client, but should also send new streamheaders to first client */
 404   buf = gst_buffer_new_and_alloc (8);
 405   gst_buffer_fill (buf, 0, "deadbabe", 8);
 406   gst_pad_push (mysrcpad, buf);
 407
 408   fail_unless_read ("first client", pfd1[0], 6, "second");
 409   fail_unless_read ("first client", pfd1[0], 6, "header");
 410   fail_unless_read ("first client", pfd1[0], 8, "deadbabe");
 411
 412   /* new streamheader data */
 413   fail_unless_read ("second client", pfd2[0], 6, "second");
 414   fail_unless_read ("second client", pfd2[0], 6, "header");
 415   /* we missed the f00d buffer */
 416   fail_unless_read ("second client", pfd2[0], 8, "deadbabe");
 417   //wait_bytes_served (sink, 36);
 418
 419   GST_DEBUG ("cleaning up multifdsink");
 420   g_signal_emit_by_name (sink, "remove", pfd1[1]);
 421   g_signal_emit_by_name (sink, "remove", pfd2[1]);
 422   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 423
 424   /* setting to NULL should have cleared the streamheader */
 425   ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 1);
 426   ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 1);
 427   gst_buffer_unref (hbuf1);
 428   gst_buffer_unref (hbuf2);
 429   cleanup_multifdsink (sink);
 430
 431   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 432   gst_caps_unref (caps);
 433 }
 434
 435 GST_END_TEST;
 436
 437 static GstBuffer *
 438 gst_new_buffer (int i)
 439 {
 440   GstMapInfo info;
 441   gchar *data;
 442
 443   GstBuffer *buffer = gst_buffer_new_and_alloc (16);
 444
 445   /* copy some id */
 446   g_assert (gst_buffer_map (buffer, &info, GST_MAP_WRITE));
 447   data = (gchar *) info.data;
 448   g_snprintf (data, 16, "deadbee%08x", i);
 449   gst_buffer_unmap (buffer, &info);
 450
 451   return buffer;
 452 }
 453
 454
 455 /* keep 100 bytes and burst 80 bytes to clients */
 456 GST_START_TEST (test_burst_client_bytes)
 457 {
 458   GstElement *sink;
 459   GstCaps *caps;
 460   int pfd1[2];
 461   int pfd2[2];
 462   int pfd3[2];
 463   gint i;
 464   guint buffers_queued;
 465
 466   sink = setup_multifdsink ();
 467   /* make sure we keep at least 100 bytes at all times */
 468   g_object_set (sink, "bytes-min", 100, NULL);
 469   g_object_set (sink, "sync-method", 3, NULL);  /* 3 = burst */
 470   g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
 471   g_object_set (sink, "burst-value", (guint64) 80, NULL);
 472
 473   fail_if (pipe (pfd1) == -1);
 474   fail_if (pipe (pfd2) == -1);
 475   fail_if (pipe (pfd3) == -1);
 476
 477   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 478
 479   caps = gst_caps_from_string ("application/x-gst-check");
 480   gst_pad_set_caps (mysrcpad, caps);
 481   GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
 482
 483   /* push buffers in, 9 * 16 bytes = 144 bytes */
 484   for (i = 0; i < 9; i++) {
 485     GstBuffer *buffer = gst_new_buffer (i);
 486
 487     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 488   }
 489
 490   /* check that at least 7 buffers (112 bytes) are in the queue */
 491   g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
 492   fail_if (buffers_queued != 7);
 493
 494   /* now add the clients */
 495   g_signal_emit_by_name (sink, "add", pfd1[1]);
 496   g_signal_emit_by_name (sink, "add_full", pfd2[1], GST_SYNC_METHOD_BURST,
 497       GST_FORMAT_BYTES, (guint64) 50, GST_FORMAT_BYTES, (guint64) 200);
 498   g_signal_emit_by_name (sink, "add_full", pfd3[1], GST_SYNC_METHOD_BURST,
 499       GST_FORMAT_BYTES, (guint64) 50, GST_FORMAT_BYTES, (guint64) 50);
 500
 501   /* push last buffer to make client fds ready for reading */
 502   for (i = 9; i < 10; i++) {
 503     GstBuffer *buffer = gst_new_buffer (i);
 504
 505     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 506   }
 507
 508   /* now we should only read the last 5 buffers (5 * 16 = 80 bytes) */
 509   GST_DEBUG ("Reading from client 1");
 510   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000005");
 511   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000006");
 512   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000007");
 513   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000008");
 514   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000009");
 515
 516   /* second client only bursts 50 bytes = 4 buffers (we get 4 buffers since
 517    * the max alows it) */
 518   GST_DEBUG ("Reading from client 2");
 519   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000006");
 520   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000007");
 521   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000008");
 522   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000009");
 523
 524   /* third client only bursts 50 bytes = 4 buffers, we can't send
 525    * more than 50 bytes so we only get 3 buffers (48 bytes). */
 526   GST_DEBUG ("Reading from client 3");
 527   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000007");
 528   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000008");
 529   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000009");
 530
 531   GST_DEBUG ("cleaning up multifdsink");
 532   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 533   cleanup_multifdsink (sink);
 534
 535   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 536   gst_caps_unref (caps);
 537 }
 538
 539 GST_END_TEST;
 540
 541 /* keep 100 bytes and burst 80 bytes to clients */
 542 GST_START_TEST (test_burst_client_bytes_keyframe)
 543 {
 544   GstElement *sink;
 545   GstCaps *caps;
 546   int pfd1[2];
 547   int pfd2[2];
 548   int pfd3[2];
 549   gint i;
 550   guint buffers_queued;
 551
 552   sink = setup_multifdsink ();
 553   /* make sure we keep at least 100 bytes at all times */
 554   g_object_set (sink, "bytes-min", 100, NULL);
 555   g_object_set (sink, "sync-method", 4, NULL);  /* 4 = burst_keyframe */
 556   g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
 557   g_object_set (sink, "burst-value", (guint64) 80, NULL);
 558
 559   fail_if (pipe (pfd1) == -1);
 560   fail_if (pipe (pfd2) == -1);
 561   fail_if (pipe (pfd3) == -1);
 562
 563   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 564
 565   caps = gst_caps_from_string ("application/x-gst-check");
 566   gst_pad_set_caps (mysrcpad, caps);
 567   GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
 568
 569   /* push buffers in, 9 * 16 bytes = 144 bytes */
 570   for (i = 0; i < 9; i++) {
 571     GstBuffer *buffer = gst_new_buffer (i);
 572
 573     /* mark most buffers as delta */
 574     if (i != 0 && i != 4 && i != 8)
 575       GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 576
 577     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 578   }
 579
 580   /* check that at least 7 buffers (112 bytes) are in the queue */
 581   g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
 582   fail_if (buffers_queued != 7);
 583
 584   /* now add the clients */
 585   g_signal_emit_by_name (sink, "add", pfd1[1]);
 586   g_signal_emit_by_name (sink, "add_full", pfd2[1],
 587       GST_SYNC_METHOD_BURST_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
 588       GST_FORMAT_BYTES, (guint64) 90);
 589   g_signal_emit_by_name (sink, "add_full", pfd3[1],
 590       GST_SYNC_METHOD_BURST_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
 591       GST_FORMAT_BYTES, (guint64) 50);
 592
 593   /* push last buffer to make client fds ready for reading */
 594   for (i = 9; i < 10; i++) {
 595     GstBuffer *buffer = gst_new_buffer (i);
 596
 597     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 598
 599     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 600   }
 601
 602   /* now we should only read the last 6 buffers (min 5 * 16 = 80 bytes),
 603    * keyframe at buffer 4 */
 604   GST_DEBUG ("Reading from client 1");
 605   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000004");
 606   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000005");
 607   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000006");
 608   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000007");
 609   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000008");
 610   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000009");
 611
 612   /* second client only bursts 50 bytes = 4 buffers, there is
 613    * no keyframe above min and below max, so get one below min */
 614   GST_DEBUG ("Reading from client 2");
 615   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000008");
 616   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000009");
 617
 618   /* third client only bursts 50 bytes = 4 buffers, we can't send
 619    * more than 50 bytes so we only get 2 buffers (32 bytes). */
 620   GST_DEBUG ("Reading from client 3");
 621   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000008");
 622   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000009");
 623
 624   GST_DEBUG ("cleaning up multifdsink");
 625   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 626   cleanup_multifdsink (sink);
 627
 628   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 629   gst_caps_unref (caps);
 630 }
 631
 632 GST_END_TEST;
 633
 634 /* keep 100 bytes and burst 80 bytes to clients */
 635 GST_START_TEST (test_burst_client_bytes_with_keyframe)
 636 {
 637   GstElement *sink;
 638   GstCaps *caps;
 639   int pfd1[2];
 640   int pfd2[2];
 641   int pfd3[2];
 642   gint i;
 643   guint buffers_queued;
 644
 645   sink = setup_multifdsink ();
 646   /* make sure we keep at least 100 bytes at all times */
 647   g_object_set (sink, "bytes-min", 100, NULL);
 648   g_object_set (sink, "sync-method", 5, NULL);  /* 5 = burst_with_keyframe */
 649   g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
 650   g_object_set (sink, "burst-value", (guint64) 80, NULL);
 651
 652   fail_if (pipe (pfd1) == -1);
 653   fail_if (pipe (pfd2) == -1);
 654   fail_if (pipe (pfd3) == -1);
 655
 656   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 657
 658   caps = gst_caps_from_string ("application/x-gst-check");
 659   gst_pad_set_caps (mysrcpad, caps);
 660   GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
 661
 662   /* push buffers in, 9 * 16 bytes = 144 bytes */
 663   for (i = 0; i < 9; i++) {
 664     GstBuffer *buffer = gst_new_buffer (i);
 665
 666     /* mark most buffers as delta */
 667     if (i != 0 && i != 4 && i != 8)
 668       GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 669
 670     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 671   }
 672
 673   /* check that at least 7 buffers (112 bytes) are in the queue */
 674   g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
 675   fail_if (buffers_queued != 7);
 676
 677   /* now add the clients */
 678   g_signal_emit_by_name (sink, "add", pfd1[1]);
 679   g_signal_emit_by_name (sink, "add_full", pfd2[1],
 680       GST_SYNC_METHOD_BURST_WITH_KEYFRAME, GST_FORMAT_BYTES,
 681       (guint64) 50, GST_FORMAT_BYTES, (guint64) 90);
 682   g_signal_emit_by_name (sink, "add_full", pfd3[1],
 683       GST_SYNC_METHOD_BURST_WITH_KEYFRAME, GST_FORMAT_BYTES,
 684       (guint64) 50, GST_FORMAT_BYTES, (guint64) 50);
 685
 686   /* push last buffer to make client fds ready for reading */
 687   for (i = 9; i < 10; i++) {
 688     GstBuffer *buffer = gst_new_buffer (i);
 689
 690     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 691
 692     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 693   }
 694
 695   /* now we should only read the last 6 buffers (min 5 * 16 = 80 bytes),
 696    * keyframe at buffer 4 */
 697   GST_DEBUG ("Reading from client 1");
 698   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000004");
 699   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000005");
 700   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000006");
 701   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000007");
 702   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000008");
 703   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000009");
 704
 705   /* second client only bursts 50 bytes = 4 buffers, there is
 706    * no keyframe above min and below max, so send min */
 707   GST_DEBUG ("Reading from client 2");
 708   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000006");
 709   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000007");
 710   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000008");
 711   fail_unless_read ("client 2", pfd2[0], 16, "deadbee00000009");
 712
 713   /* third client only bursts 50 bytes = 4 buffers, we can't send
 714    * more than 50 bytes so we only get 3 buffers (48 bytes). */
 715   GST_DEBUG ("Reading from client 3");
 716   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000007");
 717   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000008");
 718   fail_unless_read ("client 3", pfd3[0], 16, "deadbee00000009");
 719
 720   GST_DEBUG ("cleaning up multifdsink");
 721   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 722   cleanup_multifdsink (sink);
 723
 724   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 725   gst_caps_unref (caps);
 726 }
 727
 728 GST_END_TEST;
 729
 730 /* Check that we can get data when multifdsink is configured in next-keyframe
 731  * mode */
 732 GST_START_TEST (test_client_next_keyframe)
 733 {
 734   GstElement *sink;
 735   GstCaps *caps;
 736   int pfd1[2];
 737   gint i;
 738
 739   sink = setup_multifdsink ();
 740   g_object_set (sink, "sync-method", 1, NULL);  /* 1 = next-keyframe */
 741
 742   fail_if (pipe (pfd1) == -1);
 743
 744   ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
 745
 746   caps = gst_caps_from_string ("application/x-gst-check");
 747   gst_pad_set_caps (mysrcpad, caps);
 748   GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
 749
 750   /* now add our client */
 751   g_signal_emit_by_name (sink, "add", pfd1[1]);
 752
 753   /* push buffers in: keyframe, then non-keyframe */
 754   for (i = 0; i < 2; i++) {
 755     GstBuffer *buffer = gst_new_buffer (i);
 756     if (i > 0)
 757       GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
 758
 759     fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
 760   }
 761
 762   /* now we should be able to read some data */
 763   GST_DEBUG ("Reading from client 1");
 764   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000000");
 765   fail_unless_read ("client 1", pfd1[0], 16, "deadbee00000001");
 766
 767   GST_DEBUG ("cleaning up multifdsink");
 768   ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
 769   cleanup_multifdsink (sink);
 770
 771   ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
 772   gst_caps_unref (caps);
 773 }
 774
 775 GST_END_TEST;
 776
 777 /* FIXME: add test simulating chained oggs where:
 778  * sync-method is burst-on-connect
 779  * (when multifdsink actually does burst-on-connect based on byte size, not
 780    "last keyframe" which any frame for audio :))
 781  * an old client still needs to read from before the new streamheaders
 782  * a new client gets the new streamheaders
 783  */
 784 static Suite *
 785 multifdsink_suite (void)
 786 {
 787   Suite *s = suite_create ("multifdsink");
 788   TCase *tc_chain = tcase_create ("general");
 789
 790   suite_add_tcase (s, tc_chain);
 791   tcase_add_test (tc_chain, test_no_clients);
 792   tcase_add_test (tc_chain, test_add_client);
 793   tcase_add_test (tc_chain, test_streamheader);
 794   tcase_add_test (tc_chain, test_change_streamheader);
 795   tcase_add_test (tc_chain, test_burst_client_bytes);
 796   tcase_add_test (tc_chain, test_burst_client_bytes_keyframe);
 797   tcase_add_test (tc_chain, test_burst_client_bytes_with_keyframe);
 798   tcase_add_test (tc_chain, test_client_next_keyframe);
 799
 800   return s;
 801 }
 802
 803 GST_CHECK_MAIN (multifdsink);