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 #include "gstcoreelementselements.h"
  54
  55 GST_DEBUG_CATEGORY_STATIC (gst_concat_debug);
  56 #define GST_CAT_DEFAULT gst_concat_debug
  57
  58 G_GNUC_INTERNAL GType gst_concat_pad_get_type (void);
  59
  60 #define GST_TYPE_CONCAT_PAD (gst_concat_pad_get_type())
  61 #define GST_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_CONCAT_PAD, GstConcatPad))
  62 #define GST_CONCAT_PAD_CAST(obj) ((GstConcatPad *)(obj))
  63 #define GST_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_CONCAT_PAD, GstConcatPadClass))
  64 #define GST_IS_CONCAT_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_CONCAT_PAD))
  65 #define GST_IS_CONCAT_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_CONCAT_PAD))
  66
  67 typedef struct _GstConcatPad GstConcatPad;
  68 typedef struct _GstConcatPadClass GstConcatPadClass;
  69
  70 struct _GstConcatPad
  71 {
  72   GstPad parent;
  73
  74   GstSegment segment;
  75
  76   /* Protected by the concat lock */
  77   gboolean flushing;
  78 };
  79
  80 struct _GstConcatPadClass
  81 {
  82   GstPadClass parent;
  83 };
  84
  85 G_DEFINE_TYPE (GstConcatPad, gst_concat_pad, GST_TYPE_PAD);
  86
  87 static void
  88 gst_concat_pad_class_init (GstConcatPadClass * klass)
  89 {
  90 }
  91
  92 static void
  93 gst_concat_pad_init (GstConcatPad * self)
  94 {
  95   gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
  96   self->flushing = FALSE;
  97 }
  98
  99 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink_%u",
 100     GST_PAD_SINK,
 101     GST_PAD_REQUEST,
 102     GST_STATIC_CAPS_ANY);
 103
 104 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
 105     GST_PAD_SRC,
 106     GST_PAD_ALWAYS,
 107     GST_STATIC_CAPS_ANY);
 108
 109 enum
 110 {
 111   PROP_0,
 112   PROP_ACTIVE_PAD,
 113   PROP_ADJUST_BASE
 114 };
 115
 116 #define DEFAULT_ADJUST_BASE TRUE
 117
 118 #define _do_init \
 119   GST_DEBUG_CATEGORY_INIT (gst_concat_debug, "concat", 0, "concat element");
 120 #define gst_concat_parent_class parent_class
 121 G_DEFINE_TYPE_WITH_CODE (GstConcat, gst_concat, GST_TYPE_ELEMENT, _do_init);
 122 GST_ELEMENT_REGISTER_DEFINE (concat, "concat", GST_RANK_NONE, GST_TYPE_CONCAT);
 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 sink 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_static_pad_template (gstelement_class, &sink_template);
 182   gst_element_class_add_static_pad_template (gstelement_class, &src_template);
 183
 184   gstelement_class->request_new_pad =
 185       GST_DEBUG_FUNCPTR (gst_concat_request_new_pad);
 186   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_concat_release_pad);
 187   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_concat_change_state);
 188 }
 189
 190 static void
 191 gst_concat_init (GstConcat * self)
 192 {
 193   g_mutex_init (&self->lock);
 194   g_cond_init (&self->cond);
 195
 196   self->srcpad = gst_pad_new_from_static_template (&src_template, "src");
 197   gst_pad_set_event_function (self->srcpad,
 198       GST_DEBUG_FUNCPTR (gst_concat_src_event));
 199   gst_pad_set_query_function (self->srcpad,
 200       GST_DEBUG_FUNCPTR (gst_concat_src_query));
 201   gst_pad_use_fixed_caps (self->srcpad);
 202
 203   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
 204
 205   self->adjust_base = DEFAULT_ADJUST_BASE;
 206 }
 207
 208 static void
 209 gst_concat_dispose (GObject * object)
 210 {
 211   GstConcat *self = GST_CONCAT (object);
 212   GList *item;
 213
 214   gst_object_replace ((GstObject **) & self->current_sinkpad, NULL);
 215
 216 restart:
 217   for (item = GST_ELEMENT_PADS (object); item; item = g_list_next (item)) {
 218     GstPad *pad = GST_PAD (item->data);
 219
 220     if (GST_PAD_IS_SINK (pad)) {
 221       gst_element_release_request_pad (GST_ELEMENT (object), pad);
 222       goto restart;
 223     }
 224   }
 225
 226   G_OBJECT_CLASS (parent_class)->dispose (object);
 227 }
 228
 229 static void
 230 gst_concat_finalize (GObject * object)
 231 {
 232   GstConcat *self = GST_CONCAT (object);
 233
 234   g_mutex_clear (&self->lock);
 235   g_cond_clear (&self->cond);
 236
 237   G_OBJECT_CLASS (parent_class)->finalize (object);
 238 }
 239
 240 static void
 241 gst_concat_get_property (GObject * object, guint prop_id, GValue * value,
 242     GParamSpec * pspec)
 243 {
 244   GstConcat *self = GST_CONCAT (object);
 245
 246   switch (prop_id) {
 247     case PROP_ACTIVE_PAD:{
 248       g_mutex_lock (&self->lock);
 249       g_value_set_object (value, self->current_sinkpad);
 250       g_mutex_unlock (&self->lock);
 251       break;
 252     }
 253     case PROP_ADJUST_BASE:{
 254       g_mutex_lock (&self->lock);
 255       g_value_set_boolean (value, self->adjust_base);
 256       g_mutex_unlock (&self->lock);
 257       break;
 258     }
 259     default:
 260       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 261       break;
 262   }
 263 }
 264
 265 static void
 266 gst_concat_set_property (GObject * object, guint prop_id, const GValue * value,
 267     GParamSpec * pspec)
 268 {
 269   GstConcat *self = GST_CONCAT (object);
 270
 271   switch (prop_id) {
 272     case PROP_ADJUST_BASE:{
 273       g_mutex_lock (&self->lock);
 274       self->adjust_base = g_value_get_boolean (value);
 275       g_mutex_unlock (&self->lock);
 276       break;
 277     }
 278     default:
 279       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 280       break;
 281   }
 282 }
 283
 284 static GstPad *
 285 gst_concat_request_new_pad (GstElement * element, GstPadTemplate * templ,
 286     const gchar * name, const GstCaps * caps)
 287 {
 288   GstConcat *self = GST_CONCAT (element);
 289   GstPad *sinkpad;
 290   gchar *pad_name;
 291   gboolean do_notify = FALSE;
 292
 293   GST_DEBUG_OBJECT (element, "requesting pad");
 294
 295   g_mutex_lock (&self->lock);
 296   pad_name = g_strdup_printf ("sink_%u", self->pad_count);
 297   self->pad_count++;
 298   g_mutex_unlock (&self->lock);
 299
 300   sinkpad = GST_PAD_CAST (g_object_new (GST_TYPE_CONCAT_PAD,
 301           "name", pad_name, "direction", templ->direction, "template", templ,
 302           NULL));
 303   g_free (pad_name);
 304
 305   gst_pad_set_chain_function (sinkpad,
 306       GST_DEBUG_FUNCPTR (gst_concat_sink_chain));
 307   gst_pad_set_event_function (sinkpad,
 308       GST_DEBUG_FUNCPTR (gst_concat_sink_event));
 309   gst_pad_set_query_function (sinkpad,
 310       GST_DEBUG_FUNCPTR (gst_concat_sink_query));
 311   GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_CAPS);
 312   GST_OBJECT_FLAG_SET (sinkpad, GST_PAD_FLAG_PROXY_ALLOCATION);
 313
 314   gst_pad_set_active (sinkpad, TRUE);
 315
 316   g_mutex_lock (&self->lock);
 317   self->sinkpads = g_list_prepend (self->sinkpads, gst_object_ref (sinkpad));
 318   if (!self->current_sinkpad) {
 319     do_notify = TRUE;
 320     self->current_sinkpad = gst_object_ref (sinkpad);
 321   }
 322   g_mutex_unlock (&self->lock);
 323
 324   gst_element_add_pad (element, sinkpad);
 325
 326   if (do_notify)
 327     gst_concat_notify_active_pad (self);
 328
 329   GST_DEBUG_OBJECT (sinkpad, "requested pad");
 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 (pad, "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     if (self->current_sinkpad == NULL && g_list_length (self->sinkpads) == 1) {
 403       GST_LOG_OBJECT (spad, "Sole pad waiting, switching");
 404       /* If we are the only sinkpad, take active pad ownership */
 405       self->current_sinkpad = gst_object_ref (self->sinkpads->data);
 406       break;
 407     }
 408     g_cond_wait (&self->cond, &self->lock);
 409     if (spad->flushing) {
 410       g_mutex_unlock (&self->lock);
 411       GST_DEBUG_OBJECT (spad, "Flushing");
 412       return FALSE;
 413     }
 414   }
 415   /* This pad can only become not the current sinkpad from
 416    * a) This streaming thread (we hold the stream lock)
 417    * b) Releasing the pad (takes the stream lock, see above)
 418    *
 419    * Unlocking here is thus safe and we can safely push
 420    * serialized data to our srcpad
 421    */
 422   GST_DEBUG_OBJECT (spad, "Now the current sinkpad");
 423   g_mutex_unlock (&self->lock);
 424
 425   return TRUE;
 426 }
 427
 428 static GstFlowReturn
 429 gst_concat_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 430 {
 431   GstFlowReturn ret;
 432   GstConcat *self = GST_CONCAT (parent);
 433   GstConcatPad *spad = GST_CONCAT_PAD (pad);
 434
 435   GST_LOG_OBJECT (pad, "received buffer %" GST_PTR_FORMAT, buffer);
 436
 437   if (!gst_concat_pad_wait (spad, self))
 438     return GST_FLOW_FLUSHING;
 439
 440   if (self->last_stop == GST_CLOCK_TIME_NONE)
 441     self->last_stop = spad->segment.start;
 442
 443   if (self->format == GST_FORMAT_TIME) {
 444     GstClockTime start_time = GST_BUFFER_TIMESTAMP (buffer);
 445     GstClockTime end_time = GST_CLOCK_TIME_NONE;
 446
 447     if (start_time != GST_CLOCK_TIME_NONE)
 448       end_time = start_time;
 449     if (GST_BUFFER_DURATION_IS_VALID (buffer))
 450       end_time += GST_BUFFER_DURATION (buffer);
 451
 452     if (end_time != GST_CLOCK_TIME_NONE && end_time > self->last_stop)
 453       self->last_stop = end_time;
 454   } else {
 455     self->last_stop += gst_buffer_get_size (buffer);
 456   }
 457
 458   ret = gst_pad_push (self->srcpad, buffer);
 459
 460   GST_LOG_OBJECT (pad, "handled buffer %s, last_stop %" GST_TIME_FORMAT,
 461       gst_flow_get_name (ret), GST_TIME_ARGS (self->last_stop));
 462
 463   return ret;
 464 }
 465
 466 /* Returns FALSE if no further pad, must be called with concat lock */
 467 static gboolean
 468 gst_concat_switch_pad (GstConcat * self)
 469 {
 470   GList *l;
 471   gboolean next;
 472   GstSegment segment;
 473   gint64 last_stop;
 474
 475   segment = GST_CONCAT_PAD (self->current_sinkpad)->segment;
 476
 477   last_stop = self->last_stop;
 478   if (last_stop == GST_CLOCK_TIME_NONE)
 479     last_stop = segment.stop;
 480   if (last_stop == GST_CLOCK_TIME_NONE)
 481     last_stop = segment.start;
 482   g_assert (last_stop != GST_CLOCK_TIME_NONE);
 483
 484   if (last_stop > segment.stop)
 485     last_stop = segment.stop;
 486
 487   if (segment.format == GST_FORMAT_TIME)
 488     last_stop =
 489         gst_segment_to_running_time (&segment, segment.format, last_stop);
 490   else
 491     last_stop += segment.start;
 492
 493   self->current_start_offset += last_stop;
 494
 495   for (l = self->sinkpads; l; l = l->next) {
 496     if ((gpointer) self->current_sinkpad == l->data) {
 497       l = l->prev;
 498       GST_DEBUG_OBJECT (self,
 499           "Switching from pad %" GST_PTR_FORMAT " to %" GST_PTR_FORMAT,
 500           self->current_sinkpad, l ? l->data : NULL);
 501       gst_object_unref (self->current_sinkpad);
 502       self->current_sinkpad = l ? gst_object_ref (l->data) : NULL;
 503       g_cond_broadcast (&self->cond);
 504       break;
 505     }
 506   }
 507
 508   next = self->current_sinkpad != NULL;
 509
 510   self->last_stop = GST_CLOCK_TIME_NONE;
 511
 512   return next;
 513 }
 514
 515 static void
 516 gst_concat_notify_active_pad (GstConcat * self)
 517 {
 518   g_object_notify_by_pspec ((GObject *) self, pspec_active_pad);
 519 }
 520
 521 static gboolean
 522 gst_concat_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 523 {
 524   GstConcat *self = GST_CONCAT (parent);
 525   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 526   gboolean ret = TRUE;
 527   gboolean adjust_base;
 528
 529   GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
 530
 531   g_mutex_lock (&self->lock);
 532   adjust_base = self->adjust_base;
 533   g_mutex_unlock (&self->lock);
 534
 535   switch (GST_EVENT_TYPE (event)) {
 536     case GST_EVENT_STREAM_START:{
 537       if (!gst_concat_pad_wait (spad, self)) {
 538         ret = FALSE;
 539         gst_event_replace (&event, NULL);
 540       }
 541       break;
 542     }
 543     case GST_EVENT_SEGMENT:{
 544       /* Drop segment event, we create our own one */
 545       gst_event_copy_segment (event, &spad->segment);
 546       gst_event_replace (&event, NULL);
 547
 548       g_mutex_lock (&self->lock);
 549       if (self->format == GST_FORMAT_UNDEFINED) {
 550         if (spad->segment.format != GST_FORMAT_TIME
 551             && spad->segment.format != GST_FORMAT_BYTES) {
 552           g_mutex_unlock (&self->lock);
 553           GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
 554               ("Can only operate in TIME or BYTES format"));
 555           ret = FALSE;
 556           break;
 557         }
 558         self->format = spad->segment.format;
 559         GST_DEBUG_OBJECT (self, "Operating in %s format",
 560             gst_format_get_name (self->format));
 561       } else if (self->format != spad->segment.format) {
 562         g_mutex_unlock (&self->lock);
 563         GST_ELEMENT_ERROR (self, CORE, FAILED, (NULL),
 564             ("Operating in %s format but new pad has %s",
 565                 gst_format_get_name (self->format),
 566                 gst_format_get_name (spad->segment.format)));
 567         ret = FALSE;
 568         break;
 569       }
 570
 571       g_mutex_unlock (&self->lock);
 572
 573       if (!gst_concat_pad_wait (spad, self)) {
 574         ret = FALSE;
 575       } else {
 576         GstSegment segment = spad->segment;
 577
 578         g_mutex_lock (&self->lock);
 579
 580         if (adjust_base) {
 581           /* We know no duration */
 582           segment.duration = -1;
 583
 584           /* Update segment values to be continuous with last stream */
 585           if (self->format == GST_FORMAT_TIME) {
 586             GST_DEBUG_OBJECT (self,
 587                 "Updating segment base %" GST_TIME_FORMAT " + %" GST_TIME_FORMAT
 588                 " = %" GST_TIME_FORMAT, GST_TIME_ARGS (segment.base),
 589                 GST_TIME_ARGS (self->current_start_offset),
 590                 GST_TIME_ARGS (segment.base + self->current_start_offset));
 591             segment.base += self->current_start_offset;
 592           } else {
 593             /* Shift start/stop byte position */
 594             GST_DEBUG_OBJECT (self,
 595                 "Updating segment start %" G_GUINT64_FORMAT " + %"
 596                 G_GUINT64_FORMAT " = %" G_GUINT64_FORMAT, segment.start,
 597                 self->current_start_offset,
 598                 segment.start + self->current_start_offset);
 599             segment.start += self->current_start_offset;
 600             if (segment.stop != -1) {
 601               GST_DEBUG_OBJECT (self,
 602                   "Updating segment stop %" G_GUINT64_FORMAT " + %"
 603                   G_GUINT64_FORMAT " = %" G_GUINT64_FORMAT, segment.stop,
 604                   self->current_start_offset,
 605                   segment.stop + self->current_start_offset);
 606               segment.stop += self->current_start_offset;
 607             }
 608           }
 609         }
 610         event = gst_event_new_segment (&segment);
 611         gst_event_set_seqnum (event, gst_event_get_seqnum (event));
 612
 613         g_mutex_unlock (&self->lock);
 614       }
 615       break;
 616     }
 617     case GST_EVENT_EOS:{
 618       gst_event_replace (&event, NULL);
 619
 620       if (!gst_concat_pad_wait (spad, self)) {
 621         ret = FALSE;
 622       } else {
 623         gboolean next;
 624
 625         g_mutex_lock (&self->lock);
 626         next = gst_concat_switch_pad (self);
 627         g_mutex_unlock (&self->lock);
 628         ret = TRUE;
 629
 630         gst_concat_notify_active_pad (self);
 631
 632         if (!next) {
 633           event = gst_event_new_eos ();
 634         } else {
 635           gst_element_post_message (GST_ELEMENT_CAST (self),
 636               gst_message_new_duration_changed (GST_OBJECT_CAST (self)));
 637         }
 638       }
 639       break;
 640     }
 641     case GST_EVENT_FLUSH_START:{
 642       gboolean forward;
 643
 644       g_mutex_lock (&self->lock);
 645       spad->flushing = TRUE;
 646       g_cond_broadcast (&self->cond);
 647       forward = (self->current_sinkpad == GST_PAD_CAST (spad));
 648       if (!forward && g_list_length (self->sinkpads) == 1)
 649         forward = TRUE;
 650       g_mutex_unlock (&self->lock);
 651
 652       if (!forward)
 653         gst_event_replace (&event, NULL);
 654       break;
 655     }
 656     case GST_EVENT_FLUSH_STOP:{
 657       gboolean forward;
 658
 659       gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
 660       spad->flushing = FALSE;
 661
 662       g_mutex_lock (&self->lock);
 663       forward = (self->current_sinkpad == GST_PAD_CAST (spad));
 664       if (!forward && g_list_length (self->sinkpads) == 1)
 665         forward = TRUE;
 666       g_mutex_unlock (&self->lock);
 667
 668       if (forward) {
 669         gboolean reset_time;
 670
 671         gst_event_parse_flush_stop (event, &reset_time);
 672         if (reset_time) {
 673           GST_DEBUG_OBJECT (self,
 674               "resetting start offset to 0 after flushing with reset_time = TRUE");
 675           self->current_start_offset = 0;
 676           self->last_stop = GST_CLOCK_TIME_NONE;
 677         }
 678       } else {
 679         gst_event_replace (&event, NULL);
 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_replace (&event, NULL);
 687         ret = FALSE;
 688       }
 689       break;
 690     }
 691   }
 692
 693   if (event) {
 694     g_mutex_lock (&self->lock);
 695     if (self->adjust_base && self->format == GST_FORMAT_TIME) {
 696       gint64 offset;
 697
 698       event = gst_event_make_writable (event);
 699       offset = gst_event_get_running_time_offset (event);
 700       offset += self->current_start_offset;
 701
 702       gst_event_set_running_time_offset (event, offset);
 703     }
 704     g_mutex_unlock (&self->lock);
 705     ret = gst_pad_event_default (pad, parent, event);
 706   }
 707
 708   return ret;
 709 }
 710
 711 static gboolean
 712 gst_concat_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
 713 {
 714   GstConcat *self = GST_CONCAT (parent);
 715   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 716   gboolean ret = TRUE;
 717
 718   GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
 719
 720   switch (GST_QUERY_TYPE (query)) {
 721     default:
 722       /* Wait for other serialized queries before forwarding */
 723       if (GST_QUERY_IS_SERIALIZED (query) && !gst_concat_pad_wait (spad, self)) {
 724         ret = FALSE;
 725       } else {
 726         ret = gst_pad_query_default (pad, parent, query);
 727       }
 728       break;
 729   }
 730
 731   return ret;
 732 }
 733
 734 static gboolean
 735 gst_concat_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 736 {
 737   GstConcat *self = GST_CONCAT (parent);
 738   gboolean ret = TRUE;
 739   GstPad *sinkpad = NULL;
 740
 741   GST_LOG_OBJECT (pad, "received event %" GST_PTR_FORMAT, event);
 742
 743   switch (GST_EVENT_TYPE (event)) {
 744     case GST_EVENT_SEEK:{
 745       g_mutex_lock (&self->lock);
 746       if ((sinkpad = self->current_sinkpad))
 747         gst_object_ref (sinkpad);
 748       /* If no current active sinkpad but only one sinkpad, try reactivating that pad */
 749       if (sinkpad == NULL && g_list_length (self->sinkpads) == 1) {
 750         sinkpad = gst_object_ref (self->sinkpads->data);
 751       }
 752       g_mutex_unlock (&self->lock);
 753       if (!sinkpad) {
 754         gst_event_replace (&event, NULL);
 755         ret = FALSE;
 756       }
 757       break;
 758     }
 759     case GST_EVENT_QOS:{
 760       GstPad *sinkpad = NULL;
 761
 762       g_mutex_lock (&self->lock);
 763       if ((sinkpad = self->current_sinkpad))
 764         gst_object_ref (sinkpad);
 765       g_mutex_unlock (&self->lock);
 766
 767
 768       if (!sinkpad) {
 769         gst_event_replace (&event, NULL);
 770         ret = FALSE;
 771       }
 772       break;
 773     }
 774     case GST_EVENT_FLUSH_STOP:{
 775       gboolean reset_time;
 776
 777       gst_event_parse_flush_stop (event, &reset_time);
 778       if (reset_time) {
 779         GST_DEBUG_OBJECT (self,
 780             "resetting start offset to 0 after flushing with reset_time = TRUE");
 781         self->current_start_offset = 0;
 782       }
 783       break;
 784     }
 785     default:
 786       break;
 787   }
 788
 789   if (event) {
 790     g_mutex_lock (&self->lock);
 791     if (self->adjust_base && self->format == GST_FORMAT_TIME) {
 792       gint64 offset;
 793
 794       event = gst_event_make_writable (event);
 795       offset = gst_event_get_running_time_offset (event);
 796       offset -= self->current_start_offset;
 797
 798       gst_event_set_running_time_offset (event, offset);
 799     }
 800     g_mutex_unlock (&self->lock);
 801
 802     if (sinkpad)
 803       ret = gst_pad_push_event (sinkpad, event);
 804     else
 805       ret = gst_pad_event_default (pad, parent, event);
 806   }
 807
 808   if (sinkpad)
 809     gst_object_unref (sinkpad);
 810
 811   return ret;
 812 }
 813
 814 static gboolean
 815 gst_concat_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
 816 {
 817   gboolean ret = TRUE;
 818
 819   GST_LOG_OBJECT (pad, "received query %" GST_PTR_FORMAT, query);
 820
 821   switch (GST_QUERY_TYPE (query)) {
 822     default:
 823       ret = gst_pad_query_default (pad, parent, query);
 824       break;
 825   }
 826
 827   return ret;
 828 }
 829
 830 static void
 831 reset_pad (const GValue * data, gpointer user_data)
 832 {
 833   GstPad *pad = g_value_get_object (data);
 834   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 835
 836   gst_segment_init (&spad->segment, GST_FORMAT_UNDEFINED);
 837   spad->flushing = FALSE;
 838 }
 839
 840 static void
 841 unblock_pad (const GValue * data, gpointer user_data)
 842 {
 843   GstPad *pad = g_value_get_object (data);
 844   GstConcatPad *spad = GST_CONCAT_PAD_CAST (pad);
 845
 846   spad->flushing = TRUE;
 847 }
 848
 849 static GstStateChangeReturn
 850 gst_concat_change_state (GstElement * element, GstStateChange transition)
 851 {
 852   GstConcat *self = GST_CONCAT (element);
 853   GstStateChangeReturn ret;
 854
 855   switch (transition) {
 856     case GST_STATE_CHANGE_READY_TO_PAUSED:{
 857       GstIterator *iter = gst_element_iterate_sink_pads (element);
 858       GstIteratorResult res;
 859
 860       self->format = GST_FORMAT_UNDEFINED;
 861       self->current_start_offset = 0;
 862       self->last_stop = GST_CLOCK_TIME_NONE;
 863
 864       while ((res =
 865               gst_iterator_foreach (iter, reset_pad,
 866                   NULL)) == GST_ITERATOR_RESYNC)
 867         gst_iterator_resync (iter);
 868       gst_iterator_free (iter);
 869
 870       if (res == GST_ITERATOR_ERROR)
 871         return GST_STATE_CHANGE_FAILURE;
 872       break;
 873     }
 874     case GST_STATE_CHANGE_PAUSED_TO_READY:{
 875       GstIterator *iter = gst_element_iterate_sink_pads (element);
 876       GstIteratorResult res;
 877
 878       g_mutex_lock (&self->lock);
 879       while ((res =
 880               gst_iterator_foreach (iter, unblock_pad,
 881                   NULL)) == GST_ITERATOR_RESYNC)
 882         gst_iterator_resync (iter);
 883       gst_iterator_free (iter);
 884       g_cond_broadcast (&self->cond);
 885       g_mutex_unlock (&self->lock);
 886
 887       if (res == GST_ITERATOR_ERROR)
 888         return GST_STATE_CHANGE_FAILURE;
 889
 890       break;
 891     }
 892     default:
 893       break;
 894   }
 895
 896   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 897
 898   return ret;
 899 }