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