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