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