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