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