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