libs/gst/base/gstaggregator.c

   1 /* GStreamer aggregator base class
   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. Subclasses should at least implement
  33  *    the #GstAggregatorClass.aggregate() virtual method.
  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 static void gst_aggregator_set_latency_property (GstAggregator * agg,
  73     gint64 latency);
  74 static gint64 gst_aggregator_get_latency_property (GstAggregator * agg);
  75
  76
  77 /* Locking order, locks in this element must always be taken in this order
  78  *
  79  * standard sink pad stream lock -> GST_PAD_STREAM_LOCK (aggpad)
  80  * Aggregator pad flush lock -> PAD_FLUSH_LOCK(aggpad)
  81  * standard src pad stream lock -> GST_PAD_STREAM_LOCK (srcpad)
  82  * Aggregator src lock -> SRC_LOCK(agg) w/ SRC_WAIT/BROADCAST
  83  * standard element object lock -> GST_OBJECT_LOCK(agg)
  84  * Aggregator pad lock -> PAD_LOCK (aggpad) w/ PAD_WAIT/BROADCAST_EVENT(aggpad)
  85  * standard src pad object lock -> GST_OBJECT_LOCK(srcpad)
  86  * standard sink pad object lock -> GST_OBJECT_LOCK(aggpad)
  87  */
  88
  89
  90 GST_DEBUG_CATEGORY_STATIC (aggregator_debug);
  91 #define GST_CAT_DEFAULT aggregator_debug
  92
  93 /* GstAggregatorPad definitions */
  94 #define PAD_LOCK(pad)   G_STMT_START {                                  \
  95   GST_TRACE_OBJECT (pad, "Taking PAD lock from thread %p",              \
  96         g_thread_self());                                               \
  97   g_mutex_lock(&pad->priv->lock);                                       \
  98   GST_TRACE_OBJECT (pad, "Took PAD lock from thread %p",                \
  99         g_thread_self());                                               \
 100   } G_STMT_END
 101
 102 #define PAD_UNLOCK(pad)  G_STMT_START {                                 \
 103   GST_TRACE_OBJECT (pad, "Releasing PAD lock from thread %p",           \
 104       g_thread_self());                                                 \
 105   g_mutex_unlock(&pad->priv->lock);                                     \
 106   GST_TRACE_OBJECT (pad, "Release PAD lock from thread %p",             \
 107         g_thread_self());                                               \
 108   } G_STMT_END
 109
 110
 111 #define PAD_WAIT_EVENT(pad)   G_STMT_START {                            \
 112   GST_LOG_OBJECT (pad, "Waiting for EVENT on thread %p",                \
 113         g_thread_self());                                               \
 114   g_cond_wait(&(((GstAggregatorPad* )pad)->priv->event_cond),           \
 115       (&((GstAggregatorPad*)pad)->priv->lock));                         \
 116   GST_LOG_OBJECT (pad, "DONE Waiting for EVENT on thread %p",           \
 117         g_thread_self());                                               \
 118   } G_STMT_END
 119
 120 #define PAD_BROADCAST_EVENT(pad) G_STMT_START {                        \
 121   GST_LOG_OBJECT (pad, "Signaling EVENT from thread %p",               \
 122         g_thread_self());                                              \
 123   g_cond_broadcast(&(((GstAggregatorPad* )pad)->priv->event_cond));    \
 124   } G_STMT_END
 125
 126
 127 #define PAD_FLUSH_LOCK(pad)     G_STMT_START {                          \
 128   GST_TRACE_OBJECT (pad, "Taking lock from thread %p",                  \
 129         g_thread_self());                                               \
 130   g_mutex_lock(&pad->priv->flush_lock);                                 \
 131   GST_TRACE_OBJECT (pad, "Took lock from thread %p",                    \
 132         g_thread_self());                                               \
 133   } G_STMT_END
 134
 135 #define PAD_FLUSH_UNLOCK(pad)   G_STMT_START {                          \
 136   GST_TRACE_OBJECT (pad, "Releasing lock from thread %p",               \
 137         g_thread_self());                                               \
 138   g_mutex_unlock(&pad->priv->flush_lock);                               \
 139   GST_TRACE_OBJECT (pad, "Release lock from thread %p",                 \
 140         g_thread_self());                                               \
 141   } G_STMT_END
 142
 143 #define SRC_LOCK(self)   G_STMT_START {                             \
 144   GST_TRACE_OBJECT (self, "Taking src lock from thread %p",         \
 145       g_thread_self());                                             \
 146   g_mutex_lock(&self->priv->src_lock);                              \
 147   GST_TRACE_OBJECT (self, "Took src lock from thread %p",           \
 148         g_thread_self());                                           \
 149   } G_STMT_END
 150
 151 #define SRC_UNLOCK(self)  G_STMT_START {                            \
 152   GST_TRACE_OBJECT (self, "Releasing src lock from thread %p",      \
 153         g_thread_self());                                           \
 154   g_mutex_unlock(&self->priv->src_lock);                            \
 155   GST_TRACE_OBJECT (self, "Released src lock from thread %p",       \
 156         g_thread_self());                                           \
 157   } G_STMT_END
 158
 159 #define SRC_WAIT(self) G_STMT_START {                               \
 160   GST_LOG_OBJECT (self, "Waiting for src on thread %p",             \
 161         g_thread_self());                                           \
 162   g_cond_wait(&(self->priv->src_cond), &(self->priv->src_lock));    \
 163   GST_LOG_OBJECT (self, "DONE Waiting for src on thread %p",        \
 164         g_thread_self());                                           \
 165   } G_STMT_END
 166
 167 #define SRC_BROADCAST(self) G_STMT_START {                          \
 168     GST_LOG_OBJECT (self, "Signaling src from thread %p",           \
 169         g_thread_self());                                           \
 170     if (self->priv->aggregate_id)                                   \
 171       gst_clock_id_unschedule (self->priv->aggregate_id);           \
 172     g_cond_broadcast(&(self->priv->src_cond));                      \
 173   } G_STMT_END
 174
 175 struct _GstAggregatorPadPrivate
 176 {
 177   /* To always be used atomically */
 178   gboolean flushing;
 179
 180   /* Following fields are protected by the PAD_LOCK */
 181   gboolean pending_flush_start;
 182   gboolean pending_flush_stop;
 183   gboolean pending_eos;
 184
 185   GstBuffer *buffer;
 186   gboolean eos;
 187
 188   GMutex lock;
 189   GCond event_cond;
 190   /* This lock prevents a flush start processing happening while
 191    * the chain function is also happening.
 192    */
 193   GMutex flush_lock;
 194 };
 195
 196 static gboolean
 197 gst_aggregator_pad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
 198 {
 199   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 200
 201   PAD_LOCK (aggpad);
 202   aggpad->priv->eos = FALSE;
 203   aggpad->priv->flushing = FALSE;
 204   PAD_UNLOCK (aggpad);
 205
 206   if (klass->flush)
 207     return klass->flush (aggpad, agg);
 208
 209   return TRUE;
 210 }
 211
 212 /*************************************
 213  * GstAggregator implementation  *
 214  *************************************/
 215 static GstElementClass *aggregator_parent_class = NULL;
 216
 217 /* All members are protected by the object lock unless otherwise noted */
 218
 219 struct _GstAggregatorPrivate
 220 {
 221   gint padcount;
 222
 223   /* Our state is >= PAUSED */
 224   gboolean running;             /* protected by src_lock */
 225
 226   gint seqnum;
 227   gboolean send_stream_start;   /* protected by srcpad stream lock */
 228   gboolean send_segment;
 229   gboolean flush_seeking;
 230   gboolean pending_flush_start;
 231   gboolean send_eos;            /* protected by srcpad stream lock */
 232   GstFlowReturn flow_return;
 233
 234   GstCaps *srccaps;             /* protected by the srcpad stream lock */
 235
 236   GstTagList *tags;
 237   gboolean tags_changed;
 238
 239   gboolean latency_live;        /* protected by src_lock */
 240   GstClockTime latency_min;     /* protected by src_lock */
 241   GstClockTime latency_max;     /* protected by src_lock */
 242
 243   GstClockTime sub_latency_min; /* protected by src_lock */
 244   GstClockTime sub_latency_max; /* protected by src_lock */
 245
 246   /* aggregate */
 247   GstClockID aggregate_id;      /* protected by src_lock */
 248   GMutex src_lock;
 249   GCond src_cond;
 250
 251   /* properties */
 252   gint64 latency;
 253 };
 254
 255 typedef struct
 256 {
 257   GstEvent *event;
 258   gboolean result;
 259   gboolean flush;
 260
 261   gboolean one_actually_seeked;
 262 } EventData;
 263
 264 #define DEFAULT_LATENCY        0
 265
 266 enum
 267 {
 268   PROP_0,
 269   PROP_LATENCY,
 270   PROP_LAST
 271 };
 272
 273 /**
 274  * gst_aggregator_iterate_sinkpads:
 275  * @self: The #GstAggregator
 276  * @func: (scope call): The function to call.
 277  * @user_data: (closure): The data to pass to @func.
 278  *
 279  * Iterate the sinkpads of aggregator to call a function on them.
 280  *
 281  * This method guarantees that @func will be called only once for each
 282  * sink pad.
 283  */
 284 gboolean
 285 gst_aggregator_iterate_sinkpads (GstAggregator * self,
 286     GstAggregatorPadForeachFunc func, gpointer user_data)
 287 {
 288   gboolean result = FALSE;
 289   GstIterator *iter;
 290   gboolean done = FALSE;
 291   GValue item = { 0, };
 292   GList *seen_pads = NULL;
 293
 294   iter = gst_element_iterate_sink_pads (GST_ELEMENT (self));
 295
 296   if (!iter)
 297     goto no_iter;
 298
 299   while (!done) {
 300     switch (gst_iterator_next (iter, &item)) {
 301       case GST_ITERATOR_OK:
 302       {
 303         GstAggregatorPad *pad;
 304
 305         pad = g_value_get_object (&item);
 306
 307         /* if already pushed, skip. FIXME, find something faster to tag pads */
 308         if (pad == NULL || g_list_find (seen_pads, pad)) {
 309           g_value_reset (&item);
 310           break;
 311         }
 312
 313         GST_LOG_OBJECT (pad, "calling function %s on pad",
 314             GST_DEBUG_FUNCPTR_NAME (func));
 315
 316         result = func (self, pad, user_data);
 317
 318         done = !result;
 319
 320         seen_pads = g_list_prepend (seen_pads, pad);
 321
 322         g_value_reset (&item);
 323         break;
 324       }
 325       case GST_ITERATOR_RESYNC:
 326         gst_iterator_resync (iter);
 327         break;
 328       case GST_ITERATOR_ERROR:
 329         GST_ERROR_OBJECT (self,
 330             "Could not iterate over internally linked pads");
 331         done = TRUE;
 332         break;
 333       case GST_ITERATOR_DONE:
 334         done = TRUE;
 335         break;
 336     }
 337   }
 338   g_value_unset (&item);
 339   gst_iterator_free (iter);
 340
 341   if (seen_pads == NULL) {
 342     GST_DEBUG_OBJECT (self, "No pad seen");
 343     return FALSE;
 344   }
 345
 346   g_list_free (seen_pads);
 347
 348 no_iter:
 349   return result;
 350 }
 351
 352 static gboolean
 353 gst_aggregator_check_pads_ready (GstAggregator * self)
 354 {
 355   GstAggregatorPad *pad;
 356   GList *l, *sinkpads;
 357
 358   GST_LOG_OBJECT (self, "checking pads");
 359
 360   GST_OBJECT_LOCK (self);
 361
 362   sinkpads = GST_ELEMENT_CAST (self)->sinkpads;
 363   if (sinkpads == NULL)
 364     goto no_sinkpads;
 365
 366   for (l = sinkpads; l != NULL; l = l->next) {
 367     pad = l->data;
 368
 369     PAD_LOCK (pad);
 370     if (pad->priv->buffer == NULL && !pad->priv->eos) {
 371       PAD_UNLOCK (pad);
 372       goto pad_not_ready;
 373     }
 374     PAD_UNLOCK (pad);
 375
 376   }
 377
 378   GST_OBJECT_UNLOCK (self);
 379   GST_LOG_OBJECT (self, "pads are ready");
 380   return TRUE;
 381
 382 no_sinkpads:
 383   {
 384     GST_LOG_OBJECT (self, "pads not ready: no sink pads");
 385     GST_OBJECT_UNLOCK (self);
 386     return FALSE;
 387   }
 388 pad_not_ready:
 389   {
 390     GST_LOG_OBJECT (pad, "pad not ready to be aggregated yet");
 391     GST_OBJECT_UNLOCK (self);
 392     return FALSE;
 393   }
 394 }
 395
 396 static void
 397 gst_aggregator_reset_flow_values (GstAggregator * self)
 398 {
 399   GST_OBJECT_LOCK (self);
 400   self->priv->flow_return = GST_FLOW_FLUSHING;
 401   self->priv->send_stream_start = TRUE;
 402   self->priv->send_segment = TRUE;
 403   gst_segment_init (&self->segment, GST_FORMAT_TIME);
 404   GST_OBJECT_UNLOCK (self);
 405 }
 406
 407 static inline void
 408 gst_aggregator_push_mandatory_events (GstAggregator * self)
 409 {
 410   GstAggregatorPrivate *priv = self->priv;
 411   GstEvent *segment = NULL;
 412   GstEvent *tags = NULL;
 413
 414   if (self->priv->send_stream_start) {
 415     gchar s_id[32];
 416
 417     GST_INFO_OBJECT (self, "pushing stream start");
 418     /* stream-start (FIXME: create id based on input ids) */
 419     g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
 420     if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
 421       GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
 422     }
 423     self->priv->send_stream_start = FALSE;
 424   }
 425
 426   if (self->priv->srccaps) {
 427
 428     GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
 429         self->priv->srccaps);
 430     if (!gst_pad_push_event (self->srcpad,
 431             gst_event_new_caps (self->priv->srccaps))) {
 432       GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
 433     }
 434     gst_caps_unref (self->priv->srccaps);
 435     self->priv->srccaps = NULL;
 436   }
 437
 438   GST_OBJECT_LOCK (self);
 439   if (self->priv->send_segment && !self->priv->flush_seeking) {
 440     segment = gst_event_new_segment (&self->segment);
 441
 442     if (!self->priv->seqnum)
 443       self->priv->seqnum = gst_event_get_seqnum (segment);
 444     else
 445       gst_event_set_seqnum (segment, self->priv->seqnum);
 446     self->priv->send_segment = FALSE;
 447
 448     GST_DEBUG_OBJECT (self, "pushing segment %" GST_PTR_FORMAT, segment);
 449   }
 450
 451   if (priv->tags && priv->tags_changed && !self->priv->flush_seeking) {
 452     tags = gst_event_new_tag (gst_tag_list_ref (priv->tags));
 453     priv->tags_changed = FALSE;
 454   }
 455   GST_OBJECT_UNLOCK (self);
 456
 457   if (segment)
 458     gst_pad_push_event (self->srcpad, segment);
 459   if (tags)
 460     gst_pad_push_event (self->srcpad, tags);
 461
 462 }
 463
 464 /**
 465  * gst_aggregator_set_src_caps:
 466  * @self: The #GstAggregator
 467  * @caps: The #GstCaps to set on the src pad.
 468  *
 469  * Sets the caps to be used on the src pad.
 470  */
 471 void
 472 gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps)
 473 {
 474   GST_PAD_STREAM_LOCK (self->srcpad);
 475   gst_caps_replace (&self->priv->srccaps, caps);
 476   gst_aggregator_push_mandatory_events (self);
 477   GST_PAD_STREAM_UNLOCK (self->srcpad);
 478 }
 479
 480 /**
 481  * gst_aggregator_finish_buffer:
 482  * @self: The #GstAggregator
 483  * @buffer: (transfer full): the #GstBuffer to push.
 484  *
 485  * This method will push the provided output buffer downstream. If needed,
 486  * mandatory events such as stream-start, caps, and segment events will be
 487  * sent before pushing the buffer.
 488  */
 489 GstFlowReturn
 490 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 491 {
 492   gst_aggregator_push_mandatory_events (self);
 493
 494   GST_OBJECT_LOCK (self);
 495   if (!self->priv->flush_seeking && gst_pad_is_active (self->srcpad)) {
 496     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 497     GST_OBJECT_UNLOCK (self);
 498     return gst_pad_push (self->srcpad, buffer);
 499   } else {
 500     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 501         self->priv->flush_seeking, gst_pad_is_active (self->srcpad));
 502     GST_OBJECT_UNLOCK (self);
 503     gst_buffer_unref (buffer);
 504     return GST_FLOW_OK;
 505   }
 506 }
 507
 508 static void
 509 gst_aggregator_push_eos (GstAggregator * self)
 510 {
 511   GstEvent *event;
 512   gst_aggregator_push_mandatory_events (self);
 513
 514   event = gst_event_new_eos ();
 515
 516   GST_OBJECT_LOCK (self);
 517   self->priv->send_eos = FALSE;
 518   gst_event_set_seqnum (event, self->priv->seqnum);
 519   GST_OBJECT_UNLOCK (self);
 520
 521   gst_pad_push_event (self->srcpad, event);
 522 }
 523
 524 static GstClockTime
 525 gst_aggregator_get_next_time (GstAggregator * self)
 526 {
 527   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 528
 529   if (klass->get_next_time)
 530     return klass->get_next_time (self);
 531
 532   return GST_CLOCK_TIME_NONE;
 533 }
 534
 535 static gboolean
 536 gst_aggregator_wait_and_check (GstAggregator * self, gboolean * timeout)
 537 {
 538   GstClockTime latency_max, latency_min;
 539   GstClockTime start;
 540   gboolean live, res;
 541
 542   *timeout = FALSE;
 543
 544   SRC_LOCK (self);
 545
 546   gst_aggregator_get_latency_unlocked (self, &live, &latency_min, &latency_max);
 547
 548   if (gst_aggregator_check_pads_ready (self)) {
 549     GST_DEBUG_OBJECT (self, "all pads have data");
 550     SRC_UNLOCK (self);
 551
 552     return TRUE;
 553   }
 554
 555   /* Before waiting, check if we're actually still running */
 556   if (!self->priv->running || !self->priv->send_eos) {
 557     SRC_UNLOCK (self);
 558
 559     return FALSE;
 560   }
 561
 562   start = gst_aggregator_get_next_time (self);
 563
 564   if (!live || !GST_IS_CLOCK (GST_ELEMENT_CLOCK (self))
 565       || !GST_CLOCK_TIME_IS_VALID (start)) {
 566     /* We wake up here when something happened, and below
 567      * then check if we're ready now. If we return FALSE,
 568      * we will be directly called again.
 569      */
 570     SRC_WAIT (self);
 571   } else {
 572     GstClockTime base_time, time;
 573     GstClock *clock;
 574     GstClockReturn status;
 575     GstClockTimeDiff jitter;
 576
 577     GST_DEBUG_OBJECT (self, "got subclass start time: %" GST_TIME_FORMAT,
 578         GST_TIME_ARGS (start));
 579
 580     GST_OBJECT_LOCK (self);
 581     base_time = GST_ELEMENT_CAST (self)->base_time;
 582     clock = GST_ELEMENT_CLOCK (self);
 583     if (clock)
 584       gst_object_ref (clock);
 585     GST_OBJECT_UNLOCK (self);
 586
 587     time = base_time + start;
 588     time += latency_min;
 589
 590     GST_DEBUG_OBJECT (self, "possibly waiting for clock to reach %"
 591         GST_TIME_FORMAT " (base %" GST_TIME_FORMAT " start %" GST_TIME_FORMAT
 592         " latency max %" GST_TIME_FORMAT " latency min %" GST_TIME_FORMAT
 593         " current %" GST_TIME_FORMAT ")", GST_TIME_ARGS (time),
 594         GST_TIME_ARGS (GST_ELEMENT_CAST (self)->base_time),
 595         GST_TIME_ARGS (start), GST_TIME_ARGS (latency_max),
 596         GST_TIME_ARGS (latency_min),
 597         GST_TIME_ARGS (gst_clock_get_time (clock)));
 598
 599
 600     self->priv->aggregate_id = gst_clock_new_single_shot_id (clock, time);
 601     gst_object_unref (clock);
 602     SRC_UNLOCK (self);
 603
 604     jitter = 0;
 605     status = gst_clock_id_wait (self->priv->aggregate_id, &jitter);
 606
 607     SRC_LOCK (self);
 608     if (self->priv->aggregate_id) {
 609       gst_clock_id_unref (self->priv->aggregate_id);
 610       self->priv->aggregate_id = NULL;
 611     }
 612
 613     GST_DEBUG_OBJECT (self,
 614         "clock returned %d (jitter: %s%" GST_TIME_FORMAT ")",
 615         status, (jitter < 0 ? "-" : " "),
 616         GST_TIME_ARGS ((jitter < 0 ? -jitter : jitter)));
 617
 618     /* we timed out */
 619     if (status == GST_CLOCK_OK || status == GST_CLOCK_EARLY) {
 620       SRC_UNLOCK (self);
 621       *timeout = TRUE;
 622       return TRUE;
 623     }
 624   }
 625
 626   res = gst_aggregator_check_pads_ready (self);
 627   SRC_UNLOCK (self);
 628
 629   return res;
 630 }
 631
 632 static void
 633 gst_aggregator_aggregate_func (GstAggregator * self)
 634 {
 635   GstAggregatorPrivate *priv = self->priv;
 636   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 637   gboolean timeout = FALSE;
 638
 639   if (self->priv->running == FALSE) {
 640     GST_DEBUG_OBJECT (self, "Not running anymore");
 641     return;
 642   }
 643
 644   GST_LOG_OBJECT (self, "Checking aggregate");
 645   while (priv->send_eos && priv->running) {
 646     GstFlowReturn flow_return;
 647
 648     if (!gst_aggregator_wait_and_check (self, &timeout))
 649       continue;
 650
 651     GST_TRACE_OBJECT (self, "Actually aggregating!");
 652
 653     flow_return = klass->aggregate (self, timeout);
 654
 655     GST_OBJECT_LOCK (self);
 656     if (flow_return == GST_FLOW_FLUSHING && priv->flush_seeking)
 657       priv->flow_return = GST_FLOW_OK;
 658     else
 659       priv->flow_return = flow_return;
 660     GST_OBJECT_UNLOCK (self);
 661
 662     if (flow_return == GST_FLOW_EOS) {
 663       gst_aggregator_push_eos (self);
 664     }
 665
 666     GST_LOG_OBJECT (self, "flow return is %s", gst_flow_get_name (flow_return));
 667
 668     if (flow_return != GST_FLOW_OK)
 669       break;
 670   }
 671
 672   /* Pause the task here, the only ways to get here are:
 673    * 1) We're stopping, in which case the task is stopped anyway
 674    * 2) We got a flow error above, in which case it might take
 675    *    some time to forward the flow return upstream and we
 676    *    would otherwise call the task function over and over
 677    *    again without doing anything
 678    */
 679   gst_pad_pause_task (self->srcpad);
 680 }
 681
 682 static gboolean
 683 gst_aggregator_start (GstAggregator * self)
 684 {
 685   GstAggregatorClass *klass;
 686   gboolean result;
 687
 688   self->priv->running = TRUE;
 689   self->priv->send_stream_start = TRUE;
 690   self->priv->send_segment = TRUE;
 691   self->priv->send_eos = TRUE;
 692   self->priv->srccaps = NULL;
 693   self->priv->flow_return = GST_FLOW_OK;
 694
 695   klass = GST_AGGREGATOR_GET_CLASS (self);
 696
 697   if (klass->start)
 698     result = klass->start (self);
 699   else
 700     result = TRUE;
 701
 702   return result;
 703 }
 704
 705 static gboolean
 706 _check_pending_flush_stop (GstAggregatorPad * pad)
 707 {
 708   gboolean res;
 709
 710   PAD_LOCK (pad);
 711   res = (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
 712   PAD_UNLOCK (pad);
 713
 714   return res;
 715 }
 716
 717 static gboolean
 718 gst_aggregator_stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
 719 {
 720   gboolean res = TRUE;
 721
 722   GST_INFO_OBJECT (self, "%s srcpad task",
 723       flush_start ? "Pausing" : "Stopping");
 724
 725   SRC_LOCK (self);
 726   self->priv->running = FALSE;
 727   SRC_BROADCAST (self);
 728   SRC_UNLOCK (self);
 729
 730   if (flush_start) {
 731     res = gst_pad_push_event (self->srcpad, flush_start);
 732   }
 733
 734   gst_pad_stop_task (self->srcpad);
 735
 736   return res;
 737 }
 738
 739 static void
 740 gst_aggregator_start_srcpad_task (GstAggregator * self)
 741 {
 742   GST_INFO_OBJECT (self, "Starting srcpad task");
 743
 744   self->priv->running = TRUE;
 745   gst_pad_start_task (GST_PAD (self->srcpad),
 746       (GstTaskFunction) gst_aggregator_aggregate_func, self, NULL);
 747 }
 748
 749 static GstFlowReturn
 750 gst_aggregator_flush (GstAggregator * self)
 751 {
 752   GstFlowReturn ret = GST_FLOW_OK;
 753   GstAggregatorPrivate *priv = self->priv;
 754   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 755
 756   GST_DEBUG_OBJECT (self, "Flushing everything");
 757   GST_OBJECT_LOCK (self);
 758   priv->send_segment = TRUE;
 759   priv->flush_seeking = FALSE;
 760   priv->tags_changed = FALSE;
 761   GST_OBJECT_UNLOCK (self);
 762   if (klass->flush)
 763     ret = klass->flush (self);
 764
 765   return ret;
 766 }
 767
 768
 769 /* Called with GstAggregator's object lock held */
 770
 771 static gboolean
 772 gst_aggregator_all_flush_stop_received_locked (GstAggregator * self)
 773 {
 774   GList *tmp;
 775   GstAggregatorPad *tmppad;
 776
 777   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
 778     tmppad = (GstAggregatorPad *) tmp->data;
 779
 780     if (_check_pending_flush_stop (tmppad) == FALSE) {
 781       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
 782           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
 783       return FALSE;
 784     }
 785   }
 786
 787   return TRUE;
 788 }
 789
 790 static void
 791 gst_aggregator_flush_start (GstAggregator * self, GstAggregatorPad * aggpad,
 792     GstEvent * event)
 793 {
 794   GstAggregatorPrivate *priv = self->priv;
 795   GstAggregatorPadPrivate *padpriv = aggpad->priv;
 796
 797   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 798
 799   /*  Remove pad buffer and wake up the streaming thread */
 800   gst_aggregator_pad_drop_buffer (aggpad);
 801
 802   PAD_FLUSH_LOCK (aggpad);
 803   PAD_LOCK (aggpad);
 804   if (padpriv->pending_flush_start) {
 805     GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
 806
 807     padpriv->pending_flush_start = FALSE;
 808     padpriv->pending_flush_stop = TRUE;
 809   }
 810   PAD_UNLOCK (aggpad);
 811
 812   GST_OBJECT_LOCK (self);
 813   if (priv->flush_seeking) {
 814     /* If flush_seeking we forward the first FLUSH_START */
 815     if (priv->pending_flush_start) {
 816       priv->pending_flush_start = FALSE;
 817       GST_OBJECT_UNLOCK (self);
 818
 819       GST_INFO_OBJECT (self, "Flushing, pausing srcpad task");
 820       gst_aggregator_stop_srcpad_task (self, event);
 821       priv->flow_return = GST_FLOW_OK;
 822
 823       GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
 824       GST_PAD_STREAM_LOCK (self->srcpad);
 825       GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
 826       event = NULL;
 827     } else {
 828       GST_OBJECT_UNLOCK (self);
 829       gst_event_unref (event);
 830     }
 831   } else {
 832     GST_OBJECT_UNLOCK (self);
 833     gst_event_unref (event);
 834   }
 835   PAD_FLUSH_UNLOCK (aggpad);
 836
 837   gst_aggregator_pad_drop_buffer (aggpad);
 838 }
 839
 840 /* GstAggregator vmethods default implementations */
 841 static gboolean
 842 gst_aggregator_default_sink_event (GstAggregator * self,
 843     GstAggregatorPad * aggpad, GstEvent * event)
 844 {
 845   gboolean res = TRUE;
 846   GstPad *pad = GST_PAD (aggpad);
 847   GstAggregatorPrivate *priv = self->priv;
 848
 849   switch (GST_EVENT_TYPE (event)) {
 850     case GST_EVENT_FLUSH_START:
 851     {
 852       gst_aggregator_flush_start (self, aggpad, event);
 853       /* We forward only in one case: right after flush_seeking */
 854       event = NULL;
 855       goto eat;
 856     }
 857     case GST_EVENT_FLUSH_STOP:
 858     {
 859       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
 860
 861       gst_aggregator_pad_flush (aggpad, self);
 862       GST_OBJECT_LOCK (self);
 863       if (priv->flush_seeking) {
 864         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
 865         if (gst_aggregator_all_flush_stop_received_locked (self)) {
 866           GST_OBJECT_UNLOCK (self);
 867           /* That means we received FLUSH_STOP/FLUSH_STOP on
 868            * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
 869           gst_aggregator_flush (self);
 870           gst_pad_push_event (self->srcpad, event);
 871           event = NULL;
 872           SRC_LOCK (self);
 873           priv->send_eos = TRUE;
 874           SRC_BROADCAST (self);
 875           SRC_UNLOCK (self);
 876
 877           GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
 878           GST_PAD_STREAM_UNLOCK (self->srcpad);
 879           gst_aggregator_start_srcpad_task (self);
 880         } else {
 881           GST_OBJECT_UNLOCK (self);
 882         }
 883       } else {
 884         GST_OBJECT_UNLOCK (self);
 885       }
 886
 887       /* We never forward the event */
 888       goto eat;
 889     }
 890     case GST_EVENT_EOS:
 891     {
 892       GST_DEBUG_OBJECT (aggpad, "EOS");
 893
 894       /* We still have a buffer, and we don't want the subclass to have to
 895        * check for it. Mark pending_eos, eos will be set when steal_buffer is
 896        * called
 897        */
 898       SRC_LOCK (self);
 899       PAD_LOCK (aggpad);
 900       if (!aggpad->priv->buffer) {
 901         aggpad->priv->eos = TRUE;
 902       } else {
 903         aggpad->priv->pending_eos = TRUE;
 904       }
 905       PAD_UNLOCK (aggpad);
 906
 907       SRC_BROADCAST (self);
 908       SRC_UNLOCK (self);
 909       goto eat;
 910     }
 911     case GST_EVENT_SEGMENT:
 912     {
 913       GST_OBJECT_LOCK (aggpad);
 914       gst_event_copy_segment (event, &aggpad->segment);
 915       GST_OBJECT_UNLOCK (aggpad);
 916
 917       GST_OBJECT_LOCK (self);
 918       self->priv->seqnum = gst_event_get_seqnum (event);
 919       GST_OBJECT_UNLOCK (self);
 920       goto eat;
 921     }
 922     case GST_EVENT_STREAM_START:
 923     {
 924       goto eat;
 925     }
 926     case GST_EVENT_GAP:
 927     {
 928       /* FIXME: need API to handle GAP events properly */
 929       GST_FIXME_OBJECT (self, "implement support for GAP events");
 930       /* don't forward GAP events downstream */
 931       goto eat;
 932     }
 933     case GST_EVENT_TAG:
 934     {
 935       GstTagList *tags;
 936
 937       gst_event_parse_tag (event, &tags);
 938
 939       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
 940         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
 941         gst_event_unref (event);
 942         event = NULL;
 943         goto eat;
 944       }
 945       break;
 946     }
 947     default:
 948     {
 949       break;
 950     }
 951   }
 952
 953   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
 954   return gst_pad_event_default (pad, GST_OBJECT (self), event);
 955
 956 eat:
 957   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
 958   if (event)
 959     gst_event_unref (event);
 960
 961   return res;
 962 }
 963
 964 static inline gboolean
 965 gst_aggregator_stop_pad (GstAggregator * self, GstAggregatorPad * pad,
 966     gpointer unused_udata)
 967 {
 968   gst_aggregator_pad_flush (pad, self);
 969
 970   return TRUE;
 971 }
 972
 973 static gboolean
 974 gst_aggregator_stop (GstAggregator * agg)
 975 {
 976   GstAggregatorClass *klass;
 977   gboolean result;
 978
 979   gst_aggregator_reset_flow_values (agg);
 980
 981   gst_aggregator_iterate_sinkpads (agg, gst_aggregator_stop_pad, NULL);
 982
 983   klass = GST_AGGREGATOR_GET_CLASS (agg);
 984
 985   if (klass->stop)
 986     result = klass->stop (agg);
 987   else
 988     result = TRUE;
 989
 990   if (agg->priv->tags)
 991     gst_tag_list_unref (agg->priv->tags);
 992   agg->priv->tags = NULL;
 993
 994   return result;
 995 }
 996
 997 /* GstElement vmethods implementations */
 998 static GstStateChangeReturn
 999 gst_aggregator_change_state (GstElement * element, GstStateChange transition)
1000 {
1001   GstStateChangeReturn ret;
1002   GstAggregator *self = GST_AGGREGATOR (element);
1003
1004   switch (transition) {
1005     case GST_STATE_CHANGE_READY_TO_PAUSED:
1006       if (!gst_aggregator_start (self))
1007         goto error_start;
1008       break;
1009     default:
1010       break;
1011   }
1012
1013   if ((ret =
1014           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
1015               transition)) == GST_STATE_CHANGE_FAILURE)
1016     goto failure;
1017
1018
1019   switch (transition) {
1020     case GST_STATE_CHANGE_PAUSED_TO_READY:
1021       if (!gst_aggregator_stop (self)) {
1022         /* What to do in this case? Error out? */
1023         GST_ERROR_OBJECT (self, "Subclass failed to stop.");
1024       }
1025       break;
1026     default:
1027       break;
1028   }
1029
1030   return ret;
1031
1032 /* ERRORS */
1033 failure:
1034   {
1035     GST_ERROR_OBJECT (element, "parent failed state change");
1036     return ret;
1037   }
1038 error_start:
1039   {
1040     GST_ERROR_OBJECT (element, "Subclass failed to start");
1041     return GST_STATE_CHANGE_FAILURE;
1042   }
1043 }
1044
1045 static void
1046 gst_aggregator_release_pad (GstElement * element, GstPad * pad)
1047 {
1048   GstAggregator *self = GST_AGGREGATOR (element);
1049
1050   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1051
1052   GST_INFO_OBJECT (pad, "Removing pad");
1053
1054   SRC_LOCK (self);
1055   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1056   gst_aggregator_pad_drop_buffer (aggpad);
1057   gst_element_remove_pad (element, pad);
1058
1059   SRC_BROADCAST (self);
1060   SRC_UNLOCK (self);
1061 }
1062
1063 static GstPad *
1064 gst_aggregator_request_new_pad (GstElement * element,
1065     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
1066 {
1067   GstAggregator *self;
1068   GstAggregatorPad *agg_pad;
1069
1070   GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
1071   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
1072
1073   self = GST_AGGREGATOR (element);
1074
1075   if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
1076     gint serial = 0;
1077     gchar *name = NULL;
1078
1079     GST_OBJECT_LOCK (element);
1080     if (req_name == NULL || strlen (req_name) < 6
1081         || !g_str_has_prefix (req_name, "sink_")) {
1082       /* no name given when requesting the pad, use next available int */
1083       priv->padcount++;
1084     } else {
1085       /* parse serial number from requested padname */
1086       serial = g_ascii_strtoull (&req_name[5], NULL, 10);
1087       if (serial >= priv->padcount)
1088         priv->padcount = serial;
1089     }
1090
1091     name = g_strdup_printf ("sink_%u", priv->padcount);
1092     agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
1093         "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
1094     g_free (name);
1095
1096     GST_OBJECT_UNLOCK (element);
1097
1098   } else {
1099     return NULL;
1100   }
1101
1102   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
1103
1104   if (priv->running)
1105     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
1106
1107   /* add the pad to the element */
1108   gst_element_add_pad (element, GST_PAD (agg_pad));
1109
1110   return GST_PAD (agg_pad);
1111 }
1112
1113 static gboolean
1114 gst_aggregator_query_latency (GstAggregator * self, GstQuery * query)
1115 {
1116   gboolean query_ret, live;
1117   GstClockTime our_latency, min, max;
1118
1119   query_ret = gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1120
1121   if (!query_ret) {
1122     GST_WARNING_OBJECT (self, "Latency query failed");
1123     return FALSE;
1124   }
1125
1126   gst_query_parse_latency (query, &live, &min, &max);
1127
1128   SRC_LOCK (self);
1129   our_latency = self->priv->latency;
1130
1131   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (min))) {
1132     GST_ERROR_OBJECT (self, "Invalid minimum latency %" GST_TIME_FORMAT
1133         ". Please file a bug at " PACKAGE_BUGREPORT ".", GST_TIME_ARGS (min));
1134     SRC_UNLOCK (self);
1135     return FALSE;
1136   }
1137
1138   if (min > max && GST_CLOCK_TIME_IS_VALID (max)) {
1139     GST_ELEMENT_WARNING (self, CORE, CLOCK, (NULL),
1140         ("Impossible to configure latency: max %" GST_TIME_FORMAT " < min %"
1141             GST_TIME_FORMAT ". Add queues or other buffering elements.",
1142             GST_TIME_ARGS (max), GST_TIME_ARGS (min)));
1143     SRC_UNLOCK (self);
1144     return FALSE;
1145   }
1146
1147   self->priv->latency_live = live;
1148   self->priv->latency_min = min;
1149   self->priv->latency_max = max;
1150
1151   /* add our own */
1152   min += our_latency;
1153   min += self->priv->sub_latency_min;
1154   if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1155       && GST_CLOCK_TIME_IS_VALID (max))
1156     max += self->priv->sub_latency_max;
1157   else
1158     max = GST_CLOCK_TIME_NONE;
1159
1160   if (live && min > max) {
1161     GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1162         ("%s", "Latency too big"),
1163         ("The requested latency value is too big for the current pipeline.  "
1164             "Limiting to %" G_GINT64_FORMAT, max));
1165     min = max;
1166     /* FIXME: This could in theory become negative, but in
1167      * that case all is lost anyway */
1168     self->priv->latency -= min - max;
1169     /* FIXME: shouldn't we g_object_notify() the change here? */
1170   }
1171
1172   SRC_BROADCAST (self);
1173   SRC_UNLOCK (self);
1174
1175   GST_DEBUG_OBJECT (self, "configured latency live:%s min:%" G_GINT64_FORMAT
1176       " max:%" G_GINT64_FORMAT, live ? "true" : "false", min, max);
1177
1178   gst_query_set_latency (query, live, min, max);
1179
1180   return query_ret;
1181 }
1182
1183 /**
1184  * gst_aggregator_get_latency_unlocked:
1185  * @self: a #GstAggregator
1186  * @live: (out) (allow-none): whether @self is live
1187  * @min_latency: (out) (allow-none): the configured minimum latency of @self
1188  * @max_latency: (out) (allow-none): the configured maximum latency of @self
1189  *
1190  * Retreives the latency values reported by @self in response to the latency
1191  * query.
1192  *
1193  * Typically only called by subclasses.
1194  *
1195  * MUST be called with the src_lock held.
1196  */
1197 void
1198 gst_aggregator_get_latency_unlocked (GstAggregator * self, gboolean * live,
1199     GstClockTime * min_latency, GstClockTime * max_latency)
1200 {
1201   GstClockTime min, max;
1202
1203   g_return_if_fail (GST_IS_AGGREGATOR (self));
1204
1205   /* latency_min is never GST_CLOCK_TIME_NONE by construction */
1206   min = self->priv->latency_min;
1207   max = self->priv->latency_max;
1208
1209   /* add our own */
1210   min += self->priv->latency;
1211   min += self->priv->sub_latency_min;
1212   if (GST_CLOCK_TIME_IS_VALID (max)
1213       && GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max))
1214     max += self->priv->sub_latency_max;
1215   else
1216     max = GST_CLOCK_TIME_NONE;
1217
1218   if (live)
1219     *live = self->priv->latency_live;
1220   if (min_latency)
1221     *min_latency = min;
1222   if (max_latency)
1223     *max_latency = max;
1224 }
1225
1226 static gboolean
1227 gst_aggregator_send_event (GstElement * element, GstEvent * event)
1228 {
1229   GstAggregator *self = GST_AGGREGATOR (element);
1230
1231   GST_STATE_LOCK (element);
1232   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK &&
1233       GST_STATE (element) < GST_STATE_PAUSED) {
1234     gdouble rate;
1235     GstFormat fmt;
1236     GstSeekFlags flags;
1237     GstSeekType start_type, stop_type;
1238     gint64 start, stop;
1239
1240     gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1241         &start, &stop_type, &stop);
1242
1243     GST_OBJECT_LOCK (self);
1244     gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1245         stop_type, stop, NULL);
1246     self->priv->seqnum = gst_event_get_seqnum (event);
1247     GST_OBJECT_UNLOCK (self);
1248
1249     GST_DEBUG_OBJECT (element, "Storing segment %" GST_PTR_FORMAT, event);
1250   }
1251   GST_STATE_UNLOCK (element);
1252
1253
1254   return GST_ELEMENT_CLASS (aggregator_parent_class)->send_event (element,
1255       event);
1256 }
1257
1258 static gboolean
1259 gst_aggregator_default_src_query (GstAggregator * self, GstQuery * query)
1260 {
1261   gboolean res = TRUE;
1262
1263   switch (GST_QUERY_TYPE (query)) {
1264     case GST_QUERY_SEEKING:
1265     {
1266       GstFormat format;
1267
1268       /* don't pass it along as some (file)sink might claim it does
1269        * whereas with a collectpads in between that will not likely work */
1270       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
1271       gst_query_set_seeking (query, format, FALSE, 0, -1);
1272       res = TRUE;
1273
1274       goto discard;
1275     }
1276     case GST_QUERY_LATENCY:
1277       return gst_aggregator_query_latency (self, query);
1278     default:
1279       break;
1280   }
1281
1282   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1283
1284 discard:
1285   return res;
1286 }
1287
1288 static gboolean
1289 gst_aggregator_event_forward_func (GstPad * pad, gpointer user_data)
1290 {
1291   EventData *evdata = user_data;
1292   gboolean ret = TRUE;
1293   GstPad *peer = gst_pad_get_peer (pad);
1294   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1295
1296   if (peer) {
1297     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
1298     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
1299     gst_object_unref (peer);
1300   }
1301
1302   if (ret == FALSE) {
1303     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
1304       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
1305
1306     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
1307       GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
1308
1309       if (gst_pad_query (peer, seeking)) {
1310         gboolean seekable;
1311
1312         gst_query_parse_seeking (seeking, NULL, &seekable, NULL, NULL);
1313
1314         if (seekable == FALSE) {
1315           GST_INFO_OBJECT (pad,
1316               "Source not seekable, We failed but it does not matter!");
1317
1318           ret = TRUE;
1319         }
1320       } else {
1321         GST_ERROR_OBJECT (pad, "Query seeking FAILED");
1322       }
1323
1324       gst_query_unref (seeking);
1325     }
1326
1327     if (evdata->flush) {
1328       PAD_LOCK (aggpad);
1329       aggpad->priv->pending_flush_start = FALSE;
1330       aggpad->priv->pending_flush_stop = FALSE;
1331       PAD_UNLOCK (aggpad);
1332     }
1333   } else {
1334     evdata->one_actually_seeked = TRUE;
1335   }
1336
1337   evdata->result &= ret;
1338
1339   /* Always send to all pads */
1340   return FALSE;
1341 }
1342
1343 static EventData
1344 gst_aggregator_forward_event_to_all_sinkpads (GstAggregator * self,
1345     GstEvent * event, gboolean flush)
1346 {
1347   EventData evdata;
1348
1349   evdata.event = event;
1350   evdata.result = TRUE;
1351   evdata.flush = flush;
1352   evdata.one_actually_seeked = FALSE;
1353
1354   /* We first need to set all pads as flushing in a first pass
1355    * as flush_start flush_stop is sometimes sent synchronously
1356    * while we send the seek event */
1357   if (flush) {
1358     GList *l;
1359
1360     GST_OBJECT_LOCK (self);
1361     for (l = GST_ELEMENT_CAST (self)->sinkpads; l != NULL; l = l->next) {
1362       GstAggregatorPad *pad = l->data;
1363
1364       PAD_LOCK (pad);
1365       pad->priv->pending_flush_start = TRUE;
1366       pad->priv->pending_flush_stop = FALSE;
1367       PAD_UNLOCK (pad);
1368     }
1369     GST_OBJECT_UNLOCK (self);
1370   }
1371
1372   gst_pad_forward (self->srcpad, gst_aggregator_event_forward_func, &evdata);
1373
1374   gst_event_unref (event);
1375
1376   return evdata;
1377 }
1378
1379 static gboolean
1380 gst_aggregator_do_seek (GstAggregator * self, GstEvent * event)
1381 {
1382   gdouble rate;
1383   GstFormat fmt;
1384   GstSeekFlags flags;
1385   GstSeekType start_type, stop_type;
1386   gint64 start, stop;
1387   gboolean flush;
1388   EventData evdata;
1389   GstAggregatorPrivate *priv = self->priv;
1390
1391   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1392       &start, &stop_type, &stop);
1393
1394   GST_INFO_OBJECT (self, "starting SEEK");
1395
1396   flush = flags & GST_SEEK_FLAG_FLUSH;
1397
1398   GST_OBJECT_LOCK (self);
1399   if (flush) {
1400     priv->pending_flush_start = TRUE;
1401     priv->flush_seeking = TRUE;
1402   }
1403
1404   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1405       stop_type, stop, NULL);
1406   GST_OBJECT_UNLOCK (self);
1407
1408   /* forward the seek upstream */
1409   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, flush);
1410   event = NULL;
1411
1412   if (!evdata.result || !evdata.one_actually_seeked) {
1413     GST_OBJECT_LOCK (self);
1414     priv->flush_seeking = FALSE;
1415     priv->pending_flush_start = FALSE;
1416     GST_OBJECT_UNLOCK (self);
1417   }
1418
1419   GST_INFO_OBJECT (self, "seek done, result: %d", evdata.result);
1420
1421   return evdata.result;
1422 }
1423
1424 static gboolean
1425 gst_aggregator_default_src_event (GstAggregator * self, GstEvent * event)
1426 {
1427   EventData evdata;
1428   gboolean res = TRUE;
1429
1430   switch (GST_EVENT_TYPE (event)) {
1431     case GST_EVENT_SEEK:
1432     {
1433       gst_event_ref (event);
1434       res = gst_aggregator_do_seek (self, event);
1435       gst_event_unref (event);
1436       event = NULL;
1437       goto done;
1438     }
1439     case GST_EVENT_NAVIGATION:
1440     {
1441       /* navigation is rather pointless. */
1442       res = FALSE;
1443       gst_event_unref (event);
1444       goto done;
1445     }
1446     default:
1447     {
1448       break;
1449     }
1450   }
1451
1452   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE);
1453   res = evdata.result;
1454
1455 done:
1456   return res;
1457 }
1458
1459 static gboolean
1460 gst_aggregator_src_pad_event_func (GstPad * pad, GstObject * parent,
1461     GstEvent * event)
1462 {
1463   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1464
1465   return klass->src_event (GST_AGGREGATOR (parent), event);
1466 }
1467
1468 static gboolean
1469 gst_aggregator_src_pad_query_func (GstPad * pad, GstObject * parent,
1470     GstQuery * query)
1471 {
1472   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1473
1474   return klass->src_query (GST_AGGREGATOR (parent), query);
1475 }
1476
1477 static gboolean
1478 gst_aggregator_src_pad_activate_mode_func (GstPad * pad,
1479     GstObject * parent, GstPadMode mode, gboolean active)
1480 {
1481   GstAggregator *self = GST_AGGREGATOR (parent);
1482   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1483
1484   if (klass->src_activate) {
1485     if (klass->src_activate (self, mode, active) == FALSE) {
1486       return FALSE;
1487     }
1488   }
1489
1490   if (active == TRUE) {
1491     switch (mode) {
1492       case GST_PAD_MODE_PUSH:
1493       {
1494         GST_INFO_OBJECT (pad, "Activating pad!");
1495         gst_aggregator_start_srcpad_task (self);
1496         return TRUE;
1497       }
1498       default:
1499       {
1500         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
1501         return FALSE;
1502       }
1503     }
1504   }
1505
1506   /* deactivating */
1507   GST_INFO_OBJECT (self, "Deactivating srcpad");
1508   gst_aggregator_stop_srcpad_task (self, FALSE);
1509
1510   return TRUE;
1511 }
1512
1513 static gboolean
1514 gst_aggregator_default_sink_query (GstAggregator * self,
1515     GstAggregatorPad * aggpad, GstQuery * query)
1516 {
1517   GstPad *pad = GST_PAD (aggpad);
1518
1519   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1520 }
1521
1522 static void
1523 gst_aggregator_finalize (GObject * object)
1524 {
1525   GstAggregator *self = (GstAggregator *) object;
1526
1527   g_mutex_clear (&self->priv->src_lock);
1528   g_cond_clear (&self->priv->src_cond);
1529
1530   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
1531 }
1532
1533 /*
1534  * gst_aggregator_set_latency_property:
1535  * @agg: a #GstAggregator
1536  * @latency: the new latency value.
1537  *
1538  * Sets the new latency value to @latency. This value is used to limit the
1539  * amount of time a pad waits for data to appear before considering the pad
1540  * as unresponsive.
1541  */
1542 static void
1543 gst_aggregator_set_latency_property (GstAggregator * self, gint64 latency)
1544 {
1545   gboolean changed;
1546   GstClockTime min, max;
1547
1548   g_return_if_fail (GST_IS_AGGREGATOR (self));
1549   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (latency));
1550
1551   SRC_LOCK (self);
1552   if (self->priv->latency_live) {
1553     min = self->priv->latency_min;
1554     max = self->priv->latency_max;
1555     /* add our own */
1556     min += latency;
1557     min += self->priv->sub_latency_min;
1558     if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1559         && GST_CLOCK_TIME_IS_VALID (max))
1560       max += self->priv->sub_latency_max;
1561     else
1562       max = GST_CLOCK_TIME_NONE;
1563
1564     if (GST_CLOCK_TIME_IS_VALID (max) && min > max) {
1565       GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1566           ("%s", "Latency too big"),
1567           ("The requested latency value is too big for the latency in the "
1568               "current pipeline.  Limiting to %" G_GINT64_FORMAT, max));
1569       /* FIXME: This could in theory become negative, but in
1570        * that case all is lost anyway */
1571       latency -= min - max;
1572       /* FIXME: shouldn't we g_object_notify() the change here? */
1573     }
1574   }
1575
1576   changed = (self->priv->latency != latency);
1577   self->priv->latency = latency;
1578
1579   if (changed)
1580     SRC_BROADCAST (self);
1581   SRC_UNLOCK (self);
1582
1583   if (changed)
1584     gst_element_post_message (GST_ELEMENT_CAST (self),
1585         gst_message_new_latency (GST_OBJECT_CAST (self)));
1586 }
1587
1588 /*
1589  * gst_aggregator_get_latency_property:
1590  * @agg: a #GstAggregator
1591  *
1592  * Gets the latency value. See gst_aggregator_set_latency for
1593  * more details.
1594  *
1595  * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
1596  * before a pad is deemed unresponsive. A value of -1 means an
1597  * unlimited time.
1598  */
1599 static gint64
1600 gst_aggregator_get_latency_property (GstAggregator * agg)
1601 {
1602   gint64 res;
1603
1604   g_return_val_if_fail (GST_IS_AGGREGATOR (agg), -1);
1605
1606   GST_OBJECT_LOCK (agg);
1607   res = agg->priv->latency;
1608   GST_OBJECT_UNLOCK (agg);
1609
1610   return res;
1611 }
1612
1613 static void
1614 gst_aggregator_set_property (GObject * object, guint prop_id,
1615     const GValue * value, GParamSpec * pspec)
1616 {
1617   GstAggregator *agg = GST_AGGREGATOR (object);
1618
1619   switch (prop_id) {
1620     case PROP_LATENCY:
1621       gst_aggregator_set_latency_property (agg, g_value_get_int64 (value));
1622       break;
1623     default:
1624       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1625       break;
1626   }
1627 }
1628
1629 static void
1630 gst_aggregator_get_property (GObject * object, guint prop_id,
1631     GValue * value, GParamSpec * pspec)
1632 {
1633   GstAggregator *agg = GST_AGGREGATOR (object);
1634
1635   switch (prop_id) {
1636     case PROP_LATENCY:
1637       g_value_set_int64 (value, gst_aggregator_get_latency_property (agg));
1638       break;
1639     default:
1640       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1641       break;
1642   }
1643 }
1644
1645 /* GObject vmethods implementations */
1646 static void
1647 gst_aggregator_class_init (GstAggregatorClass * klass)
1648 {
1649   GObjectClass *gobject_class = (GObjectClass *) klass;
1650   GstElementClass *gstelement_class = (GstElementClass *) klass;
1651
1652   aggregator_parent_class = g_type_class_peek_parent (klass);
1653   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1654
1655   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1656       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1657
1658   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1659
1660   klass->sink_event = gst_aggregator_default_sink_event;
1661   klass->sink_query = gst_aggregator_default_sink_query;
1662
1663   klass->src_event = gst_aggregator_default_src_event;
1664   klass->src_query = gst_aggregator_default_src_query;
1665
1666   gstelement_class->request_new_pad =
1667       GST_DEBUG_FUNCPTR (gst_aggregator_request_new_pad);
1668   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_aggregator_send_event);
1669   gstelement_class->release_pad =
1670       GST_DEBUG_FUNCPTR (gst_aggregator_release_pad);
1671   gstelement_class->change_state =
1672       GST_DEBUG_FUNCPTR (gst_aggregator_change_state);
1673
1674   gobject_class->set_property = gst_aggregator_set_property;
1675   gobject_class->get_property = gst_aggregator_get_property;
1676   gobject_class->finalize = gst_aggregator_finalize;
1677
1678   g_object_class_install_property (gobject_class, PROP_LATENCY,
1679       g_param_spec_int64 ("latency", "Buffer latency",
1680           "Additional latency in live mode to allow upstream "
1681           "to take longer to produce buffers for the current "
1682           "position", 0,
1683           (G_MAXLONG == G_MAXINT64) ? G_MAXINT64 : (G_MAXLONG * GST_SECOND - 1),
1684           DEFAULT_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1685
1686   GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_stop_pad);
1687 }
1688
1689 static void
1690 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1691 {
1692   GstPadTemplate *pad_template;
1693   GstAggregatorPrivate *priv;
1694
1695   g_return_if_fail (klass->aggregate != NULL);
1696
1697   self->priv =
1698       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1699       GstAggregatorPrivate);
1700
1701   priv = self->priv;
1702
1703   pad_template =
1704       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1705   g_return_if_fail (pad_template != NULL);
1706
1707   priv->padcount = -1;
1708   priv->tags_changed = FALSE;
1709
1710   self->priv->latency_live = FALSE;
1711   self->priv->latency_min = self->priv->sub_latency_min = 0;
1712   self->priv->latency_max = self->priv->sub_latency_max = 0;
1713   gst_aggregator_reset_flow_values (self);
1714
1715   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1716
1717   gst_pad_set_event_function (self->srcpad,
1718       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_event_func));
1719   gst_pad_set_query_function (self->srcpad,
1720       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_query_func));
1721   gst_pad_set_activatemode_function (self->srcpad,
1722       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_activate_mode_func));
1723
1724   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1725
1726   self->priv->latency = DEFAULT_LATENCY;
1727
1728   g_mutex_init (&self->priv->src_lock);
1729   g_cond_init (&self->priv->src_cond);
1730 }
1731
1732 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1733  * method to get to the padtemplates */
1734 GType
1735 gst_aggregator_get_type (void)
1736 {
1737   static volatile gsize type = 0;
1738
1739   if (g_once_init_enter (&type)) {
1740     GType _type;
1741     static const GTypeInfo info = {
1742       sizeof (GstAggregatorClass),
1743       NULL,
1744       NULL,
1745       (GClassInitFunc) gst_aggregator_class_init,
1746       NULL,
1747       NULL,
1748       sizeof (GstAggregator),
1749       0,
1750       (GInstanceInitFunc) gst_aggregator_init,
1751     };
1752
1753     _type = g_type_register_static (GST_TYPE_ELEMENT,
1754         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1755     g_once_init_leave (&type, _type);
1756   }
1757   return type;
1758 }
1759
1760 static GstFlowReturn
1761 gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1762 {
1763   GstBuffer *actual_buf = buffer;
1764   GstAggregator *self = GST_AGGREGATOR (object);
1765   GstAggregatorPrivate *priv = self->priv;
1766   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1767   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1768   GstFlowReturn flow_return;
1769
1770   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1771
1772   PAD_FLUSH_LOCK (aggpad);
1773
1774   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1775     goto flushing;
1776
1777   PAD_LOCK (aggpad);
1778   if (aggpad->priv->pending_eos == TRUE)
1779     goto eos;
1780
1781   while (aggpad->priv->buffer
1782       && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1783     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1784     PAD_WAIT_EVENT (aggpad);
1785   }
1786   PAD_UNLOCK (aggpad);
1787
1788   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1789     goto flushing;
1790
1791   if (aggclass->clip) {
1792     aggclass->clip (self, aggpad, buffer, &actual_buf);
1793   }
1794
1795   SRC_LOCK (self);
1796   PAD_LOCK (aggpad);
1797   if (aggpad->priv->buffer)
1798     gst_buffer_unref (aggpad->priv->buffer);
1799   aggpad->priv->buffer = actual_buf;
1800   PAD_UNLOCK (aggpad);
1801   PAD_FLUSH_UNLOCK (aggpad);
1802
1803   SRC_BROADCAST (self);
1804   SRC_UNLOCK (self);
1805
1806   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1807
1808   GST_OBJECT_LOCK (self);
1809   flow_return = priv->flow_return;
1810   GST_OBJECT_UNLOCK (self);
1811
1812   return flow_return;
1813
1814 flushing:
1815   PAD_FLUSH_UNLOCK (aggpad);
1816
1817   gst_buffer_unref (buffer);
1818   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1819
1820   return GST_FLOW_FLUSHING;
1821
1822 eos:
1823   PAD_UNLOCK (aggpad);
1824   PAD_FLUSH_UNLOCK (aggpad);
1825
1826   gst_buffer_unref (buffer);
1827   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1828
1829   return GST_FLOW_EOS;
1830 }
1831
1832 static gboolean
1833 gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
1834     GstQuery * query)
1835 {
1836   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1837   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1838
1839   if (GST_QUERY_IS_SERIALIZED (query)) {
1840     PAD_LOCK (aggpad);
1841
1842     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE) {
1843       PAD_UNLOCK (aggpad);
1844       goto flushing;
1845     }
1846
1847     while (aggpad->priv->buffer
1848         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1849       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1850       PAD_WAIT_EVENT (aggpad);
1851     }
1852     PAD_UNLOCK (aggpad);
1853
1854     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1855       goto flushing;
1856   }
1857
1858   return klass->sink_query (GST_AGGREGATOR (parent),
1859       GST_AGGREGATOR_PAD (pad), query);
1860
1861 flushing:
1862   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping query");
1863   return FALSE;
1864 }
1865
1866 static gboolean
1867 gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
1868     GstEvent * event)
1869 {
1870   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1871   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1872
1873   if (GST_EVENT_IS_SERIALIZED (event) && GST_EVENT_TYPE (event) != GST_EVENT_EOS
1874       && GST_EVENT_TYPE (event) != GST_EVENT_SEGMENT_DONE) {
1875     PAD_LOCK (aggpad);
1876
1877     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1878         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
1879       PAD_UNLOCK (aggpad);
1880       goto flushing;
1881     }
1882
1883     while (aggpad->priv->buffer
1884         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1885       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1886       PAD_WAIT_EVENT (aggpad);
1887     }
1888     PAD_UNLOCK (aggpad);
1889
1890     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1891         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP)
1892       goto flushing;
1893   }
1894
1895   return klass->sink_event (GST_AGGREGATOR (parent),
1896       GST_AGGREGATOR_PAD (pad), event);
1897
1898 flushing:
1899   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping event");
1900   if (GST_EVENT_IS_STICKY (event))
1901     gst_pad_store_sticky_event (pad, event);
1902   gst_event_unref (event);
1903   return FALSE;
1904 }
1905
1906 static gboolean
1907 gst_aggregator_pad_activate_mode_func (GstPad * pad,
1908     GstObject * parent, GstPadMode mode, gboolean active)
1909 {
1910   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1911
1912   if (active == FALSE) {
1913     PAD_LOCK (aggpad);
1914     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1915     gst_buffer_replace (&aggpad->priv->buffer, NULL);
1916     PAD_BROADCAST_EVENT (aggpad);
1917     PAD_UNLOCK (aggpad);
1918   } else {
1919     PAD_LOCK (aggpad);
1920     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1921     PAD_BROADCAST_EVENT (aggpad);
1922     PAD_UNLOCK (aggpad);
1923   }
1924
1925   return TRUE;
1926 }
1927
1928 /***********************************
1929  * GstAggregatorPad implementation  *
1930  ************************************/
1931 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1932
1933 static void
1934 gst_aggregator_pad_constructed (GObject * object)
1935 {
1936   GstPad *pad = GST_PAD (object);
1937
1938   gst_pad_set_chain_function (pad,
1939       GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
1940   gst_pad_set_event_function (pad,
1941       GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func));
1942   gst_pad_set_query_function (pad,
1943       GST_DEBUG_FUNCPTR (gst_aggregator_pad_query_func));
1944   gst_pad_set_activatemode_function (pad,
1945       GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
1946 }
1947
1948 static void
1949 gst_aggregator_pad_finalize (GObject * object)
1950 {
1951   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1952
1953   g_cond_clear (&pad->priv->event_cond);
1954   g_mutex_clear (&pad->priv->flush_lock);
1955   g_mutex_clear (&pad->priv->lock);
1956
1957   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->finalize (object);
1958 }
1959
1960 static void
1961 gst_aggregator_pad_dispose (GObject * object)
1962 {
1963   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1964
1965   gst_aggregator_pad_drop_buffer (pad);
1966
1967   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->dispose (object);
1968 }
1969
1970 static void
1971 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
1972 {
1973   GObjectClass *gobject_class = (GObjectClass *) klass;
1974
1975   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
1976
1977   gobject_class->constructed = gst_aggregator_pad_constructed;
1978   gobject_class->finalize = gst_aggregator_pad_finalize;
1979   gobject_class->dispose = gst_aggregator_pad_dispose;
1980 }
1981
1982 static void
1983 gst_aggregator_pad_init (GstAggregatorPad * pad)
1984 {
1985   pad->priv =
1986       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
1987       GstAggregatorPadPrivate);
1988
1989   pad->priv->buffer = NULL;
1990   g_cond_init (&pad->priv->event_cond);
1991
1992   g_mutex_init (&pad->priv->flush_lock);
1993   g_mutex_init (&pad->priv->lock);
1994 }
1995
1996 /**
1997  * gst_aggregator_pad_steal_buffer:
1998  * @pad: the pad to get buffer from
1999  *
2000  * Steal the ref to the buffer currently queued in @pad.
2001  *
2002  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
2003  *   queued. You should unref the buffer after usage.
2004  */
2005 GstBuffer *
2006 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
2007 {
2008   GstBuffer *buffer = NULL;
2009
2010   PAD_LOCK (pad);
2011   if (pad->priv->buffer) {
2012     GST_TRACE_OBJECT (pad, "Consuming buffer");
2013     buffer = pad->priv->buffer;
2014     pad->priv->buffer = NULL;
2015     if (pad->priv->pending_eos) {
2016       pad->priv->pending_eos = FALSE;
2017       pad->priv->eos = TRUE;
2018     }
2019     PAD_BROADCAST_EVENT (pad);
2020     GST_DEBUG_OBJECT (pad, "Consumed: %" GST_PTR_FORMAT, buffer);
2021   }
2022   PAD_UNLOCK (pad);
2023
2024   return buffer;
2025 }
2026
2027 /**
2028  * gst_aggregator_pad_drop_buffer:
2029  * @pad: the pad where to drop any pending buffer
2030  *
2031  * Drop the buffer currently queued in @pad.
2032  *
2033  * Returns: TRUE if there was a buffer queued in @pad, or FALSE if not.
2034  */
2035 gboolean
2036 gst_aggregator_pad_drop_buffer (GstAggregatorPad * pad)
2037 {
2038   GstBuffer *buf;
2039
2040   buf = gst_aggregator_pad_steal_buffer (pad);
2041
2042   if (buf == NULL)
2043     return FALSE;
2044
2045   gst_buffer_unref (buf);
2046   return TRUE;
2047 }
2048
2049 /**
2050  * gst_aggregator_pad_get_buffer:
2051  * @pad: the pad to get buffer from
2052  *
2053  * Returns: (transfer full): A reference to the buffer in @pad or
2054  * NULL if no buffer was queued. You should unref the buffer after
2055  * usage.
2056  */
2057 GstBuffer *
2058 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
2059 {
2060   GstBuffer *buffer = NULL;
2061
2062   PAD_LOCK (pad);
2063   if (pad->priv->buffer)
2064     buffer = gst_buffer_ref (pad->priv->buffer);
2065   PAD_UNLOCK (pad);
2066
2067   return buffer;
2068 }
2069
2070 gboolean
2071 gst_aggregator_pad_is_eos (GstAggregatorPad * pad)
2072 {
2073   gboolean is_eos;
2074
2075   PAD_LOCK (pad);
2076   is_eos = pad->priv->eos;
2077   PAD_UNLOCK (pad);
2078
2079   return is_eos;
2080 }
2081
2082 /**
2083  * gst_aggregator_merge_tags:
2084  * @self: a #GstAggregator
2085  * @tags: a #GstTagList to merge
2086  * @mode: the #GstTagMergeMode to use
2087  *
2088  * Adds tags to so-called pending tags, which will be processed
2089  * before pushing out data downstream.
2090  *
2091  * Note that this is provided for convenience, and the subclass is
2092  * not required to use this and can still do tag handling on its own.
2093  *
2094  * MT safe.
2095  */
2096 void
2097 gst_aggregator_merge_tags (GstAggregator * self,
2098     const GstTagList * tags, GstTagMergeMode mode)
2099 {
2100   GstTagList *otags;
2101
2102   g_return_if_fail (GST_IS_AGGREGATOR (self));
2103   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
2104
2105   /* FIXME Check if we can use OBJECT lock here! */
2106   GST_OBJECT_LOCK (self);
2107   if (tags)
2108     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
2109   otags = self->priv->tags;
2110   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
2111   if (otags)
2112     gst_tag_list_unref (otags);
2113   self->priv->tags_changed = TRUE;
2114   GST_OBJECT_UNLOCK (self);
2115 }
2116
2117 /**
2118  * gst_aggregator_set_latency:
2119  * @self: a #GstAggregator
2120  * @min_latency: minimum latency
2121  * @max_latency: maximum latency
2122  *
2123  * Lets #GstAggregator sub-classes tell the baseclass what their internal
2124  * latency is. Will also post a LATENCY message on the bus so the pipeline
2125  * can reconfigure its global latency.
2126  */
2127 void
2128 gst_aggregator_set_latency (GstAggregator * self,
2129     GstClockTime min_latency, GstClockTime max_latency)
2130 {
2131   gboolean changed = FALSE;
2132
2133   g_return_if_fail (GST_IS_AGGREGATOR (self));
2134   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (min_latency));
2135   g_return_if_fail (max_latency >= min_latency);
2136
2137   SRC_LOCK (self);
2138   if (self->priv->sub_latency_min != min_latency) {
2139     self->priv->sub_latency_min = min_latency;
2140     changed = TRUE;
2141   }
2142   if (self->priv->sub_latency_max != max_latency) {
2143     self->priv->sub_latency_max = max_latency;
2144     changed = TRUE;
2145   }
2146
2147   if (changed)
2148     SRC_BROADCAST (self);
2149   SRC_UNLOCK (self);
2150
2151   if (changed) {
2152     gst_element_post_message (GST_ELEMENT_CAST (self),
2153         gst_message_new_latency (GST_OBJECT_CAST (self)));
2154   }
2155 }