plugins/elements/gstconcat.c

   1 /* GStreamer concat element
   2  *
   3  *  Copyright (c) 2014 Sebastian Dröge <sebastian@centricular.com>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  *
  20  */
  21 /**
  22  * SECTION:element-concat
  23  * @see_also: #GstFunnel
  24  *
  25  * Concatenates streams together to one continous stream.
  26  *
  27  * All streams but the current one are blocked until the current one
  28  * finished with %GST_EVENT_EOS. Then the next stream is enabled, while
  29  * keeping the running time continous for %GST_FORMAT_TIME segments or
  30  * keeping the segment continous for %GST_FORMAT_BYTES segments.
  31  *
  32  * Streams are switched in the order in which the sinkpads were requested.
  33  *
  34  * By default, the stream segment's base values are adjusted to ensure
  35  * the segment transitions between streams are continuous. In some cases,
  36  * it may be desirable to turn off these adjustments (for example, because
  37  * another downstream element like a streamsynchronizer adjusts the base
  38  * values on its own). The adjust-value property can be used for this purpose.
  39  *
  40  * <refsect2>
  41  * <title>Example launch line</title>
  42  * |[
  43  * gst-launch-1.0 concat name=c ! xvimagesink  videotestsrc num-buffers=100 ! c.   videotestsrc num-buffers=100 pattern=ball ! c.
  44  * ]| Plays two video streams one after another.
  45  * </refsect2>
  46  */
  47
  48 #ifdef HAVE_CONFIG_H
  49 #include "config.h"
  50 #endif
  51
  52 #include "gstconcat.h"
  53
  54 GST_DEBUG_CATEGORY_STATIC (gst_concat_debug);
  55 #define GST_CAT_DEFAULT gst_concat_debug
  56
  57 G_GNUC_INTERNAL GType gst_concat_pad_get_type (void);
  58
  59 #define GST_TYPE_CONCAT_PAD (gst_concat_pad_get_type())
  60 #define GST_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_CONCAT_PAD, GstConcatPad))
  61 #define GST_CONCAT_PAD_CAST(obj) ((GstConcatPad *)(obj))
  62 #define GST_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_CONCAT_PAD, GstConcatPadClass))
  63 #define GST_IS_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_CONCAT_PAD))
  64 #define GST_IS_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_CONCAT_PAD))
  65
  66 typedef struct _GstConcatPad GstConcatPad;
  67 typedef struct _GstConcatPadClass GstConcatPadClass;
  68
  69 struct _GstConcatPad
  70 {
  71   GstPad parent;
  72
  73   GstSegment segment;
  74
  75   /* Protected by the concat lock */
  76   gboolean flushing;
  77 };
  78
  79 struct _GstConcatPadClass
  80 {
  81   GstPadClass parent;
  82 };
  83
  84 G_DEFINE_TYPE (GstConcatPad, gst_concat_pad, GST_TYPE_PAD);
  85
  86 static void
  87 gst_concat_pad_class_init (GstConcatPadClass * klass)
  88 {
  89 }
  90
  91 static void
  92 gst_concat_pad_init (GstConcatPad * self)
  93 {
  94   gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
  95   self->flushing = FALSE;
  96 }
  97
  98 static GstStaticPadTemplate concat_sink_template =
  99 GST_STATIC_PAD_TEMPLATE ("sink_%u",
 100     GST_PAD_SINK,
 101     GST_PAD_REQUEST,
 102     GST_STATIC_CAPS_ANY);
 103
 104 static GstStaticPadTemplate concat_src_template =
 105 GST_STATIC_PAD_TEMPLATE ("src",
 106     GST_PAD_SRC,
 107     GST_PAD_ALWAYS,
 108     GST_STATIC_CAPS_ANY);
 109
 110 enum
 111 {
 112   PROP_0,
 113   PROP_ACTIVE_PAD,
 114   PROP_ADJUST_BASE
 115 };
 116
 117 #define DEFAULT_ADJUST_BASE TRUE
 118
 119 #define _do_init \
 120   GST_DEBUG_CATEGORY_INIT (gst_concat_debug, "concat", 0, "concat element");
 121 #define gst_concat_parent_class parent_class
 122 G_DEFINE_TYPE_WITH_CODE (GstConcat, gst_concat, GST_TYPE_ELEMENT, _do_init);
 123
 124 static void gst_concat_dispose (GObject * object);
 125 static void gst_concat_finalize (GObject * object);
 126 static void gst_concat_get_property (GObject * object,
 127     guint prop_id, GValue * value, GParamSpec * pspec);
 128 static void gst_concat_set_property (GObject * object,
 129     guint prop_id, const GValue * value, GParamSpec * pspec);
 130
 131 static GstStateChangeReturn gst_concat_change_state (GstElement * element,
 132     GstStateChange transition);
 133 static GstPad *gst_concat_request_new_pad (GstElement * element,
 134     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
 135 static void gst_concat_release_pad (GstElement * element, GstPad * pad);
 136
 137 static GstFlowReturn gst_concat_sink_chain (GstPad * pad, GstObject * parent,
 138     GstBuffer * buffer);
 139 static gboolean gst_concat_sink_event (GstPad * pad, GstObject * parent,
 140     GstEvent * event);
 141 static gboolean gst_concat_sink_query (GstPad * pad, GstObject * parent,
 142     GstQuery * query);
 143
 144 static gboolean gst_concat_src_event (GstPad * pad, GstObject * parent,
 145     GstEvent * event);
 146 static gboolean gst_concat_src_query (GstPad * pad, GstObject * parent,
 147     GstQuery * query);
 148
 149 static gboolean gst_concat_switch_pad (GstConcat * self);
 150
 151 static void gst_concat_notify_active_pad (GstConcat * self);
 152
 153 static GParamSpec *pspec_active_pad = NULL;
 154
 155 static void
 156 gst_concat_class_init (GstConcatClass * klass)
 157 {
 158   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 159   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 160
 161   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_concat_dispose);
 162   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_concat_finalize);
 163
 164   gobject_class->get_property = gst_concat_get_property;
 165   gobject_class->set_property = gst_concat_set_property;
 166
 167   pspec_active_pad = g_param_spec_object ("active-pad", "Active pad",
 168       "Currently active src pad", GST_TYPE_PAD, G_PARAM_READABLE |
 169       G_PARAM_STATIC_STRINGS);
 170   g_object_class_install_property (gobject_class, PROP_ACTIVE_PAD,
 171       pspec_active_pad);
 172   g_object_class_install_property (gobject_class, PROP_ADJUST_BASE,
 173       g_param_spec_boolean ("adjust-base", "Adjust segment base",
 174           "Adjust the base value of segments to ensure they are adjacent",
 175           DEFAULT_ADJUST_BASE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 176
 177   gst_element_class_set_static_metadata (gstelement_class,
 178       "Concat", "Generic", "Concatenate multiple streams",
 179       "Sebastian Dröge <sebastian@centricular.com>");
 180
 181   gst_element_class_add_pad_template (gstelement_class,
 182       gst_static_pad_template_get (&concat_sink_template));
 183   gst_element_class_add_pad_template (gstelement_class,
 184       gst_static_pad_template_get (&concat_src_template));
 185
 186   gstelement_class->request_new_pad =
 187       GST_DEBUG_FUNCPTR (gst_concat_request_new_pad);
 188   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_concat_release_pad);
 189   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_concat_change_state);
 190 }
 191
 192 static void
 193 gst_concat_init (GstConcat * self)
 194 {
 195   g_mutex_init (&self->lock);
 196   g_cond_init (&self->cond);
 197
 198   self->srcpad = gst_pad_new_from_static_template (&concat_src_template, "src");
 199   gst_pad_set_event_function (self->srcpad,
 200       GST_DEBUG_FUNCPTR (gst_concat_src_event));
 201   gst_pad_set_query_function (self->srcpad,
 202       GST_DEBUG_FUNCPTR (gst_concat_src_query));
 203   gst_pad_use_fixed_caps (self->srcpad);
 204
 205   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
 206
 207   self->adjust_base = DEFAULT_ADJUST_BASE;
 208 }
 209
 210 static void
 211 gst_concat_dispose (GObject * object)
 212 {
 213   GstConcat *self = GST_CONCAT (object);
 214   GList *item;
 215
 216   gst_object_replace ((GstObject **) & self->current_sinkpad, NULL);
 217
 218 restart:
 219   for (item = GST_ELEMENT_PADS (object); item; item = g_list_next (item)) {
 220     GstPad *pad = GST_PAD (item->data);
 221
 222     if (GST_PAD_IS_SINK (pad)) {
 223       gst_element_release_request_pad (GST_ELEMENT (object), pad);
 224       goto restart;
 225     }
 226   }
 227
 228   G_OBJECT_CLASS (parent_class)->dispose (object);
 229 }
 230
 231 static void
 232 gst_concat_finalize (GObject * object)
 233 {
 234   GstConcat *self = GST_CONCAT (object);
 235
 236   g_mutex_clear (&self->lock);
 237   g_cond_clear (&self->cond);
 238
 239   G_OBJECT_CLASS (parent_class)->finalize (object);
 240 }
 241
 242 static void
 243 gst_concat_get_property (GObject * object, guint prop_id, GValue * value,
 244     GParamSpec * pspec)
 245 {
 246   GstConcat *self = GST_CONCAT (object);
 247
 248   switch (prop_id) {
 249     case PROP_ACTIVE_PAD:{
 250       g_mutex_lock (&self->lock);
 251       g_value_set_object (value, self->current_sinkpad);
 252       g_mutex_unlock (&self->lock);
 253       break;
 254     }
 255     case PROP_ADJUST_BASE:{
 256       g_mutex_lock (&self->lock);
 257       g_value_set_boolean (value, self->adjust_base);
 258       g_mutex_unlock (&self->lock);
 259       break;
 260     }
 261     default:
 262       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 263       break;
 264   }
 265 }
 266
 267 static void
 268 gst_concat_set_property (GObject * object, guint prop_id, const GValue * value,
 269     GParamSpec * pspec)
 270 {
 271   GstConcat *self = GST_CONCAT (object);
 272
 273   switch (prop_id) {
 274     case PROP_ADJUST_BASE:{
 275       g_mutex_lock (&self->lock);
 276       self->adjust_base = g_value_get_boolean (value);
 277       g_mutex_unlock (&self->lock);
 278       break;
 279     }
 280     default:
 281       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 282       break;
 283   }
 284 }
 285
 286 static GstPad *
 287 gst_concat_request_new_pad (GstElement * element, GstPadTemplate * templ,
 288     const gchar * name, const GstCaps * caps)
 289 {
 290   GstConcat *self = GST_CONCAT (element);
 291   GstPad *sinkpad;
 292   gchar *pad_name;
 293   gboolean do_notify = FALSE;
 294
 295   GST_DEBUG_OBJECT (element, "requesting pad");
 296
 297   g_mutex_lock (&self->lock);
 298   pad_name = g_strdup_printf ("sink_%u", self->pad_count);
 299   self->pad_count++;
 300   g_mutex_unlock (&self->lock);
 301
 302   sinkpad = GST_PAD_CAST (g_object_new (GST_TYPE_CONCAT_PAD,
 303           "name", pad_name, "direction", templ->direction, "template", templ,
 304           NULL));
 305   g_free (pad_name);
 306
 307   gst_pad_set_chain_function (sinkpad,
 308       GST_DEBUG_FUNCPTR (gst_concat_sink_chain));
 309   gst_pad_set_event_function (sinkpad,
 310       GST_DEBUG_FUNCPTR (gst_concat_sink_event));
 311   gst_pad_set_query_function (sinkpad,
 312       GST_DEBUG_FUNCPTR (gst_concat_sink_query));
 313   GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_CAPS);
 314   GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_ALLOCATION);
 315
 316   gst_pad_set_active (sinkpad, TRUE);
 317
 318   g_mutex_lock (&self->lock);
 319   self->sinkpads = g_list_prepend (self->sinkpads, gst_object_ref (sinkpad));
 320   if (!self->current_sinkpad) {
 321     do_notify = TRUE;
 322     self->current_sinkpad = gst_object_ref (sinkpad);
 323   }
 324   g_mutex_unlock (&self->lock);
 325
 326   gst_element_add_pad (element, sinkpad);
 327
 328   if (do_notify)
 329     gst_concat_notify_active_pad (self);
 330
 331   return sinkpad;
 332 }
 333
 334 static void
 335 gst_concat_release_pad (GstElement * element, GstPad * pad)
 336 {
 337   GstConcat *self = GST_CONCAT (element);
 338   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 339   GList *l;
 340   gboolean current_pad_removed = FALSE;
 341   gboolean eos = FALSE;
 342   gboolean do_notify = FALSE;
 343
 344   GST_DEBUG_OBJECT (self, "releasing pad");
 345
 346   g_mutex_lock (&self->lock);
 347   spad->flushing = TRUE;
 348   g_cond_broadcast (&self->cond);
 349   g_mutex_unlock (&self->lock);
 350
 351   gst_pad_set_active (pad, FALSE);
 352
 353   /* Now the pad is definitely not running anymore */
 354
 355   g_mutex_lock (&self->lock);
 356   if (self->current_sinkpad == GST_PAD_CAST (spad)) {
 357     eos = ! !gst_concat_switch_pad (self);
 358     current_pad_removed = TRUE;
 359     do_notify = TRUE;
 360   }
 361
 362   for (l = self->sinkpads; l; l = l->next) {
 363     if ((gpointer) spad == l->data) {
 364       gst_object_unref (spad);
 365       self->sinkpads = g_list_delete_link (self->sinkpads, l);
 366       break;
 367     }
 368   }
 369   g_mutex_unlock (&self->lock);
 370
 371   gst_element_remove_pad (GST_ELEMENT_CAST (self), pad);
 372
 373   if (do_notify)
 374     gst_concat_notify_active_pad (self);
 375
 376   if (GST_STATE (self) > GST_STATE_READY) {
 377     if (current_pad_removed && !eos)
 378       gst_element_post_message (GST_ELEMENT_CAST (self),
 379           gst_message_new_duration_changed (GST_OBJECT_CAST (self)));
 380
 381     /* FIXME: Sending EOS from application thread */
 382     if (eos)
 383       gst_pad_push_event (self->srcpad, gst_event_new_eos ());
 384   }
 385 }
 386
 387 /* Returns FALSE if flushing
 388  * Must be called from the pad's streaming thread
 389  */
 390 static gboolean
 391 gst_concat_pad_wait (GstConcatPad * spad, GstConcat * self)
 392 {
 393   g_mutex_lock (&self->lock);
 394   if (spad->flushing) {
 395     g_mutex_unlock (&self->lock);
 396     GST_DEBUG_OBJECT (spad, "Flushing");
 397     return FALSE;
 398   }
 399
 400   while (spad != GST_CONCAT_PAD_CAST (self->current_sinkpad)) {
 401     GST_TRACE_OBJECT (spad, "Not the current sinkpad - waiting");
 402     g_cond_wait (&self->cond, &self->lock);
 403     if (spad->flushing) {
 404       g_mutex_unlock (&self->lock);
 405       GST_DEBUG_OBJECT (spad, "Flushing");
 406       return FALSE;
 407     }
 408   }
 409   /* This pad can only become not the current sinkpad from
 410    * a) This streaming thread (we hold the stream lock)
 411    * b) Releasing the pad (takes the stream lock, see above)
 412    *
 413    * Unlocking here is thus safe and we can safely push
 414    * serialized data to our srcpad
 415    */
 416   GST_DEBUG_OBJECT (spad, "Now the current sinkpad");
 417   g_mutex_unlock (&self->lock);
 418
 419   return TRUE;
 420 }
 421
 422 static GstFlowReturn
 423 gst_concat_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 424 {
 425   GstFlowReturn ret;
 426   GstConcat *self = GST_CONCAT (parent);
 427   GstConcatPad *spad = GST_CONCAT_PAD (pad);
 428
 429   GST_LOG_OBJECT (pad, "received buffer %p", buffer);
 430
 431   if (!gst_concat_pad_wait (spad, self))
 432     return GST_FLOW_FLUSHING;
 433
 434   if (self->last_stop == GST_CLOCK_TIME_NONE)
 435     self->last_stop = spad->segment.start;
 436
 437   if (self->format == GST_FORMAT_TIME) {
 438     GstClockTime start_time = GST_BUFFER_TIMESTAMP (buffer);
 439     GstClockTime end_time = GST_CLOCK_TIME_NONE;
 440
 441     if (start_time != GST_CLOCK_TIME_NONE)
 442       end_time = start_time;
 443     if (GST_BUFFER_DURATION_IS_VALID (buffer))
 444       end_time += GST_BUFFER_DURATION (buffer);
 445
 446     if (end_time != GST_CLOCK_TIME_NONE && end_time > self->last_stop)
 447       self->last_stop = end_time;
 448   } else {
 449     self->last_stop += gst_buffer_get_size (buffer);
 450   }
 451
 452   ret = gst_pad_push (self->srcpad, buffer);
 453
 454   GST_LOG_OBJECT (pad, "handled buffer %s", gst_flow_get_name (ret));
 455
 456   return ret;
 457 }
 458
 459 /* Returns FALSE if no further pad, must be called with concat lock */
 460 static gboolean
 461 gst_concat_switch_pad (GstConcat * self)
 462 {
 463   GList *l;
 464   gboolean next;
 465   GstSegment segment;
 466   gint64 last_stop;
 467
 468   segment = GST_CONCAT_PAD (self->current_sinkpad)->segment;
 469
 470   last_stop = self->last_stop;
 471   if (last_stop == GST_CLOCK_TIME_NONE)
 472     last_stop = segment.stop;
 473   if (last_stop == GST_CLOCK_TIME_NONE)
 474     last_stop = segment.start;
 475   g_assert (last_stop != GST_CLOCK_TIME_NONE);
 476
 477   if (last_stop > segment.stop)
 478     last_stop = segment.stop;
 479
 480   if (segment.format == GST_FORMAT_TIME)
 481     last_stop =
 482         gst_segment_to_running_time (&segment, segment.format, last_stop);
 483   else
 484     last_stop += segment.start;
 485
 486   self->current_start_offset += last_stop;
 487
 488   for (l = self->sinkpads; l; l = l->next) {
 489     if ((gpointer) self->current_sinkpad == l->data) {
 490       l = l->prev;
 491       GST_DEBUG_OBJECT (self,
 492           "Switching from pad %" GST_PTR_FORMAT " to %" GST_PTR_FORMAT,
 493           self->current_sinkpad, l ? l->data : NULL);
 494       gst_object_unref (self->current_sinkpad);
 495       self->current_sinkpad = l ? gst_object_ref (l->data) : NULL;
 496       g_cond_broadcast (&self->cond);
 497       break;
 498     }
 499   }
 500
 501   next = self->current_sinkpad != NULL;
 502
 503   self->last_stop = GST_CLOCK_TIME_NONE;
 504
 505   return next;
 506 }
 507
 508 static void
 509 gst_concat_notify_active_pad (GstConcat * self)
 510 {
 511   g_object_notify_by_pspec ((GObject *) self, pspec_active_pad);
 512 }
 513
 514 static gboolean
 515 gst_concat_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 516 {
 517   GstConcat *self = GST_CONCAT (parent);
 518   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 519   gboolean ret = TRUE;
 520
 521   GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
 522
 523   switch (GST_EVENT_TYPE (event)) {
 524     case GST_EVENT_STREAM_START:{
 525       if (!gst_concat_pad_wait (spad, self)) {
 526         ret = FALSE;
 527         gst_event_unref (event);
 528       } else {
 529         ret = gst_pad_event_default (pad, parent, event);
 530       }
 531       break;
 532     }
 533     case GST_EVENT_SEGMENT:{
 534       gboolean adjust_base;
 535
 536       /* Drop segment event, we create our own one */
 537       gst_event_copy_segment (event, &spad->segment);
 538       gst_event_unref (event);
 539
 540       g_mutex_lock (&self->lock);
 541       adjust_base = self->adjust_base;
 542       if (self->format == GST_FORMAT_UNDEFINED) {
 543         if (spad->segment.format != GST_FORMAT_TIME
 544             && spad->segment.format != GST_FORMAT_BYTES) {
 545           g_mutex_unlock (&self->lock);
 546           GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
 547               ("Can only operate in TIME or BYTES format"));
 548           ret = FALSE;
 549           break;
 550         }
 551         self->format = spad->segment.format;
 552         GST_DEBUG_OBJECT (self, "Operating in %s format",
 553             gst_format_get_name (self->format));
 554         g_mutex_unlock (&self->lock);
 555       } else if (self->format != spad->segment.format) {
 556         g_mutex_unlock (&self->lock);
 557         GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
 558             ("Operating in %s format but new pad has %s",
 559                 gst_format_get_name (self->format),
 560                 gst_format_get_name (spad->segment.format)));
 561         ret = FALSE;
 562       } else {
 563         g_mutex_unlock (&self->lock);
 564       }
 565
 566       if (!gst_concat_pad_wait (spad, self)) {
 567         ret = FALSE;
 568       } else {
 569         GstSegment segment = spad->segment;
 570
 571         if (adjust_base) {
 572           /* We know no duration */
 573           segment.duration = -1;
 574
 575           /* Update segment values to be continous with last stream */
 576           if (self->format == GST_FORMAT_TIME) {
 577             segment.base += self->current_start_offset;
 578           } else {
 579             /* Shift start/stop byte position */
 580             segment.start += self->current_start_offset;
 581             if (segment.stop != -1)
 582               segment.stop += self->current_start_offset;
 583           }
 584         }
 585
 586         gst_pad_push_event (self->srcpad, gst_event_new_segment (&segment));
 587       }
 588       break;
 589     }
 590     case GST_EVENT_EOS:{
 591       gst_event_unref (event);
 592
 593       if (!gst_concat_pad_wait (spad, self)) {
 594         ret = FALSE;
 595       } else {
 596         gboolean next;
 597
 598         g_mutex_lock (&self->lock);
 599         next = gst_concat_switch_pad (self);
 600         g_mutex_unlock (&self->lock);
 601         ret = TRUE;
 602
 603         gst_concat_notify_active_pad (self);
 604
 605         if (!next) {
 606           gst_pad_push_event (self->srcpad, gst_event_new_eos ());
 607         } else {
 608           gst_element_post_message (GST_ELEMENT_CAST (self),
 609               gst_message_new_duration_changed (GST_OBJECT_CAST (self)));
 610         }
 611       }
 612       break;
 613     }
 614     case GST_EVENT_FLUSH_START:{
 615       gboolean forward;
 616
 617       g_mutex_lock (&self->lock);
 618       spad->flushing = TRUE;
 619       g_cond_broadcast (&self->cond);
 620       forward = (self->current_sinkpad == GST_PAD_CAST (spad));
 621       g_mutex_unlock (&self->lock);
 622
 623       if (forward)
 624         ret = gst_pad_event_default (pad, parent, event);
 625       else
 626         gst_event_unref (event);
 627       break;
 628     }
 629     case GST_EVENT_FLUSH_STOP:{
 630       gboolean forward;
 631
 632       gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
 633       spad->flushing = FALSE;
 634
 635       g_mutex_lock (&self->lock);
 636       forward = (self->current_sinkpad == GST_PAD_CAST (spad));
 637       g_mutex_unlock (&self->lock);
 638
 639       if (forward) {
 640         gboolean reset_time;
 641
 642         gst_event_parse_flush_stop (event, &reset_time);
 643         if (reset_time) {
 644           GST_DEBUG_OBJECT (self,
 645               "resetting start offset to 0 after flushing with reset_time = TRUE");
 646           self->current_start_offset = 0;
 647         }
 648         ret = gst_pad_event_default (pad, parent, event);
 649       } else {
 650         gst_event_unref (event);
 651       }
 652       break;
 653     }
 654     default:{
 655       /* Wait for other serialized events before forwarding */
 656       if (GST_EVENT_IS_SERIALIZED (event) && !gst_concat_pad_wait (spad, self)) {
 657         gst_event_unref (event);
 658         ret = FALSE;
 659       } else {
 660         ret = gst_pad_event_default (pad, parent, event);
 661       }
 662       break;
 663     }
 664   }
 665
 666   return ret;
 667 }
 668
 669 static gboolean
 670 gst_concat_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
 671 {
 672   GstConcat *self = GST_CONCAT (parent);
 673   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 674   gboolean ret = TRUE;
 675
 676   GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
 677
 678   switch (GST_QUERY_TYPE (query)) {
 679     default:
 680       /* Wait for other serialized queries before forwarding */
 681       if (GST_QUERY_IS_SERIALIZED (query) && !gst_concat_pad_wait (spad, self)) {
 682         ret = FALSE;
 683       } else {
 684         ret = gst_pad_query_default (pad, parent, query);
 685       }
 686       break;
 687   }
 688
 689   return ret;
 690 }
 691
 692 static gboolean
 693 gst_concat_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 694 {
 695   GstConcat *self = GST_CONCAT (parent);
 696   gboolean ret = TRUE;
 697
 698   GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
 699
 700   switch (GST_EVENT_TYPE (event)) {
 701     case GST_EVENT_SEEK:{
 702       GstPad *sinkpad = NULL;
 703
 704       g_mutex_lock (&self->lock);
 705       if ((sinkpad = self->current_sinkpad))
 706         gst_object_ref (sinkpad);
 707       g_mutex_unlock (&self->lock);
 708       if (sinkpad) {
 709         ret = gst_pad_push_event (sinkpad, event);
 710         gst_object_unref (sinkpad);
 711       } else {
 712         gst_event_unref (event);
 713         ret = FALSE;
 714       }
 715       break;
 716     }
 717     case GST_EVENT_QOS:{
 718       GstQOSType type;
 719       GstClockTimeDiff diff;
 720       GstClockTime timestamp;
 721       gdouble proportion;
 722
 723       gst_event_parse_qos (event, &type, &proportion, &diff, &timestamp);
 724       gst_event_unref (event);
 725
 726       if (timestamp != GST_CLOCK_TIME_NONE
 727           && timestamp > self->current_start_offset) {
 728         timestamp -= self->current_start_offset;
 729         event = gst_event_new_qos (type, proportion, diff, timestamp);
 730         ret = gst_pad_push_event (self->current_sinkpad, event);
 731       } else {
 732         ret = FALSE;
 733       }
 734       break;
 735     }
 736     case GST_EVENT_FLUSH_STOP:{
 737       gboolean reset_time;
 738
 739       gst_event_parse_flush_stop (event, &reset_time);
 740       if (reset_time) {
 741         GST_DEBUG_OBJECT (self,
 742             "resetting start offset to 0 after flushing with reset_time = TRUE");
 743         self->current_start_offset = 0;
 744       }
 745
 746       ret = gst_pad_event_default (pad, parent, event);
 747       break;
 748     }
 749     default:
 750       ret = gst_pad_event_default (pad, parent, event);
 751       break;
 752   }
 753
 754   return ret;
 755 }
 756
 757 static gboolean
 758 gst_concat_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
 759 {
 760   gboolean ret = TRUE;
 761
 762   GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
 763
 764   switch (GST_QUERY_TYPE (query)) {
 765     default:
 766       ret = gst_pad_query_default (pad, parent, query);
 767       break;
 768   }
 769
 770   return ret;
 771 }
 772
 773 static void
 774 reset_pad (const GValue * data, gpointer user_data)
 775 {
 776   GstPad *pad = g_value_get_object (data);
 777   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 778
 779   gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
 780   spad->flushing = FALSE;
 781 }
 782
 783 static void
 784 unblock_pad (const GValue * data, gpointer user_data)
 785 {
 786   GstPad *pad = g_value_get_object (data);
 787   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 788
 789   spad->flushing = TRUE;
 790 }
 791
 792 static GstStateChangeReturn
 793 gst_concat_change_state (GstElement * element, GstStateChange transition)
 794 {
 795   GstConcat *self = GST_CONCAT (element);
 796   GstStateChangeReturn ret;
 797
 798   switch (transition) {
 799     case GST_STATE_CHANGE_READY_TO_PAUSED:{
 800       self->format = GST_FORMAT_UNDEFINED;
 801       self->current_start_offset = 0;
 802       self->last_stop = GST_CLOCK_TIME_NONE;
 803       break;
 804     }
 805     case GST_STATE_CHANGE_PAUSED_TO_READY:{
 806       GstIterator *iter = gst_element_iterate_sink_pads (element);
 807       GstIteratorResult res;
 808
 809       g_mutex_lock (&self->lock);
 810       do {
 811         res = gst_iterator_foreach (iter, unblock_pad, NULL);
 812       } while (res == GST_ITERATOR_RESYNC);
 813
 814       gst_iterator_free (iter);
 815       g_cond_broadcast (&self->cond);
 816       g_mutex_unlock (&self->lock);
 817
 818       if (res == GST_ITERATOR_ERROR)
 819         return GST_STATE_CHANGE_FAILURE;
 820
 821       break;
 822     }
 823     default:
 824       break;
 825   }
 826
 827   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 828   if (ret == GST_STATE_CHANGE_FAILURE)
 829     return ret;
 830
 831   switch (transition) {
 832     case GST_STATE_CHANGE_PAUSED_TO_READY:{
 833       GstIterator *iter = gst_element_iterate_sink_pads (element);
 834       GstIteratorResult res;
 835
 836       do {
 837         res = gst_iterator_foreach (iter, reset_pad, NULL);
 838       } while (res == GST_ITERATOR_RESYNC);
 839
 840       gst_iterator_free (iter);
 841
 842       if (res == GST_ITERATOR_ERROR)
 843         return GST_STATE_CHANGE_FAILURE;
 844
 845       break;
 846     }
 847     default:
 848       break;
 849   }
 850
 851   return ret;
 852 }