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