libs/gst/base/gstaggregator.c

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