libs/gst/base/gstaggregator.c

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