gst/playback/gststreamsynchronizer.c

   1 /* GStreamer
   2  * Copyright (C) 2010 Sebastian Dröge <sebastian.droege@collabora.co.uk>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19
  20 #ifdef HAVE_CONFIG_H
  21 #include "config.h"
  22 #endif
  23
  24 #include "gststreamsynchronizer.h"
  25
  26 GST_DEBUG_CATEGORY_STATIC (stream_synchronizer_debug);
  27 #define GST_CAT_DEFAULT stream_synchronizer_debug
  28
  29 #define GST_STREAM_SYNCHRONIZER_LOCK(obj) G_STMT_START {                   \
  30     GST_LOG_OBJECT (obj,                                                \
  31                     "locking from thread %p",                           \
  32                     g_thread_self ());                                  \
  33     g_mutex_lock (GST_STREAM_SYNCHRONIZER_CAST(obj)->lock);                \
  34     GST_LOG_OBJECT (obj,                                                \
  35                     "locked from thread %p",                            \
  36                     g_thread_self ());                                  \
  37 } G_STMT_END
  38
  39 #define GST_STREAM_SYNCHRONIZER_UNLOCK(obj) G_STMT_START {                 \
  40     GST_LOG_OBJECT (obj,                                                \
  41                     "unlocking from thread %p",                         \
  42                     g_thread_self ());                                  \
  43     g_mutex_unlock (GST_STREAM_SYNCHRONIZER_CAST(obj)->lock);              \
  44 } G_STMT_END
  45
  46 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src_%d",
  47     GST_PAD_SRC,
  48     GST_PAD_SOMETIMES,
  49     GST_STATIC_CAPS_ANY);
  50 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink_%d",
  51     GST_PAD_SINK,
  52     GST_PAD_REQUEST,
  53     GST_STATIC_CAPS_ANY);
  54
  55 static const gboolean passthrough = TRUE;
  56
  57 GST_BOILERPLATE (GstStreamSynchronizer, gst_stream_synchronizer,
  58     GstElement, GST_TYPE_ELEMENT);
  59
  60 typedef struct
  61 {
  62   GstStreamSynchronizer *transform;
  63   guint stream_number;
  64   GstPad *srcpad;
  65   GstPad *sinkpad;
  66   GstSegment segment;
  67
  68   gboolean wait;
  69   gboolean new_stream;
  70   gboolean drop_discont;
  71   gboolean is_eos;
  72   gboolean seen_data;
  73
  74   gint64 running_time_diff;
  75 } GstStream;
  76
  77 /* Must be called with lock! */
  78 static GstPad *
  79 gst_stream_get_other_pad (GstStream * stream, GstPad * pad)
  80 {
  81   if (stream->sinkpad == pad)
  82     return gst_object_ref (stream->srcpad);
  83   else if (stream->srcpad == pad)
  84     return gst_object_ref (stream->sinkpad);
  85
  86   return NULL;
  87 }
  88
  89 static GstPad *
  90 gst_stream_get_other_pad_from_pad (GstPad * pad)
  91 {
  92   GstStreamSynchronizer *self =
  93       GST_STREAM_SYNCHRONIZER (gst_pad_get_parent (pad));
  94   GstStream *stream;
  95   GstPad *opad = NULL;
  96
  97   GST_STREAM_SYNCHRONIZER_LOCK (self);
  98   stream = gst_pad_get_element_private (pad);
  99   if (!stream)
 100     goto out;
 101
 102   opad = gst_stream_get_other_pad (stream, pad);
 103
 104 out:
 105   GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 106   gst_object_unref (self);
 107
 108   if (!opad)
 109     GST_WARNING_OBJECT (pad, "Trying to get other pad after releasing");
 110
 111   return opad;
 112 }
 113
 114 /* Generic pad functions */
 115 static GstIterator *
 116 gst_stream_synchronizer_iterate_internal_links (GstPad * pad)
 117 {
 118   GstIterator *it = NULL;
 119   GstPad *opad;
 120
 121   opad = gst_stream_get_other_pad_from_pad (pad);
 122   if (opad) {
 123     it = gst_iterator_new_single (GST_TYPE_PAD, opad,
 124         (GstCopyFunction) gst_object_ref, (GFreeFunc) gst_object_unref);
 125     gst_object_unref (opad);
 126   }
 127
 128   return it;
 129 }
 130
 131 static gboolean
 132 gst_stream_synchronizer_query (GstPad * pad, GstQuery * query)
 133 {
 134   GstPad *opad;
 135   gboolean ret = FALSE;
 136
 137   GST_LOG_OBJECT (pad, "Handling query %s", GST_QUERY_TYPE_NAME (query));
 138
 139   opad = gst_stream_get_other_pad_from_pad (pad);
 140   if (opad) {
 141     ret = gst_pad_peer_query (opad, query);
 142     gst_object_unref (opad);
 143   }
 144
 145   return ret;
 146 }
 147
 148 static GstCaps *
 149 gst_stream_synchronizer_getcaps (GstPad * pad)
 150 {
 151   GstPad *opad;
 152   GstCaps *ret = NULL;
 153
 154   opad = gst_stream_get_other_pad_from_pad (pad);
 155   if (opad) {
 156     ret = gst_pad_peer_get_caps (opad);
 157     gst_object_unref (opad);
 158   }
 159
 160   if (ret == NULL)
 161     ret = gst_caps_new_any ();
 162
 163   GST_LOG_OBJECT (pad, "Returning caps: %" GST_PTR_FORMAT, ret);
 164
 165   return ret;
 166 }
 167
 168 static gboolean
 169 gst_stream_synchronizer_acceptcaps (GstPad * pad, GstCaps * caps)
 170 {
 171   GstPad *opad;
 172   gboolean ret = FALSE;
 173
 174   opad = gst_stream_get_other_pad_from_pad (pad);
 175   if (opad) {
 176     ret = gst_pad_peer_accept_caps (opad, caps);
 177     gst_object_unref (opad);
 178   }
 179
 180   GST_LOG_OBJECT (pad, "Caps%s accepted: %" GST_PTR_FORMAT, (ret ? "" : " not"),
 181       caps);
 182
 183   return ret;
 184 }
 185
 186 /* srcpad functions */
 187 static gboolean
 188 gst_stream_synchronizer_src_event (GstPad * pad, GstEvent * event)
 189 {
 190   GstStreamSynchronizer *self =
 191       GST_STREAM_SYNCHRONIZER (gst_pad_get_parent (pad));
 192   GstPad *opad;
 193   gboolean ret = FALSE;
 194
 195   if (passthrough)
 196     goto skip_adjustments;
 197
 198   GST_LOG_OBJECT (pad, "Handling event %s: %" GST_PTR_FORMAT,
 199       GST_EVENT_TYPE_NAME (event), event->structure);
 200
 201   switch (GST_EVENT_TYPE (event)) {
 202     case GST_EVENT_QOS:{
 203       gdouble proportion;
 204       GstClockTimeDiff diff;
 205       GstClockTime timestamp;
 206       gint64 running_time_diff;
 207       GstStream *stream;
 208
 209       gst_event_parse_qos (event, &proportion, &diff, &timestamp);
 210       gst_event_unref (event);
 211
 212       GST_STREAM_SYNCHRONIZER_LOCK (self);
 213       stream = gst_pad_get_element_private (pad);
 214       if (stream)
 215         running_time_diff = stream->running_time_diff;
 216       else
 217         running_time_diff = -1;
 218       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 219
 220       if (running_time_diff == -1) {
 221         GST_WARNING_OBJECT (pad, "QOS event before group start");
 222         goto out;
 223       } else if (timestamp < running_time_diff) {
 224         GST_DEBUG_OBJECT (pad, "QOS event from previous group");
 225         goto out;
 226       }
 227
 228       GST_LOG_OBJECT (pad,
 229           "Adjusting QOS event: %" GST_TIME_FORMAT " - %" GST_TIME_FORMAT " = %"
 230           GST_TIME_FORMAT, GST_TIME_ARGS (timestamp),
 231           GST_TIME_ARGS (running_time_diff),
 232           GST_TIME_ARGS (timestamp - running_time_diff));
 233
 234       timestamp -= running_time_diff;
 235
 236       /* That case is invalid for QoS events */
 237       if (diff < 0 && -diff > timestamp) {
 238         GST_DEBUG_OBJECT (pad, "QOS event from previous group");
 239         ret = TRUE;
 240         goto out;
 241       }
 242
 243       event = gst_event_new_qos (proportion, diff, timestamp);
 244       break;
 245     }
 246     default:
 247       break;
 248   }
 249
 250 skip_adjustments:
 251
 252   opad = gst_stream_get_other_pad_from_pad (pad);
 253   if (opad) {
 254     ret = gst_pad_push_event (opad, event);
 255     gst_object_unref (opad);
 256   }
 257
 258 out:
 259   gst_object_unref (self);
 260
 261   return ret;
 262 }
 263
 264 /* sinkpad functions */
 265 static gboolean
 266 gst_stream_synchronizer_sink_event (GstPad * pad, GstEvent * event)
 267 {
 268   GstStreamSynchronizer *self =
 269       GST_STREAM_SYNCHRONIZER (gst_pad_get_parent (pad));
 270   GstPad *opad;
 271   gboolean ret = FALSE;
 272
 273   if (passthrough)
 274     goto skip_adjustments;
 275
 276   GST_LOG_OBJECT (pad, "Handling event %s: %" GST_PTR_FORMAT,
 277       GST_EVENT_TYPE_NAME (event), event->structure);
 278
 279   switch (GST_EVENT_TYPE (event)) {
 280     case GST_EVENT_SINK_MESSAGE:{
 281       GstMessage *message;
 282
 283       gst_event_parse_sink_message (event, &message);
 284       if (message->structure
 285           && gst_structure_has_name (message->structure,
 286               "playbin2-stream-changed")) {
 287         GstStream *stream;
 288
 289         GST_STREAM_SYNCHRONIZER_LOCK (self);
 290         stream = gst_pad_get_element_private (pad);
 291         if (stream) {
 292           GList *l;
 293           gboolean all_wait = TRUE;
 294
 295           GST_DEBUG_OBJECT (pad, "Stream %d changed", stream->stream_number);
 296
 297           stream->is_eos = FALSE;
 298           stream->wait = TRUE;
 299           stream->new_stream = TRUE;
 300
 301           for (l = self->streams; l; l = l->next) {
 302             GstStream *ostream = l->data;
 303
 304             all_wait = all_wait && ostream->wait;
 305             if (!all_wait)
 306               break;
 307           }
 308           if (all_wait) {
 309             gint64 last_stop = 0;
 310
 311             GST_DEBUG_OBJECT (self, "All streams have changed -- unblocking");
 312
 313             for (l = self->streams; l; l = l->next) {
 314               GstStream *ostream = l->data;
 315               gint64 stop_running_time;
 316               gint64 last_stop_running_time;
 317
 318               ostream->wait = FALSE;
 319
 320               stop_running_time =
 321                   gst_segment_to_running_time (&ostream->segment,
 322                   GST_FORMAT_TIME, ostream->segment.stop);
 323               last_stop_running_time =
 324                   gst_segment_to_running_time (&ostream->segment,
 325                   GST_FORMAT_TIME, ostream->segment.last_stop);
 326               last_stop =
 327                   MAX (last_stop, MAX (stop_running_time,
 328                       last_stop_running_time));
 329             }
 330             last_stop = MAX (0, last_stop);
 331             self->group_start_time = MAX (self->group_start_time, last_stop);
 332
 333             GST_DEBUG_OBJECT (self, "New group start time: %" GST_TIME_FORMAT,
 334                 GST_TIME_ARGS (self->group_start_time));
 335
 336             g_cond_broadcast (self->stream_finish_cond);
 337           }
 338         }
 339         GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 340       }
 341       gst_message_unref (message);
 342       break;
 343     }
 344     case GST_EVENT_NEWSEGMENT:{
 345       GstStream *stream;
 346       gboolean update;
 347       gdouble rate, applied_rate;
 348       GstFormat format;
 349       gint64 start, stop, position;
 350
 351       gst_event_parse_new_segment_full (event,
 352           &update, &rate, &applied_rate, &format, &start, &stop, &position);
 353
 354       GST_STREAM_SYNCHRONIZER_LOCK (self);
 355       stream = gst_pad_get_element_private (pad);
 356       if (stream) {
 357         if (stream->wait) {
 358           GST_DEBUG_OBJECT (pad, "Stream %d is waiting", stream->stream_number);
 359           g_cond_wait (self->stream_finish_cond, self->lock);
 360           stream = gst_pad_get_element_private (pad);
 361           if (stream)
 362             stream->wait = FALSE;
 363         }
 364       }
 365
 366       if (self->shutdown) {
 367         GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 368         gst_event_unref (event);
 369         goto done;
 370       }
 371
 372       if (stream && format == GST_FORMAT_TIME) {
 373         if (stream->new_stream) {
 374           gint64 last_stop_running_time = 0;
 375           gint64 stop_running_time = 0;
 376
 377           if (stream->segment.format == GST_FORMAT_TIME) {
 378             last_stop_running_time =
 379                 gst_segment_to_running_time (&stream->segment, GST_FORMAT_TIME,
 380                 stream->segment.last_stop);
 381             last_stop_running_time = MAX (last_stop_running_time, 0);
 382             stop_running_time =
 383                 gst_segment_to_running_time (&stream->segment, GST_FORMAT_TIME,
 384                 stream->segment.stop);
 385             stop_running_time = MAX (last_stop_running_time, 0);
 386
 387             if (stop_running_time != last_stop_running_time) {
 388               GST_WARNING_OBJECT (pad,
 389                   "Gap between last_stop and segment stop: %" GST_TIME_FORMAT
 390                   " != %" GST_TIME_FORMAT, GST_TIME_ARGS (stop_running_time),
 391                   GST_TIME_ARGS (last_stop_running_time));
 392             }
 393
 394             if (stop_running_time < last_stop_running_time) {
 395               GST_DEBUG_OBJECT (pad, "Updating stop position");
 396               gst_pad_push_event (stream->srcpad,
 397                   gst_event_new_new_segment_full (TRUE, stream->segment.rate,
 398                       stream->segment.applied_rate, GST_FORMAT_TIME,
 399                       stream->segment.start, stream->segment.last_stop,
 400                       stream->segment.time));
 401               gst_segment_set_newsegment_full (&stream->segment, TRUE,
 402                   stream->segment.rate, stream->segment.applied_rate,
 403                   GST_FORMAT_TIME, stream->segment.start,
 404                   stream->segment.last_stop, stream->segment.time);
 405             }
 406             stop_running_time = MAX (stop_running_time, last_stop_running_time);
 407             GST_DEBUG_OBJECT (pad,
 408                 "Stop running time of last group: %" GST_TIME_FORMAT,
 409                 GST_TIME_ARGS (stop_running_time));
 410           }
 411           stream->new_stream = FALSE;
 412           stream->drop_discont = TRUE;
 413
 414           if (stop_running_time < self->group_start_time) {
 415             gint64 diff = self->group_start_time - stop_running_time;
 416
 417             GST_DEBUG_OBJECT (pad,
 418                 "Advancing running time for other streams by: %"
 419                 GST_TIME_FORMAT, GST_TIME_ARGS (diff));
 420             gst_pad_push_event (stream->srcpad,
 421                 gst_event_new_new_segment_full (FALSE, 1.0, 1.0,
 422                     GST_FORMAT_TIME, 0, diff, 0));
 423             gst_segment_set_newsegment_full (&stream->segment, FALSE, 1.0, 1.0,
 424                 GST_FORMAT_TIME, 0, diff, 0);
 425           }
 426         }
 427
 428         GST_DEBUG_OBJECT (pad, "Segment was: %" GST_SEGMENT_FORMAT,
 429             &stream->segment);
 430         gst_segment_set_newsegment_full (&stream->segment, update, rate,
 431             applied_rate, format, start, stop, position);
 432         GST_DEBUG_OBJECT (pad, "Segment now is: %" GST_SEGMENT_FORMAT,
 433             &stream->segment);
 434
 435         GST_DEBUG_OBJECT (pad, "Stream start running time: %" GST_TIME_FORMAT,
 436             GST_TIME_ARGS (stream->segment.accum));
 437         stream->running_time_diff = stream->segment.accum;
 438       } else if (stream) {
 439         GST_WARNING_OBJECT (pad, "Non-TIME segment: %s",
 440             gst_format_get_name (format));
 441         gst_segment_init (&stream->segment, GST_FORMAT_UNDEFINED);
 442       }
 443       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 444       break;
 445     }
 446     case GST_EVENT_FLUSH_STOP:{
 447       GstStream *stream;
 448
 449       GST_STREAM_SYNCHRONIZER_LOCK (self);
 450       stream = gst_pad_get_element_private (pad);
 451       if (stream) {
 452         GST_DEBUG_OBJECT (pad, "Resetting segment for stream %d",
 453             stream->stream_number);
 454         gst_segment_init (&stream->segment, GST_FORMAT_UNDEFINED);
 455
 456         stream->is_eos = FALSE;
 457         stream->wait = FALSE;
 458         stream->new_stream = FALSE;
 459         stream->drop_discont = FALSE;
 460         stream->seen_data = FALSE;
 461       }
 462       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 463       break;
 464     }
 465     case GST_EVENT_EOS:{
 466       GstStream *stream;
 467       GList *l;
 468       gboolean all_eos = TRUE;
 469       gboolean seen_data;
 470       GSList *pads = NULL;
 471       GstPad *srcpad;
 472
 473       GST_STREAM_SYNCHRONIZER_LOCK (self);
 474       stream = gst_pad_get_element_private (pad);
 475       if (stream) {
 476         GST_DEBUG_OBJECT (pad, "Have EOS for stream %d", stream->stream_number);
 477         stream->is_eos = TRUE;
 478       }
 479
 480       seen_data = stream->seen_data;
 481       srcpad = gst_object_ref (stream->srcpad);
 482
 483       for (l = self->streams; l; l = l->next) {
 484         GstStream *ostream = l->data;
 485
 486         all_eos = all_eos && ostream->is_eos;
 487         if (!all_eos)
 488           break;
 489       }
 490
 491       if (all_eos) {
 492         GST_DEBUG_OBJECT (self, "All streams are EOS -- forwarding");
 493         for (l = self->streams; l; l = l->next) {
 494           GstStream *ostream = l->data;
 495           /* local snapshot of current pads */
 496           gst_object_ref (ostream->srcpad);
 497           pads = g_slist_prepend (pads, ostream->srcpad);
 498         }
 499       }
 500       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 501       /* drop lock when sending eos, which may block in e.g. preroll */
 502       if (pads) {
 503         GstPad *pad;
 504         GSList *epad;
 505
 506         ret = TRUE;
 507         epad = pads;
 508         while (epad) {
 509           pad = epad->data;
 510           GST_DEBUG_OBJECT (pad, "Pushing EOS");
 511           ret = ret && gst_pad_push_event (pad, gst_event_new_eos ());
 512           gst_object_unref (pad);
 513           epad = g_slist_next (epad);
 514         }
 515         g_slist_free (pads);
 516       } else {
 517         /* if EOS, but no data has passed, then send something to replace EOS
 518          * for preroll purposes */
 519         if (!seen_data) {
 520           GstBuffer *buf = gst_buffer_new ();
 521
 522           GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_PREROLL);
 523           gst_pad_push (srcpad, buf);
 524         }
 525       }
 526       gst_object_unref (srcpad);
 527       goto done;
 528       break;
 529     }
 530     default:
 531       break;
 532   }
 533
 534 skip_adjustments:
 535
 536   opad = gst_stream_get_other_pad_from_pad (pad);
 537   if (opad) {
 538     ret = gst_pad_push_event (opad, event);
 539     gst_object_unref (opad);
 540   }
 541
 542 done:
 543   gst_object_unref (self);
 544
 545   return ret;
 546 }
 547
 548 static GstFlowReturn
 549 gst_stream_synchronizer_sink_bufferalloc (GstPad * pad, guint64 offset,
 550     guint size, GstCaps * caps, GstBuffer ** buf)
 551 {
 552   GstPad *opad;
 553   GstFlowReturn ret = GST_FLOW_ERROR;
 554
 555   GST_LOG_OBJECT (pad, "Allocating buffer: size=%u", size);
 556
 557   opad = gst_stream_get_other_pad_from_pad (pad);
 558   if (opad) {
 559     ret = gst_pad_alloc_buffer (opad, offset, size, caps, buf);
 560     gst_object_unref (opad);
 561   }
 562
 563   GST_LOG_OBJECT (pad, "Allocation: %s", gst_flow_get_name (ret));
 564
 565   return ret;
 566 }
 567
 568 static GstFlowReturn
 569 gst_stream_synchronizer_sink_chain (GstPad * pad, GstBuffer * buffer)
 570 {
 571   GstStreamSynchronizer *self =
 572       GST_STREAM_SYNCHRONIZER (gst_pad_get_parent (pad));
 573   GstPad *opad;
 574   GstFlowReturn ret = GST_FLOW_ERROR;
 575   GstStream *stream;
 576   GstClockTime timestamp = GST_CLOCK_TIME_NONE;
 577   GstClockTime timestamp_end = GST_CLOCK_TIME_NONE;
 578
 579   if (passthrough) {
 580     opad = gst_stream_get_other_pad_from_pad (pad);
 581     if (opad) {
 582       ret = gst_pad_push (opad, buffer);
 583       gst_object_unref (opad);
 584     }
 585     goto done;
 586   }
 587
 588   GST_LOG_OBJECT (pad, "Handling buffer %p: size=%u, timestamp=%"
 589       GST_TIME_FORMAT " duration=%" GST_TIME_FORMAT
 590       " offset=%" G_GUINT64_FORMAT " offset_end=%" G_GUINT64_FORMAT,
 591       buffer, GST_BUFFER_SIZE (buffer),
 592       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
 593       GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)),
 594       GST_BUFFER_OFFSET (buffer), GST_BUFFER_OFFSET_END (buffer));
 595
 596   timestamp = GST_BUFFER_TIMESTAMP (buffer);
 597   if (GST_BUFFER_TIMESTAMP_IS_VALID (buffer)
 598       && GST_BUFFER_DURATION_IS_VALID (buffer))
 599     timestamp_end = timestamp + GST_BUFFER_DURATION (buffer);
 600
 601   GST_STREAM_SYNCHRONIZER_LOCK (self);
 602   stream = gst_pad_get_element_private (pad);
 603
 604   stream->seen_data = TRUE;
 605   if (stream && stream->drop_discont) {
 606     buffer = gst_buffer_make_metadata_writable (buffer);
 607     GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
 608     stream->drop_discont = FALSE;
 609   }
 610
 611   if (stream && stream->segment.format == GST_FORMAT_TIME
 612       && GST_CLOCK_TIME_IS_VALID (timestamp)) {
 613     GST_LOG_OBJECT (pad,
 614         "Updating last-stop from %" GST_TIME_FORMAT " to %" GST_TIME_FORMAT,
 615         GST_TIME_ARGS (stream->segment.last_stop), GST_TIME_ARGS (timestamp));
 616     gst_segment_set_last_stop (&stream->segment, GST_FORMAT_TIME, timestamp);
 617   }
 618   GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 619
 620   opad = gst_stream_get_other_pad_from_pad (pad);
 621   if (opad) {
 622     ret = gst_pad_push (opad, buffer);
 623     gst_object_unref (opad);
 624   }
 625
 626   GST_LOG_OBJECT (pad, "Push returned: %s", gst_flow_get_name (ret));
 627   if (ret == GST_FLOW_OK) {
 628     GList *l;
 629
 630     GST_STREAM_SYNCHRONIZER_LOCK (self);
 631     stream = gst_pad_get_element_private (pad);
 632     if (stream && stream->segment.format == GST_FORMAT_TIME
 633         && GST_CLOCK_TIME_IS_VALID (timestamp_end)) {
 634       GST_LOG_OBJECT (pad,
 635           "Updating last-stop from %" GST_TIME_FORMAT " to %" GST_TIME_FORMAT,
 636           GST_TIME_ARGS (stream->segment.last_stop),
 637           GST_TIME_ARGS (timestamp_end));
 638       gst_segment_set_last_stop (&stream->segment, GST_FORMAT_TIME,
 639           timestamp_end);
 640     }
 641
 642     /* Advance EOS streams if necessary. For non-EOS
 643      * streams the demuxers should already do this! */
 644     for (l = self->streams; l; l = l->next) {
 645       GstStream *ostream = l->data;
 646       gint64 last_stop;
 647
 648       if (!ostream->is_eos || ostream->segment.format != GST_FORMAT_TIME)
 649         continue;
 650
 651       if (ostream->segment.last_stop != -1)
 652         last_stop = ostream->segment.last_stop;
 653       else
 654         last_stop = ostream->segment.start;
 655
 656       /* Is there a 1 second lag? */
 657       if (last_stop != -1 && last_stop + GST_SECOND < timestamp_end) {
 658         gint64 new_start, new_stop;
 659
 660         new_start = timestamp_end - GST_SECOND;
 661         if (ostream->segment.stop == -1)
 662           new_stop = -1;
 663         else
 664           new_stop = MAX (new_start, ostream->segment.stop);
 665
 666         GST_DEBUG_OBJECT (ostream->sinkpad,
 667             "Advancing stream %u from %" GST_TIME_FORMAT " to %"
 668             GST_TIME_FORMAT, ostream->stream_number, GST_TIME_ARGS (last_stop),
 669             GST_TIME_ARGS (new_start));
 670
 671         gst_pad_push_event (ostream->srcpad,
 672             gst_event_new_new_segment_full (TRUE, ostream->segment.rate,
 673                 ostream->segment.applied_rate, ostream->segment.format,
 674                 new_start, new_stop, new_start));
 675         gst_segment_set_newsegment_full (&ostream->segment, TRUE,
 676             ostream->segment.rate, ostream->segment.applied_rate,
 677             ostream->segment.format, new_start, new_stop, new_start);
 678         gst_segment_set_last_stop (&ostream->segment, GST_FORMAT_TIME,
 679             new_start);
 680       }
 681     }
 682     GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 683   }
 684
 685 done:
 686
 687   gst_object_unref (self);
 688
 689   return ret;
 690 }
 691
 692 /* GstElement vfuncs */
 693 static GstPad *
 694 gst_stream_synchronizer_request_new_pad (GstElement * element,
 695     GstPadTemplate * temp, const gchar * name)
 696 {
 697   GstStreamSynchronizer *self = GST_STREAM_SYNCHRONIZER (element);
 698   GstStream *stream;
 699   gchar *tmp;
 700
 701   GST_STREAM_SYNCHRONIZER_LOCK (self);
 702   GST_DEBUG_OBJECT (self, "Requesting new pad for stream %d",
 703       self->current_stream_number);
 704
 705   stream = g_slice_new0 (GstStream);
 706   stream->transform = self;
 707   stream->stream_number = self->current_stream_number;
 708
 709   tmp = g_strdup_printf ("sink_%d", self->current_stream_number);
 710   stream->sinkpad = gst_pad_new_from_static_template (&sinktemplate, tmp);
 711   g_free (tmp);
 712   gst_pad_set_element_private (stream->sinkpad, stream);
 713   gst_pad_set_iterate_internal_links_function (stream->sinkpad,
 714       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_iterate_internal_links));
 715   gst_pad_set_query_function (stream->sinkpad,
 716       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_query));
 717   gst_pad_set_getcaps_function (stream->sinkpad,
 718       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_getcaps));
 719   gst_pad_set_acceptcaps_function (stream->sinkpad,
 720       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_acceptcaps));
 721   gst_pad_set_event_function (stream->sinkpad,
 722       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_sink_event));
 723   gst_pad_set_chain_function (stream->sinkpad,
 724       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_sink_chain));
 725   gst_pad_set_bufferalloc_function (stream->sinkpad,
 726       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_sink_bufferalloc));
 727
 728   tmp = g_strdup_printf ("src_%d", self->current_stream_number);
 729   stream->srcpad = gst_pad_new_from_static_template (&srctemplate, tmp);
 730   g_free (tmp);
 731   gst_pad_set_element_private (stream->srcpad, stream);
 732   gst_pad_set_iterate_internal_links_function (stream->srcpad,
 733       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_iterate_internal_links));
 734   gst_pad_set_query_function (stream->srcpad,
 735       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_query));
 736   gst_pad_set_getcaps_function (stream->srcpad,
 737       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_getcaps));
 738   gst_pad_set_acceptcaps_function (stream->srcpad,
 739       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_acceptcaps));
 740   gst_pad_set_event_function (stream->srcpad,
 741       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_src_event));
 742
 743   gst_segment_init (&stream->segment, GST_FORMAT_UNDEFINED);
 744
 745   self->streams = g_list_prepend (self->streams, stream);
 746   self->current_stream_number++;
 747   GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 748
 749   /* Add pads and activate unless we're going to NULL */
 750   g_static_rec_mutex_lock (GST_STATE_GET_LOCK (self));
 751   if (GST_STATE_TARGET (self) != GST_STATE_NULL) {
 752     gst_pad_set_active (stream->srcpad, TRUE);
 753     gst_pad_set_active (stream->sinkpad, TRUE);
 754   }
 755   gst_element_add_pad (GST_ELEMENT_CAST (self), stream->srcpad);
 756   gst_element_add_pad (GST_ELEMENT_CAST (self), stream->sinkpad);
 757   g_static_rec_mutex_unlock (GST_STATE_GET_LOCK (self));
 758
 759   return stream->sinkpad;
 760 }
 761
 762 /* Must be called with lock! */
 763 static void
 764 gst_stream_synchronizer_release_stream (GstStreamSynchronizer * self,
 765     GstStream * stream)
 766 {
 767   GList *l;
 768
 769   GST_DEBUG_OBJECT (self, "Releasing stream %d", stream->stream_number);
 770
 771   for (l = self->streams; l; l = l->next) {
 772     if (l->data == stream) {
 773       self->streams = g_list_delete_link (self->streams, l);
 774       break;
 775     }
 776   }
 777   g_assert (l != NULL);
 778
 779   /* we can drop the lock, since stream exists now only local.
 780    * Moreover, we should drop, to prevent deadlock with STREAM_LOCK
 781    * (due to reverse lock order) when deactivating pads */
 782   GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 783
 784   gst_pad_set_element_private (stream->srcpad, NULL);
 785   gst_pad_set_element_private (stream->sinkpad, NULL);
 786   gst_pad_set_active (stream->srcpad, FALSE);
 787   gst_element_remove_pad (GST_ELEMENT_CAST (self), stream->srcpad);
 788   gst_pad_set_active (stream->sinkpad, FALSE);
 789   gst_element_remove_pad (GST_ELEMENT_CAST (self), stream->sinkpad);
 790
 791   if (stream->segment.format == GST_FORMAT_TIME) {
 792     gint64 stop_running_time;
 793     gint64 last_stop_running_time;
 794
 795     stop_running_time =
 796         gst_segment_to_running_time (&stream->segment, GST_FORMAT_TIME,
 797         stream->segment.stop);
 798     last_stop_running_time =
 799         gst_segment_to_running_time (&stream->segment, GST_FORMAT_TIME,
 800         stream->segment.last_stop);
 801     stop_running_time = MAX (stop_running_time, last_stop_running_time);
 802
 803     GST_DEBUG_OBJECT (stream->sinkpad,
 804         "Stop running time was: %" GST_TIME_FORMAT,
 805         GST_TIME_ARGS (stop_running_time));
 806
 807     self->group_start_time = MAX (self->group_start_time, stop_running_time);
 808   }
 809
 810   g_slice_free (GstStream, stream);
 811
 812   /* NOTE: In theory we have to check here if all streams
 813    * are EOS but the one that was removed wasn't and then
 814    * send EOS downstream. But due to the way how playsink
 815    * works this is not necessary and will only cause problems
 816    * for gapless playback. playsink will only add/remove pads
 817    * when it's reconfigured, which happens when the streams
 818    * change
 819    */
 820
 821   /* lock for good measure, since the caller had it */
 822   GST_STREAM_SYNCHRONIZER_LOCK (self);
 823 }
 824
 825 static void
 826 gst_stream_synchronizer_release_pad (GstElement * element, GstPad * pad)
 827 {
 828   GstStreamSynchronizer *self = GST_STREAM_SYNCHRONIZER (element);
 829   GstStream *stream;
 830
 831   GST_STREAM_SYNCHRONIZER_LOCK (self);
 832   stream = gst_pad_get_element_private (pad);
 833   if (stream) {
 834     g_assert (stream->sinkpad == pad);
 835
 836     gst_stream_synchronizer_release_stream (self, stream);
 837   }
 838   GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 839 }
 840
 841 static GstStateChangeReturn
 842 gst_stream_synchronizer_change_state (GstElement * element,
 843     GstStateChange transition)
 844 {
 845   GstStreamSynchronizer *self = GST_STREAM_SYNCHRONIZER (element);
 846   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
 847
 848   switch (transition) {
 849     case GST_STATE_CHANGE_NULL_TO_READY:
 850       GST_DEBUG_OBJECT (self, "State change NULL->READY");
 851       self->shutdown = FALSE;
 852       break;
 853     case GST_STATE_CHANGE_READY_TO_PAUSED:
 854       GST_DEBUG_OBJECT (self, "State change READY->PAUSED");
 855       self->group_start_time = 0;
 856       self->shutdown = FALSE;
 857       break;
 858     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
 859       GST_DEBUG_OBJECT (self, "State change PAUSED->PLAYING");
 860       break;
 861     case GST_STATE_CHANGE_PAUSED_TO_READY:
 862       GST_DEBUG_OBJECT (self, "State change READY->NULL");
 863
 864       GST_STREAM_SYNCHRONIZER_LOCK (self);
 865       g_cond_broadcast (self->stream_finish_cond);
 866       self->shutdown = TRUE;
 867       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 868     default:
 869       break;
 870   }
 871
 872   {
 873     GstStateChangeReturn bret;
 874
 875     bret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 876     GST_DEBUG_OBJECT (self, "Base class state changed returned: %d", bret);
 877     if (G_UNLIKELY (bret == GST_STATE_CHANGE_FAILURE))
 878       return ret;
 879   }
 880
 881   switch (transition) {
 882     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
 883       GST_DEBUG_OBJECT (self, "State change PLAYING->PAUSED");
 884       break;
 885     case GST_STATE_CHANGE_PAUSED_TO_READY:{
 886       GList *l;
 887
 888       GST_DEBUG_OBJECT (self, "State change PAUSED->READY");
 889       self->group_start_time = 0;
 890
 891       GST_STREAM_SYNCHRONIZER_LOCK (self);
 892       for (l = self->streams; l; l = l->next) {
 893         GstStream *stream = l->data;
 894
 895         gst_segment_init (&stream->segment, GST_FORMAT_UNDEFINED);
 896         stream->wait = FALSE;
 897         stream->new_stream = FALSE;
 898         stream->drop_discont = FALSE;
 899         stream->is_eos = FALSE;
 900       }
 901       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 902       break;
 903     }
 904     case GST_STATE_CHANGE_READY_TO_NULL:{
 905       GST_DEBUG_OBJECT (self, "State change READY->NULL");
 906
 907       GST_STREAM_SYNCHRONIZER_LOCK (self);
 908       while (self->streams)
 909         gst_stream_synchronizer_release_stream (self, self->streams->data);
 910       self->current_stream_number = 0;
 911       GST_STREAM_SYNCHRONIZER_UNLOCK (self);
 912       break;
 913     }
 914     default:
 915       break;
 916   }
 917
 918   return ret;
 919 }
 920
 921 /* GObject vfuncs */
 922 static void
 923 gst_stream_synchronizer_finalize (GObject * object)
 924 {
 925   GstStreamSynchronizer *self = GST_STREAM_SYNCHRONIZER (object);
 926
 927   if (self->lock) {
 928     g_mutex_free (self->lock);
 929     self->lock = NULL;
 930   }
 931
 932   if (self->stream_finish_cond) {
 933     g_cond_free (self->stream_finish_cond);
 934     self->stream_finish_cond = NULL;
 935   }
 936
 937   G_OBJECT_CLASS (parent_class)->finalize (object);
 938 }
 939
 940 /* GObject type initialization */
 941 static void
 942 gst_stream_synchronizer_init (GstStreamSynchronizer * self,
 943     GstStreamSynchronizerClass * klass)
 944 {
 945   self->lock = g_mutex_new ();
 946   self->stream_finish_cond = g_cond_new ();
 947 }
 948
 949 static void
 950 gst_stream_synchronizer_base_init (gpointer g_class)
 951 {
 952   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 953
 954   gst_element_class_add_pad_template (gstelement_class,
 955       gst_static_pad_template_get (&srctemplate));
 956   gst_element_class_add_pad_template (gstelement_class,
 957       gst_static_pad_template_get (&sinktemplate));
 958
 959   gst_element_class_set_details_simple (gstelement_class,
 960       "Stream Synchronizer", "Generic",
 961       "Synchronizes a group of streams to have equal durations and starting points",
 962       "Sebastian Dröge <sebastian.droege@collabora.co.uk>");
 963 }
 964
 965 static void
 966 gst_stream_synchronizer_class_init (GstStreamSynchronizerClass * klass)
 967 {
 968   GObjectClass *gobject_class = (GObjectClass *) klass;
 969   GstElementClass *element_class = (GstElementClass *) klass;
 970
 971   GST_DEBUG_CATEGORY_INIT (stream_synchronizer_debug,
 972       "streamsynchronizer", 0, "Stream Synchronizer");
 973
 974   gobject_class->finalize = gst_stream_synchronizer_finalize;
 975
 976   element_class->change_state =
 977       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_change_state);
 978   element_class->request_new_pad =
 979       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_request_new_pad);
 980   element_class->release_pad =
 981       GST_DEBUG_FUNCPTR (gst_stream_synchronizer_release_pad);
 982 }