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_WARNING_OBJECT (self, "Invalid minimum latency, using 0");
1133     min = 0;
1134   }
1135
1136   if (G_UNLIKELY (min > max)) {
1137     GST_WARNING_OBJECT (self, "Minimum latency is greater than maximum latency "
1138         "(%" G_GINT64_FORMAT " > %" G_GINT64_FORMAT "). "
1139         "Clamping it at the maximum latency", min, max);
1140     min = max;
1141   }
1142
1143   self->priv->latency_live = live;
1144   self->priv->latency_min = min;
1145   self->priv->latency_max = max;
1146
1147   /* add our own */
1148   min += our_latency;
1149   min += self->priv->sub_latency_min;
1150   if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1151       && GST_CLOCK_TIME_IS_VALID (max))
1152     max += self->priv->sub_latency_max;
1153   else
1154     max = GST_CLOCK_TIME_NONE;
1155
1156   if (live && min > max) {
1157     GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1158         ("%s", "Latency too big"),
1159         ("The requested latency value is too big for the current pipeline.  "
1160             "Limiting to %" G_GINT64_FORMAT, max));
1161     min = max;
1162     /* FIXME: This could in theory become negative, but in
1163      * that case all is lost anyway */
1164     self->priv->latency -= min - max;
1165     /* FIXME: shouldn't we g_object_notify() the change here? */
1166   }
1167
1168   SRC_BROADCAST (self);
1169   SRC_UNLOCK (self);
1170
1171   GST_DEBUG_OBJECT (self, "configured latency live:%s min:%" G_GINT64_FORMAT
1172       " max:%" G_GINT64_FORMAT, live ? "true" : "false", min, max);
1173
1174   gst_query_set_latency (query, live, min, max);
1175
1176   return query_ret;
1177 }
1178
1179 /**
1180  * gst_aggregator_get_latency_unlocked:
1181  * @self: a #GstAggregator
1182  * @live: (out) (allow-none): whether @self is live
1183  * @min_latency: (out) (allow-none): the configured minimum latency of @self
1184  * @max_latency: (out) (allow-none): the configured maximum latency of @self
1185  *
1186  * Retreives the latency values reported by @self in response to the latency
1187  * query.
1188  *
1189  * Typically only called by subclasses.
1190  *
1191  * MUST be called with the src_lock held.
1192  */
1193 void
1194 gst_aggregator_get_latency_unlocked (GstAggregator * self, gboolean * live,
1195     GstClockTime * min_latency, GstClockTime * max_latency)
1196 {
1197   GstClockTime min, max;
1198
1199   g_return_if_fail (GST_IS_AGGREGATOR (self));
1200
1201   /* latency_min is never GST_CLOCK_TIME_NONE by construction */
1202   min = self->priv->latency_min;
1203   max = self->priv->latency_max;
1204
1205   /* add our own */
1206   min += self->priv->latency;
1207   min += self->priv->sub_latency_min;
1208   if (GST_CLOCK_TIME_IS_VALID (max)
1209       && GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max))
1210     max += self->priv->sub_latency_max;
1211   else
1212     max = GST_CLOCK_TIME_NONE;
1213
1214   if (live)
1215     *live = self->priv->latency_live;
1216   if (min_latency)
1217     *min_latency = min;
1218   if (max_latency)
1219     *max_latency = max;
1220 }
1221
1222 static gboolean
1223 gst_aggregator_send_event (GstElement * element, GstEvent * event)
1224 {
1225   GstAggregator *self = GST_AGGREGATOR (element);
1226
1227   GST_STATE_LOCK (element);
1228   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK &&
1229       GST_STATE (element) < GST_STATE_PAUSED) {
1230     gdouble rate;
1231     GstFormat fmt;
1232     GstSeekFlags flags;
1233     GstSeekType start_type, stop_type;
1234     gint64 start, stop;
1235
1236     gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1237         &start, &stop_type, &stop);
1238
1239     GST_OBJECT_LOCK (self);
1240     gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1241         stop_type, stop, NULL);
1242     self->priv->seqnum = gst_event_get_seqnum (event);
1243     GST_OBJECT_UNLOCK (self);
1244
1245     GST_DEBUG_OBJECT (element, "Storing segment %" GST_PTR_FORMAT, event);
1246   }
1247   GST_STATE_UNLOCK (element);
1248
1249
1250   return GST_ELEMENT_CLASS (aggregator_parent_class)->send_event (element,
1251       event);
1252 }
1253
1254 static gboolean
1255 gst_aggregator_default_src_query (GstAggregator * self, GstQuery * query)
1256 {
1257   gboolean res = TRUE;
1258
1259   switch (GST_QUERY_TYPE (query)) {
1260     case GST_QUERY_SEEKING:
1261     {
1262       GstFormat format;
1263
1264       /* don't pass it along as some (file)sink might claim it does
1265        * whereas with a collectpads in between that will not likely work */
1266       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
1267       gst_query_set_seeking (query, format, FALSE, 0, -1);
1268       res = TRUE;
1269
1270       goto discard;
1271     }
1272     case GST_QUERY_LATENCY:
1273       return gst_aggregator_query_latency (self, query);
1274     default:
1275       break;
1276   }
1277
1278   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1279
1280 discard:
1281   return res;
1282 }
1283
1284 static gboolean
1285 gst_aggregator_event_forward_func (GstPad * pad, gpointer user_data)
1286 {
1287   EventData *evdata = user_data;
1288   gboolean ret = TRUE;
1289   GstPad *peer = gst_pad_get_peer (pad);
1290   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1291
1292   if (peer) {
1293     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
1294     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
1295     gst_object_unref (peer);
1296   }
1297
1298   if (ret == FALSE) {
1299     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
1300       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
1301
1302     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
1303       GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
1304
1305       if (gst_pad_query (peer, seeking)) {
1306         gboolean seekable;
1307
1308         gst_query_parse_seeking (seeking, NULL, &seekable, NULL, NULL);
1309
1310         if (seekable == FALSE) {
1311           GST_INFO_OBJECT (pad,
1312               "Source not seekable, We failed but it does not matter!");
1313
1314           ret = TRUE;
1315         }
1316       } else {
1317         GST_ERROR_OBJECT (pad, "Query seeking FAILED");
1318       }
1319
1320       gst_query_unref (seeking);
1321     }
1322
1323     if (evdata->flush) {
1324       PAD_LOCK (aggpad);
1325       aggpad->priv->pending_flush_start = FALSE;
1326       aggpad->priv->pending_flush_stop = FALSE;
1327       PAD_UNLOCK (aggpad);
1328     }
1329   } else {
1330     evdata->one_actually_seeked = TRUE;
1331   }
1332
1333   evdata->result &= ret;
1334
1335   /* Always send to all pads */
1336   return FALSE;
1337 }
1338
1339 static EventData
1340 gst_aggregator_forward_event_to_all_sinkpads (GstAggregator * self,
1341     GstEvent * event, gboolean flush)
1342 {
1343   EventData evdata;
1344
1345   evdata.event = event;
1346   evdata.result = TRUE;
1347   evdata.flush = flush;
1348   evdata.one_actually_seeked = FALSE;
1349
1350   /* We first need to set all pads as flushing in a first pass
1351    * as flush_start flush_stop is sometimes sent synchronously
1352    * while we send the seek event */
1353   if (flush) {
1354     GList *l;
1355
1356     GST_OBJECT_LOCK (self);
1357     for (l = GST_ELEMENT_CAST (self)->sinkpads; l != NULL; l = l->next) {
1358       GstAggregatorPad *pad = l->data;
1359
1360       PAD_LOCK (pad);
1361       pad->priv->pending_flush_start = TRUE;
1362       pad->priv->pending_flush_stop = FALSE;
1363       PAD_UNLOCK (pad);
1364     }
1365     GST_OBJECT_UNLOCK (self);
1366   }
1367
1368   gst_pad_forward (self->srcpad, gst_aggregator_event_forward_func, &evdata);
1369
1370   gst_event_unref (event);
1371
1372   return evdata;
1373 }
1374
1375 static gboolean
1376 gst_aggregator_do_seek (GstAggregator * self, GstEvent * event)
1377 {
1378   gdouble rate;
1379   GstFormat fmt;
1380   GstSeekFlags flags;
1381   GstSeekType start_type, stop_type;
1382   gint64 start, stop;
1383   gboolean flush;
1384   EventData evdata;
1385   GstAggregatorPrivate *priv = self->priv;
1386
1387   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1388       &start, &stop_type, &stop);
1389
1390   GST_INFO_OBJECT (self, "starting SEEK");
1391
1392   flush = flags & GST_SEEK_FLAG_FLUSH;
1393
1394   GST_OBJECT_LOCK (self);
1395   if (flush) {
1396     priv->pending_flush_start = TRUE;
1397     priv->flush_seeking = TRUE;
1398   }
1399
1400   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1401       stop_type, stop, NULL);
1402   GST_OBJECT_UNLOCK (self);
1403
1404   /* forward the seek upstream */
1405   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, flush);
1406   event = NULL;
1407
1408   if (!evdata.result || !evdata.one_actually_seeked) {
1409     GST_OBJECT_LOCK (self);
1410     priv->flush_seeking = FALSE;
1411     priv->pending_flush_start = FALSE;
1412     GST_OBJECT_UNLOCK (self);
1413   }
1414
1415   GST_INFO_OBJECT (self, "seek done, result: %d", evdata.result);
1416
1417   return evdata.result;
1418 }
1419
1420 static gboolean
1421 gst_aggregator_default_src_event (GstAggregator * self, GstEvent * event)
1422 {
1423   EventData evdata;
1424   gboolean res = TRUE;
1425
1426   switch (GST_EVENT_TYPE (event)) {
1427     case GST_EVENT_SEEK:
1428     {
1429       gst_event_ref (event);
1430       res = gst_aggregator_do_seek (self, event);
1431       gst_event_unref (event);
1432       event = NULL;
1433       goto done;
1434     }
1435     case GST_EVENT_NAVIGATION:
1436     {
1437       /* navigation is rather pointless. */
1438       res = FALSE;
1439       gst_event_unref (event);
1440       goto done;
1441     }
1442     default:
1443     {
1444       break;
1445     }
1446   }
1447
1448   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE);
1449   res = evdata.result;
1450
1451 done:
1452   return res;
1453 }
1454
1455 static gboolean
1456 gst_aggregator_src_pad_event_func (GstPad * pad, GstObject * parent,
1457     GstEvent * event)
1458 {
1459   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1460
1461   return klass->src_event (GST_AGGREGATOR (parent), event);
1462 }
1463
1464 static gboolean
1465 gst_aggregator_src_pad_query_func (GstPad * pad, GstObject * parent,
1466     GstQuery * query)
1467 {
1468   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1469
1470   return klass->src_query (GST_AGGREGATOR (parent), query);
1471 }
1472
1473 static gboolean
1474 gst_aggregator_src_pad_activate_mode_func (GstPad * pad,
1475     GstObject * parent, GstPadMode mode, gboolean active)
1476 {
1477   GstAggregator *self = GST_AGGREGATOR (parent);
1478   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1479
1480   if (klass->src_activate) {
1481     if (klass->src_activate (self, mode, active) == FALSE) {
1482       return FALSE;
1483     }
1484   }
1485
1486   if (active == TRUE) {
1487     switch (mode) {
1488       case GST_PAD_MODE_PUSH:
1489       {
1490         GST_INFO_OBJECT (pad, "Activating pad!");
1491         gst_aggregator_start_srcpad_task (self);
1492         return TRUE;
1493       }
1494       default:
1495       {
1496         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
1497         return FALSE;
1498       }
1499     }
1500   }
1501
1502   /* deactivating */
1503   GST_INFO_OBJECT (self, "Deactivating srcpad");
1504   gst_aggregator_stop_srcpad_task (self, FALSE);
1505
1506   return TRUE;
1507 }
1508
1509 static gboolean
1510 gst_aggregator_default_sink_query (GstAggregator * self,
1511     GstAggregatorPad * aggpad, GstQuery * query)
1512 {
1513   GstPad *pad = GST_PAD (aggpad);
1514
1515   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1516 }
1517
1518 static void
1519 gst_aggregator_finalize (GObject * object)
1520 {
1521   GstAggregator *self = (GstAggregator *) object;
1522
1523   g_mutex_clear (&self->priv->src_lock);
1524   g_cond_clear (&self->priv->src_cond);
1525
1526   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
1527 }
1528
1529 /*
1530  * gst_aggregator_set_latency_property:
1531  * @agg: a #GstAggregator
1532  * @latency: the new latency value.
1533  *
1534  * Sets the new latency value to @latency. This value is used to limit the
1535  * amount of time a pad waits for data to appear before considering the pad
1536  * as unresponsive.
1537  */
1538 static void
1539 gst_aggregator_set_latency_property (GstAggregator * self, gint64 latency)
1540 {
1541   gboolean changed;
1542   GstClockTime min, max;
1543
1544   g_return_if_fail (GST_IS_AGGREGATOR (self));
1545   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (latency));
1546
1547   SRC_LOCK (self);
1548   if (self->priv->latency_live) {
1549     min = self->priv->latency_min;
1550     max = self->priv->latency_max;
1551     /* add our own */
1552     min += latency;
1553     min += self->priv->sub_latency_min;
1554     if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1555         && GST_CLOCK_TIME_IS_VALID (max))
1556       max += self->priv->sub_latency_max;
1557     else
1558       max = GST_CLOCK_TIME_NONE;
1559
1560     if (GST_CLOCK_TIME_IS_VALID (max) && min > max) {
1561       GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1562           ("%s", "Latency too big"),
1563           ("The requested latency value is too big for the latency in the "
1564               "current pipeline.  Limiting to %" G_GINT64_FORMAT, max));
1565       /* FIXME: This could in theory become negative, but in
1566        * that case all is lost anyway */
1567       latency -= min - max;
1568       /* FIXME: shouldn't we g_object_notify() the change here? */
1569     }
1570   }
1571
1572   changed = (self->priv->latency != latency);
1573   self->priv->latency = latency;
1574
1575   if (changed)
1576     SRC_BROADCAST (self);
1577   SRC_UNLOCK (self);
1578
1579   if (changed)
1580     gst_element_post_message (GST_ELEMENT_CAST (self),
1581         gst_message_new_latency (GST_OBJECT_CAST (self)));
1582 }
1583
1584 /*
1585  * gst_aggregator_get_latency_property:
1586  * @agg: a #GstAggregator
1587  *
1588  * Gets the latency value. See gst_aggregator_set_latency for
1589  * more details.
1590  *
1591  * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
1592  * before a pad is deemed unresponsive. A value of -1 means an
1593  * unlimited time.
1594  */
1595 static gint64
1596 gst_aggregator_get_latency_property (GstAggregator * agg)
1597 {
1598   gint64 res;
1599
1600   g_return_val_if_fail (GST_IS_AGGREGATOR (agg), -1);
1601
1602   GST_OBJECT_LOCK (agg);
1603   res = agg->priv->latency;
1604   GST_OBJECT_UNLOCK (agg);
1605
1606   return res;
1607 }
1608
1609 static void
1610 gst_aggregator_set_property (GObject * object, guint prop_id,
1611     const GValue * value, GParamSpec * pspec)
1612 {
1613   GstAggregator *agg = GST_AGGREGATOR (object);
1614
1615   switch (prop_id) {
1616     case PROP_LATENCY:
1617       gst_aggregator_set_latency_property (agg, g_value_get_int64 (value));
1618       break;
1619     default:
1620       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1621       break;
1622   }
1623 }
1624
1625 static void
1626 gst_aggregator_get_property (GObject * object, guint prop_id,
1627     GValue * value, GParamSpec * pspec)
1628 {
1629   GstAggregator *agg = GST_AGGREGATOR (object);
1630
1631   switch (prop_id) {
1632     case PROP_LATENCY:
1633       g_value_set_int64 (value, gst_aggregator_get_latency_property (agg));
1634       break;
1635     default:
1636       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1637       break;
1638   }
1639 }
1640
1641 /* GObject vmethods implementations */
1642 static void
1643 gst_aggregator_class_init (GstAggregatorClass * klass)
1644 {
1645   GObjectClass *gobject_class = (GObjectClass *) klass;
1646   GstElementClass *gstelement_class = (GstElementClass *) klass;
1647
1648   aggregator_parent_class = g_type_class_peek_parent (klass);
1649   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1650
1651   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1652       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1653
1654   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1655
1656   klass->sink_event = gst_aggregator_default_sink_event;
1657   klass->sink_query = gst_aggregator_default_sink_query;
1658
1659   klass->src_event = gst_aggregator_default_src_event;
1660   klass->src_query = gst_aggregator_default_src_query;
1661
1662   gstelement_class->request_new_pad =
1663       GST_DEBUG_FUNCPTR (gst_aggregator_request_new_pad);
1664   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_aggregator_send_event);
1665   gstelement_class->release_pad =
1666       GST_DEBUG_FUNCPTR (gst_aggregator_release_pad);
1667   gstelement_class->change_state =
1668       GST_DEBUG_FUNCPTR (gst_aggregator_change_state);
1669
1670   gobject_class->set_property = gst_aggregator_set_property;
1671   gobject_class->get_property = gst_aggregator_get_property;
1672   gobject_class->finalize = gst_aggregator_finalize;
1673
1674   g_object_class_install_property (gobject_class, PROP_LATENCY,
1675       g_param_spec_int64 ("latency", "Buffer latency",
1676           "Additional latency in live mode to allow upstream "
1677           "to take longer to produce buffers for the current "
1678           "position", 0,
1679           (G_MAXLONG == G_MAXINT64) ? G_MAXINT64 : (G_MAXLONG * GST_SECOND - 1),
1680           DEFAULT_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1681
1682   GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_stop_pad);
1683 }
1684
1685 static void
1686 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1687 {
1688   GstPadTemplate *pad_template;
1689   GstAggregatorPrivate *priv;
1690
1691   g_return_if_fail (klass->aggregate != NULL);
1692
1693   self->priv =
1694       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1695       GstAggregatorPrivate);
1696
1697   priv = self->priv;
1698
1699   pad_template =
1700       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1701   g_return_if_fail (pad_template != NULL);
1702
1703   priv->padcount = -1;
1704   priv->tags_changed = FALSE;
1705
1706   self->priv->latency_live = FALSE;
1707   self->priv->latency_min = self->priv->sub_latency_min = 0;
1708   self->priv->latency_max = self->priv->sub_latency_max = 0;
1709   gst_aggregator_reset_flow_values (self);
1710
1711   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1712
1713   gst_pad_set_event_function (self->srcpad,
1714       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_event_func));
1715   gst_pad_set_query_function (self->srcpad,
1716       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_query_func));
1717   gst_pad_set_activatemode_function (self->srcpad,
1718       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_activate_mode_func));
1719
1720   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1721
1722   self->priv->latency = DEFAULT_LATENCY;
1723
1724   g_mutex_init (&self->priv->src_lock);
1725   g_cond_init (&self->priv->src_cond);
1726 }
1727
1728 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1729  * method to get to the padtemplates */
1730 GType
1731 gst_aggregator_get_type (void)
1732 {
1733   static volatile gsize type = 0;
1734
1735   if (g_once_init_enter (&type)) {
1736     GType _type;
1737     static const GTypeInfo info = {
1738       sizeof (GstAggregatorClass),
1739       NULL,
1740       NULL,
1741       (GClassInitFunc) gst_aggregator_class_init,
1742       NULL,
1743       NULL,
1744       sizeof (GstAggregator),
1745       0,
1746       (GInstanceInitFunc) gst_aggregator_init,
1747     };
1748
1749     _type = g_type_register_static (GST_TYPE_ELEMENT,
1750         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1751     g_once_init_leave (&type, _type);
1752   }
1753   return type;
1754 }
1755
1756 static GstFlowReturn
1757 gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1758 {
1759   GstBuffer *actual_buf = buffer;
1760   GstAggregator *self = GST_AGGREGATOR (object);
1761   GstAggregatorPrivate *priv = self->priv;
1762   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1763   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1764   GstFlowReturn flow_return;
1765
1766   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1767
1768   PAD_FLUSH_LOCK (aggpad);
1769
1770   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1771     goto flushing;
1772
1773   PAD_LOCK (aggpad);
1774   if (aggpad->priv->pending_eos == TRUE)
1775     goto eos;
1776
1777   while (aggpad->priv->buffer
1778       && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1779     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1780     PAD_WAIT_EVENT (aggpad);
1781   }
1782   PAD_UNLOCK (aggpad);
1783
1784   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1785     goto flushing;
1786
1787   if (aggclass->clip) {
1788     aggclass->clip (self, aggpad, buffer, &actual_buf);
1789   }
1790
1791   SRC_LOCK (self);
1792   PAD_LOCK (aggpad);
1793   if (aggpad->priv->buffer)
1794     gst_buffer_unref (aggpad->priv->buffer);
1795   aggpad->priv->buffer = actual_buf;
1796   PAD_UNLOCK (aggpad);
1797   PAD_FLUSH_UNLOCK (aggpad);
1798
1799   SRC_BROADCAST (self);
1800   SRC_UNLOCK (self);
1801
1802   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1803
1804   GST_OBJECT_LOCK (self);
1805   flow_return = priv->flow_return;
1806   GST_OBJECT_UNLOCK (self);
1807
1808   return flow_return;
1809
1810 flushing:
1811   PAD_FLUSH_UNLOCK (aggpad);
1812
1813   gst_buffer_unref (buffer);
1814   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1815
1816   return GST_FLOW_FLUSHING;
1817
1818 eos:
1819   PAD_UNLOCK (aggpad);
1820   PAD_FLUSH_UNLOCK (aggpad);
1821
1822   gst_buffer_unref (buffer);
1823   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1824
1825   return GST_FLOW_EOS;
1826 }
1827
1828 static gboolean
1829 gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
1830     GstQuery * query)
1831 {
1832   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1833   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1834
1835   if (GST_QUERY_IS_SERIALIZED (query)) {
1836     PAD_LOCK (aggpad);
1837
1838     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE) {
1839       PAD_UNLOCK (aggpad);
1840       goto flushing;
1841     }
1842
1843     while (aggpad->priv->buffer
1844         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1845       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1846       PAD_WAIT_EVENT (aggpad);
1847     }
1848     PAD_UNLOCK (aggpad);
1849
1850     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1851       goto flushing;
1852   }
1853
1854   return klass->sink_query (GST_AGGREGATOR (parent),
1855       GST_AGGREGATOR_PAD (pad), query);
1856
1857 flushing:
1858   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping query");
1859   return FALSE;
1860 }
1861
1862 static gboolean
1863 gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
1864     GstEvent * event)
1865 {
1866   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1867   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1868
1869   if (GST_EVENT_IS_SERIALIZED (event) && GST_EVENT_TYPE (event) != GST_EVENT_EOS
1870       && GST_EVENT_TYPE (event) != GST_EVENT_SEGMENT_DONE) {
1871     PAD_LOCK (aggpad);
1872
1873     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1874         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
1875       PAD_UNLOCK (aggpad);
1876       goto flushing;
1877     }
1878
1879     while (aggpad->priv->buffer
1880         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1881       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1882       PAD_WAIT_EVENT (aggpad);
1883     }
1884     PAD_UNLOCK (aggpad);
1885
1886     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1887         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP)
1888       goto flushing;
1889   }
1890
1891   return klass->sink_event (GST_AGGREGATOR (parent),
1892       GST_AGGREGATOR_PAD (pad), event);
1893
1894 flushing:
1895   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping event");
1896   if (GST_EVENT_IS_STICKY (event))
1897     gst_pad_store_sticky_event (pad, event);
1898   gst_event_unref (event);
1899   return FALSE;
1900 }
1901
1902 static gboolean
1903 gst_aggregator_pad_activate_mode_func (GstPad * pad,
1904     GstObject * parent, GstPadMode mode, gboolean active)
1905 {
1906   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1907
1908   if (active == FALSE) {
1909     PAD_LOCK (aggpad);
1910     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1911     gst_buffer_replace (&aggpad->priv->buffer, NULL);
1912     PAD_BROADCAST_EVENT (aggpad);
1913     PAD_UNLOCK (aggpad);
1914   } else {
1915     PAD_LOCK (aggpad);
1916     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1917     PAD_BROADCAST_EVENT (aggpad);
1918     PAD_UNLOCK (aggpad);
1919   }
1920
1921   return TRUE;
1922 }
1923
1924 /***********************************
1925  * GstAggregatorPad implementation  *
1926  ************************************/
1927 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1928
1929 static void
1930 gst_aggregator_pad_constructed (GObject * object)
1931 {
1932   GstPad *pad = GST_PAD (object);
1933
1934   gst_pad_set_chain_function (pad,
1935       GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
1936   gst_pad_set_event_function (pad,
1937       GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func));
1938   gst_pad_set_query_function (pad,
1939       GST_DEBUG_FUNCPTR (gst_aggregator_pad_query_func));
1940   gst_pad_set_activatemode_function (pad,
1941       GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
1942 }
1943
1944 static void
1945 gst_aggregator_pad_finalize (GObject * object)
1946 {
1947   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1948
1949   g_cond_clear (&pad->priv->event_cond);
1950   g_mutex_clear (&pad->priv->flush_lock);
1951   g_mutex_clear (&pad->priv->lock);
1952
1953   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->finalize (object);
1954 }
1955
1956 static void
1957 gst_aggregator_pad_dispose (GObject * object)
1958 {
1959   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1960
1961   gst_aggregator_pad_drop_buffer (pad);
1962
1963   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->dispose (object);
1964 }
1965
1966 static void
1967 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
1968 {
1969   GObjectClass *gobject_class = (GObjectClass *) klass;
1970
1971   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
1972
1973   gobject_class->constructed = gst_aggregator_pad_constructed;
1974   gobject_class->finalize = gst_aggregator_pad_finalize;
1975   gobject_class->dispose = gst_aggregator_pad_dispose;
1976 }
1977
1978 static void
1979 gst_aggregator_pad_init (GstAggregatorPad * pad)
1980 {
1981   pad->priv =
1982       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
1983       GstAggregatorPadPrivate);
1984
1985   pad->priv->buffer = NULL;
1986   g_cond_init (&pad->priv->event_cond);
1987
1988   g_mutex_init (&pad->priv->flush_lock);
1989   g_mutex_init (&pad->priv->lock);
1990 }
1991
1992 /**
1993  * gst_aggregator_pad_steal_buffer:
1994  * @pad: the pad to get buffer from
1995  *
1996  * Steal the ref to the buffer currently queued in @pad.
1997  *
1998  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
1999  *   queued. You should unref the buffer after usage.
2000  */
2001 GstBuffer *
2002 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
2003 {
2004   GstBuffer *buffer = NULL;
2005
2006   PAD_LOCK (pad);
2007   if (pad->priv->buffer) {
2008     GST_TRACE_OBJECT (pad, "Consuming buffer");
2009     buffer = pad->priv->buffer;
2010     pad->priv->buffer = NULL;
2011     if (pad->priv->pending_eos) {
2012       pad->priv->pending_eos = FALSE;
2013       pad->priv->eos = TRUE;
2014     }
2015     PAD_BROADCAST_EVENT (pad);
2016     GST_DEBUG_OBJECT (pad, "Consumed: %" GST_PTR_FORMAT, buffer);
2017   }
2018   PAD_UNLOCK (pad);
2019
2020   return buffer;
2021 }
2022
2023 /**
2024  * gst_aggregator_pad_drop_buffer:
2025  * @pad: the pad where to drop any pending buffer
2026  *
2027  * Drop the buffer currently queued in @pad.
2028  *
2029  * Returns: TRUE if there was a buffer queued in @pad, or FALSE if not.
2030  */
2031 gboolean
2032 gst_aggregator_pad_drop_buffer (GstAggregatorPad * pad)
2033 {
2034   GstBuffer *buf;
2035
2036   buf = gst_aggregator_pad_steal_buffer (pad);
2037
2038   if (buf == NULL)
2039     return FALSE;
2040
2041   gst_buffer_unref (buf);
2042   return TRUE;
2043 }
2044
2045 /**
2046  * gst_aggregator_pad_get_buffer:
2047  * @pad: the pad to get buffer from
2048  *
2049  * Returns: (transfer full): A reference to the buffer in @pad or
2050  * NULL if no buffer was queued. You should unref the buffer after
2051  * usage.
2052  */
2053 GstBuffer *
2054 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
2055 {
2056   GstBuffer *buffer = NULL;
2057
2058   PAD_LOCK (pad);
2059   if (pad->priv->buffer)
2060     buffer = gst_buffer_ref (pad->priv->buffer);
2061   PAD_UNLOCK (pad);
2062
2063   return buffer;
2064 }
2065
2066 gboolean
2067 gst_aggregator_pad_is_eos (GstAggregatorPad * pad)
2068 {
2069   gboolean is_eos;
2070
2071   PAD_LOCK (pad);
2072   is_eos = pad->priv->eos;
2073   PAD_UNLOCK (pad);
2074
2075   return is_eos;
2076 }
2077
2078 /**
2079  * gst_aggregator_merge_tags:
2080  * @self: a #GstAggregator
2081  * @tags: a #GstTagList to merge
2082  * @mode: the #GstTagMergeMode to use
2083  *
2084  * Adds tags to so-called pending tags, which will be processed
2085  * before pushing out data downstream.
2086  *
2087  * Note that this is provided for convenience, and the subclass is
2088  * not required to use this and can still do tag handling on its own.
2089  *
2090  * MT safe.
2091  */
2092 void
2093 gst_aggregator_merge_tags (GstAggregator * self,
2094     const GstTagList * tags, GstTagMergeMode mode)
2095 {
2096   GstTagList *otags;
2097
2098   g_return_if_fail (GST_IS_AGGREGATOR (self));
2099   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
2100
2101   /* FIXME Check if we can use OBJECT lock here! */
2102   GST_OBJECT_LOCK (self);
2103   if (tags)
2104     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
2105   otags = self->priv->tags;
2106   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
2107   if (otags)
2108     gst_tag_list_unref (otags);
2109   self->priv->tags_changed = TRUE;
2110   GST_OBJECT_UNLOCK (self);
2111 }
2112
2113 /**
2114  * gst_aggregator_set_latency:
2115  * @self: a #GstAggregator
2116  * @min_latency: minimum latency
2117  * @max_latency: maximum latency
2118  *
2119  * Lets #GstAggregator sub-classes tell the baseclass what their internal
2120  * latency is. Will also post a LATENCY message on the bus so the pipeline
2121  * can reconfigure its global latency.
2122  */
2123 void
2124 gst_aggregator_set_latency (GstAggregator * self,
2125     GstClockTime min_latency, GstClockTime max_latency)
2126 {
2127   gboolean changed = FALSE;
2128
2129   g_return_if_fail (GST_IS_AGGREGATOR (self));
2130   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (min_latency));
2131   g_return_if_fail (max_latency >= min_latency);
2132
2133   SRC_LOCK (self);
2134   if (self->priv->sub_latency_min != min_latency) {
2135     self->priv->sub_latency_min = min_latency;
2136     changed = TRUE;
2137   }
2138   if (self->priv->sub_latency_max != max_latency) {
2139     self->priv->sub_latency_max = max_latency;
2140     changed = TRUE;
2141   }
2142
2143   if (changed)
2144     SRC_BROADCAST (self);
2145   SRC_UNLOCK (self);
2146
2147   if (changed) {
2148     gst_element_post_message (GST_ELEMENT_CAST (self),
2149         gst_message_new_latency (GST_OBJECT_CAST (self)));
2150   }
2151 }