libs/gst/base/gstaggregator.c

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