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