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