libs/gst/base/gstaggregator.c

   1 /* GStreamer
   2  * Copyright (C) 2014 Mathieu Duponchelle <mathieu.duponchelle@opencreed.com>
   3  * Copyright (C) 2014 Thibault Saunier <tsaunier@gnome.org>
   4  *
   5  * gstaggregator.c:
   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  * SECTION: gstaggregator
  24  * @short_description: manages a set of pads with the purpose of
  25  * aggregating their buffers.
  26  * @see_also: gstcollectpads for historical reasons.
  27  *
  28  * Manages a set of pads with the purpose of aggregating their buffers.
  29  * Control is given to the subclass when all pads have data.
  30  * <itemizedlist>
  31  *  <listitem><para>
  32  *    Base class for mixers and muxers. Implementers should at least implement
  33  *    the aggregate () vmethod.
  34  *  </para></listitem>
  35  *  <listitem><para>
  36  *    When data is queued on all pads, tha aggregate vmethod is called.
  37  *  </para></listitem>
  38  *  <listitem><para>
  39  *    One can peek at the data on any given GstAggregatorPad with the
  40  *    gst_aggregator_pad_get_buffer () method, and take ownership of it
  41  *    with the gst_aggregator_pad_steal_buffer () method. When a buffer
  42  *    has been taken with steal_buffer (), a new buffer can be queued
  43  *    on that pad.
  44  *  </para></listitem>
  45  *  <listitem><para>
  46  *    If the subclass wishes to push a buffer downstream in its aggregate
  47  *    implementation, it should do so through the
  48  *    gst_aggregator_finish_buffer () method. This method will take care
  49  *    of sending and ordering mandatory events such as stream start, caps
  50  *    and segment.
  51  *  </para></listitem>
  52  *  <listitem><para>
  53  *    Same goes for EOS events, which should not be pushed directly by the
  54  *    subclass, it should instead return GST_FLOW_EOS in its aggregate
  55  *    implementation.
  56  *  </para></listitem>
  57  * </itemizedlist>
  58  */
  59
  60 #ifdef HAVE_CONFIG_H
  61 #  include "config.h"
  62 #endif
  63
  64 #include <string.h>             /* strlen */
  65
  66 #include "gstaggregator.h"
  67
  68
  69 /*  Might become API */
  70 static void gst_aggregator_merge_tags (GstAggregator * aggregator,
  71     const GstTagList * tags, GstTagMergeMode mode);
  72
  73 GST_DEBUG_CATEGORY_STATIC (aggregator_debug);
  74 #define GST_CAT_DEFAULT aggregator_debug
  75
  76 /* GstAggregatorPad definitions */
  77 #define PAD_LOCK_EVENT(pad)   G_STMT_START {                            \
  78   GST_LOG_OBJECT (pad, "Taking EVENT lock from thread %p",              \
  79         g_thread_self());                                               \
  80   g_mutex_lock(&pad->priv->event_lock);                                 \
  81   GST_LOG_OBJECT (pad, "Took EVENT lock from thread %p",              \
  82         g_thread_self());                                               \
  83   } G_STMT_END
  84
  85 #define PAD_UNLOCK_EVENT(pad)  G_STMT_START {                           \
  86   GST_LOG_OBJECT (pad, "Releasing EVENT lock from thread %p",          \
  87         g_thread_self());                                               \
  88   g_mutex_unlock(&pad->priv->event_lock);                               \
  89   GST_LOG_OBJECT (pad, "Release EVENT lock from thread %p",          \
  90         g_thread_self());                                               \
  91   } G_STMT_END
  92
  93
  94 #define PAD_WAIT_EVENT(pad)   G_STMT_START {                            \
  95   GST_LOG_OBJECT (pad, "Waiting for EVENT on thread %p",               \
  96         g_thread_self());                                               \
  97   g_cond_wait(&(((GstAggregatorPad* )pad)->priv->event_cond),       \
  98       &(pad->priv->event_lock));                                        \
  99   GST_LOG_OBJECT (pad, "DONE Waiting for EVENT on thread %p",               \
 100         g_thread_self());                                               \
 101   } G_STMT_END
 102
 103 #define PAD_BROADCAST_EVENT(pad) {                                          \
 104   GST_LOG_OBJECT (pad, "Signaling EVENT from thread %p",               \
 105         g_thread_self());                                                   \
 106   g_cond_broadcast(&(((GstAggregatorPad* )pad)->priv->event_cond)); \
 107   }
 108
 109 struct _GstAggregatorPadPrivate
 110 {
 111   gboolean pending_flush_start;
 112   gboolean pending_flush_stop;
 113   gboolean pending_eos;
 114   gboolean flushing;
 115
 116   GMutex event_lock;
 117   GCond event_cond;
 118 };
 119
 120 static gboolean
 121 _aggpad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
 122 {
 123   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 124
 125   aggpad->eos = FALSE;
 126   aggpad->priv->flushing = FALSE;
 127
 128   if (klass->flush)
 129     return klass->flush (aggpad, agg);
 130
 131   return TRUE;
 132 }
 133
 134 /*************************************
 135  * GstAggregator implementation  *
 136  *************************************/
 137 static GstElementClass *aggregator_parent_class = NULL;
 138
 139 #define MAIN_CONTEXT_LOCK(self) G_STMT_START {                       \
 140   GST_LOG_OBJECT (self, "Getting MAIN_CONTEXT_LOCK in thread %p",    \
 141         g_thread_self());                                            \
 142   g_mutex_lock(&((GstAggregator*)self)->priv->mcontext_lock);    \
 143   GST_LOG_OBJECT (self, "Got MAIN_CONTEXT_LOCK in thread %p",        \
 144         g_thread_self());                                            \
 145 } G_STMT_END
 146
 147 #define MAIN_CONTEXT_UNLOCK(self) G_STMT_START {                     \
 148   g_mutex_unlock(&((GstAggregator*)self)->priv->mcontext_lock);  \
 149   GST_LOG_OBJECT (self, "Unlocked MAIN_CONTEXT_LOCK in thread %p",   \
 150         g_thread_self());                                            \
 151 } G_STMT_END
 152
 153 struct _GstAggregatorPrivate
 154 {
 155   gint padcount;
 156
 157   GMainContext *mcontext;
 158
 159   /* Our state is >= PAUSED */
 160   gboolean running;
 161
 162   /* Ensure that when we remove all sources from the maincontext
 163    * we can not add any source, avoiding:
 164    * "g_source_attach: assertion '!SOURCE_DESTROYED (source)' failed" */
 165   GMutex mcontext_lock;
 166
 167   gboolean send_stream_start;
 168   gboolean send_segment;
 169   gboolean flush_seeking;
 170   gboolean pending_flush_start;
 171   gboolean send_eos;
 172   GstFlowReturn flow_return;
 173
 174   GstCaps *srccaps;
 175
 176   GstTagList *tags;
 177   gboolean tags_changed;
 178 };
 179
 180 typedef struct
 181 {
 182   GstEvent *event;
 183   gboolean result;
 184   gboolean flush;
 185 } EventData;
 186
 187 /**
 188  * gst_aggregator_iterate_sinkpads:
 189  * @self: The #GstAggregator
 190  * @func: The function to call.
 191  * @user_data: The data to pass to @func.
 192  *
 193  * Iterate the sinkpads of aggregator to call a function on them.
 194  *
 195  * This method guarantees that @func will be called only once for each
 196  * sink pad.
 197  */
 198 gboolean
 199 gst_aggregator_iterate_sinkpads (GstAggregator * self,
 200     GstAggregatorPadForeachFunc func, gpointer user_data)
 201 {
 202   gboolean result = FALSE;
 203   GstIterator *iter;
 204   gboolean done = FALSE;
 205   GValue item = { 0, };
 206   GList *seen_pads = NULL;
 207
 208   iter = gst_element_iterate_sink_pads (GST_ELEMENT (self));
 209
 210   if (!iter)
 211     goto no_iter;
 212
 213   while (!done) {
 214     switch (gst_iterator_next (iter, &item)) {
 215       case GST_ITERATOR_OK:
 216       {
 217         GstPad *pad;
 218
 219         pad = g_value_get_object (&item);
 220
 221         /* if already pushed, skip. FIXME, find something faster to tag pads */
 222         if (pad == NULL || g_list_find (seen_pads, pad)) {
 223           g_value_reset (&item);
 224           break;
 225         }
 226
 227         GST_LOG_OBJECT (self, "calling function on pad %s:%s",
 228             GST_DEBUG_PAD_NAME (pad));
 229         result = func (self, pad, user_data);
 230
 231         done = !result;
 232
 233         seen_pads = g_list_prepend (seen_pads, pad);
 234
 235         g_value_reset (&item);
 236         break;
 237       }
 238       case GST_ITERATOR_RESYNC:
 239         gst_iterator_resync (iter);
 240         break;
 241       case GST_ITERATOR_ERROR:
 242         GST_ERROR_OBJECT (self,
 243             "Could not iterate over internally linked pads");
 244         done = TRUE;
 245         break;
 246       case GST_ITERATOR_DONE:
 247         done = TRUE;
 248         break;
 249     }
 250   }
 251   g_value_unset (&item);
 252   gst_iterator_free (iter);
 253
 254   if (seen_pads == NULL) {
 255     GST_DEBUG_OBJECT (self, "No pad seen");
 256     return FALSE;
 257   }
 258
 259   g_list_free (seen_pads);
 260
 261 no_iter:
 262   return result;
 263 }
 264
 265 static inline gboolean
 266 _check_all_pads_with_data_or_eos (GstAggregator * self,
 267     GstAggregatorPad * aggpad)
 268 {
 269   if (aggpad->buffer || aggpad->eos) {
 270     return TRUE;
 271   }
 272
 273   GST_LOG_OBJECT (aggpad, "Not ready to be aggregated");
 274
 275   return FALSE;
 276 }
 277
 278 /**
 279  * gst_aggregator_set_src_caps:
 280  * @self: The #GstAggregator
 281  * @caps: The #GstCaps to set later on the src pad.
 282  *
 283  * Sets the caps to be used on the src pad.
 284  */
 285 void
 286 gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps)
 287 {
 288   gst_caps_replace (&self->priv->srccaps, caps);
 289 }
 290
 291 static void
 292 _reset_flow_values (GstAggregator * self)
 293 {
 294   self->priv->send_stream_start = TRUE;
 295   self->priv->send_segment = TRUE;
 296   gst_segment_init (&self->segment, GST_FORMAT_TIME);
 297 }
 298
 299 static inline void
 300 _push_mandatory_events (GstAggregator * self)
 301 {
 302   GstAggregatorPrivate *priv = self->priv;
 303
 304   if (g_atomic_int_get (&self->priv->send_stream_start)) {
 305     gchar s_id[32];
 306
 307     GST_INFO_OBJECT (self, "pushing stream start");
 308     /* stream-start (FIXME: create id based on input ids) */
 309     g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
 310     if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
 311       GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
 312     }
 313     g_atomic_int_set (&self->priv->send_stream_start, FALSE);
 314   }
 315
 316   if (self->priv->srccaps) {
 317
 318     GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
 319         self->priv->srccaps);
 320     if (!gst_pad_push_event (self->srcpad,
 321             gst_event_new_caps (self->priv->srccaps))) {
 322       GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
 323     }
 324     self->priv->srccaps = NULL;
 325   }
 326
 327   if (g_atomic_int_get (&self->priv->send_segment)) {
 328     if (!g_atomic_int_get (&self->priv->flush_seeking)) {
 329       GST_INFO_OBJECT (self, "pushing segment");
 330       gst_pad_push_event (self->srcpad, gst_event_new_segment (&self->segment));
 331       g_atomic_int_set (&self->priv->send_segment, FALSE);
 332     }
 333   }
 334
 335   if (priv->tags && priv->tags_changed) {
 336     gst_pad_push_event (self->srcpad,
 337         gst_event_new_tag (gst_tag_list_ref (priv->tags)));
 338     priv->tags_changed = FALSE;
 339   }
 340 }
 341
 342 /**
 343  * gst_aggregator_finish_buffer:
 344  * @self: The #GstAggregator
 345  * @buffer: the #GstBuffer to push.
 346  *
 347  * This method will take care of sending mandatory events before pushing
 348  * the provided buffer.
 349  */
 350 GstFlowReturn
 351 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 352 {
 353   _push_mandatory_events (self);
 354
 355   if (!g_atomic_int_get (&self->priv->flush_seeking) &&
 356       gst_pad_is_active (self->srcpad)) {
 357     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 358     return gst_pad_push (self->srcpad, buffer);
 359   } else {
 360     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 361         g_atomic_int_get (&self->priv->flush_seeking),
 362         gst_pad_is_active (self->srcpad));
 363     return GST_FLOW_OK;
 364   }
 365 }
 366
 367 static void
 368 _push_eos (GstAggregator * self)
 369 {
 370   _push_mandatory_events (self);
 371
 372   self->priv->send_eos = FALSE;
 373   gst_pad_push_event (self->srcpad, gst_event_new_eos ());
 374 }
 375
 376 static void
 377 _remove_all_sources (GstAggregator * self)
 378 {
 379   GSource *source;
 380
 381   MAIN_CONTEXT_LOCK (self);
 382   while ((source =
 383           g_main_context_find_source_by_user_data (self->priv->mcontext,
 384               self))) {
 385     g_source_destroy (source);
 386   }
 387   MAIN_CONTEXT_UNLOCK (self);
 388 }
 389
 390 static gboolean
 391 aggregate_func (GstAggregator * self)
 392 {
 393   GstAggregatorPrivate *priv = self->priv;
 394   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 395
 396   GST_LOG_OBJECT (self, "Checking aggregate");
 397   while (priv->send_eos && gst_aggregator_iterate_sinkpads (self,
 398           (GstAggregatorPadForeachFunc) _check_all_pads_with_data_or_eos,
 399           NULL) && priv->running) {
 400     GST_TRACE_OBJECT (self, "Actually aggregating!");
 401
 402     priv->flow_return = klass->aggregate (self);
 403
 404     if (priv->flow_return == GST_FLOW_EOS) {
 405       g_main_context_wakeup (self->priv->mcontext);
 406       _remove_all_sources (self);
 407       _push_eos (self);
 408     }
 409
 410     if (priv->flow_return == GST_FLOW_FLUSHING &&
 411         g_atomic_int_get (&priv->flush_seeking))
 412       priv->flow_return = GST_FLOW_OK;
 413
 414     GST_LOG_OBJECT (self, "flow return is %s",
 415         gst_flow_get_name (priv->flow_return));
 416
 417     if (priv->flow_return != GST_FLOW_OK)
 418       break;
 419   }
 420
 421   return G_SOURCE_REMOVE;
 422 }
 423
 424 static void
 425 iterate_main_context_func (GstAggregator * self)
 426 {
 427   if (self->priv->running == FALSE) {
 428     GST_DEBUG_OBJECT (self, "Not running anymore");
 429
 430     return;
 431   }
 432
 433   g_main_context_iteration (self->priv->mcontext, TRUE);
 434 }
 435
 436 static gboolean
 437 _start (GstAggregator * self)
 438 {
 439   self->priv->running = TRUE;
 440   self->priv->send_stream_start = TRUE;
 441   self->priv->send_segment = TRUE;
 442   self->priv->send_eos = TRUE;
 443   self->priv->srccaps = NULL;
 444
 445   return TRUE;
 446 }
 447
 448 static gboolean
 449 _check_pending_flush_stop (GstAggregatorPad * pad)
 450 {
 451   return (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
 452 }
 453
 454 static gboolean
 455 _stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
 456 {
 457   gboolean res = TRUE;
 458
 459   GST_INFO_OBJECT (self, "%s srcpad task",
 460       flush_start ? "Pausing" : "Stopping");
 461
 462   self->priv->running = FALSE;
 463
 464   /*  Clean the stack of GSource set on the MainContext */
 465   g_main_context_wakeup (self->priv->mcontext);
 466   _remove_all_sources (self);
 467   if (flush_start) {
 468     res = gst_pad_push_event (self->srcpad, flush_start);
 469   }
 470
 471   gst_pad_stop_task (self->srcpad);
 472
 473   return res;
 474 }
 475
 476 static void
 477 _start_srcpad_task (GstAggregator * self)
 478 {
 479   GST_INFO_OBJECT (self, "Starting srcpad task");
 480
 481   self->priv->running = TRUE;
 482   gst_pad_start_task (GST_PAD (self->srcpad),
 483       (GstTaskFunction) iterate_main_context_func, self, NULL);
 484 }
 485
 486 static inline void
 487 _add_aggregate_gsource (GstAggregator * self)
 488 {
 489   MAIN_CONTEXT_LOCK (self);
 490   g_main_context_invoke (self->priv->mcontext, (GSourceFunc) aggregate_func,
 491       self);
 492   MAIN_CONTEXT_UNLOCK (self);
 493 }
 494
 495 static GstFlowReturn
 496 _flush (GstAggregator * self)
 497 {
 498   GstFlowReturn ret = GST_FLOW_OK;
 499   GstAggregatorPrivate *priv = self->priv;
 500   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 501
 502   GST_DEBUG_OBJECT (self, "Flushing everything");
 503   g_atomic_int_set (&priv->send_segment, TRUE);
 504   g_atomic_int_set (&priv->flush_seeking, FALSE);
 505   g_atomic_int_set (&priv->tags_changed, FALSE);
 506   if (klass->flush)
 507     ret = klass->flush (self);
 508
 509   return ret;
 510 }
 511
 512 static gboolean
 513 _all_flush_stop_received (GstAggregator * self)
 514 {
 515   GList *tmp;
 516   GstAggregatorPad *tmppad;
 517
 518   GST_OBJECT_LOCK (self);
 519   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
 520     tmppad = (GstAggregatorPad *) tmp->data;
 521
 522     if (_check_pending_flush_stop (tmppad) == FALSE) {
 523       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
 524           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
 525       GST_OBJECT_UNLOCK (self);
 526       return FALSE;
 527     }
 528   }
 529   GST_OBJECT_UNLOCK (self);
 530
 531   return TRUE;
 532 }
 533
 534 /* GstAggregator vmethods default implementations */
 535 static gboolean
 536 _sink_event (GstAggregator * self, GstAggregatorPad * aggpad, GstEvent * event)
 537 {
 538   gboolean res = TRUE;
 539   GstPad *pad = GST_PAD (aggpad);
 540   GstAggregatorPrivate *priv = self->priv;
 541   GstAggregatorPadPrivate *padpriv = aggpad->priv;
 542
 543   switch (GST_EVENT_TYPE (event)) {
 544     case GST_EVENT_FLUSH_START:
 545     {
 546       GstBuffer *tmpbuf;
 547
 548       g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 549       /*  Remove pad buffer and wake up the streaming thread */
 550       tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 551       gst_buffer_replace (&tmpbuf, NULL);
 552       if (g_atomic_int_compare_and_exchange (&padpriv->pending_flush_start,
 553               TRUE, FALSE) == TRUE) {
 554         GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
 555         g_atomic_int_set (&padpriv->pending_flush_stop, TRUE);
 556       }
 557
 558       if (g_atomic_int_get (&priv->flush_seeking)) {
 559         /* If flush_seeking we forward the first FLUSH_START */
 560         if (g_atomic_int_compare_and_exchange (&priv->pending_flush_start,
 561                 TRUE, FALSE) == TRUE) {
 562
 563           GST_DEBUG_OBJECT (self, "Flushing, pausing srcpad task");
 564           priv->flow_return = GST_FLOW_OK;
 565           _stop_srcpad_task (self, event);
 566
 567           GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
 568           GST_PAD_STREAM_LOCK (self->srcpad);
 569           GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
 570           event = NULL;
 571           goto eat;
 572         }
 573       }
 574
 575       /* We forward only in one case: right after flush_seeking */
 576       goto eat;
 577     }
 578     case GST_EVENT_FLUSH_STOP:
 579     {
 580       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
 581
 582       _aggpad_flush (aggpad, self);
 583       if (g_atomic_int_get (&priv->flush_seeking)) {
 584         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
 585
 586         if (g_atomic_int_get (&priv->flush_seeking)) {
 587           if (_all_flush_stop_received (self)) {
 588             /* That means we received FLUSH_STOP/FLUSH_STOP on
 589              * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
 590             _flush (self);
 591             gst_pad_push_event (self->srcpad, event);
 592             priv->send_eos = TRUE;
 593             event = NULL;
 594             _add_aggregate_gsource (self);
 595
 596             GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
 597             GST_PAD_STREAM_UNLOCK (self->srcpad);
 598             _start_srcpad_task (self);
 599           }
 600         }
 601       }
 602
 603       /* We never forward the event */
 604       goto eat;
 605     }
 606     case GST_EVENT_EOS:
 607     {
 608       GST_DEBUG_OBJECT (aggpad, "EOS");
 609
 610       /* We still have a buffer, and we don't want the subclass to have to
 611        * check for it. Mark pending_eos, eos will be set when steal_buffer is
 612        * called
 613        */
 614       PAD_LOCK_EVENT (aggpad);
 615       if (!aggpad->buffer) {
 616         aggpad->eos = TRUE;
 617       } else {
 618         aggpad->priv->pending_eos = TRUE;
 619       }
 620       PAD_UNLOCK_EVENT (aggpad);
 621
 622       _add_aggregate_gsource (self);
 623       goto eat;
 624     }
 625     case GST_EVENT_SEGMENT:
 626     {
 627       gst_event_copy_segment (event, &aggpad->segment);
 628       PAD_UNLOCK_EVENT (aggpad);
 629
 630       goto eat;
 631     }
 632     case GST_EVENT_STREAM_START:
 633     {
 634       goto eat;
 635     }
 636     case GST_EVENT_TAG:
 637     {
 638       GstTagList *tags;
 639
 640       gst_event_parse_tag (event, &tags);
 641
 642       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
 643         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
 644         gst_event_unref (event);
 645         event = NULL;
 646         goto eat;
 647       }
 648       break;
 649     }
 650     default:
 651     {
 652       break;
 653     }
 654   }
 655
 656   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
 657   return gst_pad_event_default (pad, GST_OBJECT (self), event);
 658
 659 eat:
 660   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
 661   if (event)
 662     gst_event_unref (event);
 663
 664   return res;
 665 }
 666
 667 static gboolean
 668 _stop (GstAggregator * agg)
 669 {
 670   _reset_flow_values (agg);
 671
 672   return TRUE;
 673 }
 674
 675 /* GstElement vmethods implementations */
 676 static GstStateChangeReturn
 677 _change_state (GstElement * element, GstStateChange transition)
 678 {
 679   GstStateChangeReturn ret;
 680   GstAggregator *self = GST_AGGREGATOR (element);
 681   GstAggregatorClass *agg_class = GST_AGGREGATOR_GET_CLASS (self);
 682
 683
 684   switch (transition) {
 685     case GST_STATE_CHANGE_READY_TO_PAUSED:
 686       agg_class->start (self);
 687       break;
 688     default:
 689       break;
 690   }
 691
 692   if ((ret =
 693           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
 694               transition)) == GST_STATE_CHANGE_FAILURE)
 695     goto failure;
 696
 697
 698   switch (transition) {
 699     case GST_STATE_CHANGE_PAUSED_TO_READY:
 700       agg_class->stop (self);
 701       break;
 702     default:
 703       break;
 704   }
 705
 706   return ret;
 707
 708 failure:
 709   {
 710     GST_ERROR_OBJECT (element, "parent failed state change");
 711     return ret;
 712   }
 713 }
 714
 715 static void
 716 _release_pad (GstElement * element, GstPad * pad)
 717 {
 718   GstBuffer *tmpbuf;
 719
 720   GstAggregator *self = GST_AGGREGATOR (element);
 721   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
 722
 723   GST_INFO_OBJECT (pad, "Removing pad");
 724
 725   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 726   tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 727   gst_buffer_replace (&tmpbuf, NULL);
 728   gst_element_remove_pad (element, pad);
 729
 730   /* Something changed make sure we try to aggregate */
 731   _add_aggregate_gsource (self);
 732 }
 733
 734 static GstPad *
 735 _request_new_pad (GstElement * element,
 736     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
 737 {
 738   GstAggregator *self;
 739   GstAggregatorPad *agg_pad;
 740
 741   GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
 742   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
 743
 744   self = GST_AGGREGATOR (element);
 745
 746   if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
 747     guint serial = 0;
 748     gchar *name = NULL;
 749
 750     GST_OBJECT_LOCK (element);
 751     if (req_name == NULL || strlen (req_name) < 6
 752         || !g_str_has_prefix (req_name, "sink_")) {
 753       /* no name given when requesting the pad, use next available int */
 754       priv->padcount++;
 755     } else {
 756       /* parse serial number from requested padname */
 757       serial = g_ascii_strtoull (&req_name[5], NULL, 10);
 758       if (serial >= priv->padcount)
 759         priv->padcount = serial + 1;
 760     }
 761
 762     name = g_strdup_printf ("sink_%u", priv->padcount);
 763     agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
 764         "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
 765     g_free (name);
 766     GST_OBJECT_UNLOCK (element);
 767
 768   } else {
 769     return NULL;
 770   }
 771
 772   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
 773
 774   if (priv->running)
 775     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
 776
 777   /* add the pad to the element */
 778   gst_element_add_pad (element, GST_PAD (agg_pad));
 779
 780   return GST_PAD (agg_pad);
 781 }
 782
 783 static gboolean
 784 _src_query (GstAggregator * self, GstQuery * query)
 785 {
 786   gboolean res = TRUE;
 787
 788   switch (GST_QUERY_TYPE (query)) {
 789     case GST_QUERY_SEEKING:
 790     {
 791       GstFormat format;
 792
 793       /* don't pass it along as some (file)sink might claim it does
 794        * whereas with a collectpads in between that will not likely work */
 795       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
 796       gst_query_set_seeking (query, format, FALSE, 0, -1);
 797       res = TRUE;
 798
 799       goto discard;
 800     }
 801     default:
 802       break;
 803   }
 804
 805   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
 806
 807 discard:
 808   return res;
 809 }
 810
 811 static gboolean
 812 event_forward_func (GstPad * pad, EventData * evdata)
 813 {
 814   gboolean ret = TRUE;
 815   GstPad *peer = gst_pad_get_peer (pad);
 816   GstAggregatorPadPrivate *padpriv = GST_AGGREGATOR_PAD (pad)->priv;
 817
 818   if (peer) {
 819     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
 820     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
 821     gst_object_unref (peer);
 822   }
 823
 824   evdata->result &= ret;
 825
 826   if (ret == FALSE) {
 827     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
 828       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
 829     else
 830       GST_INFO_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
 831
 832     if (evdata->flush) {
 833       padpriv->pending_flush_start = FALSE;
 834       padpriv->pending_flush_stop = FALSE;
 835     }
 836   }
 837
 838   /* Always send to all pads */
 839   return FALSE;
 840 }
 841
 842 static gboolean
 843 _set_flush_pending (GstAggregator * self, GstAggregatorPad * pad,
 844     gpointer udata)
 845 {
 846   pad->priv->pending_flush_start = TRUE;
 847   pad->priv->pending_flush_stop = FALSE;
 848
 849   return TRUE;
 850 }
 851
 852 static gboolean
 853 _forward_event_to_all_sinkpads (GstAggregator * self, GstEvent * event,
 854     gboolean flush)
 855 {
 856   EventData evdata;
 857
 858   evdata.event = event;
 859   evdata.result = TRUE;
 860   evdata.flush = flush;
 861
 862   /* We first need to set all pads as flushing in a first pass
 863    * as flush_start flush_stop is sometimes sent synchronously
 864    * while we send the seek event */
 865   if (flush)
 866     gst_aggregator_iterate_sinkpads (self,
 867         (GstAggregatorPadForeachFunc) _set_flush_pending, NULL);
 868   gst_pad_forward (self->srcpad, (GstPadForwardFunction) event_forward_func,
 869       &evdata);
 870
 871   gst_event_unref (event);
 872
 873   return evdata.result;
 874 }
 875
 876 static gboolean
 877 _do_seek (GstAggregator * self, GstEvent * event)
 878 {
 879   gdouble rate;
 880   GstFormat fmt;
 881   GstSeekFlags flags;
 882   GstSeekType start_type, stop_type;
 883   gint64 start, stop;
 884   gboolean flush;
 885   gboolean res;
 886   GstAggregatorPrivate *priv = self->priv;
 887
 888   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
 889       &start, &stop_type, &stop);
 890
 891   GST_INFO_OBJECT (self, "starting SEEK");
 892
 893   flush = flags & GST_SEEK_FLAG_FLUSH;
 894
 895   if (flush) {
 896     g_atomic_int_set (&priv->pending_flush_start, TRUE);
 897     g_atomic_int_set (&priv->flush_seeking, TRUE);
 898   }
 899
 900   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
 901       stop_type, stop, NULL);
 902
 903   /* forward the seek upstream */
 904   res = _forward_event_to_all_sinkpads (self, event, flush);
 905   event = NULL;
 906
 907   if (!res) {
 908     g_atomic_int_set (&priv->flush_seeking, FALSE);
 909     g_atomic_int_set (&priv->pending_flush_start, FALSE);
 910   }
 911
 912   GST_INFO_OBJECT (self, "seek done, result: %d", res);
 913
 914   return res;
 915 }
 916
 917 static gboolean
 918 _src_event (GstAggregator * self, GstEvent * event)
 919 {
 920   gboolean res = TRUE;
 921
 922   switch (GST_EVENT_TYPE (event)) {
 923     case GST_EVENT_SEEK:
 924     {
 925       res = _do_seek (self, event);
 926       event = NULL;
 927       goto done;
 928     }
 929     case GST_EVENT_NAVIGATION:
 930     {
 931       /* navigation is rather pointless. */
 932       res = FALSE;
 933       gst_event_unref (event);
 934       goto done;
 935     }
 936     default:
 937     {
 938       break;
 939     }
 940   }
 941
 942   return _forward_event_to_all_sinkpads (self, event, FALSE);
 943
 944 done:
 945   return res;
 946 }
 947
 948 static gboolean
 949 src_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
 950 {
 951   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 952
 953   return klass->src_event (GST_AGGREGATOR (parent), event);
 954 }
 955
 956 static gboolean
 957 src_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
 958 {
 959   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 960
 961   return klass->src_query (GST_AGGREGATOR (parent), query);
 962 }
 963
 964 static gboolean
 965 src_activate_mode (GstPad * pad,
 966     GstObject * parent, GstPadMode mode, gboolean active)
 967 {
 968   GstAggregator *self = GST_AGGREGATOR (parent);
 969   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 970
 971   if (klass->src_activate) {
 972     if (klass->src_activate (self, mode, active) == FALSE) {
 973       return FALSE;
 974     }
 975   }
 976
 977   if (active == TRUE) {
 978     switch (mode) {
 979       case GST_PAD_MODE_PUSH:
 980       {
 981         GST_INFO_OBJECT (pad, "Activating pad!");
 982         _start_srcpad_task (self);
 983         return TRUE;
 984       }
 985       default:
 986       {
 987         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
 988         return FALSE;
 989       }
 990     }
 991   }
 992
 993   /* deactivating */
 994   GST_INFO_OBJECT (self, "Deactivating srcpad");
 995   _stop_srcpad_task (self, FALSE);
 996
 997   return TRUE;
 998 }
 999
1000 static gboolean
1001 _sink_query (GstAggregator * self, GstAggregatorPad * aggpad, GstQuery * query)
1002 {
1003   GstPad *pad = GST_PAD (aggpad);
1004
1005   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1006 }
1007
1008 /* GObject vmethods implementations */
1009 static void
1010 gst_aggregator_class_init (GstAggregatorClass * klass)
1011 {
1012   GstElementClass *gstelement_class = (GstElementClass *) klass;
1013
1014   aggregator_parent_class = g_type_class_peek_parent (klass);
1015   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1016
1017   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1018       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1019
1020   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1021   klass->start = _start;
1022   klass->stop = _stop;
1023
1024   klass->sink_event = _sink_event;
1025   klass->sink_query = _sink_query;
1026
1027   klass->src_event = _src_event;
1028   klass->src_query = _src_query;
1029
1030   gstelement_class->request_new_pad = GST_DEBUG_FUNCPTR (_request_new_pad);
1031   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (_release_pad);
1032   gstelement_class->change_state = GST_DEBUG_FUNCPTR (_change_state);
1033 }
1034
1035 static void
1036 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1037 {
1038   GstPadTemplate *pad_template;
1039   GstAggregatorPrivate *priv;
1040
1041   g_return_if_fail (klass->aggregate != NULL);
1042
1043   self->priv =
1044       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1045       GstAggregatorPrivate);
1046
1047   priv = self->priv;
1048
1049   pad_template =
1050       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1051   g_return_if_fail (pad_template != NULL);
1052
1053   priv->padcount = -1;
1054   priv->tags_changed = FALSE;
1055   _reset_flow_values (self);
1056
1057   priv->mcontext = g_main_context_new ();
1058   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1059
1060   gst_pad_set_event_function (self->srcpad,
1061       GST_DEBUG_FUNCPTR ((GstPadEventFunction) src_event_func));
1062   gst_pad_set_query_function (self->srcpad,
1063       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) src_query_func));
1064   gst_pad_set_activatemode_function (self->srcpad,
1065       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) src_activate_mode));
1066
1067   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1068 }
1069
1070 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1071  * method to get to the padtemplates */
1072 GType
1073 gst_aggregator_get_type (void)
1074 {
1075   static volatile gsize type = 0;
1076
1077   if (g_once_init_enter (&type)) {
1078     GType _type;
1079     static const GTypeInfo info = {
1080       sizeof (GstAggregatorClass),
1081       NULL,
1082       NULL,
1083       (GClassInitFunc) gst_aggregator_class_init,
1084       NULL,
1085       NULL,
1086       sizeof (GstAggregator),
1087       0,
1088       (GInstanceInitFunc) gst_aggregator_init,
1089     };
1090
1091     _type = g_type_register_static (GST_TYPE_ELEMENT,
1092         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1093     g_once_init_leave (&type, _type);
1094   }
1095   return type;
1096 }
1097
1098 static GstFlowReturn
1099 _chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1100 {
1101   GstBuffer *actual_buf = buffer;
1102   GstAggregator *self = GST_AGGREGATOR (object);
1103   GstAggregatorPrivate *priv = self->priv;
1104   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1105   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1106
1107   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1108
1109   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1110     goto flushing;
1111
1112   if (g_atomic_int_get (&aggpad->priv->pending_eos) == TRUE)
1113     goto eos;
1114
1115   PAD_LOCK_EVENT (aggpad);
1116   if (aggpad->buffer) {
1117     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1118     PAD_WAIT_EVENT (aggpad);
1119   }
1120   PAD_UNLOCK_EVENT (aggpad);
1121
1122   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1123     goto flushing;
1124
1125
1126   if (aggclass->clip) {
1127     aggclass->clip (self, aggpad, buffer, &actual_buf);
1128   }
1129
1130   PAD_LOCK_EVENT (aggpad);
1131   if (aggpad->buffer)
1132     gst_buffer_unref (aggpad->buffer);
1133   aggpad->buffer = actual_buf;
1134   PAD_UNLOCK_EVENT (aggpad);
1135
1136   _add_aggregate_gsource (self);
1137
1138   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1139
1140   return priv->flow_return;
1141
1142 flushing:
1143
1144   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1145
1146   return GST_FLOW_FLUSHING;
1147
1148 eos:
1149
1150   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1151
1152   return GST_FLOW_EOS;
1153 }
1154
1155 static gboolean
1156 pad_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
1157 {
1158   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1159
1160   return klass->sink_query (GST_AGGREGATOR (parent),
1161       GST_AGGREGATOR_PAD (pad), query);
1162 }
1163
1164 static gboolean
1165 pad_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
1166 {
1167   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1168
1169   return klass->sink_event (GST_AGGREGATOR (parent),
1170       GST_AGGREGATOR_PAD (pad), event);
1171 }
1172
1173 static gboolean
1174 pad_activate_mode_func (GstPad * pad,
1175     GstObject * parent, GstPadMode mode, gboolean active)
1176 {
1177   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1178
1179   if (active == FALSE) {
1180     PAD_LOCK_EVENT (aggpad);
1181     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1182     gst_buffer_replace (&aggpad->buffer, NULL);
1183     PAD_BROADCAST_EVENT (aggpad);
1184     PAD_UNLOCK_EVENT (aggpad);
1185   } else {
1186     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1187     PAD_LOCK_EVENT (aggpad);
1188     PAD_BROADCAST_EVENT (aggpad);
1189     PAD_UNLOCK_EVENT (aggpad);
1190   }
1191
1192   return TRUE;
1193 }
1194
1195 /***********************************
1196  * GstAggregatorPad implementation  *
1197  ************************************/
1198 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1199
1200 static void
1201 _pad_constructed (GObject * object)
1202 {
1203   GstPad *pad = GST_PAD (object);
1204
1205   gst_pad_set_chain_function (pad,
1206       GST_DEBUG_FUNCPTR ((GstPadChainFunction) _chain));
1207   gst_pad_set_event_function (pad,
1208       GST_DEBUG_FUNCPTR ((GstPadEventFunction) pad_event_func));
1209   gst_pad_set_query_function (pad,
1210       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) pad_query_func));
1211   gst_pad_set_activatemode_function (pad,
1212       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) pad_activate_mode_func));
1213 }
1214
1215 static void
1216 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
1217 {
1218   GObjectClass *gobject_class = (GObjectClass *) klass;
1219
1220   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
1221
1222   gobject_class->constructed = GST_DEBUG_FUNCPTR (_pad_constructed);
1223 }
1224
1225 static void
1226 gst_aggregator_pad_init (GstAggregatorPad * pad)
1227 {
1228   pad->priv =
1229       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
1230       GstAggregatorPadPrivate);
1231
1232   pad->buffer = NULL;
1233   g_mutex_init (&pad->priv->event_lock);
1234   g_cond_init (&pad->priv->event_cond);
1235
1236 }
1237
1238 /**
1239  * gst_aggregator_pad_steal_buffer:
1240  * @pad: the pad to get buffer from
1241  *
1242  * Steal the ref to the buffer currently queued in @pad.
1243  *
1244  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
1245  *   queued. You should unref the buffer after usage.
1246  */
1247 GstBuffer *
1248 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
1249 {
1250   GstBuffer *buffer = NULL;
1251
1252   PAD_LOCK_EVENT (pad);
1253   if (pad->buffer) {
1254     GST_TRACE_OBJECT (pad, "Consuming buffer");
1255     buffer = pad->buffer;
1256     pad->buffer = NULL;
1257     if (pad->priv->pending_eos) {
1258       pad->priv->pending_eos = FALSE;
1259       pad->eos = TRUE;
1260     }
1261     PAD_BROADCAST_EVENT (pad);
1262     GST_DEBUG_OBJECT (pad, "Consummed: %" GST_PTR_FORMAT, buffer);
1263   }
1264   PAD_UNLOCK_EVENT (pad);
1265
1266   return buffer;
1267 }
1268
1269 /**
1270  * gst_aggregator_pad_get_buffer:
1271  * @pad: the pad to get buffer from
1272  *
1273  * Returns: (transfer full): A reference to the buffer in @pad or
1274  * NULL if no buffer was queued. You should unref the buffer after
1275  * usage.
1276  */
1277 GstBuffer *
1278 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
1279 {
1280   GstBuffer *buffer = NULL;
1281
1282   PAD_LOCK_EVENT (pad);
1283   if (pad->buffer)
1284     buffer = gst_buffer_ref (pad->buffer);
1285   PAD_UNLOCK_EVENT (pad);
1286
1287   return buffer;
1288 }
1289
1290 /**
1291  * gst_aggregator_merge_tags:
1292  * @self: a #GstAggregator
1293  * @tags: a #GstTagList to merge
1294  * @mode: the #GstTagMergeMode to use
1295  *
1296  * Adds tags to so-called pending tags, which will be processed
1297  * before pushing out data downstream.
1298  *
1299  * Note that this is provided for convenience, and the subclass is
1300  * not required to use this and can still do tag handling on its own.
1301  *
1302  * MT safe.
1303  */
1304 void
1305 gst_aggregator_merge_tags (GstAggregator * self,
1306     const GstTagList * tags, GstTagMergeMode mode)
1307 {
1308   GstTagList *otags;
1309
1310   g_return_if_fail (GST_IS_AGGREGATOR (self));
1311   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
1312
1313   /* FIXME Check if we can use OBJECT lock here! */
1314   GST_OBJECT_LOCK (self);
1315   if (tags)
1316     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
1317   otags = self->priv->tags;
1318   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
1319   if (otags)
1320     gst_tag_list_unref (otags);
1321   self->priv->tags_changed = TRUE;
1322   GST_OBJECT_UNLOCK (self);
1323 }