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  * @short_description: manages a set of pads with the purpose of
  25  * aggregating their buffers.
  26  * @see_also: gstcollectpads for historical reasons.
  27  *
  28  * Manages a set of pads with the purpose of aggregating their buffers.
  29  * Control is given to the subclass when all pads have data.
  30  * <itemizedlist>
  31  *  <listitem><para>
  32  *    Base class for mixers and muxers. Subclasses should at least implement
  33  *    the #GstAggregatorClass.aggregate() virtual method.
  34  *  </para></listitem>
  35  *  <listitem><para>
  36  *    When data is queued on all pads, tha aggregate vmethod is called.
  37  *  </para></listitem>
  38  *  <listitem><para>
  39  *    One can peek at the data on any given GstAggregatorPad with the
  40  *    gst_aggregator_pad_get_buffer () method, and take ownership of it
  41  *    with the gst_aggregator_pad_steal_buffer () method. When a buffer
  42  *    has been taken with steal_buffer (), a new buffer can be queued
  43  *    on that pad.
  44  *  </para></listitem>
  45  *  <listitem><para>
  46  *    If the subclass wishes to push a buffer downstream in its aggregate
  47  *    implementation, it should do so through the
  48  *    gst_aggregator_finish_buffer () method. This method will take care
  49  *    of sending and ordering mandatory events such as stream start, caps
  50  *    and segment.
  51  *  </para></listitem>
  52  *  <listitem><para>
  53  *    Same goes for EOS events, which should not be pushed directly by the
  54  *    subclass, it should instead return GST_FLOW_EOS in its aggregate
  55  *    implementation.
  56  *  </para></listitem>
  57  * </itemizedlist>
  58  */
  59
  60 #ifdef HAVE_CONFIG_H
  61 #  include "config.h"
  62 #endif
  63
  64 #include <string.h>             /* strlen */
  65
  66 #include "gstaggregator.h"
  67
  68
  69 /*  Might become API */
  70 static void gst_aggregator_merge_tags (GstAggregator * aggregator,
  71     const GstTagList * tags, GstTagMergeMode mode);
  72 static void gst_aggregator_set_latency_property (GstAggregator * agg,
  73     gint64 latency);
  74 static gint64 gst_aggregator_get_latency_property (GstAggregator * agg);
  75
  76
  77 GST_DEBUG_CATEGORY_STATIC (aggregator_debug);
  78 #define GST_CAT_DEFAULT aggregator_debug
  79
  80 /* GstAggregatorPad definitions */
  81 #define PAD_LOCK(pad)   G_STMT_START {                                  \
  82   GST_TRACE_OBJECT (pad, "Taking PAD lock from thread %p",              \
  83         g_thread_self());                                               \
  84   g_mutex_lock(&pad->priv->lock);                                       \
  85   GST_TRACE_OBJECT (pad, "Took PAD lock from thread %p",                \
  86         g_thread_self());                                               \
  87   } G_STMT_END
  88
  89 #define PAD_UNLOCK(pad)  G_STMT_START {                                 \
  90   GST_TRACE_OBJECT (pad, "Releasing PAD lock from thread %p",           \
  91       g_thread_self());                                                 \
  92   g_mutex_unlock(&pad->priv->lock);                                     \
  93   GST_TRACE_OBJECT (pad, "Release PAD lock from thread %p",             \
  94         g_thread_self());                                               \
  95   } G_STMT_END
  96
  97
  98 #define PAD_WAIT_EVENT(pad)   G_STMT_START {                            \
  99   GST_LOG_OBJECT (pad, "Waiting for EVENT on thread %p",                \
 100         g_thread_self());                                               \
 101   g_cond_wait(&(((GstAggregatorPad* )pad)->priv->event_cond),           \
 102       (&((GstAggregatorPad*)pad)->priv->lock));                         \
 103   GST_LOG_OBJECT (pad, "DONE Waiting for EVENT on thread %p",           \
 104         g_thread_self());                                               \
 105   } G_STMT_END
 106
 107 #define PAD_BROADCAST_EVENT(pad) G_STMT_START {                        \
 108   GST_LOG_OBJECT (pad, "Signaling EVENT from thread %p",               \
 109         g_thread_self());                                              \
 110   g_cond_broadcast(&(((GstAggregatorPad* )pad)->priv->event_cond));    \
 111   } G_STMT_END
 112
 113
 114 #define PAD_FLUSH_LOCK(pad)     G_STMT_START {                          \
 115   GST_TRACE_OBJECT (pad, "Taking lock from thread %p",                  \
 116         g_thread_self());                                               \
 117   g_mutex_lock(&pad->priv->flush_lock);                                 \
 118   GST_TRACE_OBJECT (pad, "Took lock from thread %p",                    \
 119         g_thread_self());                                               \
 120   } G_STMT_END
 121
 122 #define PAD_FLUSH_UNLOCK(pad)   G_STMT_START {                          \
 123   GST_TRACE_OBJECT (pad, "Releasing lock from thread %p",               \
 124         g_thread_self());                                               \
 125   g_mutex_unlock(&pad->priv->flush_lock);                               \
 126   GST_TRACE_OBJECT (pad, "Release lock from thread %p",                 \
 127         g_thread_self());                                               \
 128   } G_STMT_END
 129
 130 #define SRC_STREAM_LOCK(self)   G_STMT_START {                             \
 131   GST_TRACE_OBJECT (self, "Taking src STREAM lock from thread %p",         \
 132         g_thread_self());                                                  \
 133   g_mutex_lock(&self->priv->src_lock);                                     \
 134   GST_TRACE_OBJECT (self, "Took src STREAM lock from thread %p",           \
 135         g_thread_self());                                                  \
 136   } G_STMT_END
 137
 138 #define SRC_STREAM_UNLOCK(self)  G_STMT_START {                            \
 139   GST_TRACE_OBJECT (self, "Releasing src STREAM lock from thread %p",      \
 140         g_thread_self());                                                  \
 141   g_mutex_unlock(&self->priv->src_lock);                                   \
 142   GST_TRACE_OBJECT (self, "Released src STREAM lock from thread %p",       \
 143         g_thread_self());                                                  \
 144   } G_STMT_END
 145
 146 #define SRC_STREAM_WAIT(self) G_STMT_START {                               \
 147   GST_LOG_OBJECT (self, "Waiting for src STREAM on thread %p",             \
 148         g_thread_self());                                                  \
 149   g_cond_wait(&(self->priv->src_cond), &(self->priv->src_lock));           \
 150   GST_LOG_OBJECT (self, "DONE Waiting for src STREAM on thread %p",        \
 151         g_thread_self());                                                  \
 152   } G_STMT_END
 153
 154 #define SRC_STREAM_BROADCAST(self) G_STMT_START {                 \
 155     GST_LOG_OBJECT (self, "Signaling src STREAM from thread %p",           \
 156         g_thread_self());                                                  \
 157     if (self->priv->aggregate_id)                                          \
 158       gst_clock_id_unschedule (self->priv->aggregate_id);                  \
 159     g_cond_broadcast(&(self->priv->src_cond));                             \
 160   } G_STMT_END
 161
 162 struct _GstAggregatorPadPrivate
 163 {
 164   /* To always be used atomically */
 165   gboolean flushing;
 166
 167   /* Following fields are protected by the PAD_LOCK */
 168   gboolean pending_flush_start;
 169   gboolean pending_flush_stop;
 170   gboolean pending_eos;
 171
 172   GstBuffer *buffer;
 173   gboolean eos;
 174
 175   GMutex lock;
 176   GCond event_cond;
 177   /* This lock prevents a flush start processing happening while
 178    * the chain function is also happening.
 179    */
 180   GMutex flush_lock;
 181 };
 182
 183 static gboolean
 184 gst_aggregator_pad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
 185 {
 186   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 187
 188   PAD_LOCK (aggpad);
 189   aggpad->priv->eos = FALSE;
 190   aggpad->priv->flushing = FALSE;
 191   PAD_UNLOCK (aggpad);
 192
 193   if (klass->flush)
 194     return klass->flush (aggpad, agg);
 195
 196   return TRUE;
 197 }
 198
 199 /*************************************
 200  * GstAggregator implementation  *
 201  *************************************/
 202 static GstElementClass *aggregator_parent_class = NULL;
 203
 204 /* All members are protected by the object lock unless otherwise noted */
 205
 206 struct _GstAggregatorPrivate
 207 {
 208   gint padcount;
 209
 210   /* Our state is >= PAUSED */
 211   gboolean running;             /* protected by SRC_STREAM_LOCK */
 212
 213   gint seqnum;
 214   gboolean send_stream_start;   /* protected by srcpad stream lock */
 215   gboolean send_segment;
 216   gboolean flush_seeking;
 217   gboolean pending_flush_start;
 218   gboolean send_eos;            /* protected by srcpad stream lock */
 219   GstFlowReturn flow_return;
 220
 221   GstCaps *srccaps;             /* protected by the srcpad stream lock */
 222
 223   GstTagList *tags;
 224   gboolean tags_changed;
 225
 226   gboolean latency_live;
 227   GstClockTime latency_min;
 228   GstClockTime latency_max;
 229
 230   GstClockTime sub_latency_min;
 231   GstClockTime sub_latency_max;
 232
 233   /* aggregate */
 234   GstClockID aggregate_id;      /* protected by src_lock */
 235   GMutex src_lock;
 236   GCond src_cond;
 237
 238   /* properties */
 239   gint64 latency;
 240 };
 241
 242 typedef struct
 243 {
 244   GstEvent *event;
 245   gboolean result;
 246   gboolean flush;
 247
 248   gboolean one_actually_seeked;
 249 } EventData;
 250
 251 #define DEFAULT_LATENCY        0
 252
 253 enum
 254 {
 255   PROP_0,
 256   PROP_LATENCY,
 257   PROP_LAST
 258 };
 259
 260 /**
 261  * gst_aggregator_iterate_sinkpads:
 262  * @self: The #GstAggregator
 263  * @func: (scope call): The function to call.
 264  * @user_data: (closure): The data to pass to @func.
 265  *
 266  * Iterate the sinkpads of aggregator to call a function on them.
 267  *
 268  * This method guarantees that @func will be called only once for each
 269  * sink pad.
 270  */
 271 gboolean
 272 gst_aggregator_iterate_sinkpads (GstAggregator * self,
 273     GstAggregatorPadForeachFunc func, gpointer user_data)
 274 {
 275   gboolean result = FALSE;
 276   GstIterator *iter;
 277   gboolean done = FALSE;
 278   GValue item = { 0, };
 279   GList *seen_pads = NULL;
 280
 281   iter = gst_element_iterate_sink_pads (GST_ELEMENT (self));
 282
 283   if (!iter)
 284     goto no_iter;
 285
 286   while (!done) {
 287     switch (gst_iterator_next (iter, &item)) {
 288       case GST_ITERATOR_OK:
 289       {
 290         GstAggregatorPad *pad;
 291
 292         pad = g_value_get_object (&item);
 293
 294         /* if already pushed, skip. FIXME, find something faster to tag pads */
 295         if (pad == NULL || g_list_find (seen_pads, pad)) {
 296           g_value_reset (&item);
 297           break;
 298         }
 299
 300         GST_LOG_OBJECT (pad, "calling function %s on pad",
 301             GST_DEBUG_FUNCPTR_NAME (func));
 302
 303         result = func (self, pad, user_data);
 304
 305         done = !result;
 306
 307         seen_pads = g_list_prepend (seen_pads, pad);
 308
 309         g_value_reset (&item);
 310         break;
 311       }
 312       case GST_ITERATOR_RESYNC:
 313         gst_iterator_resync (iter);
 314         break;
 315       case GST_ITERATOR_ERROR:
 316         GST_ERROR_OBJECT (self,
 317             "Could not iterate over internally linked pads");
 318         done = TRUE;
 319         break;
 320       case GST_ITERATOR_DONE:
 321         done = TRUE;
 322         break;
 323     }
 324   }
 325   g_value_unset (&item);
 326   gst_iterator_free (iter);
 327
 328   if (seen_pads == NULL) {
 329     GST_DEBUG_OBJECT (self, "No pad seen");
 330     return FALSE;
 331   }
 332
 333   g_list_free (seen_pads);
 334
 335 no_iter:
 336   return result;
 337 }
 338
 339 static gboolean
 340 gst_aggregator_check_pads_ready (GstAggregator * self)
 341 {
 342   GstAggregatorPad *pad;
 343   GList *l, *sinkpads;
 344
 345   GST_LOG_OBJECT (self, "checking pads");
 346
 347   GST_OBJECT_LOCK (self);
 348
 349   sinkpads = GST_ELEMENT_CAST (self)->sinkpads;
 350   if (sinkpads == NULL)
 351     goto no_sinkpads;
 352
 353   for (l = sinkpads; l != NULL; l = l->next) {
 354     pad = l->data;
 355
 356     PAD_LOCK (pad);
 357     if (pad->priv->buffer == NULL && !pad->priv->eos) {
 358       PAD_UNLOCK (pad);
 359       goto pad_not_ready;
 360     }
 361     PAD_UNLOCK (pad);
 362
 363   }
 364
 365   GST_OBJECT_UNLOCK (self);
 366   GST_LOG_OBJECT (self, "pads are ready");
 367   return TRUE;
 368
 369 no_sinkpads:
 370   {
 371     GST_LOG_OBJECT (self, "pads not ready: no sink pads");
 372     GST_OBJECT_UNLOCK (self);
 373     return FALSE;
 374   }
 375 pad_not_ready:
 376   {
 377     GST_LOG_OBJECT (pad, "pad not ready to be aggregated yet");
 378     GST_OBJECT_UNLOCK (self);
 379     return FALSE;
 380   }
 381 }
 382
 383 static void
 384 gst_aggregator_reset_flow_values (GstAggregator * self)
 385 {
 386   GST_OBJECT_LOCK (self);
 387   self->priv->flow_return = GST_FLOW_FLUSHING;
 388   self->priv->send_stream_start = TRUE;
 389   self->priv->send_segment = TRUE;
 390   gst_segment_init (&self->segment, GST_FORMAT_TIME);
 391   GST_OBJECT_UNLOCK (self);
 392 }
 393
 394 static inline void
 395 gst_aggregator_push_mandatory_events (GstAggregator * self)
 396 {
 397   GstAggregatorPrivate *priv = self->priv;
 398   GstEvent *segment = NULL;
 399   GstEvent *tags = NULL;
 400
 401   if (self->priv->send_stream_start) {
 402     gchar s_id[32];
 403
 404     GST_INFO_OBJECT (self, "pushing stream start");
 405     /* stream-start (FIXME: create id based on input ids) */
 406     g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
 407     if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
 408       GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
 409     }
 410     self->priv->send_stream_start = FALSE;
 411   }
 412
 413   if (self->priv->srccaps) {
 414
 415     GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
 416         self->priv->srccaps);
 417     if (!gst_pad_push_event (self->srcpad,
 418             gst_event_new_caps (self->priv->srccaps))) {
 419       GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
 420     }
 421     gst_caps_unref (self->priv->srccaps);
 422     self->priv->srccaps = NULL;
 423   }
 424
 425   GST_OBJECT_LOCK (self);
 426   if (self->priv->send_segment && !self->priv->flush_seeking) {
 427     segment = gst_event_new_segment (&self->segment);
 428
 429     if (!self->priv->seqnum)
 430       self->priv->seqnum = gst_event_get_seqnum (segment);
 431     else
 432       gst_event_set_seqnum (segment, self->priv->seqnum);
 433     self->priv->send_segment = FALSE;
 434
 435     GST_DEBUG_OBJECT (self, "pushing segment %" GST_PTR_FORMAT, segment);
 436   }
 437
 438   if (priv->tags && priv->tags_changed) {
 439     tags = gst_event_new_tag (gst_tag_list_ref (priv->tags));
 440     priv->tags_changed = FALSE;
 441   }
 442   GST_OBJECT_UNLOCK (self);
 443
 444   if (segment)
 445     gst_pad_push_event (self->srcpad, segment);
 446   if (tags)
 447     gst_pad_push_event (self->srcpad, tags);
 448
 449 }
 450
 451 /**
 452  * gst_aggregator_set_src_caps:
 453  * @self: The #GstAggregator
 454  * @caps: The #GstCaps to set on the src pad.
 455  *
 456  * Sets the caps to be used on the src pad.
 457  */
 458 void
 459 gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps)
 460 {
 461   GST_PAD_STREAM_LOCK (self->srcpad);
 462   gst_caps_replace (&self->priv->srccaps, caps);
 463   gst_aggregator_push_mandatory_events (self);
 464   GST_PAD_STREAM_UNLOCK (self->srcpad);
 465 }
 466
 467 /**
 468  * gst_aggregator_finish_buffer:
 469  * @self: The #GstAggregator
 470  * @buffer: (transfer full): the #GstBuffer to push.
 471  *
 472  * This method will push the provided output buffer downstream. If needed,
 473  * mandatory events such as stream-start, caps, and segment events will be
 474  * sent before pushing the buffer.
 475  */
 476 GstFlowReturn
 477 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 478 {
 479   gst_aggregator_push_mandatory_events (self);
 480
 481   GST_OBJECT_LOCK (self);
 482   if (!self->priv->flush_seeking && gst_pad_is_active (self->srcpad)) {
 483     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 484     GST_OBJECT_UNLOCK (self);
 485     return gst_pad_push (self->srcpad, buffer);
 486   } else {
 487     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 488         self->priv->flush_seeking, gst_pad_is_active (self->srcpad));
 489     GST_OBJECT_UNLOCK (self);
 490     gst_buffer_unref (buffer);
 491     return GST_FLOW_OK;
 492   }
 493 }
 494
 495 static void
 496 gst_aggregator_push_eos (GstAggregator * self)
 497 {
 498   GstEvent *event;
 499   gst_aggregator_push_mandatory_events (self);
 500
 501   event = gst_event_new_eos ();
 502
 503   GST_OBJECT_LOCK (self);
 504   self->priv->send_eos = FALSE;
 505   gst_event_set_seqnum (event, self->priv->seqnum);
 506   GST_OBJECT_UNLOCK (self);
 507
 508   gst_pad_push_event (self->srcpad, event);
 509 }
 510
 511 static GstClockTime
 512 gst_aggregator_get_next_time (GstAggregator * self)
 513 {
 514   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 515
 516   if (klass->get_next_time)
 517     return klass->get_next_time (self);
 518
 519   return GST_CLOCK_TIME_NONE;
 520 }
 521
 522 /* called with the src STREAM lock */
 523 static gboolean
 524 gst_aggregator_wait_and_check (GstAggregator * self, gboolean * timeout)
 525 {
 526   GstClockTime latency_max, latency_min;
 527   GstClockTime start;
 528   gboolean live, res;
 529
 530   *timeout = FALSE;
 531
 532   SRC_STREAM_LOCK (self);
 533
 534   GST_OBJECT_LOCK (self);
 535   gst_aggregator_get_latency_unlocked (self, &live, &latency_min, &latency_max);
 536   GST_OBJECT_UNLOCK (self);
 537
 538   if (gst_aggregator_check_pads_ready (self)) {
 539     GST_DEBUG_OBJECT (self, "all pads have data");
 540     SRC_STREAM_UNLOCK (self);
 541
 542     return TRUE;
 543   }
 544
 545   /* Before waiting, check if we're actually still running */
 546   if (!self->priv->running || !self->priv->send_eos) {
 547     SRC_STREAM_UNLOCK (self);
 548
 549     return FALSE;
 550   }
 551
 552   start = gst_aggregator_get_next_time (self);
 553
 554   if (!live || !GST_IS_CLOCK (GST_ELEMENT_CLOCK (self))
 555       || !GST_CLOCK_TIME_IS_VALID (start)) {
 556     /* We wake up here when something happened, and below
 557      * then check if we're ready now. If we return FALSE,
 558      * we will be directly called again.
 559      */
 560     SRC_STREAM_WAIT (self);
 561   } else {
 562     GstClockTime base_time, time;
 563     GstClock *clock;
 564     GstClockReturn status;
 565     GstClockTimeDiff jitter;
 566
 567     GST_DEBUG_OBJECT (self, "got subclass start time: %" GST_TIME_FORMAT,
 568         GST_TIME_ARGS (start));
 569
 570     GST_OBJECT_LOCK (self);
 571     base_time = GST_ELEMENT_CAST (self)->base_time;
 572     clock = GST_ELEMENT_CLOCK (self);
 573     if (clock)
 574       gst_object_ref (clock);
 575
 576     time = base_time + start;
 577     time += latency_min;
 578
 579     GST_DEBUG_OBJECT (self, "possibly waiting for clock to reach %"
 580         GST_TIME_FORMAT " (base %" GST_TIME_FORMAT " start %" GST_TIME_FORMAT
 581         " latency max %" GST_TIME_FORMAT " latency min %" GST_TIME_FORMAT
 582         " current %" GST_TIME_FORMAT ")", GST_TIME_ARGS (time),
 583         GST_TIME_ARGS (GST_ELEMENT_CAST (self)->base_time),
 584         GST_TIME_ARGS (start), GST_TIME_ARGS (latency_max),
 585         GST_TIME_ARGS (latency_min),
 586         GST_TIME_ARGS (gst_clock_get_time (clock)));
 587
 588     GST_OBJECT_UNLOCK (self);
 589
 590     self->priv->aggregate_id = gst_clock_new_single_shot_id (clock, time);
 591     gst_object_unref (clock);
 592     SRC_STREAM_UNLOCK (self);
 593
 594     jitter = 0;
 595     status = gst_clock_id_wait (self->priv->aggregate_id, &jitter);
 596
 597     SRC_STREAM_LOCK (self);
 598     if (self->priv->aggregate_id) {
 599       gst_clock_id_unref (self->priv->aggregate_id);
 600       self->priv->aggregate_id = NULL;
 601     }
 602
 603     GST_DEBUG_OBJECT (self,
 604         "clock returned %d (jitter: %s%" GST_TIME_FORMAT ")",
 605         status, (jitter < 0 ? "-" : " "),
 606         GST_TIME_ARGS ((jitter < 0 ? -jitter : jitter)));
 607
 608     /* we timed out */
 609     if (status == GST_CLOCK_OK || status == GST_CLOCK_EARLY) {
 610       SRC_STREAM_UNLOCK (self);
 611       *timeout = TRUE;
 612       return TRUE;
 613     }
 614   }
 615
 616   res = gst_aggregator_check_pads_ready (self);
 617   SRC_STREAM_UNLOCK (self);
 618
 619   return res;
 620 }
 621
 622 static void
 623 gst_aggregator_aggregate_func (GstAggregator * self)
 624 {
 625   GstAggregatorPrivate *priv = self->priv;
 626   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 627   gboolean timeout = FALSE;
 628
 629   if (self->priv->running == FALSE) {
 630     GST_DEBUG_OBJECT (self, "Not running anymore");
 631     return;
 632   }
 633
 634   GST_LOG_OBJECT (self, "Checking aggregate");
 635   while (priv->send_eos && priv->running) {
 636     GstFlowReturn flow_return;
 637
 638     if (!gst_aggregator_wait_and_check (self, &timeout))
 639       continue;
 640
 641     GST_TRACE_OBJECT (self, "Actually aggregating!");
 642
 643     flow_return = klass->aggregate (self, timeout);
 644
 645     GST_OBJECT_LOCK (self);
 646     if (flow_return == GST_FLOW_FLUSHING && priv->flush_seeking)
 647       priv->flow_return = GST_FLOW_OK;
 648     else
 649       priv->flow_return = flow_return;
 650     GST_OBJECT_UNLOCK (self);
 651
 652     if (flow_return == GST_FLOW_EOS) {
 653       gst_aggregator_push_eos (self);
 654     }
 655
 656     GST_LOG_OBJECT (self, "flow return is %s", gst_flow_get_name (flow_return));
 657
 658     if (flow_return != GST_FLOW_OK)
 659       break;
 660   }
 661
 662   /* Pause the task here, the only ways to get here are:
 663    * 1) We're stopping, in which case the task is stopped anyway
 664    * 2) We got a flow error above, in which case it might take
 665    *    some time to forward the flow return upstream and we
 666    *    would otherwise call the task function over and over
 667    *    again without doing anything
 668    */
 669   gst_pad_pause_task (self->srcpad);
 670 }
 671
 672 static gboolean
 673 gst_aggregator_start (GstAggregator * self)
 674 {
 675   GstAggregatorClass *klass;
 676   gboolean result;
 677
 678   self->priv->running = TRUE;
 679   self->priv->send_stream_start = TRUE;
 680   self->priv->send_segment = TRUE;
 681   self->priv->send_eos = TRUE;
 682   self->priv->srccaps = NULL;
 683   self->priv->flow_return = GST_FLOW_OK;
 684
 685   klass = GST_AGGREGATOR_GET_CLASS (self);
 686
 687   if (klass->start)
 688     result = klass->start (self);
 689   else
 690     result = TRUE;
 691
 692   return result;
 693 }
 694
 695 static gboolean
 696 _check_pending_flush_stop (GstAggregatorPad * pad)
 697 {
 698   gboolean res;
 699
 700   PAD_LOCK (pad);
 701   res = (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
 702   PAD_UNLOCK (pad);
 703
 704   return res;
 705 }
 706
 707 static gboolean
 708 gst_aggregator_stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
 709 {
 710   gboolean res = TRUE;
 711
 712   GST_INFO_OBJECT (self, "%s srcpad task",
 713       flush_start ? "Pausing" : "Stopping");
 714
 715   SRC_STREAM_LOCK (self);
 716   self->priv->running = FALSE;
 717   SRC_STREAM_BROADCAST (self);
 718   SRC_STREAM_UNLOCK (self);
 719
 720   if (flush_start) {
 721     res = gst_pad_push_event (self->srcpad, flush_start);
 722   }
 723
 724   gst_pad_stop_task (self->srcpad);
 725
 726   return res;
 727 }
 728
 729 static void
 730 gst_aggregator_start_srcpad_task (GstAggregator * self)
 731 {
 732   GST_INFO_OBJECT (self, "Starting srcpad task");
 733
 734   self->priv->running = TRUE;
 735   gst_pad_start_task (GST_PAD (self->srcpad),
 736       (GstTaskFunction) gst_aggregator_aggregate_func, self, NULL);
 737 }
 738
 739 static GstFlowReturn
 740 gst_aggregator_flush (GstAggregator * self)
 741 {
 742   GstFlowReturn ret = GST_FLOW_OK;
 743   GstAggregatorPrivate *priv = self->priv;
 744   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 745
 746   GST_DEBUG_OBJECT (self, "Flushing everything");
 747   GST_OBJECT_LOCK (self);
 748   priv->send_segment = TRUE;
 749   priv->flush_seeking = FALSE;
 750   priv->tags_changed = FALSE;
 751   GST_OBJECT_UNLOCK (self);
 752   if (klass->flush)
 753     ret = klass->flush (self);
 754
 755   return ret;
 756 }
 757
 758
 759 /* Called with GstAggregator's object lock held */
 760
 761 static gboolean
 762 gst_aggregator_all_flush_stop_received_locked (GstAggregator * self)
 763 {
 764   GList *tmp;
 765   GstAggregatorPad *tmppad;
 766
 767   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
 768     tmppad = (GstAggregatorPad *) tmp->data;
 769
 770     if (_check_pending_flush_stop (tmppad) == FALSE) {
 771       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
 772           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
 773       return FALSE;
 774     }
 775   }
 776
 777   return TRUE;
 778 }
 779
 780 static void
 781 gst_aggregator_flush_start (GstAggregator * self, GstAggregatorPad * aggpad,
 782     GstEvent * event)
 783 {
 784   GstAggregatorPrivate *priv = self->priv;
 785   GstAggregatorPadPrivate *padpriv = aggpad->priv;
 786
 787   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 788
 789   /*  Remove pad buffer and wake up the streaming thread */
 790   gst_aggregator_pad_drop_buffer (aggpad);
 791
 792   PAD_FLUSH_LOCK (aggpad);
 793   PAD_LOCK (aggpad);
 794   if (padpriv->pending_flush_start) {
 795     GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
 796
 797     padpriv->pending_flush_start = FALSE;
 798     padpriv->pending_flush_stop = TRUE;
 799   }
 800   PAD_UNLOCK (aggpad);
 801
 802   GST_OBJECT_LOCK (self);
 803   if (priv->flush_seeking) {
 804     /* If flush_seeking we forward the first FLUSH_START */
 805     if (priv->pending_flush_start) {
 806       priv->pending_flush_start = FALSE;
 807       GST_OBJECT_UNLOCK (self);
 808
 809       GST_INFO_OBJECT (self, "Flushing, pausing srcpad task");
 810       gst_aggregator_stop_srcpad_task (self, event);
 811       priv->flow_return = GST_FLOW_OK;
 812
 813       GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
 814       GST_PAD_STREAM_LOCK (self->srcpad);
 815       GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
 816       event = NULL;
 817     } else {
 818       GST_OBJECT_UNLOCK (self);
 819       gst_event_unref (event);
 820     }
 821   } else {
 822     GST_OBJECT_UNLOCK (self);
 823     gst_event_unref (event);
 824   }
 825   PAD_FLUSH_UNLOCK (aggpad);
 826
 827   gst_aggregator_pad_drop_buffer (aggpad);
 828 }
 829
 830 /* GstAggregator vmethods default implementations */
 831 static gboolean
 832 gst_aggregator_default_sink_event (GstAggregator * self,
 833     GstAggregatorPad * aggpad, GstEvent * event)
 834 {
 835   gboolean res = TRUE;
 836   GstPad *pad = GST_PAD (aggpad);
 837   GstAggregatorPrivate *priv = self->priv;
 838
 839   switch (GST_EVENT_TYPE (event)) {
 840     case GST_EVENT_FLUSH_START:
 841     {
 842       gst_aggregator_flush_start (self, aggpad, event);
 843       /* We forward only in one case: right after flush_seeking */
 844       event = NULL;
 845       goto eat;
 846     }
 847     case GST_EVENT_FLUSH_STOP:
 848     {
 849       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
 850
 851       gst_aggregator_pad_flush (aggpad, self);
 852       GST_OBJECT_LOCK (self);
 853       if (priv->flush_seeking) {
 854         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
 855         if (gst_aggregator_all_flush_stop_received_locked (self)) {
 856           GST_OBJECT_UNLOCK (self);
 857           /* That means we received FLUSH_STOP/FLUSH_STOP on
 858            * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
 859           gst_aggregator_flush (self);
 860           gst_pad_push_event (self->srcpad, event);
 861           event = NULL;
 862           SRC_STREAM_LOCK (self);
 863           priv->send_eos = TRUE;
 864           SRC_STREAM_BROADCAST (self);
 865           SRC_STREAM_UNLOCK (self);
 866
 867           GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
 868           GST_PAD_STREAM_UNLOCK (self->srcpad);
 869           gst_aggregator_start_srcpad_task (self);
 870         } else {
 871           GST_OBJECT_UNLOCK (self);
 872         }
 873       } else {
 874         GST_OBJECT_UNLOCK (self);
 875       }
 876
 877       /* We never forward the event */
 878       goto eat;
 879     }
 880     case GST_EVENT_EOS:
 881     {
 882       GST_DEBUG_OBJECT (aggpad, "EOS");
 883
 884       /* We still have a buffer, and we don't want the subclass to have to
 885        * check for it. Mark pending_eos, eos will be set when steal_buffer is
 886        * called
 887        */
 888       SRC_STREAM_LOCK (self);
 889       PAD_LOCK (aggpad);
 890       if (!aggpad->priv->buffer) {
 891         aggpad->priv->eos = TRUE;
 892       } else {
 893         aggpad->priv->pending_eos = TRUE;
 894       }
 895       PAD_UNLOCK (aggpad);
 896
 897       SRC_STREAM_BROADCAST (self);
 898       SRC_STREAM_UNLOCK (self);
 899       goto eat;
 900     }
 901     case GST_EVENT_SEGMENT:
 902     {
 903       GST_OBJECT_LOCK (aggpad);
 904       gst_event_copy_segment (event, &aggpad->segment);
 905       GST_OBJECT_UNLOCK (aggpad);
 906
 907       GST_OBJECT_LOCK (self);
 908       self->priv->seqnum = gst_event_get_seqnum (event);
 909       GST_OBJECT_UNLOCK (self);
 910       goto eat;
 911     }
 912     case GST_EVENT_STREAM_START:
 913     {
 914       goto eat;
 915     }
 916     case GST_EVENT_GAP:
 917     {
 918       /* FIXME: need API to handle GAP events properly */
 919       GST_FIXME_OBJECT (self, "implement support for GAP events");
 920       /* don't forward GAP events downstream */
 921       goto eat;
 922     }
 923     case GST_EVENT_TAG:
 924     {
 925       GstTagList *tags;
 926
 927       gst_event_parse_tag (event, &tags);
 928
 929       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
 930         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
 931         gst_event_unref (event);
 932         event = NULL;
 933         goto eat;
 934       }
 935       break;
 936     }
 937     default:
 938     {
 939       break;
 940     }
 941   }
 942
 943   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
 944   return gst_pad_event_default (pad, GST_OBJECT (self), event);
 945
 946 eat:
 947   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
 948   if (event)
 949     gst_event_unref (event);
 950
 951   return res;
 952 }
 953
 954 static inline gboolean
 955 gst_aggregator_stop_pad (GstAggregator * self, GstAggregatorPad * pad,
 956     gpointer unused_udata)
 957 {
 958   gst_aggregator_pad_flush (pad, self);
 959
 960   return TRUE;
 961 }
 962
 963 static gboolean
 964 gst_aggregator_stop (GstAggregator * agg)
 965 {
 966   GstAggregatorClass *klass;
 967   gboolean result;
 968
 969   gst_aggregator_reset_flow_values (agg);
 970
 971   gst_aggregator_iterate_sinkpads (agg, gst_aggregator_stop_pad, NULL);
 972
 973   klass = GST_AGGREGATOR_GET_CLASS (agg);
 974
 975   if (klass->stop)
 976     result = klass->stop (agg);
 977   else
 978     result = TRUE;
 979
 980   if (agg->priv->tags)
 981     gst_tag_list_unref (agg->priv->tags);
 982   agg->priv->tags = NULL;
 983
 984   return result;
 985 }
 986
 987 /* GstElement vmethods implementations */
 988 static GstStateChangeReturn
 989 gst_aggregator_change_state (GstElement * element, GstStateChange transition)
 990 {
 991   GstStateChangeReturn ret;
 992   GstAggregator *self = GST_AGGREGATOR (element);
 993
 994   switch (transition) {
 995     case GST_STATE_CHANGE_READY_TO_PAUSED:
 996       if (!gst_aggregator_start (self))
 997         goto error_start;
 998       break;
 999     default:
1000       break;
1001   }
1002
1003   if ((ret =
1004           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
1005               transition)) == GST_STATE_CHANGE_FAILURE)
1006     goto failure;
1007
1008
1009   switch (transition) {
1010     case GST_STATE_CHANGE_PAUSED_TO_READY:
1011       if (!gst_aggregator_stop (self)) {
1012         /* What to do in this case? Error out? */
1013         GST_ERROR_OBJECT (self, "Subclass failed to stop.");
1014       }
1015       break;
1016     default:
1017       break;
1018   }
1019
1020   return ret;
1021
1022 /* ERRORS */
1023 failure:
1024   {
1025     GST_ERROR_OBJECT (element, "parent failed state change");
1026     return ret;
1027   }
1028 error_start:
1029   {
1030     GST_ERROR_OBJECT (element, "Subclass failed to start");
1031     return GST_STATE_CHANGE_FAILURE;
1032   }
1033 }
1034
1035 static void
1036 gst_aggregator_release_pad (GstElement * element, GstPad * pad)
1037 {
1038   GstAggregator *self = GST_AGGREGATOR (element);
1039
1040   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1041
1042   GST_INFO_OBJECT (pad, "Removing pad");
1043
1044   SRC_STREAM_LOCK (self);
1045   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1046   gst_aggregator_pad_drop_buffer (aggpad);
1047   gst_element_remove_pad (element, pad);
1048
1049   SRC_STREAM_BROADCAST (self);
1050   SRC_STREAM_UNLOCK (self);
1051 }
1052
1053 static GstPad *
1054 gst_aggregator_request_new_pad (GstElement * element,
1055     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
1056 {
1057   GstAggregator *self;
1058   GstAggregatorPad *agg_pad;
1059
1060   GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
1061   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
1062
1063   self = GST_AGGREGATOR (element);
1064
1065   if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
1066     gint serial = 0;
1067     gchar *name = NULL;
1068
1069     GST_OBJECT_LOCK (element);
1070     if (req_name == NULL || strlen (req_name) < 6
1071         || !g_str_has_prefix (req_name, "sink_")) {
1072       /* no name given when requesting the pad, use next available int */
1073       priv->padcount++;
1074     } else {
1075       /* parse serial number from requested padname */
1076       serial = g_ascii_strtoull (&req_name[5], NULL, 10);
1077       if (serial >= priv->padcount)
1078         priv->padcount = serial;
1079     }
1080
1081     name = g_strdup_printf ("sink_%u", priv->padcount);
1082     agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
1083         "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
1084     g_free (name);
1085
1086     GST_OBJECT_UNLOCK (element);
1087
1088   } else {
1089     return NULL;
1090   }
1091
1092   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
1093
1094   if (priv->running)
1095     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
1096
1097   /* add the pad to the element */
1098   gst_element_add_pad (element, GST_PAD (agg_pad));
1099
1100   return GST_PAD (agg_pad);
1101 }
1102
1103 typedef struct
1104 {
1105   GstClockTime min, max;
1106   gboolean live;
1107 } LatencyData;
1108
1109 static gboolean
1110 gst_aggregator_query_sink_latency_foreach (GstAggregator * self,
1111     GstAggregatorPad * pad, gpointer user_data)
1112 {
1113   LatencyData *data = user_data;
1114   GstClockTime min, max;
1115   GstQuery *query;
1116   gboolean live, res;
1117
1118   query = gst_query_new_latency ();
1119   res = gst_pad_peer_query (GST_PAD_CAST (pad), query);
1120
1121   if (res) {
1122     gst_query_parse_latency (query, &live, &min, &max);
1123
1124     GST_LOG_OBJECT (pad, "got latency live:%s min:%" G_GINT64_FORMAT
1125         " max:%" G_GINT64_FORMAT, live ? "true" : "false", min, max);
1126
1127     if (live) {
1128       if (min > data->min)
1129         data->min = min;
1130
1131       if (data->max == GST_CLOCK_TIME_NONE)
1132         data->max = max;
1133       else if (max < data->max)
1134         data->max = max;
1135
1136       data->live = TRUE;
1137     }
1138   }
1139
1140   gst_query_unref (query);
1141
1142   return TRUE;
1143 }
1144
1145 /**
1146  * gst_aggregator_get_latency_unlocked:
1147  * @self: a #GstAggregator
1148  * @live: (out) (allow-none): whether @self is live
1149  * @min_latency: (out) (allow-none): the configured minimum latency of @self
1150  * @max_latency: (out) (allow-none): the configured maximum latency of @self
1151  *
1152  * Retreives the latency values reported by @self in response to the latency
1153  * query.
1154  *
1155  * Typically only called by subclasses.
1156  *
1157  * MUST be called with the object lock held.
1158  */
1159 void
1160 gst_aggregator_get_latency_unlocked (GstAggregator * self, gboolean * live,
1161     GstClockTime * min_latency, GstClockTime * max_latency)
1162 {
1163   GstClockTime min, max;
1164
1165   g_return_if_fail (GST_IS_AGGREGATOR (self));
1166
1167   /* latency_min is never GST_CLOCK_TIME_NONE by construction */
1168   min = self->priv->latency_min;
1169   max = self->priv->latency_max;
1170
1171   /* add our own */
1172   min += self->priv->latency;
1173   min += self->priv->sub_latency_min;
1174   if (GST_CLOCK_TIME_IS_VALID (max)
1175       && GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max))
1176     max += self->priv->sub_latency_max;
1177   else
1178     max = GST_CLOCK_TIME_NONE;
1179
1180   if (live)
1181     *live = self->priv->latency_live;
1182   if (min_latency)
1183     *min_latency = min;
1184   if (max_latency)
1185     *max_latency = max;
1186 }
1187
1188 static gboolean
1189 gst_aggregator_query_latency (GstAggregator * self, GstQuery * query)
1190 {
1191   GstClockTime our_latency;
1192   LatencyData data;
1193
1194   data.min = 0;
1195   data.max = GST_CLOCK_TIME_NONE;
1196   data.live = FALSE;
1197
1198   /* query upstream's latency */
1199   SRC_STREAM_LOCK (self);
1200   gst_aggregator_iterate_sinkpads (self,
1201       gst_aggregator_query_sink_latency_foreach, &data);
1202   SRC_STREAM_UNLOCK (self);
1203
1204   GST_OBJECT_LOCK (self);
1205   our_latency = self->priv->latency;
1206
1207   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (data.min))) {
1208     GST_WARNING_OBJECT (self, "Invalid minimum latency, using 0");
1209     data.min = 0;
1210   }
1211
1212   if (G_UNLIKELY (data.min > data.max)) {
1213     GST_WARNING_OBJECT (self, "Minimum latency is greater than maximum latency "
1214         "(%" G_GINT64_FORMAT " > %" G_GINT64_FORMAT "). "
1215         "Clamping it at the maximum latency", data.min, data.max);
1216     data.min = data.max;
1217   }
1218
1219   self->priv->latency_live = data.live;
1220   self->priv->latency_min = data.min;
1221   self->priv->latency_max = data.max;
1222
1223   /* add our own */
1224   data.min += our_latency;
1225   data.min += self->priv->sub_latency_min;
1226   if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1227       && GST_CLOCK_TIME_IS_VALID (data.max))
1228     data.max += self->priv->sub_latency_max;
1229   else
1230     data.max = GST_CLOCK_TIME_NONE;
1231
1232   if (data.live && data.min > data.max) {
1233     GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1234         ("%s", "Latency too big"),
1235         ("The requested latency value is too big for the current pipeline.  "
1236             "Limiting to %" G_GINT64_FORMAT, data.max));
1237     data.min = data.max;
1238     /* FIXME: This could in theory become negative, but in
1239      * that case all is lost anyway */
1240     self->priv->latency -= data.min - data.max;
1241     /* FIXME: shouldn't we g_object_notify() the change here? */
1242   }
1243
1244   GST_OBJECT_UNLOCK (self);
1245
1246   GST_DEBUG_OBJECT (self, "configured latency live:%s min:%" G_GINT64_FORMAT
1247       " max:%" G_GINT64_FORMAT, data.live ? "true" : "false", data.min,
1248       data.max);
1249
1250   gst_query_set_latency (query, data.live, data.min, data.max);
1251
1252   return TRUE;
1253 }
1254
1255 static gboolean
1256 gst_aggregator_send_event (GstElement * element, GstEvent * event)
1257 {
1258   GstAggregator *self = GST_AGGREGATOR (element);
1259
1260   GST_STATE_LOCK (element);
1261   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK &&
1262       GST_STATE (element) < GST_STATE_PAUSED) {
1263     gdouble rate;
1264     GstFormat fmt;
1265     GstSeekFlags flags;
1266     GstSeekType start_type, stop_type;
1267     gint64 start, stop;
1268
1269     gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1270         &start, &stop_type, &stop);
1271
1272     GST_OBJECT_LOCK (self);
1273     gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1274         stop_type, stop, NULL);
1275     self->priv->seqnum = gst_event_get_seqnum (event);
1276     GST_OBJECT_UNLOCK (self);
1277
1278     GST_DEBUG_OBJECT (element, "Storing segment %" GST_PTR_FORMAT, event);
1279   }
1280   GST_STATE_UNLOCK (element);
1281
1282
1283   return GST_ELEMENT_CLASS (aggregator_parent_class)->send_event (element,
1284       event);
1285 }
1286
1287 static gboolean
1288 gst_aggregator_default_src_query (GstAggregator * self, GstQuery * query)
1289 {
1290   gboolean res = TRUE;
1291
1292   switch (GST_QUERY_TYPE (query)) {
1293     case GST_QUERY_SEEKING:
1294     {
1295       GstFormat format;
1296
1297       /* don't pass it along as some (file)sink might claim it does
1298        * whereas with a collectpads in between that will not likely work */
1299       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
1300       gst_query_set_seeking (query, format, FALSE, 0, -1);
1301       res = TRUE;
1302
1303       goto discard;
1304     }
1305     case GST_QUERY_LATENCY:
1306     {
1307       gboolean ret;
1308
1309       ret = gst_aggregator_query_latency (self, query);
1310       /* Wake up the src thread again, due to changed latencies
1311        * or changed live-ness we might have to adjust if we wait
1312        * on a deadline at all and how long.
1313        * This is only to unschedule the clock id, we don't really care
1314        * about the GCond here.
1315        */
1316       SRC_STREAM_LOCK (self);
1317       SRC_STREAM_BROADCAST (self);
1318       SRC_STREAM_UNLOCK (self);
1319       return ret;
1320     }
1321     default:
1322       break;
1323   }
1324
1325   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
1326
1327 discard:
1328   return res;
1329 }
1330
1331 static gboolean
1332 gst_aggregator_event_forward_func (GstPad * pad, gpointer user_data)
1333 {
1334   EventData *evdata = user_data;
1335   gboolean ret = TRUE;
1336   GstPad *peer = gst_pad_get_peer (pad);
1337   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1338
1339   if (peer) {
1340     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
1341     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
1342     gst_object_unref (peer);
1343   }
1344
1345   if (ret == FALSE) {
1346     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
1347       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
1348
1349     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
1350       GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
1351
1352       if (gst_pad_query (peer, seeking)) {
1353         gboolean seekable;
1354
1355         gst_query_parse_seeking (seeking, NULL, &seekable, NULL, NULL);
1356
1357         if (seekable == FALSE) {
1358           GST_INFO_OBJECT (pad,
1359               "Source not seekable, We failed but it does not matter!");
1360
1361           ret = TRUE;
1362         }
1363       } else {
1364         GST_ERROR_OBJECT (pad, "Query seeking FAILED");
1365       }
1366
1367       gst_query_unref (seeking);
1368     }
1369
1370     if (evdata->flush) {
1371       PAD_LOCK (aggpad);
1372       aggpad->priv->pending_flush_start = FALSE;
1373       aggpad->priv->pending_flush_stop = FALSE;
1374       PAD_UNLOCK (aggpad);
1375     }
1376   } else {
1377     evdata->one_actually_seeked = TRUE;
1378   }
1379
1380   evdata->result &= ret;
1381
1382   /* Always send to all pads */
1383   return FALSE;
1384 }
1385
1386 static EventData
1387 gst_aggregator_forward_event_to_all_sinkpads (GstAggregator * self,
1388     GstEvent * event, gboolean flush)
1389 {
1390   EventData evdata;
1391
1392   evdata.event = event;
1393   evdata.result = TRUE;
1394   evdata.flush = flush;
1395   evdata.one_actually_seeked = FALSE;
1396
1397   /* We first need to set all pads as flushing in a first pass
1398    * as flush_start flush_stop is sometimes sent synchronously
1399    * while we send the seek event */
1400   if (flush) {
1401     GList *l;
1402
1403     GST_OBJECT_LOCK (self);
1404     for (l = GST_ELEMENT_CAST (self)->sinkpads; l != NULL; l = l->next) {
1405       GstAggregatorPad *pad = l->data;
1406
1407       PAD_LOCK (pad);
1408       pad->priv->pending_flush_start = TRUE;
1409       pad->priv->pending_flush_stop = FALSE;
1410       PAD_UNLOCK (pad);
1411     }
1412     GST_OBJECT_UNLOCK (self);
1413   }
1414
1415   gst_pad_forward (self->srcpad, gst_aggregator_event_forward_func, &evdata);
1416
1417   gst_event_unref (event);
1418
1419   return evdata;
1420 }
1421
1422 static gboolean
1423 gst_aggregator_do_seek (GstAggregator * self, GstEvent * event)
1424 {
1425   gdouble rate;
1426   GstFormat fmt;
1427   GstSeekFlags flags;
1428   GstSeekType start_type, stop_type;
1429   gint64 start, stop;
1430   gboolean flush;
1431   EventData evdata;
1432   GstAggregatorPrivate *priv = self->priv;
1433
1434   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1435       &start, &stop_type, &stop);
1436
1437   GST_INFO_OBJECT (self, "starting SEEK");
1438
1439   flush = flags & GST_SEEK_FLAG_FLUSH;
1440
1441   GST_OBJECT_LOCK (self);
1442   if (flush) {
1443     priv->pending_flush_start = TRUE;
1444     priv->flush_seeking = TRUE;
1445   }
1446
1447   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1448       stop_type, stop, NULL);
1449   GST_OBJECT_UNLOCK (self);
1450
1451   /* forward the seek upstream */
1452   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, flush);
1453   event = NULL;
1454
1455   if (!evdata.result || !evdata.one_actually_seeked) {
1456     GST_OBJECT_LOCK (self);
1457     priv->flush_seeking = FALSE;
1458     priv->pending_flush_start = FALSE;
1459     GST_OBJECT_UNLOCK (self);
1460   }
1461
1462   GST_INFO_OBJECT (self, "seek done, result: %d", evdata.result);
1463
1464   return evdata.result;
1465 }
1466
1467 static gboolean
1468 gst_aggregator_default_src_event (GstAggregator * self, GstEvent * event)
1469 {
1470   EventData evdata;
1471   gboolean res = TRUE;
1472
1473   switch (GST_EVENT_TYPE (event)) {
1474     case GST_EVENT_SEEK:
1475     {
1476       gst_event_ref (event);
1477       res = gst_aggregator_do_seek (self, event);
1478       gst_event_unref (event);
1479       event = NULL;
1480       goto done;
1481     }
1482     case GST_EVENT_NAVIGATION:
1483     {
1484       /* navigation is rather pointless. */
1485       res = FALSE;
1486       gst_event_unref (event);
1487       goto done;
1488     }
1489     default:
1490     {
1491       break;
1492     }
1493   }
1494
1495   evdata = gst_aggregator_forward_event_to_all_sinkpads (self, event, FALSE);
1496   res = evdata.result;
1497
1498 done:
1499   return res;
1500 }
1501
1502 static gboolean
1503 gst_aggregator_src_pad_event_func (GstPad * pad, GstObject * parent,
1504     GstEvent * event)
1505 {
1506   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1507
1508   return klass->src_event (GST_AGGREGATOR (parent), event);
1509 }
1510
1511 static gboolean
1512 gst_aggregator_src_pad_query_func (GstPad * pad, GstObject * parent,
1513     GstQuery * query)
1514 {
1515   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1516
1517   return klass->src_query (GST_AGGREGATOR (parent), query);
1518 }
1519
1520 static gboolean
1521 gst_aggregator_src_pad_activate_mode_func (GstPad * pad,
1522     GstObject * parent, GstPadMode mode, gboolean active)
1523 {
1524   GstAggregator *self = GST_AGGREGATOR (parent);
1525   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1526
1527   if (klass->src_activate) {
1528     if (klass->src_activate (self, mode, active) == FALSE) {
1529       return FALSE;
1530     }
1531   }
1532
1533   if (active == TRUE) {
1534     switch (mode) {
1535       case GST_PAD_MODE_PUSH:
1536       {
1537         GST_INFO_OBJECT (pad, "Activating pad!");
1538         gst_aggregator_start_srcpad_task (self);
1539         return TRUE;
1540       }
1541       default:
1542       {
1543         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
1544         return FALSE;
1545       }
1546     }
1547   }
1548
1549   /* deactivating */
1550   GST_INFO_OBJECT (self, "Deactivating srcpad");
1551   gst_aggregator_stop_srcpad_task (self, FALSE);
1552
1553   return TRUE;
1554 }
1555
1556 static gboolean
1557 gst_aggregator_default_sink_query (GstAggregator * self,
1558     GstAggregatorPad * aggpad, GstQuery * query)
1559 {
1560   GstPad *pad = GST_PAD (aggpad);
1561
1562   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1563 }
1564
1565 static void
1566 gst_aggregator_finalize (GObject * object)
1567 {
1568   GstAggregator *self = (GstAggregator *) object;
1569
1570   g_mutex_clear (&self->priv->src_lock);
1571   g_cond_clear (&self->priv->src_cond);
1572
1573   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
1574 }
1575
1576 /*
1577  * gst_aggregator_set_latency_property:
1578  * @agg: a #GstAggregator
1579  * @latency: the new latency value.
1580  *
1581  * Sets the new latency value to @latency. This value is used to limit the
1582  * amount of time a pad waits for data to appear before considering the pad
1583  * as unresponsive.
1584  */
1585 static void
1586 gst_aggregator_set_latency_property (GstAggregator * self, gint64 latency)
1587 {
1588   gboolean changed;
1589   GstClockTime min, max;
1590
1591   g_return_if_fail (GST_IS_AGGREGATOR (self));
1592   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (latency));
1593
1594   GST_OBJECT_LOCK (self);
1595   if (self->priv->latency_live) {
1596     min = self->priv->latency_min;
1597     max = self->priv->latency_max;
1598     /* add our own */
1599     min += latency;
1600     min += self->priv->sub_latency_min;
1601     if (GST_CLOCK_TIME_IS_VALID (self->priv->sub_latency_max)
1602         && GST_CLOCK_TIME_IS_VALID (max))
1603       max += self->priv->sub_latency_max;
1604     else
1605       max = GST_CLOCK_TIME_NONE;
1606
1607     if (GST_CLOCK_TIME_IS_VALID (max) && min > max) {
1608       GST_ELEMENT_WARNING (self, CORE, NEGOTIATION,
1609           ("%s", "Latency too big"),
1610           ("The requested latency value is too big for the latency in the "
1611               "current pipeline.  Limiting to %" G_GINT64_FORMAT, max));
1612       /* FIXME: This could in theory become negative, but in
1613        * that case all is lost anyway */
1614       latency -= min - max;
1615       /* FIXME: shouldn't we g_object_notify() the change here? */
1616     }
1617   }
1618
1619   changed = (self->priv->latency != latency);
1620   self->priv->latency = latency;
1621   GST_OBJECT_UNLOCK (self);
1622
1623   if (changed)
1624     gst_element_post_message (GST_ELEMENT_CAST (self),
1625         gst_message_new_latency (GST_OBJECT_CAST (self)));
1626 }
1627
1628 /*
1629  * gst_aggregator_get_latency_property:
1630  * @agg: a #GstAggregator
1631  *
1632  * Gets the latency value. See gst_aggregator_set_latency for
1633  * more details.
1634  *
1635  * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
1636  * before a pad is deemed unresponsive. A value of -1 means an
1637  * unlimited time.
1638  */
1639 static gint64
1640 gst_aggregator_get_latency_property (GstAggregator * agg)
1641 {
1642   gint64 res;
1643
1644   g_return_val_if_fail (GST_IS_AGGREGATOR (agg), -1);
1645
1646   GST_OBJECT_LOCK (agg);
1647   res = agg->priv->latency;
1648   GST_OBJECT_UNLOCK (agg);
1649
1650   return res;
1651 }
1652
1653 static void
1654 gst_aggregator_set_property (GObject * object, guint prop_id,
1655     const GValue * value, GParamSpec * pspec)
1656 {
1657   GstAggregator *agg = GST_AGGREGATOR (object);
1658
1659   switch (prop_id) {
1660     case PROP_LATENCY:
1661       gst_aggregator_set_latency_property (agg, g_value_get_int64 (value));
1662       break;
1663     default:
1664       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1665       break;
1666   }
1667 }
1668
1669 static void
1670 gst_aggregator_get_property (GObject * object, guint prop_id,
1671     GValue * value, GParamSpec * pspec)
1672 {
1673   GstAggregator *agg = GST_AGGREGATOR (object);
1674
1675   switch (prop_id) {
1676     case PROP_LATENCY:
1677       g_value_set_int64 (value, gst_aggregator_get_latency_property (agg));
1678       break;
1679     default:
1680       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1681       break;
1682   }
1683 }
1684
1685 /* GObject vmethods implementations */
1686 static void
1687 gst_aggregator_class_init (GstAggregatorClass * klass)
1688 {
1689   GObjectClass *gobject_class = (GObjectClass *) klass;
1690   GstElementClass *gstelement_class = (GstElementClass *) klass;
1691
1692   aggregator_parent_class = g_type_class_peek_parent (klass);
1693   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1694
1695   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1696       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1697
1698   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1699
1700   klass->sink_event = gst_aggregator_default_sink_event;
1701   klass->sink_query = gst_aggregator_default_sink_query;
1702
1703   klass->src_event = gst_aggregator_default_src_event;
1704   klass->src_query = gst_aggregator_default_src_query;
1705
1706   gstelement_class->request_new_pad =
1707       GST_DEBUG_FUNCPTR (gst_aggregator_request_new_pad);
1708   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_aggregator_send_event);
1709   gstelement_class->release_pad =
1710       GST_DEBUG_FUNCPTR (gst_aggregator_release_pad);
1711   gstelement_class->change_state =
1712       GST_DEBUG_FUNCPTR (gst_aggregator_change_state);
1713
1714   gobject_class->set_property = gst_aggregator_set_property;
1715   gobject_class->get_property = gst_aggregator_get_property;
1716   gobject_class->finalize = gst_aggregator_finalize;
1717
1718   g_object_class_install_property (gobject_class, PROP_LATENCY,
1719       g_param_spec_int64 ("latency", "Buffer latency",
1720           "Additional latency in live mode to allow upstream "
1721           "to take longer to produce buffers for the current "
1722           "position", 0,
1723           (G_MAXLONG == G_MAXINT64) ? G_MAXINT64 : (G_MAXLONG * GST_SECOND - 1),
1724           DEFAULT_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1725
1726   GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_stop_pad);
1727   GST_DEBUG_REGISTER_FUNCPTR (gst_aggregator_query_sink_latency_foreach);
1728 }
1729
1730 static void
1731 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1732 {
1733   GstPadTemplate *pad_template;
1734   GstAggregatorPrivate *priv;
1735
1736   g_return_if_fail (klass->aggregate != NULL);
1737
1738   self->priv =
1739       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1740       GstAggregatorPrivate);
1741
1742   priv = self->priv;
1743
1744   pad_template =
1745       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1746   g_return_if_fail (pad_template != NULL);
1747
1748   priv->padcount = -1;
1749   priv->tags_changed = FALSE;
1750
1751   self->priv->latency_live = FALSE;
1752   self->priv->latency_min = self->priv->sub_latency_min = 0;
1753   self->priv->latency_max = self->priv->sub_latency_max = 0;
1754   gst_aggregator_reset_flow_values (self);
1755
1756   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1757
1758   gst_pad_set_event_function (self->srcpad,
1759       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_event_func));
1760   gst_pad_set_query_function (self->srcpad,
1761       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_query_func));
1762   gst_pad_set_activatemode_function (self->srcpad,
1763       GST_DEBUG_FUNCPTR (gst_aggregator_src_pad_activate_mode_func));
1764
1765   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1766
1767   self->priv->latency = DEFAULT_LATENCY;
1768
1769   g_mutex_init (&self->priv->src_lock);
1770   g_cond_init (&self->priv->src_cond);
1771 }
1772
1773 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1774  * method to get to the padtemplates */
1775 GType
1776 gst_aggregator_get_type (void)
1777 {
1778   static volatile gsize type = 0;
1779
1780   if (g_once_init_enter (&type)) {
1781     GType _type;
1782     static const GTypeInfo info = {
1783       sizeof (GstAggregatorClass),
1784       NULL,
1785       NULL,
1786       (GClassInitFunc) gst_aggregator_class_init,
1787       NULL,
1788       NULL,
1789       sizeof (GstAggregator),
1790       0,
1791       (GInstanceInitFunc) gst_aggregator_init,
1792     };
1793
1794     _type = g_type_register_static (GST_TYPE_ELEMENT,
1795         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1796     g_once_init_leave (&type, _type);
1797   }
1798   return type;
1799 }
1800
1801 static GstFlowReturn
1802 gst_aggregator_pad_chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1803 {
1804   GstBuffer *actual_buf = buffer;
1805   GstAggregator *self = GST_AGGREGATOR (object);
1806   GstAggregatorPrivate *priv = self->priv;
1807   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1808   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1809   GstFlowReturn flow_return;
1810
1811   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1812
1813   PAD_FLUSH_LOCK (aggpad);
1814
1815   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1816     goto flushing;
1817
1818   PAD_LOCK (aggpad);
1819   if (aggpad->priv->pending_eos == TRUE)
1820     goto eos;
1821
1822   while (aggpad->priv->buffer
1823       && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1824     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1825     PAD_WAIT_EVENT (aggpad);
1826   }
1827   PAD_UNLOCK (aggpad);
1828
1829   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1830     goto flushing;
1831
1832   if (aggclass->clip) {
1833     aggclass->clip (self, aggpad, buffer, &actual_buf);
1834   }
1835
1836   SRC_STREAM_LOCK (self);
1837   PAD_LOCK (aggpad);
1838   if (aggpad->priv->buffer)
1839     gst_buffer_unref (aggpad->priv->buffer);
1840   aggpad->priv->buffer = actual_buf;
1841   PAD_UNLOCK (aggpad);
1842   PAD_FLUSH_UNLOCK (aggpad);
1843
1844   SRC_STREAM_BROADCAST (self);
1845   SRC_STREAM_UNLOCK (self);
1846
1847   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1848
1849   GST_OBJECT_LOCK (self);
1850   flow_return = priv->flow_return;
1851   GST_OBJECT_UNLOCK (self);
1852
1853   return flow_return;
1854
1855 flushing:
1856   PAD_FLUSH_UNLOCK (aggpad);
1857
1858   gst_buffer_unref (buffer);
1859   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1860
1861   return GST_FLOW_FLUSHING;
1862
1863 eos:
1864   PAD_UNLOCK (aggpad);
1865   PAD_FLUSH_UNLOCK (aggpad);
1866
1867   gst_buffer_unref (buffer);
1868   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1869
1870   return GST_FLOW_EOS;
1871 }
1872
1873 static gboolean
1874 gst_aggregator_pad_query_func (GstPad * pad, GstObject * parent,
1875     GstQuery * query)
1876 {
1877   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1878   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1879
1880   if (GST_QUERY_IS_SERIALIZED (query)) {
1881     PAD_LOCK (aggpad);
1882
1883     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE) {
1884       PAD_UNLOCK (aggpad);
1885       goto flushing;
1886     }
1887
1888     while (aggpad->priv->buffer
1889         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1890       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1891       PAD_WAIT_EVENT (aggpad);
1892     }
1893     PAD_UNLOCK (aggpad);
1894
1895     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1896       goto flushing;
1897   }
1898
1899   return klass->sink_query (GST_AGGREGATOR (parent),
1900       GST_AGGREGATOR_PAD (pad), query);
1901
1902 flushing:
1903   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping query");
1904   return FALSE;
1905 }
1906
1907 static gboolean
1908 gst_aggregator_pad_event_func (GstPad * pad, GstObject * parent,
1909     GstEvent * event)
1910 {
1911   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1912   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1913
1914   if (GST_EVENT_IS_SERIALIZED (event) && GST_EVENT_TYPE (event) != GST_EVENT_EOS
1915       && GST_EVENT_TYPE (event) != GST_EVENT_SEGMENT_DONE) {
1916     PAD_LOCK (aggpad);
1917
1918     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1919         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP) {
1920       PAD_UNLOCK (aggpad);
1921       goto flushing;
1922     }
1923
1924     while (aggpad->priv->buffer
1925         && g_atomic_int_get (&aggpad->priv->flushing) == FALSE) {
1926       GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1927       PAD_WAIT_EVENT (aggpad);
1928     }
1929     PAD_UNLOCK (aggpad);
1930
1931     if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE
1932         && GST_EVENT_TYPE (event) != GST_EVENT_FLUSH_STOP)
1933       goto flushing;
1934   }
1935
1936   return klass->sink_event (GST_AGGREGATOR (parent),
1937       GST_AGGREGATOR_PAD (pad), event);
1938
1939 flushing:
1940   GST_DEBUG_OBJECT (aggpad, "Pad is flushing, dropping event");
1941   if (GST_EVENT_IS_STICKY (event))
1942     gst_pad_store_sticky_event (pad, event);
1943   gst_event_unref (event);
1944   return FALSE;
1945 }
1946
1947 static gboolean
1948 gst_aggregator_pad_activate_mode_func (GstPad * pad,
1949     GstObject * parent, GstPadMode mode, gboolean active)
1950 {
1951   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1952
1953   if (active == FALSE) {
1954     PAD_LOCK (aggpad);
1955     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1956     gst_buffer_replace (&aggpad->priv->buffer, NULL);
1957     PAD_BROADCAST_EVENT (aggpad);
1958     PAD_UNLOCK (aggpad);
1959   } else {
1960     PAD_LOCK (aggpad);
1961     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1962     PAD_BROADCAST_EVENT (aggpad);
1963     PAD_UNLOCK (aggpad);
1964   }
1965
1966   return TRUE;
1967 }
1968
1969 /***********************************
1970  * GstAggregatorPad implementation  *
1971  ************************************/
1972 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1973
1974 static void
1975 gst_aggregator_pad_constructed (GObject * object)
1976 {
1977   GstPad *pad = GST_PAD (object);
1978
1979   gst_pad_set_chain_function (pad,
1980       GST_DEBUG_FUNCPTR (gst_aggregator_pad_chain));
1981   gst_pad_set_event_function (pad,
1982       GST_DEBUG_FUNCPTR (gst_aggregator_pad_event_func));
1983   gst_pad_set_query_function (pad,
1984       GST_DEBUG_FUNCPTR (gst_aggregator_pad_query_func));
1985   gst_pad_set_activatemode_function (pad,
1986       GST_DEBUG_FUNCPTR (gst_aggregator_pad_activate_mode_func));
1987 }
1988
1989 static void
1990 gst_aggregator_pad_finalize (GObject * object)
1991 {
1992   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1993
1994   g_cond_clear (&pad->priv->event_cond);
1995   g_mutex_clear (&pad->priv->flush_lock);
1996   g_mutex_clear (&pad->priv->lock);
1997
1998   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->finalize (object);
1999 }
2000
2001 static void
2002 gst_aggregator_pad_dispose (GObject * object)
2003 {
2004   GstAggregatorPad *pad = (GstAggregatorPad *) object;
2005
2006   gst_aggregator_pad_drop_buffer (pad);
2007
2008   G_OBJECT_CLASS (gst_aggregator_pad_parent_class)->dispose (object);
2009 }
2010
2011 static void
2012 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
2013 {
2014   GObjectClass *gobject_class = (GObjectClass *) klass;
2015
2016   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
2017
2018   gobject_class->constructed = gst_aggregator_pad_constructed;
2019   gobject_class->finalize = gst_aggregator_pad_finalize;
2020   gobject_class->dispose = gst_aggregator_pad_dispose;
2021 }
2022
2023 static void
2024 gst_aggregator_pad_init (GstAggregatorPad * pad)
2025 {
2026   pad->priv =
2027       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
2028       GstAggregatorPadPrivate);
2029
2030   pad->priv->buffer = NULL;
2031   g_cond_init (&pad->priv->event_cond);
2032
2033   g_mutex_init (&pad->priv->flush_lock);
2034   g_mutex_init (&pad->priv->lock);
2035 }
2036
2037 /**
2038  * gst_aggregator_pad_steal_buffer:
2039  * @pad: the pad to get buffer from
2040  *
2041  * Steal the ref to the buffer currently queued in @pad.
2042  *
2043  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
2044  *   queued. You should unref the buffer after usage.
2045  */
2046 GstBuffer *
2047 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
2048 {
2049   GstBuffer *buffer = NULL;
2050
2051   PAD_LOCK (pad);
2052   if (pad->priv->buffer) {
2053     GST_TRACE_OBJECT (pad, "Consuming buffer");
2054     buffer = pad->priv->buffer;
2055     pad->priv->buffer = NULL;
2056     if (pad->priv->pending_eos) {
2057       pad->priv->pending_eos = FALSE;
2058       pad->priv->eos = TRUE;
2059     }
2060     PAD_BROADCAST_EVENT (pad);
2061     GST_DEBUG_OBJECT (pad, "Consumed: %" GST_PTR_FORMAT, buffer);
2062   }
2063   PAD_UNLOCK (pad);
2064
2065   return buffer;
2066 }
2067
2068 /**
2069  * gst_aggregator_pad_drop_buffer:
2070  * @pad: the pad where to drop any pending buffer
2071  *
2072  * Drop the buffer currently queued in @pad.
2073  *
2074  * Returns: TRUE if there was a buffer queued in @pad, or FALSE if not.
2075  */
2076 gboolean
2077 gst_aggregator_pad_drop_buffer (GstAggregatorPad * pad)
2078 {
2079   GstBuffer *buf;
2080
2081   buf = gst_aggregator_pad_steal_buffer (pad);
2082
2083   if (buf == NULL)
2084     return FALSE;
2085
2086   gst_buffer_unref (buf);
2087   return TRUE;
2088 }
2089
2090 /**
2091  * gst_aggregator_pad_get_buffer:
2092  * @pad: the pad to get buffer from
2093  *
2094  * Returns: (transfer full): A reference to the buffer in @pad or
2095  * NULL if no buffer was queued. You should unref the buffer after
2096  * usage.
2097  */
2098 GstBuffer *
2099 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
2100 {
2101   GstBuffer *buffer = NULL;
2102
2103   PAD_LOCK (pad);
2104   if (pad->priv->buffer)
2105     buffer = gst_buffer_ref (pad->priv->buffer);
2106   PAD_UNLOCK (pad);
2107
2108   return buffer;
2109 }
2110
2111 gboolean
2112 gst_aggregator_pad_is_eos (GstAggregatorPad * pad)
2113 {
2114   gboolean is_eos;
2115
2116   PAD_LOCK (pad);
2117   is_eos = pad->priv->eos;
2118   PAD_UNLOCK (pad);
2119
2120   return is_eos;
2121 }
2122
2123 /**
2124  * gst_aggregator_merge_tags:
2125  * @self: a #GstAggregator
2126  * @tags: a #GstTagList to merge
2127  * @mode: the #GstTagMergeMode to use
2128  *
2129  * Adds tags to so-called pending tags, which will be processed
2130  * before pushing out data downstream.
2131  *
2132  * Note that this is provided for convenience, and the subclass is
2133  * not required to use this and can still do tag handling on its own.
2134  *
2135  * MT safe.
2136  */
2137 void
2138 gst_aggregator_merge_tags (GstAggregator * self,
2139     const GstTagList * tags, GstTagMergeMode mode)
2140 {
2141   GstTagList *otags;
2142
2143   g_return_if_fail (GST_IS_AGGREGATOR (self));
2144   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
2145
2146   /* FIXME Check if we can use OBJECT lock here! */
2147   GST_OBJECT_LOCK (self);
2148   if (tags)
2149     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
2150   otags = self->priv->tags;
2151   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
2152   if (otags)
2153     gst_tag_list_unref (otags);
2154   self->priv->tags_changed = TRUE;
2155   GST_OBJECT_UNLOCK (self);
2156 }
2157
2158 /**
2159  * gst_aggregator_set_latency:
2160  * @self: a #GstAggregator
2161  * @min_latency: minimum latency
2162  * @max_latency: maximum latency
2163  *
2164  * Lets #GstAggregator sub-classes tell the baseclass what their internal
2165  * latency is. Will also post a LATENCY message on the bus so the pipeline
2166  * can reconfigure its global latency.
2167  */
2168 void
2169 gst_aggregator_set_latency (GstAggregator * self,
2170     GstClockTime min_latency, GstClockTime max_latency)
2171 {
2172   gboolean changed = FALSE;
2173
2174   g_return_if_fail (GST_IS_AGGREGATOR (self));
2175   g_return_if_fail (GST_CLOCK_TIME_IS_VALID (min_latency));
2176   g_return_if_fail (max_latency >= min_latency);
2177
2178   GST_OBJECT_LOCK (self);
2179   if (self->priv->sub_latency_min != min_latency) {
2180     self->priv->sub_latency_min = min_latency;
2181     changed = TRUE;
2182   }
2183   if (self->priv->sub_latency_max != max_latency) {
2184     self->priv->sub_latency_max = max_latency;
2185     changed = TRUE;
2186   }
2187   GST_OBJECT_UNLOCK (self);
2188
2189   if (changed) {
2190     gst_element_post_message (GST_ELEMENT_CAST (self),
2191         gst_message_new_latency (GST_OBJECT_CAST (self)));
2192   }
2193 }