libs/gst/base/gstaggregator.c

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