libs/gst/base/gstbasesink.c

   1 /* GStreamer
   2  * Copyright (C) 2005-2007 Wim Taymans <wim.taymans@gmail.com>
   3  *
   4  * gstbasesink.c: Base class for sink elements
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 /**
  23  * SECTION:gstbasesink
  24  * @short_description: Base class for sink elements
  25  * @see_also: #GstBaseTransform, #GstBaseSource
  26  *
  27  * #GstBaseSink is the base class for sink elements in GStreamer, such as
  28  * xvimagesink or filesink. It is a layer on top of #GstElement that provides a
  29  * simplified interface to plugin writers. #GstBaseSink handles many details
  30  * for you, for example: preroll, clock synchronization, state changes,
  31  * activation in push or pull mode, and queries.
  32  *
  33  * In most cases, when writing sink elements, there is no need to implement
  34  * class methods from #GstElement or to set functions on pads, because the
  35  * #GstBaseSink infrastructure should be sufficient.
  36  *
  37  * #GstBaseSink provides support for exactly one sink pad, which should be
  38  * named "sink". A sink implementation (subclass of #GstBaseSink) should
  39  * install a pad template in its base_init function, like so:
  40  * <programlisting>
  41  * static void
  42  * my_element_base_init (gpointer g_class)
  43  * {
  44  *   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
  45  *
  46  *   // sinktemplate should be a #GstStaticPadTemplate with direction
  47  *   // #GST_PAD_SINK and name "sink"
  48  *   gst_element_class_add_pad_template (gstelement_class,
  49  *       gst_static_pad_template_get (&amp;sinktemplate));
  50  *   // see #GstElementDetails
  51  *   gst_element_class_set_details (gstelement_class, &amp;details);
  52  * }
  53  * </programlisting>
  54  *
  55  * #GstBaseSink will handle the prerolling correctly. This means that it will
  56  * return #GST_STATE_CHANGE_ASYNC from a state change to PAUSED until the first
  57  * buffer arrives in this element. The base class will call the
  58  * #GstBaseSink::preroll vmethod with this preroll buffer and will then commit
  59  * the state change to the next asynchronously pending state.
  60  *
  61  * When the element is set to PLAYING, #GstBaseSink will synchronise on the
  62  * clock using the times returned from ::get_times. If this function returns
  63  * #GST_CLOCK_TIME_NONE for the start time, no synchronisation will be done.
  64  * Synchronisation can be disabled entirely by setting the object "sync"
  65  * property to %FALSE.
  66  *
  67  * After synchronisation the virtual method #GstBaseSink::render will be called.
  68  * Subclasses should minimally implement this method.
  69  *
  70  * Since 0.10.3 subclasses that synchronise on the clock in the ::render method
  71  * are supported as well. These classes typically receive a buffer in the render
  72  * method and can then potentially block on the clock while rendering. A typical
  73  * example is an audiosink. Since 0.10.11 these subclasses can use
  74  * gst_base_sink_wait_preroll() to perform the blocking wait.
  75  *
  76  * Upon receiving the EOS event in the PLAYING state, #GstBaseSink will wait
  77  * for the clock to reach the time indicated by the stop time of the last
  78  * ::get_times call before posting an EOS message. When the element receives
  79  * EOS in PAUSED, preroll completes, the event is queued and an EOS message is
  80  * posted when going to PLAYING.
  81  *
  82  * #GstBaseSink will internally use the #GST_EVENT_NEWSEGMENT events to schedule
  83  * synchronisation and clipping of buffers. Buffers that fall completely outside
  84  * of the current segment are dropped. Buffers that fall partially in the
  85  * segment are rendered (and prerolled). Subclasses should do any subbuffer
  86  * clipping themselves when needed.
  87  *
  88  * #GstBaseSink will by default report the current playback position in
  89  * #GST_FORMAT_TIME based on the current clock time and segment information.
  90  * If no clock has been set on the element, the query will be forwarded
  91  * upstream.
  92  *
  93  * The ::set_caps function will be called when the subclass should configure
  94  * itself to process a specific media type.
  95  *
  96  * The ::start and ::stop virtual methods will be called when resources should
  97  * be allocated. Any ::preroll, ::render  and ::set_caps function will be
  98  * called between the ::start and ::stop calls.
  99  *
 100  * The ::event virtual method will be called when an event is received by
 101  * #GstBaseSink. Normally this method should only be overriden by very specific
 102  * elements (such as file sinks) which need to handle the newsegment event
 103  * specially.
 104  *
 105  * #GstBaseSink provides an overridable ::buffer_alloc function that can be
 106  * used by sinks that want to do reverse negotiation or to provide
 107  * custom buffers (hardware buffers for example) to upstream elements.
 108  *
 109  * The ::unlock method is called when the elements should unblock any blocking
 110  * operations they perform in the ::render method. This is mostly useful when
 111  * the ::render method performs a blocking write on a file descriptor, for
 112  * example.
 113  *
 114  * The max-lateness property affects how the sink deals with buffers that
 115  * arrive too late in the sink. A buffer arrives too late in the sink when
 116  * the presentation time (as a combination of the last segment, buffer
 117  * timestamp and element base_time) plus the duration is before the current
 118  * time of the clock.
 119  * If the frame is later than max-lateness, the sink will drop the buffer
 120  * without calling the render method.
 121  * This feature is disabled if sync is disabled, the ::get-times method does
 122  * not return a valid start time or max-lateness is set to -1 (the default).
 123  * Subclasses can use gst_base_sink_set_max_lateness() to configure the
 124  * max-lateness value.
 125  *
 126  * The qos property will enable the quality-of-service features of the basesink
 127  * which gather statistics about the real-time performance of the clock
 128  * synchronisation. For each buffer received in the sink, statistics are
 129  * gathered and a QOS event is sent upstream with these numbers. This
 130  * information can then be used by upstream elements to reduce their processing
 131  * rate, for example.
 132  *
 133  * Since 0.10.15 the async property can be used to instruct the sink to never
 134  * perform an ASYNC state change. This feature is mostly usable when dealing
 135  * with non-synchronized streams or sparse streams.
 136  *
 137  * Last reviewed on 2007-08-29 (0.10.15)
 138  */
 139
 140 #ifdef HAVE_CONFIG_H
 141 #  include "config.h"
 142 #endif
 143
 144 #include "gstbasesink.h"
 145 #include <gst/gstmarshal.h>
 146 #include <gst/gst_private.h>
 147 #include <gst/gst-i18n-lib.h>
 148
 149 GST_DEBUG_CATEGORY_STATIC (gst_base_sink_debug);
 150 #define GST_CAT_DEFAULT gst_base_sink_debug
 151
 152 #define GST_BASE_SINK_GET_PRIVATE(obj)  \
 153    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BASE_SINK, GstBaseSinkPrivate))
 154
 155 #define GST_FLOW_STEP GST_FLOW_CUSTOM_ERROR
 156
 157 typedef struct
 158 {
 159   gboolean valid;               /* if this info is valid */
 160   guint32 seqnum;               /* the seqnum of the STEP event */
 161   GstFormat format;             /* the format of the amount */
 162   guint64 amount;               /* the total amount of data to skip */
 163   guint64 position;             /* the position in the stepped data */
 164   guint64 duration;             /* the duration in time of the skipped data */
 165   guint64 start;                /* running_time of the start */
 166   gdouble rate;                 /* rate of skipping */
 167   gdouble start_rate;           /* rate before skipping */
 168   gboolean flush;               /* if this was a flushing step */
 169   gboolean intermediate;        /* if this is an intermediate step */
 170   gboolean need_preroll;        /* if we need preroll after this step */
 171 } GstStepInfo;
 172
 173 /* FIXME, some stuff in ABI.data and other in Private...
 174  * Make up your mind please.
 175  */
 176 struct _GstBaseSinkPrivate
 177 {
 178   gint qos_enabled;             /* ATOMIC */
 179   gboolean async_enabled;
 180   GstClockTimeDiff ts_offset;
 181   GstClockTime render_delay;
 182
 183   /* start, stop of current buffer, stream time, used to report position */
 184   GstClockTime current_sstart;
 185   GstClockTime current_sstop;
 186
 187   /* start, stop and jitter of current buffer, running time */
 188   GstClockTime current_rstart;
 189   GstClockTime current_rstop;
 190   GstClockTimeDiff current_jitter;
 191
 192   /* EOS sync time in running time */
 193   GstClockTime eos_rtime;
 194
 195   /* last buffer that arrived in time, running time */
 196   GstClockTime last_in_time;
 197   /* when the last buffer left the sink, running time */
 198   GstClockTime last_left;
 199
 200   /* running averages go here these are done on running time */
 201   GstClockTime avg_pt;
 202   GstClockTime avg_duration;
 203   gdouble avg_rate;
 204
 205   /* these are done on system time. avg_jitter and avg_render are
 206    * compared to eachother to see if the rendering time takes a
 207    * huge amount of the processing, If so we are flooded with
 208    * buffers. */
 209   GstClockTime last_left_systime;
 210   GstClockTime avg_jitter;
 211   GstClockTime start, stop;
 212   GstClockTime avg_render;
 213
 214   /* number of rendered and dropped frames */
 215   guint64 rendered;
 216   guint64 dropped;
 217
 218   /* latency stuff */
 219   GstClockTime latency;
 220
 221   /* if we already commited the state */
 222   gboolean commited;
 223
 224   /* when we received EOS */
 225   gboolean received_eos;
 226
 227   /* when we are prerolled and able to report latency */
 228   gboolean have_latency;
 229
 230   /* the last buffer we prerolled or rendered. Useful for making snapshots */
 231   GstBuffer *last_buffer;
 232
 233   /* caps for pull based scheduling */
 234   GstCaps *pull_caps;
 235
 236   /* blocksize for pulling */
 237   guint blocksize;
 238
 239   gboolean discont;
 240
 241   /* seqnum of the stream */
 242   guint32 seqnum;
 243
 244   gboolean call_preroll;
 245   gboolean step_unlock;
 246
 247   /* we have a pending and a current step operation */
 248   GstStepInfo current_step;
 249   GstStepInfo pending_step;
 250 };
 251
 252 #define DO_RUNNING_AVG(avg,val,size) (((val) + ((size)-1) * (avg)) / (size))
 253
 254 /* generic running average, this has a neutral window size */
 255 #define UPDATE_RUNNING_AVG(avg,val)   DO_RUNNING_AVG(avg,val,8)
 256
 257 /* the windows for these running averages are experimentally obtained.
 258  * possitive values get averaged more while negative values use a small
 259  * window so we can react faster to badness. */
 260 #define UPDATE_RUNNING_AVG_P(avg,val) DO_RUNNING_AVG(avg,val,16)
 261 #define UPDATE_RUNNING_AVG_N(avg,val) DO_RUNNING_AVG(avg,val,4)
 262
 263 /* BaseSink properties */
 264
 265 #define DEFAULT_CAN_ACTIVATE_PULL FALSE /* fixme: enable me */
 266 #define DEFAULT_CAN_ACTIVATE_PUSH TRUE
 267
 268 #define DEFAULT_PREROLL_QUEUE_LEN       0
 269 #define DEFAULT_SYNC                    TRUE
 270 #define DEFAULT_MAX_LATENESS            -1
 271 #define DEFAULT_QOS                     FALSE
 272 #define DEFAULT_ASYNC                   TRUE
 273 #define DEFAULT_TS_OFFSET               0
 274 #define DEFAULT_BLOCKSIZE               4096
 275 #define DEFAULT_RENDER_DELAY            0
 276
 277 enum
 278 {
 279   PROP_0,
 280   PROP_PREROLL_QUEUE_LEN,
 281   PROP_SYNC,
 282   PROP_MAX_LATENESS,
 283   PROP_QOS,
 284   PROP_ASYNC,
 285   PROP_TS_OFFSET,
 286   PROP_LAST_BUFFER,
 287   PROP_BLOCKSIZE,
 288   PROP_RENDER_DELAY,
 289   PROP_LAST
 290 };
 291
 292 static GstElementClass *parent_class = NULL;
 293
 294 static void gst_base_sink_class_init (GstBaseSinkClass * klass);
 295 static void gst_base_sink_init (GstBaseSink * trans, gpointer g_class);
 296 static void gst_base_sink_finalize (GObject * object);
 297
 298 GType
 299 gst_base_sink_get_type (void)
 300 {
 301   static volatile gsize base_sink_type = 0;
 302
 303   if (g_once_init_enter (&base_sink_type)) {
 304     GType _type;
 305     static const GTypeInfo base_sink_info = {
 306       sizeof (GstBaseSinkClass),
 307       NULL,
 308       NULL,
 309       (GClassInitFunc) gst_base_sink_class_init,
 310       NULL,
 311       NULL,
 312       sizeof (GstBaseSink),
 313       0,
 314       (GInstanceInitFunc) gst_base_sink_init,
 315     };
 316
 317     _type = g_type_register_static (GST_TYPE_ELEMENT,
 318         "GstBaseSink", &base_sink_info, G_TYPE_FLAG_ABSTRACT);
 319     g_once_init_leave (&base_sink_type, _type);
 320   }
 321   return base_sink_type;
 322 }
 323
 324 static void gst_base_sink_set_property (GObject * object, guint prop_id,
 325     const GValue * value, GParamSpec * pspec);
 326 static void gst_base_sink_get_property (GObject * object, guint prop_id,
 327     GValue * value, GParamSpec * pspec);
 328
 329 static gboolean gst_base_sink_send_event (GstElement * element,
 330     GstEvent * event);
 331 static gboolean gst_base_sink_query (GstElement * element, GstQuery * query);
 332
 333 static GstCaps *gst_base_sink_get_caps (GstBaseSink * sink);
 334 static gboolean gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps);
 335 static GstFlowReturn gst_base_sink_buffer_alloc (GstBaseSink * sink,
 336     guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
 337 static void gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
 338     GstClockTime * start, GstClockTime * end);
 339 static gboolean gst_base_sink_set_flushing (GstBaseSink * basesink,
 340     GstPad * pad, gboolean flushing);
 341 static gboolean gst_base_sink_default_activate_pull (GstBaseSink * basesink,
 342     gboolean active);
 343 static gboolean gst_base_sink_default_do_seek (GstBaseSink * sink,
 344     GstSegment * segment);
 345 static gboolean gst_base_sink_default_prepare_seek_segment (GstBaseSink * sink,
 346     GstEvent * event, GstSegment * segment);
 347
 348 static GstStateChangeReturn gst_base_sink_change_state (GstElement * element,
 349     GstStateChange transition);
 350
 351 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
 352 static void gst_base_sink_loop (GstPad * pad);
 353 static gboolean gst_base_sink_pad_activate (GstPad * pad);
 354 static gboolean gst_base_sink_pad_activate_push (GstPad * pad, gboolean active);
 355 static gboolean gst_base_sink_pad_activate_pull (GstPad * pad, gboolean active);
 356 static gboolean gst_base_sink_event (GstPad * pad, GstEvent * event);
 357 static gboolean gst_base_sink_peer_query (GstBaseSink * sink, GstQuery * query);
 358
 359 static gboolean gst_base_sink_negotiate_pull (GstBaseSink * basesink);
 360
 361 /* check if an object was too late */
 362 static gboolean gst_base_sink_is_too_late (GstBaseSink * basesink,
 363     GstMiniObject * obj, GstClockTime start, GstClockTime stop,
 364     GstClockReturn status, GstClockTimeDiff jitter);
 365 static GstFlowReturn gst_base_sink_preroll_object (GstBaseSink * basesink,
 366     GstMiniObject * obj);
 367
 368 static void
 369 gst_base_sink_class_init (GstBaseSinkClass * klass)
 370 {
 371   GObjectClass *gobject_class;
 372   GstElementClass *gstelement_class;
 373
 374   gobject_class = G_OBJECT_CLASS (klass);
 375   gstelement_class = GST_ELEMENT_CLASS (klass);
 376
 377   GST_DEBUG_CATEGORY_INIT (gst_base_sink_debug, "basesink", 0,
 378       "basesink element");
 379
 380   g_type_class_add_private (klass, sizeof (GstBaseSinkPrivate));
 381
 382   parent_class = g_type_class_peek_parent (klass);
 383
 384   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_sink_finalize);
 385   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_base_sink_set_property);
 386   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_base_sink_get_property);
 387
 388   /* FIXME, this next value should be configured using an event from the
 389    * upstream element, ie, the BUFFER_SIZE event. */
 390   g_object_class_install_property (gobject_class, PROP_PREROLL_QUEUE_LEN,
 391       g_param_spec_uint ("preroll-queue-len", "Preroll queue length",
 392           "Number of buffers to queue during preroll", 0, G_MAXUINT,
 393           DEFAULT_PREROLL_QUEUE_LEN,
 394           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 395
 396   g_object_class_install_property (gobject_class, PROP_SYNC,
 397       g_param_spec_boolean ("sync", "Sync", "Sync on the clock", DEFAULT_SYNC,
 398           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 399
 400   g_object_class_install_property (gobject_class, PROP_MAX_LATENESS,
 401       g_param_spec_int64 ("max-lateness", "Max Lateness",
 402           "Maximum number of nanoseconds that a buffer can be late before it "
 403           "is dropped (-1 unlimited)", -1, G_MAXINT64, DEFAULT_MAX_LATENESS,
 404           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 405
 406   g_object_class_install_property (gobject_class, PROP_QOS,
 407       g_param_spec_boolean ("qos", "Qos",
 408           "Generate Quality-of-Service events upstream", DEFAULT_QOS,
 409           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 410   /**
 411    * GstBaseSink:async
 412    *
 413    * If set to #TRUE, the basesink will perform asynchronous state changes.
 414    * When set to #FALSE, the sink will not signal the parent when it prerolls.
 415    * Use this option when dealing with sparse streams or when synchronisation is
 416    * not required.
 417    *
 418    * Since: 0.10.15
 419    */
 420   g_object_class_install_property (gobject_class, PROP_ASYNC,
 421       g_param_spec_boolean ("async", "Async",
 422           "Go asynchronously to PAUSED", DEFAULT_ASYNC,
 423           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 424   /**
 425    * GstBaseSink:ts-offset
 426    *
 427    * Controls the final synchronisation, a negative value will render the buffer
 428    * earlier while a positive value delays playback. This property can be
 429    * used to fix synchronisation in bad files.
 430    *
 431    * Since: 0.10.15
 432    */
 433   g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
 434       g_param_spec_int64 ("ts-offset", "TS Offset",
 435           "Timestamp offset in nanoseconds", G_MININT64, G_MAXINT64,
 436           DEFAULT_TS_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 437   /**
 438    * GstBaseSink:last-buffer
 439    *
 440    * The last buffer that arrived in the sink and was used for preroll or for
 441    * rendering. This property can be used to generate thumbnails. This property
 442    * can be NULL when the sink has not yet received a bufer.
 443    *
 444    * Since: 0.10.15
 445    */
 446   g_object_class_install_property (gobject_class, PROP_LAST_BUFFER,
 447       gst_param_spec_mini_object ("last-buffer", "Last Buffer",
 448           "The last buffer received in the sink", GST_TYPE_BUFFER,
 449           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 450   /**
 451    * GstBaseSink:blocksize
 452    *
 453    * The amount of bytes to pull when operating in pull mode.
 454    *
 455    * Since: 0.10.22
 456    */
 457   g_object_class_install_property (gobject_class, PROP_BLOCKSIZE,
 458       g_param_spec_uint ("blocksize", "Block size",
 459           "Size in bytes to pull per buffer (0 = default)", 0, G_MAXUINT,
 460           DEFAULT_BLOCKSIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 461   /**
 462    * GstBaseSink:render-delay
 463    *
 464    * The additional delay between synchronisation and actual rendering of the
 465    * media. This property will add additional latency to the device in order to
 466    * make other sinks compensate for the delay.
 467    *
 468    * Since: 0.10.22
 469    */
 470   g_object_class_install_property (gobject_class, PROP_RENDER_DELAY,
 471       g_param_spec_uint64 ("render-delay", "Render Delay",
 472           "Additional render delay of the sink in nanoseconds", 0, G_MAXUINT64,
 473           DEFAULT_RENDER_DELAY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 474
 475   gstelement_class->change_state =
 476       GST_DEBUG_FUNCPTR (gst_base_sink_change_state);
 477   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_base_sink_send_event);
 478   gstelement_class->query = GST_DEBUG_FUNCPTR (gst_base_sink_query);
 479
 480   klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_sink_get_caps);
 481   klass->set_caps = GST_DEBUG_FUNCPTR (gst_base_sink_set_caps);
 482   klass->buffer_alloc = GST_DEBUG_FUNCPTR (gst_base_sink_buffer_alloc);
 483   klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
 484   klass->activate_pull =
 485       GST_DEBUG_FUNCPTR (gst_base_sink_default_activate_pull);
 486 }
 487
 488 static GstCaps *
 489 gst_base_sink_pad_getcaps (GstPad * pad)
 490 {
 491   GstBaseSinkClass *bclass;
 492   GstBaseSink *bsink;
 493   GstCaps *caps = NULL;
 494
 495   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 496   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 497
 498   if (bsink->pad_mode == GST_ACTIVATE_PULL) {
 499     /* if we are operating in pull mode we only accept the negotiated caps */
 500     GST_OBJECT_LOCK (pad);
 501     if ((caps = GST_PAD_CAPS (pad)))
 502       gst_caps_ref (caps);
 503     GST_OBJECT_UNLOCK (pad);
 504   }
 505   if (caps == NULL) {
 506     if (bclass->get_caps)
 507       caps = bclass->get_caps (bsink);
 508
 509     if (caps == NULL) {
 510       GstPadTemplate *pad_template;
 511
 512       pad_template =
 513           gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass),
 514           "sink");
 515       if (pad_template != NULL) {
 516         caps = gst_caps_ref (gst_pad_template_get_caps (pad_template));
 517       }
 518     }
 519   }
 520   gst_object_unref (bsink);
 521
 522   return caps;
 523 }
 524
 525 static gboolean
 526 gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps)
 527 {
 528   GstBaseSinkClass *bclass;
 529   GstBaseSink *bsink;
 530   gboolean res = TRUE;
 531
 532   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 533   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 534
 535   if (res && bclass->set_caps)
 536     res = bclass->set_caps (bsink, caps);
 537
 538   gst_object_unref (bsink);
 539
 540   return res;
 541 }
 542
 543 static void
 544 gst_base_sink_pad_fixate (GstPad * pad, GstCaps * caps)
 545 {
 546   GstBaseSinkClass *bclass;
 547   GstBaseSink *bsink;
 548
 549   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 550   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 551
 552   if (bclass->fixate)
 553     bclass->fixate (bsink, caps);
 554
 555   gst_object_unref (bsink);
 556 }
 557
 558 static GstFlowReturn
 559 gst_base_sink_pad_buffer_alloc (GstPad * pad, guint64 offset, guint size,
 560     GstCaps * caps, GstBuffer ** buf)
 561 {
 562   GstBaseSinkClass *bclass;
 563   GstBaseSink *bsink;
 564   GstFlowReturn result = GST_FLOW_OK;
 565
 566   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 567   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 568
 569   if (bclass->buffer_alloc)
 570     result = bclass->buffer_alloc (bsink, offset, size, caps, buf);
 571   else
 572     *buf = NULL;                /* fallback in gstpad.c will allocate generic buffer */
 573
 574   gst_object_unref (bsink);
 575
 576   return result;
 577 }
 578
 579 static void
 580 gst_base_sink_init (GstBaseSink * basesink, gpointer g_class)
 581 {
 582   GstPadTemplate *pad_template;
 583   GstBaseSinkPrivate *priv;
 584
 585   basesink->priv = priv = GST_BASE_SINK_GET_PRIVATE (basesink);
 586
 587   pad_template =
 588       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "sink");
 589   g_return_if_fail (pad_template != NULL);
 590
 591   basesink->sinkpad = gst_pad_new_from_template (pad_template, "sink");
 592
 593   gst_pad_set_getcaps_function (basesink->sinkpad,
 594       GST_DEBUG_FUNCPTR (gst_base_sink_pad_getcaps));
 595   gst_pad_set_setcaps_function (basesink->sinkpad,
 596       GST_DEBUG_FUNCPTR (gst_base_sink_pad_setcaps));
 597   gst_pad_set_fixatecaps_function (basesink->sinkpad,
 598       GST_DEBUG_FUNCPTR (gst_base_sink_pad_fixate));
 599   gst_pad_set_bufferalloc_function (basesink->sinkpad,
 600       GST_DEBUG_FUNCPTR (gst_base_sink_pad_buffer_alloc));
 601   gst_pad_set_activate_function (basesink->sinkpad,
 602       GST_DEBUG_FUNCPTR (gst_base_sink_pad_activate));
 603   gst_pad_set_activatepush_function (basesink->sinkpad,
 604       GST_DEBUG_FUNCPTR (gst_base_sink_pad_activate_push));
 605   gst_pad_set_activatepull_function (basesink->sinkpad,
 606       GST_DEBUG_FUNCPTR (gst_base_sink_pad_activate_pull));
 607   gst_pad_set_event_function (basesink->sinkpad,
 608       GST_DEBUG_FUNCPTR (gst_base_sink_event));
 609   gst_pad_set_chain_function (basesink->sinkpad,
 610       GST_DEBUG_FUNCPTR (gst_base_sink_chain));
 611   gst_element_add_pad (GST_ELEMENT_CAST (basesink), basesink->sinkpad);
 612
 613   basesink->pad_mode = GST_ACTIVATE_NONE;
 614   basesink->preroll_queue = g_queue_new ();
 615   basesink->abidata.ABI.clip_segment = gst_segment_new ();
 616   priv->have_latency = FALSE;
 617
 618   basesink->can_activate_push = DEFAULT_CAN_ACTIVATE_PUSH;
 619   basesink->can_activate_pull = DEFAULT_CAN_ACTIVATE_PULL;
 620
 621   basesink->sync = DEFAULT_SYNC;
 622   basesink->abidata.ABI.max_lateness = DEFAULT_MAX_LATENESS;
 623   g_atomic_int_set (&priv->qos_enabled, DEFAULT_QOS);
 624   priv->async_enabled = DEFAULT_ASYNC;
 625   priv->ts_offset = DEFAULT_TS_OFFSET;
 626   priv->render_delay = DEFAULT_RENDER_DELAY;
 627   priv->blocksize = DEFAULT_BLOCKSIZE;
 628
 629   GST_OBJECT_FLAG_SET (basesink, GST_ELEMENT_IS_SINK);
 630 }
 631
 632 static void
 633 gst_base_sink_finalize (GObject * object)
 634 {
 635   GstBaseSink *basesink;
 636
 637   basesink = GST_BASE_SINK (object);
 638
 639   g_queue_free (basesink->preroll_queue);
 640   gst_segment_free (basesink->abidata.ABI.clip_segment);
 641
 642   G_OBJECT_CLASS (parent_class)->finalize (object);
 643 }
 644
 645 /**
 646  * gst_base_sink_set_sync:
 647  * @sink: the sink
 648  * @sync: the new sync value.
 649  *
 650  * Configures @sink to synchronize on the clock or not. When
 651  * @sync is FALSE, incomming samples will be played as fast as
 652  * possible. If @sync is TRUE, the timestamps of the incomming
 653  * buffers will be used to schedule the exact render time of its
 654  * contents.
 655  *
 656  * Since: 0.10.4
 657  */
 658 void
 659 gst_base_sink_set_sync (GstBaseSink * sink, gboolean sync)
 660 {
 661   g_return_if_fail (GST_IS_BASE_SINK (sink));
 662
 663   GST_OBJECT_LOCK (sink);
 664   sink->sync = sync;
 665   GST_OBJECT_UNLOCK (sink);
 666 }
 667
 668 /**
 669  * gst_base_sink_get_sync:
 670  * @sink: the sink
 671  *
 672  * Checks if @sink is currently configured to synchronize against the
 673  * clock.
 674  *
 675  * Returns: TRUE if the sink is configured to synchronize against the clock.
 676  *
 677  * Since: 0.10.4
 678  */
 679 gboolean
 680 gst_base_sink_get_sync (GstBaseSink * sink)
 681 {
 682   gboolean res;
 683
 684   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
 685
 686   GST_OBJECT_LOCK (sink);
 687   res = sink->sync;
 688   GST_OBJECT_UNLOCK (sink);
 689
 690   return res;
 691 }
 692
 693 /**
 694  * gst_base_sink_set_max_lateness:
 695  * @sink: the sink
 696  * @max_lateness: the new max lateness value.
 697  *
 698  * Sets the new max lateness value to @max_lateness. This value is
 699  * used to decide if a buffer should be dropped or not based on the
 700  * buffer timestamp and the current clock time. A value of -1 means
 701  * an unlimited time.
 702  *
 703  * Since: 0.10.4
 704  */
 705 void
 706 gst_base_sink_set_max_lateness (GstBaseSink * sink, gint64 max_lateness)
 707 {
 708   g_return_if_fail (GST_IS_BASE_SINK (sink));
 709
 710   GST_OBJECT_LOCK (sink);
 711   sink->abidata.ABI.max_lateness = max_lateness;
 712   GST_OBJECT_UNLOCK (sink);
 713 }
 714
 715 /**
 716  * gst_base_sink_get_max_lateness:
 717  * @sink: the sink
 718  *
 719  * Gets the max lateness value. See gst_base_sink_set_max_lateness for
 720  * more details.
 721  *
 722  * Returns: The maximum time in nanoseconds that a buffer can be late
 723  * before it is dropped and not rendered. A value of -1 means an
 724  * unlimited time.
 725  *
 726  * Since: 0.10.4
 727  */
 728 gint64
 729 gst_base_sink_get_max_lateness (GstBaseSink * sink)
 730 {
 731   gint64 res;
 732
 733   g_return_val_if_fail (GST_IS_BASE_SINK (sink), -1);
 734
 735   GST_OBJECT_LOCK (sink);
 736   res = sink->abidata.ABI.max_lateness;
 737   GST_OBJECT_UNLOCK (sink);
 738
 739   return res;
 740 }
 741
 742 /**
 743  * gst_base_sink_set_qos_enabled:
 744  * @sink: the sink
 745  * @enabled: the new qos value.
 746  *
 747  * Configures @sink to send Quality-of-Service events upstream.
 748  *
 749  * Since: 0.10.5
 750  */
 751 void
 752 gst_base_sink_set_qos_enabled (GstBaseSink * sink, gboolean enabled)
 753 {
 754   g_return_if_fail (GST_IS_BASE_SINK (sink));
 755
 756   g_atomic_int_set (&sink->priv->qos_enabled, enabled);
 757 }
 758
 759 /**
 760  * gst_base_sink_is_qos_enabled:
 761  * @sink: the sink
 762  *
 763  * Checks if @sink is currently configured to send Quality-of-Service events
 764  * upstream.
 765  *
 766  * Returns: TRUE if the sink is configured to perform Quality-of-Service.
 767  *
 768  * Since: 0.10.5
 769  */
 770 gboolean
 771 gst_base_sink_is_qos_enabled (GstBaseSink * sink)
 772 {
 773   gboolean res;
 774
 775   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
 776
 777   res = g_atomic_int_get (&sink->priv->qos_enabled);
 778
 779   return res;
 780 }
 781
 782 /**
 783  * gst_base_sink_set_async_enabled:
 784  * @sink: the sink
 785  * @enabled: the new async value.
 786  *
 787  * Configures @sink to perform all state changes asynchronusly. When async is
 788  * disabled, the sink will immediatly go to PAUSED instead of waiting for a
 789  * preroll buffer. This feature is usefull if the sink does not synchronize
 790  * against the clock or when it is dealing with sparse streams.
 791  *
 792  * Since: 0.10.15
 793  */
 794 void
 795 gst_base_sink_set_async_enabled (GstBaseSink * sink, gboolean enabled)
 796 {
 797   g_return_if_fail (GST_IS_BASE_SINK (sink));
 798
 799   GST_PAD_PREROLL_LOCK (sink->sinkpad);
 800   sink->priv->async_enabled = enabled;
 801   GST_LOG_OBJECT (sink, "set async enabled to %d", enabled);
 802   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
 803 }
 804
 805 /**
 806  * gst_base_sink_is_async_enabled:
 807  * @sink: the sink
 808  *
 809  * Checks if @sink is currently configured to perform asynchronous state
 810  * changes to PAUSED.
 811  *
 812  * Returns: TRUE if the sink is configured to perform asynchronous state
 813  * changes.
 814  *
 815  * Since: 0.10.15
 816  */
 817 gboolean
 818 gst_base_sink_is_async_enabled (GstBaseSink * sink)
 819 {
 820   gboolean res;
 821
 822   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
 823
 824   GST_PAD_PREROLL_LOCK (sink->sinkpad);
 825   res = sink->priv->async_enabled;
 826   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
 827
 828   return res;
 829 }
 830
 831 /**
 832  * gst_base_sink_set_ts_offset:
 833  * @sink: the sink
 834  * @offset: the new offset
 835  *
 836  * Adjust the synchronisation of @sink with @offset. A negative value will
 837  * render buffers earlier than their timestamp. A positive value will delay
 838  * rendering. This function can be used to fix playback of badly timestamped
 839  * buffers.
 840  *
 841  * Since: 0.10.15
 842  */
 843 void
 844 gst_base_sink_set_ts_offset (GstBaseSink * sink, GstClockTimeDiff offset)
 845 {
 846   g_return_if_fail (GST_IS_BASE_SINK (sink));
 847
 848   GST_OBJECT_LOCK (sink);
 849   sink->priv->ts_offset = offset;
 850   GST_LOG_OBJECT (sink, "set time offset to %" G_GINT64_FORMAT, offset);
 851   GST_OBJECT_UNLOCK (sink);
 852 }
 853
 854 /**
 855  * gst_base_sink_get_ts_offset:
 856  * @sink: the sink
 857  *
 858  * Get the synchronisation offset of @sink.
 859  *
 860  * Returns: The synchronisation offset.
 861  *
 862  * Since: 0.10.15
 863  */
 864 GstClockTimeDiff
 865 gst_base_sink_get_ts_offset (GstBaseSink * sink)
 866 {
 867   GstClockTimeDiff res;
 868
 869   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
 870
 871   GST_OBJECT_LOCK (sink);
 872   res = sink->priv->ts_offset;
 873   GST_OBJECT_UNLOCK (sink);
 874
 875   return res;
 876 }
 877
 878 /**
 879  * gst_base_sink_get_last_buffer:
 880  * @sink: the sink
 881  *
 882  * Get the last buffer that arrived in the sink and was used for preroll or for
 883  * rendering. This property can be used to generate thumbnails.
 884  *
 885  * The #GstCaps on the buffer can be used to determine the type of the buffer.
 886  *
 887  * Returns: a #GstBuffer. gst_buffer_unref() after usage. This function returns
 888  * NULL when no buffer has arrived in the sink yet or when the sink is not in
 889  * PAUSED or PLAYING.
 890  *
 891  * Since: 0.10.15
 892  */
 893 GstBuffer *
 894 gst_base_sink_get_last_buffer (GstBaseSink * sink)
 895 {
 896   GstBuffer *res;
 897
 898   g_return_val_if_fail (GST_IS_BASE_SINK (sink), NULL);
 899
 900   GST_OBJECT_LOCK (sink);
 901   if ((res = sink->priv->last_buffer))
 902     gst_buffer_ref (res);
 903   GST_OBJECT_UNLOCK (sink);
 904
 905   return res;
 906 }
 907
 908 static void
 909 gst_base_sink_set_last_buffer (GstBaseSink * sink, GstBuffer * buffer)
 910 {
 911   GstBuffer *old;
 912
 913   GST_OBJECT_LOCK (sink);
 914   old = sink->priv->last_buffer;
 915   if (G_LIKELY (old != buffer)) {
 916     GST_DEBUG_OBJECT (sink, "setting last buffer to %p", buffer);
 917     if (G_LIKELY (buffer))
 918       gst_buffer_ref (buffer);
 919     sink->priv->last_buffer = buffer;
 920   } else {
 921     old = NULL;
 922   }
 923   GST_OBJECT_UNLOCK (sink);
 924
 925   /* avoid unreffing with the lock because cleanup code might want to take the
 926    * lock too */
 927   if (G_LIKELY (old))
 928     gst_buffer_unref (old);
 929 }
 930
 931 /**
 932  * gst_base_sink_get_latency:
 933  * @sink: the sink
 934  *
 935  * Get the currently configured latency.
 936  *
 937  * Returns: The configured latency.
 938  *
 939  * Since: 0.10.12
 940  */
 941 GstClockTime
 942 gst_base_sink_get_latency (GstBaseSink * sink)
 943 {
 944   GstClockTime res;
 945
 946   GST_OBJECT_LOCK (sink);
 947   res = sink->priv->latency;
 948   GST_OBJECT_UNLOCK (sink);
 949
 950   return res;
 951 }
 952
 953 /**
 954  * gst_base_sink_query_latency:
 955  * @sink: the sink
 956  * @live: if the sink is live
 957  * @upstream_live: if an upstream element is live
 958  * @min_latency: the min latency of the upstream elements
 959  * @max_latency: the max latency of the upstream elements
 960  *
 961  * Query the sink for the latency parameters. The latency will be queried from
 962  * the upstream elements. @live will be TRUE if @sink is configured to
 963  * synchronize against the clock. @upstream_live will be TRUE if an upstream
 964  * element is live.
 965  *
 966  * If both @live and @upstream_live are TRUE, the sink will want to compensate
 967  * for the latency introduced by the upstream elements by setting the
 968  * @min_latency to a strictly possitive value.
 969  *
 970  * This function is mostly used by subclasses.
 971  *
 972  * Returns: TRUE if the query succeeded.
 973  *
 974  * Since: 0.10.12
 975  */
 976 gboolean
 977 gst_base_sink_query_latency (GstBaseSink * sink, gboolean * live,
 978     gboolean * upstream_live, GstClockTime * min_latency,
 979     GstClockTime * max_latency)
 980 {
 981   gboolean l, us_live, res, have_latency;
 982   GstClockTime min, max, render_delay;
 983   GstQuery *query;
 984   GstClockTime us_min, us_max;
 985
 986   /* we are live when we sync to the clock */
 987   GST_OBJECT_LOCK (sink);
 988   l = sink->sync;
 989   have_latency = sink->priv->have_latency;
 990   render_delay = sink->priv->render_delay;
 991   GST_OBJECT_UNLOCK (sink);
 992
 993   /* assume no latency */
 994   min = 0;
 995   max = -1;
 996   us_live = FALSE;
 997
 998   if (have_latency) {
 999     GST_DEBUG_OBJECT (sink, "we are ready for LATENCY query");
1000     /* we are ready for a latency query this is when we preroll or when we are
1001      * not async. */
1002     query = gst_query_new_latency ();
1003
1004     /* ask the peer for the latency */
1005     if ((res = gst_base_sink_peer_query (sink, query))) {
1006       /* get upstream min and max latency */
1007       gst_query_parse_latency (query, &us_live, &us_min, &us_max);
1008
1009       if (us_live) {
1010         /* upstream live, use its latency, subclasses should use these
1011          * values to create the complete latency. */
1012         min = us_min;
1013         max = us_max;
1014       }
1015       if (l) {
1016         /* we need to add the render delay if we are live */
1017         if (min != -1)
1018           min += render_delay;
1019         if (max != -1)
1020           max += render_delay;
1021       }
1022     }
1023     gst_query_unref (query);
1024   } else {
1025     GST_DEBUG_OBJECT (sink, "we are not yet ready for LATENCY query");
1026     res = FALSE;
1027   }
1028
1029   /* not live, we tried to do the query, if it failed we return TRUE anyway */
1030   if (!res) {
1031     if (!l) {
1032       res = TRUE;
1033       GST_DEBUG_OBJECT (sink, "latency query failed but we are not live");
1034     } else {
1035       GST_DEBUG_OBJECT (sink, "latency query failed and we are live");
1036     }
1037   }
1038
1039   if (res) {
1040     GST_DEBUG_OBJECT (sink, "latency query: live: %d, have_latency %d,"
1041         " upstream: %d, min %" GST_TIME_FORMAT ", max %" GST_TIME_FORMAT, l,
1042         have_latency, us_live, GST_TIME_ARGS (min), GST_TIME_ARGS (max));
1043
1044     if (live)
1045       *live = l;
1046     if (upstream_live)
1047       *upstream_live = us_live;
1048     if (min_latency)
1049       *min_latency = min;
1050     if (max_latency)
1051       *max_latency = max;
1052   }
1053   return res;
1054 }
1055
1056 /**
1057  * gst_base_sink_set_render_delay:
1058  * @sink: a #GstBaseSink
1059  * @delay: the new delay
1060  *
1061  * Set the render delay in @sink to @delay. The render delay is the time
1062  * between actual rendering of a buffer and its synchronisation time. Some
1063  * devices might delay media rendering which can be compensated for with this
1064  * function.
1065  *
1066  * After calling this function, this sink will report additional latency and
1067  * other sinks will adjust their latency to delay the rendering of their media.
1068  *
1069  * This function is usually called by subclasses.
1070  *
1071  * Since: 0.10.21
1072  */
1073 void
1074 gst_base_sink_set_render_delay (GstBaseSink * sink, GstClockTime delay)
1075 {
1076   GstClockTime old_render_delay;
1077
1078   g_return_if_fail (GST_IS_BASE_SINK (sink));
1079
1080   GST_OBJECT_LOCK (sink);
1081   old_render_delay = sink->priv->render_delay;
1082   sink->priv->render_delay = delay;
1083   GST_LOG_OBJECT (sink, "set render delay to %" GST_TIME_FORMAT,
1084       GST_TIME_ARGS (delay));
1085   GST_OBJECT_UNLOCK (sink);
1086
1087   if (delay != old_render_delay) {
1088     GST_DEBUG_OBJECT (sink, "posting latency changed");
1089     gst_element_post_message (GST_ELEMENT_CAST (sink),
1090         gst_message_new_latency (GST_OBJECT_CAST (sink)));
1091   }
1092 }
1093
1094 /**
1095  * gst_base_sink_get_render_delay:
1096  * @sink: a #GstBaseSink
1097  *
1098  * Get the render delay of @sink. see gst_base_sink_set_render_delay() for more
1099  * information about the render delay.
1100  *
1101  * Returns: the render delay of @sink.
1102  *
1103  * Since: 0.10.21
1104  */
1105 GstClockTime
1106 gst_base_sink_get_render_delay (GstBaseSink * sink)
1107 {
1108   GstClockTimeDiff res;
1109
1110   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
1111
1112   GST_OBJECT_LOCK (sink);
1113   res = sink->priv->render_delay;
1114   GST_OBJECT_UNLOCK (sink);
1115
1116   return res;
1117 }
1118
1119 /**
1120  * gst_base_sink_set_blocksize:
1121  * @sink: a #GstBaseSink
1122  * @blocksize: the blocksize in bytes
1123  *
1124  * Set the number of bytes that the sink will pull when it is operating in pull
1125  * mode.
1126  *
1127  * Since: 0.10.22
1128  */
1129 void
1130 gst_base_sink_set_blocksize (GstBaseSink * sink, guint blocksize)
1131 {
1132   g_return_if_fail (GST_IS_BASE_SINK (sink));
1133
1134   GST_OBJECT_LOCK (sink);
1135   sink->priv->blocksize = blocksize;
1136   GST_LOG_OBJECT (sink, "set blocksize to %u", blocksize);
1137   GST_OBJECT_UNLOCK (sink);
1138 }
1139
1140 /**
1141  * gst_base_sink_get_blocksize:
1142  * @sink: a #GstBaseSink
1143  *
1144  * Get the number of bytes that the sink will pull when it is operating in pull
1145  * mode.
1146  *
1147  * Returns: the number of bytes @sink will pull in pull mode.
1148  *
1149  * Since: 0.10.22
1150  */
1151 guint
1152 gst_base_sink_get_blocksize (GstBaseSink * sink)
1153 {
1154   guint res;
1155
1156   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
1157
1158   GST_OBJECT_LOCK (sink);
1159   res = sink->priv->blocksize;
1160   GST_OBJECT_UNLOCK (sink);
1161
1162   return res;
1163 }
1164
1165 static void
1166 gst_base_sink_set_property (GObject * object, guint prop_id,
1167     const GValue * value, GParamSpec * pspec)
1168 {
1169   GstBaseSink *sink = GST_BASE_SINK (object);
1170
1171   switch (prop_id) {
1172     case PROP_PREROLL_QUEUE_LEN:
1173       /* preroll lock necessary to serialize with finish_preroll */
1174       GST_PAD_PREROLL_LOCK (sink->sinkpad);
1175       sink->preroll_queue_max_len = g_value_get_uint (value);
1176       GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
1177       break;
1178     case PROP_SYNC:
1179       gst_base_sink_set_sync (sink, g_value_get_boolean (value));
1180       break;
1181     case PROP_MAX_LATENESS:
1182       gst_base_sink_set_max_lateness (sink, g_value_get_int64 (value));
1183       break;
1184     case PROP_QOS:
1185       gst_base_sink_set_qos_enabled (sink, g_value_get_boolean (value));
1186       break;
1187     case PROP_ASYNC:
1188       gst_base_sink_set_async_enabled (sink, g_value_get_boolean (value));
1189       break;
1190     case PROP_TS_OFFSET:
1191       gst_base_sink_set_ts_offset (sink, g_value_get_int64 (value));
1192       break;
1193     case PROP_BLOCKSIZE:
1194       gst_base_sink_set_blocksize (sink, g_value_get_uint (value));
1195       break;
1196     case PROP_RENDER_DELAY:
1197       gst_base_sink_set_render_delay (sink, g_value_get_uint64 (value));
1198       break;
1199     default:
1200       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1201       break;
1202   }
1203 }
1204
1205 static void
1206 gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
1207     GParamSpec * pspec)
1208 {
1209   GstBaseSink *sink = GST_BASE_SINK (object);
1210
1211   switch (prop_id) {
1212     case PROP_PREROLL_QUEUE_LEN:
1213       GST_PAD_PREROLL_LOCK (sink->sinkpad);
1214       g_value_set_uint (value, sink->preroll_queue_max_len);
1215       GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
1216       break;
1217     case PROP_SYNC:
1218       g_value_set_boolean (value, gst_base_sink_get_sync (sink));
1219       break;
1220     case PROP_MAX_LATENESS:
1221       g_value_set_int64 (value, gst_base_sink_get_max_lateness (sink));
1222       break;
1223     case PROP_QOS:
1224       g_value_set_boolean (value, gst_base_sink_is_qos_enabled (sink));
1225       break;
1226     case PROP_ASYNC:
1227       g_value_set_boolean (value, gst_base_sink_is_async_enabled (sink));
1228       break;
1229     case PROP_TS_OFFSET:
1230       g_value_set_int64 (value, gst_base_sink_get_ts_offset (sink));
1231       break;
1232     case PROP_LAST_BUFFER:
1233       gst_value_take_buffer (value, gst_base_sink_get_last_buffer (sink));
1234       break;
1235     case PROP_BLOCKSIZE:
1236       g_value_set_uint (value, gst_base_sink_get_blocksize (sink));
1237       break;
1238     case PROP_RENDER_DELAY:
1239       g_value_set_uint64 (value, gst_base_sink_get_render_delay (sink));
1240       break;
1241     default:
1242       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1243       break;
1244   }
1245 }
1246
1247
1248 static GstCaps *
1249 gst_base_sink_get_caps (GstBaseSink * sink)
1250 {
1251   return NULL;
1252 }
1253
1254 static gboolean
1255 gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
1256 {
1257   return TRUE;
1258 }
1259
1260 static GstFlowReturn
1261 gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
1262     GstCaps * caps, GstBuffer ** buf)
1263 {
1264   *buf = NULL;
1265   return GST_FLOW_OK;
1266 }
1267
1268 /* with PREROLL_LOCK, STREAM_LOCK */
1269 static void
1270 gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
1271 {
1272   GstMiniObject *obj;
1273
1274   GST_DEBUG_OBJECT (basesink, "flushing queue %p", basesink);
1275   while ((obj = g_queue_pop_head (basesink->preroll_queue))) {
1276     GST_DEBUG_OBJECT (basesink, "popped %p", obj);
1277     gst_mini_object_unref (obj);
1278   }
1279   /* we can't have EOS anymore now */
1280   basesink->eos = FALSE;
1281   basesink->priv->received_eos = FALSE;
1282   basesink->have_preroll = FALSE;
1283   basesink->priv->step_unlock = FALSE;
1284   basesink->eos_queued = FALSE;
1285   basesink->preroll_queued = 0;
1286   basesink->buffers_queued = 0;
1287   basesink->events_queued = 0;
1288   /* can't report latency anymore until we preroll again */
1289   if (basesink->priv->async_enabled) {
1290     GST_OBJECT_LOCK (basesink);
1291     basesink->priv->have_latency = FALSE;
1292     GST_OBJECT_UNLOCK (basesink);
1293   }
1294   /* and signal any waiters now */
1295   GST_PAD_PREROLL_SIGNAL (pad);
1296 }
1297
1298 /* with STREAM_LOCK, configures given segment with the event information. */
1299 static void
1300 gst_base_sink_configure_segment (GstBaseSink * basesink, GstPad * pad,
1301     GstEvent * event, GstSegment * segment)
1302 {
1303   gboolean update;
1304   gdouble rate, arate;
1305   GstFormat format;
1306   gint64 start;
1307   gint64 stop;
1308   gint64 time;
1309
1310   /* the newsegment event is needed to bring the buffer timestamps to the
1311    * stream time and to drop samples outside of the playback segment. */
1312   gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format,
1313       &start, &stop, &time);
1314
1315   /* The segment is protected with both the STREAM_LOCK and the OBJECT_LOCK.
1316    * We protect with the OBJECT_LOCK so that we can use the values to
1317    * safely answer a POSITION query. */
1318   GST_OBJECT_LOCK (basesink);
1319   gst_segment_set_newsegment_full (segment, update, rate, arate, format, start,
1320       stop, time);
1321
1322   if (format == GST_FORMAT_TIME) {
1323     GST_DEBUG_OBJECT (basesink,
1324         "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
1325         "format GST_FORMAT_TIME, "
1326         "%" GST_TIME_FORMAT " -- %" GST_TIME_FORMAT
1327         ", time %" GST_TIME_FORMAT ", accum %" GST_TIME_FORMAT,
1328         update, rate, arate, GST_TIME_ARGS (segment->start),
1329         GST_TIME_ARGS (segment->stop), GST_TIME_ARGS (segment->time),
1330         GST_TIME_ARGS (segment->accum));
1331   } else {
1332     GST_DEBUG_OBJECT (basesink,
1333         "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
1334         "format %d, "
1335         "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
1336         G_GINT64_FORMAT ", accum %" G_GINT64_FORMAT, update, rate, arate,
1337         segment->format, segment->start, segment->stop, segment->time,
1338         segment->accum);
1339   }
1340   GST_OBJECT_UNLOCK (basesink);
1341 }
1342
1343 /* with PREROLL_LOCK, STREAM_LOCK */
1344 static gboolean
1345 gst_base_sink_commit_state (GstBaseSink * basesink)
1346 {
1347   /* commit state and proceed to next pending state */
1348   GstState current, next, pending, post_pending;
1349   gboolean post_paused = FALSE;
1350   gboolean post_async_done = FALSE;
1351   gboolean post_playing = FALSE;
1352
1353   /* we are certainly not playing async anymore now */
1354   basesink->playing_async = FALSE;
1355
1356   GST_OBJECT_LOCK (basesink);
1357   current = GST_STATE (basesink);
1358   next = GST_STATE_NEXT (basesink);
1359   pending = GST_STATE_PENDING (basesink);
1360   post_pending = pending;
1361
1362   switch (pending) {
1363     case GST_STATE_PLAYING:
1364     {
1365       GstBaseSinkClass *bclass;
1366       GstStateChangeReturn ret;
1367
1368       bclass = GST_BASE_SINK_GET_CLASS (basesink);
1369
1370       GST_DEBUG_OBJECT (basesink, "commiting state to PLAYING");
1371
1372       basesink->need_preroll = FALSE;
1373       post_async_done = TRUE;
1374       basesink->priv->commited = TRUE;
1375       post_playing = TRUE;
1376       /* post PAUSED too when we were READY */
1377       if (current == GST_STATE_READY) {
1378         post_paused = TRUE;
1379       }
1380
1381       /* make sure we notify the subclass of async playing */
1382       if (bclass->async_play) {
1383         GST_WARNING_OBJECT (basesink, "deprecated async_play");
1384         ret = bclass->async_play (basesink);
1385         if (ret == GST_STATE_CHANGE_FAILURE)
1386           goto async_failed;
1387       }
1388       break;
1389     }
1390     case GST_STATE_PAUSED:
1391       GST_DEBUG_OBJECT (basesink, "commiting state to PAUSED");
1392       post_paused = TRUE;
1393       post_async_done = TRUE;
1394       basesink->priv->commited = TRUE;
1395       post_pending = GST_STATE_VOID_PENDING;
1396       break;
1397     case GST_STATE_READY:
1398     case GST_STATE_NULL:
1399       goto stopping;
1400     case GST_STATE_VOID_PENDING:
1401       goto nothing_pending;
1402     default:
1403       break;
1404   }
1405
1406   /* we can report latency queries now */
1407   basesink->priv->have_latency = TRUE;
1408
1409   GST_STATE (basesink) = pending;
1410   GST_STATE_NEXT (basesink) = GST_STATE_VOID_PENDING;
1411   GST_STATE_PENDING (basesink) = GST_STATE_VOID_PENDING;
1412   GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_SUCCESS;
1413   GST_OBJECT_UNLOCK (basesink);
1414
1415   if (post_paused) {
1416     GST_DEBUG_OBJECT (basesink, "posting PAUSED state change message");
1417     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1418         gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
1419             current, next, post_pending));
1420   }
1421   if (post_async_done) {
1422     GST_DEBUG_OBJECT (basesink, "posting async-done message");
1423     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1424         gst_message_new_async_done (GST_OBJECT_CAST (basesink)));
1425   }
1426   if (post_playing) {
1427     GST_DEBUG_OBJECT (basesink, "posting PLAYING state change message");
1428     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1429         gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
1430             next, pending, GST_STATE_VOID_PENDING));
1431   }
1432
1433   GST_STATE_BROADCAST (basesink);
1434
1435   return TRUE;
1436
1437 nothing_pending:
1438   {
1439     /* Depending on the state, set our vars. We get in this situation when the
1440      * state change function got a change to update the state vars before the
1441      * streaming thread did. This is fine but we need to make sure that we
1442      * update the need_preroll var since it was TRUE when we got here and might
1443      * become FALSE if we got to PLAYING. */
1444     GST_DEBUG_OBJECT (basesink, "nothing to commit, now in %s",
1445         gst_element_state_get_name (current));
1446     switch (current) {
1447       case GST_STATE_PLAYING:
1448         basesink->need_preroll = FALSE;
1449         break;
1450       case GST_STATE_PAUSED:
1451         basesink->need_preroll = TRUE;
1452         break;
1453       default:
1454         basesink->need_preroll = FALSE;
1455         basesink->flushing = TRUE;
1456         break;
1457     }
1458     /* we can report latency queries now */
1459     basesink->priv->have_latency = TRUE;
1460     GST_OBJECT_UNLOCK (basesink);
1461     return TRUE;
1462   }
1463 stopping:
1464   {
1465     /* app is going to READY */
1466     GST_DEBUG_OBJECT (basesink, "stopping");
1467     basesink->need_preroll = FALSE;
1468     basesink->flushing = TRUE;
1469     GST_OBJECT_UNLOCK (basesink);
1470     return FALSE;
1471   }
1472 async_failed:
1473   {
1474     GST_DEBUG_OBJECT (basesink, "async commit failed");
1475     GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_FAILURE;
1476     GST_OBJECT_UNLOCK (basesink);
1477     return FALSE;
1478   }
1479 }
1480
1481 static void
1482 start_stepping (GstBaseSink * sink, GstSegment * segment,
1483     GstStepInfo * pending, GstStepInfo * current)
1484 {
1485   GST_DEBUG_OBJECT (sink, "update pending step");
1486
1487   GST_OBJECT_LOCK (sink);
1488   memcpy (current, pending, sizeof (GstStepInfo));
1489   pending->valid = FALSE;
1490   GST_OBJECT_UNLOCK (sink);
1491
1492   /* get the running time of where we paused and remember it */
1493   current->start = gst_element_get_start_time (GST_ELEMENT_CAST (sink));
1494   gst_segment_set_running_time (segment, GST_FORMAT_TIME, current->start);
1495
1496   /* set the new rate for the remainder of the segment */
1497   current->start_rate = segment->rate;
1498   segment->rate *= current->rate;
1499   segment->abs_rate = ABS (segment->rate);
1500
1501   GST_DEBUG_OBJECT (sink, "step started at running_time %" GST_TIME_FORMAT,
1502       GST_TIME_ARGS (current->start));
1503
1504   if (current->amount == -1) {
1505     GST_DEBUG_OBJECT (sink, "step amount == -1, stop stepping");
1506     current->valid = FALSE;
1507   } else {
1508     GST_DEBUG_OBJECT (sink, "step amount: %" G_GUINT64_FORMAT ", format: %s, "
1509         "rate: %f", current->amount, gst_format_get_name (current->format),
1510         current->rate);
1511   }
1512 }
1513
1514 static void
1515 stop_stepping (GstBaseSink * sink, GstSegment * segment,
1516     GstStepInfo * current, guint64 cstart, guint64 cstop, gint64 * rstart,
1517     gint64 * rstop)
1518 {
1519   gint64 stop, position;
1520   GstMessage *message;
1521
1522   GST_DEBUG_OBJECT (sink, "step complete");
1523
1524   if (segment->rate > 0.0)
1525     stop = *rstart;
1526   else
1527     stop = *rstop;
1528
1529   GST_DEBUG_OBJECT (sink,
1530       "step stop at running_time %" GST_TIME_FORMAT, GST_TIME_ARGS (stop));
1531
1532   if (stop == -1)
1533     current->duration = current->position;
1534   else
1535     current->duration = stop - current->start;
1536
1537   GST_DEBUG_OBJECT (sink, "step elapsed running_time %" GST_TIME_FORMAT,
1538       GST_TIME_ARGS (current->duration));
1539
1540   /* now move the segment to the new running time */
1541   position = current->start + current->duration;
1542   gst_segment_set_running_time (segment, GST_FORMAT_TIME, position);
1543   gst_element_set_start_time (GST_ELEMENT_CAST (sink), position);
1544
1545   /* restore the previous rate */
1546   segment->rate = current->start_rate;
1547   segment->abs_rate = ABS (segment->rate);
1548
1549   if (current->flush) {
1550     /* and remove the accumulated time we flushed */
1551     segment->accum = current->start;
1552   }
1553
1554   /* the clip segment is used for position report in paused... */
1555   memcpy (sink->abidata.ABI.clip_segment, segment, sizeof (GstSegment));
1556
1557   /* post the step done when we know the stepped duration in TIME */
1558   message =
1559       gst_message_new_step_done (GST_OBJECT_CAST (sink), current->format,
1560       current->amount, current->rate, current->flush, current->intermediate,
1561       current->duration);
1562   gst_message_set_seqnum (message, current->seqnum);
1563   gst_element_post_message (GST_ELEMENT_CAST (sink), message);
1564
1565   if (!current->intermediate)
1566     sink->need_preroll = current->need_preroll;
1567
1568   /* and the current step info finished and becomes invalid */
1569   current->valid = FALSE;
1570 }
1571
1572 static gboolean
1573 handle_stepping (GstBaseSink * sink, GstSegment * segment,
1574     GstStepInfo * current, gint64 * cstart, gint64 * cstop, gint64 * rstart,
1575     gint64 * rstop)
1576 {
1577   GstBaseSinkPrivate *priv;
1578   gboolean step_end = FALSE;
1579
1580   priv = sink->priv;
1581
1582   /* see if we need to skip this buffer because of stepping */
1583   switch (current->format) {
1584     case GST_FORMAT_TIME:
1585     {
1586       guint64 end;
1587       gint64 first, last;
1588
1589       if (segment->rate > 0.0) {
1590         first = *rstart;
1591         last = *rstop;
1592       } else {
1593         first = *rstop;
1594         last = *rstart;
1595       }
1596
1597       end = current->start + current->amount;
1598       current->position = first - current->start;
1599
1600       if (G_UNLIKELY (segment->abs_rate != 1.0))
1601         current->position /= segment->abs_rate;
1602
1603       GST_DEBUG_OBJECT (sink,
1604           "buffer: %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
1605           GST_TIME_ARGS (first), GST_TIME_ARGS (last));
1606       GST_DEBUG_OBJECT (sink,
1607           "got time step %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT "/%"
1608           GST_TIME_FORMAT, GST_TIME_ARGS (current->position),
1609           GST_TIME_ARGS (last - current->start),
1610           GST_TIME_ARGS (current->amount));
1611
1612       if (current->position >= current->amount || last >= end) {
1613         GST_DEBUG_OBJECT (sink, "step ended, we need clipping");
1614         step_end = TRUE;
1615         if (segment->rate > 0.0) {
1616           *rstart = end;
1617           *cstart = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1618         } else {
1619           *rstop = end;
1620           *cstop = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1621         }
1622       }
1623       GST_DEBUG_OBJECT (sink,
1624           "cstart %" GST_TIME_FORMAT ", rstart %" GST_TIME_FORMAT,
1625           GST_TIME_ARGS (*cstart), GST_TIME_ARGS (*rstart));
1626       GST_DEBUG_OBJECT (sink,
1627           "cstop %" GST_TIME_FORMAT ", rstop %" GST_TIME_FORMAT,
1628           GST_TIME_ARGS (*cstop), GST_TIME_ARGS (*rstop));
1629       break;
1630     }
1631     case GST_FORMAT_BUFFERS:
1632       GST_DEBUG_OBJECT (sink,
1633           "got default step %" G_GUINT64_FORMAT "/%" G_GUINT64_FORMAT,
1634           current->position, current->amount);
1635
1636       if (current->position < current->amount) {
1637         current->position++;
1638       } else {
1639         step_end = TRUE;
1640       }
1641       break;
1642     case GST_FORMAT_DEFAULT:
1643     default:
1644       GST_DEBUG_OBJECT (sink,
1645           "got unknown step %" G_GUINT64_FORMAT "/%" G_GUINT64_FORMAT,
1646           current->position, current->amount);
1647       break;
1648   }
1649   return step_end;
1650 }
1651
1652 /* with STREAM_LOCK, PREROLL_LOCK
1653  *
1654  * Returns TRUE if the object needs synchronisation and takes therefore
1655  * part in prerolling.
1656  *
1657  * rsstart/rsstop contain the start/stop in stream time.
1658  * rrstart/rrstop contain the start/stop in running time.
1659  */
1660 static gboolean
1661 gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
1662     GstClockTime * rsstart, GstClockTime * rsstop,
1663     GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync,
1664     gboolean * stepped, GstSegment * segment, GstStepInfo * step)
1665 {
1666   GstBaseSinkClass *bclass;
1667   GstBuffer *buffer;
1668   GstClockTime start, stop;     /* raw start/stop timestamps */
1669   gint64 cstart, cstop;         /* clipped raw timestamps */
1670   gint64 rstart, rstop;         /* clipped timestamps converted to running time */
1671   GstClockTime sstart, sstop;   /* clipped timestamps converted to stream time */
1672   GstFormat format;
1673   GstBaseSinkPrivate *priv;
1674   gboolean step_end;
1675
1676   priv = basesink->priv;
1677
1678   /* start with nothing */
1679   start = stop = -1;
1680
1681   step_end = FALSE;
1682
1683   if (G_UNLIKELY (GST_IS_EVENT (obj))) {
1684     GstEvent *event = GST_EVENT_CAST (obj);
1685
1686     switch (GST_EVENT_TYPE (event)) {
1687         /* EOS event needs syncing */
1688       case GST_EVENT_EOS:
1689       {
1690         if (basesink->segment.rate >= 0.0) {
1691           sstart = sstop = priv->current_sstop;
1692           if (sstart == -1) {
1693             /* we have not seen a buffer yet, use the segment values */
1694             sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
1695                 basesink->segment.format, basesink->segment.stop);
1696           }
1697         } else {
1698           sstart = sstop = priv->current_sstart;
1699           if (sstart == -1) {
1700             /* we have not seen a buffer yet, use the segment values */
1701             sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
1702                 basesink->segment.format, basesink->segment.start);
1703           }
1704         }
1705
1706         rstart = rstop = priv->eos_rtime;
1707         *do_sync = rstart != -1;
1708         GST_DEBUG_OBJECT (basesink, "sync times for EOS %" GST_TIME_FORMAT,
1709             GST_TIME_ARGS (rstart));
1710         /* if we are stepping, we end now */
1711         step_end = step->valid;
1712         goto eos_done;
1713       }
1714       default:
1715         /* other events do not need syncing */
1716         /* FIXME, maybe NEWSEGMENT might need synchronisation
1717          * since the POSITION query depends on accumulated times and
1718          * we cannot accumulate the current segment before the previous
1719          * one completed.
1720          */
1721         return FALSE;
1722     }
1723   }
1724
1725   /* else do buffer sync code */
1726   buffer = GST_BUFFER_CAST (obj);
1727
1728   bclass = GST_BASE_SINK_GET_CLASS (basesink);
1729
1730   /* just get the times to see if we need syncing, if the start returns -1 we
1731    * don't sync. */
1732   if (bclass->get_times)
1733     bclass->get_times (basesink, buffer, &start, &stop);
1734
1735   if (start == -1) {
1736     /* we don't need to sync but we still want to get the timestamps for
1737      * tracking the position */
1738     gst_base_sink_get_times (basesink, buffer, &start, &stop);
1739     *do_sync = FALSE;
1740   } else {
1741     *do_sync = TRUE;
1742   }
1743
1744   GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
1745       ", stop: %" GST_TIME_FORMAT ", do_sync %d", GST_TIME_ARGS (start),
1746       GST_TIME_ARGS (stop), *do_sync);
1747
1748   /* collect segment and format for code clarity */
1749   format = segment->format;
1750
1751   /* no timestamp clipping if we did not get a TIME segment format */
1752   if (G_UNLIKELY (format != GST_FORMAT_TIME)) {
1753     cstart = start;
1754     cstop = stop;
1755     /* do running and stream time in TIME format */
1756     format = GST_FORMAT_TIME;
1757     GST_LOG_OBJECT (basesink, "not time format, don't clip");
1758     goto do_times;
1759   }
1760
1761   /* clip, only when we know about time */
1762   if (G_UNLIKELY (!gst_segment_clip (segment, GST_FORMAT_TIME,
1763               (gint64) start, (gint64) stop, &cstart, &cstop)))
1764     goto out_of_segment;
1765
1766   if (G_UNLIKELY (start != cstart || stop != cstop)) {
1767     GST_DEBUG_OBJECT (basesink, "clipped to: start %" GST_TIME_FORMAT
1768         ", stop: %" GST_TIME_FORMAT, GST_TIME_ARGS (cstart),
1769         GST_TIME_ARGS (cstop));
1770   }
1771
1772   /* set last stop position */
1773   if (G_LIKELY (cstop != GST_CLOCK_TIME_NONE))
1774     gst_segment_set_last_stop (segment, GST_FORMAT_TIME, cstop);
1775   else
1776     gst_segment_set_last_stop (segment, GST_FORMAT_TIME, cstart);
1777
1778 do_times:
1779   rstart = gst_segment_to_running_time (segment, format, cstart);
1780   rstop = gst_segment_to_running_time (segment, format, cstop);
1781
1782   if (G_UNLIKELY (step->valid)) {
1783     if (!(step_end = handle_stepping (basesink, segment, step, &cstart, &cstop,
1784                 &rstart, &rstop))) {
1785       /* step is still busy, we discard data when we are flushing */
1786       *stepped = step->flush;
1787     }
1788   }
1789   /* this can produce wrong values if we accumulated non-TIME segments. If this happens,
1790    * upstream is behaving very badly */
1791   sstart = gst_segment_to_stream_time (segment, format, cstart);
1792   sstop = gst_segment_to_stream_time (segment, format, cstop);
1793
1794 eos_done:
1795   /* done label only called when doing EOS, we also stop stepping then */
1796   if (step_end)
1797     stop_stepping (basesink, segment, step, cstart, cstop, &rstart, &rstop);
1798
1799   /* save times */
1800   *rsstart = sstart;
1801   *rsstop = sstop;
1802   *rrstart = rstart;
1803   *rrstop = rstop;
1804
1805   /* buffers and EOS always need syncing and preroll */
1806   return TRUE;
1807
1808   /* special cases */
1809 out_of_segment:
1810   {
1811     /* we usually clip in the chain function already but stepping could cause
1812      * the segment to be updated later. we return FALSE so that we don't try
1813      * to sync on it. */
1814     GST_LOG_OBJECT (basesink, "buffer skipped, not in segment");
1815     return FALSE;
1816   }
1817 }
1818
1819 /* with STREAM_LOCK, PREROLL_LOCK, LOCK
1820  * adjust a timestamp with the latency and timestamp offset */
1821 static GstClockTime
1822 gst_base_sink_adjust_time (GstBaseSink * basesink, GstClockTime time)
1823 {
1824   GstClockTimeDiff ts_offset;
1825
1826   /* don't do anything funny with invalid timestamps */
1827   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time)))
1828     return time;
1829
1830   time += basesink->priv->latency;
1831
1832   /* apply offset, be carefull for underflows */
1833   ts_offset = basesink->priv->ts_offset;
1834   if (ts_offset < 0) {
1835     ts_offset = -ts_offset;
1836     if (ts_offset < time)
1837       time -= ts_offset;
1838     else
1839       time = 0;
1840   } else
1841     time += ts_offset;
1842
1843   return time;
1844 }
1845
1846 /**
1847  * gst_base_sink_wait_clock:
1848  * @sink: the sink
1849  * @time: the running_time to be reached
1850  * @jitter: the jitter to be filled with time diff (can be NULL)
1851  *
1852  * This function will block until @time is reached. It is usually called by
1853  * subclasses that use their own internal synchronisation.
1854  *
1855  * If @time is not valid, no sycnhronisation is done and #GST_CLOCK_BADTIME is
1856  * returned. Likewise, if synchronisation is disabled in the element or there
1857  * is no clock, no synchronisation is done and #GST_CLOCK_BADTIME is returned.
1858  *
1859  * This function should only be called with the PREROLL_LOCK held, like when
1860  * receiving an EOS event in the ::event vmethod or when receiving a buffer in
1861  * the ::render vmethod.
1862  *
1863  * The @time argument should be the running_time of when this method should
1864  * return and is not adjusted with any latency or offset configured in the
1865  * sink.
1866  *
1867  * Since 0.10.20
1868  *
1869  * Returns: #GstClockReturn
1870  */
1871 GstClockReturn
1872 gst_base_sink_wait_clock (GstBaseSink * sink, GstClockTime time,
1873     GstClockTimeDiff * jitter)
1874 {
1875   GstClockID id;
1876   GstClockReturn ret;
1877   GstClock *clock;
1878
1879   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time)))
1880     goto invalid_time;
1881
1882   GST_OBJECT_LOCK (sink);
1883   if (G_UNLIKELY (!sink->sync))
1884     goto no_sync;
1885
1886   if (G_UNLIKELY ((clock = GST_ELEMENT_CLOCK (sink)) == NULL))
1887     goto no_clock;
1888
1889   /* add base_time to running_time to get the time against the clock */
1890   time += GST_ELEMENT_CAST (sink)->base_time;
1891
1892   id = gst_clock_new_single_shot_id (clock, time);
1893   GST_OBJECT_UNLOCK (sink);
1894
1895   /* A blocking wait is performed on the clock. We save the ClockID
1896    * so we can unlock the entry at any time. While we are blocking, we
1897    * release the PREROLL_LOCK so that other threads can interrupt the
1898    * entry. */
1899   sink->clock_id = id;
1900   /* release the preroll lock while waiting */
1901   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
1902
1903   ret = gst_clock_id_wait (id, jitter);
1904
1905   GST_PAD_PREROLL_LOCK (sink->sinkpad);
1906   gst_clock_id_unref (id);
1907   sink->clock_id = NULL;
1908
1909   return ret;
1910
1911   /* no syncing needed */
1912 invalid_time:
1913   {
1914     GST_DEBUG_OBJECT (sink, "time not valid, no sync needed");
1915     return GST_CLOCK_BADTIME;
1916   }
1917 no_sync:
1918   {
1919     GST_DEBUG_OBJECT (sink, "sync disabled");
1920     GST_OBJECT_UNLOCK (sink);
1921     return GST_CLOCK_BADTIME;
1922   }
1923 no_clock:
1924   {
1925     GST_DEBUG_OBJECT (sink, "no clock, can't sync");
1926     GST_OBJECT_UNLOCK (sink);
1927     return GST_CLOCK_BADTIME;
1928   }
1929 }
1930
1931 /**
1932  * gst_base_sink_wait_preroll:
1933  * @sink: the sink
1934  *
1935  * If the #GstBaseSinkClass::render method performs its own synchronisation against
1936  * the clock it must unblock when going from PLAYING to the PAUSED state and call
1937  * this method before continuing to render the remaining data.
1938  *
1939  * This function will block until a state change to PLAYING happens (in which
1940  * case this function returns #GST_FLOW_OK) or the processing must be stopped due
1941  * to a state change to READY or a FLUSH event (in which case this function
1942  * returns #GST_FLOW_WRONG_STATE).
1943  *
1944  * This function should only be called with the PREROLL_LOCK held, like in the
1945  * render function.
1946  *
1947  * Since: 0.10.11
1948  *
1949  * Returns: #GST_FLOW_OK if the preroll completed and processing can
1950  * continue. Any other return value should be returned from the render vmethod.
1951  */
1952 GstFlowReturn
1953 gst_base_sink_wait_preroll (GstBaseSink * sink)
1954 {
1955   sink->have_preroll = TRUE;
1956   GST_DEBUG_OBJECT (sink, "waiting in preroll for flush or PLAYING");
1957   /* block until the state changes, or we get a flush, or something */
1958   GST_PAD_PREROLL_WAIT (sink->sinkpad);
1959   sink->have_preroll = FALSE;
1960   if (G_UNLIKELY (sink->flushing))
1961     goto stopping;
1962   if (G_UNLIKELY (sink->priv->step_unlock))
1963     goto step_unlocked;
1964   GST_DEBUG_OBJECT (sink, "continue after preroll");
1965
1966   return GST_FLOW_OK;
1967
1968   /* ERRORS */
1969 stopping:
1970   {
1971     GST_DEBUG_OBJECT (sink, "preroll interrupted because of flush");
1972     return GST_FLOW_WRONG_STATE;
1973   }
1974 step_unlocked:
1975   {
1976     sink->priv->step_unlock = FALSE;
1977     GST_DEBUG_OBJECT (sink, "preroll interrupted because of step");
1978     return GST_FLOW_STEP;
1979   }
1980 }
1981
1982 /**
1983  * gst_base_sink_do_preroll:
1984  * @sink: the sink
1985  * @obj: the object that caused the preroll
1986  *
1987  * If the @sink spawns its own thread for pulling buffers from upstream it
1988  * should call this method after it has pulled a buffer. If the element needed
1989  * to preroll, this function will perform the preroll and will then block
1990  * until the element state is changed.
1991  *
1992  * This function should be called with the PREROLL_LOCK held.
1993  *
1994  * Since 0.10.22
1995  *
1996  * Returns: #GST_FLOW_OK if the preroll completed and processing can
1997  * continue. Any other return value should be returned from the render vmethod.
1998  */
1999 GstFlowReturn
2000 gst_base_sink_do_preroll (GstBaseSink * sink, GstMiniObject * obj)
2001 {
2002   GstFlowReturn ret;
2003
2004   while (G_UNLIKELY (sink->need_preroll)) {
2005     GST_DEBUG_OBJECT (sink, "prerolling object %p", obj);
2006
2007     ret = gst_base_sink_preroll_object (sink, obj);
2008     if (ret != GST_FLOW_OK)
2009       goto preroll_failed;
2010
2011     /* need to recheck here because the commit state could have
2012      * made us not need the preroll anymore */
2013     if (G_LIKELY (sink->need_preroll)) {
2014       /* block until the state changes, or we get a flush, or something */
2015       ret = gst_base_sink_wait_preroll (sink);
2016       if (ret != GST_FLOW_OK) {
2017         if (ret == GST_FLOW_STEP)
2018           ret = GST_FLOW_OK;
2019         else
2020           goto preroll_failed;
2021       }
2022     }
2023   }
2024   return GST_FLOW_OK;
2025
2026   /* ERRORS */
2027 preroll_failed:
2028   {
2029     GST_DEBUG_OBJECT (sink, "preroll failed %d", ret);
2030     return ret;
2031   }
2032 }
2033
2034 /**
2035  * gst_base_sink_wait_eos:
2036  * @sink: the sink
2037  * @time: the running_time to be reached
2038  * @jitter: the jitter to be filled with time diff (can be NULL)
2039  *
2040  * This function will block until @time is reached. It is usually called by
2041  * subclasses that use their own internal synchronisation but want to let the
2042  * EOS be handled by the base class.
2043  *
2044  * This function should only be called with the PREROLL_LOCK held, like when
2045  * receiving an EOS event in the ::event vmethod.
2046  *
2047  * The @time argument should be the running_time of when the EOS should happen
2048  * and will be adjusted with any latency and offset configured in the sink.
2049  *
2050  * Since 0.10.15
2051  *
2052  * Returns: #GstFlowReturn
2053  */
2054 GstFlowReturn
2055 gst_base_sink_wait_eos (GstBaseSink * sink, GstClockTime time,
2056     GstClockTimeDiff * jitter)
2057 {
2058   GstClockReturn status;
2059   GstFlowReturn ret;
2060
2061   do {
2062     GstClockTime stime;
2063
2064     GST_DEBUG_OBJECT (sink, "checking preroll");
2065
2066     /* first wait for the playing state before we can continue */
2067     if (G_UNLIKELY (sink->need_preroll)) {
2068       ret = gst_base_sink_wait_preroll (sink);
2069       if (ret != GST_FLOW_OK) {
2070         if (ret == GST_FLOW_STEP)
2071           ret = GST_FLOW_OK;
2072         else
2073           goto flushing;
2074       }
2075     }
2076
2077     /* preroll done, we can sync since we are in PLAYING now. */
2078     GST_DEBUG_OBJECT (sink, "possibly waiting for clock to reach %"
2079         GST_TIME_FORMAT, GST_TIME_ARGS (time));
2080
2081     /* compensate for latency and ts_offset. We don't adjust for render delay
2082      * because we don't interact with the device on EOS normally. */
2083     stime = gst_base_sink_adjust_time (sink, time);
2084
2085     /* wait for the clock, this can be interrupted because we got shut down or
2086      * we PAUSED. */
2087     status = gst_base_sink_wait_clock (sink, stime, jitter);
2088
2089     GST_DEBUG_OBJECT (sink, "clock returned %d", status);
2090
2091     /* invalid time, no clock or sync disabled, just continue then */
2092     if (status == GST_CLOCK_BADTIME)
2093       break;
2094
2095     /* waiting could have been interrupted and we can be flushing now */
2096     if (G_UNLIKELY (sink->flushing))
2097       goto flushing;
2098
2099     /* retry if we got unscheduled, which means we did not reach the timeout
2100      * yet. if some other error occures, we continue. */
2101   } while (status == GST_CLOCK_UNSCHEDULED);
2102
2103   GST_DEBUG_OBJECT (sink, "end of stream");
2104
2105   return GST_FLOW_OK;
2106
2107   /* ERRORS */
2108 flushing:
2109   {
2110     GST_DEBUG_OBJECT (sink, "we are flushing");
2111     return GST_FLOW_WRONG_STATE;
2112   }
2113 }
2114
2115 /* with STREAM_LOCK, PREROLL_LOCK
2116  *
2117  * Make sure we are in PLAYING and synchronize an object to the clock.
2118  *
2119  * If we need preroll, we are not in PLAYING. We try to commit the state
2120  * if needed and then block if we still are not PLAYING.
2121  *
2122  * We start waiting on the clock in PLAYING. If we got interrupted, we
2123  * immediatly try to re-preroll.
2124  *
2125  * Some objects do not need synchronisation (most events) and so this function
2126  * immediatly returns GST_FLOW_OK.
2127  *
2128  * for objects that arrive later than max-lateness to be synchronized to the
2129  * clock have the @late boolean set to TRUE.
2130  *
2131  * This function keeps a running average of the jitter (the diff between the
2132  * clock time and the requested sync time). The jitter is negative for
2133  * objects that arrive in time and positive for late buffers.
2134  *
2135  * does not take ownership of obj.
2136  */
2137 static GstFlowReturn
2138 gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
2139     GstMiniObject * obj, gboolean * late)
2140 {
2141   GstClockTimeDiff jitter;
2142   gboolean syncable;
2143   GstClockReturn status = GST_CLOCK_OK;
2144   GstClockTime rstart, rstop, sstart, sstop, stime;
2145   gboolean do_sync;
2146   GstBaseSinkPrivate *priv;
2147   GstFlowReturn ret;
2148   GstStepInfo *current, *pending;
2149   gboolean stepped;
2150
2151   priv = basesink->priv;
2152
2153 do_step:
2154   sstart = sstop = rstart = rstop = -1;
2155   do_sync = TRUE;
2156   stepped = FALSE;
2157
2158   priv->current_rstart = -1;
2159
2160   /* get stepping info */
2161   current = &priv->current_step;
2162   pending = &priv->pending_step;
2163
2164   /* get timing information for this object against the render segment */
2165   syncable = gst_base_sink_get_sync_times (basesink, obj,
2166       &sstart, &sstop, &rstart, &rstop, &do_sync, &stepped, &basesink->segment,
2167       current);
2168
2169   if (G_UNLIKELY (stepped))
2170     goto step_skipped;
2171
2172   /* a syncable object needs to participate in preroll and
2173    * clocking. All buffers and EOS are syncable. */
2174   if (G_UNLIKELY (!syncable))
2175     goto not_syncable;
2176
2177   /* store timing info for current object */
2178   priv->current_rstart = rstart;
2179   priv->current_rstop = (rstop != -1 ? rstop : rstart);
2180
2181   /* save sync time for eos when the previous object needed sync */
2182   priv->eos_rtime = (do_sync ? priv->current_rstop : -1);
2183
2184 again:
2185   /* first do preroll, this makes sure we commit our state
2186    * to PAUSED and can continue to PLAYING. We cannot perform
2187    * any clock sync in PAUSED because there is no clock. */
2188   ret = gst_base_sink_do_preroll (basesink, obj);
2189   if (G_UNLIKELY (ret != GST_FLOW_OK))
2190     goto preroll_failed;
2191
2192   /* After rendering we store the position of the last buffer so that we can use
2193    * it to report the position. We need to take the lock here. */
2194   GST_OBJECT_LOCK (basesink);
2195   priv->current_sstart = sstart;
2196   priv->current_sstop = (sstop != -1 ? sstop : sstart);
2197   GST_OBJECT_UNLOCK (basesink);
2198
2199   /* update the segment with a pending step if the current one is invalid and we
2200    * have a new pending one. We only accept new step updates after a preroll */
2201   if (G_UNLIKELY (pending->valid && !current->valid)) {
2202     start_stepping (basesink, &basesink->segment, pending, current);
2203     goto do_step;
2204   }
2205
2206   if (!do_sync)
2207     goto done;
2208
2209   /* adjust for latency */
2210   stime = gst_base_sink_adjust_time (basesink, rstart);
2211
2212   /* adjust for render-delay, avoid underflows */
2213   if (stime != -1) {
2214     if (stime > priv->render_delay)
2215       stime -= priv->render_delay;
2216     else
2217       stime = 0;
2218   }
2219
2220   /* preroll done, we can sync since we are in PLAYING now. */
2221   GST_DEBUG_OBJECT (basesink, "possibly waiting for clock to reach %"
2222       GST_TIME_FORMAT ", adjusted %" GST_TIME_FORMAT,
2223       GST_TIME_ARGS (rstart), GST_TIME_ARGS (stime));
2224
2225   /* This function will return immediatly if start == -1, no clock
2226    * or sync is disabled with GST_CLOCK_BADTIME. */
2227   status = gst_base_sink_wait_clock (basesink, stime, &jitter);
2228
2229   GST_DEBUG_OBJECT (basesink, "clock returned %d", status);
2230
2231   /* invalid time, no clock or sync disabled, just render */
2232   if (status == GST_CLOCK_BADTIME)
2233     goto done;
2234
2235   /* waiting could have been interrupted and we can be flushing now */
2236   if (G_UNLIKELY (basesink->flushing))
2237     goto flushing;
2238
2239   /* check for unlocked by a state change, we are not flushing so
2240    * we can try to preroll on the current buffer. */
2241   if (G_UNLIKELY (status == GST_CLOCK_UNSCHEDULED)) {
2242     GST_DEBUG_OBJECT (basesink, "unscheduled, waiting some more");
2243     priv->call_preroll = TRUE;
2244     goto again;
2245   }
2246
2247   /* successful syncing done, record observation */
2248   priv->current_jitter = jitter;
2249
2250   /* check if the object should be dropped */
2251   *late = gst_base_sink_is_too_late (basesink, obj, rstart, rstop,
2252       status, jitter);
2253
2254 done:
2255   return GST_FLOW_OK;
2256
2257   /* ERRORS */
2258 step_skipped:
2259   {
2260     GST_DEBUG_OBJECT (basesink, "skipped stepped object %p", obj);
2261     *late = TRUE;
2262     return GST_FLOW_OK;
2263   }
2264 not_syncable:
2265   {
2266     GST_DEBUG_OBJECT (basesink, "non syncable object %p", obj);
2267     return GST_FLOW_OK;
2268   }
2269 flushing:
2270   {
2271     GST_DEBUG_OBJECT (basesink, "we are flushing");
2272     return GST_FLOW_WRONG_STATE;
2273   }
2274 preroll_failed:
2275   {
2276     GST_DEBUG_OBJECT (basesink, "preroll failed");
2277     return ret;
2278   }
2279 }
2280
2281 static gboolean
2282 gst_base_sink_send_qos (GstBaseSink * basesink,
2283     gdouble proportion, GstClockTime time, GstClockTimeDiff diff)
2284 {
2285   GstEvent *event;
2286   gboolean res;
2287
2288   /* generate Quality-of-Service event */
2289   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
2290       "qos: proportion: %lf, diff %" G_GINT64_FORMAT ", timestamp %"
2291       GST_TIME_FORMAT, proportion, diff, GST_TIME_ARGS (time));
2292
2293   event = gst_event_new_qos (proportion, diff, time);
2294
2295   /* send upstream */
2296   res = gst_pad_push_event (basesink->sinkpad, event);
2297
2298   return res;
2299 }
2300
2301 static void
2302 gst_base_sink_perform_qos (GstBaseSink * sink, gboolean dropped)
2303 {
2304   GstBaseSinkPrivate *priv;
2305   GstClockTime start, stop;
2306   GstClockTimeDiff jitter;
2307   GstClockTime pt, entered, left;
2308   GstClockTime duration;
2309   gdouble rate;
2310
2311   priv = sink->priv;
2312
2313   start = priv->current_rstart;
2314
2315   /* if Quality-of-Service disabled, do nothing */
2316   if (!g_atomic_int_get (&priv->qos_enabled) || start == -1)
2317     return;
2318
2319   stop = priv->current_rstop;
2320   jitter = priv->current_jitter;
2321
2322   if (jitter < 0) {
2323     /* this is the time the buffer entered the sink */
2324     if (start < -jitter)
2325       entered = 0;
2326     else
2327       entered = start + jitter;
2328     left = start;
2329   } else {
2330     /* this is the time the buffer entered the sink */
2331     entered = start + jitter;
2332     /* this is the time the buffer left the sink */
2333     left = start + jitter;
2334   }
2335
2336   /* calculate duration of the buffer */
2337   if (stop != -1)
2338     duration = stop - start;
2339   else
2340     duration = -1;
2341
2342   /* if we have the time when the last buffer left us, calculate
2343    * processing time */
2344   if (priv->last_left != -1) {
2345     if (entered > priv->last_left) {
2346       pt = entered - priv->last_left;
2347     } else {
2348       pt = 0;
2349     }
2350   } else {
2351     pt = priv->avg_pt;
2352   }
2353
2354   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "start: %" GST_TIME_FORMAT
2355       ", entered %" GST_TIME_FORMAT ", left %" GST_TIME_FORMAT ", pt: %"
2356       GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT ",jitter %"
2357       G_GINT64_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (entered),
2358       GST_TIME_ARGS (left), GST_TIME_ARGS (pt), GST_TIME_ARGS (duration),
2359       jitter);
2360
2361   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "avg_duration: %" GST_TIME_FORMAT
2362       ", avg_pt: %" GST_TIME_FORMAT ", avg_rate: %g",
2363       GST_TIME_ARGS (priv->avg_duration), GST_TIME_ARGS (priv->avg_pt),
2364       priv->avg_rate);
2365
2366   /* collect running averages. for first observations, we copy the
2367    * values */
2368   if (priv->avg_duration == -1)
2369     priv->avg_duration = duration;
2370   else
2371     priv->avg_duration = UPDATE_RUNNING_AVG (priv->avg_duration, duration);
2372
2373   if (priv->avg_pt == -1)
2374     priv->avg_pt = pt;
2375   else
2376     priv->avg_pt = UPDATE_RUNNING_AVG (priv->avg_pt, pt);
2377
2378   if (priv->avg_duration != 0)
2379     rate =
2380         gst_guint64_to_gdouble (priv->avg_pt) /
2381         gst_guint64_to_gdouble (priv->avg_duration);
2382   else
2383     rate = 0.0;
2384
2385   if (priv->last_left != -1) {
2386     if (dropped || priv->avg_rate < 0.0) {
2387       priv->avg_rate = rate;
2388     } else {
2389       if (rate > 1.0)
2390         priv->avg_rate = UPDATE_RUNNING_AVG_N (priv->avg_rate, rate);
2391       else
2392         priv->avg_rate = UPDATE_RUNNING_AVG_P (priv->avg_rate, rate);
2393     }
2394   }
2395
2396   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink,
2397       "updated: avg_duration: %" GST_TIME_FORMAT ", avg_pt: %" GST_TIME_FORMAT
2398       ", avg_rate: %g", GST_TIME_ARGS (priv->avg_duration),
2399       GST_TIME_ARGS (priv->avg_pt), priv->avg_rate);
2400
2401
2402   if (priv->avg_rate >= 0.0) {
2403     /* if we have a valid rate, start sending QoS messages */
2404     if (priv->current_jitter < 0) {
2405       /* make sure we never go below 0 when adding the jitter to the
2406        * timestamp. */
2407       if (priv->current_rstart < -priv->current_jitter)
2408         priv->current_jitter = -priv->current_rstart;
2409     }
2410     gst_base_sink_send_qos (sink, priv->avg_rate, priv->current_rstart,
2411         priv->current_jitter);
2412   }
2413
2414   /* record when this buffer will leave us */
2415   priv->last_left = left;
2416 }
2417
2418 /* reset all qos measuring */
2419 static void
2420 gst_base_sink_reset_qos (GstBaseSink * sink)
2421 {
2422   GstBaseSinkPrivate *priv;
2423
2424   priv = sink->priv;
2425
2426   priv->last_in_time = -1;
2427   priv->last_left = -1;
2428   priv->avg_duration = -1;
2429   priv->avg_pt = -1;
2430   priv->avg_rate = -1.0;
2431   priv->avg_render = -1;
2432   priv->rendered = 0;
2433   priv->dropped = 0;
2434
2435 }
2436
2437 /* Checks if the object was scheduled too late.
2438  *
2439  * start/stop contain the raw timestamp start and stop values
2440  * of the object.
2441  *
2442  * status and jitter contain the return values from the clock wait.
2443  *
2444  * returns TRUE if the buffer was too late.
2445  */
2446 static gboolean
2447 gst_base_sink_is_too_late (GstBaseSink * basesink, GstMiniObject * obj,
2448     GstClockTime start, GstClockTime stop,
2449     GstClockReturn status, GstClockTimeDiff jitter)
2450 {
2451   gboolean late;
2452   gint64 max_lateness;
2453   GstBaseSinkPrivate *priv;
2454
2455   priv = basesink->priv;
2456
2457   late = FALSE;
2458
2459   /* only for objects that were too late */
2460   if (G_LIKELY (status != GST_CLOCK_EARLY))
2461     goto in_time;
2462
2463   max_lateness = basesink->abidata.ABI.max_lateness;
2464
2465   /* check if frame dropping is enabled */
2466   if (max_lateness == -1)
2467     goto no_drop;
2468
2469   /* only check for buffers */
2470   if (G_UNLIKELY (!GST_IS_BUFFER (obj)))
2471     goto not_buffer;
2472
2473   /* can't do check if we don't have a timestamp */
2474   if (G_UNLIKELY (start == -1))
2475     goto no_timestamp;
2476
2477   /* we can add a valid stop time */
2478   if (stop != -1)
2479     max_lateness += stop;
2480   else
2481     max_lateness += start;
2482
2483   /* if the jitter bigger than duration and lateness we are too late */
2484   if ((late = start + jitter > max_lateness)) {
2485     GST_DEBUG_OBJECT (basesink, "buffer is too late %" GST_TIME_FORMAT
2486         " > %" GST_TIME_FORMAT, GST_TIME_ARGS (start + jitter),
2487         GST_TIME_ARGS (max_lateness));
2488     /* !!emergency!!, if we did not receive anything valid for more than a
2489      * second, render it anyway so the user sees something */
2490     if (priv->last_in_time != -1 && start - priv->last_in_time > GST_SECOND) {
2491       late = FALSE;
2492       GST_ELEMENT_WARNING (basesink, CORE, CLOCK,
2493           (_("A lot of buffers are being dropped.")),
2494           ("There may be a timestamping problem, or this computer is too slow."));
2495       GST_DEBUG_OBJECT (basesink,
2496           "**emergency** last buffer at %" GST_TIME_FORMAT " > GST_SECOND",
2497           GST_TIME_ARGS (priv->last_in_time));
2498     }
2499   }
2500
2501 done:
2502   if (!late) {
2503     priv->last_in_time = start;
2504   }
2505   return late;
2506
2507   /* all is fine */
2508 in_time:
2509   {
2510     GST_DEBUG_OBJECT (basesink, "object was scheduled in time");
2511     goto done;
2512   }
2513 no_drop:
2514   {
2515     GST_DEBUG_OBJECT (basesink, "frame dropping disabled");
2516     goto done;
2517   }
2518 not_buffer:
2519   {
2520     GST_DEBUG_OBJECT (basesink, "object is not a buffer");
2521     return FALSE;
2522   }
2523 no_timestamp:
2524   {
2525     GST_DEBUG_OBJECT (basesink, "buffer has no timestamp");
2526     return FALSE;
2527   }
2528 }
2529
2530 /* called before and after calling the render vmethod. It keeps track of how
2531  * much time was spent in the render method and is used to check if we are
2532  * flooded */
2533 static void
2534 gst_base_sink_do_render_stats (GstBaseSink * basesink, gboolean start)
2535 {
2536   GstBaseSinkPrivate *priv;
2537
2538   priv = basesink->priv;
2539
2540   if (start) {
2541     priv->start = gst_util_get_timestamp ();
2542   } else {
2543     GstClockTime elapsed;
2544
2545     priv->stop = gst_util_get_timestamp ();
2546
2547     elapsed = GST_CLOCK_DIFF (priv->start, priv->stop);
2548
2549     if (priv->avg_render == -1)
2550       priv->avg_render = elapsed;
2551     else
2552       priv->avg_render = UPDATE_RUNNING_AVG (priv->avg_render, elapsed);
2553
2554     GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
2555         "avg_render: %" GST_TIME_FORMAT, GST_TIME_ARGS (priv->avg_render));
2556   }
2557 }
2558
2559 /* with STREAM_LOCK, PREROLL_LOCK,
2560  *
2561  * Synchronize the object on the clock and then render it.
2562  *
2563  * takes ownership of obj.
2564  */
2565 static GstFlowReturn
2566 gst_base_sink_render_object (GstBaseSink * basesink, GstPad * pad,
2567     GstMiniObject * obj)
2568 {
2569   GstFlowReturn ret;
2570   GstBaseSinkClass *bclass;
2571   gboolean late;
2572
2573   GstBaseSinkPrivate *priv;
2574
2575   priv = basesink->priv;
2576
2577 again:
2578   late = FALSE;
2579   ret = GST_FLOW_OK;
2580
2581   /* synchronize this object, non syncable objects return OK
2582    * immediatly. */
2583   ret = gst_base_sink_do_sync (basesink, pad, obj, &late);
2584   if (G_UNLIKELY (ret != GST_FLOW_OK))
2585     goto sync_failed;
2586
2587   /* and now render, event or buffer. */
2588   if (G_LIKELY (GST_IS_BUFFER (obj))) {
2589     GstBuffer *buf;
2590
2591     /* drop late buffers unconditionally, let's hope it's unlikely */
2592     if (G_UNLIKELY (late))
2593       goto dropped;
2594
2595     buf = GST_BUFFER_CAST (obj);
2596
2597     gst_base_sink_set_last_buffer (basesink, buf);
2598
2599     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2600
2601     if (G_LIKELY (bclass->render)) {
2602       gint do_qos;
2603
2604       /* read once, to get same value before and after */
2605       do_qos = g_atomic_int_get (&priv->qos_enabled);
2606
2607       GST_DEBUG_OBJECT (basesink, "rendering buffer %p", obj);
2608
2609       /* record rendering time for QoS and stats */
2610       if (do_qos)
2611         gst_base_sink_do_render_stats (basesink, TRUE);
2612
2613       ret = bclass->render (basesink, buf);
2614
2615       if (do_qos)
2616         gst_base_sink_do_render_stats (basesink, FALSE);
2617
2618       if (ret == GST_FLOW_STEP)
2619         goto again;
2620
2621       priv->rendered++;
2622     }
2623   } else {
2624     GstEvent *event = GST_EVENT_CAST (obj);
2625     gboolean event_res = TRUE;
2626     GstEventType type;
2627
2628     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2629
2630     type = GST_EVENT_TYPE (event);
2631
2632     GST_DEBUG_OBJECT (basesink, "rendering event %p, type %s", obj,
2633         gst_event_type_get_name (type));
2634
2635     if (bclass->event)
2636       event_res = bclass->event (basesink, event);
2637
2638     /* when we get here we could be flushing again when the event handler calls
2639      * _wait_eos(). We have to ignore this object in that case. */
2640     if (G_UNLIKELY (basesink->flushing))
2641       goto flushing;
2642
2643     if (G_LIKELY (event_res)) {
2644       guint32 seqnum;
2645
2646       seqnum = basesink->priv->seqnum = gst_event_get_seqnum (event);
2647       GST_DEBUG_OBJECT (basesink, "Got seqnum #%" G_GUINT32_FORMAT, seqnum);
2648
2649       switch (type) {
2650         case GST_EVENT_EOS:
2651         {
2652           GstMessage *message;
2653
2654           /* the EOS event is completely handled so we mark
2655            * ourselves as being in the EOS state. eos is also
2656            * protected by the object lock so we can read it when
2657            * answering the POSITION query. */
2658           GST_OBJECT_LOCK (basesink);
2659           basesink->eos = TRUE;
2660           GST_OBJECT_UNLOCK (basesink);
2661
2662           /* ok, now we can post the message */
2663           GST_DEBUG_OBJECT (basesink, "Now posting EOS");
2664
2665           message = gst_message_new_eos (GST_OBJECT_CAST (basesink));
2666           gst_message_set_seqnum (message, seqnum);
2667           gst_element_post_message (GST_ELEMENT_CAST (basesink), message);
2668           break;
2669         }
2670         case GST_EVENT_NEWSEGMENT:
2671           /* configure the segment */
2672           gst_base_sink_configure_segment (basesink, pad, event,
2673               &basesink->segment);
2674           break;
2675         default:
2676           break;
2677       }
2678     }
2679   }
2680
2681 done:
2682   gst_base_sink_perform_qos (basesink, late);
2683
2684   GST_DEBUG_OBJECT (basesink, "object unref after render %p", obj);
2685   gst_mini_object_unref (obj);
2686
2687   return ret;
2688
2689   /* ERRORS */
2690 sync_failed:
2691   {
2692     GST_DEBUG_OBJECT (basesink, "do_sync returned %s", gst_flow_get_name (ret));
2693     goto done;
2694   }
2695 dropped:
2696   {
2697     priv->dropped++;
2698     GST_DEBUG_OBJECT (basesink, "buffer late, dropping");
2699     goto done;
2700   }
2701 flushing:
2702   {
2703     GST_DEBUG_OBJECT (basesink, "we are flushing, ignore object");
2704     gst_mini_object_unref (obj);
2705     return GST_FLOW_WRONG_STATE;
2706   }
2707 }
2708
2709 /* with STREAM_LOCK, PREROLL_LOCK
2710  *
2711  * Perform preroll on the given object. For buffers this means
2712  * calling the preroll subclass method.
2713  * If that succeeds, the state will be commited.
2714  *
2715  * function does not take ownership of obj.
2716  */
2717 static GstFlowReturn
2718 gst_base_sink_preroll_object (GstBaseSink * basesink, GstMiniObject * obj)
2719 {
2720   GstFlowReturn ret;
2721
2722   GST_DEBUG_OBJECT (basesink, "prerolling object %p", obj);
2723
2724   /* if it's a buffer, we need to call the preroll method */
2725   if (G_LIKELY (GST_IS_BUFFER (obj)) && basesink->priv->call_preroll) {
2726     GstBaseSinkClass *bclass;
2727     GstBuffer *buf;
2728     GstClockTime timestamp;
2729
2730     buf = GST_BUFFER_CAST (obj);
2731     timestamp = GST_BUFFER_TIMESTAMP (buf);
2732
2733     GST_DEBUG_OBJECT (basesink, "preroll buffer %" GST_TIME_FORMAT,
2734         GST_TIME_ARGS (timestamp));
2735
2736     gst_base_sink_set_last_buffer (basesink, buf);
2737
2738     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2739     if (bclass->preroll)
2740       if ((ret = bclass->preroll (basesink, buf)) != GST_FLOW_OK)
2741         goto preroll_failed;
2742
2743     basesink->priv->call_preroll = FALSE;
2744   }
2745
2746   /* commit state */
2747   if (G_LIKELY (basesink->playing_async)) {
2748     if (G_UNLIKELY (!gst_base_sink_commit_state (basesink)))
2749       goto stopping;
2750   }
2751
2752   return GST_FLOW_OK;
2753
2754   /* ERRORS */
2755 preroll_failed:
2756   {
2757     GST_DEBUG_OBJECT (basesink, "preroll failed, abort state");
2758     gst_element_abort_state (GST_ELEMENT_CAST (basesink));
2759     return ret;
2760   }
2761 stopping:
2762   {
2763     GST_DEBUG_OBJECT (basesink, "stopping while commiting state");
2764     return GST_FLOW_WRONG_STATE;
2765   }
2766 }
2767
2768 /* with STREAM_LOCK, PREROLL_LOCK
2769  *
2770  * Queue an object for rendering.
2771  * The first prerollable object queued will complete the preroll. If the
2772  * preroll queue if filled, we render all the objects in the queue.
2773  *
2774  * This function takes ownership of the object.
2775  */
2776 static GstFlowReturn
2777 gst_base_sink_queue_object_unlocked (GstBaseSink * basesink, GstPad * pad,
2778     GstMiniObject * obj, gboolean prerollable)
2779 {
2780   GstFlowReturn ret = GST_FLOW_OK;
2781   gint length;
2782   GQueue *q;
2783
2784   if (G_UNLIKELY (basesink->need_preroll)) {
2785     if (G_LIKELY (prerollable))
2786       basesink->preroll_queued++;
2787
2788     length = basesink->preroll_queued;
2789
2790     GST_DEBUG_OBJECT (basesink, "now %d prerolled items", length);
2791
2792     /* first prerollable item needs to finish the preroll */
2793     if (length == 1) {
2794       ret = gst_base_sink_preroll_object (basesink, obj);
2795       if (G_UNLIKELY (ret != GST_FLOW_OK))
2796         goto preroll_failed;
2797     }
2798     /* need to recheck if we need preroll, commmit state during preroll
2799      * could have made us not need more preroll. */
2800     if (G_UNLIKELY (basesink->need_preroll)) {
2801       /* see if we can render now, if we can't add the object to the preroll
2802        * queue. */
2803       if (G_UNLIKELY (length <= basesink->preroll_queue_max_len))
2804         goto more_preroll;
2805     }
2806   }
2807
2808   /* we can start rendering (or blocking) the queued object
2809    * if any. */
2810   q = basesink->preroll_queue;
2811   while (G_UNLIKELY (!g_queue_is_empty (q))) {
2812     GstMiniObject *o;
2813
2814     o = g_queue_pop_head (q);
2815     GST_DEBUG_OBJECT (basesink, "rendering queued object %p", o);
2816
2817     /* do something with the return value */
2818     ret = gst_base_sink_render_object (basesink, pad, o);
2819     if (ret != GST_FLOW_OK)
2820       goto dequeue_failed;
2821   }
2822
2823   /* now render the object */
2824   ret = gst_base_sink_render_object (basesink, pad, obj);
2825   basesink->preroll_queued = 0;
2826
2827   return ret;
2828
2829   /* special cases */
2830 preroll_failed:
2831   {
2832     GST_DEBUG_OBJECT (basesink, "preroll failed, reason %s",
2833         gst_flow_get_name (ret));
2834     gst_mini_object_unref (obj);
2835     return ret;
2836   }
2837 more_preroll:
2838   {
2839     /* add object to the queue and return */
2840     GST_DEBUG_OBJECT (basesink, "need more preroll data %d <= %d",
2841         length, basesink->preroll_queue_max_len);
2842     g_queue_push_tail (basesink->preroll_queue, obj);
2843     return GST_FLOW_OK;
2844   }
2845 dequeue_failed:
2846   {
2847     GST_DEBUG_OBJECT (basesink, "rendering queued objects failed, reason %s",
2848         gst_flow_get_name (ret));
2849     gst_mini_object_unref (obj);
2850     return ret;
2851   }
2852 }
2853
2854 /* with STREAM_LOCK
2855  *
2856  * This function grabs the PREROLL_LOCK and adds the object to
2857  * the queue.
2858  *
2859  * This function takes ownership of obj.
2860  */
2861 static GstFlowReturn
2862 gst_base_sink_queue_object (GstBaseSink * basesink, GstPad * pad,
2863     GstMiniObject * obj, gboolean prerollable)
2864 {
2865   GstFlowReturn ret;
2866
2867   GST_PAD_PREROLL_LOCK (pad);
2868   if (G_UNLIKELY (basesink->flushing))
2869     goto flushing;
2870
2871   if (G_UNLIKELY (basesink->priv->received_eos))
2872     goto was_eos;
2873
2874   ret = gst_base_sink_queue_object_unlocked (basesink, pad, obj, prerollable);
2875   GST_PAD_PREROLL_UNLOCK (pad);
2876
2877   return ret;
2878
2879   /* ERRORS */
2880 flushing:
2881   {
2882     GST_DEBUG_OBJECT (basesink, "sink is flushing");
2883     GST_PAD_PREROLL_UNLOCK (pad);
2884     gst_mini_object_unref (obj);
2885     return GST_FLOW_WRONG_STATE;
2886   }
2887 was_eos:
2888   {
2889     GST_DEBUG_OBJECT (basesink,
2890         "we are EOS, dropping object, return UNEXPECTED");
2891     GST_PAD_PREROLL_UNLOCK (pad);
2892     gst_mini_object_unref (obj);
2893     return GST_FLOW_UNEXPECTED;
2894   }
2895 }
2896
2897 static void
2898 gst_base_sink_flush_start (GstBaseSink * basesink, GstPad * pad)
2899 {
2900   /* make sure we are not blocked on the clock also clear any pending
2901    * eos state. */
2902   gst_base_sink_set_flushing (basesink, pad, TRUE);
2903
2904   /* we grab the stream lock but that is not needed since setting the
2905    * sink to flushing would make sure no state commit is being done
2906    * anymore */
2907   GST_PAD_STREAM_LOCK (pad);
2908   gst_base_sink_reset_qos (basesink);
2909   if (basesink->priv->async_enabled) {
2910     /* and we need to commit our state again on the next
2911      * prerolled buffer */
2912     basesink->playing_async = TRUE;
2913     gst_element_lost_state (GST_ELEMENT_CAST (basesink));
2914   } else {
2915     basesink->priv->have_latency = TRUE;
2916     basesink->need_preroll = FALSE;
2917   }
2918   gst_base_sink_set_last_buffer (basesink, NULL);
2919   GST_PAD_STREAM_UNLOCK (pad);
2920 }
2921
2922 static void
2923 gst_base_sink_flush_stop (GstBaseSink * basesink, GstPad * pad)
2924 {
2925   /* unset flushing so we can accept new data, this also flushes out any EOS
2926    * event. */
2927   gst_base_sink_set_flushing (basesink, pad, FALSE);
2928
2929   /* for position reporting */
2930   GST_OBJECT_LOCK (basesink);
2931   basesink->priv->current_sstart = -1;
2932   basesink->priv->current_sstop = -1;
2933   basesink->priv->eos_rtime = -1;
2934   basesink->priv->call_preroll = TRUE;
2935   basesink->priv->current_step.valid = FALSE;
2936   basesink->priv->pending_step.valid = FALSE;
2937   if (basesink->pad_mode == GST_ACTIVATE_PUSH) {
2938     /* we need new segment info after the flush. */
2939     basesink->have_newsegment = FALSE;
2940     gst_segment_init (&basesink->segment, GST_FORMAT_UNDEFINED);
2941     gst_segment_init (basesink->abidata.ABI.clip_segment, GST_FORMAT_UNDEFINED);
2942   }
2943   GST_OBJECT_UNLOCK (basesink);
2944 }
2945
2946 static gboolean
2947 gst_base_sink_event (GstPad * pad, GstEvent * event)
2948 {
2949   GstBaseSink *basesink;
2950   gboolean result = TRUE;
2951   GstBaseSinkClass *bclass;
2952
2953   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
2954
2955   bclass = GST_BASE_SINK_GET_CLASS (basesink);
2956
2957   GST_DEBUG_OBJECT (basesink, "event %p (%s)", event,
2958       GST_EVENT_TYPE_NAME (event));
2959
2960   switch (GST_EVENT_TYPE (event)) {
2961     case GST_EVENT_EOS:
2962     {
2963       GstFlowReturn ret;
2964
2965       GST_PAD_PREROLL_LOCK (pad);
2966       if (G_UNLIKELY (basesink->flushing))
2967         goto flushing;
2968
2969       if (G_UNLIKELY (basesink->priv->received_eos)) {
2970         /* we can't accept anything when we are EOS */
2971         result = FALSE;
2972         gst_event_unref (event);
2973       } else {
2974         /* we set the received EOS flag here so that we can use it when testing if
2975          * we are prerolled and to refuse more buffers. */
2976         basesink->priv->received_eos = TRUE;
2977
2978         /* EOS is a prerollable object, we call the unlocked version because it
2979          * does not check the received_eos flag. */
2980         ret = gst_base_sink_queue_object_unlocked (basesink, pad,
2981             GST_MINI_OBJECT_CAST (event), TRUE);
2982         if (G_UNLIKELY (ret != GST_FLOW_OK))
2983           result = FALSE;
2984       }
2985       GST_PAD_PREROLL_UNLOCK (pad);
2986       break;
2987     }
2988     case GST_EVENT_NEWSEGMENT:
2989     {
2990       GstFlowReturn ret;
2991
2992       GST_DEBUG_OBJECT (basesink, "newsegment %p", event);
2993
2994       GST_PAD_PREROLL_LOCK (pad);
2995       if (G_UNLIKELY (basesink->flushing))
2996         goto flushing;
2997
2998       if (G_UNLIKELY (basesink->priv->received_eos)) {
2999         /* we can't accept anything when we are EOS */
3000         result = FALSE;
3001         gst_event_unref (event);
3002       } else {
3003         /* the new segment is a non prerollable item and does not block anything,
3004          * we need to configure the current clipping segment and insert the event
3005          * in the queue to serialize it with the buffers for rendering. */
3006         gst_base_sink_configure_segment (basesink, pad, event,
3007             basesink->abidata.ABI.clip_segment);
3008
3009         ret =
3010             gst_base_sink_queue_object_unlocked (basesink, pad,
3011             GST_MINI_OBJECT_CAST (event), FALSE);
3012         if (G_UNLIKELY (ret != GST_FLOW_OK))
3013           result = FALSE;
3014         else {
3015           GST_OBJECT_LOCK (basesink);
3016           basesink->have_newsegment = TRUE;
3017           GST_OBJECT_UNLOCK (basesink);
3018         }
3019       }
3020       GST_PAD_PREROLL_UNLOCK (pad);
3021       break;
3022     }
3023     case GST_EVENT_FLUSH_START:
3024       if (bclass->event)
3025         bclass->event (basesink, event);
3026
3027       GST_DEBUG_OBJECT (basesink, "flush-start %p", event);
3028
3029       gst_base_sink_flush_start (basesink, pad);
3030
3031       gst_event_unref (event);
3032       break;
3033     case GST_EVENT_FLUSH_STOP:
3034       if (bclass->event)
3035         bclass->event (basesink, event);
3036
3037       GST_DEBUG_OBJECT (basesink, "flush-stop %p", event);
3038
3039       gst_base_sink_flush_stop (basesink, pad);
3040
3041       gst_event_unref (event);
3042       break;
3043     default:
3044       /* other events are sent to queue or subclass depending on if they
3045        * are serialized. */
3046       if (GST_EVENT_IS_SERIALIZED (event)) {
3047         gst_base_sink_queue_object (basesink, pad,
3048             GST_MINI_OBJECT_CAST (event), FALSE);
3049       } else {
3050         if (bclass->event)
3051           bclass->event (basesink, event);
3052         gst_event_unref (event);
3053       }
3054       break;
3055   }
3056 done:
3057   gst_object_unref (basesink);
3058
3059   return result;
3060
3061   /* ERRORS */
3062 flushing:
3063   {
3064     GST_DEBUG_OBJECT (basesink, "we are flushing");
3065     GST_PAD_PREROLL_UNLOCK (pad);
3066     result = FALSE;
3067     gst_event_unref (event);
3068     goto done;
3069   }
3070 }
3071
3072 /* default implementation to calculate the start and end
3073  * timestamps on a buffer, subclasses can override
3074  */
3075 static void
3076 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
3077     GstClockTime * start, GstClockTime * end)
3078 {
3079   GstClockTime timestamp, duration;
3080
3081   timestamp = GST_BUFFER_TIMESTAMP (buffer);
3082   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
3083
3084     /* get duration to calculate end time */
3085     duration = GST_BUFFER_DURATION (buffer);
3086     if (GST_CLOCK_TIME_IS_VALID (duration)) {
3087       *end = timestamp + duration;
3088     }
3089     *start = timestamp;
3090   }
3091 }
3092
3093 /* must be called with PREROLL_LOCK */
3094 static gboolean
3095 gst_base_sink_needs_preroll (GstBaseSink * basesink)
3096 {
3097   gboolean is_prerolled, res;
3098
3099   /* we have 2 cases where the PREROLL_LOCK is released:
3100    *  1) we are blocking in the PREROLL_LOCK and thus are prerolled.
3101    *  2) we are syncing on the clock
3102    */
3103   is_prerolled = basesink->have_preroll || basesink->priv->received_eos;
3104   res = !is_prerolled;
3105
3106   GST_DEBUG_OBJECT (basesink, "have_preroll: %d, EOS: %d => needs preroll: %d",
3107       basesink->have_preroll, basesink->priv->received_eos, res);
3108
3109   return res;
3110 }
3111
3112 /* with STREAM_LOCK, PREROLL_LOCK
3113  *
3114  * Takes a buffer and compare the timestamps with the last segment.
3115  * If the buffer falls outside of the segment boundaries, drop it.
3116  * Else queue the buffer for preroll and rendering.
3117  *
3118  * This function takes ownership of the buffer.
3119  */
3120 static GstFlowReturn
3121 gst_base_sink_chain_unlocked (GstBaseSink * basesink, GstPad * pad,
3122     GstBuffer * buf)
3123 {
3124   GstBaseSinkClass *bclass;
3125   GstFlowReturn result;
3126   GstClockTime start = GST_CLOCK_TIME_NONE, end = GST_CLOCK_TIME_NONE;
3127   GstSegment *clip_segment;
3128
3129   if (G_UNLIKELY (basesink->flushing))
3130     goto flushing;
3131
3132   if (G_UNLIKELY (basesink->priv->received_eos))
3133     goto was_eos;
3134
3135   /* for code clarity */
3136   clip_segment = basesink->abidata.ABI.clip_segment;
3137
3138   if (G_UNLIKELY (!basesink->have_newsegment)) {
3139     gboolean sync;
3140
3141     sync = gst_base_sink_get_sync (basesink);
3142     if (sync) {
3143       GST_ELEMENT_WARNING (basesink, STREAM, FAILED,
3144           (_("Internal data flow problem.")),
3145           ("Received buffer without a new-segment. Assuming timestamps start from 0."));
3146     }
3147
3148     /* this means this sink will assume timestamps start from 0 */
3149     GST_OBJECT_LOCK (basesink);
3150     clip_segment->start = 0;
3151     clip_segment->stop = -1;
3152     basesink->segment.start = 0;
3153     basesink->segment.stop = -1;
3154     basesink->have_newsegment = TRUE;
3155     GST_OBJECT_UNLOCK (basesink);
3156   }
3157
3158   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3159
3160   /* check if the buffer needs to be dropped, we first ask the subclass for the
3161    * start and end */
3162   if (bclass->get_times)
3163     bclass->get_times (basesink, buf, &start, &end);
3164
3165   if (start == -1) {
3166     /* if the subclass does not want sync, we use our own values so that we at
3167      * least clip the buffer to the segment */
3168     gst_base_sink_get_times (basesink, buf, &start, &end);
3169   }
3170
3171   GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
3172       ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
3173
3174   /* a dropped buffer does not participate in anything */
3175   if (GST_CLOCK_TIME_IS_VALID (start) &&
3176       (clip_segment->format == GST_FORMAT_TIME)) {
3177     if (G_UNLIKELY (!gst_segment_clip (clip_segment,
3178                 GST_FORMAT_TIME, (gint64) start, (gint64) end, NULL, NULL)))
3179       goto out_of_segment;
3180   }
3181
3182   /* now we can process the buffer in the queue, this function takes ownership
3183    * of the buffer */
3184   result = gst_base_sink_queue_object_unlocked (basesink, pad,
3185       GST_MINI_OBJECT_CAST (buf), TRUE);
3186
3187   return result;
3188
3189   /* ERRORS */
3190 flushing:
3191   {
3192     GST_DEBUG_OBJECT (basesink, "sink is flushing");
3193     gst_buffer_unref (buf);
3194     return GST_FLOW_WRONG_STATE;
3195   }
3196 was_eos:
3197   {
3198     GST_DEBUG_OBJECT (basesink,
3199         "we are EOS, dropping object, return UNEXPECTED");
3200     gst_buffer_unref (buf);
3201     return GST_FLOW_UNEXPECTED;
3202   }
3203 out_of_segment:
3204   {
3205     GST_DEBUG_OBJECT (basesink, "dropping buffer, out of clipping segment");
3206     gst_buffer_unref (buf);
3207     return GST_FLOW_OK;
3208   }
3209 }
3210
3211 /* with STREAM_LOCK
3212  */
3213 static GstFlowReturn
3214 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
3215 {
3216   GstBaseSink *basesink;
3217   GstFlowReturn result;
3218
3219   basesink = GST_BASE_SINK (GST_OBJECT_PARENT (pad));
3220
3221   if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH))
3222     goto wrong_mode;
3223
3224   GST_PAD_PREROLL_LOCK (pad);
3225   result = gst_base_sink_chain_unlocked (basesink, pad, buf);
3226   GST_PAD_PREROLL_UNLOCK (pad);
3227
3228 done:
3229   return result;
3230
3231   /* ERRORS */
3232 wrong_mode:
3233   {
3234     GST_OBJECT_LOCK (pad);
3235     GST_WARNING_OBJECT (basesink,
3236         "Push on pad %s:%s, but it was not activated in push mode",
3237         GST_DEBUG_PAD_NAME (pad));
3238     GST_OBJECT_UNLOCK (pad);
3239     gst_buffer_unref (buf);
3240     /* we don't post an error message this will signal to the peer
3241      * pushing that EOS is reached. */
3242     result = GST_FLOW_UNEXPECTED;
3243     goto done;
3244   }
3245 }
3246
3247 static gboolean
3248 gst_base_sink_default_do_seek (GstBaseSink * sink, GstSegment * segment)
3249 {
3250   gboolean res = TRUE;
3251
3252   /* update our offset if the start/stop position was updated */
3253   if (segment->format == GST_FORMAT_BYTES) {
3254     segment->time = segment->start;
3255   } else if (segment->start == 0) {
3256     /* seek to start, we can implement a default for this. */
3257     segment->time = 0;
3258   } else {
3259     res = FALSE;
3260     GST_INFO_OBJECT (sink, "Can't do a default seek");
3261   }
3262
3263   return res;
3264 }
3265
3266 #define SEEK_TYPE_IS_RELATIVE(t) (((t) != GST_SEEK_TYPE_NONE) && ((t) != GST_SEEK_TYPE_SET))
3267
3268 static gboolean
3269 gst_base_sink_default_prepare_seek_segment (GstBaseSink * sink,
3270     GstEvent * event, GstSegment * segment)
3271 {
3272   /* By default, we try one of 2 things:
3273    *   - For absolute seek positions, convert the requested position to our
3274    *     configured processing format and place it in the output segment \
3275    *   - For relative seek positions, convert our current (input) values to the
3276    *     seek format, adjust by the relative seek offset and then convert back to
3277    *     the processing format
3278    */
3279   GstSeekType cur_type, stop_type;
3280   gint64 cur, stop;
3281   GstSeekFlags flags;
3282   GstFormat seek_format, dest_format;
3283   gdouble rate;
3284   gboolean update;
3285   gboolean res = TRUE;
3286
3287   gst_event_parse_seek (event, &rate, &seek_format, &flags,
3288       &cur_type, &cur, &stop_type, &stop);
3289   dest_format = segment->format;
3290
3291   if (seek_format == dest_format) {
3292     gst_segment_set_seek (segment, rate, seek_format, flags,
3293         cur_type, cur, stop_type, stop, &update);
3294     return TRUE;
3295   }
3296
3297   if (cur_type != GST_SEEK_TYPE_NONE) {
3298     /* FIXME: Handle seek_cur & seek_end by converting the input segment vals */
3299     res =
3300         gst_pad_query_convert (sink->sinkpad, seek_format, cur, &dest_format,
3301         &cur);
3302     cur_type = GST_SEEK_TYPE_SET;
3303   }
3304
3305   if (res && stop_type != GST_SEEK_TYPE_NONE) {
3306     /* FIXME: Handle seek_cur & seek_end by converting the input segment vals */
3307     res =
3308         gst_pad_query_convert (sink->sinkpad, seek_format, stop, &dest_format,
3309         &stop);
3310     stop_type = GST_SEEK_TYPE_SET;
3311   }
3312
3313   /* And finally, configure our output segment in the desired format */
3314   gst_segment_set_seek (segment, rate, dest_format, flags, cur_type, cur,
3315       stop_type, stop, &update);
3316
3317   if (!res)
3318     goto no_format;
3319
3320   return res;
3321
3322 no_format:
3323   {
3324     GST_DEBUG_OBJECT (sink, "undefined format given, seek aborted.");
3325     return FALSE;
3326   }
3327 }
3328
3329 /* perform a seek, only executed in pull mode */
3330 static gboolean
3331 gst_base_sink_perform_seek (GstBaseSink * sink, GstPad * pad, GstEvent * event)
3332 {
3333   gboolean flush;
3334   gdouble rate;
3335   GstFormat seek_format, dest_format;
3336   GstSeekFlags flags;
3337   GstSeekType cur_type, stop_type;
3338   gboolean seekseg_configured = FALSE;
3339   gint64 cur, stop;
3340   gboolean update, res = TRUE;
3341   GstSegment seeksegment;
3342
3343   dest_format = sink->segment.format;
3344
3345   if (event) {
3346     GST_DEBUG_OBJECT (sink, "performing seek with event %p", event);
3347     gst_event_parse_seek (event, &rate, &seek_format, &flags,
3348         &cur_type, &cur, &stop_type, &stop);
3349
3350     flush = flags & GST_SEEK_FLAG_FLUSH;
3351   } else {
3352     GST_DEBUG_OBJECT (sink, "performing seek without event");
3353     flush = FALSE;
3354   }
3355
3356   if (flush) {
3357     GST_DEBUG_OBJECT (sink, "flushing upstream");
3358     gst_pad_push_event (pad, gst_event_new_flush_start ());
3359     gst_base_sink_flush_start (sink, pad);
3360   } else {
3361     GST_DEBUG_OBJECT (sink, "pausing pulling thread");
3362   }
3363
3364   GST_PAD_STREAM_LOCK (pad);
3365
3366   /* If we configured the seeksegment above, don't overwrite it now. Otherwise
3367    * copy the current segment info into the temp segment that we can actually
3368    * attempt the seek with. We only update the real segment if the seek suceeds. */
3369   if (!seekseg_configured) {
3370     memcpy (&seeksegment, &sink->segment, sizeof (GstSegment));
3371
3372     /* now configure the final seek segment */
3373     if (event) {
3374       if (sink->segment.format != seek_format) {
3375         /* OK, here's where we give the subclass a chance to convert the relative
3376          * seek into an absolute one in the processing format. We set up any
3377          * absolute seek above, before taking the stream lock. */
3378         if (!gst_base_sink_default_prepare_seek_segment (sink, event,
3379                 &seeksegment)) {
3380           GST_DEBUG_OBJECT (sink,
3381               "Preparing the seek failed after flushing. " "Aborting seek");
3382           res = FALSE;
3383         }
3384       } else {
3385         /* The seek format matches our processing format, no need to ask the
3386          * the subclass to configure the segment. */
3387         gst_segment_set_seek (&seeksegment, rate, seek_format, flags,
3388             cur_type, cur, stop_type, stop, &update);
3389       }
3390     }
3391     /* Else, no seek event passed, so we're just (re)starting the
3392        current segment. */
3393   }
3394
3395   if (res) {
3396     GST_DEBUG_OBJECT (sink, "segment configured from %" G_GINT64_FORMAT
3397         " to %" G_GINT64_FORMAT ", position %" G_GINT64_FORMAT,
3398         seeksegment.start, seeksegment.stop, seeksegment.last_stop);
3399
3400     /* do the seek, segment.last_stop contains the new position. */
3401     res = gst_base_sink_default_do_seek (sink, &seeksegment);
3402   }
3403
3404
3405   if (flush) {
3406     GST_DEBUG_OBJECT (sink, "stop flushing upstream");
3407     gst_pad_push_event (pad, gst_event_new_flush_stop ());
3408     gst_base_sink_flush_stop (sink, pad);
3409   } else if (res && sink->abidata.ABI.running) {
3410     /* we are running the current segment and doing a non-flushing seek,
3411      * close the segment first based on the last_stop. */
3412     GST_DEBUG_OBJECT (sink, "closing running segment %" G_GINT64_FORMAT
3413         " to %" G_GINT64_FORMAT, sink->segment.start, sink->segment.last_stop);
3414   }
3415
3416   /* The subclass must have converted the segment to the processing format
3417    * by now */
3418   if (res && seeksegment.format != dest_format) {
3419     GST_DEBUG_OBJECT (sink, "Subclass failed to prepare a seek segment "
3420         "in the correct format. Aborting seek.");
3421     res = FALSE;
3422   }
3423
3424   /* if successfull seek, we update our real segment and push
3425    * out the new segment. */
3426   if (res) {
3427     memcpy (&sink->segment, &seeksegment, sizeof (GstSegment));
3428
3429     if (sink->segment.flags & GST_SEEK_FLAG_SEGMENT) {
3430       gst_element_post_message (GST_ELEMENT (sink),
3431           gst_message_new_segment_start (GST_OBJECT (sink),
3432               sink->segment.format, sink->segment.last_stop));
3433     }
3434   }
3435
3436   sink->priv->discont = TRUE;
3437   sink->abidata.ABI.running = TRUE;
3438
3439   GST_PAD_STREAM_UNLOCK (pad);
3440
3441   return res;
3442 }
3443
3444 static void
3445 set_step_info (GstBaseSink * sink, GstStepInfo * current, GstStepInfo * pending,
3446     guint seqnum, GstFormat format, guint64 amount, gdouble rate,
3447     gboolean flush, gboolean intermediate)
3448 {
3449   GST_OBJECT_LOCK (sink);
3450   pending->seqnum = seqnum;
3451   pending->format = format;
3452   pending->amount = amount;
3453   pending->position = 0;
3454   pending->rate = rate;
3455   pending->flush = flush;
3456   pending->intermediate = intermediate;
3457   pending->valid = TRUE;
3458   /* flush invalidates the current stepping segment */
3459   if (flush)
3460     current->valid = FALSE;
3461   GST_OBJECT_UNLOCK (sink);
3462 }
3463
3464 static gboolean
3465 gst_base_sink_perform_step (GstBaseSink * sink, GstPad * pad, GstEvent * event)
3466 {
3467   GstBaseSinkPrivate *priv;
3468   GstBaseSinkClass *bclass;
3469   gboolean flush, intermediate;
3470   gdouble rate;
3471   GstFormat format;
3472   guint64 amount;
3473   guint seqnum;
3474   GstStepInfo *pending, *current;
3475
3476   bclass = GST_BASE_SINK_GET_CLASS (sink);
3477   priv = sink->priv;
3478
3479   GST_DEBUG_OBJECT (sink, "performing step with event %p", event);
3480
3481   gst_event_parse_step (event, &format, &amount, &rate, &flush, &intermediate);
3482   seqnum = gst_event_get_seqnum (event);
3483
3484   pending = &priv->pending_step;
3485   current = &priv->current_step;
3486
3487   if (flush) {
3488     /* we need to call ::unlock before locking PREROLL_LOCK
3489      * since we lock it before going into ::render */
3490     if (bclass->unlock)
3491       bclass->unlock (sink);
3492
3493     GST_PAD_PREROLL_LOCK (sink->sinkpad);
3494     /* now that we have the PREROLL lock, clear our unlock request */
3495     if (bclass->unlock_stop)
3496       bclass->unlock_stop (sink);
3497
3498     /* update the stepinfo and make it valid */
3499     set_step_info (sink, current, pending, seqnum, format, amount, rate, flush,
3500         intermediate);
3501
3502     if (sink->priv->async_enabled) {
3503       /* and we need to commit our state again on the next
3504        * prerolled buffer */
3505       sink->playing_async = TRUE;
3506       priv->pending_step.need_preroll = TRUE;
3507       sink->need_preroll = FALSE;
3508       gst_element_lost_state_full (GST_ELEMENT_CAST (sink), FALSE);
3509     } else {
3510       sink->priv->have_latency = TRUE;
3511       sink->need_preroll = FALSE;
3512     }
3513     priv->current_sstart = -1;
3514     priv->current_sstop = -1;
3515     priv->eos_rtime = -1;
3516     priv->call_preroll = TRUE;
3517     gst_base_sink_set_last_buffer (sink, NULL);
3518     gst_base_sink_reset_qos (sink);
3519
3520     if (sink->clock_id) {
3521       gst_clock_id_unschedule (sink->clock_id);
3522     }
3523
3524     if (sink->have_preroll) {
3525       GST_DEBUG_OBJECT (sink, "signal waiter");
3526       priv->step_unlock = TRUE;
3527       GST_PAD_PREROLL_SIGNAL (sink->sinkpad);
3528     }
3529     GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
3530   } else {
3531     /* update the stepinfo and make it valid */
3532     set_step_info (sink, current, pending, seqnum, format, amount, rate, flush,
3533         intermediate);
3534   }
3535
3536   return TRUE;
3537 }
3538
3539 /* with STREAM_LOCK
3540  */
3541 static void
3542 gst_base_sink_loop (GstPad * pad)
3543 {
3544   GstBaseSink *basesink;
3545   GstBuffer *buf = NULL;
3546   GstFlowReturn result;
3547   guint blocksize;
3548   guint64 offset;
3549
3550   basesink = GST_BASE_SINK (GST_OBJECT_PARENT (pad));
3551
3552   g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
3553
3554   if ((blocksize = basesink->priv->blocksize) == 0)
3555     blocksize = -1;
3556
3557   offset = basesink->segment.last_stop;
3558
3559   GST_DEBUG_OBJECT (basesink, "pulling %" G_GUINT64_FORMAT ", %u",
3560       offset, blocksize);
3561
3562   result = gst_pad_pull_range (pad, offset, blocksize, &buf);
3563   if (G_UNLIKELY (result != GST_FLOW_OK))
3564     goto paused;
3565
3566   if (G_UNLIKELY (buf == NULL))
3567     goto no_buffer;
3568
3569   offset += GST_BUFFER_SIZE (buf);
3570
3571   gst_segment_set_last_stop (&basesink->segment, GST_FORMAT_BYTES, offset);
3572
3573   GST_PAD_PREROLL_LOCK (pad);
3574   result = gst_base_sink_chain_unlocked (basesink, pad, buf);
3575   GST_PAD_PREROLL_UNLOCK (pad);
3576   if (G_UNLIKELY (result != GST_FLOW_OK))
3577     goto paused;
3578
3579   return;
3580
3581   /* ERRORS */
3582 paused:
3583   {
3584     GST_LOG_OBJECT (basesink, "pausing task, reason %s",
3585         gst_flow_get_name (result));
3586     gst_pad_pause_task (pad);
3587     /* fatal errors and NOT_LINKED cause EOS */
3588     if (GST_FLOW_IS_FATAL (result) || result == GST_FLOW_NOT_LINKED) {
3589       if (result == GST_FLOW_UNEXPECTED) {
3590         /* perform EOS logic */
3591         if (basesink->segment.flags & GST_SEEK_FLAG_SEGMENT) {
3592           gst_element_post_message (GST_ELEMENT_CAST (basesink),
3593               gst_message_new_segment_done (GST_OBJECT_CAST (basesink),
3594                   basesink->segment.format, basesink->segment.last_stop));
3595         } else {
3596           gst_base_sink_event (pad, gst_event_new_eos ());
3597         }
3598       } else {
3599         /* for fatal errors we post an error message, post the error
3600          * first so the app knows about the error first. */
3601         GST_ELEMENT_ERROR (basesink, STREAM, FAILED,
3602             (_("Internal data stream error.")),
3603             ("stream stopped, reason %s", gst_flow_get_name (result)));
3604         gst_base_sink_event (pad, gst_event_new_eos ());
3605       }
3606     }
3607     return;
3608   }
3609 no_buffer:
3610   {
3611     GST_LOG_OBJECT (basesink, "no buffer, pausing");
3612     GST_ELEMENT_ERROR (basesink, STREAM, FAILED,
3613         (_("Internal data flow error.")), ("element returned NULL buffer"));
3614     result = GST_FLOW_ERROR;
3615     goto paused;
3616   }
3617 }
3618
3619 static gboolean
3620 gst_base_sink_set_flushing (GstBaseSink * basesink, GstPad * pad,
3621     gboolean flushing)
3622 {
3623   GstBaseSinkClass *bclass;
3624
3625   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3626
3627   if (flushing) {
3628     /* unlock any subclasses, we need to do this before grabbing the
3629      * PREROLL_LOCK since we hold this lock before going into ::render. */
3630     if (bclass->unlock)
3631       bclass->unlock (basesink);
3632   }
3633
3634   GST_PAD_PREROLL_LOCK (pad);
3635   basesink->flushing = flushing;
3636   if (flushing) {
3637     /* step 1, now that we have the PREROLL lock, clear our unlock request */
3638     if (bclass->unlock_stop)
3639       bclass->unlock_stop (basesink);
3640
3641     /* set need_preroll before we unblock the clock. If the clock is unblocked
3642      * before timing out, we can reuse the buffer for preroll. */
3643     basesink->need_preroll = TRUE;
3644
3645     /* step 2, unblock clock sync (if any) or any other blocking thing */
3646     if (basesink->clock_id) {
3647       gst_clock_id_unschedule (basesink->clock_id);
3648     }
3649
3650     /* flush out the data thread if it's locked in finish_preroll, this will
3651      * also flush out the EOS state */
3652     GST_DEBUG_OBJECT (basesink,
3653         "flushing out data thread, need preroll to TRUE");
3654     gst_base_sink_preroll_queue_flush (basesink, pad);
3655   }
3656   GST_PAD_PREROLL_UNLOCK (pad);
3657
3658   return TRUE;
3659 }
3660
3661 static gboolean
3662 gst_base_sink_default_activate_pull (GstBaseSink * basesink, gboolean active)
3663 {
3664   gboolean result;
3665
3666   if (active) {
3667     /* start task */
3668     result = gst_pad_start_task (basesink->sinkpad,
3669         (GstTaskFunction) gst_base_sink_loop, basesink->sinkpad);
3670   } else {
3671     /* step 2, make sure streaming finishes */
3672     result = gst_pad_stop_task (basesink->sinkpad);
3673   }
3674
3675   return result;
3676 }
3677
3678 static gboolean
3679 gst_base_sink_pad_activate (GstPad * pad)
3680 {
3681   gboolean result = FALSE;
3682   GstBaseSink *basesink;
3683
3684   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3685
3686   GST_DEBUG_OBJECT (basesink, "Trying pull mode first");
3687
3688   gst_base_sink_set_flushing (basesink, pad, FALSE);
3689
3690   /* we need to have the pull mode enabled */
3691   if (!basesink->can_activate_pull) {
3692     GST_DEBUG_OBJECT (basesink, "pull mode disabled");
3693     goto fallback;
3694   }
3695
3696   /* check if downstreams supports pull mode at all */
3697   if (!gst_pad_check_pull_range (pad)) {
3698     GST_DEBUG_OBJECT (basesink, "pull mode not supported");
3699     goto fallback;
3700   }
3701
3702   /* set the pad mode before starting the task so that it's in the
3703    * correct state for the new thread. also the sink set_caps and get_caps
3704    * function checks this */
3705   basesink->pad_mode = GST_ACTIVATE_PULL;
3706
3707   /* we first try to negotiate a format so that when we try to activate
3708    * downstream, it knows about our format */
3709   if (!gst_base_sink_negotiate_pull (basesink)) {
3710     GST_DEBUG_OBJECT (basesink, "failed to negotiate in pull mode");
3711     goto fallback;
3712   }
3713
3714   /* ok activate now */
3715   if (!gst_pad_activate_pull (pad, TRUE)) {
3716     /* clear any pending caps */
3717     GST_OBJECT_LOCK (basesink);
3718     gst_caps_replace (&basesink->priv->pull_caps, NULL);
3719     GST_OBJECT_UNLOCK (basesink);
3720     GST_DEBUG_OBJECT (basesink, "failed to activate in pull mode");
3721     goto fallback;
3722   }
3723
3724   GST_DEBUG_OBJECT (basesink, "Success activating pull mode");
3725   result = TRUE;
3726   goto done;
3727
3728   /* push mode fallback */
3729 fallback:
3730   GST_DEBUG_OBJECT (basesink, "Falling back to push mode");
3731   if ((result = gst_pad_activate_push (pad, TRUE))) {
3732     GST_DEBUG_OBJECT (basesink, "Success activating push mode");
3733   }
3734
3735 done:
3736   if (!result) {
3737     GST_WARNING_OBJECT (basesink, "Could not activate pad in either mode");
3738     gst_base_sink_set_flushing (basesink, pad, TRUE);
3739   }
3740
3741   gst_object_unref (basesink);
3742
3743   return result;
3744 }
3745
3746 static gboolean
3747 gst_base_sink_pad_activate_push (GstPad * pad, gboolean active)
3748 {
3749   gboolean result;
3750   GstBaseSink *basesink;
3751
3752   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3753
3754   if (active) {
3755     if (!basesink->can_activate_push) {
3756       result = FALSE;
3757       basesink->pad_mode = GST_ACTIVATE_NONE;
3758     } else {
3759       result = TRUE;
3760       basesink->pad_mode = GST_ACTIVATE_PUSH;
3761     }
3762   } else {
3763     if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH)) {
3764       g_warning ("Internal GStreamer activation error!!!");
3765       result = FALSE;
3766     } else {
3767       gst_base_sink_set_flushing (basesink, pad, TRUE);
3768       result = TRUE;
3769       basesink->pad_mode = GST_ACTIVATE_NONE;
3770     }
3771   }
3772
3773   gst_object_unref (basesink);
3774
3775   return result;
3776 }
3777
3778 static gboolean
3779 gst_base_sink_negotiate_pull (GstBaseSink * basesink)
3780 {
3781   GstCaps *caps;
3782   gboolean result;
3783
3784   result = FALSE;
3785
3786   /* this returns the intersection between our caps and the peer caps. If there
3787    * is no peer, it returns NULL and we can't operate in pull mode so we can
3788    * fail the negotiation. */
3789   caps = gst_pad_get_allowed_caps (GST_BASE_SINK_PAD (basesink));
3790   if (caps == NULL || gst_caps_is_empty (caps))
3791     goto no_caps_possible;
3792
3793   GST_DEBUG_OBJECT (basesink, "allowed caps: %" GST_PTR_FORMAT, caps);
3794
3795   caps = gst_caps_make_writable (caps);
3796   /* get the first (prefered) format */
3797   gst_caps_truncate (caps);
3798   /* try to fixate */
3799   gst_pad_fixate_caps (GST_BASE_SINK_PAD (basesink), caps);
3800
3801   GST_DEBUG_OBJECT (basesink, "fixated to: %" GST_PTR_FORMAT, caps);
3802
3803   if (gst_caps_is_any (caps)) {
3804     GST_DEBUG_OBJECT (basesink, "caps were ANY after fixating, "
3805         "allowing pull()");
3806     /* neither side has template caps in this case, so they are prepared for
3807        pull() without setcaps() */
3808     result = TRUE;
3809   } else if (gst_caps_is_fixed (caps)) {
3810     if (!gst_pad_set_caps (GST_BASE_SINK_PAD (basesink), caps))
3811       goto could_not_set_caps;
3812
3813     GST_OBJECT_LOCK (basesink);
3814     gst_caps_replace (&basesink->priv->pull_caps, caps);
3815     GST_OBJECT_UNLOCK (basesink);
3816
3817     result = TRUE;
3818   }
3819
3820   gst_caps_unref (caps);
3821
3822   return result;
3823
3824 no_caps_possible:
3825   {
3826     GST_INFO_OBJECT (basesink, "Pipeline could not agree on caps");
3827     GST_DEBUG_OBJECT (basesink, "get_allowed_caps() returned EMPTY");
3828     if (caps)
3829       gst_caps_unref (caps);
3830     return FALSE;
3831   }
3832 could_not_set_caps:
3833   {
3834     GST_INFO_OBJECT (basesink, "Could not set caps: %" GST_PTR_FORMAT, caps);
3835     gst_caps_unref (caps);
3836     return FALSE;
3837   }
3838 }
3839
3840 /* this won't get called until we implement an activate function */
3841 static gboolean
3842 gst_base_sink_pad_activate_pull (GstPad * pad, gboolean active)
3843 {
3844   gboolean result = FALSE;
3845   GstBaseSink *basesink;
3846   GstBaseSinkClass *bclass;
3847
3848   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3849   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3850
3851   if (active) {
3852     GstFormat format;
3853     gint64 duration;
3854
3855     /* we mark we have a newsegment here because pull based
3856      * mode works just fine without having a newsegment before the
3857      * first buffer */
3858     format = GST_FORMAT_BYTES;
3859
3860     gst_segment_init (&basesink->segment, format);
3861     gst_segment_init (basesink->abidata.ABI.clip_segment, format);
3862     GST_OBJECT_LOCK (basesink);
3863     basesink->have_newsegment = TRUE;
3864     GST_OBJECT_UNLOCK (basesink);
3865
3866     /* get the peer duration in bytes */
3867     result = gst_pad_query_peer_duration (pad, &format, &duration);
3868     if (result) {
3869       GST_DEBUG_OBJECT (basesink,
3870           "setting duration in bytes to %" G_GINT64_FORMAT, duration);
3871       gst_segment_set_duration (basesink->abidata.ABI.clip_segment, format,
3872           duration);
3873       gst_segment_set_duration (&basesink->segment, format, duration);
3874     } else {
3875       GST_DEBUG_OBJECT (basesink, "unknown duration");
3876     }
3877
3878     if (bclass->activate_pull)
3879       result = bclass->activate_pull (basesink, TRUE);
3880     else
3881       result = FALSE;
3882
3883     if (!result)
3884       goto activate_failed;
3885
3886   } else {
3887     if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PULL)) {
3888       g_warning ("Internal GStreamer activation error!!!");
3889       result = FALSE;
3890     } else {
3891       result = gst_base_sink_set_flushing (basesink, pad, TRUE);
3892       if (bclass->activate_pull)
3893         result &= bclass->activate_pull (basesink, FALSE);
3894       basesink->pad_mode = GST_ACTIVATE_NONE;
3895       /* clear any pending caps */
3896       GST_OBJECT_LOCK (basesink);
3897       gst_caps_replace (&basesink->priv->pull_caps, NULL);
3898       GST_OBJECT_UNLOCK (basesink);
3899     }
3900   }
3901   gst_object_unref (basesink);
3902
3903   return result;
3904
3905   /* ERRORS */
3906 activate_failed:
3907   {
3908     /* reset, as starting the thread failed */
3909     basesink->pad_mode = GST_ACTIVATE_NONE;
3910
3911     GST_ERROR_OBJECT (basesink, "subclass failed to activate in pull mode");
3912     return FALSE;
3913   }
3914 }
3915
3916 /* send an event to our sinkpad peer. */
3917 static gboolean
3918 gst_base_sink_send_event (GstElement * element, GstEvent * event)
3919 {
3920   GstPad *pad;
3921   GstBaseSink *basesink = GST_BASE_SINK (element);
3922   gboolean forward, result = TRUE;
3923   GstActivateMode mode;
3924
3925   GST_OBJECT_LOCK (element);
3926   /* get the pad and the scheduling mode */
3927   pad = gst_object_ref (basesink->sinkpad);
3928   mode = basesink->pad_mode;
3929   GST_OBJECT_UNLOCK (element);
3930
3931   /* only push UPSTREAM events upstream */
3932   forward = GST_EVENT_IS_UPSTREAM (event);
3933
3934   switch (GST_EVENT_TYPE (event)) {
3935     case GST_EVENT_LATENCY:
3936     {
3937       GstClockTime latency;
3938
3939       gst_event_parse_latency (event, &latency);
3940
3941       /* store the latency. We use this to adjust the running_time before syncing
3942        * it to the clock. */
3943       GST_OBJECT_LOCK (element);
3944       basesink->priv->latency = latency;
3945       if (!basesink->priv->have_latency)
3946         forward = FALSE;
3947       GST_OBJECT_UNLOCK (element);
3948       GST_DEBUG_OBJECT (basesink, "latency set to %" GST_TIME_FORMAT,
3949           GST_TIME_ARGS (latency));
3950
3951       /* We forward this event so that all elements know about the global pipeline
3952        * latency. This is interesting for an element when it wants to figure out
3953        * when a particular piece of data will be rendered. */
3954       break;
3955     }
3956     case GST_EVENT_SEEK:
3957       /* in pull mode we will execute the seek */
3958       if (mode == GST_ACTIVATE_PULL)
3959         result = gst_base_sink_perform_seek (basesink, pad, event);
3960       break;
3961     case GST_EVENT_STEP:
3962       result = gst_base_sink_perform_step (basesink, pad, event);
3963       forward = FALSE;
3964       break;
3965     default:
3966       break;
3967   }
3968
3969   if (forward) {
3970     result = gst_pad_push_event (pad, event);
3971   } else {
3972     /* not forwarded, unref the event */
3973     gst_event_unref (event);
3974   }
3975
3976   gst_object_unref (pad);
3977   return result;
3978 }
3979
3980 static gboolean
3981 gst_base_sink_peer_query (GstBaseSink * sink, GstQuery * query)
3982 {
3983   GstPad *peer;
3984   gboolean res = FALSE;
3985
3986   if ((peer = gst_pad_get_peer (sink->sinkpad))) {
3987     res = gst_pad_query (peer, query);
3988     gst_object_unref (peer);
3989   }
3990   return res;
3991 }
3992
3993 /* get the end position of the last seen object, this is used
3994  * for EOS and for making sure that we don't report a position we
3995  * have not reached yet. With LOCK. */
3996 static gboolean
3997 gst_base_sink_get_position_last (GstBaseSink * basesink, GstFormat format,
3998     gint64 * cur)
3999 {
4000   GstFormat oformat;
4001   GstSegment *segment;
4002   gboolean ret = TRUE;
4003
4004   segment = &basesink->segment;
4005   oformat = segment->format;
4006
4007   if (oformat == GST_FORMAT_TIME) {
4008     /* return last observed stream time, we keep the stream time around in the
4009      * time format. */
4010     *cur = basesink->priv->current_sstop;
4011   } else {
4012     /* convert last stop to stream time */
4013     *cur = gst_segment_to_stream_time (segment, oformat, segment->last_stop);
4014   }
4015
4016   if (*cur != -1 && oformat != format) {
4017     GST_OBJECT_UNLOCK (basesink);
4018     /* convert to the target format if we need to, release lock first */
4019     ret =
4020         gst_pad_query_convert (basesink->sinkpad, oformat, *cur, &format, cur);
4021     if (!ret)
4022       *cur = -1;
4023     GST_OBJECT_LOCK (basesink);
4024   }
4025
4026   GST_DEBUG_OBJECT (basesink, "POSITION: %" GST_TIME_FORMAT,
4027       GST_TIME_ARGS (*cur));
4028
4029   return ret;
4030 }
4031
4032 /* get the position when we are PAUSED, this is the stream time of the buffer
4033  * that prerolled. If no buffer is prerolled (we are still flushing), this
4034  * value will be -1. With LOCK. */
4035 static gboolean
4036 gst_base_sink_get_position_paused (GstBaseSink * basesink, GstFormat format,
4037     gint64 * cur)
4038 {
4039   gboolean res;
4040   gint64 time;
4041   GstSegment *segment;
4042   GstFormat oformat;
4043
4044   /* we don't use the clip segment in pull mode, when seeking we update the
4045    * main segment directly with the new segment values without it having to be
4046    * activated by the rendering after preroll */
4047   if (basesink->pad_mode == GST_ACTIVATE_PUSH)
4048     segment = basesink->abidata.ABI.clip_segment;
4049   else
4050     segment = &basesink->segment;
4051   oformat = segment->format;
4052
4053   if (oformat == GST_FORMAT_TIME) {
4054     *cur = basesink->priv->current_sstart;
4055     if (segment->rate < 0.0 && basesink->priv->current_sstop != -1) {
4056       /* for reverse playback we prefer the stream time stop position if we have
4057        * one */
4058       *cur = basesink->priv->current_sstop;
4059     }
4060   } else {
4061     *cur = gst_segment_to_stream_time (segment, oformat, segment->last_stop);
4062   }
4063
4064   time = segment->time;
4065
4066   if (*cur != -1) {
4067     *cur = MAX (*cur, time);
4068     GST_DEBUG_OBJECT (basesink, "POSITION as max: %" GST_TIME_FORMAT
4069         ", time %" GST_TIME_FORMAT, GST_TIME_ARGS (*cur), GST_TIME_ARGS (time));
4070   } else {
4071     /* we have no buffer, use the segment times. */
4072     if (segment->rate >= 0.0) {
4073       /* forward, next position is always the time of the segment */
4074       *cur = time;
4075       GST_DEBUG_OBJECT (basesink, "POSITION as time: %" GST_TIME_FORMAT,
4076           GST_TIME_ARGS (*cur));
4077     } else {
4078       /* reverse, next expected timestamp is segment->stop. We use the function
4079        * to get things right for negative applied_rates. */
4080       *cur = gst_segment_to_stream_time (segment, oformat, segment->stop);
4081       GST_DEBUG_OBJECT (basesink, "reverse POSITION: %" GST_TIME_FORMAT,
4082           GST_TIME_ARGS (*cur));
4083     }
4084   }
4085
4086   res = (*cur != -1);
4087   if (res && oformat != format) {
4088     GST_OBJECT_UNLOCK (basesink);
4089     res =
4090         gst_pad_query_convert (basesink->sinkpad, oformat, *cur, &format, cur);
4091     if (!res)
4092       *cur = -1;
4093     GST_OBJECT_LOCK (basesink);
4094   }
4095
4096   return res;
4097 }
4098
4099 static gboolean
4100 gst_base_sink_get_position (GstBaseSink * basesink, GstFormat format,
4101     gint64 * cur, gboolean * upstream)
4102 {
4103   GstClock *clock;
4104   gboolean res = FALSE;
4105   GstFormat oformat, tformat;
4106   GstClockTime now, base, latency;
4107   gint64 time, accum, duration;
4108   gdouble rate;
4109   gint64 last;
4110
4111   GST_OBJECT_LOCK (basesink);
4112   /* our intermediate time format */
4113   tformat = GST_FORMAT_TIME;
4114   /* get the format in the segment */
4115   oformat = basesink->segment.format;
4116
4117   /* can only give answer based on the clock if not EOS */
4118   if (G_UNLIKELY (basesink->eos))
4119     goto in_eos;
4120
4121   /* we can only get the segment when we are not NULL or READY */
4122   if (!basesink->have_newsegment)
4123     goto wrong_state;
4124
4125   /* when not in PLAYING or when we're busy with a state change, we
4126    * cannot read from the clock so we report time based on the
4127    * last seen timestamp. */
4128   if (GST_STATE (basesink) != GST_STATE_PLAYING ||
4129       GST_STATE_PENDING (basesink) != GST_STATE_VOID_PENDING)
4130     goto in_pause;
4131
4132   /* we need to sync on the clock. */
4133   if (basesink->sync == FALSE)
4134     goto no_sync;
4135
4136   /* and we need a clock */
4137   if (G_UNLIKELY ((clock = GST_ELEMENT_CLOCK (basesink)) == NULL))
4138     goto no_sync;
4139
4140   /* collect all data we need holding the lock */
4141   if (GST_CLOCK_TIME_IS_VALID (basesink->segment.time))
4142     time = basesink->segment.time;
4143   else
4144     time = 0;
4145
4146   if (GST_CLOCK_TIME_IS_VALID (basesink->segment.stop))
4147     duration = basesink->segment.stop - basesink->segment.start;
4148   else
4149     duration = 0;
4150
4151   base = GST_ELEMENT_CAST (basesink)->base_time;
4152   accum = basesink->segment.accum;
4153   rate = basesink->segment.rate * basesink->segment.applied_rate;
4154   latency = basesink->priv->latency;
4155
4156   gst_object_ref (clock);
4157
4158   /* this function might release the LOCK */
4159   gst_base_sink_get_position_last (basesink, format, &last);
4160
4161   /* need to release the object lock before we can get the time,
4162    * a clock might take the LOCK of the provider, which could be
4163    * a basesink subclass. */
4164   GST_OBJECT_UNLOCK (basesink);
4165
4166   now = gst_clock_get_time (clock);
4167
4168   if (oformat != tformat) {
4169     /* convert accum, time and duration to time */
4170     if (!gst_pad_query_convert (basesink->sinkpad, oformat, accum, &tformat,
4171             &accum))
4172       goto convert_failed;
4173     if (!gst_pad_query_convert (basesink->sinkpad, oformat, duration, &tformat,
4174             &duration))
4175       goto convert_failed;
4176     if (!gst_pad_query_convert (basesink->sinkpad, oformat, time, &tformat,
4177             &time))
4178       goto convert_failed;
4179   }
4180
4181   /* subtract base time and accumulated time from the clock time.
4182    * Make sure we don't go negative. This is the current time in
4183    * the segment which we need to scale with the combined
4184    * rate and applied rate. */
4185   base += accum;
4186   base += latency;
4187   base = MIN (now, base);
4188
4189   /* for negative rates we need to count back from from the segment
4190    * duration. */
4191   if (rate < 0.0)
4192     time += duration;
4193
4194   *cur = time + gst_guint64_to_gdouble (now - base) * rate;
4195
4196   /* never report more than last seen position */
4197   if (last != -1)
4198     *cur = MIN (last, *cur);
4199
4200   gst_object_unref (clock);
4201
4202   GST_DEBUG_OBJECT (basesink,
4203       "now %" GST_TIME_FORMAT " - base %" GST_TIME_FORMAT " - accum %"
4204       GST_TIME_FORMAT " + time %" GST_TIME_FORMAT,
4205       GST_TIME_ARGS (now), GST_TIME_ARGS (base),
4206       GST_TIME_ARGS (accum), GST_TIME_ARGS (time));
4207
4208   if (oformat != format) {
4209     /* convert time to final format */
4210     if (!gst_pad_query_convert (basesink->sinkpad, tformat, *cur, &format, cur))
4211       goto convert_failed;
4212   }
4213
4214   res = TRUE;
4215
4216 done:
4217   GST_DEBUG_OBJECT (basesink, "res: %d, POSITION: %" GST_TIME_FORMAT,
4218       res, GST_TIME_ARGS (*cur));
4219   return res;
4220
4221   /* special cases */
4222 in_eos:
4223   {
4224     GST_DEBUG_OBJECT (basesink, "position in EOS");
4225     res = gst_base_sink_get_position_last (basesink, format, cur);
4226     GST_OBJECT_UNLOCK (basesink);
4227     goto done;
4228   }
4229 in_pause:
4230   {
4231     GST_DEBUG_OBJECT (basesink, "position in PAUSED");
4232     res = gst_base_sink_get_position_paused (basesink, format, cur);
4233     GST_OBJECT_UNLOCK (basesink);
4234     goto done;
4235   }
4236 wrong_state:
4237   {
4238     /* in NULL or READY we always return FALSE and -1 */
4239     GST_DEBUG_OBJECT (basesink, "position in wrong state, return -1");
4240     res = FALSE;
4241     *cur = -1;
4242     GST_OBJECT_UNLOCK (basesink);
4243     goto done;
4244   }
4245 no_sync:
4246   {
4247     /* report last seen timestamp if any, else ask upstream to answer */
4248     if ((*cur = basesink->priv->current_sstart) != -1)
4249       res = TRUE;
4250     else
4251       *upstream = TRUE;
4252
4253     GST_DEBUG_OBJECT (basesink, "no sync, res %d, POSITION %" GST_TIME_FORMAT,
4254         res, GST_TIME_ARGS (*cur));
4255     GST_OBJECT_UNLOCK (basesink);
4256     return res;
4257   }
4258 convert_failed:
4259   {
4260     GST_DEBUG_OBJECT (basesink, "convert failed, try upstream");
4261     *upstream = TRUE;
4262     return FALSE;
4263   }
4264 }
4265
4266 static gboolean
4267 gst_base_sink_query (GstElement * element, GstQuery * query)
4268 {
4269   gboolean res = FALSE;
4270
4271   GstBaseSink *basesink = GST_BASE_SINK (element);
4272
4273   switch (GST_QUERY_TYPE (query)) {
4274     case GST_QUERY_POSITION:
4275     {
4276       gint64 cur = 0;
4277       GstFormat format;
4278       gboolean upstream = FALSE;
4279
4280       gst_query_parse_position (query, &format, NULL);
4281
4282       GST_DEBUG_OBJECT (basesink, "position format %d", format);
4283
4284       /* first try to get the position based on the clock */
4285       if ((res =
4286               gst_base_sink_get_position (basesink, format, &cur, &upstream))) {
4287         gst_query_set_position (query, format, cur);
4288       } else if (upstream) {
4289         /* fallback to peer query */
4290         res = gst_base_sink_peer_query (basesink, query);
4291       }
4292       break;
4293     }
4294     case GST_QUERY_DURATION:
4295     {
4296       GstFormat format, uformat;
4297       gint64 duration, uduration;
4298
4299       gst_query_parse_duration (query, &format, NULL);
4300
4301       GST_DEBUG_OBJECT (basesink, "duration query in format %s",
4302           gst_format_get_name (format));
4303
4304       if (basesink->pad_mode == GST_ACTIVATE_PULL) {
4305         uformat = GST_FORMAT_BYTES;
4306
4307         /* get the duration in bytes, in pull mode that's all we are sure to
4308          * know. We have to explicitly get this value from upstream instead of
4309          * using our cached value because it might change. Duration caching
4310          * should be done at a higher level. */
4311         res = gst_pad_query_peer_duration (basesink->sinkpad, &uformat,
4312             &uduration);
4313         if (res) {
4314           gst_segment_set_duration (&basesink->segment, uformat, uduration);
4315           if (format != uformat) {
4316             /* convert to the requested format */
4317             res = gst_pad_query_convert (basesink->sinkpad, uformat, uduration,
4318                 &format, &duration);
4319           } else {
4320             duration = uduration;
4321           }
4322           if (res) {
4323             /* set the result */
4324             gst_query_set_duration (query, format, duration);
4325           }
4326         }
4327       } else {
4328         /* in push mode we simply forward upstream */
4329         res = gst_base_sink_peer_query (basesink, query);
4330       }
4331       break;
4332     }
4333     case GST_QUERY_LATENCY:
4334     {
4335       gboolean live, us_live;
4336       GstClockTime min, max;
4337
4338       if ((res = gst_base_sink_query_latency (basesink, &live, &us_live, &min,
4339                   &max))) {
4340         gst_query_set_latency (query, live, min, max);
4341       }
4342       break;
4343     }
4344     case GST_QUERY_JITTER:
4345       break;
4346     case GST_QUERY_RATE:
4347       /* gst_query_set_rate (query, basesink->segment_rate); */
4348       res = TRUE;
4349       break;
4350     case GST_QUERY_SEGMENT:
4351     {
4352       /* FIXME, bring start/stop to stream time */
4353       gst_query_set_segment (query, basesink->segment.rate,
4354           GST_FORMAT_TIME, basesink->segment.start, basesink->segment.stop);
4355       break;
4356     }
4357     case GST_QUERY_SEEKING:
4358     case GST_QUERY_CONVERT:
4359     case GST_QUERY_FORMATS:
4360     default:
4361       res = gst_base_sink_peer_query (basesink, query);
4362       break;
4363   }
4364   return res;
4365 }
4366
4367 static GstStateChangeReturn
4368 gst_base_sink_change_state (GstElement * element, GstStateChange transition)
4369 {
4370   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
4371   GstBaseSink *basesink = GST_BASE_SINK (element);
4372   GstBaseSinkClass *bclass;
4373   GstBaseSinkPrivate *priv;
4374
4375   priv = basesink->priv;
4376
4377   bclass = GST_BASE_SINK_GET_CLASS (basesink);
4378
4379   switch (transition) {
4380     case GST_STATE_CHANGE_NULL_TO_READY:
4381       if (bclass->start)
4382         if (!bclass->start (basesink))
4383           goto start_failed;
4384       break;
4385     case GST_STATE_CHANGE_READY_TO_PAUSED:
4386       /* need to complete preroll before this state change completes, there
4387        * is no data flow in READY so we can safely assume we need to preroll. */
4388       GST_PAD_PREROLL_LOCK (basesink->sinkpad);
4389       GST_DEBUG_OBJECT (basesink, "READY to PAUSED");
4390       basesink->have_newsegment = FALSE;
4391       gst_segment_init (&basesink->segment, GST_FORMAT_UNDEFINED);
4392       gst_segment_init (basesink->abidata.ABI.clip_segment,
4393           GST_FORMAT_UNDEFINED);
4394       basesink->offset = 0;
4395       basesink->have_preroll = FALSE;
4396       priv->step_unlock = FALSE;
4397       basesink->need_preroll = TRUE;
4398       basesink->playing_async = TRUE;
4399       priv->current_sstart = -1;
4400       priv->current_sstop = -1;
4401       priv->eos_rtime = -1;
4402       priv->latency = 0;
4403       basesink->eos = FALSE;
4404       priv->received_eos = FALSE;
4405       gst_base_sink_reset_qos (basesink);
4406       priv->commited = FALSE;
4407       priv->call_preroll = TRUE;
4408       priv->current_step.valid = FALSE;
4409       priv->pending_step.valid = FALSE;
4410       if (priv->async_enabled) {
4411         GST_DEBUG_OBJECT (basesink, "doing async state change");
4412         /* when async enabled, post async-start message and return ASYNC from
4413          * the state change function */
4414         ret = GST_STATE_CHANGE_ASYNC;
4415         gst_element_post_message (GST_ELEMENT_CAST (basesink),
4416             gst_message_new_async_start (GST_OBJECT_CAST (basesink), FALSE));
4417       } else {
4418         priv->have_latency = TRUE;
4419       }
4420       GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);
4421       break;
4422     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
4423       GST_PAD_PREROLL_LOCK (basesink->sinkpad);
4424       if (!gst_base_sink_needs_preroll (basesink)) {
4425         GST_DEBUG_OBJECT (basesink, "PAUSED to PLAYING, don't need preroll");
4426         /* no preroll needed anymore now. */
4427         basesink->playing_async = FALSE;
4428         basesink->need_preroll = FALSE;
4429         if (basesink->eos) {
4430           GstMessage *message;
4431
4432           /* need to post EOS message here */
4433           GST_DEBUG_OBJECT (basesink, "Now posting EOS");
4434           message = gst_message_new_eos (GST_OBJECT_CAST (basesink));
4435           gst_message_set_seqnum (message, basesink->priv->seqnum);
4436           gst_element_post_message (GST_ELEMENT_CAST (basesink), message);
4437         } else {
4438           GST_DEBUG_OBJECT (basesink, "signal preroll");
4439           GST_PAD_PREROLL_SIGNAL (basesink->sinkpad);
4440         }
4441       } else {
4442         GST_DEBUG_OBJECT (basesink, "PAUSED to PLAYING, we are not prerolled");
4443         basesink->need_preroll = TRUE;
4444         basesink->playing_async = TRUE;
4445         priv->call_preroll = TRUE;
4446         priv->commited = FALSE;
4447         if (priv->async_enabled) {
4448           GST_DEBUG_OBJECT (basesink, "doing async state change");
4449           ret = GST_STATE_CHANGE_ASYNC;
4450           gst_element_post_message (GST_ELEMENT_CAST (basesink),
4451               gst_message_new_async_start (GST_OBJECT_CAST (basesink), FALSE));
4452         }
4453       }
4454       GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);
4455       break;
4456     default:
4457       break;
4458   }
4459
4460   {
4461     GstStateChangeReturn bret;
4462
4463     bret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
4464     if (G_UNLIKELY (bret == GST_STATE_CHANGE_FAILURE))
4465       goto activate_failed;
4466   }
4467
4468   switch (transition) {
4469     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
4470       GST_DEBUG_OBJECT (basesink, "PLAYING to PAUSED");
4471       /* FIXME, make sure we cannot enter _render first */
4472
4473       /* we need to call ::unlock before locking PREROLL_LOCK
4474        * since we lock it before going into ::render */
4475       if (bclass->unlock)
4476         bclass->unlock (basesink);
4477
4478       GST_PAD_PREROLL_LOCK (basesink->sinkpad);
4479       /* now that we have the PREROLL lock, clear our unlock request */
4480       if (bclass->unlock_stop)
4481         bclass->unlock_stop (basesink);
4482
4483       /* we need preroll again and we set the flag before unlocking the clockid
4484        * because if the clockid is unlocked before a current buffer expired, we
4485        * can use that buffer to preroll with */
4486       basesink->need_preroll = TRUE;
4487
4488       if (basesink->clock_id) {
4489         gst_clock_id_unschedule (basesink->clock_id);
4490       }
4491
4492       /* if we don't have a preroll buffer we need to wait for a preroll and
4493        * return ASYNC. */
4494       if (!gst_base_sink_needs_preroll (basesink)) {
4495         GST_DEBUG_OBJECT (basesink, "PLAYING to PAUSED, we are prerolled");
4496         basesink->playing_async = FALSE;
4497       } else {
4498         if (GST_STATE_TARGET (GST_ELEMENT (basesink)) <= GST_STATE_READY) {
4499           ret = GST_STATE_CHANGE_SUCCESS;
4500         } else {
4501           GST_DEBUG_OBJECT (basesink,
4502               "PLAYING to PAUSED, we are not prerolled");
4503           basesink->playing_async = TRUE;
4504           priv->commited = FALSE;
4505           priv->call_preroll = TRUE;
4506           if (priv->async_enabled) {
4507             GST_DEBUG_OBJECT (basesink, "doing async state change");
4508             ret = GST_STATE_CHANGE_ASYNC;
4509             gst_element_post_message (GST_ELEMENT_CAST (basesink),
4510                 gst_message_new_async_start (GST_OBJECT_CAST (basesink),
4511                     FALSE));
4512           }
4513         }
4514       }
4515       GST_DEBUG_OBJECT (basesink, "rendered: %" G_GUINT64_FORMAT
4516           ", dropped: %" G_GUINT64_FORMAT, priv->rendered, priv->dropped);
4517
4518       gst_base_sink_reset_qos (basesink);
4519       GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);
4520       break;
4521     case GST_STATE_CHANGE_PAUSED_TO_READY:
4522       GST_PAD_PREROLL_LOCK (basesink->sinkpad);
4523       /* start by reseting our position state with the object lock so that the
4524        * position query gets the right idea. We do this before we post the
4525        * messages so that the message handlers pick this up. */
4526       GST_OBJECT_LOCK (basesink);
4527       basesink->have_newsegment = FALSE;
4528       priv->current_sstart = -1;
4529       priv->current_sstop = -1;
4530       priv->have_latency = FALSE;
4531       GST_OBJECT_UNLOCK (basesink);
4532
4533       gst_base_sink_set_last_buffer (basesink, NULL);
4534       priv->call_preroll = FALSE;
4535
4536       if (!priv->commited) {
4537         if (priv->async_enabled) {
4538           GST_DEBUG_OBJECT (basesink, "PAUSED to READY, posting async-done");
4539
4540           gst_element_post_message (GST_ELEMENT_CAST (basesink),
4541               gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
4542                   GST_STATE_PLAYING, GST_STATE_PAUSED, GST_STATE_READY));
4543
4544           gst_element_post_message (GST_ELEMENT_CAST (basesink),
4545               gst_message_new_async_done (GST_OBJECT_CAST (basesink)));
4546         }
4547         priv->commited = TRUE;
4548       } else {
4549         GST_DEBUG_OBJECT (basesink, "PAUSED to READY, don't need_preroll");
4550       }
4551       GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);
4552       break;
4553     case GST_STATE_CHANGE_READY_TO_NULL:
4554       if (bclass->stop) {
4555         if (!bclass->stop (basesink)) {
4556           GST_WARNING_OBJECT (basesink, "failed to stop");
4557         }
4558       }
4559       gst_base_sink_set_last_buffer (basesink, NULL);
4560       priv->call_preroll = FALSE;
4561       break;
4562     default:
4563       break;
4564   }
4565
4566   return ret;
4567
4568   /* ERRORS */
4569 start_failed:
4570   {
4571     GST_DEBUG_OBJECT (basesink, "failed to start");
4572     return GST_STATE_CHANGE_FAILURE;
4573   }
4574 activate_failed:
4575   {
4576     GST_DEBUG_OBJECT (basesink,
4577         "element failed to change states -- activation problem?");
4578     return GST_STATE_CHANGE_FAILURE;
4579   }
4580 }