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