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