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   GstBuffer *clipped_buffer;
 216   guint num_buffers;
 217   GstClockTime head_position;
 218   GstClockTime tail_position;
 219   GstClockTime head_time;
 220   GstClockTime tail_time;
 221   GstClockTime time_level;
 222   GstSegment head_segment;      /* segment before the queue */
 223
 224   gboolean negotiated;
 225
 226   gboolean eos;
 227
 228   GMutex lock;
 229   GCond event_cond;
 230   /* This lock prevents a flush start processing happening while
 231    * the chain function is also happening.
 232    */
 233   GMutex flush_lock;
 234 };
 235
 236 /* Must be called with PAD_LOCK held */
 237 static void
 238 gst_aggregator_pad_reset_unlocked (GstAggregatorPad * aggpad)
 239 {
 240   aggpad->priv->pending_eos = FALSE;
 241   aggpad->priv->eos = FALSE;
 242   aggpad->priv->flow_return = GST_FLOW_OK;
 243   GST_OBJECT_LOCK (aggpad);
 244   gst_segment_init (&aggpad->segment, GST_FORMAT_UNDEFINED);
 245   gst_segment_init (&aggpad->priv->head_segment, GST_FORMAT_UNDEFINED);
 246   GST_OBJECT_UNLOCK (aggpad);
 247   aggpad->priv->head_position = GST_CLOCK_TIME_NONE;
 248   aggpad->priv->tail_position = GST_CLOCK_TIME_NONE;
 249   aggpad->priv->head_time = GST_CLOCK_TIME_NONE;
 250   aggpad->priv->tail_time = GST_CLOCK_TIME_NONE;
 251   aggpad->priv->time_level = 0;
 252   aggpad->priv->first_buffer = TRUE;
 253 }
 254
 255 static gboolean
 256 gst_aggregator_pad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
 257 {
 258   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 259
 260   PAD_LOCK (aggpad);
 261   gst_aggregator_pad_reset_unlocked (aggpad);
 262   PAD_UNLOCK (aggpad);
 263
 264   if (klass->flush)
 265     return klass->flush (aggpad, agg);
 266
 267   return TRUE;
 268 }
 269
 270 /*************************************
 271  * GstAggregator implementation  *
 272  *************************************/
 273 static GstElementClass *aggregator_parent_class = NULL;
 274
 275 /* All members are protected by the object lock unless otherwise noted */
 276
 277 struct _GstAggregatorPrivate
 278 {
 279   gint max_padserial;
 280
 281   /* Our state is >= PAUSED */
 282   gboolean running;             /* protected by src_lock */
 283
 284   gint seqnum;
 285   gboolean send_stream_start;   /* protected by srcpad stream lock */
 286   gboolean send_segment;
 287   gboolean flush_seeking;
 288   gboolean pending_flush_start;
 289   gboolean send_eos;            /* protected by srcpad stream lock */
 290
 291   GstCaps *srccaps;             /* protected by the srcpad stream lock */
 292
 293   GstTagList *tags;
 294   gboolean tags_changed;
 295
 296   gboolean peer_latency_live;   /* protected by src_lock */
 297   GstClockTime peer_latency_min;        /* protected by src_lock */
 298   GstClockTime peer_latency_max;        /* protected by src_lock */
 299   gboolean has_peer_latency;    /* protected by src_lock */
 300
 301   GstClockTime sub_latency_min; /* protected by src_lock */
 302   GstClockTime sub_latency_max; /* protected by src_lock */
 303
 304   /* aggregate */
 305   GstClockID aggregate_id;      /* protected by src_lock */
 306   GMutex src_lock;
 307   GCond src_cond;
 308
 309   gboolean first_buffer;        /* protected by object lock */
 310   GstAggregatorStartTimeSelection start_time_selection;
 311   GstClockTime start_time;
 312
 313   /* protected by the object lock */
 314   GstQuery *allocation_query;
 315   GstAllocator *allocator;
 316   GstBufferPool *pool;
 317   GstAllocationParams allocation_params;
 318
 319   /* properties */
 320   gint64 latency;               /* protected by both src_lock and all pad locks */
 321 };
 322
 323 typedef struct
 324 {
 325   GstEvent *event;
 326   gboolean result;
 327   gboolean flush;
 328   gboolean only_to_active_pads;
 329
 330   gboolean one_actually_seeked;
 331 } EventData;
 332
 333 #define DEFAULT_LATENCY              0
 334 #define DEFAULT_START_TIME_SELECTION GST_AGGREGATOR_START_TIME_SELECTION_ZERO
 335 #define DEFAULT_START_TIME           (-1)
 336
 337 enum
 338 {
 339   PROP_0,
 340   PROP_LATENCY,
 341   PROP_START_TIME_SELECTION,
 342   PROP_START_TIME,
 343   PROP_LAST
 344 };
 345
 346 static GstFlowReturn gst_aggregator_pad_chain_internal (GstAggregator * self,
 347     GstAggregatorPad * aggpad, GstBuffer * buffer, gboolean head);
 348
 349 /**
 350  * gst_aggregator_iterate_sinkpads:
 351  * @self: The #GstAggregator
 352  * @func: (scope call): The function to call.
 353  * @user_data: (closure): The data to pass to @func.
 354  *
 355  * Iterate the sinkpads of aggregator to call a function on them.
 356  *
 357  * This method guarantees that @func will be called only once for each
 358  * sink pad.
 359  */
 360 gboolean
 361 gst_aggregator_iterate_sinkpads (GstAggregator * self,
 362     GstAggregatorPadForeachFunc func, gpointer user_data)
 363 {
 364   gboolean result = FALSE;
 365   GstIterator *iter;
 366   gboolean done = FALSE;
 367   GValue item = { 0, };
 368   GList *seen_pads = NULL;
 369
 370   iter = gst_element_iterate_sink_pads (GST_ELEMENT (self));
 371
 372   if (!iter)
 373     goto no_iter;
 374
 375   while (!done) {
 376     switch (gst_iterator_next (iter, &item)) {
 377       case GST_ITERATOR_OK:
 378       {
 379         GstAggregatorPad *pad;
 380
 381         pad = g_value_get_object (&item);
 382
 383         /* if already pushed, skip. FIXME, find something faster to tag pads */
 384         if (pad == NULL || g_list_find (seen_pads, pad)) {
 385           g_value_reset (&item);
 386           break;
 387         }
 388
 389         GST_LOG_OBJECT (pad, "calling function %s on pad",
 390             GST_DEBUG_FUNCPTR_NAME (func));
 391
 392         result = func (self, pad, user_data);
 393
 394         done = !result;
 395
 396         seen_pads = g_list_prepend (seen_pads, pad);
 397
 398         g_value_reset (&item);
 399         break;
 400       }
 401       case GST_ITERATOR_RESYNC:
 402         gst_iterator_resync (iter);
 403         break;
 404       case GST_ITERATOR_ERROR:
 405         GST_ERROR_OBJECT (self,
 406             "Could not iterate over internally linked pads");
 407         done = TRUE;
 408         break;
 409       case GST_ITERATOR_DONE:
 410         done = TRUE;
 411         break;
 412     }
 413   }
 414   g_value_unset (&item);
 415   gst_iterator_free (iter);
 416
 417   if (seen_pads == NULL) {
 418     GST_DEBUG_OBJECT (self, "No pad seen");
 419     return FALSE;
 420   }
 421
 422   g_list_free (seen_pads);
 423
 424 no_iter:
 425   return result;
 426 }
 427
 428 static gboolean
 429 gst_aggregator_pad_queue_is_empty (GstAggregatorPad * pad)
 430 {
 431   return (g_queue_peek_tail (&pad->priv->buffers) == NULL &&
 432       pad->priv->clipped_buffer == NULL);
 433 }
 434
 435 static gboolean
 436 gst_aggregator_check_pads_ready (GstAggregator * self)
 437 {
 438   GstAggregatorPad *pad;
 439   GList *l, *sinkpads;
 440   gboolean have_buffer = TRUE;
 441   gboolean have_event = FALSE;
 442
 443   GST_LOG_OBJECT (self, "checking pads");
 444
 445   GST_OBJECT_LOCK (self);
 446
 447   sinkpads = GST_ELEMENT_CAST (self)->sinkpads;
 448   if (sinkpads == NULL)
 449     goto no_sinkpads;
 450
 451   for (l = sinkpads; l != NULL; l = l->next) {
 452     pad = l->data;
 453
 454     PAD_LOCK (pad);
 455
 456     if (pad->priv->num_buffers == 0) {
 457       if (!gst_aggregator_pad_queue_is_empty (pad))
 458         have_event = TRUE;
 459       if (!pad->priv->eos) {
 460         have_buffer = FALSE;
 461
 462         /* If not live we need data on all pads, so leave the loop */
 463         if (!self->priv->peer_latency_live) {
 464           PAD_UNLOCK (pad);
 465           goto pad_not_ready;
 466         }
 467       }
 468     } else if (self->priv->peer_latency_live) {
 469       /* In live mode, having a single pad with buffers is enough to
 470        * generate a start time from it. In non-live mode all pads need
 471        * to have a buffer
 472        */
 473       self->priv->first_buffer = FALSE;
 474     }
 475
 476     PAD_UNLOCK (pad);
 477   }
 478
 479   if (!have_buffer && !have_event)
 480     goto pad_not_ready;
 481
 482   if (have_buffer)
 483     self->priv->first_buffer = FALSE;
 484
 485   GST_OBJECT_UNLOCK (self);
 486   GST_LOG_OBJECT (self, "pads are ready");
 487   return TRUE;
 488
 489 no_sinkpads:
 490   {
 491     GST_LOG_OBJECT (self, "pads not ready: no sink pads");
 492     GST_OBJECT_UNLOCK (self);
 493     return FALSE;
 494   }
 495 pad_not_ready:
 496   {
 497     if (have_event)
 498       GST_LOG_OBJECT (pad, "pad not ready to be aggregated yet,"
 499           " but waking up for serialized event");
 500     else
 501       GST_LOG_OBJECT (pad, "pad not ready to be aggregated yet");
 502     GST_OBJECT_UNLOCK (self);
 503     return have_event;
 504   }
 505 }
 506
 507 static void
 508 gst_aggregator_reset_flow_values (GstAggregator * self)
 509 {
 510   GST_OBJECT_LOCK (self);
 511   self->priv->send_stream_start = TRUE;
 512   self->priv->send_segment = TRUE;
 513   gst_segment_init (&self->segment, GST_FORMAT_TIME);
 514   self->priv->first_buffer = TRUE;
 515   GST_OBJECT_UNLOCK (self);
 516 }
 517
 518 static inline void
 519 gst_aggregator_push_mandatory_events (GstAggregator * self)
 520 {
 521   GstAggregatorPrivate *priv = self->priv;
 522   GstEvent *segment = NULL;
 523   GstEvent *tags = NULL;
 524
 525   if (self->priv->send_stream_start) {
 526     gchar s_id[32];
 527
 528     GST_INFO_OBJECT (self, "pushing stream start");
 529     /* stream-start (FIXME: create id based on input ids) */
 530     g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
 531     if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
 532       GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
 533     }
 534     self->priv->send_stream_start = FALSE;
 535   }
 536
 537   if (self->priv->srccaps) {
 538
 539     GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
 540         self->priv->srccaps);
 541     if (!gst_pad_push_event (self->srcpad,
 542             gst_event_new_caps (self->priv->srccaps))) {
 543       GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
 544     }
 545     gst_caps_unref (self->priv->srccaps);
 546     self->priv->srccaps = NULL;
 547   }
 548
 549   GST_OBJECT_LOCK (self);
 550   if (self->priv->send_segment && !self->priv->flush_seeking) {
 551     segment = gst_event_new_segment (&self->segment);
 552
 553     if (!self->priv->seqnum)
 554       self->priv->seqnum = gst_event_get_seqnum (segment);
 555     else
 556       gst_event_set_seqnum (segment, self->priv->seqnum);
 557     self->priv->send_segment = FALSE;
 558
 559     GST_DEBUG_OBJECT (self, "pushing segment %" GST_PTR_FORMAT, segment);
 560   }
 561
 562   if (priv->tags && priv->tags_changed && !self->priv->flush_seeking) {
 563     tags = gst_event_new_tag (gst_tag_list_ref (priv->tags));
 564     priv->tags_changed = FALSE;
 565   }
 566   GST_OBJECT_UNLOCK (self);
 567
 568   if (segment)
 569     gst_pad_push_event (self->srcpad, segment);
 570   if (tags)
 571     gst_pad_push_event (self->srcpad, tags);
 572
 573 }
 574
 575 /**
 576  * gst_aggregator_set_src_caps:
 577  * @self: The #GstAggregator
 578  * @caps: The #GstCaps to set on the src pad.
 579  *
 580  * Sets the caps to be used on the src pad.
 581  */
 582 void
 583 gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps)
 584 {
 585   GST_PAD_STREAM_LOCK (self->srcpad);
 586   gst_caps_replace (&self->priv->srccaps, caps);
 587   gst_aggregator_push_mandatory_events (self);
 588   GST_PAD_STREAM_UNLOCK (self->srcpad);
 589 }
 590
 591 /**
 592  * gst_aggregator_finish_buffer:
 593  * @self: The #GstAggregator
 594  * @buffer: (transfer full): the #GstBuffer to push.
 595  *
 596  * This method will push the provided output buffer downstream. If needed,
 597  * mandatory events such as stream-start, caps, and segment events will be
 598  * sent before pushing the buffer.
 599  */
 600 GstFlowReturn
 601 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 602 {
 603   gst_aggregator_push_mandatory_events (self);
 604
 605   GST_OBJECT_LOCK (self);
 606   if (!self->priv->flush_seeking && gst_pad_is_active (self->srcpad)) {
 607     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 608     GST_OBJECT_UNLOCK (self);
 609     return gst_pad_push (self->srcpad, buffer);
 610   } else {
 611     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 612         self->priv->flush_seeking, gst_pad_is_active (self->srcpad));
 613     GST_OBJECT_UNLOCK (self);
 614     gst_buffer_unref (buffer);
 615     return GST_FLOW_OK;
 616   }
 617 }
 618
 619 static void
 620 gst_aggregator_push_eos (GstAggregator * self)
 621 {
 622   GstEvent *event;
 623   gst_aggregator_push_mandatory_events (self);
 624
 625   event = gst_event_new_eos ();
 626
 627   GST_OBJECT_LOCK (self);
 628   self->priv->send_eos = FALSE;
 629   gst_event_set_seqnum (event, self->priv->seqnum);
 630   GST_OBJECT_UNLOCK (self);
 631
 632   gst_pad_push_event (self->srcpad, event);
 633 }
 634
 635 static GstClockTime
 636 gst_aggregator_get_next_time (GstAggregator * self)
 637 {
 638   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 639
 640   if (klass->get_next_time)
 641     return klass->get_next_time (self);
 642
 643   return GST_CLOCK_TIME_NONE;
 644 }
 645
 646 static gboolean
 647 gst_aggregator_wait_and_check (GstAggregator * self, gboolean * timeout)
 648 {
 649   GstClockTime latency;
 650   GstClockTime start;
 651   gboolean res;
 652
 653   *timeout = FALSE;
 654
 655   SRC_LOCK (self);
 656
 657   latency = gst_aggregator_get_latency_unlocked (self);
 658
 659   if (gst_aggregator_check_pads_ready (self)) {
 660     GST_DEBUG_OBJECT (self, "all pads have data");
 661     SRC_UNLOCK (self);
 662
 663     return TRUE;
 664   }
 665
 666   /* Before waiting, check if we're actually still running */
 667   if (!self->priv->running || !self->priv->send_eos) {
 668     SRC_UNLOCK (self);
 669
 670     return FALSE;
 671   }
 672
 673   start = gst_aggregator_get_next_time (self);
 674
 675   /* If we're not live, or if we use the running time
 676    * of the first buffer as start time, we wait until
 677    * all pads have buffers.
 678    * Otherwise (i.e. if we are live!), we wait on the clock
 679    * and if a pad does not have a buffer in time we ignore
 680    * that pad.
 681    */
 682   GST_OBJECT_LOCK (self);
 683   if (!GST_CLOCK_TIME_IS_VALID (latency) ||
 684       !GST_IS_CLOCK (GST_ELEMENT_CLOCK (self)) ||
 685       !GST_CLOCK_TIME_IS_VALID (start) ||
 686       (self->priv->first_buffer
 687           && self->priv->start_time_selection ==
 688           GST_AGGREGATOR_START_TIME_SELECTION_FIRST)) {
 689     /* We wake up here when something happened, and below
 690      * then check if we're ready now. If we return FALSE,
 691      * we will be directly called again.
 692      */
 693     GST_OBJECT_UNLOCK (self);
 694     SRC_WAIT (self);
 695   } else {
 696     GstClockTime base_time, time;
 697     GstClock *clock;
 698     GstClockReturn status;
 699     GstClockTimeDiff jitter;
 700
 701     GST_DEBUG_OBJECT (self, "got subclass start time: %" GST_TIME_FORMAT,
 702         GST_TIME_ARGS (start));
 703
 704     base_time = GST_ELEMENT_CAST (self)->base_time;
 705     clock = gst_object_ref (GST_ELEMENT_CLOCK (self));
 706     GST_OBJECT_UNLOCK (self);
 707
 708     time = base_time + start;
 709     time += latency;
 710
 711     GST_DEBUG_OBJECT (self, "possibly waiting for clock to reach %"
 712         GST_TIME_FORMAT " (base %" GST_TIME_FORMAT " start %" GST_TIME_FORMAT
 713         " latency %" GST_TIME_FORMAT " current %" GST_TIME_FORMAT ")",
 714         GST_TIME_ARGS (time),
 715         GST_TIME_ARGS (base_time),
 716         GST_TIME_ARGS (start), GST_TIME_ARGS (latency),
 717         GST_TIME_ARGS (gst_clock_get_time (clock)));
 718
 719     self->priv->aggregate_id = gst_clock_new_single_shot_id (clock, time);
 720     gst_object_unref (clock);
 721     SRC_UNLOCK (self);
 722
 723     jitter = 0;
 724     status = gst_clock_id_wait (self->priv->aggregate_id, &jitter);
 725
 726     SRC_LOCK (self);
 727     if (self->priv->aggregate_id) {
 728       gst_clock_id_unref (self->priv->aggregate_id);
 729       self->priv->aggregate_id = NULL;
 730     }
 731
 732     GST_DEBUG_OBJECT (self,
 733         "clock returned %d (jitter: %" GST_STIME_FORMAT ")",
 734         status, GST_STIME_ARGS (jitter));
 735
 736     /* we timed out */
 737     if (status == GST_CLOCK_OK || status == GST_CLOCK_EARLY) {
 738       SRC_UNLOCK (self);
 739       *timeout = TRUE;
 740       return TRUE;
 741     }
 742   }
 743
 744   res = gst_aggregator_check_pads_ready (self);
 745   SRC_UNLOCK (self);
 746
 747   return res;
 748 }
 749
 750 static gboolean
 751 check_events (GstAggregator * self, GstAggregatorPad * pad, gpointer user_data)
 752 {
 753   GstEvent *event = NULL;
 754   GstQuery *query = NULL;
 755   GstAggregatorClass *klass = NULL;
 756   gboolean *processed_event = user_data;
 757
 758   do {
 759     event = NULL;
 760
 761     PAD_LOCK (pad);
 762     if (pad->priv->num_buffers == 0 && pad->priv->pending_eos) {
 763       pad->priv->pending_eos = FALSE;
 764       pad->priv->eos = TRUE;
 765     }
 766     if (pad->priv->clipped_buffer == NULL &&
 767         !GST_IS_BUFFER (g_queue_peek_tail (&pad->priv->buffers))) {
 768       if (GST_IS_EVENT (g_queue_peek_tail (&pad->priv->buffers)))
 769         event = gst_event_ref (g_queue_peek_tail (&pad->priv->buffers));
 770       if (GST_IS_QUERY (g_queue_peek_tail (&pad->priv->buffers)))
 771         query = g_queue_peek_tail (&pad->priv->buffers);
 772     }
 773     PAD_UNLOCK (pad);
 774     if (event || query) {
 775       gboolean ret;
 776
 777       if (processed_event)
 778         *processed_event = TRUE;
 779       if (klass == NULL)
 780         klass = GST_AGGREGATOR_GET_CLASS (self);
 781
 782       GST_LOG_OBJECT (pad, "Processing %" GST_PTR_FORMAT, event);
 783
 784       if (event) {
 785         gst_event_ref (event);
 786         ret = klass->sink_event (self, pad, event);
 787
 788         PAD_LOCK (pad);
 789         if (GST_EVENT_TYPE (event) == GST_EVENT_CAPS)
 790           pad->priv->negotiated = ret;
 791         if (g_queue_peek_tail (&pad->priv->buffers) == event)
 792           gst_event_unref (g_queue_pop_tail (&pad->priv->buffers));
 793         gst_event_unref (event);
 794       }
 795
 796       if (query) {
 797         GST_LOG_OBJECT (pad, "Processing %" GST_PTR_FORMAT, event);
 798         ret = klass->sink_query (self, pad, query);
 799
 800         PAD_LOCK (pad);
 801         if (g_queue_peek_tail (&pad->priv->buffers) == query) {
 802           GstStructure *s;
 803
 804           s = gst_query_writable_structure (query);
 805           gst_structure_set (s, "gst-aggregator-retval", G_TYPE_BOOLEAN, ret,
 806               NULL);
 807           g_queue_pop_tail (&pad->priv->buffers);
 808         }
 809       }
 810
 811       PAD_BROADCAST_EVENT (pad);
 812       PAD_UNLOCK (pad);
 813     }
 814     if (query) {
 815       if (processed_event)
 816         *processed_event = TRUE;
 817       if (klass == NULL)
 818         klass = GST_AGGREGATOR_GET_CLASS (self);
 819     }
 820   } while (event != NULL);
 821
 822   return TRUE;
 823 }
 824
 825 static void
 826 gst_aggregator_pad_set_flushing (GstAggregatorPad * aggpad,
 827     GstFlowReturn flow_return, gboolean full)
 828 {
 829   GList *item;
 830
 831   PAD_LOCK (aggpad);
 832   if (flow_return == GST_FLOW_NOT_LINKED)
 833     aggpad->priv->flow_return = MIN (flow_return, aggpad->priv->flow_return);
 834   else
 835     aggpad->priv->flow_return = flow_return;
 836
 837   item = g_queue_peek_head_link (&aggpad->priv->buffers);
 838   while (item) {
 839     GList *next = item->next;
 840
 841     /* In partial flush, we do like the pad, we get rid of non-sticky events
 842      * and EOS/SEGMENT.
 843      */
 844     if (full || GST_IS_BUFFER (item->data) ||
 845         GST_EVENT_TYPE (item->data) == GST_EVENT_EOS ||
 846         GST_EVENT_TYPE (item->data) == GST_EVENT_SEGMENT ||
 847         !GST_EVENT_IS_STICKY (item->data)) {
 848       if (!GST_IS_QUERY (item->data))
 849         gst_mini_object_unref (item->data);
 850       g_queue_delete_link (&aggpad->priv->buffers, item);
 851     }
 852     item = next;
 853   }
 854   aggpad->priv->num_buffers = 0;
 855   gst_buffer_replace (&aggpad->priv->clipped_buffer, NULL);
 856
 857   PAD_BROADCAST_EVENT (aggpad);
 858   PAD_UNLOCK (aggpad);
 859 }
 860
 861 static GstFlowReturn
 862 gst_aggregator_default_update_src_caps (GstAggregator * agg, GstCaps * caps,
 863     GstCaps ** ret)
 864 {
 865   *ret = gst_caps_ref (caps);
 866
 867   return GST_FLOW_OK;
 868 }
 869
 870 static GstCaps *
 871 gst_aggregator_default_fixate_src_caps (GstAggregator * agg, GstCaps * caps)
 872 {
 873   caps = gst_caps_fixate (caps);
 874
 875   return caps;
 876 }
 877
 878 static gboolean
 879 gst_aggregator_default_negotiated_src_caps (GstAggregator * agg, GstCaps * caps)
 880 {
 881   return TRUE;
 882 }
 883
 884
 885 /* takes ownership of the pool, allocator and query */
 886 static gboolean
 887 gst_aggregator_set_allocation (GstAggregator * self,
 888     GstBufferPool * pool, GstAllocator * allocator,
 889     GstAllocationParams * params, GstQuery * query)
 890 {
 891   GstAllocator *oldalloc;
 892   GstBufferPool *oldpool;
 893   GstQuery *oldquery;
 894
 895   GST_DEBUG ("storing allocation query");
 896
 897   GST_OBJECT_LOCK (self);
 898   oldpool = self->priv->pool;
 899   self->priv->pool = pool;
 900
 901   oldalloc = self->priv->allocator;
 902   self->priv->allocator = allocator;
 903
 904   oldquery = self->priv->allocation_query;
 905   self->priv->allocation_query = query;
 906
 907   if (params)
 908     self->priv->allocation_params = *params;
 909   else
 910     gst_allocation_params_init (&self->priv->allocation_params);
 911   GST_OBJECT_UNLOCK (self);
 912
 913   if (oldpool) {
 914     GST_DEBUG_OBJECT (self, "deactivating old pool %p", oldpool);
 915     gst_buffer_pool_set_active (oldpool, FALSE);
 916     gst_object_unref (oldpool);
 917   }
 918   if (oldalloc) {
 919     gst_object_unref (oldalloc);
 920   }
 921   if (oldquery) {
 922     gst_query_unref (oldquery);
 923   }
 924   return TRUE;
 925 }
 926
 927
 928 static gboolean
 929 gst_aggregator_decide_allocation (GstAggregator * self, GstQuery * query)
 930 {
 931   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (self);
 932
 933   if (aggclass->decide_allocation)
 934     if (!aggclass->decide_allocation (self, query))
 935       return FALSE;
 936
 937   return TRUE;
 938 }
 939
 940 static gboolean
 941 gst_aggregator_do_allocation (GstAggregator * self, GstCaps * caps)
 942 {
 943   GstQuery *query;
 944   gboolean result = TRUE;
 945   GstBufferPool *pool = NULL;
 946   GstAllocator *allocator;
 947   GstAllocationParams params;
 948
 949   /* find a pool for the negotiated caps now */
 950   GST_DEBUG_OBJECT (self, "doing allocation query");
 951   query = gst_query_new_allocation (caps, TRUE);
 952   if (!gst_pad_peer_query (self->srcpad, query)) {
 953     /* not a problem, just debug a little */
 954     GST_DEBUG_OBJECT (self, "peer ALLOCATION query failed");
 955   }
 956
 957   GST_DEBUG_OBJECT (self, "calling decide_allocation");
 958   result = gst_aggregator_decide_allocation (self, query);
 959
 960   GST_DEBUG_OBJECT (self, "ALLOCATION (%d) params: %" GST_PTR_FORMAT, result,
 961       query);
 962
 963   if (!result)
 964     goto no_decide_allocation;
 965
 966   /* we got configuration from our peer or the decide_allocation method,
 967    * parse them */
 968   if (gst_query_get_n_allocation_params (query) > 0) {
 969     gst_query_parse_nth_allocation_param (query, 0, &allocator, &params);
 970   } else {
 971     allocator = NULL;
 972     gst_allocation_params_init (&params);
 973   }
 974
 975   if (gst_query_get_n_allocation_pools (query) > 0)
 976     gst_query_parse_nth_allocation_pool (query, 0, &pool, NULL, NULL, NULL);
 977
 978   /* now store */
 979   result =
 980       gst_aggregator_set_allocation (self, pool, allocator, &params, query);
 981
 982   return result;
 983
 984   /* Errors */
 985 no_decide_allocation:
 986   {
 987     GST_WARNING_OBJECT (self, "Failed to decide allocation");
 988     gst_query_unref (query);
 989
 990     return result;
 991   }
 992
 993 }
 994
 995 /* WITH SRC_LOCK held */
 996 static GstFlowReturn
 997 gst_aggregator_update_src_caps (GstAggregator * self)
 998 {
 999   GstAggregatorClass *agg_klass = GST_AGGREGATOR_GET_CLASS (self);
1000   GstCaps *downstream_caps, *template_caps, *caps = NULL;
1001   GstFlowReturn ret = GST_FLOW_OK;
1002
1003   template_caps = gst_pad_get_pad_template_caps (self->srcpad);
1004   downstream_caps = gst_pad_peer_query_caps (self->srcpad, template_caps);
1005
1006   if (gst_caps_is_empty (downstream_caps)) {
1007     GST_INFO_OBJECT (self, "Downstream caps (%"
1008         GST_PTR_FORMAT ") not compatible with pad template caps (%"
1009         GST_PTR_FORMAT ")", downstream_caps, template_caps);
1010     ret = GST_FLOW_NOT_NEGOTIATED;
1011     goto done;
1012   }
1013
1014   g_assert (agg_klass->update_src_caps);
1015   GST_DEBUG_OBJECT (self, "updating caps from %" GST_PTR_FORMAT,
1016       downstream_caps);
1017   ret = agg_klass->update_src_caps (self, downstream_caps, &caps);
1018   if (ret < GST_FLOW_OK) {
1019     GST_WARNING_OBJECT (self, "Subclass failed to update provided caps");
1020     goto done;
1021   }
1022   if ((caps == NULL || gst_caps_is_empty (caps)) && ret >= GST_FLOW_OK) {
1023     ret = GST_FLOW_NOT_NEGOTIATED;
1024     goto done;
1025   }
1026   GST_DEBUG_OBJECT (self, "               to %" GST_PTR_FORMAT, caps);
1027
1028 #ifdef GST_ENABLE_EXTRA_CHECKS
1029   if (!gst_caps_is_subset (caps, template_caps)) {
1030     GstCaps *intersection;
1031
1032     GST_ERROR_OBJECT (self,
1033         "update_src_caps returned caps %" GST_PTR_FORMAT
1034         " which are not a real subset of the template caps %"
1035         GST_PTR_FORMAT, caps, template_caps);
1036     g_warning ("%s: update_src_caps returned caps which are not a real "
1037         "subset of the filter caps", GST_ELEMENT_NAME (self));
1038
1039     intersection =
1040         gst_caps_intersect_full (template_caps, caps, GST_CAPS_INTERSECT_FIRST);
1041     gst_caps_unref (caps);
1042     caps = intersection;
1043   }
1044 #endif
1045
1046   if (gst_caps_is_any (caps)) {
1047     goto done;
1048   }
1049
1050   if (!gst_caps_is_fixed (caps)) {
1051     g_assert (agg_klass->fixate_src_caps);
1052
1053     GST_DEBUG_OBJECT (self, "fixate caps from %" GST_PTR_FORMAT, caps);
1054     if (!(caps = agg_klass->fixate_src_caps (self, caps))) {
1055       GST_WARNING_OBJECT (self, "Subclass failed to fixate provided caps");
1056       ret = GST_FLOW_NOT_NEGOTIATED;
1057       goto done;
1058     }
1059     GST_DEBUG_OBJECT (self, "             to %" GST_PTR_FORMAT, caps);
1060   }
1061
1062   if (agg_klass->negotiated_src_caps) {
1063     if (!agg_klass->negotiated_src_caps (self, caps)) {
1064       GST_WARNING_OBJECT (self, "Subclass failed to accept negotiated caps");
1065       ret = GST_FLOW_NOT_NEGOTIATED;
1066       goto done;
1067     }
1068   }
1069
1070   gst_aggregator_set_src_caps (self, caps);
1071
1072   if (!gst_aggregator_do_allocation (self, caps)) {
1073     GST_WARNING_OBJECT (self, "Allocation negotiation failed");
1074     ret = GST_FLOW_NOT_NEGOTIATED;
1075   }
1076
1077 done:
1078   gst_caps_unref (downstream_caps);
1079   gst_caps_unref (template_caps);
1080
1081   if (caps)
1082     gst_caps_unref (caps);
1083
1084   return ret;
1085 }
1086
1087 static void
1088 gst_aggregator_aggregate_func (GstAggregator * self)
1089 {
1090   GstAggregatorPrivate *priv = self->priv;
1091   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
1092   gboolean timeout = FALSE;
1093
1094   if (self->priv->running == FALSE) {
1095     GST_DEBUG_OBJECT (self, "Not running anymore");
1096     return;
1097   }
1098
1099   GST_LOG_OBJECT (self, "Checking aggregate");
1100   while (priv->send_eos && priv->running) {
1101     GstFlowReturn flow_return = GST_FLOW_OK;
1102     gboolean processed_event = FALSE;
1103
1104     gst_aggregator_iterate_sinkpads (self, check_events, NULL);
1105
1106     if (!gst_aggregator_wait_and_check (self, &timeout))
1107       continue;
1108
1109     gst_aggregator_iterate_sinkpads (self, check_events, &processed_event);
1110     if (processed_event)
1111       continue;
1112
1113     if (gst_pad_check_reconfigure (GST_AGGREGATOR_SRC_PAD (self))) {
1114       flow_return = gst_aggregator_update_src_caps (self);
1115       if (flow_return != GST_FLOW_OK)
1116         gst_pad_mark_reconfigure (GST_AGGREGATOR_SRC_PAD (self));
1117     }
1118
1119     if (timeout || flow_return >= GST_FLOW_OK) {
1120       GST_TRACE_OBJECT (self, "Actually aggregating!");
1121       flow_return = klass->aggregate (self, timeout);
1122     }
1123
1124     if (flow_return == GST_AGGREGATOR_FLOW_NEED_DATA)
1125       continue;
1126
1127     GST_OBJECT_LOCK (self);
1128     if (flow_return == GST_FLOW_FLUSHING && priv->flush_seeking) {
1129       /* We don't want to set the pads to flushing, but we want to
1130        * stop the thread, so just break here */
1131       GST_OBJECT_UNLOCK (self);
1132       break;
1133     }
1134     GST_OBJECT_UNLOCK (self);
1135
1136     if (flow_return == GST_FLOW_EOS || flow_return == GST_FLOW_ERROR) {
1137       gst_aggregator_push_eos (self);
1138     }
1139
1140     GST_LOG_OBJECT (self, "flow return is %s", gst_flow_get_name (flow_return));
1141
1142     if (flow_return != GST_FLOW_OK) {
1143       GList *item;
1144
1145       GST_OBJECT_LOCK (self);
1146       for (item = GST_ELEMENT (self)->sinkpads; item; item = item->next) {
1147         GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (item->data);
1148
1149         gst_aggregator_pad_set_flushing (aggpad, flow_return, TRUE);
1150       }
1151       GST_OBJECT_UNLOCK (self);
1152       break;
1153     }
1154   }
1155
1156   /* Pause the task here, the only ways to get here are:
1157    * 1) We're stopping, in which case the task is stopped anyway
1158    * 2) We got a flow error above, in which case it might take
1159    *    some time to forward the flow return upstream and we
1160    *    would otherwise call the task function over and over
1161    *    again without doing anything
1162    */
1163   gst_pad_pause_task (self->srcpad);
1164 }
1165
1166 static gboolean
1167 gst_aggregator_start (GstAggregator * self)
1168 {
1169   GstAggregatorClass *klass;
1170   gboolean result;
1171
1172   self->priv->send_stream_start = TRUE;
1173   self->priv->send_segment = TRUE;
1174   self->priv->send_eos = TRUE;
1175   self->priv->srccaps = NULL;
1176
1177   gst_aggregator_set_allocation (self, NULL, NULL, NULL, NULL);
1178
1179   klass = GST_AGGREGATOR_GET_CLASS (self);
1180
1181   if (klass->start)
1182     result = klass->start (self);
1183   else
1184     result = TRUE;
1185
1186   return result;
1187 }
1188
1189 static gboolean
1190 _check_pending_flush_stop (GstAggregatorPad * pad)
1191 {
1192   gboolean res;
1193
1194   PAD_LOCK (pad);
1195   res = (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
1196   PAD_UNLOCK (pad);
1197
1198   return res;
1199 }
1200
1201 static gboolean
1202 gst_aggregator_stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
1203 {
1204   gboolean res = TRUE;
1205
1206   GST_INFO_OBJECT (self, "%s srcpad task",
1207       flush_start ? "Pausing" : "Stopping");
1208
1209   SRC_LOCK (self);
1210   self->priv->running = FALSE;
1211   SRC_BROADCAST (self);
1212   SRC_UNLOCK (self);
1213
1214   if (flush_start) {
1215     res = gst_pad_push_event (self->srcpad, flush_start);
1216   }
1217
1218   gst_pad_stop_task (self->srcpad);
1219
1220   return res;
1221 }
1222
1223 static void
1224 gst_aggregator_start_srcpad_task (GstAggregator * self)
1225 {
1226   GST_INFO_OBJECT (self, "Starting srcpad task");
1227
1228   self->priv->running = TRUE;
1229   gst_pad_start_task (GST_PAD (self->srcpad),
1230       (GstTaskFunction) gst_aggregator_aggregate_func, self, NULL);
1231 }
1232
1233 static GstFlowReturn
1234 gst_aggregator_flush (GstAggregator * self)
1235 {
1236   GstFlowReturn ret = GST_FLOW_OK;
1237   GstAggregatorPrivate *priv = self->priv;
1238   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
1239
1240   GST_DEBUG_OBJECT (self, "Flushing everything");
1241   GST_OBJECT_LOCK (self);
1242   priv->send_segment = TRUE;
1243   priv->flush_seeking = FALSE;
1244   priv->tags_changed = FALSE;
1245   GST_OBJECT_UNLOCK (self);
1246   if (klass->flush)
1247     ret = klass->flush (self);
1248
1249   return ret;
1250 }
1251
1252
1253 /* Called with GstAggregator's object lock held */
1254
1255 static gboolean
1256 gst_aggregator_all_flush_stop_received_locked (GstAggregator * self)
1257 {
1258   GList *tmp;
1259   GstAggregatorPad *tmppad;
1260
1261   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
1262     tmppad = (GstAggregatorPad *) tmp->data;
1263
1264     if (_check_pending_flush_stop (tmppad) == FALSE) {
1265       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
1266           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
1267       return FALSE;
1268     }
1269   }
1270
1271   return TRUE;
1272 }
1273
1274 static void
1275 gst_aggregator_flush_start (GstAggregator * self, GstAggregatorPad * aggpad,
1276     GstEvent * event)
1277 {
1278   GstAggregatorPrivate *priv = self->priv;
1279   GstAggregatorPadPrivate *padpriv = aggpad->priv;
1280
1281   gst_aggregator_pad_set_flushing (aggpad, GST_FLOW_FLUSHING, FALSE);
1282
1283   PAD_FLUSH_LOCK (aggpad);
1284   PAD_LOCK (aggpad);
1285   if (padpriv->pending_flush_start) {
1286     GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
1287
1288     padpriv->pending_flush_start = FALSE;
1289     padpriv->pending_flush_stop = TRUE;
1290   }
1291   PAD_UNLOCK (aggpad);
1292
1293   GST_OBJECT_LOCK (self);
1294   if (priv->flush_seeking) {
1295     /* If flush_seeking we forward the first FLUSH_START */
1296     if (priv->pending_flush_start) {
1297       priv->pending_flush_start = FALSE;
1298       GST_OBJECT_UNLOCK (self);
1299
1300       GST_INFO_OBJECT (self, "Flushing, pausing srcpad task");
1301       gst_aggregator_stop_srcpad_task (self, event);
1302
1303       GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
1304       GST_PAD_STREAM_LOCK (self->srcpad);
1305       GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
1306       event = NULL;
1307     } else {
1308       GST_OBJECT_UNLOCK (self);
1309       gst_event_unref (event);
1310     }
1311   } else {
1312     GST_OBJECT_UNLOCK (self);
1313     gst_event_unref (event);
1314   }
1315   PAD_FLUSH_UNLOCK (aggpad);
1316 }
1317
1318 /* Must be called with the the PAD_LOCK held */
1319 static void
1320 update_time_level (GstAggregatorPad * aggpad, gboolean head)
1321 {
1322   if (head) {
1323     if (GST_CLOCK_TIME_IS_VALID (aggpad->priv->head_position) &&
1324         aggpad->priv->head_segment.format == GST_FORMAT_TIME)
1325       aggpad->priv->head_time =
1326           gst_segment_to_running_time (&aggpad->priv->head_segment,
1327           GST_FORMAT_TIME, aggpad->priv->head_position);
1328     else
1329       aggpad->priv->head_time = GST_CLOCK_TIME_NONE;
1330
1331     if (!GST_CLOCK_TIME_IS_VALID (aggpad->priv->tail_time))
1332       aggpad->priv->tail_time = aggpad->priv->head_time;
1333   } else {
1334     if (GST_CLOCK_TIME_IS_VALID (aggpad->priv->tail_position) &&
1335         aggpad->segment.format == GST_FORMAT_TIME)
1336       aggpad->priv->tail_time =
1337           gst_segment_to_running_time (&aggpad->segment,
1338           GST_FORMAT_TIME, aggpad->priv->tail_position);
1339     else
1340       aggpad->priv->tail_time = aggpad->priv->head_time;
1341   }
1342
1343   if (aggpad->priv->head_time == GST_CLOCK_TIME_NONE ||
1344       aggpad->priv->tail_time == GST_CLOCK_TIME_NONE) {
1345     aggpad->priv->time_level = 0;
1346     return;
1347   }
1348
1349   if (aggpad->priv->tail_time > aggpad->priv->head_time)
1350     aggpad->priv->time_level = 0;
1351   else
1352     aggpad->priv->time_level = aggpad->priv->head_time -
1353         aggpad->priv->tail_time;
1354 }
1355
1356
1357 /* GstAggregator vmethods default implementations */
1358 static gboolean
1359 gst_aggregator_default_sink_event (GstAggregator * self,
1360     GstAggregatorPad * aggpad, GstEvent * event)
1361 {
1362   gboolean res = TRUE;
1363   GstPad *pad = GST_PAD (aggpad);
1364   GstAggregatorPrivate *priv = self->priv;
1365
1366   switch (GST_EVENT_TYPE (event)) {
1367     case GST_EVENT_FLUSH_START:
1368     {
1369       gst_aggregator_flush_start (self, aggpad, event);
1370       /* We forward only in one case: right after flush_seeking */
1371       event = NULL;
1372       goto eat;
1373     }
1374     case GST_EVENT_FLUSH_STOP:
1375     {
1376       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
1377
1378       gst_aggregator_pad_flush (aggpad, self);
1379       GST_OBJECT_LOCK (self);
1380       if (priv->flush_seeking) {
1381         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
1382         if (gst_aggregator_all_flush_stop_received_locked (self)) {
1383           GST_OBJECT_UNLOCK (self);
1384           /* That means we received FLUSH_STOP/FLUSH_STOP on
1385            * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
1386           gst_aggregator_flush (self);
1387           gst_pad_push_event (self->srcpad, event);
1388           event = NULL;
1389           SRC_LOCK (self);
1390           priv->send_eos = TRUE;
1391           SRC_BROADCAST (self);
1392           SRC_UNLOCK (self);
1393
1394           GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
1395           GST_PAD_STREAM_UNLOCK (self->srcpad);
1396           gst_aggregator_start_srcpad_task (self);
1397         } else {
1398           GST_OBJECT_UNLOCK (self);
1399         }
1400       } else {
1401         GST_OBJECT_UNLOCK (self);
1402       }
1403
1404       /* We never forward the event */
1405       goto eat;
1406     }
1407     case GST_EVENT_EOS:
1408     {
1409       GST_DEBUG_OBJECT (aggpad, "EOS");
1410
1411       /* We still have a buffer, and we don't want the subclass to have to
1412        * check for it. Mark pending_eos, eos will be set when steal_buffer is
1413        * called
1414        */
1415       SRC_LOCK (self);
1416       PAD_LOCK (aggpad);
1417       if (aggpad->priv->num_buffers == 0) {
1418         aggpad->priv->eos = TRUE;
1419       } else {
1420         aggpad->priv->pending_eos = TRUE;
1421       }
1422       PAD_UNLOCK (aggpad);
1423
1424       SRC_BROADCAST (self);
1425       SRC_UNLOCK (self);
1426       goto eat;
1427     }
1428     case GST_EVENT_SEGMENT:
1429     {
1430       PAD_LOCK (aggpad);
1431       GST_OBJECT_LOCK (aggpad);
1432       gst_event_copy_segment (event, &aggpad->segment);
1433       /* We've got a new segment, tail_position is now meaningless
1434        * and may interfere with the time_level calculation
1435        */
1436       aggpad->priv->tail_position = GST_CLOCK_TIME_NONE;
1437       update_time_level (aggpad, FALSE);
1438       GST_OBJECT_UNLOCK (aggpad);
1439       PAD_UNLOCK (aggpad);
1440
1441       GST_OBJECT_LOCK (self);
1442       self->priv->seqnum = gst_event_get_seqnum (event);
1443       GST_OBJECT_UNLOCK (self);
1444       goto eat;
1445     }
1446     case GST_EVENT_STREAM_START:
1447     {
1448       goto eat;
1449     }
1450     case GST_EVENT_GAP:
1451     {
1452       GstClockTime pts, endpts;
1453       GstClockTime duration;
1454       GstBuffer *gapbuf;
1455
1456       gst_event_parse_gap (event, &pts, &duration);
1457       gapbuf = gst_buffer_new ();
1458
1459       if (GST_CLOCK_TIME_IS_VALID (duration))
1460         endpts = pts + duration;
1461       else
1462         endpts = GST_CLOCK_TIME_NONE;
1463
1464       GST_OBJECT_LOCK (aggpad);
1465       res = gst_segment_clip (&aggpad->segment, GST_FORMAT_TIME, pts, endpts,
1466           &pts, &endpts);
1467       GST_OBJECT_UNLOCK (aggpad);
1468
1469       if (!res) {
1470         GST_WARNING_OBJECT (self, "GAP event outside segment, dropping");
1471         goto eat;
1472       }
1473
1474       if (GST_CLOCK_TIME_IS_VALID (endpts) && GST_CLOCK_TIME_IS_VALID (pts))
1475         duration = endpts - pts;
1476       else
1477         duration = GST_CLOCK_TIME_NONE;
1478
1479       GST_BUFFER_PTS (gapbuf) = pts;
1480       GST_BUFFER_DURATION (gapbuf) = duration;
1481       GST_BUFFER_FLAG_SET (gapbuf, GST_BUFFER_FLAG_GAP);
1482       GST_BUFFER_FLAG_SET (gapbuf, GST_BUFFER_FLAG_DROPPABLE);
1483
1484       if (gst_aggregator_pad_chain_internal (self, aggpad, gapbuf, FALSE) !=
1485           GST_FLOW_OK) {
1486         GST_WARNING_OBJECT (self, "Failed to chain gap buffer");
1487         res = FALSE;
1488       }
1489
1490       goto eat;
1491     }
1492     case GST_EVENT_TAG:
1493     {
1494       GstTagList *tags;
1495
1496       gst_event_parse_tag (event, &tags);
1497
1498       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
1499         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
1500         gst_event_unref (event);
1501         event = NULL;
1502         goto eat;
1503       }
1504       break;
1505     }
1506     default:
1507     {
1508       break;
1509     }
1510   }
1511
1512   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
1513   return gst_pad_event_default (pad, GST_OBJECT (self), event);
1514
1515 eat:
1516   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
1517   if (event)
1518     gst_event_unref (event);
1519
1520   return res;
1521 }
1522
1523 static inline gboolean
1524 gst_aggregator_stop_pad (GstAggregator * self, GstAggregatorPad * pad,
1525     gpointer unused_udata)
1526 {
1527   gst_aggregator_pad_flush (pad, self);
1528
1529   PAD_LOCK (pad);
1530   pad->priv->flow_return = GST_FLOW_FLUSHING;
1531   pad->priv->negotiated = FALSE;
1532   PAD_BROADCAST_EVENT (pad);
1533   PAD_UNLOCK (pad);
1534
1535   return TRUE;
1536 }
1537
1538 static gboolean
1539 gst_aggregator_stop (GstAggregator * agg)
1540 {
1541   GstAggregatorClass *klass;
1542   gboolean result;
1543
1544   gst_aggregator_reset_flow_values (agg);
1545
1546   gst_aggregator_iterate_sinkpads (agg, gst_aggregator_stop_pad, NULL);
1547
1548   klass = GST_AGGREGATOR_GET_CLASS (agg);
1549
1550   if (klass->stop)
1551     result = klass->stop (agg);
1552   else
1553     result = TRUE;
1554
1555   agg->priv->has_peer_latency = FALSE;
1556   agg->priv->peer_latency_live = FALSE;
1557   agg->priv->peer_latency_min = agg->priv->peer_latency_max = FALSE;
1558
1559   if (agg->priv->tags)
1560     gst_tag_list_unref (agg->priv->tags);
1561   agg->priv->tags = NULL;
1562
1563   gst_aggregator_set_allocation (agg, NULL, NULL, NULL, NULL);
1564
1565   return result;
1566 }
1567
1568 /* GstElement vmethods implementations */
1569 static GstStateChangeReturn
1570 gst_aggregator_change_state (GstElement * element, GstStateChange transition)
1571 {
1572   GstStateChangeReturn ret;
1573   GstAggregator *self = GST_AGGREGATOR (element);
1574
1575   switch (transition) {
1576     case GST_STATE_CHANGE_READY_TO_PAUSED:
1577       if (!gst_aggregator_start (self))
1578         goto error_start;
1579       break;
1580     default:
1581       break;
1582   }
1583
1584   if ((ret =
1585           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
1586               transition)) == GST_STATE_CHANGE_FAILURE)
1587     goto failure;
1588
1589
1590   switch (transition) {
1591     case GST_STATE_CHANGE_PAUSED_TO_READY:
1592       if (!gst_aggregator_stop (self)) {
1593         /* What to do in this case? Error out? */
1594         GST_ERROR_OBJECT (self, "Subclass failed to stop.");
1595       }
1596       break;
1597     default:
1598       break;
1599   }
1600
1601   return ret;
1602
1603 /* ERRORS */
1604 failure:
1605   {
1606     GST_ERROR_OBJECT (element, "parent failed state change");
1607     return ret;
1608   }
1609 error_start:
1610   {
1611     GST_ERROR_OBJECT (element, "Subclass failed to start");
1612     return GST_STATE_CHANGE_FAILURE;
1613   }
1614 }
1615
1616 static void
1617 gst_aggregator_release_pad (GstElement * element, GstPad * pad)
1618 {
1619   GstAggregator *self = GST_AGGREGATOR (element);
1620   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1621
1622   GST_INFO_OBJECT (pad, "Removing pad");
1623
1624   SRC_LOCK (self);
1625   gst_aggregator_pad_set_flushing (aggpad, GST_FLOW_FLUSHING, TRUE);
1626   gst_element_remove_pad (element, pad);
1627
1628   self->priv->has_peer_latency = FALSE;
1629   SRC_BROADCAST (self);
1630   SRC_UNLOCK (self);
1631 }
1632
1633 static GstAggregatorPad *
1634 gst_aggregator_default_create_new_pad (GstAggregator * self,
1635     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
1636 {
1637   GstAggregatorPad *agg_pad;
1638   GstAggregatorPrivate *priv = self->priv;
1639   gint serial = 0;
1640   gchar *name = NULL;
1641
1642   if (templ->direction != GST_PAD_SINK)
1643     goto not_sink;
1644
1645   if (templ->presence != GST_PAD_REQUEST)
1646     goto not_request;
1647
1648   GST_OBJECT_LOCK (self);
1649   if (req_name == NULL || strlen (req_name) < 6
1650       || !g_str_has_prefix (req_name, "sink_")) {
1651     /* no name given when requesting the pad, use next available int */
1652     serial = ++priv->max_padserial;
1653   } else {
1654     /* parse serial number from requested padname */
1655     serial = g_ascii_strtoull (&req_name[5], NULL, 10);
1656     if (serial > priv->max_padserial)
1657       priv->max_padserial = serial;
1658   }
1659
1660   name = g_strdup_printf ("sink_%u", serial);
1661   agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
1662       "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
1663   g_free (name);
1664
1665   GST_OBJECT_UNLOCK (self);
1666
1667   return agg_pad;
1668
1669   /* errors */
1670 not_sink:
1671   {
1672     GST_WARNING_OBJECT (self, "request new pad that is not a SINK pad");
1673     return NULL;
1674   }
1675 not_request:
1676   {
1677     GST_WARNING_OBJECT (self, "request new pad that is not a REQUEST pad");
1678     return NULL;
1679   }
1680 }
1681
1682 static GstPad *
1683 gst_aggregator_request_new_pad (GstElement * element,
1684     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
1685 {
1686   GstAggregator *self;
1687   GstAggregatorPad *agg_pad;
1688   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (element);
1689   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
1690
1691   self = GST_AGGREGATOR (element);
1692
1693   agg_pad = klass->create_new_pad (self, templ, req_name, caps);
1694   if (!agg_pad) {
1695     GST_ERROR_OBJECT (element, "Couldn't create new pad");
1696     return NULL;
1697   }
1698
1699   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
1700
1701   if (priv->running)
1702     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
1703
1704   /* add the pad to the element */
1705   gst_element_add_pad (element, GST_PAD (agg_pad));
1706
1707   return GST_PAD (agg_pad);
1708 }
1709
1710 /* Must be called with SRC_LOCK held */
1711
1712 static gboolean
1713 gst_aggregator_query_latency_unlocked (GstAggregator * self, GstQuery * query)
1714 {
1715   gboolean query_ret, live;
1716   GstClockTime our_latency, min, max;
1717
1718   query_ret = gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1719
1720   if (!query_ret) {
1721     GST_WARNING_OBJECT (self, "Latency query failed");
1722     return FALSE;
1723   }
1724
1725   gst_query_parse_latency (query, &live, &min, &max);
1726
1727   our_latency = self->priv->latency;
1728
1729   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (min))) {
1730     GST_ERROR_OBJECT (self, "Invalid minimum latency %" GST_TIME_FORMAT
1731         ". Please file a bug at " PACKAGE_BUGREPORT ".", GST_TIME_ARGS (min));
1732     return FALSE;
1733   }
1734
1735   if (min > max && GST_CLOCK_TIME_IS_VALID (max)) {
1736     GST_ELEMENT_WARNING (self, CORE, CLOCK, (NULL),
1737         ("Impossible to configure latency: max %" GST_TIME_FORMAT " < min %"
1738             GST_TIME_FORMAT ". Add queues or other buffering elements.",
1739             GST_TIME_ARGS (max), GST_TIME_ARGS (min)));
1740     return FALSE;
1741   }
1742
1743   self->priv->peer_latency_live = live;
1744   self->priv->peer_latency_min = min;
1745   self->priv->peer_latency_max = max;
1746   self->priv->has_peer_latency = TRUE;
1747
1748   /* add our own */
1749   min += our_latency;
1750   min += self->priv->sub_latency_min;
1751   if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1752       && GST_CLOCK_TIME_IS_VALID (max))
1753     max += self->priv->sub_latency_max + our_latency;
1754   else
1755     max = GST_CLOCK_TIME_NONE;
1756
1757   SRC_BROADCAST (self);
1758
1759   GST_DEBUG_OBJECT (self, "configured latency live:%s min:%" G_GINT64_FORMAT
1760       " max:%" G_GINT64_FORMAT, live ? "true" : "false", min, max);
1761
1762   gst_query_set_latency (query, live, min, max);
1763
1764   return query_ret;
1765 }
1766
1767 /*
1768  * MUST be called with the src_lock held.
1769  *
1770  * See  gst_aggregator_get_latency() for doc
1771  */
1772 static GstClockTime
1773 gst_aggregator_get_latency_unlocked (GstAggregator * self)
1774 {
1775   GstClockTime latency;
1776
1777   g_return_val_if_fail (GST_IS_AGGREGATOR (self), 0);
1778
1779   if (!self->priv->has_peer_latency) {
1780     GstQuery *query = gst_query_new_latency ();
1781     gboolean ret;
1782
1783     ret = gst_aggregator_query_latency_unlocked (self, query);
1784     gst_query_unref (query);
1785     if (!ret)
1786       return GST_CLOCK_TIME_NONE;
1787   }
1788
1789   if (!self->priv->has_peer_latency || !self->priv->peer_latency_live)
1790     return GST_CLOCK_TIME_NONE;
1791
1792   /* latency_min is never GST_CLOCK_TIME_NONE by construction */
1793   latency = self->priv->peer_latency_min;
1794
1795   /* add our own */
1796   latency += self->priv->latency;
1797   latency += self->priv->sub_latency_min;
1798
1799   return latency;
1800 }
1801
1802 /**
1803  * gst_aggregator_get_latency:
1804  * @self: a #GstAggregator
1805  *
1806  * Retrieves the latency values reported by @self in response to the latency
1807  * query, or %GST_CLOCK_TIME_NONE if there is not live source connected and the element
1808  * will not wait for the clock.
1809  *
1810  * Typically only called by subclasses.
1811  *
1812  * Returns: The latency or %GST_CLOCK_TIME_NONE if the element does not sync
1813  */
1814 GstClockTime
1815 gst_aggregator_get_latency (GstAggregator * self)
1816 {
1817   GstClockTime ret;
1818
1819   SRC_LOCK (self);
1820   ret = gst_aggregator_get_latency_unlocked (self);
1821   SRC_UNLOCK (self);
1822
1823   return ret;
1824 }
1825
1826 static gboolean
1827 gst_aggregator_send_event (GstElement * element, GstEvent * event)
1828 {
1829   GstAggregator *self = GST_AGGREGATOR (element);
1830
1831   GST_STATE_LOCK (element);
1832   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK &&
1833       GST_STATE (element) < GST_STATE_PAUSED) {
1834     gdouble rate;
1835     GstFormat fmt;
1836     GstSeekFlags flags;
1837     GstSeekType start_type, stop_type;
1838     gint64 start, stop;
1839
1840     gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1841         &start, &stop_type, &stop);
1842
1843     GST_OBJECT_LOCK (self);
1844     gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1845         stop_type, stop, NULL);
1846     self->priv->seqnum = gst_event_get_seqnum (event);
1847     self->priv->first_buffer = FALSE;
1848     GST_OBJECT_UNLOCK (self);
1849
1850     GST_DEBUG_OBJECT (element, "Storing segment %" GST_PTR_FORMAT, event);
1851   }
1852   GST_STATE_UNLOCK (element);
1853
1854
1855   return GST_ELEMENT_CLASS (aggregator_parent_class)->send_event (element,
1856       event);
1857 }
1858
1859 static gboolean
1860 gst_aggregator_default_src_query (GstAggregator * self, GstQuery * query)
1861 {
1862   gboolean res = TRUE;
1863
1864   switch (GST_QUERY_TYPE (query)) {
1865     case GST_QUERY_SEEKING:
1866     {
1867       GstFormat format;
1868
1869       /* don't pass it along as some (file)sink might claim it does
1870        * whereas with a collectpads in between that will not likely work */
1871       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
1872       gst_query_set_seeking (query, format, FALSE, 0, -1);
1873       res = TRUE;
1874
1875       break;
1876     }
1877     case GST_QUERY_LATENCY:
1878       SRC_LOCK (self);
1879       res = gst_aggregator_query_latency_unlocked (self, query);
1880       SRC_UNLOCK (self);
1881       break;
1882     default:
1883       return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1884   }
1885
1886   return res;
1887 }
1888
1889 static gboolean
1890 gst_aggregator_event_forward_func (GstPad * pad, gpointer user_data)
1891 {
1892   EventData *evdata = user_data;
1893   gboolean ret = TRUE;
1894   GstPad *peer = gst_pad_get_peer (pad);
1895   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1896
1897   if (peer) {
1898     if (evdata->only_to_active_pads && aggpad->priv->first_buffer) {
1899       GST_DEBUG_OBJECT (pad, "not sending event to inactive pad");
1900       ret = TRUE;
1901     } else {
1902       ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
1903       GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
1904       gst_object_unref (peer);
1905     }
1906   }
1907
1908   if (ret == FALSE) {
1909     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
1910       GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
1911
1912       GST_DEBUG_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
1913
1914       if (gst_pad_query (peer, seeking)) {
1915         gboolean seekable;
1916
1917         gst_query_parse_seeking (seeking, NULL, &seekable, NULL, NULL);
1918
1919         if (seekable == FALSE) {
1920           GST_INFO_OBJECT (pad,
1921               "Source not seekable, We failed but it does not matter!");
1922
1923           ret = TRUE;
1924         }
1925       } else {
1926         GST_ERROR_OBJECT (pad, "Query seeking FAILED");
1927       }
1928
1929       gst_query_unref (seeking);
1930     }
1931
1932     if (evdata->flush) {
1933       PAD_LOCK (aggpad);
1934       aggpad->priv->pending_flush_start = FALSE;
1935       aggpad->priv->pending_flush_stop = FALSE;
1936       PAD_UNLOCK (aggpad);
1937     }
1938   } else {
1939     evdata->one_actually_seeked = TRUE;
1940   }
1941
1942   evdata->result &= ret;
1943
1944   /* Always send to all pads */
1945   return FALSE;
1946 }
1947
1948 static EventData
1949 gst_aggregator_forward_event_to_all_sinkpads (GstAggregator * self,
1950     GstEvent * event, gboolean flush, gboolean only_to_active_pads)
1951 {
1952   EventData evdata;
1953
1954   evdata.event = event;
1955   evdata.result = TRUE;
1956   evdata.flush = flush;
1957   evdata.one_actually_seeked = FALSE;
1958   evdata.only_to_active_pads = only_to_active_pads;
1959
1960   /* We first need to set all pads as flushing in a first pass
1961    * as flush_start flush_stop is sometimes sent synchronously
1962    * while we send the seek event */
1963   if (flush) {
1964     GList *l;
1965
1966     GST_OBJECT_LOCK (self);
1967     for (l = GST_ELEMENT_CAST (self)->sinkpads; l != NULL; l = l->next) {
1968       GstAggregatorPad *pad = l->data;
1969
1970       PAD_LOCK (pad);
1971       pad->priv->pending_flush_start = TRUE;
1972       pad->priv->pending_flush_stop = FALSE;
1973       PAD_UNLOCK (pad);
1974     }
1975     GST_OBJECT_UNLOCK (self);
1976   }
1977
1978   gst_pad_forward (self->srcpad, gst_aggregator_event_forward_func, &evdata);
1979
1980   gst_event_unref (event);
1981
1982   return evdata;
1983 }
1984
1985 static gboolean
1986 gst_aggregator_do_seek (GstAggregator * self, GstEvent * event)
1987 {
1988   gdouble rate;
1989   GstFormat fmt;
1990   GstSeekFlags flags;
1991   GstSeekType start_type, stop_type;
1992   gint64 start, stop;
1993   gboolean flush;
1994   EventData evdata;
1995   GstAggregatorPrivate *priv = self->priv;
1996
1997   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1998       &start, &stop_type, &stop);
1999
2000   GST_INFO_OBJECT (self, "starting SEEK");
2001
2002   flush = flags & GST_SEEK_FLAG_FLUSH;
2003
2004   GST_OBJECT_LOCK (self);
2005   if (flush) {
2006     priv->pending_flush_start = TRUE;
2007     priv->flush_seeking = TRUE;
2008   }
2009
2010   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
2011       stop_type, stop, NULL);
2012
2013   /* Seeking sets a position */
2014   self->priv->first_buffer = FALSE;
2015   GST_OBJECT_UNLOCK (self);
2016
2017   /* forward the seek upstream */
2018   evdata =
2019       gst_aggregator_forward_event_to_all_sinkpads (self, event, flush, FALSE);
2020   event = NULL;
2021
2022   if (!evdata.result || !evdata.one_actually_seeked) {
2023     GST_OBJECT_LOCK (self);
2024     priv->flush_seeking = FALSE;
2025     priv->pending_flush_start = FALSE;
2026     GST_OBJECT_UNLOCK (self);
2027   }
2028
2029   GST_INFO_OBJECT (self, "seek done, result: %d", evdata.result);
2030
2031   return evdata.result;
2032 }
2033
2034 static gboolean
2035 gst_aggregator_default_src_event (GstAggregator * self, GstEvent * event)
2036 {
2037   EventData evdata;
2038   gboolean res = TRUE;
2039
2040   switch (GST_EVENT_TYPE (event)) {
2041     case GST_EVENT_SEEK:
2042     {
2043       gst_event_ref (event);
2044       res = gst_aggregator_do_seek (self, event);
2045       gst_event_unref (event);
2046       event = NULL;
2047       goto done;
2048     }
2049     case GST_EVENT_NAVIGATION:
2050     {
2051       /* navigation is rather pointless. */
2052       res = FALSE;
2053       gst_event_unref (event);
2054       goto done;
2055     }
2056     default:
2057     {
2058       break;
2059     }
2060   }
2061
2062   /* Don't forward QOS events to pads that had no active buffer yet. Otherwise
2063    * they will receive a QOS event that has earliest_time=0 (because we can't
2064    * have negative timestamps), and consider their buffer as too late */
2065   evdata =
2066       gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE,
2067       GST_EVENT_TYPE (event) == GST_EVENT_QOS);
2068   res = evdata.result;
2069
2070 done:
2071   return res;
2072 }
2073
2074 static gboolean
2075 gst_aggregator_src_pad_event_func (GstPad * pad, GstObject * parent,
2076     GstEvent * event)
2077 {
2078   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
2079
2080   return klass->src_event (GST_AGGREGATOR (parent), event);
2081 }
2082
2083 static gboolean
2084 gst_aggregator_src_pad_query_func (GstPad * pad, GstObject * parent,
2085     GstQuery * query)
2086 {
2087   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
2088
2089   return klass->src_query (GST_AGGREGATOR (parent), query);
2090 }
2091
2092 static gboolean
2093 gst_aggregator_src_pad_activate_mode_func (GstPad * pad,
2094     GstObject * parent, GstPadMode mode, gboolean active)
2095 {
2096   GstAggregator *self = GST_AGGREGATOR (parent);
2097   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
2098
2099   if (klass->src_activate) {
2100     if (klass->src_activate (self, mode, active) == FALSE) {
2101       return FALSE;
2102     }
2103   }
2104
2105   if (active == TRUE) {
2106     switch (mode) {
2107       case GST_PAD_MODE_PUSH:
2108       {
2109         GST_INFO_OBJECT (pad, "Activating pad!");
2110         gst_aggregator_start_srcpad_task (self);
2111         return TRUE;
2112       }
2113       default:
2114       {
2115         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
2116         return FALSE;
2117       }
2118     }
2119   }
2120
2121   /* deactivating */
2122   GST_INFO_OBJECT (self, "Deactivating srcpad");
2123   gst_aggregator_stop_srcpad_task (self, FALSE);
2124
2125   return TRUE;
2126 }
2127
2128 static gboolean
2129 gst_aggregator_default_sink_query (GstAggregator * self,
2130     GstAggregatorPad * aggpad, GstQuery * query)
2131 {
2132   GstPad *pad = GST_PAD (aggpad);
2133
2134   if (GST_QUERY_TYPE (query) == GST_QUERY_ALLOCATION) {
2135     GstQuery *decide_query = NULL;
2136     GstAggregatorClass *agg_class;
2137     gboolean ret;
2138
2139     GST_OBJECT_LOCK (self);
2140     PAD_LOCK (aggpad);
2141     if (G_UNLIKELY (!aggpad->priv->negotiated)) {
2142       GST_DEBUG_OBJECT (self,
2143           "not negotiated yet, can't answer ALLOCATION query");
2144       PAD_UNLOCK (aggpad);
2145       GST_OBJECT_UNLOCK (self);
2146
2147       return FALSE;
2148     }
2149
2150     if ((decide_query = self->priv->allocation_query))
2151       gst_query_ref (decide_query);
2152     PAD_UNLOCK (aggpad);
2153     GST_OBJECT_UNLOCK (self);
2154
2155     GST_DEBUG_OBJECT (self,
2156         "calling propose allocation with query %" GST_PTR_FORMAT, decide_query);
2157
2158     agg_class = GST_AGGREGATOR_GET_CLASS (self);
2159
2160     /* pass the query to the propose_allocation vmethod if any */
2161     if (agg_class->propose_allocation)
2162       ret = agg_class->propose_allocation (self, aggpad, decide_query, query);
2163     else
2164       ret = FALSE;
2165
2166     if (decide_query)
2167       gst_query_unref (decide_query);
2168
2169     GST_DEBUG_OBJECT (self, "ALLOCATION ret %d, %" GST_PTR_FORMAT, ret, query);
2170     return ret;
2171   }
2172
2173   return gst_pad_query_default (pad, GST_OBJECT (self), query);
2174 }
2175
2176 static void
2177 gst_aggregator_finalize (GObject * object)
2178 {
2179   GstAggregator *self = (GstAggregator *) object;
2180
2181   g_mutex_clear (&self->priv->src_lock);
2182   g_cond_clear (&self->priv->src_cond);
2183
2184   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
2185 }
2186
2187 /*
2188  * gst_aggregator_set_latency_property:
2189  * @agg: a #GstAggregator
2190  * @latency: the new latency value (in nanoseconds).
2191  *
2192  * Sets the new latency value to @latency. This value is used to limit the
2193  * amount of time a pad waits for data to appear before considering the pad
2194  * as unresponsive.
2195  */
2196 static void
2197 gst_aggregator_set_latency_property (GstAggregator * self, gint64 latency)
2198 {
2199   gboolean changed;
2200
2201   g_return_if_fail (GST_IS_AGGREGATOR (self));
2202   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (latency));
2203
2204   SRC_LOCK (self);
2205   changed = (self->priv->latency != latency);
2206
2207   if (changed) {
2208     GList *item;
2209
2210     GST_OBJECT_LOCK (self);
2211     /* First lock all the pads */
2212     for (item = GST_ELEMENT_CAST (self)->sinkpads; item; item = item->next) {
2213       GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (item->data);
2214       PAD_LOCK (aggpad);
2215     }
2216
2217     self->priv->latency = latency;
2218
2219     SRC_BROADCAST (self);
2220
2221     /* Now wake up the pads */
2222     for (item = GST_ELEMENT_CAST (self)->sinkpads; item; item = item->next) {
2223       GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (item->data);
2224       PAD_BROADCAST_EVENT (aggpad);
2225       PAD_UNLOCK (aggpad);
2226     }
2227     GST_OBJECT_UNLOCK (self);
2228   }
2229
2230   SRC_UNLOCK (self);
2231
2232   if (changed)
2233     gst_element_post_message (GST_ELEMENT_CAST (self),
2234         gst_message_new_latency (GST_OBJECT_CAST (self)));
2235 }
2236
2237 /*
2238  * gst_aggregator_get_latency_property:
2239  * @agg: a #GstAggregator
2240  *
2241  * Gets the latency value. See gst_aggregator_set_latency for
2242  * more details.
2243  *
2244  * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
2245  * before a pad is deemed unresponsive. A value of -1 means an
2246  * unlimited time.
2247  */
2248 static gint64
2249 gst_aggregator_get_latency_property (GstAggregator * agg)
2250 {
2251   gint64 res;
2252
2253   g_return_val_if_fail (GST_IS_AGGREGATOR (agg), -1);
2254
2255   GST_OBJECT_LOCK (agg);
2256   res = agg->priv->latency;
2257   GST_OBJECT_UNLOCK (agg);
2258
2259   return res;
2260 }
2261
2262 static void
2263 gst_aggregator_set_property (GObject * object, guint prop_id,
2264     const GValue * value, GParamSpec * pspec)
2265 {
2266   GstAggregator *agg = GST_AGGREGATOR (object);
2267
2268   switch (prop_id) {
2269     case PROP_LATENCY:
2270       gst_aggregator_set_latency_property (agg, g_value_get_int64 (value));
2271       break;
2272     case PROP_START_TIME_SELECTION:
2273       agg->priv->start_time_selection = g_value_get_enum (value);
2274       break;
2275     case PROP_START_TIME:
2276       agg->priv->start_time = g_value_get_uint64 (value);
2277       break;
2278     default:
2279       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2280       break;
2281   }
2282 }
2283
2284 static void
2285 gst_aggregator_get_property (GObject * object, guint prop_id,
2286     GValue * value, GParamSpec * pspec)
2287 {
2288   GstAggregator *agg = GST_AGGREGATOR (object);
2289
2290   switch (prop_id) {
2291     case PROP_LATENCY:
2292       g_value_set_int64 (value, gst_aggregator_get_latency_property (agg));
2293       break;
2294     case PROP_START_TIME_SELECTION:
2295       g_value_set_enum (value, agg->priv->start_time_selection);
2296       break;
2297     case PROP_START_TIME:
2298       g_value_set_uint64 (value, agg->priv->start_time);
2299       break;
2300     default:
2301       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2302       break;
2303   }
2304 }
2305
2306 /* GObject vmethods implementations */
2307 static void
2308 gst_aggregator_class_init (GstAggregatorClass * klass)
2309 {
2310   GObjectClass *gobject_class = (GObjectClass *) klass;
2311   GstElementClass *gstelement_class = (GstElementClass *) klass;
2312
2313   aggregator_parent_class = g_type_class_peek_parent (klass);
2314   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
2315
2316   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
2317       GST_DEBUG_FG_MAGENTA, "GstAggregator");
2318
2319   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
2320
2321   klass->sink_event = gst_aggregator_default_sink_event;
2322   klass->sink_query = gst_aggregator_default_sink_query;
2323
2324   klass->src_event = gst_aggregator_default_src_event;
2325   klass->src_query = gst_aggregator_default_src_query;
2326
2327   klass->create_new_pad = gst_aggregator_default_create_new_pad;
2328   klass->update_src_caps = gst_aggregator_default_update_src_caps;
2329   klass->fixate_src_caps = gst_aggregator_default_fixate_src_caps;
2330   klass->negotiated_src_caps = gst_aggregator_default_negotiated_src_caps;
2331
2332   gstelement_class->request_new_pad =
2333       GST_DEBUG_FUNCPTR (gst_aggregator_request_new_pad);
2334   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_aggregator_send_event);
2335   gstelement_class->release_pad =
2336       GST_DEBUG_FUNCPTR (gst_aggregator_release_pad);
2337   gstelement_class->change_state =
2338       GST_DEBUG_FUNCPTR (gst_aggregator_change_state);
2339
2340   gobject_class->set_property = gst_aggregator_set_property;
2341   gobject_class->get_property = gst_aggregator_get_property;
2342   gobject_class->finalize = gst_aggregator_finalize;
2343
2344   g_object_class_install_property (gobject_class, PROP_LATENCY,
2345       g_param_spec_int64 ("latency", "Buffer latency",
2346           "Additional latency in live mode to allow upstream "
2347           "to take longer to produce buffers for the current "
2348           "position (in nanoseconds)", 0,
2349           (G_MAXLONG == G_MAXINT64) ? G_MAXINT64 : (G_MAXLONG * GST_SECOND - 1),
2350           DEFAULT_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2351
2352   g_object_class_install_property (gobject_class, PROP_START_TIME_SELECTION,
2353       g_param_spec_enum ("start-time-selection", "Start Time Selection",
2354           "Decides which start time is output",
2355           gst_aggregator_start_time_selection_get_type (),
2356           DEFAULT_START_TIME_SELECTION,
2357           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2358
2359   g_object_class_install_property (gobject_class, PROP_START_TIME,
2360       g_param_spec_uint64 ("start-time", "Start Time",
2361           "Start time to use if start-time-selection=set", 0,
2362           G_MAXUINT64,
2363           DEFAULT_START_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2364
2365   GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_stop_pad);
2366 }
2367
2368 static void
2369 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
2370 {
2371   GstPadTemplate *pad_template;
2372   GstAggregatorPrivate *priv;
2373
2374   g_return_if_fail (klass->aggregate != NULL);
2375
2376   self->priv =
2377       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
2378       GstAggregatorPrivate);
2379
2380   priv = self->priv;
2381
2382   pad_template =
2383       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
2384   g_return_if_fail (pad_template != NULL);
2385
2386   priv->max_padserial = -1;
2387   priv->tags_changed = FALSE;
2388
2389   self->priv->peer_latency_live = FALSE;
2390   self->priv->peer_latency_min = self->priv->sub_latency_min = 0;
2391   self->priv->peer_latency_max = self->priv->sub_latency_max = 0;
2392   self->priv->has_peer_latency = FALSE;
2393   gst_aggregator_reset_flow_values (self);
2394
2395   self->srcpad = gst_pad_new_from_template (pad_template, "src");
2396
2397   gst_pad_set_event_function (self->srcpad,
2398       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_event_func));
2399   gst_pad_set_query_function (self->srcpad,
2400       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_query_func));
2401   gst_pad_set_activatemode_function (self->srcpad,
2402       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_activate_mode_func));
2403
2404   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
2405
2406   self->priv->latency = DEFAULT_LATENCY;
2407   self->priv->start_time_selection = DEFAULT_START_TIME_SELECTION;
2408   self->priv->start_time = DEFAULT_START_TIME;
2409
2410   g_mutex_init (&self->priv->src_lock);
2411   g_cond_init (&self->priv->src_cond);
2412 }
2413
2414 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
2415  * method to get to the padtemplates */
2416 GType
2417 gst_aggregator_get_type (void)
2418 {
2419   static volatile gsize type = 0;
2420
2421   if (g_once_init_enter (&type)) {
2422     GType _type;
2423     static const GTypeInfo info = {
2424       sizeof (GstAggregatorClass),
2425       NULL,
2426       NULL,
2427       (GClassInitFunc) gst_aggregator_class_init,
2428       NULL,
2429       NULL,
2430       sizeof (GstAggregator),
2431       0,
2432       (GInstanceInitFunc) gst_aggregator_init,
2433     };
2434
2435     _type = g_type_register_static (GST_TYPE_ELEMENT,
2436         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
2437     g_once_init_leave (&type, _type);
2438   }
2439   return type;
2440 }
2441
2442 /* Must be called with SRC lock and PAD lock held */
2443 static gboolean
2444 gst_aggregator_pad_has_space (GstAggregator * self, GstAggregatorPad * aggpad)
2445 {
2446   /* Empty queue always has space */
2447   if (aggpad->priv->num_buffers == 0 && aggpad->priv->clipped_buffer == NULL)
2448     return TRUE;
2449
2450   /* We also want at least two buffers, one is being processed and one is ready
2451    * for the next iteration when we operate in live mode. */
2452   if (self->priv->peer_latency_live && aggpad->priv->num_buffers < 2)
2453     return TRUE;
2454
2455   /* zero latency, if there is a buffer, it's full */
2456   if (self->priv->latency == 0)
2457     return FALSE;
2458
2459   /* Allow no more buffers than the latency */
2460   return (aggpad->priv->time_level <= self->priv->latency);
2461 }
2462
2463 /* Must be called with the PAD_LOCK held */
2464 static void
2465 apply_buffer (GstAggregatorPad * aggpad, GstBuffer * buffer, gboolean head)
2466 {
2467   GstClockTime timestamp;
2468
2469   if (GST_BUFFER_DTS_IS_VALID (buffer))
2470     timestamp = GST_BUFFER_DTS (buffer);
2471   else
2472     timestamp = GST_BUFFER_PTS (buffer);
2473
2474   if (timestamp == GST_CLOCK_TIME_NONE) {
2475     if (head)
2476       timestamp = aggpad->priv->head_position;
2477     else
2478       timestamp = aggpad->priv->tail_position;
2479   }
2480
2481   /* add duration */
2482   if (GST_BUFFER_DURATION_IS_VALID (buffer))
2483     timestamp += GST_BUFFER_DURATION (buffer);
2484
2485   if (head)
2486     aggpad->priv->head_position = timestamp;
2487   else
2488     aggpad->priv->tail_position = timestamp;
2489
2490   update_time_level (aggpad, head);
2491 }
2492
2493 static GstFlowReturn
2494 gst_aggregator_pad_chain_internal (GstAggregator * self,
2495     GstAggregatorPad * aggpad, GstBuffer * buffer, gboolean head)
2496 {
2497   GstFlowReturn flow_return;
2498   GstClockTime buf_pts;
2499
2500   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
2501
2502   PAD_FLUSH_LOCK (aggpad);
2503
2504   PAD_LOCK (aggpad);
2505   flow_return = aggpad->priv->flow_return;
2506   if (flow_return != GST_FLOW_OK)
2507     goto flushing;
2508
2509   if (aggpad->priv->pending_eos == TRUE)
2510     goto eos;
2511
2512   PAD_UNLOCK (aggpad);
2513
2514   buf_pts = GST_BUFFER_PTS (buffer);
2515
2516   for (;;) {
2517     SRC_LOCK (self);
2518     GST_OBJECT_LOCK (self);
2519     PAD_LOCK (aggpad);
2520
2521     if (aggpad->priv->first_buffer) {
2522       self->priv->has_peer_latency = FALSE;
2523       aggpad->priv->first_buffer = FALSE;
2524     }
2525
2526     if (gst_aggregator_pad_has_space (self, aggpad)
2527         && aggpad->priv->flow_return == GST_FLOW_OK) {
2528       if (head)
2529         g_queue_push_head (&aggpad->priv->buffers, buffer);
2530       else
2531         g_queue_push_tail (&aggpad->priv->buffers, buffer);
2532       apply_buffer (aggpad, buffer, head);
2533       aggpad->priv->num_buffers++;
2534       buffer = NULL;
2535       SRC_BROADCAST (self);
2536       break;
2537     }
2538
2539     flow_return = aggpad->priv->flow_return;
2540     if (flow_return != GST_FLOW_OK) {
2541       GST_OBJECT_UNLOCK (self);
2542       SRC_UNLOCK (self);
2543       goto flushing;
2544     }
2545     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
2546     GST_OBJECT_UNLOCK (self);
2547     SRC_UNLOCK (self);
2548     PAD_WAIT_EVENT (aggpad);
2549
2550     PAD_UNLOCK (aggpad);
2551   }
2552
2553   if (self->priv->first_buffer) {
2554     GstClockTime start_time;
2555
2556     switch (self->priv->start_time_selection) {
2557       case GST_AGGREGATOR_START_TIME_SELECTION_ZERO:
2558       default:
2559         start_time = 0;
2560         break;
2561       case GST_AGGREGATOR_START_TIME_SELECTION_FIRST:
2562         GST_OBJECT_LOCK (aggpad);
2563         if (aggpad->priv->head_segment.format == GST_FORMAT_TIME) {
2564           start_time = buf_pts;
2565           if (start_time != -1) {
2566             start_time = MAX (start_time, aggpad->priv->head_segment.start);
2567             start_time =
2568                 gst_segment_to_running_time (&aggpad->priv->head_segment,
2569                 GST_FORMAT_TIME, start_time);
2570           }
2571         } else {
2572           start_time = 0;
2573           GST_WARNING_OBJECT (aggpad,
2574               "Ignoring request of selecting the first start time "
2575               "as the segment is a %s segment instead of a time segment",
2576               gst_format_get_name (aggpad->segment.format));
2577         }
2578         GST_OBJECT_UNLOCK (aggpad);
2579         break;
2580       case GST_AGGREGATOR_START_TIME_SELECTION_SET:
2581         start_time = self->priv->start_time;
2582         if (start_time == -1)
2583           start_time = 0;
2584         break;
2585     }
2586
2587     if (start_time != -1) {
2588       if (self->segment.position == -1)
2589         self->segment.position = start_time;
2590       else
2591         self->segment.position = MIN (start_time, self->segment.position);
2592
2593       GST_DEBUG_OBJECT (self, "Selecting start time %" GST_TIME_FORMAT,
2594           GST_TIME_ARGS (start_time));
2595     }
2596   }
2597
2598   PAD_UNLOCK (aggpad);
2599   GST_OBJECT_UNLOCK (self);
2600   SRC_UNLOCK (self);
2601
2602   PAD_FLUSH_UNLOCK (aggpad);
2603
2604   GST_DEBUG_OBJECT (aggpad, "Done chaining");
2605
2606   return flow_return;
2607
2608 flushing:
2609   PAD_UNLOCK (aggpad);
2610   PAD_FLUSH_UNLOCK (aggpad);
2611
2612   GST_DEBUG_OBJECT (aggpad, "Pad is %s, dropping buffer",
2613       gst_flow_get_name (flow_return));
2614   if (buffer)
2615     gst_buffer_unref (buffer);
2616
2617   return flow_return;
2618
2619 eos:
2620   PAD_UNLOCK (aggpad);
2621   PAD_FLUSH_UNLOCK (aggpad);
2622
2623   gst_buffer_unref (buffer);
2624   GST_DEBUG_OBJECT (aggpad, "We are EOS already...");
2625
2626   return GST_FLOW_EOS;
2627 }
2628
2629 static GstFlowReturn
2630 gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
2631 {
2632   return gst_aggregator_pad_chain_internal (GST_AGGREGATOR_CAST (object),
2633       GST_AGGREGATOR_PAD_CAST (pad), buffer, TRUE);
2634 }
2635
2636 static gboolean
2637 gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
2638     GstQuery * query)
2639 {
2640   GstAggregator *self = GST_AGGREGATOR (parent);
2641   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
2642   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
2643
2644   if (GST_QUERY_IS_SERIALIZED (query)) {
2645     GstStructure *s;
2646     gboolean ret = FALSE;
2647
2648     SRC_LOCK (self);
2649     PAD_LOCK (aggpad);
2650
2651     if (aggpad->priv->flow_return != GST_FLOW_OK) {
2652       SRC_UNLOCK (self);
2653       goto flushing;
2654     }
2655
2656     g_queue_push_head (&aggpad->priv->buffers, query);
2657     SRC_BROADCAST (self);
2658     SRC_UNLOCK (self);
2659
2660     while (!gst_aggregator_pad_queue_is_empty (aggpad)
2661         && aggpad->priv->flow_return == GST_FLOW_OK) {
2662       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
2663       PAD_WAIT_EVENT (aggpad);
2664     }
2665
2666     s = gst_query_writable_structure (query);
2667     if (gst_structure_get_boolean (s, "gst-aggregator-retval", &ret))
2668       gst_structure_remove_field (s, "gst-aggregator-retval");
2669     else
2670       g_queue_remove (&aggpad->priv->buffers, query);
2671
2672     if (aggpad->priv->flow_return != GST_FLOW_OK)
2673       goto flushing;
2674
2675     PAD_UNLOCK (aggpad);
2676
2677     return ret;
2678   }
2679
2680   return klass->sink_query (self, aggpad, query);
2681
2682 flushing:
2683   GST_DEBUG_OBJECT (aggpad, "Pad is %s, dropping query",
2684       gst_flow_get_name (aggpad->priv->flow_return));
2685   PAD_UNLOCK (aggpad);
2686
2687   return FALSE;
2688 }
2689
2690 static GstFlowReturn
2691 gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
2692     GstEvent * event)
2693 {
2694   GstFlowReturn ret = GST_FLOW_OK;
2695   GstAggregator *self = GST_AGGREGATOR (parent);
2696   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
2697   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
2698
2699   if (GST_EVENT_IS_SERIALIZED (event) && GST_EVENT_TYPE (event) != GST_EVENT_EOS
2700       /* && GST_EVENT_TYPE (event) != GST_EVENT_SEGMENT_DONE */ ) {
2701     SRC_LOCK (self);
2702     PAD_LOCK (aggpad);
2703
2704     if (aggpad->priv->flow_return != GST_FLOW_OK
2705         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
2706       ret = aggpad->priv->flow_return;
2707       goto flushing;
2708     }
2709
2710     if (GST_EVENT_TYPE (event) == GST_EVENT_SEGMENT) {
2711       GST_OBJECT_LOCK (aggpad);
2712       gst_event_copy_segment (event, &aggpad->priv->head_segment);
2713       aggpad->priv->head_position = aggpad->priv->head_segment.position;
2714       update_time_level (aggpad, TRUE);
2715       GST_OBJECT_UNLOCK (aggpad);
2716     }
2717
2718     if (GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
2719       GST_DEBUG_OBJECT (aggpad, "Store event in queue: %" GST_PTR_FORMAT,
2720           event);
2721       g_queue_push_head (&aggpad->priv->buffers, event);
2722       event = NULL;
2723       SRC_BROADCAST (self);
2724     }
2725     PAD_UNLOCK (aggpad);
2726     SRC_UNLOCK (self);
2727   }
2728
2729   if (event) {
2730     if (!klass->sink_event (self, aggpad, event)) {
2731       /* Copied from GstPad to convert boolean to a GstFlowReturn in
2732        * the event handling func */
2733       ret = GST_FLOW_ERROR;
2734     }
2735   }
2736
2737   return ret;
2738
2739 flushing:
2740   GST_DEBUG_OBJECT (aggpad, "Pad is %s, dropping event",
2741       gst_flow_get_name (aggpad->priv->flow_return));
2742   PAD_UNLOCK (aggpad);
2743   SRC_UNLOCK (self);
2744   if (GST_EVENT_IS_STICKY (event))
2745     gst_pad_store_sticky_event (pad, event);
2746   gst_event_unref (event);
2747
2748   return ret;
2749 }
2750
2751 static gboolean
2752 gst_aggregator_pad_activate_mode_func (GstPad * pad,
2753     GstObject * parent, GstPadMode mode, gboolean active)
2754 {
2755   GstAggregator *self = GST_AGGREGATOR (parent);
2756   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
2757
2758   if (active == FALSE) {
2759     SRC_LOCK (self);
2760     gst_aggregator_pad_set_flushing (aggpad, GST_FLOW_FLUSHING, TRUE);
2761     SRC_BROADCAST (self);
2762     SRC_UNLOCK (self);
2763   } else {
2764     PAD_LOCK (aggpad);
2765     aggpad->priv->flow_return = GST_FLOW_OK;
2766     PAD_BROADCAST_EVENT (aggpad);
2767     PAD_UNLOCK (aggpad);
2768   }
2769
2770   return TRUE;
2771 }
2772
2773 /***********************************
2774  * GstAggregatorPad implementation  *
2775  ************************************/
2776 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
2777
2778 static void
2779 gst_aggregator_pad_constructed (GObject * object)
2780 {
2781   GstPad *pad = GST_PAD (object);
2782
2783   gst_pad_set_chain_function (pad,
2784       GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
2785   gst_pad_set_event_full_function_full (pad,
2786       GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func), NULL, NULL);
2787   gst_pad_set_query_function (pad,
2788       GST_DEBUG_FUNCPTR (gst_aggregator_pad_query_func));
2789   gst_pad_set_activatemode_function (pad,
2790       GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
2791 }
2792
2793 static void
2794 gst_aggregator_pad_finalize (GObject * object)
2795 {
2796   GstAggregatorPad *pad = (GstAggregatorPad *) object;
2797
2798   g_cond_clear (&pad->priv->event_cond);
2799   g_mutex_clear (&pad->priv->flush_lock);
2800   g_mutex_clear (&pad->priv->lock);
2801
2802   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->finalize (object);
2803 }
2804
2805 static void
2806 gst_aggregator_pad_dispose (GObject * object)
2807 {
2808   GstAggregatorPad *pad = (GstAggregatorPad *) object;
2809
2810   gst_aggregator_pad_set_flushing (pad, GST_FLOW_FLUSHING, TRUE);
2811
2812   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->dispose (object);
2813 }
2814
2815 static void
2816 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
2817 {
2818   GObjectClass *gobject_class = (GObjectClass *) klass;
2819
2820   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
2821
2822   gobject_class->constructed = gst_aggregator_pad_constructed;
2823   gobject_class->finalize = gst_aggregator_pad_finalize;
2824   gobject_class->dispose = gst_aggregator_pad_dispose;
2825 }
2826
2827 static void
2828 gst_aggregator_pad_init (GstAggregatorPad * pad)
2829 {
2830   pad->priv =
2831       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
2832       GstAggregatorPadPrivate);
2833
2834   g_queue_init (&pad->priv->buffers);
2835   g_cond_init (&pad->priv->event_cond);
2836
2837   g_mutex_init (&pad->priv->flush_lock);
2838   g_mutex_init (&pad->priv->lock);
2839
2840   gst_aggregator_pad_reset_unlocked (pad);
2841   pad->priv->negotiated = FALSE;
2842 }
2843
2844 /* Must be called with the PAD_LOCK held */
2845 static void
2846 gst_aggregator_pad_buffer_consumed (GstAggregatorPad * pad)
2847 {
2848   pad->priv->num_buffers--;
2849   GST_TRACE_OBJECT (pad, "Consuming buffer");
2850   if (gst_aggregator_pad_queue_is_empty (pad) && pad->priv->pending_eos) {
2851     pad->priv->pending_eos = FALSE;
2852     pad->priv->eos = TRUE;
2853   }
2854   PAD_BROADCAST_EVENT (pad);
2855 }
2856
2857 /* Must be called with the PAD_LOCK held */
2858 static void
2859 gst_aggregator_pad_clip_buffer_unlocked (GstAggregatorPad * pad)
2860 {
2861   GstAggregator *self = NULL;
2862   GstAggregatorClass *aggclass = NULL;
2863   GstBuffer *buffer = NULL;
2864
2865   while (pad->priv->clipped_buffer == NULL &&
2866       GST_IS_BUFFER (g_queue_peek_tail (&pad->priv->buffers))) {
2867     buffer = g_queue_pop_tail (&pad->priv->buffers);
2868
2869     apply_buffer (pad, buffer, FALSE);
2870
2871     /* We only take the parent here so that it's not taken if the buffer is
2872      * already clipped or if the queue is empty.
2873      */
2874     if (self == NULL) {
2875       self = GST_AGGREGATOR (gst_pad_get_parent_element (GST_PAD (pad)));
2876       if (self == NULL) {
2877         gst_buffer_unref (buffer);
2878         return;
2879       }
2880
2881       aggclass = GST_AGGREGATOR_GET_CLASS (self);
2882     }
2883
2884     if (aggclass->clip) {
2885       GST_TRACE_OBJECT (pad, "Clipping: %" GST_PTR_FORMAT, buffer);
2886
2887       buffer = aggclass->clip (self, pad, buffer);
2888
2889       if (buffer == NULL) {
2890         gst_aggregator_pad_buffer_consumed (pad);
2891         GST_TRACE_OBJECT (pad, "Clipping consumed the buffer");
2892       }
2893     }
2894
2895     pad->priv->clipped_buffer = buffer;
2896   }
2897
2898   if (self)
2899     gst_object_unref (self);
2900 }
2901
2902 /**
2903  * gst_aggregator_pad_steal_buffer:
2904  * @pad: the pad to get buffer from
2905  *
2906  * Steal the ref to the buffer currently queued in @pad.
2907  *
2908  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
2909  *   queued. You should unref the buffer after usage.
2910  */
2911 GstBuffer *
2912 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
2913 {
2914   GstBuffer *buffer;
2915
2916   PAD_LOCK (pad);
2917
2918   gst_aggregator_pad_clip_buffer_unlocked (pad);
2919
2920   buffer = pad->priv->clipped_buffer;
2921   pad->priv->clipped_buffer = NULL;
2922
2923   if (buffer) {
2924     gst_aggregator_pad_buffer_consumed (pad);
2925     GST_DEBUG_OBJECT (pad, "Consumed: %" GST_PTR_FORMAT, buffer);
2926   }
2927
2928   PAD_UNLOCK (pad);
2929
2930   return buffer;
2931 }
2932
2933 /**
2934  * gst_aggregator_pad_drop_buffer:
2935  * @pad: the pad where to drop any pending buffer
2936  *
2937  * Drop the buffer currently queued in @pad.
2938  *
2939  * Returns: TRUE if there was a buffer queued in @pad, or FALSE if not.
2940  */
2941 gboolean
2942 gst_aggregator_pad_drop_buffer (GstAggregatorPad * pad)
2943 {
2944   GstBuffer *buf;
2945
2946   buf = gst_aggregator_pad_steal_buffer (pad);
2947
2948   if (buf == NULL)
2949     return FALSE;
2950
2951   gst_buffer_unref (buf);
2952   return TRUE;
2953 }
2954
2955 /**
2956  * gst_aggregator_pad_get_buffer:
2957  * @pad: the pad to get buffer from
2958  *
2959  * Returns: (transfer full): A reference to the buffer in @pad or
2960  * NULL if no buffer was queued. You should unref the buffer after
2961  * usage.
2962  */
2963 GstBuffer *
2964 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
2965 {
2966   GstBuffer *buffer;
2967
2968   PAD_LOCK (pad);
2969
2970   gst_aggregator_pad_clip_buffer_unlocked (pad);
2971
2972   if (pad->priv->clipped_buffer) {
2973     buffer = gst_buffer_ref (pad->priv->clipped_buffer);
2974   } else {
2975     buffer = NULL;
2976   }
2977   PAD_UNLOCK (pad);
2978
2979   return buffer;
2980 }
2981
2982 gboolean
2983 gst_aggregator_pad_is_eos (GstAggregatorPad * pad)
2984 {
2985   gboolean is_eos;
2986
2987   PAD_LOCK (pad);
2988   is_eos = pad->priv->eos;
2989   PAD_UNLOCK (pad);
2990
2991   return is_eos;
2992 }
2993
2994 /**
2995  * gst_aggregator_merge_tags:
2996  * @self: a #GstAggregator
2997  * @tags: a #GstTagList to merge
2998  * @mode: the #GstTagMergeMode to use
2999  *
3000  * Adds tags to so-called pending tags, which will be processed
3001  * before pushing out data downstream.
3002  *
3003  * Note that this is provided for convenience, and the subclass is
3004  * not required to use this and can still do tag handling on its own.
3005  *
3006  * MT safe.
3007  */
3008 void
3009 gst_aggregator_merge_tags (GstAggregator * self,
3010     const GstTagList * tags, GstTagMergeMode mode)
3011 {
3012   GstTagList *otags;
3013
3014   g_return_if_fail (GST_IS_AGGREGATOR (self));
3015   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
3016
3017   /* FIXME Check if we can use OBJECT lock here! */
3018   GST_OBJECT_LOCK (self);
3019   if (tags)
3020     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
3021   otags = self->priv->tags;
3022   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
3023   if (otags)
3024     gst_tag_list_unref (otags);
3025   self->priv->tags_changed = TRUE;
3026   GST_OBJECT_UNLOCK (self);
3027 }
3028
3029 /**
3030  * gst_aggregator_set_latency:
3031  * @self: a #GstAggregator
3032  * @min_latency: minimum latency
3033  * @max_latency: maximum latency
3034  *
3035  * Lets #GstAggregator sub-classes tell the baseclass what their internal
3036  * latency is. Will also post a LATENCY message on the bus so the pipeline
3037  * can reconfigure its global latency.
3038  */
3039 void
3040 gst_aggregator_set_latency (GstAggregator * self,
3041     GstClockTime min_latency, GstClockTime max_latency)
3042 {
3043   gboolean changed = FALSE;
3044
3045   g_return_if_fail (GST_IS_AGGREGATOR (self));
3046   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (min_latency));
3047   g_return_if_fail (max_latency >= min_latency);
3048
3049   SRC_LOCK (self);
3050   if (self->priv->sub_latency_min != min_latency) {
3051     self->priv->sub_latency_min = min_latency;
3052     changed = TRUE;
3053   }
3054   if (self->priv->sub_latency_max != max_latency) {
3055     self->priv->sub_latency_max = max_latency;
3056     changed = TRUE;
3057   }
3058
3059   if (changed)
3060     SRC_BROADCAST (self);
3061   SRC_UNLOCK (self);
3062
3063   if (changed) {
3064     gst_element_post_message (GST_ELEMENT_CAST (self),
3065         gst_message_new_latency (GST_OBJECT_CAST (self)));
3066   }
3067 }
3068
3069 /**
3070  * gst_aggregator_get_buffer_pool:
3071  * @self: a #GstAggregator
3072  *
3073  * Returns: (transfer full): the instance of the #GstBufferPool used
3074  * by @trans; free it after use it
3075  */
3076 GstBufferPool *
3077 gst_aggregator_get_buffer_pool (GstAggregator * self)
3078 {
3079   GstBufferPool *pool;
3080
3081   g_return_val_if_fail (GST_IS_AGGREGATOR (self), NULL);
3082
3083   GST_OBJECT_LOCK (self);
3084   pool = self->priv->pool;
3085   if (pool)
3086     gst_object_ref (pool);
3087   GST_OBJECT_UNLOCK (self);
3088
3089   return pool;
3090 }
3091
3092 /**
3093  * gst_aggregator_get_allocator:
3094  * @self: a #GstAggregator
3095  * @allocator: (out) (allow-none) (transfer full): the #GstAllocator
3096  * used
3097  * @params: (out) (allow-none) (transfer full): the
3098  * #GstAllocationParams of @allocator
3099  *
3100  * Lets #GstAggregator sub-classes get the memory @allocator
3101  * acquired by the base class and its @params.
3102  *
3103  * Unref the @allocator after use it.
3104  */
3105 void
3106 gst_aggregator_get_allocator (GstAggregator * self,
3107     GstAllocator ** allocator, GstAllocationParams * params)
3108 {
3109   g_return_if_fail (GST_IS_AGGREGATOR (self));
3110
3111   if (allocator)
3112     *allocator = self->priv->allocator ?
3113         gst_object_ref (self->priv->allocator) : NULL;
3114
3115   if (params)
3116     *params = self->priv->allocation_params;
3117 }