gst-libs/gst/audio/gstaudioaggregator.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2001 Thomas <thomas@apestaart.org>
   4  *               2005,2006 Wim Taymans <wim@fluendo.com>
   5  *                    2013 Sebastian Dröge <sebastian@centricular.com>
   6  *                    2014 Collabora
   7  *                             Olivier Crete <olivier.crete@collabora.com>
   8  *
   9  * gstaudioaggregator.c:
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Library General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Library General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Library General Public
  22  * License along with this library; if not, write to the
  23  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  24  * Boston, MA 02110-1301, USA.
  25  */
  26 /**
  27  * SECTION: gstaudioaggregator
  28  * @short_description: manages a set of pads with the purpose of
  29  * aggregating their buffers for raw audio
  30  * @see_also: #GstAggregator
  31  *
  32  */
  33
  34
  35 #ifdef HAVE_CONFIG_H
  36 #  include "config.h"
  37 #endif
  38
  39 #include "gstaudioaggregator.h"
  40
  41 #include <string.h>
  42
  43 GST_DEBUG_CATEGORY_STATIC (audio_aggregator_debug);
  44 #define GST_CAT_DEFAULT audio_aggregator_debug
  45
  46 struct _GstAudioAggregatorPadPrivate
  47 {
  48   /* All members are protected by the pad object lock */
  49
  50   GstBuffer *buffer;            /* current buffer we're mixing,
  51                                    for comparison with collect.buffer
  52                                    to see if we need to update our
  53                                    cached values. */
  54   guint position, size;
  55
  56   guint64 output_offset;        /* Sample offset in output segment relative to
  57                                    segment.start that collect.pos refers to in the
  58                                    current buffer. */
  59
  60   guint64 next_offset;          /* Next expected sample offset in the input segment
  61                                    relative to segment.start */
  62
  63   /* Last time we noticed a discont */
  64   GstClockTime discont_time;
  65
  66   /* A new unhandled segment event has been received */
  67   gboolean new_segment;
  68 };
  69
  70
  71 /*****************************************
  72  * GstAudioAggregatorPad implementation  *
  73  *****************************************/
  74 G_DEFINE_TYPE (GstAudioAggregatorPad, gst_audio_aggregator_pad,
  75     GST_TYPE_AGGREGATOR_PAD);
  76
  77 static GstFlowReturn
  78 gst_audio_aggregator_pad_flush_pad (GstAggregatorPad * aggpad,
  79     GstAggregator * aggregator);
  80
  81 static void
  82 gst_audio_aggregator_pad_finalize (GObject * object)
  83 {
  84   GstAudioAggregatorPad *pad = (GstAudioAggregatorPad *) object;
  85
  86   gst_buffer_replace (&pad->priv->buffer, NULL);
  87
  88   G_OBJECT_CLASS (gst_audio_aggregator_pad_parent_class)->finalize (object);
  89 }
  90
  91 static void
  92 gst_audio_aggregator_pad_class_init (GstAudioAggregatorPadClass * klass)
  93 {
  94   GObjectClass *gobject_class = (GObjectClass *) klass;
  95   GstAggregatorPadClass *aggpadclass = (GstAggregatorPadClass *) klass;
  96
  97   g_type_class_add_private (klass, sizeof (GstAudioAggregatorPadPrivate));
  98
  99   gobject_class->finalize = gst_audio_aggregator_pad_finalize;
 100   aggpadclass->flush = GST_DEBUG_FUNCPTR (gst_audio_aggregator_pad_flush_pad);
 101 }
 102
 103 static void
 104 gst_audio_aggregator_pad_init (GstAudioAggregatorPad * pad)
 105 {
 106   pad->priv =
 107       G_TYPE_INSTANCE_GET_PRIVATE (pad, GST_TYPE_AUDIO_AGGREGATOR_PAD,
 108       GstAudioAggregatorPadPrivate);
 109
 110   gst_audio_info_init (&pad->info);
 111
 112   pad->priv->buffer = NULL;
 113   pad->priv->position = 0;
 114   pad->priv->size = 0;
 115   pad->priv->output_offset = -1;
 116   pad->priv->next_offset = -1;
 117   pad->priv->discont_time = GST_CLOCK_TIME_NONE;
 118 }
 119
 120
 121 static GstFlowReturn
 122 gst_audio_aggregator_pad_flush_pad (GstAggregatorPad * aggpad,
 123     GstAggregator * aggregator)
 124 {
 125   GstAudioAggregatorPad *pad = GST_AUDIO_AGGREGATOR_PAD (aggpad);
 126
 127   GST_OBJECT_LOCK (aggpad);
 128   pad->priv->position = pad->priv->size = 0;
 129   pad->priv->output_offset = pad->priv->next_offset = -1;
 130   pad->priv->discont_time = GST_CLOCK_TIME_NONE;
 131   gst_buffer_replace (&pad->priv->buffer, NULL);
 132   GST_OBJECT_UNLOCK (aggpad);
 133
 134   return GST_FLOW_OK;
 135 }
 136
 137
 138
 139 /**************************************
 140  * GstAudioAggregator implementation  *
 141  **************************************/
 142
 143 struct _GstAudioAggregatorPrivate
 144 {
 145   GMutex mutex;
 146
 147   /* All three properties are unprotected, can't be modified while streaming */
 148   /* Size in frames that is output per buffer */
 149   GstClockTime output_buffer_duration;
 150   GstClockTime alignment_threshold;
 151   GstClockTime discont_wait;
 152
 153   /* Protected by srcpad stream clock */
 154   /* Buffer starting at offset containing block_size frames */
 155   GstBuffer *current_buffer;
 156
 157   /* counters to keep track of timestamps */
 158   /* Readable with object lock, writable with both aag lock and object lock */
 159
 160   gint64 offset;                /* Sample offset starting from 0 at segment.start */
 161 };
 162
 163 #define GST_AUDIO_AGGREGATOR_LOCK(self)   g_mutex_lock (&(self)->priv->mutex);
 164 #define GST_AUDIO_AGGREGATOR_UNLOCK(self) g_mutex_unlock (&(self)->priv->mutex);
 165
 166 static void gst_audio_aggregator_set_property (GObject * object, guint prop_id,
 167     const GValue * value, GParamSpec * pspec);
 168 static void gst_audio_aggregator_get_property (GObject * object, guint prop_id,
 169     GValue * value, GParamSpec * pspec);
 170 static void gst_audio_aggregator_dispose (GObject * object);
 171
 172 static gboolean gst_audio_aggregator_src_event (GstAggregator * agg,
 173     GstEvent * event);
 174 static gboolean gst_audio_aggregator_sink_event (GstAggregator * agg,
 175     GstAggregatorPad * aggpad, GstEvent * event);
 176 static gboolean gst_audio_aggregator_src_query (GstAggregator * agg,
 177     GstQuery * query);
 178 static gboolean gst_audio_aggregator_start (GstAggregator * agg);
 179 static gboolean gst_audio_aggregator_stop (GstAggregator * agg);
 180 static GstFlowReturn gst_audio_aggregator_flush (GstAggregator * agg);
 181
 182 static GstBuffer *gst_audio_aggregator_create_output_buffer (GstAudioAggregator
 183     * aagg, guint num_frames);
 184 static GstBuffer *gst_audio_aggregator_do_clip (GstAggregator * agg,
 185     GstAggregatorPad * bpad, GstBuffer * buffer);
 186 static GstFlowReturn gst_audio_aggregator_aggregate (GstAggregator * agg,
 187     gboolean timeout);
 188 static gboolean sync_pad_values (GstAudioAggregator * aagg,
 189     GstAudioAggregatorPad * pad);
 190 static gboolean gst_audio_aggregator_negotiated_src_caps (GstAggregator * agg,
 191     GstCaps * caps);
 192
 193 #define DEFAULT_OUTPUT_BUFFER_DURATION (10 * GST_MSECOND)
 194 #define DEFAULT_ALIGNMENT_THRESHOLD   (40 * GST_MSECOND)
 195 #define DEFAULT_DISCONT_WAIT (1 * GST_SECOND)
 196
 197 enum
 198 {
 199   PROP_0,
 200   PROP_OUTPUT_BUFFER_DURATION,
 201   PROP_ALIGNMENT_THRESHOLD,
 202   PROP_DISCONT_WAIT,
 203 };
 204
 205 G_DEFINE_ABSTRACT_TYPE (GstAudioAggregator, gst_audio_aggregator,
 206     GST_TYPE_AGGREGATOR);
 207
 208 static GstClockTime
 209 gst_audio_aggregator_get_next_time (GstAggregator * agg)
 210 {
 211   GstClockTime next_time;
 212
 213   GST_OBJECT_LOCK (agg);
 214   if (agg->segment.position == -1 || agg->segment.position < agg->segment.start)
 215     next_time = agg->segment.start;
 216   else
 217     next_time = agg->segment.position;
 218
 219   if (agg->segment.stop != -1 && next_time > agg->segment.stop)
 220     next_time = agg->segment.stop;
 221
 222   next_time =
 223       gst_segment_to_running_time (&agg->segment, GST_FORMAT_TIME, next_time);
 224   GST_OBJECT_UNLOCK (agg);
 225
 226   return next_time;
 227 }
 228
 229 static void
 230 gst_audio_aggregator_class_init (GstAudioAggregatorClass * klass)
 231 {
 232   GObjectClass *gobject_class = (GObjectClass *) klass;
 233   GstAggregatorClass *gstaggregator_class = (GstAggregatorClass *) klass;
 234
 235   g_type_class_add_private (klass, sizeof (GstAudioAggregatorPrivate));
 236
 237   gobject_class->set_property = gst_audio_aggregator_set_property;
 238   gobject_class->get_property = gst_audio_aggregator_get_property;
 239   gobject_class->dispose = gst_audio_aggregator_dispose;
 240
 241   gstaggregator_class->src_event =
 242       GST_DEBUG_FUNCPTR (gst_audio_aggregator_src_event);
 243   gstaggregator_class->sink_event =
 244       GST_DEBUG_FUNCPTR (gst_audio_aggregator_sink_event);
 245   gstaggregator_class->src_query =
 246       GST_DEBUG_FUNCPTR (gst_audio_aggregator_src_query);
 247   gstaggregator_class->start = gst_audio_aggregator_start;
 248   gstaggregator_class->stop = gst_audio_aggregator_stop;
 249   gstaggregator_class->flush = gst_audio_aggregator_flush;
 250   gstaggregator_class->aggregate =
 251       GST_DEBUG_FUNCPTR (gst_audio_aggregator_aggregate);
 252   gstaggregator_class->clip = GST_DEBUG_FUNCPTR (gst_audio_aggregator_do_clip);
 253   gstaggregator_class->get_next_time = gst_audio_aggregator_get_next_time;
 254   gstaggregator_class->negotiated_src_caps =
 255       gst_audio_aggregator_negotiated_src_caps;
 256
 257   klass->create_output_buffer = gst_audio_aggregator_create_output_buffer;
 258
 259   GST_DEBUG_REGISTER_FUNCPTR (sync_pad_values);
 260
 261   GST_DEBUG_CATEGORY_INIT (audio_aggregator_debug, "audioaggregator",
 262       GST_DEBUG_FG_MAGENTA, "GstAudioAggregator");
 263
 264   g_object_class_install_property (gobject_class, PROP_OUTPUT_BUFFER_DURATION,
 265       g_param_spec_uint64 ("output-buffer-duration", "Output Buffer Duration",
 266           "Output block size in nanoseconds", 1,
 267           G_MAXUINT64, DEFAULT_OUTPUT_BUFFER_DURATION,
 268           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 269
 270   g_object_class_install_property (gobject_class, PROP_ALIGNMENT_THRESHOLD,
 271       g_param_spec_uint64 ("alignment-threshold", "Alignment Threshold",
 272           "Timestamp alignment threshold in nanoseconds", 0,
 273           G_MAXUINT64 - 1, DEFAULT_ALIGNMENT_THRESHOLD,
 274           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 275
 276   g_object_class_install_property (gobject_class, PROP_DISCONT_WAIT,
 277       g_param_spec_uint64 ("discont-wait", "Discont Wait",
 278           "Window of time in nanoseconds to wait before "
 279           "creating a discontinuity", 0,
 280           G_MAXUINT64 - 1, DEFAULT_DISCONT_WAIT,
 281           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 282 }
 283
 284 static void
 285 gst_audio_aggregator_init (GstAudioAggregator * aagg)
 286 {
 287   aagg->priv =
 288       G_TYPE_INSTANCE_GET_PRIVATE (aagg, GST_TYPE_AUDIO_AGGREGATOR,
 289       GstAudioAggregatorPrivate);
 290
 291   g_mutex_init (&aagg->priv->mutex);
 292
 293   aagg->priv->output_buffer_duration = DEFAULT_OUTPUT_BUFFER_DURATION;
 294   aagg->priv->alignment_threshold = DEFAULT_ALIGNMENT_THRESHOLD;
 295   aagg->priv->discont_wait = DEFAULT_DISCONT_WAIT;
 296
 297   aagg->current_caps = NULL;
 298   gst_audio_info_init (&aagg->info);
 299
 300   gst_aggregator_set_latency (GST_AGGREGATOR (aagg),
 301       aagg->priv->output_buffer_duration, aagg->priv->output_buffer_duration);
 302 }
 303
 304 static void
 305 gst_audio_aggregator_dispose (GObject * object)
 306 {
 307   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (object);
 308
 309   gst_caps_replace (&aagg->current_caps, NULL);
 310
 311   g_mutex_clear (&aagg->priv->mutex);
 312
 313   G_OBJECT_CLASS (gst_audio_aggregator_parent_class)->dispose (object);
 314 }
 315
 316 static void
 317 gst_audio_aggregator_set_property (GObject * object, guint prop_id,
 318     const GValue * value, GParamSpec * pspec)
 319 {
 320   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (object);
 321
 322   switch (prop_id) {
 323     case PROP_OUTPUT_BUFFER_DURATION:
 324       aagg->priv->output_buffer_duration = g_value_get_uint64 (value);
 325       gst_aggregator_set_latency (GST_AGGREGATOR (aagg),
 326           aagg->priv->output_buffer_duration,
 327           aagg->priv->output_buffer_duration);
 328       break;
 329     case PROP_ALIGNMENT_THRESHOLD:
 330       aagg->priv->alignment_threshold = g_value_get_uint64 (value);
 331       break;
 332     case PROP_DISCONT_WAIT:
 333       aagg->priv->discont_wait = g_value_get_uint64 (value);
 334       break;
 335     default:
 336       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 337       break;
 338   }
 339 }
 340
 341 static void
 342 gst_audio_aggregator_get_property (GObject * object, guint prop_id,
 343     GValue * value, GParamSpec * pspec)
 344 {
 345   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (object);
 346
 347   switch (prop_id) {
 348     case PROP_OUTPUT_BUFFER_DURATION:
 349       g_value_set_uint64 (value, aagg->priv->output_buffer_duration);
 350       break;
 351     case PROP_ALIGNMENT_THRESHOLD:
 352       g_value_set_uint64 (value, aagg->priv->alignment_threshold);
 353       break;
 354     case PROP_DISCONT_WAIT:
 355       g_value_set_uint64 (value, aagg->priv->discont_wait);
 356       break;
 357     default:
 358       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 359       break;
 360   }
 361 }
 362
 363
 364 /* event handling */
 365
 366 static gboolean
 367 gst_audio_aggregator_src_event (GstAggregator * agg, GstEvent * event)
 368 {
 369   gboolean result;
 370
 371   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 372   GST_DEBUG_OBJECT (agg->srcpad, "Got %s event on src pad",
 373       GST_EVENT_TYPE_NAME (event));
 374
 375   switch (GST_EVENT_TYPE (event)) {
 376     case GST_EVENT_QOS:
 377       /* QoS might be tricky */
 378       gst_event_unref (event);
 379       return FALSE;
 380     case GST_EVENT_NAVIGATION:
 381       /* navigation is rather pointless. */
 382       gst_event_unref (event);
 383       return FALSE;
 384       break;
 385     case GST_EVENT_SEEK:
 386     {
 387       GstSeekFlags flags;
 388       gdouble rate;
 389       GstSeekType start_type, stop_type;
 390       gint64 start, stop;
 391       GstFormat seek_format, dest_format;
 392
 393       /* parse the seek parameters */
 394       gst_event_parse_seek (event, &rate, &seek_format, &flags, &start_type,
 395           &start, &stop_type, &stop);
 396
 397       /* Check the seeking parametters before linking up */
 398       if ((start_type != GST_SEEK_TYPE_NONE)
 399           && (start_type != GST_SEEK_TYPE_SET)) {
 400         result = FALSE;
 401         GST_DEBUG_OBJECT (aagg,
 402             "seeking failed, unhandled seek type for start: %d", start_type);
 403         goto done;
 404       }
 405       if ((stop_type != GST_SEEK_TYPE_NONE) && (stop_type != GST_SEEK_TYPE_SET)) {
 406         result = FALSE;
 407         GST_DEBUG_OBJECT (aagg,
 408             "seeking failed, unhandled seek type for end: %d", stop_type);
 409         goto done;
 410       }
 411
 412       GST_OBJECT_LOCK (agg);
 413       dest_format = agg->segment.format;
 414       GST_OBJECT_UNLOCK (agg);
 415       if (seek_format != dest_format) {
 416         result = FALSE;
 417         GST_DEBUG_OBJECT (aagg,
 418             "seeking failed, unhandled seek format: %s",
 419             gst_format_get_name (seek_format));
 420         goto done;
 421       }
 422     }
 423       break;
 424     default:
 425       break;
 426   }
 427
 428   return
 429       GST_AGGREGATOR_CLASS (gst_audio_aggregator_parent_class)->src_event (agg,
 430       event);
 431
 432 done:
 433   return result;
 434 }
 435
 436
 437 static gboolean
 438 gst_audio_aggregator_sink_event (GstAggregator * agg,
 439     GstAggregatorPad * aggpad, GstEvent * event)
 440 {
 441   gboolean res = TRUE;
 442
 443   GST_DEBUG_OBJECT (aggpad, "Got %s event on sink pad",
 444       GST_EVENT_TYPE_NAME (event));
 445
 446   switch (GST_EVENT_TYPE (event)) {
 447     case GST_EVENT_SEGMENT:
 448     {
 449       const GstSegment *segment;
 450       gst_event_parse_segment (event, &segment);
 451
 452       if (segment->format != GST_FORMAT_TIME) {
 453         GST_ERROR_OBJECT (agg, "Segment of type %s are not supported,"
 454             " only TIME segments are supported",
 455             gst_format_get_name (segment->format));
 456         gst_event_unref (event);
 457         event = NULL;
 458         res = FALSE;
 459         break;
 460       }
 461
 462       GST_OBJECT_LOCK (agg);
 463       if (segment->rate != agg->segment.rate) {
 464         GST_ERROR_OBJECT (aggpad,
 465             "Got segment event with wrong rate %lf, expected %lf",
 466             segment->rate, agg->segment.rate);
 467         res = FALSE;
 468         gst_event_unref (event);
 469         event = NULL;
 470       } else if (segment->rate < 0.0) {
 471         GST_ERROR_OBJECT (aggpad, "Negative rates not supported yet");
 472         res = FALSE;
 473         gst_event_unref (event);
 474         event = NULL;
 475       } else {
 476         GstAudioAggregatorPad *pad = GST_AUDIO_AGGREGATOR_PAD (aggpad);
 477
 478         GST_OBJECT_LOCK (pad);
 479         pad->priv->new_segment = TRUE;
 480         GST_OBJECT_UNLOCK (pad);
 481       }
 482       GST_OBJECT_UNLOCK (agg);
 483
 484       break;
 485     }
 486     default:
 487       break;
 488   }
 489
 490   if (event != NULL)
 491     return
 492         GST_AGGREGATOR_CLASS (gst_audio_aggregator_parent_class)->sink_event
 493         (agg, aggpad, event);
 494
 495   return res;
 496 }
 497
 498 /* FIXME, the duration query should reflect how long you will produce
 499  * data, that is the amount of stream time until you will emit EOS.
 500  *
 501  * For synchronized mixing this is always the max of all the durations
 502  * of upstream since we emit EOS when all of them finished.
 503  *
 504  * We don't do synchronized mixing so this really depends on where the
 505  * streams where punched in and what their relative offsets are against
 506  * eachother which we can get from the first timestamps we see.
 507  *
 508  * When we add a new stream (or remove a stream) the duration might
 509  * also become invalid again and we need to post a new DURATION
 510  * message to notify this fact to the parent.
 511  * For now we take the max of all the upstream elements so the simple
 512  * cases work at least somewhat.
 513  */
 514 static gboolean
 515 gst_audio_aggregator_query_duration (GstAudioAggregator * aagg,
 516     GstQuery * query)
 517 {
 518   gint64 max;
 519   gboolean res;
 520   GstFormat format;
 521   GstIterator *it;
 522   gboolean done;
 523   GValue item = { 0, };
 524
 525   /* parse format */
 526   gst_query_parse_duration (query, &format, NULL);
 527
 528   max = -1;
 529   res = TRUE;
 530   done = FALSE;
 531
 532   it = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (aagg));
 533   while (!done) {
 534     GstIteratorResult ires;
 535
 536     ires = gst_iterator_next (it, &item);
 537     switch (ires) {
 538       case GST_ITERATOR_DONE:
 539         done = TRUE;
 540         break;
 541       case GST_ITERATOR_OK:
 542       {
 543         GstPad *pad = g_value_get_object (&item);
 544         gint64 duration;
 545
 546         /* ask sink peer for duration */
 547         res &= gst_pad_peer_query_duration (pad, format, &duration);
 548         /* take max from all valid return values */
 549         if (res) {
 550           /* valid unknown length, stop searching */
 551           if (duration == -1) {
 552             max = duration;
 553             done = TRUE;
 554           }
 555           /* else see if bigger than current max */
 556           else if (duration > max)
 557             max = duration;
 558         }
 559         g_value_reset (&item);
 560         break;
 561       }
 562       case GST_ITERATOR_RESYNC:
 563         max = -1;
 564         res = TRUE;
 565         gst_iterator_resync (it);
 566         break;
 567       default:
 568         res = FALSE;
 569         done = TRUE;
 570         break;
 571     }
 572   }
 573   g_value_unset (&item);
 574   gst_iterator_free (it);
 575
 576   if (res) {
 577     /* and store the max */
 578     GST_DEBUG_OBJECT (aagg, "Total duration in format %s: %"
 579         GST_TIME_FORMAT, gst_format_get_name (format), GST_TIME_ARGS (max));
 580     gst_query_set_duration (query, format, max);
 581   }
 582
 583   return res;
 584 }
 585
 586
 587 static gboolean
 588 gst_audio_aggregator_src_query (GstAggregator * agg, GstQuery * query)
 589 {
 590   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 591   gboolean res = FALSE;
 592
 593   switch (GST_QUERY_TYPE (query)) {
 594     case GST_QUERY_DURATION:
 595       res = gst_audio_aggregator_query_duration (aagg, query);
 596       break;
 597     case GST_QUERY_POSITION:
 598     {
 599       GstFormat format;
 600
 601       gst_query_parse_position (query, &format, NULL);
 602
 603       GST_OBJECT_LOCK (aagg);
 604
 605       switch (format) {
 606         case GST_FORMAT_TIME:
 607           gst_query_set_position (query, format,
 608               gst_segment_to_stream_time (&agg->segment, GST_FORMAT_TIME,
 609                   agg->segment.position));
 610           res = TRUE;
 611           break;
 612         case GST_FORMAT_BYTES:
 613           if (GST_AUDIO_INFO_BPF (&aagg->info)) {
 614             gst_query_set_position (query, format, aagg->priv->offset *
 615                 GST_AUDIO_INFO_BPF (&aagg->info));
 616             res = TRUE;
 617           }
 618           break;
 619         case GST_FORMAT_DEFAULT:
 620           gst_query_set_position (query, format, aagg->priv->offset);
 621           res = TRUE;
 622           break;
 623         default:
 624           break;
 625       }
 626
 627       GST_OBJECT_UNLOCK (aagg);
 628
 629       break;
 630     }
 631     default:
 632       res =
 633           GST_AGGREGATOR_CLASS (gst_audio_aggregator_parent_class)->src_query
 634           (agg, query);
 635       break;
 636   }
 637
 638   return res;
 639 }
 640
 641
 642 void
 643 gst_audio_aggregator_set_sink_caps (GstAudioAggregator * aagg,
 644     GstAudioAggregatorPad * pad, GstCaps * caps)
 645 {
 646 #ifndef G_DISABLE_ASSERT
 647   gboolean valid;
 648
 649   GST_OBJECT_LOCK (pad);
 650   valid = gst_audio_info_from_caps (&pad->info, caps);
 651   g_assert (valid);
 652   GST_OBJECT_UNLOCK (pad);
 653 #else
 654   GST_OBJECT_LOCK (pad);
 655   (void) gst_audio_info_from_caps (&pad->info, caps);
 656   GST_OBJECT_UNLOCK (pad);
 657 #endif
 658 }
 659
 660
 661 static gboolean
 662 gst_audio_aggregator_negotiated_src_caps (GstAggregator * agg, GstCaps * caps)
 663 {
 664   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 665   GstAudioInfo info;
 666
 667   if (!gst_audio_info_from_caps (&info, caps)) {
 668     GST_WARNING_OBJECT (aagg, "Rejecting invalid caps: %" GST_PTR_FORMAT, caps);
 669     return FALSE;
 670   }
 671
 672   GST_AUDIO_AGGREGATOR_LOCK (aagg);
 673   GST_OBJECT_LOCK (aagg);
 674
 675   if (!gst_audio_info_is_equal (&info, &aagg->info)) {
 676     GST_INFO_OBJECT (aagg, "setting caps to %" GST_PTR_FORMAT, caps);
 677     gst_caps_replace (&aagg->current_caps, caps);
 678
 679     memcpy (&aagg->info, &info, sizeof (info));
 680   }
 681
 682   GST_OBJECT_UNLOCK (aagg);
 683   GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
 684
 685   /* send caps event later, after stream-start event */
 686
 687   return
 688       GST_AGGREGATOR_CLASS
 689       (gst_audio_aggregator_parent_class)->negotiated_src_caps (agg, caps);
 690 }
 691
 692
 693 /* Must hold object lock and aagg lock to call */
 694
 695 static void
 696 gst_audio_aggregator_reset (GstAudioAggregator * aagg)
 697 {
 698   GstAggregator *agg = GST_AGGREGATOR (aagg);
 699
 700   GST_AUDIO_AGGREGATOR_LOCK (aagg);
 701   GST_OBJECT_LOCK (aagg);
 702   agg->segment.position = -1;
 703   aagg->priv->offset = -1;
 704   gst_audio_info_init (&aagg->info);
 705   gst_caps_replace (&aagg->current_caps, NULL);
 706   gst_buffer_replace (&aagg->priv->current_buffer, NULL);
 707   GST_OBJECT_UNLOCK (aagg);
 708   GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
 709 }
 710
 711 static gboolean
 712 gst_audio_aggregator_start (GstAggregator * agg)
 713 {
 714   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 715
 716   gst_audio_aggregator_reset (aagg);
 717
 718   return TRUE;
 719 }
 720
 721 static gboolean
 722 gst_audio_aggregator_stop (GstAggregator * agg)
 723 {
 724   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 725
 726   gst_audio_aggregator_reset (aagg);
 727
 728   return TRUE;
 729 }
 730
 731 static GstFlowReturn
 732 gst_audio_aggregator_flush (GstAggregator * agg)
 733 {
 734   GstAudioAggregator *aagg = GST_AUDIO_AGGREGATOR (agg);
 735
 736   GST_AUDIO_AGGREGATOR_LOCK (aagg);
 737   GST_OBJECT_LOCK (aagg);
 738   agg->segment.position = -1;
 739   aagg->priv->offset = -1;
 740   gst_buffer_replace (&aagg->priv->current_buffer, NULL);
 741   GST_OBJECT_UNLOCK (aagg);
 742   GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
 743
 744   return GST_FLOW_OK;
 745 }
 746
 747 static GstBuffer *
 748 gst_audio_aggregator_do_clip (GstAggregator * agg,
 749     GstAggregatorPad * bpad, GstBuffer * buffer)
 750 {
 751   GstAudioAggregatorPad *pad = GST_AUDIO_AGGREGATOR_PAD (bpad);
 752   gint rate, bpf;
 753
 754   rate = GST_AUDIO_INFO_RATE (&pad->info);
 755   bpf = GST_AUDIO_INFO_BPF (&pad->info);
 756
 757   GST_OBJECT_LOCK (bpad);
 758   buffer = gst_audio_buffer_clip (buffer, &bpad->segment, rate, bpf);
 759   GST_OBJECT_UNLOCK (bpad);
 760
 761   return buffer;
 762 }
 763
 764 /* Called with the object lock for both the element and pad held,
 765  * as well as the aagg lock
 766  */
 767 static gboolean
 768 gst_audio_aggregator_fill_buffer (GstAudioAggregator * aagg,
 769     GstAudioAggregatorPad * pad, GstBuffer * inbuf)
 770 {
 771   GstClockTime start_time, end_time;
 772   gboolean discont = FALSE;
 773   guint64 start_offset, end_offset;
 774   gint rate, bpf;
 775
 776   GstAggregator *agg = GST_AGGREGATOR (aagg);
 777   GstAggregatorPad *aggpad = GST_AGGREGATOR_PAD (pad);
 778
 779   g_assert (pad->priv->buffer == NULL);
 780
 781   rate = GST_AUDIO_INFO_RATE (&pad->info);
 782   bpf = GST_AUDIO_INFO_BPF (&pad->info);
 783
 784   pad->priv->position = 0;
 785   pad->priv->size = gst_buffer_get_size (inbuf) / bpf;
 786
 787   if (!GST_BUFFER_PTS_IS_VALID (inbuf)) {
 788     if (pad->priv->output_offset == -1)
 789       pad->priv->output_offset = aagg->priv->offset;
 790     if (pad->priv->next_offset == -1)
 791       pad->priv->next_offset = pad->priv->size;
 792     else
 793       pad->priv->next_offset += pad->priv->size;
 794     goto done;
 795   }
 796
 797   start_time = GST_BUFFER_PTS (inbuf);
 798   end_time =
 799       start_time + gst_util_uint64_scale_ceil (pad->priv->size, GST_SECOND,
 800       rate);
 801
 802   /* Clipping should've ensured this */
 803   g_assert (start_time >= aggpad->segment.start);
 804
 805   start_offset =
 806       gst_util_uint64_scale (start_time - aggpad->segment.start, rate,
 807       GST_SECOND);
 808   end_offset = start_offset + pad->priv->size;
 809
 810   if (GST_BUFFER_IS_DISCONT (inbuf)
 811       || GST_BUFFER_FLAG_IS_SET (inbuf, GST_BUFFER_FLAG_RESYNC)
 812       || pad->priv->new_segment || pad->priv->next_offset == -1) {
 813     discont = TRUE;
 814     pad->priv->new_segment = FALSE;
 815   } else {
 816     guint64 diff, max_sample_diff;
 817
 818     /* Check discont, based on audiobasesink */
 819     if (start_offset <= pad->priv->next_offset)
 820       diff = pad->priv->next_offset - start_offset;
 821     else
 822       diff = start_offset - pad->priv->next_offset;
 823
 824     max_sample_diff =
 825         gst_util_uint64_scale_int (aagg->priv->alignment_threshold, rate,
 826         GST_SECOND);
 827
 828     /* Discont! */
 829     if (G_UNLIKELY (diff >= max_sample_diff)) {
 830       if (aagg->priv->discont_wait > 0) {
 831         if (pad->priv->discont_time == GST_CLOCK_TIME_NONE) {
 832           pad->priv->discont_time = start_time;
 833         } else if (start_time - pad->priv->discont_time >=
 834             aagg->priv->discont_wait) {
 835           discont = TRUE;
 836           pad->priv->discont_time = GST_CLOCK_TIME_NONE;
 837         }
 838       } else {
 839         discont = TRUE;
 840       }
 841     } else if (G_UNLIKELY (pad->priv->discont_time != GST_CLOCK_TIME_NONE)) {
 842       /* we have had a discont, but are now back on track! */
 843       pad->priv->discont_time = GST_CLOCK_TIME_NONE;
 844     }
 845   }
 846
 847   if (discont) {
 848     /* Have discont, need resync */
 849     if (pad->priv->next_offset != -1)
 850       GST_DEBUG_OBJECT (pad, "Have discont. Expected %"
 851           G_GUINT64_FORMAT ", got %" G_GUINT64_FORMAT,
 852           pad->priv->next_offset, start_offset);
 853     pad->priv->output_offset = -1;
 854     pad->priv->next_offset = end_offset;
 855   } else {
 856     pad->priv->next_offset += pad->priv->size;
 857   }
 858
 859   if (pad->priv->output_offset == -1) {
 860     GstClockTime start_running_time;
 861     GstClockTime end_running_time;
 862     guint64 start_output_offset;
 863     guint64 end_output_offset;
 864
 865     start_running_time =
 866         gst_segment_to_running_time (&aggpad->segment,
 867         GST_FORMAT_TIME, start_time);
 868     end_running_time =
 869         gst_segment_to_running_time (&aggpad->segment,
 870         GST_FORMAT_TIME, end_time);
 871
 872     /* Convert to position in the output segment */
 873     start_output_offset =
 874         gst_segment_position_from_running_time (&agg->segment, GST_FORMAT_TIME,
 875         start_running_time);
 876     if (start_output_offset != -1)
 877       start_output_offset =
 878           gst_util_uint64_scale (start_output_offset - agg->segment.start, rate,
 879           GST_SECOND);
 880
 881     end_output_offset =
 882         gst_segment_position_from_running_time (&agg->segment, GST_FORMAT_TIME,
 883         end_running_time);
 884     if (end_output_offset != -1)
 885       end_output_offset =
 886           gst_util_uint64_scale (end_output_offset - agg->segment.start, rate,
 887           GST_SECOND);
 888
 889     if (start_output_offset == -1 && end_output_offset == -1) {
 890       /* Outside output segment, drop */
 891       gst_buffer_unref (inbuf);
 892       pad->priv->buffer = NULL;
 893       pad->priv->position = 0;
 894       pad->priv->size = 0;
 895       pad->priv->output_offset = -1;
 896       GST_DEBUG_OBJECT (pad, "Buffer outside output segment");
 897       return FALSE;
 898     }
 899
 900     /* Calculate end_output_offset if it was outside the output segment */
 901     if (end_output_offset == -1)
 902       end_output_offset = start_output_offset + pad->priv->size;
 903
 904     if (end_output_offset < aagg->priv->offset) {
 905       /* Before output segment, drop */
 906       gst_buffer_unref (inbuf);
 907       pad->priv->buffer = NULL;
 908       pad->priv->position = 0;
 909       pad->priv->size = 0;
 910       pad->priv->output_offset = -1;
 911       GST_DEBUG_OBJECT (pad,
 912           "Buffer before segment or current position: %" G_GUINT64_FORMAT " < %"
 913           G_GINT64_FORMAT, end_output_offset, aagg->priv->offset);
 914       return FALSE;
 915     }
 916
 917     if (start_output_offset == -1 || start_output_offset < aagg->priv->offset) {
 918       guint diff;
 919
 920       if (start_output_offset == -1 && end_output_offset < pad->priv->size) {
 921         diff = pad->priv->size - end_output_offset + aagg->priv->offset;
 922       } else if (start_output_offset == -1) {
 923         start_output_offset = end_output_offset - pad->priv->size;
 924
 925         if (start_output_offset < aagg->priv->offset)
 926           diff = aagg->priv->offset - start_output_offset;
 927         else
 928           diff = 0;
 929       } else {
 930         diff = aagg->priv->offset - start_output_offset;
 931       }
 932
 933       pad->priv->position += diff;
 934       if (pad->priv->position >= pad->priv->size) {
 935         /* Empty buffer, drop */
 936         gst_buffer_unref (inbuf);
 937         pad->priv->buffer = NULL;
 938         pad->priv->position = 0;
 939         pad->priv->size = 0;
 940         pad->priv->output_offset = -1;
 941         GST_DEBUG_OBJECT (pad,
 942             "Buffer before segment or current position: %" G_GUINT64_FORMAT
 943             " < %" G_GINT64_FORMAT, end_output_offset, aagg->priv->offset);
 944         return FALSE;
 945       }
 946     }
 947
 948     if (start_output_offset == -1 || start_output_offset < aagg->priv->offset)
 949       pad->priv->output_offset = aagg->priv->offset;
 950     else
 951       pad->priv->output_offset = start_output_offset;
 952
 953     GST_DEBUG_OBJECT (pad,
 954         "Buffer resynced: Pad offset %" G_GUINT64_FORMAT
 955         ", current audio aggregator offset %" G_GINT64_FORMAT,
 956         pad->priv->output_offset, aagg->priv->offset);
 957   }
 958
 959 done:
 960
 961   GST_LOG_OBJECT (pad,
 962       "Queued new buffer at offset %" G_GUINT64_FORMAT,
 963       pad->priv->output_offset);
 964   pad->priv->buffer = inbuf;
 965
 966   return TRUE;
 967 }
 968
 969 /* Called with pad object lock held */
 970
 971 static gboolean
 972 gst_audio_aggregator_mix_buffer (GstAudioAggregator * aagg,
 973     GstAudioAggregatorPad * pad, GstBuffer * inbuf, GstBuffer * outbuf)
 974 {
 975   guint overlap;
 976   guint out_start;
 977   gboolean filled;
 978   guint blocksize;
 979   guint in_offset;
 980   gboolean pad_changed = FALSE;
 981
 982   blocksize = gst_util_uint64_scale (aagg->priv->output_buffer_duration,
 983       GST_AUDIO_INFO_RATE (&aagg->info), GST_SECOND);
 984   blocksize = MAX (1, blocksize);
 985
 986   /* Overlap => mix */
 987   if (aagg->priv->offset < pad->priv->output_offset)
 988     out_start = pad->priv->output_offset - aagg->priv->offset;
 989   else
 990     out_start = 0;
 991
 992   overlap = pad->priv->size - pad->priv->position;
 993   if (overlap > blocksize - out_start)
 994     overlap = blocksize - out_start;
 995
 996   if (GST_BUFFER_FLAG_IS_SET (inbuf, GST_BUFFER_FLAG_GAP)) {
 997     /* skip gap buffer */
 998     GST_LOG_OBJECT (pad, "skipping GAP buffer");
 999     pad->priv->output_offset += pad->priv->size - pad->priv->position;
1000     pad->priv->position = pad->priv->size;
1001
1002     gst_buffer_replace (&pad->priv->buffer, NULL);
1003     return FALSE;
1004   }
1005
1006   gst_buffer_ref (inbuf);
1007   in_offset = pad->priv->position;
1008   GST_OBJECT_UNLOCK (pad);
1009   GST_OBJECT_UNLOCK (aagg);
1010
1011   filled = GST_AUDIO_AGGREGATOR_GET_CLASS (aagg)->aggregate_one_buffer (aagg,
1012       pad, inbuf, in_offset, outbuf, out_start, overlap);
1013
1014   GST_OBJECT_LOCK (aagg);
1015   GST_OBJECT_LOCK (pad);
1016
1017   pad_changed = (inbuf != pad->priv->buffer);
1018   gst_buffer_unref (inbuf);
1019
1020   if (filled)
1021     GST_BUFFER_FLAG_UNSET (outbuf, GST_BUFFER_FLAG_GAP);
1022
1023   if (pad_changed)
1024     return FALSE;
1025
1026   pad->priv->position += overlap;
1027   pad->priv->output_offset += overlap;
1028
1029   if (pad->priv->position == pad->priv->size) {
1030     /* Buffer done, drop it */
1031     gst_buffer_replace (&pad->priv->buffer, NULL);
1032     GST_LOG_OBJECT (pad, "Finished mixing buffer, waiting for next");
1033     return FALSE;
1034   }
1035
1036   return TRUE;
1037 }
1038
1039 static GstBuffer *
1040 gst_audio_aggregator_create_output_buffer (GstAudioAggregator * aagg,
1041     guint num_frames)
1042 {
1043   GstAllocator *allocator;
1044   GstAllocationParams params;
1045   GstBuffer *outbuf;
1046   GstMapInfo outmap;
1047
1048   gst_aggregator_get_allocator (GST_AGGREGATOR (aagg), &allocator, &params);
1049
1050   outbuf = gst_buffer_new_allocate (allocator, num_frames *
1051       GST_AUDIO_INFO_BPF (&aagg->info), &params);
1052
1053   if (allocator)
1054     gst_object_unref (allocator);
1055
1056   gst_buffer_map (outbuf, &outmap, GST_MAP_WRITE);
1057   gst_audio_format_fill_silence (aagg->info.finfo, outmap.data, outmap.size);
1058   gst_buffer_unmap (outbuf, &outmap);
1059
1060   return outbuf;
1061 }
1062
1063 static gboolean
1064 sync_pad_values (GstAudioAggregator * aagg, GstAudioAggregatorPad * pad)
1065 {
1066   GstAggregatorPad *bpad = GST_AGGREGATOR_PAD (pad);
1067   GstClockTime timestamp, stream_time;
1068
1069   if (pad->priv->buffer == NULL)
1070     return TRUE;
1071
1072   timestamp = GST_BUFFER_PTS (pad->priv->buffer);
1073   GST_OBJECT_LOCK (bpad);
1074   stream_time = gst_segment_to_stream_time (&bpad->segment, GST_FORMAT_TIME,
1075       timestamp);
1076   GST_OBJECT_UNLOCK (bpad);
1077
1078   /* sync object properties on stream time */
1079   /* TODO: Ideally we would want to do that on every sample */
1080   if (GST_CLOCK_TIME_IS_VALID (stream_time))
1081     gst_object_sync_values (GST_OBJECT (pad), stream_time);
1082
1083   return TRUE;
1084 }
1085
1086 static GstFlowReturn
1087 gst_audio_aggregator_aggregate (GstAggregator * agg, gboolean timeout)
1088 {
1089   /* Get all pads that have data for us and store them in a
1090    * new list.
1091    *
1092    * Calculate the current output offset/timestamp and
1093    * offset_end/timestamp_end. Allocate a silence buffer
1094    * for this and store it.
1095    *
1096    * For all pads:
1097    * 1) Once per input buffer (cached)
1098    *   1) Check discont (flag and timestamp with tolerance)
1099    *   2) If discont or new, resync. That means:
1100    *     1) Drop all start data of the buffer that comes before
1101    *        the current position/offset.
1102    *     2) Calculate the offset (output segment!) that the first
1103    *        frame of the input buffer corresponds to. Base this on
1104    *        the running time.
1105    *
1106    * 2) If the current pad's offset/offset_end overlaps with the output
1107    *    offset/offset_end, mix it at the appropiate position in the output
1108    *    buffer and advance the pad's position. Remember if this pad needs
1109    *    a new buffer to advance behind the output offset_end.
1110    *
1111    * 3) If we had no pad with a buffer, go EOS.
1112    *
1113    * 4) If we had at least one pad that did not advance behind output
1114    *    offset_end, let collected be called again for the current
1115    *    output offset/offset_end.
1116    */
1117   GstElement *element;
1118   GstAudioAggregator *aagg;
1119   GList *iter;
1120   GstFlowReturn ret;
1121   GstBuffer *outbuf = NULL;
1122   gint64 next_offset;
1123   gint64 next_timestamp;
1124   gint rate, bpf;
1125   gboolean dropped = FALSE;
1126   gboolean is_eos = TRUE;
1127   gboolean is_done = TRUE;
1128   guint blocksize;
1129
1130   element = GST_ELEMENT (agg);
1131   aagg = GST_AUDIO_AGGREGATOR (agg);
1132
1133   /* Sync pad properties to the stream time */
1134   gst_aggregator_iterate_sinkpads (agg,
1135       (GstAggregatorPadForeachFunc) sync_pad_values, NULL);
1136
1137   GST_AUDIO_AGGREGATOR_LOCK (aagg);
1138   GST_OBJECT_LOCK (agg);
1139
1140   /* Update position from the segment start/stop if needed */
1141   if (agg->segment.position == -1) {
1142     if (agg->segment.rate > 0.0)
1143       agg->segment.position = agg->segment.start;
1144     else
1145       agg->segment.position = agg->segment.stop;
1146   }
1147
1148   if (G_UNLIKELY (aagg->info.finfo->format == GST_AUDIO_FORMAT_UNKNOWN)) {
1149     if (timeout) {
1150       GST_DEBUG_OBJECT (aagg,
1151           "Got timeout before receiving any caps, don't output anything");
1152
1153       /* Advance position */
1154       if (agg->segment.rate > 0.0)
1155         agg->segment.position += aagg->priv->output_buffer_duration;
1156       else if (agg->segment.position > aagg->priv->output_buffer_duration)
1157         agg->segment.position -= aagg->priv->output_buffer_duration;
1158       else
1159         agg->segment.position = 0;
1160
1161       GST_OBJECT_UNLOCK (agg);
1162       GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1163       return GST_AGGREGATOR_FLOW_NEED_DATA;
1164     } else {
1165       GST_OBJECT_UNLOCK (agg);
1166       goto not_negotiated;
1167     }
1168   }
1169
1170   rate = GST_AUDIO_INFO_RATE (&aagg->info);
1171   bpf = GST_AUDIO_INFO_BPF (&aagg->info);
1172
1173   if (aagg->priv->offset == -1) {
1174     aagg->priv->offset =
1175         gst_util_uint64_scale (agg->segment.position - agg->segment.start, rate,
1176         GST_SECOND);
1177     GST_DEBUG_OBJECT (aagg, "Starting at offset %" G_GINT64_FORMAT,
1178         aagg->priv->offset);
1179   }
1180
1181   blocksize = gst_util_uint64_scale (aagg->priv->output_buffer_duration,
1182       rate, GST_SECOND);
1183   blocksize = MAX (1, blocksize);
1184
1185   /* for the next timestamp, use the sample counter, which will
1186    * never accumulate rounding errors */
1187
1188   /* FIXME: Reverse mixing does not work at all yet */
1189   if (agg->segment.rate > 0.0) {
1190     next_offset = aagg->priv->offset + blocksize;
1191   } else {
1192     next_offset = aagg->priv->offset - blocksize;
1193   }
1194
1195   next_timestamp =
1196       agg->segment.start + gst_util_uint64_scale (next_offset, GST_SECOND,
1197       rate);
1198
1199   if (aagg->priv->current_buffer == NULL) {
1200     GST_OBJECT_UNLOCK (agg);
1201     aagg->priv->current_buffer =
1202         GST_AUDIO_AGGREGATOR_GET_CLASS (aagg)->create_output_buffer (aagg,
1203         blocksize);
1204     /* Be careful, some things could have changed ? */
1205     GST_OBJECT_LOCK (agg);
1206     GST_BUFFER_FLAG_SET (aagg->priv->current_buffer, GST_BUFFER_FLAG_GAP);
1207   }
1208   outbuf = aagg->priv->current_buffer;
1209
1210   GST_LOG_OBJECT (agg,
1211       "Starting to mix %u samples for offset %" G_GINT64_FORMAT
1212       " with timestamp %" GST_TIME_FORMAT, blocksize,
1213       aagg->priv->offset, GST_TIME_ARGS (agg->segment.position));
1214
1215   for (iter = element->sinkpads; iter; iter = iter->next) {
1216     GstBuffer *inbuf;
1217     GstAudioAggregatorPad *pad = (GstAudioAggregatorPad *) iter->data;
1218     GstAggregatorPad *aggpad = (GstAggregatorPad *) iter->data;
1219     gboolean drop_buf = FALSE;
1220     gboolean pad_eos = gst_aggregator_pad_is_eos (aggpad);
1221
1222     if (!pad_eos)
1223       is_eos = FALSE;
1224
1225     inbuf = gst_aggregator_pad_get_buffer (aggpad);
1226
1227     GST_OBJECT_LOCK (pad);
1228     if (!inbuf) {
1229       if (timeout) {
1230         if (pad->priv->output_offset < next_offset) {
1231           gint64 diff = next_offset - pad->priv->output_offset;
1232           GST_DEBUG_OBJECT (pad, "Timeout, missing %" G_GINT64_FORMAT
1233               " frames (%" GST_TIME_FORMAT ")", diff,
1234               GST_TIME_ARGS (gst_util_uint64_scale (diff, GST_SECOND,
1235                       GST_AUDIO_INFO_RATE (&aagg->info))));
1236         }
1237       } else if (!pad_eos) {
1238         is_done = FALSE;
1239       }
1240       GST_OBJECT_UNLOCK (pad);
1241       continue;
1242     }
1243
1244     g_assert (!pad->priv->buffer || pad->priv->buffer == inbuf);
1245
1246     /* New buffer? */
1247     if (!pad->priv->buffer) {
1248       /* Takes ownership of buffer */
1249       if (!gst_audio_aggregator_fill_buffer (aagg, pad, inbuf)) {
1250         dropped = TRUE;
1251         GST_OBJECT_UNLOCK (pad);
1252         gst_aggregator_pad_drop_buffer (aggpad);
1253         continue;
1254       }
1255     } else {
1256       gst_buffer_unref (inbuf);
1257     }
1258
1259     if (!pad->priv->buffer && !dropped && pad_eos) {
1260       GST_DEBUG_OBJECT (aggpad, "Pad is in EOS state");
1261       GST_OBJECT_UNLOCK (pad);
1262       continue;
1263     }
1264
1265     g_assert (pad->priv->buffer);
1266
1267     /* This pad is lacking behind, we need to update the offset
1268      * and maybe drop the current buffer */
1269     if (pad->priv->output_offset < aagg->priv->offset) {
1270       gint64 diff = aagg->priv->offset - pad->priv->output_offset;
1271       gint64 odiff = diff;
1272
1273       if (pad->priv->position + diff > pad->priv->size)
1274         diff = pad->priv->size - pad->priv->position;
1275       pad->priv->position += diff;
1276       pad->priv->output_offset += diff;
1277
1278       if (pad->priv->position == pad->priv->size) {
1279         GST_DEBUG_OBJECT (pad, "Buffer was late by %" GST_TIME_FORMAT
1280             ", dropping %" GST_PTR_FORMAT,
1281             GST_TIME_ARGS (gst_util_uint64_scale (odiff, GST_SECOND,
1282                     GST_AUDIO_INFO_RATE (&aagg->info))), pad->priv->buffer);
1283         /* Buffer done, drop it */
1284         gst_buffer_replace (&pad->priv->buffer, NULL);
1285         dropped = TRUE;
1286         GST_OBJECT_UNLOCK (pad);
1287         gst_aggregator_pad_drop_buffer (aggpad);
1288         continue;
1289       }
1290     }
1291
1292
1293     if (pad->priv->output_offset >= aagg->priv->offset
1294         && pad->priv->output_offset <
1295         aagg->priv->offset + blocksize && pad->priv->buffer) {
1296       GST_LOG_OBJECT (aggpad, "Mixing buffer for current offset");
1297       drop_buf = !gst_audio_aggregator_mix_buffer (aagg, pad, pad->priv->buffer,
1298           outbuf);
1299       if (pad->priv->output_offset >= next_offset) {
1300         GST_LOG_OBJECT (pad,
1301             "Pad is at or after current offset: %" G_GUINT64_FORMAT " >= %"
1302             G_GINT64_FORMAT, pad->priv->output_offset, next_offset);
1303       } else {
1304         is_done = FALSE;
1305       }
1306     }
1307
1308     GST_OBJECT_UNLOCK (pad);
1309     if (drop_buf)
1310       gst_aggregator_pad_drop_buffer (aggpad);
1311
1312   }
1313   GST_OBJECT_UNLOCK (agg);
1314
1315   if (dropped) {
1316     /* We dropped a buffer, retry */
1317     GST_LOG_OBJECT (aagg, "A pad dropped a buffer, wait for the next one");
1318     GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1319     return GST_AGGREGATOR_FLOW_NEED_DATA;
1320   }
1321
1322   if (!is_done && !is_eos) {
1323     /* Get more buffers */
1324     GST_LOG_OBJECT (aagg,
1325         "We're not done yet for the current offset, waiting for more data");
1326     GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1327     return GST_AGGREGATOR_FLOW_NEED_DATA;
1328   }
1329
1330   if (is_eos) {
1331     gint64 max_offset = 0;
1332
1333     GST_DEBUG_OBJECT (aagg, "We're EOS");
1334
1335     GST_OBJECT_LOCK (agg);
1336     for (iter = GST_ELEMENT (agg)->sinkpads; iter; iter = iter->next) {
1337       GstAudioAggregatorPad *pad = GST_AUDIO_AGGREGATOR_PAD (iter->data);
1338
1339       max_offset = MAX ((gint64) max_offset, (gint64) pad->priv->output_offset);
1340     }
1341     GST_OBJECT_UNLOCK (agg);
1342
1343     /* This means EOS or nothing mixed in at all */
1344     if (aagg->priv->offset == max_offset) {
1345       gst_buffer_replace (&aagg->priv->current_buffer, NULL);
1346       GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1347       return GST_FLOW_EOS;
1348     }
1349
1350     if (max_offset <= next_offset) {
1351       GST_DEBUG_OBJECT (aagg,
1352           "Last buffer is incomplete: %" G_GUINT64_FORMAT " <= %"
1353           G_GINT64_FORMAT, max_offset, next_offset);
1354       next_offset = max_offset;
1355       next_timestamp =
1356           agg->segment.start + gst_util_uint64_scale (next_offset, GST_SECOND,
1357           rate);
1358
1359       if (next_offset > aagg->priv->offset)
1360         gst_buffer_resize (outbuf, 0, (next_offset - aagg->priv->offset) * bpf);
1361     }
1362   }
1363
1364   /* set timestamps on the output buffer */
1365   GST_OBJECT_LOCK (agg);
1366   if (agg->segment.rate > 0.0) {
1367     GST_BUFFER_PTS (outbuf) = agg->segment.position;
1368     GST_BUFFER_OFFSET (outbuf) = aagg->priv->offset;
1369     GST_BUFFER_OFFSET_END (outbuf) = next_offset;
1370     GST_BUFFER_DURATION (outbuf) = next_timestamp - agg->segment.position;
1371   } else {
1372     GST_BUFFER_PTS (outbuf) = next_timestamp;
1373     GST_BUFFER_OFFSET (outbuf) = next_offset;
1374     GST_BUFFER_OFFSET_END (outbuf) = aagg->priv->offset;
1375     GST_BUFFER_DURATION (outbuf) = agg->segment.position - next_timestamp;
1376   }
1377
1378   GST_OBJECT_UNLOCK (agg);
1379
1380   /* send it out */
1381   GST_LOG_OBJECT (aagg,
1382       "pushing outbuf %p, timestamp %" GST_TIME_FORMAT " offset %"
1383       G_GINT64_FORMAT, outbuf, GST_TIME_ARGS (GST_BUFFER_PTS (outbuf)),
1384       GST_BUFFER_OFFSET (outbuf));
1385
1386   GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1387
1388   ret = gst_aggregator_finish_buffer (agg, aagg->priv->current_buffer);
1389   aagg->priv->current_buffer = NULL;
1390
1391   GST_LOG_OBJECT (aagg, "pushed outbuf, result = %s", gst_flow_get_name (ret));
1392
1393   GST_AUDIO_AGGREGATOR_LOCK (aagg);
1394   GST_OBJECT_LOCK (agg);
1395   aagg->priv->offset = next_offset;
1396   agg->segment.position = next_timestamp;
1397
1398   /* If there was a timeout and there was a gap in data in out of the streams,
1399    * then it's a very good time to for a resync with the timestamps.
1400    */
1401   if (timeout) {
1402     for (iter = element->sinkpads; iter; iter = iter->next) {
1403       GstAudioAggregatorPad *pad = GST_AUDIO_AGGREGATOR_PAD (iter->data);
1404
1405       GST_OBJECT_LOCK (pad);
1406       if (pad->priv->output_offset < aagg->priv->offset)
1407         pad->priv->output_offset = -1;
1408       GST_OBJECT_UNLOCK (pad);
1409     }
1410   }
1411   GST_OBJECT_UNLOCK (agg);
1412   GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1413
1414   return ret;
1415   /* ERRORS */
1416 not_negotiated:
1417   {
1418     GST_AUDIO_AGGREGATOR_UNLOCK (aagg);
1419     GST_ELEMENT_ERROR (aagg, STREAM, FORMAT, (NULL),
1420         ("Unknown data received, not negotiated"));
1421     return GST_FLOW_NOT_NEGOTIATED;
1422   }
1423 }