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     gst_caps_unref (self->priv->srccaps);
 325     self->priv->srccaps = NULL;
 326   }
 327
 328   if (g_atomic_int_get (&self->priv->send_segment)) {
 329     if (!g_atomic_int_get (&self->priv->flush_seeking)) {
 330       GST_INFO_OBJECT (self, "pushing segment");
 331       gst_pad_push_event (self->srcpad, gst_event_new_segment (&self->segment));
 332       g_atomic_int_set (&self->priv->send_segment, FALSE);
 333     }
 334   }
 335
 336   if (priv->tags && priv->tags_changed) {
 337     gst_pad_push_event (self->srcpad,
 338         gst_event_new_tag (gst_tag_list_ref (priv->tags)));
 339     priv->tags_changed = FALSE;
 340   }
 341 }
 342
 343 /**
 344  * gst_aggregator_finish_buffer:
 345  * @self: The #GstAggregator
 346  * @buffer: the #GstBuffer to push.
 347  *
 348  * This method will take care of sending mandatory events before pushing
 349  * the provided buffer.
 350  */
 351 GstFlowReturn
 352 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 353 {
 354   _push_mandatory_events (self);
 355
 356   if (!g_atomic_int_get (&self->priv->flush_seeking) &&
 357       gst_pad_is_active (self->srcpad)) {
 358     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 359     return gst_pad_push (self->srcpad, buffer);
 360   } else {
 361     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 362         g_atomic_int_get (&self->priv->flush_seeking),
 363         gst_pad_is_active (self->srcpad));
 364     return GST_FLOW_OK;
 365   }
 366 }
 367
 368 static void
 369 _push_eos (GstAggregator * self)
 370 {
 371   _push_mandatory_events (self);
 372
 373   self->priv->send_eos = FALSE;
 374   gst_pad_push_event (self->srcpad, gst_event_new_eos ());
 375 }
 376
 377 static void
 378 _remove_all_sources (GstAggregator * self)
 379 {
 380   GSource *source;
 381
 382   MAIN_CONTEXT_LOCK (self);
 383   while ((source =
 384           g_main_context_find_source_by_user_data (self->priv->mcontext,
 385               self))) {
 386     g_source_destroy (source);
 387   }
 388   MAIN_CONTEXT_UNLOCK (self);
 389 }
 390
 391 static gboolean
 392 aggregate_func (GstAggregator * self)
 393 {
 394   GstAggregatorPrivate *priv = self->priv;
 395   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 396
 397   GST_LOG_OBJECT (self, "Checking aggregate");
 398   while (priv->send_eos && gst_aggregator_iterate_sinkpads (self,
 399           (GstAggregatorPadForeachFunc) _check_all_pads_with_data_or_eos,
 400           NULL) && priv->running) {
 401     GST_TRACE_OBJECT (self, "Actually aggregating!");
 402
 403     priv->flow_return = klass->aggregate (self);
 404
 405     if (priv->flow_return == GST_FLOW_EOS) {
 406       g_main_context_wakeup (self->priv->mcontext);
 407       _remove_all_sources (self);
 408       _push_eos (self);
 409     }
 410
 411     if (priv->flow_return == GST_FLOW_FLUSHING &&
 412         g_atomic_int_get (&priv->flush_seeking))
 413       priv->flow_return = GST_FLOW_OK;
 414
 415     GST_LOG_OBJECT (self, "flow return is %s",
 416         gst_flow_get_name (priv->flow_return));
 417
 418     if (priv->flow_return != GST_FLOW_OK)
 419       break;
 420   }
 421
 422   return G_SOURCE_REMOVE;
 423 }
 424
 425 static void
 426 iterate_main_context_func (GstAggregator * self)
 427 {
 428   if (self->priv->running == FALSE) {
 429     GST_DEBUG_OBJECT (self, "Not running anymore");
 430
 431     return;
 432   }
 433
 434   g_main_context_iteration (self->priv->mcontext, TRUE);
 435 }
 436
 437 static gboolean
 438 _start (GstAggregator * self)
 439 {
 440   self->priv->running = TRUE;
 441   self->priv->send_stream_start = TRUE;
 442   self->priv->send_segment = TRUE;
 443   self->priv->send_eos = TRUE;
 444   self->priv->srccaps = NULL;
 445
 446   return TRUE;
 447 }
 448
 449 static gboolean
 450 _check_pending_flush_stop (GstAggregatorPad * pad)
 451 {
 452   return (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
 453 }
 454
 455 static gboolean
 456 _stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
 457 {
 458   gboolean res = TRUE;
 459
 460   GST_INFO_OBJECT (self, "%s srcpad task",
 461       flush_start ? "Pausing" : "Stopping");
 462
 463   self->priv->running = FALSE;
 464
 465   /*  Clean the stack of GSource set on the MainContext */
 466   g_main_context_wakeup (self->priv->mcontext);
 467   _remove_all_sources (self);
 468   if (flush_start) {
 469     res = gst_pad_push_event (self->srcpad, flush_start);
 470   }
 471
 472   gst_pad_stop_task (self->srcpad);
 473
 474   return res;
 475 }
 476
 477 static void
 478 _start_srcpad_task (GstAggregator * self)
 479 {
 480   GST_INFO_OBJECT (self, "Starting srcpad task");
 481
 482   self->priv->running = TRUE;
 483   gst_pad_start_task (GST_PAD (self->srcpad),
 484       (GstTaskFunction) iterate_main_context_func, self, NULL);
 485 }
 486
 487 static inline void
 488 _add_aggregate_gsource (GstAggregator * self)
 489 {
 490   MAIN_CONTEXT_LOCK (self);
 491   g_main_context_invoke (self->priv->mcontext, (GSourceFunc) aggregate_func,
 492       self);
 493   MAIN_CONTEXT_UNLOCK (self);
 494 }
 495
 496 static GstFlowReturn
 497 _flush (GstAggregator * self)
 498 {
 499   GstFlowReturn ret = GST_FLOW_OK;
 500   GstAggregatorPrivate *priv = self->priv;
 501   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 502
 503   GST_DEBUG_OBJECT (self, "Flushing everything");
 504   g_atomic_int_set (&priv->send_segment, TRUE);
 505   g_atomic_int_set (&priv->flush_seeking, FALSE);
 506   g_atomic_int_set (&priv->tags_changed, FALSE);
 507   if (klass->flush)
 508     ret = klass->flush (self);
 509
 510   return ret;
 511 }
 512
 513 static gboolean
 514 _all_flush_stop_received (GstAggregator * self)
 515 {
 516   GList *tmp;
 517   GstAggregatorPad *tmppad;
 518
 519   GST_OBJECT_LOCK (self);
 520   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
 521     tmppad = (GstAggregatorPad *) tmp->data;
 522
 523     if (_check_pending_flush_stop (tmppad) == FALSE) {
 524       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
 525           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
 526       GST_OBJECT_UNLOCK (self);
 527       return FALSE;
 528     }
 529   }
 530   GST_OBJECT_UNLOCK (self);
 531
 532   return TRUE;
 533 }
 534
 535 /* GstAggregator vmethods default implementations */
 536 static gboolean
 537 _sink_event (GstAggregator * self, GstAggregatorPad * aggpad, GstEvent * event)
 538 {
 539   gboolean res = TRUE;
 540   GstPad *pad = GST_PAD (aggpad);
 541   GstAggregatorPrivate *priv = self->priv;
 542   GstAggregatorPadPrivate *padpriv = aggpad->priv;
 543
 544   switch (GST_EVENT_TYPE (event)) {
 545     case GST_EVENT_FLUSH_START:
 546     {
 547       GstBuffer *tmpbuf;
 548
 549       g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 550       /*  Remove pad buffer and wake up the streaming thread */
 551       tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 552       gst_buffer_replace (&tmpbuf, NULL);
 553       if (g_atomic_int_compare_and_exchange (&padpriv->pending_flush_start,
 554               TRUE, FALSE) == TRUE) {
 555         GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
 556         g_atomic_int_set (&padpriv->pending_flush_stop, TRUE);
 557       }
 558
 559       if (g_atomic_int_get (&priv->flush_seeking)) {
 560         /* If flush_seeking we forward the first FLUSH_START */
 561         if (g_atomic_int_compare_and_exchange (&priv->pending_flush_start,
 562                 TRUE, FALSE) == TRUE) {
 563
 564           GST_DEBUG_OBJECT (self, "Flushing, pausing srcpad task");
 565           priv->flow_return = GST_FLOW_OK;
 566           _stop_srcpad_task (self, event);
 567
 568           GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
 569           GST_PAD_STREAM_LOCK (self->srcpad);
 570           GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
 571           event = NULL;
 572           goto eat;
 573         }
 574       }
 575
 576       /* We forward only in one case: right after flush_seeking */
 577       goto eat;
 578     }
 579     case GST_EVENT_FLUSH_STOP:
 580     {
 581       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
 582
 583       _aggpad_flush (aggpad, self);
 584       if (g_atomic_int_get (&priv->flush_seeking)) {
 585         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
 586
 587         if (g_atomic_int_get (&priv->flush_seeking)) {
 588           if (_all_flush_stop_received (self)) {
 589             /* That means we received FLUSH_STOP/FLUSH_STOP on
 590              * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
 591             _flush (self);
 592             gst_pad_push_event (self->srcpad, event);
 593             priv->send_eos = TRUE;
 594             event = NULL;
 595             _add_aggregate_gsource (self);
 596
 597             GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
 598             GST_PAD_STREAM_UNLOCK (self->srcpad);
 599             _start_srcpad_task (self);
 600           }
 601         }
 602       }
 603
 604       /* We never forward the event */
 605       goto eat;
 606     }
 607     case GST_EVENT_EOS:
 608     {
 609       GST_DEBUG_OBJECT (aggpad, "EOS");
 610
 611       /* We still have a buffer, and we don't want the subclass to have to
 612        * check for it. Mark pending_eos, eos will be set when steal_buffer is
 613        * called
 614        */
 615       PAD_LOCK_EVENT (aggpad);
 616       if (!aggpad->buffer) {
 617         aggpad->eos = TRUE;
 618       } else {
 619         aggpad->priv->pending_eos = TRUE;
 620       }
 621       PAD_UNLOCK_EVENT (aggpad);
 622
 623       _add_aggregate_gsource (self);
 624       goto eat;
 625     }
 626     case GST_EVENT_SEGMENT:
 627     {
 628       gst_event_copy_segment (event, &aggpad->segment);
 629       PAD_UNLOCK_EVENT (aggpad);
 630
 631       goto eat;
 632     }
 633     case GST_EVENT_STREAM_START:
 634     {
 635       goto eat;
 636     }
 637     case GST_EVENT_TAG:
 638     {
 639       GstTagList *tags;
 640
 641       gst_event_parse_tag (event, &tags);
 642
 643       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
 644         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
 645         gst_event_unref (event);
 646         event = NULL;
 647         goto eat;
 648       }
 649       break;
 650     }
 651     default:
 652     {
 653       break;
 654     }
 655   }
 656
 657   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
 658   return gst_pad_event_default (pad, GST_OBJECT (self), event);
 659
 660 eat:
 661   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
 662   if (event)
 663     gst_event_unref (event);
 664
 665   return res;
 666 }
 667
 668 static gboolean
 669 _stop (GstAggregator * agg)
 670 {
 671   _reset_flow_values (agg);
 672
 673   return TRUE;
 674 }
 675
 676 /* GstElement vmethods implementations */
 677 static GstStateChangeReturn
 678 _change_state (GstElement * element, GstStateChange transition)
 679 {
 680   GstStateChangeReturn ret;
 681   GstAggregator *self = GST_AGGREGATOR (element);
 682   GstAggregatorClass *agg_class = GST_AGGREGATOR_GET_CLASS (self);
 683
 684
 685   switch (transition) {
 686     case GST_STATE_CHANGE_READY_TO_PAUSED:
 687       agg_class->start (self);
 688       break;
 689     default:
 690       break;
 691   }
 692
 693   if ((ret =
 694           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
 695               transition)) == GST_STATE_CHANGE_FAILURE)
 696     goto failure;
 697
 698
 699   switch (transition) {
 700     case GST_STATE_CHANGE_PAUSED_TO_READY:
 701       agg_class->stop (self);
 702       break;
 703     default:
 704       break;
 705   }
 706
 707   return ret;
 708
 709 failure:
 710   {
 711     GST_ERROR_OBJECT (element, "parent failed state change");
 712     return ret;
 713   }
 714 }
 715
 716 static void
 717 _release_pad (GstElement * element, GstPad * pad)
 718 {
 719   GstBuffer *tmpbuf;
 720
 721   GstAggregator *self = GST_AGGREGATOR (element);
 722   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
 723
 724   GST_INFO_OBJECT (pad, "Removing pad");
 725
 726   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 727   tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 728   gst_buffer_replace (&tmpbuf, NULL);
 729   gst_element_remove_pad (element, pad);
 730
 731   /* Something changed make sure we try to aggregate */
 732   _add_aggregate_gsource (self);
 733 }
 734
 735 static GstPad *
 736 _request_new_pad (GstElement * element,
 737     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
 738 {
 739   GstAggregator *self;
 740   GstAggregatorPad *agg_pad;
 741
 742   GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
 743   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
 744
 745   self = GST_AGGREGATOR (element);
 746
 747   if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
 748     gint serial = 0;
 749     gchar *name = NULL;
 750
 751     GST_OBJECT_LOCK (element);
 752     if (req_name == NULL || strlen (req_name) < 6
 753         || !g_str_has_prefix (req_name, "sink_")) {
 754       /* no name given when requesting the pad, use next available int */
 755       priv->padcount++;
 756     } else {
 757       /* parse serial number from requested padname */
 758       serial = g_ascii_strtoull (&req_name[5], NULL, 10);
 759       if (serial >= priv->padcount)
 760         priv->padcount = serial;
 761     }
 762
 763     name = g_strdup_printf ("sink_%u", priv->padcount);
 764     agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
 765         "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
 766     g_free (name);
 767     GST_OBJECT_UNLOCK (element);
 768
 769   } else {
 770     return NULL;
 771   }
 772
 773   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
 774
 775   if (priv->running)
 776     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
 777
 778   /* add the pad to the element */
 779   gst_element_add_pad (element, GST_PAD (agg_pad));
 780
 781   return GST_PAD (agg_pad);
 782 }
 783
 784 static gboolean
 785 _src_query (GstAggregator * self, GstQuery * query)
 786 {
 787   gboolean res = TRUE;
 788
 789   switch (GST_QUERY_TYPE (query)) {
 790     case GST_QUERY_SEEKING:
 791     {
 792       GstFormat format;
 793
 794       /* don't pass it along as some (file)sink might claim it does
 795        * whereas with a collectpads in between that will not likely work */
 796       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
 797       gst_query_set_seeking (query, format, FALSE, 0, -1);
 798       res = TRUE;
 799
 800       goto discard;
 801     }
 802     default:
 803       break;
 804   }
 805
 806   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
 807
 808 discard:
 809   return res;
 810 }
 811
 812 static gboolean
 813 event_forward_func (GstPad * pad, EventData * evdata)
 814 {
 815   gboolean ret = TRUE;
 816   GstPad *peer = gst_pad_get_peer (pad);
 817   GstAggregatorPadPrivate *padpriv = GST_AGGREGATOR_PAD (pad)->priv;
 818
 819   if (peer) {
 820     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
 821     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
 822     gst_object_unref (peer);
 823   }
 824
 825   evdata->result &= ret;
 826
 827   if (ret == FALSE) {
 828     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
 829       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
 830     else
 831       GST_INFO_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
 832
 833     if (evdata->flush) {
 834       padpriv->pending_flush_start = FALSE;
 835       padpriv->pending_flush_stop = FALSE;
 836     }
 837   }
 838
 839   /* Always send to all pads */
 840   return FALSE;
 841 }
 842
 843 static gboolean
 844 _set_flush_pending (GstAggregator * self, GstAggregatorPad * pad,
 845     gpointer udata)
 846 {
 847   pad->priv->pending_flush_start = TRUE;
 848   pad->priv->pending_flush_stop = FALSE;
 849
 850   return TRUE;
 851 }
 852
 853 static gboolean
 854 _forward_event_to_all_sinkpads (GstAggregator * self, GstEvent * event,
 855     gboolean flush)
 856 {
 857   EventData evdata;
 858
 859   evdata.event = event;
 860   evdata.result = TRUE;
 861   evdata.flush = flush;
 862
 863   /* We first need to set all pads as flushing in a first pass
 864    * as flush_start flush_stop is sometimes sent synchronously
 865    * while we send the seek event */
 866   if (flush)
 867     gst_aggregator_iterate_sinkpads (self,
 868         (GstAggregatorPadForeachFunc) _set_flush_pending, NULL);
 869   gst_pad_forward (self->srcpad, (GstPadForwardFunction) event_forward_func,
 870       &evdata);
 871
 872   gst_event_unref (event);
 873
 874   return evdata.result;
 875 }
 876
 877 static gboolean
 878 _do_seek (GstAggregator * self, GstEvent * event)
 879 {
 880   gdouble rate;
 881   GstFormat fmt;
 882   GstSeekFlags flags;
 883   GstSeekType start_type, stop_type;
 884   gint64 start, stop;
 885   gboolean flush;
 886   gboolean res;
 887   GstAggregatorPrivate *priv = self->priv;
 888
 889   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
 890       &start, &stop_type, &stop);
 891
 892   GST_INFO_OBJECT (self, "starting SEEK");
 893
 894   flush = flags & GST_SEEK_FLAG_FLUSH;
 895
 896   if (flush) {
 897     g_atomic_int_set (&priv->pending_flush_start, TRUE);
 898     g_atomic_int_set (&priv->flush_seeking, TRUE);
 899   }
 900
 901   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
 902       stop_type, stop, NULL);
 903
 904   /* forward the seek upstream */
 905   res = _forward_event_to_all_sinkpads (self, event, flush);
 906   event = NULL;
 907
 908   if (!res) {
 909     g_atomic_int_set (&priv->flush_seeking, FALSE);
 910     g_atomic_int_set (&priv->pending_flush_start, FALSE);
 911   }
 912
 913   GST_INFO_OBJECT (self, "seek done, result: %d", res);
 914
 915   return res;
 916 }
 917
 918 static gboolean
 919 _src_event (GstAggregator * self, GstEvent * event)
 920 {
 921   gboolean res = TRUE;
 922
 923   switch (GST_EVENT_TYPE (event)) {
 924     case GST_EVENT_SEEK:
 925     {
 926       res = _do_seek (self, event);
 927       event = NULL;
 928       goto done;
 929     }
 930     case GST_EVENT_NAVIGATION:
 931     {
 932       /* navigation is rather pointless. */
 933       res = FALSE;
 934       gst_event_unref (event);
 935       goto done;
 936     }
 937     default:
 938     {
 939       break;
 940     }
 941   }
 942
 943   return _forward_event_to_all_sinkpads (self, event, FALSE);
 944
 945 done:
 946   return res;
 947 }
 948
 949 static gboolean
 950 src_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
 951 {
 952   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 953
 954   return klass->src_event (GST_AGGREGATOR (parent), event);
 955 }
 956
 957 static gboolean
 958 src_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
 959 {
 960   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 961
 962   return klass->src_query (GST_AGGREGATOR (parent), query);
 963 }
 964
 965 static gboolean
 966 src_activate_mode (GstPad * pad,
 967     GstObject * parent, GstPadMode mode, gboolean active)
 968 {
 969   GstAggregator *self = GST_AGGREGATOR (parent);
 970   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
 971
 972   if (klass->src_activate) {
 973     if (klass->src_activate (self, mode, active) == FALSE) {
 974       return FALSE;
 975     }
 976   }
 977
 978   if (active == TRUE) {
 979     switch (mode) {
 980       case GST_PAD_MODE_PUSH:
 981       {
 982         GST_INFO_OBJECT (pad, "Activating pad!");
 983         _start_srcpad_task (self);
 984         return TRUE;
 985       }
 986       default:
 987       {
 988         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
 989         return FALSE;
 990       }
 991     }
 992   }
 993
 994   /* deactivating */
 995   GST_INFO_OBJECT (self, "Deactivating srcpad");
 996   _stop_srcpad_task (self, FALSE);
 997
 998   return TRUE;
 999 }
1000
1001 static gboolean
1002 _sink_query (GstAggregator * self, GstAggregatorPad * aggpad, GstQuery * query)
1003 {
1004   GstPad *pad = GST_PAD (aggpad);
1005
1006   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1007 }
1008
1009 static void
1010 gst_aggregator_finalize (GObject * object)
1011 {
1012   GstAggregator *self = (GstAggregator *) object;
1013
1014   g_mutex_clear (&self->priv->mcontext_lock);
1015
1016   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
1017 }
1018
1019 /* GObject vmethods implementations */
1020 static void
1021 gst_aggregator_class_init (GstAggregatorClass * klass)
1022 {
1023   GObjectClass *gobject_class = (GObjectClass *) klass;
1024   GstElementClass *gstelement_class = (GstElementClass *) klass;
1025
1026   aggregator_parent_class = g_type_class_peek_parent (klass);
1027   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1028
1029   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1030       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1031
1032   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1033   klass->start = _start;
1034   klass->stop = _stop;
1035
1036   klass->sink_event = _sink_event;
1037   klass->sink_query = _sink_query;
1038
1039   klass->src_event = _src_event;
1040   klass->src_query = _src_query;
1041
1042   gstelement_class->request_new_pad = GST_DEBUG_FUNCPTR (_request_new_pad);
1043   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (_release_pad);
1044   gstelement_class->change_state = GST_DEBUG_FUNCPTR (_change_state);
1045
1046   gobject_class->finalize = gst_aggregator_finalize;
1047 }
1048
1049 static void
1050 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1051 {
1052   GstPadTemplate *pad_template;
1053   GstAggregatorPrivate *priv;
1054
1055   g_return_if_fail (klass->aggregate != NULL);
1056
1057   self->priv =
1058       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1059       GstAggregatorPrivate);
1060
1061   priv = self->priv;
1062
1063   pad_template =
1064       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1065   g_return_if_fail (pad_template != NULL);
1066
1067   priv->padcount = -1;
1068   priv->tags_changed = FALSE;
1069   _reset_flow_values (self);
1070
1071   priv->mcontext = g_main_context_new ();
1072   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1073
1074   gst_pad_set_event_function (self->srcpad,
1075       GST_DEBUG_FUNCPTR ((GstPadEventFunction) src_event_func));
1076   gst_pad_set_query_function (self->srcpad,
1077       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) src_query_func));
1078   gst_pad_set_activatemode_function (self->srcpad,
1079       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) src_activate_mode));
1080
1081   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1082
1083   g_mutex_init (&self->priv->mcontext_lock);
1084 }
1085
1086 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1087  * method to get to the padtemplates */
1088 GType
1089 gst_aggregator_get_type (void)
1090 {
1091   static volatile gsize type = 0;
1092
1093   if (g_once_init_enter (&type)) {
1094     GType _type;
1095     static const GTypeInfo info = {
1096       sizeof (GstAggregatorClass),
1097       NULL,
1098       NULL,
1099       (GClassInitFunc) gst_aggregator_class_init,
1100       NULL,
1101       NULL,
1102       sizeof (GstAggregator),
1103       0,
1104       (GInstanceInitFunc) gst_aggregator_init,
1105     };
1106
1107     _type = g_type_register_static (GST_TYPE_ELEMENT,
1108         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1109     g_once_init_leave (&type, _type);
1110   }
1111   return type;
1112 }
1113
1114 static GstFlowReturn
1115 _chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1116 {
1117   GstBuffer *actual_buf = buffer;
1118   GstAggregator *self = GST_AGGREGATOR (object);
1119   GstAggregatorPrivate *priv = self->priv;
1120   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1121   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1122
1123   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1124
1125   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1126     goto flushing;
1127
1128   if (g_atomic_int_get (&aggpad->priv->pending_eos) == TRUE)
1129     goto eos;
1130
1131   PAD_LOCK_EVENT (aggpad);
1132   if (aggpad->buffer) {
1133     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1134     PAD_WAIT_EVENT (aggpad);
1135   }
1136   PAD_UNLOCK_EVENT (aggpad);
1137
1138   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1139     goto flushing;
1140
1141
1142   if (aggclass->clip) {
1143     aggclass->clip (self, aggpad, buffer, &actual_buf);
1144   }
1145
1146   PAD_LOCK_EVENT (aggpad);
1147   if (aggpad->buffer)
1148     gst_buffer_unref (aggpad->buffer);
1149   aggpad->buffer = actual_buf;
1150   PAD_UNLOCK_EVENT (aggpad);
1151
1152   _add_aggregate_gsource (self);
1153
1154   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1155
1156   return priv->flow_return;
1157
1158 flushing:
1159
1160   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1161
1162   return GST_FLOW_FLUSHING;
1163
1164 eos:
1165
1166   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1167
1168   return GST_FLOW_EOS;
1169 }
1170
1171 static gboolean
1172 pad_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
1173 {
1174   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1175
1176   return klass->sink_query (GST_AGGREGATOR (parent),
1177       GST_AGGREGATOR_PAD (pad), query);
1178 }
1179
1180 static gboolean
1181 pad_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
1182 {
1183   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1184
1185   return klass->sink_event (GST_AGGREGATOR (parent),
1186       GST_AGGREGATOR_PAD (pad), event);
1187 }
1188
1189 static gboolean
1190 pad_activate_mode_func (GstPad * pad,
1191     GstObject * parent, GstPadMode mode, gboolean active)
1192 {
1193   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1194
1195   if (active == FALSE) {
1196     PAD_LOCK_EVENT (aggpad);
1197     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1198     gst_buffer_replace (&aggpad->buffer, NULL);
1199     PAD_BROADCAST_EVENT (aggpad);
1200     PAD_UNLOCK_EVENT (aggpad);
1201   } else {
1202     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1203     PAD_LOCK_EVENT (aggpad);
1204     PAD_BROADCAST_EVENT (aggpad);
1205     PAD_UNLOCK_EVENT (aggpad);
1206   }
1207
1208   return TRUE;
1209 }
1210
1211 /***********************************
1212  * GstAggregatorPad implementation  *
1213  ************************************/
1214 static GstPadClass *aggregator_pad_parent_class = NULL;
1215 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1216
1217 static void
1218 _pad_constructed (GObject * object)
1219 {
1220   GstPad *pad = GST_PAD (object);
1221
1222   gst_pad_set_chain_function (pad,
1223       GST_DEBUG_FUNCPTR ((GstPadChainFunction) _chain));
1224   gst_pad_set_event_function (pad,
1225       GST_DEBUG_FUNCPTR ((GstPadEventFunction) pad_event_func));
1226   gst_pad_set_query_function (pad,
1227       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) pad_query_func));
1228   gst_pad_set_activatemode_function (pad,
1229       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) pad_activate_mode_func));
1230 }
1231
1232 static void
1233 gst_aggregator_pad_finalize (GObject * object)
1234 {
1235   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1236
1237   g_mutex_clear (&pad->priv->event_lock);
1238   g_cond_clear (&pad->priv->event_cond);
1239
1240   G_OBJECT_CLASS (aggregator_pad_parent_class)->finalize (object);
1241 }
1242
1243 static void
1244 gst_aggregator_pad_dispose (GObject * object)
1245 {
1246   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1247   GstBuffer *buf;
1248
1249   buf = gst_aggregator_pad_steal_buffer (pad);
1250   if (buf)
1251     gst_buffer_unref (buf);
1252
1253   G_OBJECT_CLASS (aggregator_pad_parent_class)->dispose (object);
1254 }
1255
1256 static void
1257 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
1258 {
1259   GObjectClass *gobject_class = (GObjectClass *) klass;
1260
1261   aggregator_pad_parent_class = g_type_class_peek_parent (klass);
1262   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
1263
1264   gobject_class->constructed = GST_DEBUG_FUNCPTR (_pad_constructed);
1265   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_aggregator_pad_finalize);
1266   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_aggregator_pad_dispose);
1267 }
1268
1269 static void
1270 gst_aggregator_pad_init (GstAggregatorPad * pad)
1271 {
1272   pad->priv =
1273       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
1274       GstAggregatorPadPrivate);
1275
1276   pad->buffer = NULL;
1277   g_mutex_init (&pad->priv->event_lock);
1278   g_cond_init (&pad->priv->event_cond);
1279
1280 }
1281
1282 /**
1283  * gst_aggregator_pad_steal_buffer:
1284  * @pad: the pad to get buffer from
1285  *
1286  * Steal the ref to the buffer currently queued in @pad.
1287  *
1288  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
1289  *   queued. You should unref the buffer after usage.
1290  */
1291 GstBuffer *
1292 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
1293 {
1294   GstBuffer *buffer = NULL;
1295
1296   PAD_LOCK_EVENT (pad);
1297   if (pad->buffer) {
1298     GST_TRACE_OBJECT (pad, "Consuming buffer");
1299     buffer = pad->buffer;
1300     pad->buffer = NULL;
1301     if (pad->priv->pending_eos) {
1302       pad->priv->pending_eos = FALSE;
1303       pad->eos = TRUE;
1304     }
1305     PAD_BROADCAST_EVENT (pad);
1306     GST_DEBUG_OBJECT (pad, "Consummed: %" GST_PTR_FORMAT, buffer);
1307   }
1308   PAD_UNLOCK_EVENT (pad);
1309
1310   return buffer;
1311 }
1312
1313 /**
1314  * gst_aggregator_pad_get_buffer:
1315  * @pad: the pad to get buffer from
1316  *
1317  * Returns: (transfer full): A reference to the buffer in @pad or
1318  * NULL if no buffer was queued. You should unref the buffer after
1319  * usage.
1320  */
1321 GstBuffer *
1322 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
1323 {
1324   GstBuffer *buffer = NULL;
1325
1326   PAD_LOCK_EVENT (pad);
1327   if (pad->buffer)
1328     buffer = gst_buffer_ref (pad->buffer);
1329   PAD_UNLOCK_EVENT (pad);
1330
1331   return buffer;
1332 }
1333
1334 /**
1335  * gst_aggregator_merge_tags:
1336  * @self: a #GstAggregator
1337  * @tags: a #GstTagList to merge
1338  * @mode: the #GstTagMergeMode to use
1339  *
1340  * Adds tags to so-called pending tags, which will be processed
1341  * before pushing out data downstream.
1342  *
1343  * Note that this is provided for convenience, and the subclass is
1344  * not required to use this and can still do tag handling on its own.
1345  *
1346  * MT safe.
1347  */
1348 void
1349 gst_aggregator_merge_tags (GstAggregator * self,
1350     const GstTagList * tags, GstTagMergeMode mode)
1351 {
1352   GstTagList *otags;
1353
1354   g_return_if_fail (GST_IS_AGGREGATOR (self));
1355   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
1356
1357   /* FIXME Check if we can use OBJECT lock here! */
1358   GST_OBJECT_LOCK (self);
1359   if (tags)
1360     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
1361   otags = self->priv->tags;
1362   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
1363   if (otags)
1364     gst_tag_list_unref (otags);
1365   self->priv->tags_changed = TRUE;
1366   GST_OBJECT_UNLOCK (self);
1367 }