libs/gst/base/gstaggregator.c

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