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