libs/gst/base/gstaggregator.c

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