libs/gst/base/gstaggregator.c

   1 /* GStreamer
   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. Implementers should at least implement
  33  *    the aggregate () vmethod.
  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_timeout (GstAggregator * agg, gint64 timeout);
  73 static gint64 gst_aggregator_get_timeout (GstAggregator * agg);
  74
  75
  76 GST_DEBUG_CATEGORY_STATIC (aggregator_debug);
  77 #define GST_CAT_DEFAULT aggregator_debug
  78
  79 /* GstAggregatorPad definitions */
  80 #define PAD_LOCK_EVENT(pad)   G_STMT_START {                            \
  81   GST_LOG_OBJECT (pad, "Taking EVENT lock from thread %p",              \
  82         g_thread_self());                                               \
  83   g_mutex_lock(&pad->priv->event_lock);                                 \
  84   GST_LOG_OBJECT (pad, "Took EVENT lock from thread %p",              \
  85         g_thread_self());                                               \
  86   } G_STMT_END
  87
  88 #define PAD_UNLOCK_EVENT(pad)  G_STMT_START {                           \
  89   GST_LOG_OBJECT (pad, "Releasing EVENT lock from thread %p",          \
  90         g_thread_self());                                               \
  91   g_mutex_unlock(&pad->priv->event_lock);                               \
  92   GST_LOG_OBJECT (pad, "Release EVENT lock from thread %p",          \
  93         g_thread_self());                                               \
  94   } G_STMT_END
  95
  96
  97 #define PAD_WAIT_EVENT(pad)   G_STMT_START {                            \
  98   GST_LOG_OBJECT (pad, "Waiting for EVENT on thread %p",               \
  99         g_thread_self());                                               \
 100   g_cond_wait(&(((GstAggregatorPad* )pad)->priv->event_cond),       \
 101       &(pad->priv->event_lock));                                        \
 102   GST_LOG_OBJECT (pad, "DONE Waiting for EVENT on thread %p",               \
 103         g_thread_self());                                               \
 104   } G_STMT_END
 105
 106 #define PAD_BROADCAST_EVENT(pad) {                                          \
 107   GST_LOG_OBJECT (pad, "Signaling EVENT from thread %p",               \
 108         g_thread_self());                                                   \
 109   g_cond_broadcast(&(((GstAggregatorPad* )pad)->priv->event_cond)); \
 110   }
 111
 112 #define GST_AGGREGATOR_SETCAPS_LOCK(self)   G_STMT_START {        \
 113   GST_LOG_OBJECT (self, "Taking SETCAPS lock from thread %p",   \
 114         g_thread_self());                                         \
 115   g_mutex_lock(&self->priv->setcaps_lock);                         \
 116   GST_LOG_OBJECT (self, "Took SETCAPS lock from thread %p",     \
 117         g_thread_self());                                         \
 118   } G_STMT_END
 119
 120 #define GST_AGGREGATOR_SETCAPS_UNLOCK(self)   G_STMT_START {        \
 121   GST_LOG_OBJECT (self, "Releasing SETCAPS lock from thread %p",  \
 122         g_thread_self());                                           \
 123   g_mutex_unlock(&self->priv->setcaps_lock);                         \
 124   GST_LOG_OBJECT (self, "Took SETCAPS lock from thread %p",       \
 125         g_thread_self());                                           \
 126   } G_STMT_END
 127
 128 #define PAD_STREAM_LOCK(pad)   G_STMT_START {                            \
 129   GST_LOG_OBJECT (pad, "Taking lock from thread %p",              \
 130         g_thread_self());                                               \
 131   g_mutex_lock(&pad->priv->stream_lock);                                 \
 132   GST_LOG_OBJECT (pad, "Took lock from thread %p",              \
 133         g_thread_self());                                               \
 134   } G_STMT_END
 135
 136 #define PAD_STREAM_UNLOCK(pad)  G_STMT_START {                           \
 137   GST_LOG_OBJECT (pad, "Releasing lock from thread %p",          \
 138         g_thread_self());                                               \
 139   g_mutex_unlock(&pad->priv->stream_lock);                               \
 140   GST_LOG_OBJECT (pad, "Release lock from thread %p",          \
 141         g_thread_self());                                               \
 142   } G_STMT_END
 143
 144 struct _GstAggregatorPadPrivate
 145 {
 146   gboolean pending_flush_start;
 147   gboolean pending_flush_stop;
 148   gboolean pending_eos;
 149   gboolean flushing;
 150
 151   GstClockID timeout_id;
 152
 153   GMutex event_lock;
 154   GCond event_cond;
 155
 156   GMutex stream_lock;
 157 };
 158
 159 static gboolean
 160 _aggpad_flush (GstAggregatorPad * aggpad, GstAggregator * agg)
 161 {
 162   GstAggregatorPadClass *klass = GST_AGGREGATOR_PAD_GET_CLASS (aggpad);
 163
 164   aggpad->eos = FALSE;
 165   aggpad->priv->flushing = FALSE;
 166
 167   if (klass->flush)
 168     return klass->flush (aggpad, agg);
 169
 170   return TRUE;
 171 }
 172
 173 /*************************************
 174  * GstAggregator implementation  *
 175  *************************************/
 176 static GstElementClass *aggregator_parent_class = NULL;
 177
 178 #define AGGREGATOR_QUEUE(self) (((GstAggregator*)self)->priv->queue)
 179
 180 #define QUEUE_PUSH(self) G_STMT_START {                              \
 181   GST_LOG_OBJECT (self, "Pushing to QUEUE in thread %p",             \
 182       g_thread_self());                                              \
 183   g_async_queue_push (AGGREGATOR_QUEUE (self), GINT_TO_POINTER (1)); \
 184 } G_STMT_END
 185
 186 #define QUEUE_POP(self) G_STMT_START {                               \
 187   GST_LOG_OBJECT (self, "Waiting on QUEUE in thread %p",             \
 188         g_thread_self());                                            \
 189   g_async_queue_pop (AGGREGATOR_QUEUE (self));                       \
 190   GST_LOG_OBJECT (self, "Waited on QUEUE in thread %p",              \
 191         g_thread_self());                                            \
 192 } G_STMT_END
 193
 194 #define QUEUE_FLUSH(self) G_STMT_START {                             \
 195   GST_LOG_OBJECT (self, "Flushing QUEUE in thread %p",               \
 196       g_thread_self());                                              \
 197   g_async_queue_lock (AGGREGATOR_QUEUE (self));                      \
 198   while (g_async_queue_try_pop_unlocked (AGGREGATOR_QUEUE (self)));  \
 199   g_async_queue_unlock (AGGREGATOR_QUEUE (self));                    \
 200   GST_LOG_OBJECT (self, "Flushed QUEUE in thread %p",                \
 201       g_thread_self());                                              \
 202 } G_STMT_END
 203
 204 struct _GstAggregatorPrivate
 205 {
 206   gint padcount;
 207
 208   GAsyncQueue *queue;
 209
 210   /* Our state is >= PAUSED */
 211   gboolean running;
 212
 213
 214   gint seqnum;
 215   gboolean send_stream_start;
 216   gboolean send_segment;
 217   gboolean flush_seeking;
 218   gboolean pending_flush_start;
 219   gboolean send_eos;
 220   GstFlowReturn flow_return;
 221
 222   GstCaps *srccaps;
 223
 224   GstTagList *tags;
 225   gboolean tags_changed;
 226
 227   /* Lock to prevent two src setcaps from happening at the same time  */
 228   GMutex setcaps_lock;
 229 };
 230
 231 typedef struct
 232 {
 233   GstEvent *event;
 234   gboolean result;
 235   gboolean flush;
 236
 237   gboolean one_actually_seeked;
 238 } EventData;
 239
 240 #define DEFAULT_TIMEOUT        -1
 241
 242 enum
 243 {
 244   PROP_0,
 245   PROP_TIMEOUT,
 246   PROP_LAST
 247 };
 248
 249 /**
 250  * gst_aggregator_iterate_sinkpads:
 251  * @self: The #GstAggregator
 252  * @func: The function to call.
 253  * @user_data: The data to pass to @func.
 254  *
 255  * Iterate the sinkpads of aggregator to call a function on them.
 256  *
 257  * This method guarantees that @func will be called only once for each
 258  * sink pad.
 259  */
 260 gboolean
 261 gst_aggregator_iterate_sinkpads (GstAggregator * self,
 262     GstAggregatorPadForeachFunc func, gpointer user_data)
 263 {
 264   gboolean result = FALSE;
 265   GstIterator *iter;
 266   gboolean done = FALSE;
 267   GValue item = { 0, };
 268   GList *seen_pads = NULL;
 269
 270   iter = gst_element_iterate_sink_pads (GST_ELEMENT (self));
 271
 272   if (!iter)
 273     goto no_iter;
 274
 275   while (!done) {
 276     switch (gst_iterator_next (iter, &item)) {
 277       case GST_ITERATOR_OK:
 278       {
 279         GstPad *pad;
 280
 281         pad = g_value_get_object (&item);
 282
 283         /* if already pushed, skip. FIXME, find something faster to tag pads */
 284         if (pad == NULL || g_list_find (seen_pads, pad)) {
 285           g_value_reset (&item);
 286           break;
 287         }
 288
 289         GST_LOG_OBJECT (self, "calling function on pad %s:%s",
 290             GST_DEBUG_PAD_NAME (pad));
 291         result = func (self, pad, user_data);
 292
 293         done = !result;
 294
 295         seen_pads = g_list_prepend (seen_pads, pad);
 296
 297         g_value_reset (&item);
 298         break;
 299       }
 300       case GST_ITERATOR_RESYNC:
 301         gst_iterator_resync (iter);
 302         break;
 303       case GST_ITERATOR_ERROR:
 304         GST_ERROR_OBJECT (self,
 305             "Could not iterate over internally linked pads");
 306         done = TRUE;
 307         break;
 308       case GST_ITERATOR_DONE:
 309         done = TRUE;
 310         break;
 311     }
 312   }
 313   g_value_unset (&item);
 314   gst_iterator_free (iter);
 315
 316   if (seen_pads == NULL) {
 317     GST_DEBUG_OBJECT (self, "No pad seen");
 318     return FALSE;
 319   }
 320
 321   g_list_free (seen_pads);
 322
 323 no_iter:
 324   return result;
 325 }
 326
 327 static inline gboolean
 328 _check_all_pads_with_data_or_eos_or_timeout (GstAggregator * self,
 329     GstAggregatorPad * aggpad)
 330 {
 331   if (aggpad->buffer || aggpad->eos) {
 332     return TRUE;
 333   }
 334
 335   if (g_atomic_int_get (&aggpad->unresponsive) == TRUE) {
 336     /* pad has been deemed unresponsive */
 337     return TRUE;
 338   }
 339
 340   GST_LOG_OBJECT (aggpad, "Not ready to be aggregated");
 341
 342   return FALSE;
 343 }
 344
 345 static void
 346 _reset_flow_values (GstAggregator * self)
 347 {
 348   self->priv->flow_return = GST_FLOW_FLUSHING;
 349   self->priv->send_stream_start = TRUE;
 350   self->priv->send_segment = TRUE;
 351   gst_segment_init (&self->segment, GST_FORMAT_TIME);
 352 }
 353
 354 static inline void
 355 _push_mandatory_events (GstAggregator * self)
 356 {
 357   GstAggregatorPrivate *priv = self->priv;
 358
 359   if (g_atomic_int_get (&self->priv->send_stream_start)) {
 360     gchar s_id[32];
 361
 362     GST_INFO_OBJECT (self, "pushing stream start");
 363     /* stream-start (FIXME: create id based on input ids) */
 364     g_snprintf (s_id, sizeof (s_id), "agg-%08x", g_random_int ());
 365     if (!gst_pad_push_event (self->srcpad, gst_event_new_stream_start (s_id))) {
 366       GST_WARNING_OBJECT (self->srcpad, "Sending stream start event failed");
 367     }
 368     g_atomic_int_set (&self->priv->send_stream_start, FALSE);
 369   }
 370
 371   if (self->priv->srccaps) {
 372
 373     GST_INFO_OBJECT (self, "pushing caps: %" GST_PTR_FORMAT,
 374         self->priv->srccaps);
 375     if (!gst_pad_push_event (self->srcpad,
 376             gst_event_new_caps (self->priv->srccaps))) {
 377       GST_WARNING_OBJECT (self->srcpad, "Sending caps event failed");
 378     }
 379     gst_caps_unref (self->priv->srccaps);
 380     self->priv->srccaps = NULL;
 381   }
 382
 383   if (g_atomic_int_get (&self->priv->send_segment)) {
 384     if (!g_atomic_int_get (&self->priv->flush_seeking)) {
 385       GstEvent *segev = gst_event_new_segment (&self->segment);
 386
 387       if (!self->priv->seqnum)
 388         self->priv->seqnum = gst_event_get_seqnum (segev);
 389       else
 390         gst_event_set_seqnum (segev, self->priv->seqnum);
 391
 392       GST_DEBUG_OBJECT (self, "pushing segment %" GST_PTR_FORMAT, segev);
 393       gst_pad_push_event (self->srcpad, segev);
 394       g_atomic_int_set (&self->priv->send_segment, FALSE);
 395     }
 396   }
 397
 398   if (priv->tags && priv->tags_changed) {
 399     gst_pad_push_event (self->srcpad,
 400         gst_event_new_tag (gst_tag_list_ref (priv->tags)));
 401     priv->tags_changed = FALSE;
 402   }
 403 }
 404
 405 /**
 406  * gst_aggregator_set_src_caps:
 407  * @self: The #GstAggregator
 408  * @caps: The #GstCaps to set on the src pad.
 409  *
 410  * Sets the caps to be used on the src pad.
 411  */
 412 void
 413 gst_aggregator_set_src_caps (GstAggregator * self, GstCaps * caps)
 414 {
 415   GST_AGGREGATOR_SETCAPS_LOCK (self);
 416   gst_caps_replace (&self->priv->srccaps, caps);
 417   _push_mandatory_events (self);
 418   GST_AGGREGATOR_SETCAPS_UNLOCK (self);
 419 }
 420
 421 /**
 422  * gst_aggregator_finish_buffer:
 423  * @self: The #GstAggregator
 424  * @buffer: the #GstBuffer to push.
 425  *
 426  * This method will take care of sending mandatory events before pushing
 427  * the provided buffer.
 428  */
 429 GstFlowReturn
 430 gst_aggregator_finish_buffer (GstAggregator * self, GstBuffer * buffer)
 431 {
 432   _push_mandatory_events (self);
 433
 434   if (!g_atomic_int_get (&self->priv->flush_seeking) &&
 435       gst_pad_is_active (self->srcpad)) {
 436     GST_TRACE_OBJECT (self, "pushing buffer %" GST_PTR_FORMAT, buffer);
 437     return gst_pad_push (self->srcpad, buffer);
 438   } else {
 439     GST_INFO_OBJECT (self, "Not pushing (active: %i, flushing: %i)",
 440         g_atomic_int_get (&self->priv->flush_seeking),
 441         gst_pad_is_active (self->srcpad));
 442     gst_buffer_unref (buffer);
 443     return GST_FLOW_OK;
 444   }
 445 }
 446
 447 static void
 448 _push_eos (GstAggregator * self)
 449 {
 450   GstEvent *event;
 451   _push_mandatory_events (self);
 452
 453   self->priv->send_eos = FALSE;
 454   event = gst_event_new_eos ();
 455   gst_event_set_seqnum (event, self->priv->seqnum);
 456   gst_pad_push_event (self->srcpad, event);
 457 }
 458
 459 static void
 460 aggregate_func (GstAggregator * self)
 461 {
 462   GstAggregatorPrivate *priv = self->priv;
 463   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 464
 465   if (self->priv->running == FALSE) {
 466     GST_DEBUG_OBJECT (self, "Not running anymore");
 467
 468     return;
 469   }
 470
 471   QUEUE_POP (self);
 472
 473   GST_LOG_OBJECT (self, "Checking aggregate");
 474   while (priv->send_eos && gst_aggregator_iterate_sinkpads (self,
 475           (GstAggregatorPadForeachFunc)
 476           _check_all_pads_with_data_or_eos_or_timeout, NULL) && priv->running) {
 477     GST_TRACE_OBJECT (self, "Actually aggregating!");
 478
 479     priv->flow_return = klass->aggregate (self);
 480
 481     if (priv->flow_return == GST_FLOW_EOS) {
 482       QUEUE_FLUSH (self);
 483       _push_eos (self);
 484     }
 485
 486     if (priv->flow_return == GST_FLOW_FLUSHING &&
 487         g_atomic_int_get (&priv->flush_seeking))
 488       priv->flow_return = GST_FLOW_OK;
 489
 490     GST_LOG_OBJECT (self, "flow return is %s",
 491         gst_flow_get_name (priv->flow_return));
 492
 493     if (priv->flow_return != GST_FLOW_OK)
 494       break;
 495   }
 496
 497 }
 498
 499 static gboolean
 500 _start (GstAggregator * self)
 501 {
 502   self->priv->running = TRUE;
 503   self->priv->send_stream_start = TRUE;
 504   self->priv->send_segment = TRUE;
 505   self->priv->send_eos = TRUE;
 506   self->priv->srccaps = NULL;
 507   self->priv->flow_return = GST_FLOW_OK;
 508
 509   return TRUE;
 510 }
 511
 512 static gboolean
 513 _check_pending_flush_stop (GstAggregatorPad * pad)
 514 {
 515   return (!pad->priv->pending_flush_stop && !pad->priv->pending_flush_start);
 516 }
 517
 518 static gboolean
 519 _stop_srcpad_task (GstAggregator * self, GstEvent * flush_start)
 520 {
 521   gboolean res = TRUE;
 522
 523   GST_INFO_OBJECT (self, "%s srcpad task",
 524       flush_start ? "Pausing" : "Stopping");
 525
 526   self->priv->running = FALSE;
 527   QUEUE_PUSH (self);
 528
 529   if (flush_start) {
 530     res = gst_pad_push_event (self->srcpad, flush_start);
 531   }
 532
 533   gst_pad_stop_task (self->srcpad);
 534   QUEUE_FLUSH (self);
 535
 536   return res;
 537 }
 538
 539 static void
 540 _start_srcpad_task (GstAggregator * self)
 541 {
 542   GST_INFO_OBJECT (self, "Starting srcpad task");
 543
 544   self->priv->running = TRUE;
 545   gst_pad_start_task (GST_PAD (self->srcpad),
 546       (GstTaskFunction) aggregate_func, self, NULL);
 547 }
 548
 549 static GstFlowReturn
 550 _flush (GstAggregator * self)
 551 {
 552   GstFlowReturn ret = GST_FLOW_OK;
 553   GstAggregatorPrivate *priv = self->priv;
 554   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);
 555
 556   GST_DEBUG_OBJECT (self, "Flushing everything");
 557   g_atomic_int_set (&priv->send_segment, TRUE);
 558   g_atomic_int_set (&priv->flush_seeking, FALSE);
 559   g_atomic_int_set (&priv->tags_changed, FALSE);
 560   if (klass->flush)
 561     ret = klass->flush (self);
 562
 563   return ret;
 564 }
 565
 566 static gboolean
 567 _all_flush_stop_received (GstAggregator * self)
 568 {
 569   GList *tmp;
 570   GstAggregatorPad *tmppad;
 571
 572   GST_OBJECT_LOCK (self);
 573   for (tmp = GST_ELEMENT (self)->sinkpads; tmp; tmp = tmp->next) {
 574     tmppad = (GstAggregatorPad *) tmp->data;
 575
 576     if (_check_pending_flush_stop (tmppad) == FALSE) {
 577       GST_DEBUG_OBJECT (tmppad, "Is not last %i -- %i",
 578           tmppad->priv->pending_flush_start, tmppad->priv->pending_flush_stop);
 579       GST_OBJECT_UNLOCK (self);
 580       return FALSE;
 581     }
 582   }
 583   GST_OBJECT_UNLOCK (self);
 584
 585   return TRUE;
 586 }
 587
 588 static void
 589 _flush_start (GstAggregator * self, GstAggregatorPad * aggpad, GstEvent * event)
 590 {
 591   GstBuffer *tmpbuf;
 592   GstAggregatorPrivate *priv = self->priv;
 593   GstAggregatorPadPrivate *padpriv = aggpad->priv;
 594
 595   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 596   /*  Remove pad buffer and wake up the streaming thread */
 597   tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 598   gst_buffer_replace (&tmpbuf, NULL);
 599   PAD_STREAM_LOCK (aggpad);
 600   if (g_atomic_int_compare_and_exchange (&padpriv->pending_flush_start,
 601           TRUE, FALSE) == TRUE) {
 602     GST_DEBUG_OBJECT (aggpad, "Expecting FLUSH_STOP now");
 603     g_atomic_int_set (&padpriv->pending_flush_stop, TRUE);
 604   }
 605
 606   if (g_atomic_int_get (&priv->flush_seeking)) {
 607     /* If flush_seeking we forward the first FLUSH_START */
 608     if (g_atomic_int_compare_and_exchange (&priv->pending_flush_start,
 609             TRUE, FALSE) == TRUE) {
 610
 611       GST_INFO_OBJECT (self, "Flushing, pausing srcpad task");
 612       _stop_srcpad_task (self, event);
 613       priv->flow_return = GST_FLOW_OK;
 614
 615       GST_INFO_OBJECT (self, "Getting STREAM_LOCK while seeking");
 616       GST_PAD_STREAM_LOCK (self->srcpad);
 617       GST_LOG_OBJECT (self, "GOT STREAM_LOCK");
 618       event = NULL;
 619     }
 620   } else {
 621     gst_event_unref (event);
 622   }
 623   PAD_STREAM_UNLOCK (aggpad);
 624
 625   tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 626   gst_buffer_replace (&tmpbuf, NULL);
 627 }
 628
 629 /* GstAggregator vmethods default implementations */
 630 static gboolean
 631 _sink_event (GstAggregator * self, GstAggregatorPad * aggpad, GstEvent * event)
 632 {
 633   gboolean res = TRUE;
 634   GstPad *pad = GST_PAD (aggpad);
 635   GstAggregatorPrivate *priv = self->priv;
 636
 637   switch (GST_EVENT_TYPE (event)) {
 638     case GST_EVENT_FLUSH_START:
 639     {
 640       _flush_start (self, aggpad, event);
 641       /* We forward only in one case: right after flush_seeking */
 642       event = NULL;
 643       goto eat;
 644     }
 645     case GST_EVENT_FLUSH_STOP:
 646     {
 647       GST_DEBUG_OBJECT (aggpad, "Got FLUSH_STOP");
 648
 649       _aggpad_flush (aggpad, self);
 650       if (g_atomic_int_get (&priv->flush_seeking)) {
 651         g_atomic_int_set (&aggpad->priv->pending_flush_stop, FALSE);
 652
 653         if (g_atomic_int_get (&priv->flush_seeking)) {
 654           if (_all_flush_stop_received (self)) {
 655             /* That means we received FLUSH_STOP/FLUSH_STOP on
 656              * all sinkpads -- Seeking is Done... sending FLUSH_STOP */
 657             _flush (self);
 658             gst_pad_push_event (self->srcpad, event);
 659             priv->send_eos = TRUE;
 660             event = NULL;
 661             QUEUE_PUSH (self);
 662
 663             GST_INFO_OBJECT (self, "Releasing source pad STREAM_LOCK");
 664             GST_PAD_STREAM_UNLOCK (self->srcpad);
 665             _start_srcpad_task (self);
 666           }
 667         }
 668       }
 669
 670       /* We never forward the event */
 671       goto eat;
 672     }
 673     case GST_EVENT_EOS:
 674     {
 675       GST_DEBUG_OBJECT (aggpad, "EOS");
 676
 677       /* We still have a buffer, and we don't want the subclass to have to
 678        * check for it. Mark pending_eos, eos will be set when steal_buffer is
 679        * called
 680        */
 681       PAD_LOCK_EVENT (aggpad);
 682       if (!aggpad->buffer) {
 683         aggpad->eos = TRUE;
 684       } else {
 685         aggpad->priv->pending_eos = TRUE;
 686       }
 687       PAD_UNLOCK_EVENT (aggpad);
 688
 689       QUEUE_PUSH (self);
 690       goto eat;
 691     }
 692     case GST_EVENT_SEGMENT:
 693     {
 694       PAD_LOCK_EVENT (aggpad);
 695       gst_event_copy_segment (event, &aggpad->segment);
 696       self->priv->seqnum = gst_event_get_seqnum (event);
 697       PAD_UNLOCK_EVENT (aggpad);
 698       goto eat;
 699     }
 700     case GST_EVENT_STREAM_START:
 701     {
 702       goto eat;
 703     }
 704     case GST_EVENT_TAG:
 705     {
 706       GstTagList *tags;
 707
 708       gst_event_parse_tag (event, &tags);
 709
 710       if (gst_tag_list_get_scope (tags) == GST_TAG_SCOPE_STREAM) {
 711         gst_aggregator_merge_tags (self, tags, GST_TAG_MERGE_REPLACE);
 712         gst_event_unref (event);
 713         event = NULL;
 714         goto eat;
 715       }
 716       break;
 717     }
 718     default:
 719     {
 720       break;
 721     }
 722   }
 723
 724   GST_DEBUG_OBJECT (pad, "Forwarding event: %" GST_PTR_FORMAT, event);
 725   return gst_pad_event_default (pad, GST_OBJECT (self), event);
 726
 727 eat:
 728   GST_DEBUG_OBJECT (pad, "Eating event: %" GST_PTR_FORMAT, event);
 729   if (event)
 730     gst_event_unref (event);
 731
 732   return res;
 733 }
 734
 735 static gboolean
 736 _stop_pad (GstAggregator * self, GstAggregatorPad * pad, gpointer unused_udata)
 737 {
 738   _aggpad_flush (pad, self);
 739
 740   PAD_LOCK_EVENT (pad);
 741   /* remove the timeouts */
 742   if (pad->priv->timeout_id) {
 743     gst_clock_id_unschedule (pad->priv->timeout_id);
 744     gst_clock_id_unref (pad->priv->timeout_id);
 745     pad->priv->timeout_id = NULL;
 746   }
 747   PAD_UNLOCK_EVENT (pad);
 748
 749   return TRUE;
 750 }
 751
 752 static gboolean
 753 _stop (GstAggregator * agg)
 754 {
 755   _reset_flow_values (agg);
 756
 757   gst_aggregator_iterate_sinkpads (agg,
 758       (GstAggregatorPadForeachFunc) _stop_pad, NULL);
 759
 760   if (agg->priv->tags)
 761     gst_tag_list_unref (agg->priv->tags);
 762   agg->priv->tags = NULL;
 763
 764   return TRUE;
 765 }
 766
 767 /* GstElement vmethods implementations */
 768 static GstStateChangeReturn
 769 _change_state (GstElement * element, GstStateChange transition)
 770 {
 771   GstStateChangeReturn ret;
 772   GstAggregator *self = GST_AGGREGATOR (element);
 773   GstAggregatorClass *agg_class = GST_AGGREGATOR_GET_CLASS (self);
 774
 775
 776   switch (transition) {
 777     case GST_STATE_CHANGE_READY_TO_PAUSED:
 778       agg_class->start (self);
 779       break;
 780     default:
 781       break;
 782   }
 783
 784   if ((ret =
 785           GST_ELEMENT_CLASS (aggregator_parent_class)->change_state (element,
 786               transition)) == GST_STATE_CHANGE_FAILURE)
 787     goto failure;
 788
 789
 790   switch (transition) {
 791     case GST_STATE_CHANGE_PAUSED_TO_READY:
 792       agg_class->stop (self);
 793       break;
 794     default:
 795       break;
 796   }
 797
 798   return ret;
 799
 800 failure:
 801   {
 802     GST_ERROR_OBJECT (element, "parent failed state change");
 803     return ret;
 804   }
 805 }
 806
 807 static void
 808 _release_pad (GstElement * element, GstPad * pad)
 809 {
 810   GstBuffer *tmpbuf;
 811
 812   GstAggregator *self = GST_AGGREGATOR (element);
 813   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
 814
 815   GST_INFO_OBJECT (pad, "Removing pad");
 816
 817   g_atomic_int_set (&aggpad->priv->flushing, TRUE);
 818   tmpbuf = gst_aggregator_pad_steal_buffer (aggpad);
 819   gst_buffer_replace (&tmpbuf, NULL);
 820   gst_element_remove_pad (element, pad);
 821
 822   /* Something changed make sure we try to aggregate */
 823   QUEUE_PUSH (self);
 824 }
 825
 826 static gboolean
 827 _unresponsive_timeout (GstClock * clock, GstClockTime time, GstClockID id,
 828     gpointer user_data)
 829 {
 830   GstAggregatorPad *aggpad;
 831   GstAggregator *self;
 832
 833   if (user_data == NULL)
 834     return FALSE;
 835
 836   aggpad = GST_AGGREGATOR_PAD (user_data);
 837
 838   /* avoid holding the last reference to the parent element here */
 839   PAD_LOCK_EVENT (aggpad);
 840
 841   self = GST_AGGREGATOR (gst_pad_get_parent (GST_PAD (aggpad)));
 842
 843   GST_DEBUG_OBJECT (aggpad, "marked unresponsive");
 844
 845   g_atomic_int_set (&aggpad->unresponsive, TRUE);
 846
 847   if (self) {
 848     QUEUE_PUSH (self);
 849     gst_object_unref (self);
 850   }
 851
 852   PAD_UNLOCK_EVENT (aggpad);
 853
 854   return TRUE;
 855 }
 856
 857 static GstPad *
 858 _request_new_pad (GstElement * element,
 859     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
 860 {
 861   GstAggregator *self;
 862   GstAggregatorPad *agg_pad;
 863
 864   GstElementClass *klass = GST_ELEMENT_GET_CLASS (element);
 865   GstAggregatorPrivate *priv = GST_AGGREGATOR (element)->priv;
 866
 867   self = GST_AGGREGATOR (element);
 868
 869   if (templ == gst_element_class_get_pad_template (klass, "sink_%u")) {
 870     gint serial = 0;
 871     gchar *name = NULL;
 872
 873     GST_OBJECT_LOCK (element);
 874     if (req_name == NULL || strlen (req_name) < 6
 875         || !g_str_has_prefix (req_name, "sink_")) {
 876       /* no name given when requesting the pad, use next available int */
 877       priv->padcount++;
 878     } else {
 879       /* parse serial number from requested padname */
 880       serial = g_ascii_strtoull (&req_name[5], NULL, 10);
 881       if (serial >= priv->padcount)
 882         priv->padcount = serial;
 883     }
 884
 885     name = g_strdup_printf ("sink_%u", priv->padcount);
 886     agg_pad = g_object_new (GST_AGGREGATOR_GET_CLASS (self)->sinkpads_type,
 887         "name", name, "direction", GST_PAD_SINK, "template", templ, NULL);
 888     g_free (name);
 889
 890     GST_OBJECT_UNLOCK (element);
 891
 892   } else {
 893     return NULL;
 894   }
 895
 896   GST_DEBUG_OBJECT (element, "Adding pad %s", GST_PAD_NAME (agg_pad));
 897
 898   if (priv->running)
 899     gst_pad_set_active (GST_PAD (agg_pad), TRUE);
 900
 901   /* add the pad to the element */
 902   gst_element_add_pad (element, GST_PAD (agg_pad));
 903
 904   return GST_PAD (agg_pad);
 905 }
 906
 907 static gboolean
 908 _send_event (GstElement * element, GstEvent * event)
 909 {
 910   GstAggregator *self = GST_AGGREGATOR (element);
 911
 912   GST_STATE_LOCK (element);
 913   if (GST_EVENT_TYPE (event) == GST_EVENT_SEEK &&
 914       GST_STATE (element) < GST_STATE_PAUSED) {
 915     gdouble rate;
 916     GstFormat fmt;
 917     GstSeekFlags flags;
 918     GstSeekType start_type, stop_type;
 919     gint64 start, stop;
 920
 921     gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
 922         &start, &stop_type, &stop);
 923     gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
 924         stop_type, stop, NULL);
 925
 926     self->priv->seqnum = gst_event_get_seqnum (event);
 927     GST_DEBUG_OBJECT (element, "Storing segment %" GST_PTR_FORMAT, event);
 928   }
 929   GST_STATE_UNLOCK (element);
 930
 931
 932   return GST_ELEMENT_CLASS (aggregator_parent_class)->send_event (element,
 933       event);
 934 }
 935
 936 static gboolean
 937 _src_query (GstAggregator * self, GstQuery * query)
 938 {
 939   gboolean res = TRUE;
 940
 941   switch (GST_QUERY_TYPE (query)) {
 942     case GST_QUERY_SEEKING:
 943     {
 944       GstFormat format;
 945
 946       /* don't pass it along as some (file)sink might claim it does
 947        * whereas with a collectpads in between that will not likely work */
 948       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
 949       gst_query_set_seeking (query, format, FALSE, 0, -1);
 950       res = TRUE;
 951
 952       goto discard;
 953     }
 954     default:
 955       break;
 956   }
 957
 958   return gst_pad_query_default (self->srcpad, GST_OBJECT (self), query);
 959
 960 discard:
 961   return res;
 962 }
 963
 964 static gboolean
 965 event_forward_func (GstPad * pad, EventData * evdata)
 966 {
 967   gboolean ret = TRUE;
 968   GstPad *peer = gst_pad_get_peer (pad);
 969   GstAggregatorPadPrivate *padpriv = GST_AGGREGATOR_PAD (pad)->priv;
 970
 971   if (peer) {
 972     ret = gst_pad_send_event (peer, gst_event_ref (evdata->event));
 973     GST_DEBUG_OBJECT (pad, "return of event push is %d", ret);
 974     gst_object_unref (peer);
 975   }
 976
 977   if (ret == FALSE) {
 978     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK)
 979       GST_ERROR_OBJECT (pad, "Event %" GST_PTR_FORMAT " failed", evdata->event);
 980
 981     if (GST_EVENT_TYPE (evdata->event) == GST_EVENT_SEEK) {
 982       GstQuery *seeking = gst_query_new_seeking (GST_FORMAT_TIME);
 983
 984       if (gst_pad_query (peer, seeking)) {
 985         gboolean seekable;
 986
 987         gst_query_parse_seeking (seeking, NULL, &seekable, NULL, NULL);
 988
 989         if (seekable == FALSE) {
 990           GST_INFO_OBJECT (pad,
 991               "Source not seekable, We failed but it does not matter!");
 992
 993           ret = TRUE;
 994         }
 995       } else {
 996         GST_ERROR_OBJECT (pad, "Query seeking FAILED");
 997       }
 998     }
 999
1000     if (evdata->flush) {
1001       padpriv->pending_flush_start = FALSE;
1002       padpriv->pending_flush_stop = FALSE;
1003     }
1004   } else {
1005     evdata->one_actually_seeked = TRUE;
1006   }
1007
1008   evdata->result &= ret;
1009
1010   /* Always send to all pads */
1011   return FALSE;
1012 }
1013
1014 static gboolean
1015 _set_flush_pending (GstAggregator * self, GstAggregatorPad * pad,
1016     gpointer udata)
1017 {
1018   pad->priv->pending_flush_start = TRUE;
1019   pad->priv->pending_flush_stop = FALSE;
1020
1021   return TRUE;
1022 }
1023
1024 static EventData
1025 _forward_event_to_all_sinkpads (GstAggregator * self, GstEvent * event,
1026     gboolean flush)
1027 {
1028   EventData evdata;
1029
1030   evdata.event = event;
1031   evdata.result = TRUE;
1032   evdata.flush = flush;
1033   evdata.one_actually_seeked = FALSE;
1034
1035   /* We first need to set all pads as flushing in a first pass
1036    * as flush_start flush_stop is sometimes sent synchronously
1037    * while we send the seek event */
1038   if (flush)
1039     gst_aggregator_iterate_sinkpads (self,
1040         (GstAggregatorPadForeachFunc) _set_flush_pending, NULL);
1041   gst_pad_forward (self->srcpad, (GstPadForwardFunction) event_forward_func,
1042       &evdata);
1043
1044   gst_event_unref (event);
1045
1046   return evdata;
1047 }
1048
1049 static gboolean
1050 _do_seek (GstAggregator * self, GstEvent * event)
1051 {
1052   gdouble rate;
1053   GstFormat fmt;
1054   GstSeekFlags flags;
1055   GstSeekType start_type, stop_type;
1056   gint64 start, stop;
1057   gboolean flush;
1058   EventData evdata;
1059   GstAggregatorPrivate *priv = self->priv;
1060
1061   gst_event_parse_seek (event, &rate, &fmt, &flags, &start_type,
1062       &start, &stop_type, &stop);
1063
1064   GST_INFO_OBJECT (self, "starting SEEK");
1065
1066   flush = flags & GST_SEEK_FLAG_FLUSH;
1067
1068   if (flush) {
1069     g_atomic_int_set (&priv->pending_flush_start, TRUE);
1070     g_atomic_int_set (&priv->flush_seeking, TRUE);
1071   }
1072
1073   gst_segment_do_seek (&self->segment, rate, fmt, flags, start_type, start,
1074       stop_type, stop, NULL);
1075
1076   /* forward the seek upstream */
1077   evdata = _forward_event_to_all_sinkpads (self, event, flush);
1078   event = NULL;
1079
1080   if (!evdata.result || !evdata.one_actually_seeked) {
1081     g_atomic_int_set (&priv->flush_seeking, FALSE);
1082     g_atomic_int_set (&priv->pending_flush_start, FALSE);
1083   }
1084
1085   GST_INFO_OBJECT (self, "seek done, result: %d", evdata.result);
1086
1087   return evdata.result;
1088 }
1089
1090 static gboolean
1091 _src_event (GstAggregator * self, GstEvent * event)
1092 {
1093   EventData evdata;
1094   gboolean res = TRUE;
1095
1096   switch (GST_EVENT_TYPE (event)) {
1097     case GST_EVENT_SEEK:
1098     {
1099       gst_event_ref (event);
1100       res = _do_seek (self, event);
1101       gst_event_unref (event);
1102       event = NULL;
1103       goto done;
1104     }
1105     case GST_EVENT_NAVIGATION:
1106     {
1107       /* navigation is rather pointless. */
1108       res = FALSE;
1109       gst_event_unref (event);
1110       goto done;
1111     }
1112     default:
1113     {
1114       break;
1115     }
1116   }
1117
1118   evdata = _forward_event_to_all_sinkpads (self, event, FALSE);
1119   res = evdata.result;
1120
1121 done:
1122   return res;
1123 }
1124
1125 static gboolean
1126 src_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
1127 {
1128   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1129
1130   return klass->src_event (GST_AGGREGATOR (parent), event);
1131 }
1132
1133 static gboolean
1134 src_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
1135 {
1136   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1137
1138   return klass->src_query (GST_AGGREGATOR (parent), query);
1139 }
1140
1141 static gboolean
1142 src_activate_mode (GstPad * pad,
1143     GstObject * parent, GstPadMode mode, gboolean active)
1144 {
1145   GstAggregator *self = GST_AGGREGATOR (parent);
1146   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1147
1148   if (klass->src_activate) {
1149     if (klass->src_activate (self, mode, active) == FALSE) {
1150       return FALSE;
1151     }
1152   }
1153
1154   if (active == TRUE) {
1155     switch (mode) {
1156       case GST_PAD_MODE_PUSH:
1157       {
1158         GST_INFO_OBJECT (pad, "Activating pad!");
1159         _start_srcpad_task (self);
1160         return TRUE;
1161       }
1162       default:
1163       {
1164         GST_ERROR_OBJECT (pad, "Only supported mode is PUSH");
1165         return FALSE;
1166       }
1167     }
1168   }
1169
1170   /* deactivating */
1171   GST_INFO_OBJECT (self, "Deactivating srcpad");
1172   _stop_srcpad_task (self, FALSE);
1173
1174   return TRUE;
1175 }
1176
1177 static gboolean
1178 _sink_query (GstAggregator * self, GstAggregatorPad * aggpad, GstQuery * query)
1179 {
1180   GstPad *pad = GST_PAD (aggpad);
1181
1182   return gst_pad_query_default (pad, GST_OBJECT (self), query);
1183 }
1184
1185 static void
1186 gst_aggregator_finalize (GObject * object)
1187 {
1188   GstAggregator *self = (GstAggregator *) object;
1189
1190   gst_object_unref (self->clock);
1191   g_mutex_clear (&self->priv->setcaps_lock);
1192
1193   G_OBJECT_CLASS (aggregator_parent_class)->finalize (object);
1194 }
1195
1196 static void
1197 gst_aggregator_dispose (GObject * object)
1198 {
1199   GstAggregator *self = (GstAggregator *) object;
1200
1201   G_OBJECT_CLASS (aggregator_parent_class)->dispose (object);
1202
1203   if (AGGREGATOR_QUEUE (self)) {
1204     g_async_queue_unref (AGGREGATOR_QUEUE (self));
1205     AGGREGATOR_QUEUE (self) = NULL;
1206   }
1207 }
1208
1209 /**
1210  * gst_aggregator_set_timeout:
1211  * @agg: a #GstAggregator
1212  * @timeout: the new timeout value.
1213  *
1214  * Sets the new timeout value to @timeout. This value is used to limit the
1215  * amount of time a pad waits for data to appear before considering the pad
1216  * as unresponsive.
1217  */
1218 static void
1219 gst_aggregator_set_timeout (GstAggregator * agg, gint64 timeout)
1220 {
1221   g_return_if_fail (GST_IS_AGGREGATOR (agg));
1222
1223   GST_OBJECT_LOCK (agg);
1224   agg->timeout = timeout;
1225   GST_OBJECT_UNLOCK (agg);
1226 }
1227
1228 /**
1229  * gst_aggregator_get_timeout:
1230  * @agg: a #GstAggregator
1231  *
1232  * Gets the timeout value. See gst_aggregator_set_timeout for
1233  * more details.
1234  *
1235  * Returns: The time in nanoseconds to wait for data to arrive on a sink pad
1236  * before a pad is deemed unresponsive. A value of -1 means an
1237  * unlimited time.
1238  */
1239 static gint64
1240 gst_aggregator_get_timeout (GstAggregator * agg)
1241 {
1242   gint64 res;
1243
1244   g_return_val_if_fail (GST_IS_AGGREGATOR (agg), -1);
1245
1246   GST_OBJECT_LOCK (agg);
1247   res = agg->timeout;
1248   GST_OBJECT_UNLOCK (agg);
1249
1250   return res;
1251 }
1252
1253 static void
1254 gst_aggregator_set_property (GObject * object, guint prop_id,
1255     const GValue * value, GParamSpec * pspec)
1256 {
1257   GstAggregator *agg = GST_AGGREGATOR (object);
1258
1259   switch (prop_id) {
1260     case PROP_TIMEOUT:
1261       gst_aggregator_set_timeout (agg, g_value_get_int64 (value));
1262       break;
1263     default:
1264       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1265       break;
1266   }
1267 }
1268
1269 static void
1270 gst_aggregator_get_property (GObject * object, guint prop_id,
1271     GValue * value, GParamSpec * pspec)
1272 {
1273   GstAggregator *agg = GST_AGGREGATOR (object);
1274
1275   switch (prop_id) {
1276     case PROP_TIMEOUT:
1277       g_value_set_int64 (value, gst_aggregator_get_timeout (agg));
1278       break;
1279     default:
1280       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1281       break;
1282   }
1283 }
1284
1285 /* GObject vmethods implementations */
1286 static void
1287 gst_aggregator_class_init (GstAggregatorClass * klass)
1288 {
1289   GObjectClass *gobject_class = (GObjectClass *) klass;
1290   GstElementClass *gstelement_class = (GstElementClass *) klass;
1291
1292   aggregator_parent_class = g_type_class_peek_parent (klass);
1293   g_type_class_add_private (klass, sizeof (GstAggregatorPrivate));
1294
1295   GST_DEBUG_CATEGORY_INIT (aggregator_debug, "aggregator",
1296       GST_DEBUG_FG_MAGENTA, "GstAggregator");
1297
1298   klass->sinkpads_type = GST_TYPE_AGGREGATOR_PAD;
1299   klass->start = _start;
1300   klass->stop = _stop;
1301
1302   klass->sink_event = _sink_event;
1303   klass->sink_query = _sink_query;
1304
1305   klass->src_event = _src_event;
1306   klass->src_query = _src_query;
1307
1308   gstelement_class->request_new_pad = GST_DEBUG_FUNCPTR (_request_new_pad);
1309   gstelement_class->send_event = GST_DEBUG_FUNCPTR (_send_event);
1310   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (_release_pad);
1311   gstelement_class->change_state = GST_DEBUG_FUNCPTR (_change_state);
1312
1313   gobject_class->set_property = gst_aggregator_set_property;
1314   gobject_class->get_property = gst_aggregator_get_property;
1315   gobject_class->finalize = gst_aggregator_finalize;
1316   gobject_class->dispose = gst_aggregator_dispose;
1317
1318   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
1319       g_param_spec_int64 ("timeout", "Buffer timeout",
1320           "Number of nanoseconds to wait for a buffer to arrive on a sink pad"
1321           "before the pad is deemed unresponsive (-1 unlimited)", -1,
1322           G_MAXINT64, DEFAULT_TIMEOUT,
1323           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1324 }
1325
1326 static void
1327 gst_aggregator_init (GstAggregator * self, GstAggregatorClass * klass)
1328 {
1329   GstPadTemplate *pad_template;
1330   GstAggregatorPrivate *priv;
1331
1332   g_return_if_fail (klass->aggregate != NULL);
1333
1334   self->priv =
1335       G_TYPE_INSTANCE_GET_PRIVATE (self, GST_TYPE_AGGREGATOR,
1336       GstAggregatorPrivate);
1337
1338   priv = self->priv;
1339
1340   pad_template =
1341       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
1342   g_return_if_fail (pad_template != NULL);
1343
1344   priv->padcount = -1;
1345   priv->tags_changed = FALSE;
1346   _reset_flow_values (self);
1347
1348   AGGREGATOR_QUEUE (self) = g_async_queue_new ();
1349   self->srcpad = gst_pad_new_from_template (pad_template, "src");
1350
1351   gst_pad_set_event_function (self->srcpad,
1352       GST_DEBUG_FUNCPTR ((GstPadEventFunction) src_event_func));
1353   gst_pad_set_query_function (self->srcpad,
1354       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) src_query_func));
1355   gst_pad_set_activatemode_function (self->srcpad,
1356       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) src_activate_mode));
1357
1358   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
1359
1360   self->clock = gst_system_clock_obtain ();
1361   self->timeout = -1;
1362
1363   g_mutex_init (&self->priv->setcaps_lock);
1364 }
1365
1366 /* we can't use G_DEFINE_ABSTRACT_TYPE because we need the klass in the _init
1367  * method to get to the padtemplates */
1368 GType
1369 gst_aggregator_get_type (void)
1370 {
1371   static volatile gsize type = 0;
1372
1373   if (g_once_init_enter (&type)) {
1374     GType _type;
1375     static const GTypeInfo info = {
1376       sizeof (GstAggregatorClass),
1377       NULL,
1378       NULL,
1379       (GClassInitFunc) gst_aggregator_class_init,
1380       NULL,
1381       NULL,
1382       sizeof (GstAggregator),
1383       0,
1384       (GInstanceInitFunc) gst_aggregator_init,
1385     };
1386
1387     _type = g_type_register_static (GST_TYPE_ELEMENT,
1388         "GstAggregator", &info, G_TYPE_FLAG_ABSTRACT);
1389     g_once_init_leave (&type, _type);
1390   }
1391   return type;
1392 }
1393
1394 static GstFlowReturn
1395 _chain (GstPad * pad, GstObject * object, GstBuffer * buffer)
1396 {
1397   GstBuffer *actual_buf = buffer;
1398   GstAggregator *self = GST_AGGREGATOR (object);
1399   GstAggregatorPrivate *priv = self->priv;
1400   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1401   GstAggregatorClass *aggclass = GST_AGGREGATOR_GET_CLASS (object);
1402   GstClockTime timeout = gst_aggregator_get_timeout (self);
1403   GstClockTime now;
1404
1405   GST_DEBUG_OBJECT (aggpad, "Start chaining a buffer %" GST_PTR_FORMAT, buffer);
1406   if (aggpad->priv->timeout_id) {
1407     gst_clock_id_unschedule (aggpad->priv->timeout_id);
1408     gst_clock_id_unref (aggpad->priv->timeout_id);
1409     aggpad->priv->timeout_id = NULL;
1410   }
1411   g_atomic_int_set (&aggpad->unresponsive, FALSE);
1412
1413   PAD_STREAM_LOCK (aggpad);
1414
1415   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1416     goto flushing;
1417
1418   if (g_atomic_int_get (&aggpad->priv->pending_eos) == TRUE)
1419     goto eos;
1420
1421   PAD_LOCK_EVENT (aggpad);
1422
1423   if (aggpad->buffer) {
1424     GST_DEBUG_OBJECT (aggpad, "Waiting for buffer to be consumed");
1425     PAD_WAIT_EVENT (aggpad);
1426   }
1427   PAD_UNLOCK_EVENT (aggpad);
1428
1429   if (g_atomic_int_get (&aggpad->priv->flushing) == TRUE)
1430     goto flushing;
1431
1432   if (aggclass->clip) {
1433     aggclass->clip (self, aggpad, buffer, &actual_buf);
1434   }
1435
1436   PAD_LOCK_EVENT (aggpad);
1437   if (aggpad->buffer)
1438     gst_buffer_unref (aggpad->buffer);
1439   aggpad->buffer = actual_buf;
1440   PAD_UNLOCK_EVENT (aggpad);
1441   PAD_STREAM_UNLOCK (aggpad);
1442
1443   QUEUE_PUSH (self);
1444
1445   if (GST_CLOCK_TIME_IS_VALID (timeout)) {
1446     now = gst_clock_get_time (self->clock);
1447     aggpad->priv->timeout_id =
1448         gst_clock_new_single_shot_id (self->clock, now + timeout);
1449     gst_clock_id_wait_async (aggpad->priv->timeout_id, _unresponsive_timeout,
1450         gst_object_ref (aggpad), gst_object_unref);
1451   }
1452
1453   GST_DEBUG_OBJECT (aggpad, "Done chaining");
1454
1455   return priv->flow_return;
1456
1457 flushing:
1458   PAD_STREAM_UNLOCK (aggpad);
1459
1460   gst_buffer_unref (buffer);
1461   GST_DEBUG_OBJECT (aggpad, "We are flushing");
1462
1463   return GST_FLOW_FLUSHING;
1464
1465 eos:
1466   PAD_STREAM_UNLOCK (aggpad);
1467
1468   gst_buffer_unref (buffer);
1469   GST_DEBUG_OBJECT (pad, "We are EOS already...");
1470
1471   return GST_FLOW_EOS;
1472 }
1473
1474 static gboolean
1475 pad_query_func (GstPad * pad, GstObject * parent, GstQuery * query)
1476 {
1477   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1478
1479   return klass->sink_query (GST_AGGREGATOR (parent),
1480       GST_AGGREGATOR_PAD (pad), query);
1481 }
1482
1483 static gboolean
1484 pad_event_func (GstPad * pad, GstObject * parent, GstEvent * event)
1485 {
1486   GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (parent);
1487
1488   return klass->sink_event (GST_AGGREGATOR (parent),
1489       GST_AGGREGATOR_PAD (pad), event);
1490 }
1491
1492 static gboolean
1493 pad_activate_mode_func (GstPad * pad,
1494     GstObject * parent, GstPadMode mode, gboolean active)
1495 {
1496   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
1497
1498   if (active == FALSE) {
1499     PAD_LOCK_EVENT (aggpad);
1500     g_atomic_int_set (&aggpad->priv->flushing, TRUE);
1501     gst_buffer_replace (&aggpad->buffer, NULL);
1502     PAD_BROADCAST_EVENT (aggpad);
1503     PAD_UNLOCK_EVENT (aggpad);
1504   } else {
1505     g_atomic_int_set (&aggpad->priv->flushing, FALSE);
1506     PAD_LOCK_EVENT (aggpad);
1507     PAD_BROADCAST_EVENT (aggpad);
1508     PAD_UNLOCK_EVENT (aggpad);
1509   }
1510
1511   return TRUE;
1512 }
1513
1514 /***********************************
1515  * GstAggregatorPad implementation  *
1516  ************************************/
1517 static GstPadClass *aggregator_pad_parent_class = NULL;
1518 G_DEFINE_TYPE (GstAggregatorPad, gst_aggregator_pad, GST_TYPE_PAD);
1519
1520 static void
1521 _pad_constructed (GObject * object)
1522 {
1523   GstPad *pad = GST_PAD (object);
1524
1525   gst_pad_set_chain_function (pad,
1526       GST_DEBUG_FUNCPTR ((GstPadChainFunction) _chain));
1527   gst_pad_set_event_function (pad,
1528       GST_DEBUG_FUNCPTR ((GstPadEventFunction) pad_event_func));
1529   gst_pad_set_query_function (pad,
1530       GST_DEBUG_FUNCPTR ((GstPadQueryFunction) pad_query_func));
1531   gst_pad_set_activatemode_function (pad,
1532       GST_DEBUG_FUNCPTR ((GstPadActivateModeFunction) pad_activate_mode_func));
1533 }
1534
1535 static void
1536 gst_aggregator_pad_finalize (GObject * object)
1537 {
1538   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1539
1540   g_mutex_clear (&pad->priv->event_lock);
1541   g_cond_clear (&pad->priv->event_cond);
1542   g_mutex_clear (&pad->priv->stream_lock);
1543
1544   G_OBJECT_CLASS (aggregator_pad_parent_class)->finalize (object);
1545 }
1546
1547 static void
1548 gst_aggregator_pad_dispose (GObject * object)
1549 {
1550   GstAggregatorPad *pad = (GstAggregatorPad *) object;
1551   GstBuffer *buf;
1552
1553   buf = gst_aggregator_pad_steal_buffer (pad);
1554   if (buf)
1555     gst_buffer_unref (buf);
1556
1557   G_OBJECT_CLASS (aggregator_pad_parent_class)->dispose (object);
1558 }
1559
1560 static void
1561 gst_aggregator_pad_class_init (GstAggregatorPadClass * klass)
1562 {
1563   GObjectClass *gobject_class = (GObjectClass *) klass;
1564
1565   aggregator_pad_parent_class = g_type_class_peek_parent (klass);
1566   g_type_class_add_private (klass, sizeof (GstAggregatorPadPrivate));
1567
1568   gobject_class->constructed = GST_DEBUG_FUNCPTR (_pad_constructed);
1569   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_aggregator_pad_finalize);
1570   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_aggregator_pad_dispose);
1571 }
1572
1573 static void
1574 gst_aggregator_pad_init (GstAggregatorPad * pad)
1575 {
1576   pad->priv =
1577       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AGGREGATOR_PAD,
1578       GstAggregatorPadPrivate);
1579
1580   pad->buffer = NULL;
1581   g_mutex_init (&pad->priv->event_lock);
1582   g_cond_init (&pad->priv->event_cond);
1583
1584   g_mutex_init (&pad->priv->stream_lock);
1585 }
1586
1587 /**
1588  * gst_aggregator_pad_steal_buffer:
1589  * @pad: the pad to get buffer from
1590  *
1591  * Steal the ref to the buffer currently queued in @pad.
1592  *
1593  * Returns: (transfer full): The buffer in @pad or NULL if no buffer was
1594  *   queued. You should unref the buffer after usage.
1595  */
1596 GstBuffer *
1597 gst_aggregator_pad_steal_buffer (GstAggregatorPad * pad)
1598 {
1599   GstBuffer *buffer = NULL;
1600
1601   PAD_LOCK_EVENT (pad);
1602   if (pad->buffer) {
1603     GST_TRACE_OBJECT (pad, "Consuming buffer");
1604     buffer = pad->buffer;
1605     pad->buffer = NULL;
1606     if (pad->priv->pending_eos) {
1607       pad->priv->pending_eos = FALSE;
1608       pad->eos = TRUE;
1609     }
1610     PAD_BROADCAST_EVENT (pad);
1611     GST_DEBUG_OBJECT (pad, "Consummed: %" GST_PTR_FORMAT, buffer);
1612   }
1613   PAD_UNLOCK_EVENT (pad);
1614
1615   return buffer;
1616 }
1617
1618 /**
1619  * gst_aggregator_pad_get_buffer:
1620  * @pad: the pad to get buffer from
1621  *
1622  * Returns: (transfer full): A reference to the buffer in @pad or
1623  * NULL if no buffer was queued. You should unref the buffer after
1624  * usage.
1625  */
1626 GstBuffer *
1627 gst_aggregator_pad_get_buffer (GstAggregatorPad * pad)
1628 {
1629   GstBuffer *buffer = NULL;
1630
1631   PAD_LOCK_EVENT (pad);
1632   if (pad->buffer)
1633     buffer = gst_buffer_ref (pad->buffer);
1634   PAD_UNLOCK_EVENT (pad);
1635
1636   return buffer;
1637 }
1638
1639 /**
1640  * gst_aggregator_merge_tags:
1641  * @self: a #GstAggregator
1642  * @tags: a #GstTagList to merge
1643  * @mode: the #GstTagMergeMode to use
1644  *
1645  * Adds tags to so-called pending tags, which will be processed
1646  * before pushing out data downstream.
1647  *
1648  * Note that this is provided for convenience, and the subclass is
1649  * not required to use this and can still do tag handling on its own.
1650  *
1651  * MT safe.
1652  */
1653 void
1654 gst_aggregator_merge_tags (GstAggregator * self,
1655     const GstTagList * tags, GstTagMergeMode mode)
1656 {
1657   GstTagList *otags;
1658
1659   g_return_if_fail (GST_IS_AGGREGATOR (self));
1660   g_return_if_fail (tags == NULL || GST_IS_TAG_LIST (tags));
1661
1662   /* FIXME Check if we can use OBJECT lock here! */
1663   GST_OBJECT_LOCK (self);
1664   if (tags)
1665     GST_DEBUG_OBJECT (self, "merging tags %" GST_PTR_FORMAT, tags);
1666   otags = self->priv->tags;
1667   self->priv->tags = gst_tag_list_merge (self->priv->tags, tags, mode);
1668   if (otags)
1669     gst_tag_list_unref (otags);
1670   self->priv->tags_changed = TRUE;
1671   GST_OBJECT_UNLOCK (self);
1672 }