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