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