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