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