3 * Copyright (C) 2006 Thomas Vander Stichele <thomas at apestaart dot org>
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.
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.
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.
22 #include <sys/ioctl.h>
23 #include <sys/socket.h>
24 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
25 #include <sys/filio.h>
29 #include <gst/check/gstcheck.h>
31 #include "gst/tcp/gstmultisocketsink.h"
33 static GstPad *mysrcpad;
35 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
38 GST_STATIC_CAPS ("application/x-gst-check")
42 setup_multisocketsink (void)
44 GstElement *multisocketsink;
46 GST_DEBUG ("setup_multisocketsink");
47 multisocketsink = gst_check_setup_element ("multisocketsink");
48 mysrcpad = gst_check_setup_src_pad (multisocketsink, &srctemplate);
49 GST_PAD_UNSET_FLUSHING (mysrcpad);
51 return multisocketsink;
55 cleanup_multisocketsink (GstElement * multisocketsink)
57 GST_DEBUG ("cleanup_multisocketsink");
59 gst_check_teardown_src_pad (multisocketsink);
60 gst_check_teardown_element (multisocketsink);
64 wait_bytes_served (GstElement * sink, guint64 bytes)
66 guint64 bytes_served = 0;
68 while (bytes_served != bytes) {
69 g_object_get (sink, "bytes-served", &bytes_served, NULL);
73 /* FIXME: possibly racy, since if it would write, we may not get it
75 #define fail_if_can_read(msg,fd) \
79 fail_if (ioctl (fd, FIONREAD, &avail) < 0, "%s: could not ioctl", msg); \
80 fail_if (avail > 0, "%s: has bytes available to read"); \
84 GST_START_TEST (test_no_clients)
90 sink = setup_multisocketsink ();
92 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
94 caps = gst_caps_from_string ("application/x-gst-check");
95 buffer = gst_buffer_new_and_alloc (4);
96 gst_pad_set_caps (mysrcpad, caps);
97 gst_caps_unref (caps);
98 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
100 GST_DEBUG ("cleaning up multisocketsink");
101 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
102 cleanup_multisocketsink (sink);
108 setup_handles (GSocket ** sinkhandle, GSocket ** srchandle)
110 GError *error = NULL;
114 // g_assert (*sinkhandle);
115 // g_assert (*srchandle);
117 fail_if (socketpair (PF_UNIX, SOCK_STREAM, 0, sv));
119 *sinkhandle = g_socket_new_from_fd (sv[1], &error);
121 fail_if (*sinkhandle == NULL);
122 *srchandle = g_socket_new_from_fd (sv[0], &error);
124 fail_if (*srchandle == NULL);
130 read_handle (GSocket * srchandle, void *buf, size_t count)
134 ret = g_socket_receive (srchandle, buf, count, NULL, NULL);
139 #define fail_unless_read(msg,handle,size,ref) \
141 char data[size + 1]; \
144 GST_DEBUG ("%s: reading %d bytes", msg, size); \
145 nbytes = read_handle (handle, data, size); \
147 GST_DEBUG ("%s: read %d bytes", msg, nbytes); \
148 fail_if (nbytes < size); \
149 fail_unless (memcmp (data, ref, size) == 0, \
150 "data read '%s' differs from '%s'", data, ref); \
153 #define fail_unless_num_handles(sink,num) \
156 g_object_get (sink, "num-sockets", &handles, NULL); \
157 fail_unless (handles == num, \
158 "sink has %d handles instead of expected %d", handles, num); \
161 GST_START_TEST (test_add_client)
167 GSocket *sinksocket, *srcsocket;
169 sink = setup_multisocketsink ();
170 fail_unless (setup_handles (&sinksocket, &srcsocket));
173 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
176 g_signal_emit_by_name (sink, "add", sinksocket);
178 caps = gst_caps_from_string ("application/x-gst-check");
179 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
180 GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
181 buffer = gst_buffer_new_and_alloc (4);
182 gst_pad_set_caps (mysrcpad, caps);
183 ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
184 gst_buffer_fill (buffer, 0, "dead", 4);
185 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
187 GST_DEBUG ("reading");
188 fail_if (read_handle (srcsocket, data, 4) < 4);
189 fail_unless (strncmp (data, "dead", 4) == 0);
190 wait_bytes_served (sink, 4);
192 GST_DEBUG ("cleaning up multisocketsink");
193 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
194 cleanup_multisocketsink (sink);
196 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
197 gst_caps_unref (caps);
202 /* from the given two data buffers, create two streamheader buffers and
203 * some caps that match it, and store them in the given pointers
204 * returns one ref to each of the buffers and the caps */
206 gst_multisocketsink_create_streamheader (const gchar * data1,
207 const gchar * data2, GstBuffer ** hbuf1, GstBuffer ** hbuf2,
211 GValue array = { 0 };
212 GValue value = { 0 };
213 GstStructure *structure;
214 guint size1 = strlen (data1);
215 guint size2 = strlen (data2);
217 fail_if (hbuf1 == NULL);
218 fail_if (hbuf2 == NULL);
219 fail_if (caps == NULL);
221 /* create caps with streamheader, set the caps, and push the IN_CAPS
223 *hbuf1 = gst_buffer_new_and_alloc (size1);
224 GST_BUFFER_FLAG_SET (*hbuf1, GST_BUFFER_FLAG_IN_CAPS);
225 gst_buffer_fill (*hbuf1, 0, data1, size1);
226 *hbuf2 = gst_buffer_new_and_alloc (size2);
227 GST_BUFFER_FLAG_SET (*hbuf2, GST_BUFFER_FLAG_IN_CAPS);
228 gst_buffer_fill (*hbuf2, 0, data2, size2);
230 g_value_init (&array, GST_TYPE_ARRAY);
232 g_value_init (&value, GST_TYPE_BUFFER);
233 /* we take a copy, set it on the array (which refs it), then unref our copy */
234 buf = gst_buffer_copy (*hbuf1);
235 gst_value_set_buffer (&value, buf);
236 ASSERT_BUFFER_REFCOUNT (buf, "copied buffer", 2);
237 gst_buffer_unref (buf);
238 gst_value_array_append_value (&array, &value);
239 g_value_unset (&value);
241 g_value_init (&value, GST_TYPE_BUFFER);
242 buf = gst_buffer_copy (*hbuf2);
243 gst_value_set_buffer (&value, buf);
244 ASSERT_BUFFER_REFCOUNT (buf, "copied buffer", 2);
245 gst_buffer_unref (buf);
246 gst_value_array_append_value (&array, &value);
247 g_value_unset (&value);
249 *caps = gst_caps_from_string ("application/x-gst-check");
250 structure = gst_caps_get_structure (*caps, 0);
252 gst_structure_set_value (structure, "streamheader", &array);
253 g_value_unset (&array);
254 ASSERT_CAPS_REFCOUNT (*caps, "streamheader caps", 1);
256 /* we want to keep them around for the tests */
257 gst_buffer_ref (*hbuf1);
258 gst_buffer_ref (*hbuf2);
260 GST_DEBUG ("created streamheader caps %p %" GST_PTR_FORMAT, *caps, *caps);
265 * - adds a first client
266 * - sets streamheader caps on the pad
267 * - pushes the IN_CAPS buffers
269 * - verifies that the client received all the data correctly, and did not
270 * get multiple copies of the streamheader
271 * - adds a second client
272 * - verifies that this second client receives the streamheader caps too, plus
275 GST_START_TEST (test_streamheader)
278 GstBuffer *hbuf1, *hbuf2, *buf;
282 sink = setup_multisocketsink ();
284 fail_unless (setup_handles (&socket[0], &socket[1]));
285 fail_unless (setup_handles (&socket[2], &socket[3]));
287 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
289 /* add the first client */
290 fail_unless_num_handles (sink, 0);
291 g_signal_emit_by_name (sink, "add", socket[0]);
292 fail_unless_num_handles (sink, 1);
294 /* create caps with streamheader, set the caps, and push the IN_CAPS
296 gst_multisocketsink_create_streamheader ("babe", "deadbeef", &hbuf1, &hbuf2,
298 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
299 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
300 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
301 fail_unless (gst_pad_set_caps (mysrcpad, caps));
302 /* one is ours, two from set_caps */
303 ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
305 fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
306 fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
307 // FIXME: we can't assert on the refcount because giving away the ref
308 // doesn't mean the refcount decreases
309 // ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 1);
310 // ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 1);
313 //fail_if_can_read ("first client", socket[1]);
315 /* push a non-IN_CAPS buffer, this should trigger the client receiving the
316 * first three buffers */
317 buf = gst_buffer_new_and_alloc (4);
318 gst_buffer_fill (buf, 0, "f00d", 4);
319 gst_pad_push (mysrcpad, buf);
321 fail_unless_read ("first client", socket[1], 4, "babe");
322 fail_unless_read ("first client", socket[1], 8, "deadbeef");
323 fail_unless_read ("first client", socket[1], 4, "f00d");
324 wait_bytes_served (sink, 16);
326 /* now add the second client */
327 g_signal_emit_by_name (sink, "add", socket[2]);
328 fail_unless_num_handles (sink, 2);
330 //fail_if_can_read ("second client", socket[3]);
332 /* now push another buffer, which will trigger streamheader for second
334 buf = gst_buffer_new_and_alloc (4);
335 gst_buffer_fill (buf, 0, "deaf", 4);
336 gst_pad_push (mysrcpad, buf);
338 fail_unless_read ("first client", socket[1], 4, "deaf");
340 fail_unless_read ("second client", socket[3], 4, "babe");
341 fail_unless_read ("second client", socket[3], 8, "deadbeef");
342 /* we missed the f00d buffer */
343 fail_unless_read ("second client", socket[3], 4, "deaf");
344 wait_bytes_served (sink, 36);
346 GST_DEBUG ("cleaning up multisocketsink");
348 fail_unless_num_handles (sink, 2);
349 g_signal_emit_by_name (sink, "remove", socket[0]);
350 fail_unless_num_handles (sink, 1);
351 g_signal_emit_by_name (sink, "remove", socket[2]);
352 fail_unless_num_handles (sink, 0);
354 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
355 cleanup_multisocketsink (sink);
357 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 1);
358 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 1);
359 gst_buffer_unref (hbuf1);
360 gst_buffer_unref (hbuf2);
362 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
363 gst_caps_unref (caps);
368 /* this tests changing of streamheaders
369 * - set streamheader caps on the pad
370 * - pushes the IN_CAPS buffers
372 * - add a first client
373 * - verifies that this first client receives the first streamheader caps,
375 * - change streamheader caps
376 * - verify that the first client receives the new streamheader buffers as well
378 GST_START_TEST (test_change_streamheader)
381 GstBuffer *hbuf1, *hbuf2, *buf;
385 sink = setup_multisocketsink ();
387 fail_unless (setup_handles (&socket[0], &socket[1]));
388 fail_unless (setup_handles (&socket[2], &socket[3]));
390 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
392 /* create caps with streamheader, set the caps, and push the IN_CAPS
394 gst_multisocketsink_create_streamheader ("first", "header", &hbuf1, &hbuf2,
396 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
397 fail_unless (gst_pad_set_caps (mysrcpad, caps));
398 /* one is ours, two from set_caps */
399 ASSERT_CAPS_REFCOUNT (caps, "caps", 3);
401 /* one to hold for the test and one to give away */
402 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
403 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
405 fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
406 fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
408 /* add the first client */
409 g_signal_emit_by_name (sink, "add", socket[0]);
411 /* verify this hasn't triggered a write yet */
412 /* FIXME: possibly racy, since if it would write, we may not get it
414 //fail_if_can_read ("first client, no buffer", socket[1]);
416 /* now push a buffer and read */
417 buf = gst_buffer_new_and_alloc (4);
418 gst_buffer_fill (buf, 0, "f00d", 4);
419 gst_pad_push (mysrcpad, buf);
421 fail_unless_read ("change: first client", socket[1], 5, "first");
422 fail_unless_read ("change: first client", socket[1], 6, "header");
423 fail_unless_read ("change: first client", socket[1], 4, "f00d");
424 //wait_bytes_served (sink, 16);
426 /* now add the second client */
427 g_signal_emit_by_name (sink, "add", socket[2]);
428 //fail_if_can_read ("second client, no buffer", socket[3]);
430 /* change the streamheader */
432 /* before we change, multisocketsink still has a list of the old streamheaders */
433 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
434 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
435 gst_buffer_unref (hbuf1);
436 gst_buffer_unref (hbuf2);
438 /* drop our ref to the previous caps */
439 gst_caps_unref (caps);
441 gst_multisocketsink_create_streamheader ("second", "header", &hbuf1, &hbuf2,
443 fail_unless (gst_pad_set_caps (mysrcpad, caps));
444 /* one to hold for the test and one to give away */
445 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 2);
446 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 2);
448 fail_unless (gst_pad_push (mysrcpad, hbuf1) == GST_FLOW_OK);
449 fail_unless (gst_pad_push (mysrcpad, hbuf2) == GST_FLOW_OK);
451 /* verify neither client has new data available to read */
452 //fail_if_can_read ("first client, changed streamheader", socket[1]);
453 //fail_if_can_read ("second client, changed streamheader", socket[3]);
455 /* now push another buffer, which will trigger streamheader for second
456 * client, but should also send new streamheaders to first client */
457 buf = gst_buffer_new_and_alloc (8);
458 gst_buffer_fill (buf, 0, "deadbabe", 8);
459 gst_pad_push (mysrcpad, buf);
461 fail_unless_read ("first client", socket[1], 6, "second");
462 fail_unless_read ("first client", socket[1], 6, "header");
463 fail_unless_read ("first client", socket[1], 8, "deadbabe");
465 /* new streamheader data */
466 fail_unless_read ("second client", socket[3], 6, "second");
467 fail_unless_read ("second client", socket[3], 6, "header");
468 /* we missed the f00d buffer */
469 fail_unless_read ("second client", socket[3], 8, "deadbabe");
470 //wait_bytes_served (sink, 36);
472 GST_DEBUG ("cleaning up multisocketsink");
473 g_signal_emit_by_name (sink, "remove", socket[0]);
474 g_signal_emit_by_name (sink, "remove", socket[2]);
475 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
477 /* setting to NULL should have cleared the streamheader */
478 ASSERT_BUFFER_REFCOUNT (hbuf1, "hbuf1", 1);
479 ASSERT_BUFFER_REFCOUNT (hbuf2, "hbuf2", 1);
480 gst_buffer_unref (hbuf1);
481 gst_buffer_unref (hbuf2);
482 cleanup_multisocketsink (sink);
484 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
485 gst_caps_unref (caps);
491 gst_new_buffer (int i)
496 GstBuffer *buffer = gst_buffer_new_and_alloc (16);
499 g_assert (gst_buffer_map (buffer, &info, GST_MAP_WRITE));
500 data = (gchar *) info.data;
501 g_snprintf (data, 16, "deadbee%08x", i);
502 gst_buffer_unmap (buffer, &info);
508 /* keep 100 bytes and burst 80 bytes to clients */
509 GST_START_TEST (test_burst_client_bytes)
515 guint buffers_queued;
517 sink = setup_multisocketsink ();
518 /* make sure we keep at least 100 bytes at all times */
519 g_object_set (sink, "bytes-min", 100, NULL);
520 g_object_set (sink, "sync-method", 3, NULL); /* 3 = burst */
521 g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
522 g_object_set (sink, "burst-value", (guint64) 80, NULL);
524 fail_unless (setup_handles (&socket[0], &socket[1]));
525 fail_unless (setup_handles (&socket[2], &socket[3]));
526 fail_unless (setup_handles (&socket[4], &socket[5]));
528 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
530 caps = gst_caps_from_string ("application/x-gst-check");
531 gst_pad_set_caps (mysrcpad, caps);
532 GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
534 /* push buffers in, 9 * 16 bytes = 144 bytes */
535 for (i = 0; i < 9; i++) {
536 GstBuffer *buffer = gst_new_buffer (i);
538 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
541 /* check that at least 7 buffers (112 bytes) are in the queue */
542 g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
543 fail_if (buffers_queued != 7);
545 /* now add the clients */
546 fail_unless_num_handles (sink, 0);
547 g_signal_emit_by_name (sink, "add", socket[0]);
548 fail_unless_num_handles (sink, 1);
549 g_signal_emit_by_name (sink, "add_full", socket[2], GST_SYNC_METHOD_BURST,
550 GST_FORMAT_BYTES, (guint64) 50, GST_FORMAT_BYTES, (guint64) 200);
551 g_signal_emit_by_name (sink, "add_full", socket[4], GST_SYNC_METHOD_BURST,
552 GST_FORMAT_BYTES, (guint64) 50, GST_FORMAT_BYTES, (guint64) 50);
553 fail_unless_num_handles (sink, 3);
555 /* push last buffer to make client fds ready for reading */
556 for (i = 9; i < 10; i++) {
557 GstBuffer *buffer = gst_new_buffer (i);
559 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
562 /* now we should only read the last 5 buffers (5 * 16 = 80 bytes) */
563 GST_DEBUG ("Reading from client 1");
564 fail_unless_read ("client 1", socket[1], 16, "deadbee00000005");
565 fail_unless_read ("client 1", socket[1], 16, "deadbee00000006");
566 fail_unless_read ("client 1", socket[1], 16, "deadbee00000007");
567 fail_unless_read ("client 1", socket[1], 16, "deadbee00000008");
568 fail_unless_read ("client 1", socket[1], 16, "deadbee00000009");
570 /* second client only bursts 50 bytes = 4 buffers (we get 4 buffers since
571 * the max allows it) */
572 GST_DEBUG ("Reading from client 2");
573 fail_unless_read ("client 2", socket[3], 16, "deadbee00000006");
574 fail_unless_read ("client 2", socket[3], 16, "deadbee00000007");
575 fail_unless_read ("client 2", socket[3], 16, "deadbee00000008");
576 fail_unless_read ("client 2", socket[3], 16, "deadbee00000009");
578 /* third client only bursts 50 bytes = 4 buffers, we can't send
579 * more than 50 bytes so we only get 3 buffers (48 bytes). */
580 GST_DEBUG ("Reading from client 3");
581 fail_unless_read ("client 3", socket[5], 16, "deadbee00000007");
582 fail_unless_read ("client 3", socket[5], 16, "deadbee00000008");
583 fail_unless_read ("client 3", socket[5], 16, "deadbee00000009");
585 GST_DEBUG ("cleaning up multisocketsink");
586 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
587 cleanup_multisocketsink (sink);
589 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
590 gst_caps_unref (caps);
595 /* keep 100 bytes and burst 80 bytes to clients */
596 GST_START_TEST (test_burst_client_bytes_keyframe)
602 guint buffers_queued;
604 sink = setup_multisocketsink ();
605 /* make sure we keep at least 100 bytes at all times */
606 g_object_set (sink, "bytes-min", 100, NULL);
607 g_object_set (sink, "sync-method", 4, NULL); /* 4 = burst_keyframe */
608 g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
609 g_object_set (sink, "burst-value", (guint64) 80, NULL);
611 fail_unless (setup_handles (&socket[0], &socket[1]));
612 fail_unless (setup_handles (&socket[2], &socket[3]));
613 fail_unless (setup_handles (&socket[4], &socket[5]));
615 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
617 caps = gst_caps_from_string ("application/x-gst-check");
618 GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
619 gst_pad_set_caps (mysrcpad, caps);
621 /* push buffers in, 9 * 16 bytes = 144 bytes */
622 for (i = 0; i < 9; i++) {
623 GstBuffer *buffer = gst_new_buffer (i);
625 /* mark most buffers as delta */
626 if (i != 0 && i != 4 && i != 8)
627 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
629 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
632 /* check that at least 7 buffers (112 bytes) are in the queue */
633 g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
634 fail_if (buffers_queued != 7);
636 /* now add the clients */
637 g_signal_emit_by_name (sink, "add", socket[0]);
638 g_signal_emit_by_name (sink, "add_full", socket[2],
639 GST_SYNC_METHOD_BURST_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
640 GST_FORMAT_BYTES, (guint64) 90);
641 g_signal_emit_by_name (sink, "add_full", socket[4],
642 GST_SYNC_METHOD_BURST_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
643 GST_FORMAT_BYTES, (guint64) 50);
645 /* push last buffer to make client fds ready for reading */
646 for (i = 9; i < 10; i++) {
647 GstBuffer *buffer = gst_new_buffer (i);
649 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
651 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
654 /* now we should only read the last 6 buffers (min 5 * 16 = 80 bytes),
655 * keyframe at buffer 4 */
656 GST_DEBUG ("Reading from client 1");
657 fail_unless_read ("client 1", socket[1], 16, "deadbee00000004");
658 fail_unless_read ("client 1", socket[1], 16, "deadbee00000005");
659 fail_unless_read ("client 1", socket[1], 16, "deadbee00000006");
660 fail_unless_read ("client 1", socket[1], 16, "deadbee00000007");
661 fail_unless_read ("client 1", socket[1], 16, "deadbee00000008");
662 fail_unless_read ("client 1", socket[1], 16, "deadbee00000009");
664 /* second client only bursts 50 bytes = 4 buffers, there is
665 * no keyframe above min and below max, so get one below min */
666 GST_DEBUG ("Reading from client 2");
667 fail_unless_read ("client 2", socket[3], 16, "deadbee00000008");
668 fail_unless_read ("client 2", socket[3], 16, "deadbee00000009");
670 /* third client only bursts 50 bytes = 4 buffers, we can't send
671 * more than 50 bytes so we only get 2 buffers (32 bytes). */
672 GST_DEBUG ("Reading from client 3");
673 fail_unless_read ("client 3", socket[5], 16, "deadbee00000008");
674 fail_unless_read ("client 3", socket[5], 16, "deadbee00000009");
676 GST_DEBUG ("cleaning up multisocketsink");
677 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
678 cleanup_multisocketsink (sink);
680 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
681 gst_caps_unref (caps);
688 /* keep 100 bytes and burst 80 bytes to clients */
689 GST_START_TEST (test_burst_client_bytes_with_keyframe)
695 guint buffers_queued;
697 sink = setup_multisocketsink ();
699 /* make sure we keep at least 100 bytes at all times */
700 g_object_set (sink, "bytes-min", 100, NULL);
701 g_object_set (sink, "sync-method", 5, NULL); /* 5 = burst_with_keyframe */
702 g_object_set (sink, "burst-format", GST_FORMAT_BYTES, NULL);
703 g_object_set (sink, "burst-value", (guint64) 80, NULL);
705 fail_unless (setup_handles (&socket[0], &socket[1]));
706 fail_unless (setup_handles (&socket[2], &socket[3]));
707 fail_unless (setup_handles (&socket[4], &socket[5]));
709 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
711 caps = gst_caps_from_string ("application/x-gst-check");
712 gst_pad_set_caps (mysrcpad, caps);
713 GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
715 /* push buffers in, 9 * 16 bytes = 144 bytes */
716 for (i = 0; i < 9; i++) {
717 GstBuffer *buffer = gst_new_buffer (i);
719 /* mark most buffers as delta */
720 if (i != 0 && i != 4 && i != 8)
721 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
723 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
726 /* check that at least 7 buffers (112 bytes) are in the queue */
727 g_object_get (sink, "buffers-queued", &buffers_queued, NULL);
728 fail_if (buffers_queued != 7);
730 /* now add the clients */
731 g_signal_emit_by_name (sink, "add", socket[0]);
732 g_signal_emit_by_name (sink, "add_full", socket[2],
733 GST_SYNC_METHOD_BURST_WITH_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
734 GST_FORMAT_BYTES, (guint64) 90);
735 g_signal_emit_by_name (sink, "add_full", socket[4],
736 GST_SYNC_METHOD_BURST_WITH_KEYFRAME, GST_FORMAT_BYTES, (guint64) 50,
737 GST_FORMAT_BYTES, (guint64) 50);
739 /* push last buffer to make client fds ready for reading */
740 for (i = 9; i < 10; i++) {
741 GstBuffer *buffer = gst_new_buffer (i);
743 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
745 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
748 /* now we should only read the last 6 buffers (min 5 * 16 = 80 bytes),
749 * keyframe at buffer 4 */
750 GST_DEBUG ("Reading from client 1");
751 fail_unless_read ("client 1", socket[1], 16, "deadbee00000004");
752 fail_unless_read ("client 1", socket[1], 16, "deadbee00000005");
753 fail_unless_read ("client 1", socket[1], 16, "deadbee00000006");
754 fail_unless_read ("client 1", socket[1], 16, "deadbee00000007");
755 fail_unless_read ("client 1", socket[1], 16, "deadbee00000008");
756 fail_unless_read ("client 1", socket[1], 16, "deadbee00000009");
758 /* second client only bursts 50 bytes = 4 buffers, there is
759 * no keyframe above min and below max, so send min */
760 GST_DEBUG ("Reading from client 2");
761 fail_unless_read ("client 2", socket[3], 16, "deadbee00000006");
762 fail_unless_read ("client 2", socket[3], 16, "deadbee00000007");
763 fail_unless_read ("client 2", socket[3], 16, "deadbee00000008");
764 fail_unless_read ("client 2", socket[3], 16, "deadbee00000009");
766 /* third client only bursts 50 bytes = 4 buffers, we can't send
767 * more than 50 bytes so we only get 3 buffers (48 bytes). */
768 GST_DEBUG ("Reading from client 3");
769 fail_unless_read ("client 3", socket[5], 16, "deadbee00000007");
770 fail_unless_read ("client 3", socket[5], 16, "deadbee00000008");
771 fail_unless_read ("client 3", socket[5], 16, "deadbee00000009");
773 GST_DEBUG ("cleaning up multisocketsink");
774 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
775 cleanup_multisocketsink (sink);
777 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
778 gst_caps_unref (caps);
783 /* Check that we can get data when multisocketsink is configured in next-keyframe
785 GST_START_TEST (test_client_next_keyframe)
792 sink = setup_multisocketsink ();
793 g_object_set (sink, "sync-method", 1, NULL); /* 1 = next-keyframe */
795 fail_unless (setup_handles (&socket[0], &socket[1]));
797 ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
799 caps = gst_caps_from_string ("application/x-gst-check");
800 gst_pad_set_caps (mysrcpad, caps);
801 GST_DEBUG ("Created test caps %p %" GST_PTR_FORMAT, caps, caps);
803 /* now add our client */
804 g_signal_emit_by_name (sink, "add", socket[0]);
806 /* push buffers in: keyframe, then non-keyframe */
807 for (i = 0; i < 2; i++) {
808 GstBuffer *buffer = gst_new_buffer (i);
810 GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
812 fail_unless (gst_pad_push (mysrcpad, buffer) == GST_FLOW_OK);
815 /* now we should be able to read some data */
816 GST_DEBUG ("Reading from client 1");
817 fail_unless_read ("client 1", socket[1], 16, "deadbee00000000");
818 fail_unless_read ("client 1", socket[1], 16, "deadbee00000001");
820 GST_DEBUG ("cleaning up multisocketsink");
821 ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
822 cleanup_multisocketsink (sink);
824 ASSERT_CAPS_REFCOUNT (caps, "caps", 1);
825 gst_caps_unref (caps);
830 /* FIXME: add test simulating chained oggs where:
831 * sync-method is burst-on-connect
832 * (when multisocketsink actually does burst-on-connect based on byte size, not
833 "last keyframe" which any frame for audio :))
834 * an old client still needs to read from before the new streamheaders
835 * a new client gets the new streamheaders
838 multisocketsink_suite (void)
840 Suite *s = suite_create ("multisocketsink");
841 TCase *tc_chain = tcase_create ("general");
843 suite_add_tcase (s, tc_chain);
844 tcase_add_test (tc_chain, test_no_clients);
845 tcase_add_test (tc_chain, test_add_client);
846 tcase_add_test (tc_chain, test_streamheader);
847 tcase_add_test (tc_chain, test_change_streamheader);
848 tcase_add_test (tc_chain, test_burst_client_bytes);
849 tcase_add_test (tc_chain, test_burst_client_bytes_keyframe);
850 tcase_add_test (tc_chain, test_burst_client_bytes_with_keyframe);
851 tcase_add_test (tc_chain, test_client_next_keyframe);
856 GST_CHECK_MAIN (multisocketsink);