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