libs/gst/base/gstaggregator.c

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