tests/check/libs/aggregator.c

   1 /*
   2  * aggregator.c - GstAggregator testsuite
   3  * Copyright (C) 2006 Alessandro Decina <alessandro.d@gmail.com>
   4  * Copyright (C) 2014 Mathieu Duponchelle <mathieu.duponchelle@oencreed.com>
   5  * Copyright (C) 2014 Thibault Saunier <tsaunier@opencreed.com>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #  include "config.h"
  25 #endif
  26
  27 #include <gst/check/gstcheck.h>
  28 #include <gst/base/gstaggregator.h>
  29
  30 /* dummy aggregator based element */
  31
  32 #define GST_TYPE_TEST_AGGREGATOR            (gst_test_aggregator_get_type ())
  33 #define GST_TEST_AGGREGATOR(obj)            (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_TEST_AGGREGATOR, GstTestAggregator))
  34 #define GST_TEST_AGGREGATOR_CLASS(klass)    (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_TEST_AGGREGATOR, GstTestAggregatorClass))
  35 #define GST_TEST_AGGREGATOR_GET_CLASS(obj)  (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_TEST_AGGREGATOR, GstTestAggregatorClass))
  36
  37 #define fail_error_message(msg)     \
  38   G_STMT_START {        \
  39     GError *error;        \
  40     gst_message_parse_error(msg, &error, NULL);       \
  41     fail_unless(FALSE, "Error Message from %s : %s",      \
  42     GST_OBJECT_NAME (GST_MESSAGE_SRC(msg)), error->message); \
  43     g_error_free (error);           \
  44   } G_STMT_END;
  45
  46 typedef struct _GstTestAggregator GstTestAggregator;
  47 typedef struct _GstTestAggregatorClass GstTestAggregatorClass;
  48
  49 static GType gst_test_aggregator_get_type (void);
  50
  51 #define BUFFER_DURATION 100000000       /* 10 frames per second */
  52
  53 struct _GstTestAggregator
  54 {
  55   GstAggregator parent;
  56
  57   guint64 timestamp;
  58 };
  59
  60 struct _GstTestAggregatorClass
  61 {
  62   GstAggregatorClass parent_class;
  63 };
  64
  65 static GstFlowReturn
  66 gst_test_aggregator_aggregate (GstAggregator * aggregator, gboolean timeout)
  67 {
  68   GstIterator *iter;
  69   gboolean all_eos = TRUE;
  70   GstTestAggregator *testagg;
  71   GstBuffer *buf;
  72
  73   gboolean done_iterating = FALSE;
  74
  75   testagg = GST_TEST_AGGREGATOR (aggregator);
  76
  77   iter = gst_element_iterate_sink_pads (GST_ELEMENT (testagg));
  78   while (!done_iterating) {
  79     GstBuffer *buffer;
  80     GValue value = { 0, };
  81     GstAggregatorPad *pad;
  82
  83     switch (gst_iterator_next (iter, &value)) {
  84       case GST_ITERATOR_OK:
  85         pad = g_value_get_object (&value);
  86
  87         if (gst_aggregator_pad_is_eos (pad) == FALSE)
  88           all_eos = FALSE;
  89         buffer = gst_aggregator_pad_steal_buffer (pad);
  90         gst_buffer_replace (&buffer, NULL);
  91
  92         g_value_reset (&value);
  93         break;
  94       case GST_ITERATOR_RESYNC:
  95         gst_iterator_resync (iter);
  96         break;
  97       case GST_ITERATOR_ERROR:
  98         GST_WARNING_OBJECT (testagg, "Sinkpads iteration error");
  99         done_iterating = TRUE;
 100         break;
 101       case GST_ITERATOR_DONE:
 102         done_iterating = TRUE;
 103         break;
 104     }
 105   }
 106   gst_iterator_free (iter);
 107
 108   if (all_eos == TRUE) {
 109     GST_INFO_OBJECT (testagg, "no data available, must be EOS");
 110     gst_pad_push_event (aggregator->srcpad, gst_event_new_eos ());
 111     return GST_FLOW_EOS;
 112   }
 113
 114   buf = gst_buffer_new ();
 115   GST_BUFFER_TIMESTAMP (buf) = testagg->timestamp;
 116   GST_BUFFER_DURATION (buf) = BUFFER_DURATION;
 117   testagg->timestamp += BUFFER_DURATION;
 118
 119   gst_aggregator_finish_buffer (aggregator, buf);
 120
 121   /* We just check finish_frame return FLOW_OK */
 122   return GST_FLOW_OK;
 123 }
 124
 125 #define gst_test_aggregator_parent_class parent_class
 126 G_DEFINE_TYPE (GstTestAggregator, gst_test_aggregator, GST_TYPE_AGGREGATOR);
 127
 128 static void
 129 gst_test_aggregator_class_init (GstTestAggregatorClass * klass)
 130 {
 131   GstElementClass *gstelement_class = (GstElementClass *) klass;
 132   GstAggregatorClass *base_aggregator_class = (GstAggregatorClass *) klass;
 133
 134   static GstStaticPadTemplate _src_template =
 135       GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC, GST_PAD_ALWAYS,
 136       GST_STATIC_CAPS_ANY);
 137
 138   static GstStaticPadTemplate _sink_template =
 139       GST_STATIC_PAD_TEMPLATE ("sink_%u", GST_PAD_SINK, GST_PAD_REQUEST,
 140       GST_STATIC_CAPS_ANY);
 141
 142   gst_element_class_add_pad_template (gstelement_class,
 143       gst_static_pad_template_get (&_src_template));
 144
 145   gst_element_class_add_pad_template (gstelement_class,
 146       gst_static_pad_template_get (&_sink_template));
 147
 148   gst_element_class_set_static_metadata (gstelement_class, "Aggregator",
 149       "Testing", "Combine N buffers", "Stefan Sauer <ensonic@users.sf.net>");
 150
 151   base_aggregator_class->aggregate =
 152       GST_DEBUG_FUNCPTR (gst_test_aggregator_aggregate);
 153 }
 154
 155 static void
 156 gst_test_aggregator_init (GstTestAggregator * self)
 157 {
 158   GstAggregator *agg = GST_AGGREGATOR (self);
 159   gst_segment_init (&agg->segment, GST_FORMAT_BYTES);
 160   self->timestamp = 0;
 161 }
 162
 163 static gboolean
 164 gst_test_aggregator_plugin_init (GstPlugin * plugin)
 165 {
 166   return gst_element_register (plugin, "testaggregator", GST_RANK_NONE,
 167       GST_TYPE_TEST_AGGREGATOR);
 168 }
 169
 170 static gboolean
 171 gst_test_aggregator_plugin_register (void)
 172 {
 173   return gst_plugin_register_static (GST_VERSION_MAJOR,
 174       GST_VERSION_MINOR,
 175       "testaggregator",
 176       "Combine buffers",
 177       gst_test_aggregator_plugin_init,
 178       VERSION, GST_LICENSE, PACKAGE, GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN);
 179 }
 180
 181 typedef struct
 182 {
 183   GstEvent *event;
 184   GstBuffer *buffer;
 185   GstElement *aggregator;
 186   GstPad *sinkpad, *srcpad;
 187   GstFlowReturn expected_result;
 188
 189   /*                       ------------------
 190    * -----------   --------|--              |
 191    * | srcpad | -- | sinkpad |  aggregator  |
 192    * -----------   --------|--              |
 193    *                       ------------------
 194    *  This is for 1 Chain, we can have several
 195    */
 196 } ChainData;
 197
 198 typedef struct
 199 {
 200   GMainLoop *ml;
 201   GstPad *srcpad,               /* srcpad of the GstAggregator */
 202    *sinkpad;                    /* fake sinkpad to which GstAggregator.srcpad is linked */
 203   guint timeout_id;
 204   GstElement *aggregator;
 205
 206   /* -----------------|
 207    * |             ----------    -----------
 208    * | aggregator  | srcpad | -- | sinkpad |
 209    * |             ----------    -----------
 210    * -----------------|
 211    */
 212
 213   gint flush_start_events, flush_stop_events;
 214 } TestData;
 215
 216 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
 217     GST_PAD_SRC,
 218     GST_PAD_ALWAYS,
 219     GST_STATIC_CAPS_ANY);
 220
 221 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
 222     GST_PAD_SINK,
 223     GST_PAD_ALWAYS,
 224     GST_STATIC_CAPS_ANY);
 225
 226 static gpointer
 227 push_buffer (gpointer user_data)
 228 {
 229   GstFlowReturn flow;
 230   GstCaps *caps;
 231   ChainData *chain_data = (ChainData *) user_data;
 232   GstSegment segment;
 233
 234   gst_pad_push_event (chain_data->srcpad, gst_event_new_stream_start ("test"));
 235
 236   caps = gst_caps_new_empty_simple ("foo/x-bar");
 237   gst_pad_push_event (chain_data->srcpad, gst_event_new_caps (caps));
 238   gst_caps_unref (caps);
 239
 240   gst_segment_init (&segment, GST_FORMAT_TIME);
 241   gst_pad_push_event (chain_data->srcpad, gst_event_new_segment (&segment));
 242
 243   GST_DEBUG ("Pushing buffer on pad: %s:%s",
 244       GST_DEBUG_PAD_NAME (chain_data->sinkpad));
 245   flow = gst_pad_push (chain_data->srcpad, chain_data->buffer);
 246   fail_unless (flow == chain_data->expected_result,
 247       "got flow %s instead of %s on %s:%s", gst_flow_get_name (flow),
 248       gst_flow_get_name (chain_data->expected_result),
 249       GST_DEBUG_PAD_NAME (chain_data->sinkpad));
 250   chain_data->buffer = NULL;
 251
 252   return NULL;
 253 }
 254
 255 static gpointer
 256 push_event (gpointer user_data)
 257 {
 258   ChainData *chain_data = (ChainData *) user_data;
 259
 260   GST_INFO_OBJECT (chain_data->srcpad, "Pushing event: %"
 261       GST_PTR_FORMAT, chain_data->event);
 262   fail_unless (gst_pad_push_event (chain_data->srcpad,
 263           chain_data->event) == TRUE);
 264
 265   return NULL;
 266 }
 267
 268 static gboolean
 269 _aggregate_timeout (GMainLoop * ml)
 270 {
 271   g_main_loop_quit (ml);
 272
 273   fail_unless ("No buffer found on aggregator.srcpad -> TIMEOUT" == NULL);
 274
 275   return FALSE;
 276 }
 277
 278 static gboolean
 279 _quit (GMainLoop * ml)
 280 {
 281   GST_DEBUG ("QUITING ML");
 282   g_main_loop_quit (ml);
 283
 284   return G_SOURCE_REMOVE;
 285 }
 286
 287 static GstPadProbeReturn
 288 _aggregated_cb (GstPad * pad, GstPadProbeInfo * info, GMainLoop * ml)
 289 {
 290   GST_DEBUG ("SHould quit ML");
 291   g_idle_add ((GSourceFunc) _quit, ml);
 292
 293   return GST_PAD_PROBE_REMOVE;
 294 }
 295
 296 static GstPadProbeReturn
 297 downstream_probe_cb (GstPad * pad, GstPadProbeInfo * info, TestData * test)
 298 {
 299   GST_DEBUG ("PROBING ");
 300   if (info->type & GST_PAD_PROBE_TYPE_EVENT_FLUSH) {
 301     if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_EVENT (info)) ==
 302         GST_EVENT_FLUSH_START) {
 303
 304       g_atomic_int_inc (&test->flush_start_events);
 305       GST_DEBUG ("==========> FLUSH: %i", test->flush_start_events);
 306     } else if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_EVENT (info)) ==
 307         GST_EVENT_FLUSH_STOP)
 308       g_atomic_int_inc (&test->flush_stop_events);
 309   }
 310
 311   return GST_PAD_PROBE_DROP;
 312 }
 313
 314 /*
 315  * Not thread safe, will create a new ChainData which contains
 316  * an activated src pad linked to a requested sink pad of @agg, and
 317  * a newly allocated buffer ready to be pushed. Caller needs to
 318  * clear with _chain_data_clear after.
 319  */
 320 static void
 321 _chain_data_init (ChainData * data, GstElement * agg)
 322 {
 323   static gint num_src_pads = 0;
 324   gchar *pad_name = g_strdup_printf ("src%d", num_src_pads);
 325
 326   num_src_pads += 1;
 327
 328   data->srcpad = gst_pad_new_from_static_template (&srctemplate, pad_name);
 329   g_free (pad_name);
 330   gst_pad_set_active (data->srcpad, TRUE);
 331   data->aggregator = agg;
 332   data->buffer = gst_buffer_new ();
 333   data->sinkpad = gst_element_get_request_pad (agg, "sink_%u");
 334   fail_unless (GST_IS_PAD (data->sinkpad));
 335   fail_unless (gst_pad_link (data->srcpad, data->sinkpad) == GST_PAD_LINK_OK);
 336 }
 337
 338 static void
 339 _chain_data_clear (ChainData * data)
 340 {
 341   if (data->buffer)
 342     gst_buffer_unref (data->buffer);
 343   if (data->srcpad)
 344     gst_object_unref (data->srcpad);
 345   if (data->sinkpad)
 346     gst_object_unref (data->sinkpad);
 347 }
 348
 349 static void
 350 _test_data_init (TestData * test, gboolean needs_flushing)
 351 {
 352   test->aggregator = gst_element_factory_make ("testaggregator", NULL);
 353   gst_element_set_state (test->aggregator, GST_STATE_PLAYING);
 354   test->ml = g_main_loop_new (NULL, TRUE);
 355   test->srcpad = GST_AGGREGATOR (test->aggregator)->srcpad;
 356
 357   GST_DEBUG ("Srcpad: %p", test->srcpad);
 358
 359   if (needs_flushing) {
 360     static gint num_sink_pads = 0;
 361     gchar *pad_name = g_strdup_printf ("sink%d", num_sink_pads);
 362
 363     num_sink_pads += 1;
 364     test->sinkpad = gst_pad_new_from_static_template (&sinktemplate, pad_name);
 365     g_free (pad_name);
 366     fail_unless (gst_pad_link (test->srcpad, test->sinkpad) == GST_PAD_LINK_OK);
 367     gst_pad_add_probe (test->srcpad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM |
 368         GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM |
 369         GST_PAD_PROBE_TYPE_EVENT_FLUSH,
 370         (GstPadProbeCallback) downstream_probe_cb, test, NULL);
 371   } else {
 372     gst_pad_add_probe (test->srcpad, GST_PAD_PROBE_TYPE_BUFFER,
 373         (GstPadProbeCallback) _aggregated_cb, test->ml, NULL);
 374   }
 375
 376
 377   test->timeout_id =
 378       g_timeout_add (1000, (GSourceFunc) _aggregate_timeout, test->ml);
 379 }
 380
 381 static void
 382 _test_data_clear (TestData * test)
 383 {
 384   gst_element_set_state (test->aggregator, GST_STATE_NULL);
 385   gst_object_unref (test->aggregator);
 386
 387   if (test->sinkpad)
 388     gst_object_unref (test->sinkpad);
 389
 390   g_main_loop_unref (test->ml);
 391 }
 392
 393 GST_START_TEST (test_aggregate)
 394 {
 395   GThread *thread1, *thread2;
 396
 397   ChainData data1 = { 0, };
 398   ChainData data2 = { 0, };
 399   TestData test = { 0, };
 400
 401   _test_data_init (&test, FALSE);
 402   _chain_data_init (&data1, test.aggregator);
 403   _chain_data_init (&data2, test.aggregator);
 404
 405   thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL);
 406   thread2 = g_thread_try_new ("gst-check", push_buffer, &data2, NULL);
 407
 408   g_main_loop_run (test.ml);
 409   g_source_remove (test.timeout_id);
 410
 411
 412   /* these will return immediately as when the data is popped the threads are
 413    * unlocked and will terminate */
 414   g_thread_join (thread1);
 415   g_thread_join (thread2);
 416
 417   _chain_data_clear (&data1);
 418   _chain_data_clear (&data2);
 419   _test_data_clear (&test);
 420 }
 421
 422 GST_END_TEST;
 423
 424 GST_START_TEST (test_aggregate_eos)
 425 {
 426   GThread *thread1, *thread2;
 427
 428   ChainData data1 = { 0, };
 429   ChainData data2 = { 0, };
 430   TestData test = { 0, };
 431
 432   _test_data_init (&test, FALSE);
 433   _chain_data_init (&data1, test.aggregator);
 434   _chain_data_init (&data2, test.aggregator);
 435
 436   data2.event = gst_event_new_eos ();
 437
 438   thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL);
 439   thread2 = g_thread_try_new ("gst-check", push_event, &data2, NULL);
 440
 441   g_main_loop_run (test.ml);
 442   g_source_remove (test.timeout_id);
 443
 444   /* these will return immediately as when the data is popped the threads are
 445    * unlocked and will terminate */
 446   g_thread_join (thread1);
 447   g_thread_join (thread2);
 448
 449   _chain_data_clear (&data1);
 450   _chain_data_clear (&data2);
 451   _test_data_clear (&test);
 452 }
 453
 454 GST_END_TEST;
 455
 456 #define NUM_BUFFERS 3
 457 static void
 458 handoff (GstElement * fakesink, GstBuffer * buf, GstPad * pad, guint * count)
 459 {
 460   *count = *count + 1;
 461   GST_DEBUG ("HANDOFF: %i", *count);
 462 }
 463
 464 /* Test a linear pipeline using aggregator */
 465 GST_START_TEST (test_linear_pipeline)
 466 {
 467   GstBus *bus;
 468   GstMessage *msg;
 469   GstElement *pipeline, *src, *agg, *sink;
 470
 471   gint count = 0;
 472
 473   pipeline = gst_pipeline_new ("pipeline");
 474   src = gst_check_setup_element ("fakesrc");
 475   g_object_set (src, "num-buffers", NUM_BUFFERS, "sizetype", 2, "sizemax", 4,
 476       NULL);
 477   agg = gst_check_setup_element ("testaggregator");
 478   sink = gst_check_setup_element ("fakesink");
 479   g_object_set (sink, "signal-handoffs", TRUE, NULL);
 480   g_signal_connect (sink, "handoff", (GCallback) handoff, &count);
 481
 482   fail_unless (gst_bin_add (GST_BIN (pipeline), src));
 483   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
 484   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
 485   fail_unless (gst_element_link (src, agg));
 486   fail_unless (gst_element_link (agg, sink));
 487
 488   bus = gst_element_get_bus (pipeline);
 489   fail_if (bus == NULL);
 490   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 491
 492   msg = gst_bus_poll (bus, GST_MESSAGE_EOS | GST_MESSAGE_ERROR, -1);
 493   fail_if (GST_MESSAGE_TYPE (msg) != GST_MESSAGE_EOS);
 494   gst_message_unref (msg);
 495
 496   fail_unless_equals_int (count, NUM_BUFFERS);
 497
 498   gst_element_set_state (pipeline, GST_STATE_NULL);
 499   gst_object_unref (bus);
 500   gst_object_unref (pipeline);
 501 }
 502
 503 GST_END_TEST;
 504
 505 GST_START_TEST (test_two_src_pipeline)
 506 {
 507   GstBus *bus;
 508   GstMessage *msg;
 509   GstElement *pipeline, *src, *src1, *agg, *sink;
 510
 511   gint count = 0;
 512
 513   pipeline = gst_pipeline_new ("pipeline");
 514   src = gst_element_factory_make ("fakesrc", NULL);
 515   g_object_set (src, "num-buffers", NUM_BUFFERS, "sizetype", 2, "sizemax", 4,
 516       NULL);
 517
 518   src1 = gst_element_factory_make ("fakesrc", NULL);
 519   g_object_set (src1, "num-buffers", NUM_BUFFERS + 1, "sizetype", 2, "sizemax",
 520       4, NULL);
 521
 522   agg = gst_check_setup_element ("testaggregator");
 523   sink = gst_check_setup_element ("fakesink");
 524   g_object_set (sink, "signal-handoffs", TRUE, NULL);
 525   g_signal_connect (sink, "handoff", (GCallback) handoff, &count);
 526
 527   fail_unless (gst_bin_add (GST_BIN (pipeline), src));
 528   fail_unless (gst_bin_add (GST_BIN (pipeline), src1));
 529   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
 530   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
 531   fail_unless (gst_element_link (src, agg));
 532   fail_unless (gst_element_link (src1, agg));
 533   fail_unless (gst_element_link (agg, sink));
 534
 535   bus = gst_element_get_bus (pipeline);
 536   fail_if (bus == NULL);
 537   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 538
 539   msg = gst_bus_poll (bus, GST_MESSAGE_EOS | GST_MESSAGE_ERROR, -1);
 540   fail_if (GST_MESSAGE_TYPE (msg) != GST_MESSAGE_EOS);
 541   gst_message_unref (msg);
 542
 543   fail_unless_equals_int (count, NUM_BUFFERS + 1);
 544
 545   gst_element_set_state (pipeline, GST_STATE_NULL);
 546   gst_object_unref (bus);
 547   gst_object_unref (pipeline);
 548 }
 549
 550 GST_END_TEST;
 551
 552 static GstPadProbeReturn
 553 _drop_buffer_probe_cb (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
 554 {
 555   gint wait;
 556
 557   if (GST_IS_BUFFER (info->data)) {
 558     wait = GPOINTER_TO_INT (user_data);
 559     if (wait > 0)
 560       g_usleep (wait / 1000);
 561     return GST_PAD_PROBE_DROP;
 562   }
 563
 564   return GST_PAD_PROBE_PASS;
 565 }
 566
 567 #define TIMEOUT_NUM_BUFFERS 20
 568 static void
 569 _test_timeout (gint buffer_wait)
 570 {
 571   GstBus *bus;
 572   GstMessage *msg;
 573   GstElement *pipeline, *src, *src1, *agg, *sink;
 574   GstPad *src1pad;
 575
 576   gint count = 0;
 577
 578   pipeline = gst_pipeline_new ("pipeline");
 579   src = gst_element_factory_make ("fakesrc", NULL);
 580   g_object_set (src, "num-buffers", TIMEOUT_NUM_BUFFERS, "sizetype", 2,
 581       "sizemax", 4, NULL);
 582
 583   src1 = gst_element_factory_make ("fakesrc", NULL);
 584   g_object_set (src1, "num-buffers", TIMEOUT_NUM_BUFFERS, "sizetype", 2,
 585       "sizemax", 4, NULL);
 586
 587   agg = gst_check_setup_element ("testaggregator");
 588   g_object_set (agg, "latency", G_GINT64_CONSTANT (1000) /* 1 us */ , NULL);
 589   sink = gst_check_setup_element ("fakesink");
 590   g_object_set (sink, "signal-handoffs", TRUE, NULL);
 591   g_signal_connect (sink, "handoff", (GCallback) handoff, &count);
 592
 593   fail_unless (gst_bin_add (GST_BIN (pipeline), src));
 594   fail_unless (gst_bin_add (GST_BIN (pipeline), src1));
 595   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
 596   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
 597
 598   src1pad = gst_element_get_static_pad (src1, "src");
 599   fail_if (src1pad == NULL);
 600   gst_pad_add_probe (src1pad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
 601       (GstPadProbeCallback) _drop_buffer_probe_cb,
 602       GINT_TO_POINTER (buffer_wait), NULL);
 603
 604   fail_unless (gst_element_link (src, agg));
 605   fail_unless (gst_element_link (src1, agg));
 606   fail_unless (gst_element_link (agg, sink));
 607
 608   bus = gst_element_get_bus (pipeline);
 609   fail_if (bus == NULL);
 610   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 611
 612   msg = gst_bus_poll (bus, GST_MESSAGE_EOS | GST_MESSAGE_ERROR, -1);
 613   fail_if (GST_MESSAGE_TYPE (msg) != GST_MESSAGE_EOS);
 614   gst_message_unref (msg);
 615
 616   /* cannot rely on the exact number of buffers as the timeout may produce
 617    * more buffers with the unsynchronized _aggregate() implementation in
 618    * testaggregator */
 619   fail_if (count < TIMEOUT_NUM_BUFFERS);
 620
 621   gst_element_set_state (pipeline, GST_STATE_NULL);
 622   gst_object_unref (src1pad);
 623   gst_object_unref (bus);
 624   gst_object_unref (pipeline);
 625 }
 626
 627 GST_START_TEST (test_timeout_pipeline)
 628 {
 629   _test_timeout (0);
 630 }
 631
 632 GST_END_TEST;
 633
 634 GST_START_TEST (test_timeout_pipeline_with_wait)
 635 {
 636   _test_timeout (1000000 /* 1 ms */ );
 637 }
 638
 639 GST_END_TEST;
 640
 641 GST_START_TEST (test_flushing_seek)
 642 {
 643   GstEvent *event;
 644   GThread *thread1, *thread2;
 645
 646   ChainData data1 = { 0, };
 647   ChainData data2 = { 0, };
 648   TestData test = { 0, };
 649
 650   _test_data_init (&test, TRUE);
 651
 652   /* Queue a buffer in agg:sink_1. Then do a flushing seek and check that the
 653    * new flushing seek logic is triggered. On the first FLUSH_START call the
 654    * buffers queued in collectpads should get flushed. Only one FLUSH_START and
 655    * one FLUSH_STOP should be forwarded downstream.
 656    */
 657   _chain_data_init (&data1, test.aggregator);
 658   _chain_data_init (&data2, test.aggregator);
 659   GST_BUFFER_TIMESTAMP (data2.buffer) = 0;
 660
 661   gst_segment_init (&GST_AGGREGATOR (test.aggregator)->segment,
 662       GST_FORMAT_TIME);
 663   /* now do a successful flushing seek */
 664   event = gst_event_new_seek (1, GST_FORMAT_TIME, GST_SEEK_FLAG_FLUSH,
 665       GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, 10 * GST_SECOND);
 666   fail_unless (gst_pad_send_event (test.srcpad, event));
 667
 668   /* flushing starts once one of the upstream elements sends the first
 669    * FLUSH_START */
 670   fail_unless_equals_int (test.flush_start_events, 0);
 671   fail_unless_equals_int (test.flush_stop_events, 0);
 672
 673   /* flush ogg:sink_0. This flushs collectpads, calls ::flush() and sends
 674    * FLUSH_START downstream */
 675   GST_DEBUG ("Flushing: %s:%s", GST_DEBUG_PAD_NAME (data2.sinkpad));
 676   fail_unless (gst_pad_push_event (data2.srcpad, gst_event_new_flush_start ()));
 677
 678   /* expect this buffer to be flushed */
 679   data2.expected_result = GST_FLOW_FLUSHING;
 680   thread2 = g_thread_try_new ("gst-check", push_buffer, &data2, NULL);
 681
 682   fail_unless (gst_pad_push_event (data1.srcpad, gst_event_new_flush_start ()));
 683   fail_unless_equals_int (test.flush_start_events, 1);
 684   fail_unless_equals_int (test.flush_stop_events, 0);
 685
 686   /* the first FLUSH_STOP is not forwarded downstream */
 687   fail_unless (gst_pad_push_event (data1.srcpad,
 688           gst_event_new_flush_stop (TRUE)));
 689   fail_unless_equals_int (test.flush_start_events, 1);
 690   fail_unless_equals_int (test.flush_stop_events, 0);
 691
 692   /* at this point even the other pad agg:sink_1 should be flushing so thread2
 693    * should have stopped */
 694   g_thread_join (thread2);
 695
 696   /* push a buffer on agg:sink_0 to trigger one collect after flushing to verify
 697    * that flushing completes once all the pads have been flushed */
 698   thread1 = g_thread_try_new ("gst-check", push_buffer, &data1, NULL);
 699
 700   /* flush agg:sink_1 as well. This completes the flushing seek so a FLUSH_STOP is
 701    * sent downstream */
 702   gst_pad_push_event (data2.srcpad, gst_event_new_flush_stop (TRUE));
 703
 704   /* and the last FLUSH_STOP is forwarded downstream */
 705   fail_unless_equals_int (test.flush_start_events, 1);
 706
 707   /*  Check collected */
 708   gst_pad_add_probe (test.srcpad, GST_PAD_PROBE_TYPE_BUFFER,
 709       (GstPadProbeCallback) _aggregated_cb, test.ml, NULL);
 710
 711   data2.event = gst_event_new_eos ();
 712   thread2 = g_thread_try_new ("gst-check", push_event, &data2, NULL);
 713
 714   g_main_loop_run (test.ml);
 715   g_source_remove (test.timeout_id);
 716
 717   fail_unless_equals_int (test.flush_stop_events, 1);
 718
 719   /* these will return immediately as at this point the threads have been
 720    * unlocked and are finished */
 721   g_thread_join (thread1);
 722   g_thread_join (thread2);
 723
 724   _chain_data_clear (&data1);
 725   _chain_data_clear (&data2);
 726   _test_data_clear (&test);
 727
 728 }
 729
 730 GST_END_TEST;
 731
 732 static void
 733 infinite_seek (guint num_srcs, guint num_seeks)
 734 {
 735   GstBus *bus;
 736   GstMessage *message;
 737   GstElement *pipeline, *src, *agg, *sink;
 738
 739   gint count = 0, i;
 740   gboolean seek_res, carry_on = TRUE;
 741
 742   gst_init (NULL, NULL);
 743
 744   pipeline = gst_pipeline_new ("pipeline");
 745
 746   agg = gst_check_setup_element ("testaggregator");
 747   sink = gst_check_setup_element ("fakesink");
 748
 749   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
 750   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
 751   fail_unless (gst_element_link (agg, sink));
 752
 753   for (i = 0; i < num_srcs; i++) {
 754     src = gst_element_factory_make ("fakesrc", NULL);
 755     g_object_set (src, "sizetype", 2, "sizemax", 4, NULL);
 756     fail_unless (gst_bin_add (GST_BIN (pipeline), src));
 757     fail_unless (gst_element_link (src, agg));
 758   }
 759
 760   bus = gst_element_get_bus (pipeline);
 761   fail_if (bus == NULL);
 762   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 763   while (count < num_seeks && carry_on) {
 764     message = gst_bus_poll (bus, GST_MESSAGE_ANY, GST_SECOND / 10);
 765     if (message) {
 766       switch (GST_MESSAGE_TYPE (message)) {
 767         case GST_MESSAGE_EOS:
 768         {
 769           /* we should check if we really finished here */
 770           GST_WARNING ("Got an EOS");
 771           carry_on = FALSE;
 772           break;
 773         }
 774         case GST_MESSAGE_STATE_CHANGED:
 775         {
 776           GstState new;
 777
 778           if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) {
 779             gst_message_parse_state_changed (message, NULL, &new, NULL);
 780
 781             if (new != GST_STATE_PLAYING)
 782               break;
 783
 784             GST_INFO ("Seeking (num: %i)", count);
 785             seek_res =
 786                 gst_element_seek_simple (sink, GST_FORMAT_BYTES,
 787                 GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, 0);
 788             GST_INFO ("seek result is : %d", seek_res);
 789             fail_unless (seek_res != 0);
 790             count++;
 791           }
 792
 793           break;
 794         }
 795         case GST_MESSAGE_ERROR:
 796           GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message);
 797           carry_on = FALSE;
 798           fail_error_message (message);
 799           break;
 800         default:
 801           break;
 802       }
 803       gst_message_unref (message);
 804     }
 805   }
 806
 807   gst_element_set_state (pipeline, GST_STATE_NULL);
 808   gst_object_unref (bus);
 809   gst_object_unref (pipeline);
 810 }
 811
 812 GST_START_TEST (test_infinite_seek)
 813 {
 814   infinite_seek (2, 500);
 815 }
 816
 817 GST_END_TEST;
 818
 819 GST_START_TEST (test_infinite_seek_50_src)
 820 {
 821   infinite_seek (50, 100);
 822 }
 823
 824 GST_END_TEST;
 825
 826 typedef struct
 827 {
 828   GstElement *agg, *src, *pipeline;
 829   GCond *cond;
 830   GMutex *lock;
 831 } RemoveElementData;
 832
 833 static GstPadProbeReturn
 834 pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, RemoveElementData * data)
 835 {
 836   GstPad *peer;
 837
 838   GST_INFO_OBJECT (pad, "Removing pad");
 839
 840   peer = gst_pad_get_peer (pad);
 841   gst_pad_unlink (pad, peer);
 842   gst_element_release_request_pad (data->agg, peer);
 843   fail_unless (gst_bin_remove (GST_BIN (data->pipeline), data->src));
 844   gst_object_unref (peer);
 845
 846   g_mutex_lock (data->lock);
 847   g_cond_broadcast (data->cond);
 848   g_mutex_unlock (data->lock);
 849
 850   return GST_PAD_PROBE_OK;
 851 }
 852
 853 GST_START_TEST (test_add_remove)
 854 {
 855   /* Used to notify that we removed the pad from  */
 856   GCond cond;
 857   GMutex lock;
 858
 859   GstBus *bus;
 860   GstState state;
 861   GstMessage *message;
 862   gboolean carry_on = TRUE;
 863   guint num_iterations = 100;
 864
 865   GstPad *pad;
 866   GstElement *pipeline, *src, *src1 = NULL, *agg, *sink;
 867
 868   gint count = 0;
 869
 870   gst_init (NULL, NULL);
 871   g_mutex_init (&lock);
 872   g_cond_init (&cond);
 873
 874   pipeline = gst_pipeline_new ("pipeline");
 875
 876   agg = gst_check_setup_element ("testaggregator");
 877   sink = gst_check_setup_element ("fakesink");
 878
 879   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
 880   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
 881   fail_unless (gst_element_link (agg, sink));
 882
 883   bus = gst_element_get_bus (pipeline);
 884   while (count < num_iterations) {
 885
 886     src = gst_element_factory_make ("fakesrc", NULL);
 887     g_object_set (src, "num-buffers", 100000, "sizetype", 2, "sizemax", 4,
 888         NULL);
 889     gst_element_set_locked_state (src, TRUE);
 890     fail_unless (gst_bin_add (GST_BIN (pipeline), src));
 891     fail_unless (gst_element_link (src, agg));
 892     gst_element_set_locked_state (src, FALSE);
 893     fail_unless (gst_element_sync_state_with_parent (src));
 894
 895     if (count == 0)
 896       gst_element_set_state (pipeline, GST_STATE_PLAYING);
 897
 898     /* Now make sure the seek happend */
 899     carry_on = TRUE;
 900     do {
 901       message = gst_bus_timed_pop (bus, -1);
 902       switch (GST_MESSAGE_TYPE (message)) {
 903         case GST_MESSAGE_EOS:
 904         {
 905           /* we should check if we really finished here */
 906           GST_WARNING ("Got an EOS");
 907           carry_on = FALSE;
 908           break;
 909         }
 910         case GST_MESSAGE_STATE_CHANGED:
 911         {
 912           if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) {
 913             gst_message_parse_state_changed (message, NULL, &state, NULL);
 914
 915             if (state == GST_STATE_PLAYING) {
 916               RemoveElementData data;
 917
 918               carry_on = FALSE;
 919               if (count == 0) {
 920                 GST_DEBUG ("First run, not removing any element yet");
 921
 922                 break;
 923               }
 924
 925               data.src = gst_object_ref (src1);
 926               data.agg = agg;
 927               data.lock = &lock;
 928               data.cond = &cond;
 929               data.pipeline = pipeline;
 930               pad = gst_element_get_static_pad (data.src, "src");
 931
 932               g_mutex_lock (&lock);
 933               gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
 934                   (GstPadProbeCallback) pad_probe_cb, &data, NULL);
 935               GST_INFO ("Waiting for %" GST_PTR_FORMAT " %s", pad,
 936                   gst_element_state_get_name (GST_STATE (data.src)));
 937               g_cond_wait (&cond, &lock);
 938               g_mutex_unlock (&lock);
 939               gst_object_unref (pad);
 940
 941               /*  We can not set state from the streaming thread so we
 942                *  need to make sure that the source has been removed
 943                *  before setting its state to NULL */
 944               gst_element_set_state (data.src, GST_STATE_NULL);
 945
 946               gst_object_unref (data.src);
 947             }
 948           }
 949
 950           break;
 951         }
 952         case GST_MESSAGE_ERROR:
 953         {
 954           GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message);
 955           carry_on = FALSE;
 956           fail_error_message (message);
 957           break;
 958         }
 959         default:
 960           break;
 961       }
 962
 963       gst_message_unref (message);
 964     } while (carry_on);
 965
 966     GST_INFO ("Seeking");
 967     fail_unless (gst_element_seek_simple (pipeline, GST_FORMAT_BYTES,
 968             GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, 0));
 969
 970     count++;
 971     src1 = src;
 972   }
 973   gst_element_set_state (pipeline, GST_STATE_NULL);
 974   gst_object_unref (bus);
 975   gst_object_unref (pipeline);
 976   g_mutex_clear (&lock);
 977   g_cond_clear (&cond);
 978 }
 979
 980 GST_END_TEST;
 981
 982 GST_START_TEST (test_change_state_intensive)
 983 {
 984   GstBus *bus;
 985   GstMessage *message;
 986   GstElement *pipeline, *src, *agg, *sink;
 987
 988   gint i, state_i = 0, num_srcs = 3;
 989   gboolean carry_on = TRUE, ready = FALSE;
 990   GstStateChangeReturn state_return;
 991   GstState wanted_state, wanted_states[] = {
 992     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY,
 993     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY,
 994     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_READY,
 995     GST_STATE_PAUSED, GST_STATE_READY, GST_STATE_PAUSED, GST_STATE_READY,
 996     GST_STATE_PAUSED, GST_STATE_READY, GST_STATE_PAUSED, GST_STATE_NULL,
 997     GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_NULL,
 998     GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PAUSED, GST_STATE_NULL,
 999     GST_STATE_PAUSED, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL,
1000     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL,
1001     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL,
1002     GST_STATE_PLAYING, GST_STATE_NULL, GST_STATE_PLAYING, GST_STATE_NULL,
1003   };
1004
1005   gst_init (NULL, NULL);
1006
1007   pipeline = gst_pipeline_new ("pipeline");
1008
1009   agg = gst_check_setup_element ("testaggregator");
1010   sink = gst_check_setup_element ("fakesink");
1011
1012   fail_unless (gst_bin_add (GST_BIN (pipeline), agg));
1013   fail_unless (gst_bin_add (GST_BIN (pipeline), sink));
1014   fail_unless (gst_element_link (agg, sink));
1015
1016   for (i = 0; i < num_srcs; i++) {
1017     src = gst_element_factory_make ("fakesrc", NULL);
1018     g_object_set (src, "sizetype", 2, "sizemax", 4, NULL);
1019     fail_unless (gst_bin_add (GST_BIN (pipeline), src));
1020     fail_unless (gst_element_link (src, agg));
1021   }
1022
1023   bus = gst_element_get_bus (pipeline);
1024   fail_if (bus == NULL);
1025
1026   wanted_state = wanted_states[state_i++];
1027   state_return = gst_element_set_state (pipeline, wanted_state);
1028
1029   while (state_i < G_N_ELEMENTS (wanted_states) && carry_on) {
1030     if (state_return == GST_STATE_CHANGE_SUCCESS && ready) {
1031       wanted_state = wanted_states[state_i++];
1032       fail_unless (gst_element_set_state (pipeline, wanted_state),
1033           GST_STATE_CHANGE_SUCCESS);
1034       GST_INFO ("Wanted state: %s", gst_element_state_get_name (wanted_state));
1035     }
1036
1037     message = gst_bus_poll (bus, GST_MESSAGE_ANY, GST_SECOND / 10);
1038     if (message) {
1039       switch (GST_MESSAGE_TYPE (message)) {
1040         case GST_MESSAGE_EOS:
1041         {
1042           /* we should check if we really finished here */
1043           GST_WARNING ("Got an EOS");
1044           carry_on = FALSE;
1045           break;
1046         }
1047         case GST_MESSAGE_STATE_CHANGED:
1048         {
1049           GstState new;
1050
1051           if (GST_MESSAGE_SRC (message) == GST_OBJECT (pipeline)) {
1052             gst_message_parse_state_changed (message, NULL, &new, NULL);
1053
1054             if (new != wanted_state) {
1055               ready = FALSE;
1056               break;
1057             }
1058
1059             GST_DEBUG ("State %s reached",
1060                 gst_element_state_get_name (wanted_state));
1061             wanted_state = wanted_states[state_i++];
1062             GST_DEBUG ("Wanted state: %s",
1063                 gst_element_state_get_name (wanted_state));
1064             state_return = gst_element_set_state (pipeline, wanted_state);
1065             fail_unless (state_return == GST_STATE_CHANGE_SUCCESS ||
1066                 state_return == GST_STATE_CHANGE_ASYNC);
1067             ready = TRUE;
1068           }
1069
1070           break;
1071         }
1072         case GST_MESSAGE_ERROR:
1073           GST_ERROR ("Error on the bus: %" GST_PTR_FORMAT, message);
1074           carry_on = FALSE;
1075           break;
1076         default:
1077           break;
1078       }
1079       gst_message_unref (message);
1080     }
1081   }
1082
1083   gst_element_set_state (pipeline, GST_STATE_NULL);
1084   gst_object_unref (bus);
1085   gst_object_unref (pipeline);
1086 }
1087
1088 GST_END_TEST;
1089
1090 static Suite *
1091 gst_aggregator_suite (void)
1092 {
1093   Suite *suite;
1094   TCase *general;
1095
1096   gst_test_aggregator_plugin_register ();
1097
1098   suite = suite_create ("GstAggregator");
1099
1100   general = tcase_create ("general");
1101   suite_add_tcase (suite, general);
1102   tcase_add_test (general, test_aggregate);
1103   tcase_add_test (general, test_aggregate_eos);
1104   tcase_add_test (general, test_flushing_seek);
1105   tcase_add_test (general, test_infinite_seek);
1106   tcase_add_test (general, test_infinite_seek_50_src);
1107   tcase_add_test (general, test_linear_pipeline);
1108   tcase_add_test (general, test_two_src_pipeline);
1109   tcase_add_test (general, test_timeout_pipeline);
1110   tcase_add_test (general, test_timeout_pipeline_with_wait);
1111   tcase_add_test (general, test_add_remove);
1112   tcase_add_test (general, test_change_state_intensive);
1113
1114   return suite;
1115 }
1116
1117 GST_CHECK_MAIN (gst_aggregator);