libs/gst/base/gstbasesrc.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *               2000,2005 Wim Taymans <wim@fluendo.com>
   4  *
   5  * gstbasesrc.c:
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 /**
  24  * SECTION:gstbasesrc
  25  * @title: GstBaseSrc
  26  * @short_description: Base class for getrange based source elements
  27  * @see_also: #GstPushSrc, #GstBaseTransform, #GstBaseSink
  28  *
  29  * This is a generic base class for source elements. The following
  30  * types of sources are supported:
  31  *
  32  *   * random access sources like files
  33  *   * seekable sources
  34  *   * live sources
  35  *
  36  * The source can be configured to operate in any #GstFormat with the
  37  * gst_base_src_set_format() method. The currently set format determines
  38  * the format of the internal #GstSegment and any %GST_EVENT_SEGMENT
  39  * events. The default format for #GstBaseSrc is %GST_FORMAT_BYTES.
  40  *
  41  * #GstBaseSrc always supports push mode scheduling. If the following
  42  * conditions are met, it also supports pull mode scheduling:
  43  *
  44  *   * The format is set to %GST_FORMAT_BYTES (default).
  45  *   * #GstBaseSrcClass.is_seekable() returns %TRUE.
  46  *
  47  * If all the conditions are met for operating in pull mode, #GstBaseSrc is
  48  * automatically seekable in push mode as well. The following conditions must
  49  * be met to make the element seekable in push mode when the format is not
  50  * %GST_FORMAT_BYTES:
  51  *
  52  * * #GstBaseSrcClass.is_seekable() returns %TRUE.
  53  * * #GstBaseSrcClass.query() can convert all supported seek formats to the
  54  *   internal format as set with gst_base_src_set_format().
  55  * * #GstBaseSrcClass.do_seek() is implemented, performs the seek and returns
  56  *    %TRUE.
  57  *
  58  * When the element does not meet the requirements to operate in pull mode, the
  59  * offset and length in the #GstBaseSrcClass.create() method should be ignored.
  60  * It is recommended to subclass #GstPushSrc instead, in this situation. If the
  61  * element can operate in pull mode but only with specific offsets and
  62  * lengths, it is allowed to generate an error when the wrong values are passed
  63  * to the #GstBaseSrcClass.create() function.
  64  *
  65  * #GstBaseSrc has support for live sources. Live sources are sources that when
  66  * paused discard data, such as audio or video capture devices. A typical live
  67  * source also produces data at a fixed rate and thus provides a clock to publish
  68  * this rate.
  69  * Use gst_base_src_set_live() to activate the live source mode.
  70  *
  71  * A live source does not produce data in the PAUSED state. This means that the
  72  * #GstBaseSrcClass.create() method will not be called in PAUSED but only in
  73  * PLAYING. To signal the pipeline that the element will not produce data, the
  74  * return value from the READY to PAUSED state will be
  75  * %GST_STATE_CHANGE_NO_PREROLL.
  76  *
  77  * A typical live source will timestamp the buffers it creates with the
  78  * current running time of the pipeline. This is one reason why a live source
  79  * can only produce data in the PLAYING state, when the clock is actually
  80  * distributed and running.
  81  *
  82  * Live sources that synchronize and block on the clock (an audio source, for
  83  * example) can use gst_base_src_wait_playing() when the
  84  * #GstBaseSrcClass.create() function was interrupted by a state change to
  85  * PAUSED.
  86  *
  87  * The #GstBaseSrcClass.get_times() method can be used to implement pseudo-live
  88  * sources. It only makes sense to implement the #GstBaseSrcClass.get_times()
  89  * function if the source is a live source. The #GstBaseSrcClass.get_times()
  90  * function should return timestamps starting from 0, as if it were a non-live
  91  * source. The base class will make sure that the timestamps are transformed
  92  * into the current running_time. The base source will then wait for the
  93  * calculated running_time before pushing out the buffer.
  94  *
  95  * For live sources, the base class will by default report a latency of 0.
  96  * For pseudo live sources, the base class will by default measure the difference
  97  * between the first buffer timestamp and the start time of get_times and will
  98  * report this value as the latency.
  99  * Subclasses should override the query function when this behaviour is not
 100  * acceptable.
 101  *
 102  * There is only support in #GstBaseSrc for exactly one source pad, which
 103  * should be named "src". A source implementation (subclass of #GstBaseSrc)
 104  * should install a pad template in its class_init function, like so:
 105  * |[<!-- language="C" -->
 106  * static void
 107  * my_element_class_init (GstMyElementClass *klass)
 108  * {
 109  *   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 110  *   // srctemplate should be a #GstStaticPadTemplate with direction
 111  *   // %GST_PAD_SRC and name "src"
 112  *   gst_element_class_add_static_pad_template (gstelement_class, &amp;srctemplate);
 113  *
 114  *   gst_element_class_set_static_metadata (gstelement_class,
 115  *      "Source name",
 116  *      "Source",
 117  *      "My Source element",
 118  *      "The author <my.sink@my.email>");
 119  * }
 120  * ]|
 121  *
 122  * ## Controlled shutdown of live sources in applications
 123  *
 124  * Applications that record from a live source may want to stop recording
 125  * in a controlled way, so that the recording is stopped, but the data
 126  * already in the pipeline is processed to the end (remember that many live
 127  * sources would go on recording forever otherwise). For that to happen the
 128  * application needs to make the source stop recording and send an EOS
 129  * event down the pipeline. The application would then wait for an
 130  * EOS message posted on the pipeline's bus to know when all data has
 131  * been processed and the pipeline can safely be stopped.
 132  *
 133  * An application may send an EOS event to a source element to make it
 134  * perform the EOS logic (send EOS event downstream or post a
 135  * %GST_MESSAGE_SEGMENT_DONE on the bus). This can typically be done
 136  * with the gst_element_send_event() function on the element or its parent bin.
 137  *
 138  * After the EOS has been sent to the element, the application should wait for
 139  * an EOS message to be posted on the pipeline's bus. Once this EOS message is
 140  * received, it may safely shut down the entire pipeline.
 141  *
 142  */
 143
 144 #ifdef HAVE_CONFIG_H
 145 #  include "config.h"
 146 #endif
 147
 148 #include <stdlib.h>
 149 #include <string.h>
 150
 151 #include <gst/gst_private.h>
 152 #include <gst/glib-compat-private.h>
 153
 154 #include "gstbasesrc.h"
 155 #include "gsttypefindhelper.h"
 156 #include <gst/gst-i18n-lib.h>
 157
 158 GST_DEBUG_CATEGORY_STATIC (gst_base_src_debug);
 159 #define GST_CAT_DEFAULT gst_base_src_debug
 160
 161 #define GST_LIVE_GET_LOCK(elem)               (&GST_BASE_SRC_CAST(elem)->live_lock)
 162 #define GST_LIVE_LOCK(elem)                   g_mutex_lock(GST_LIVE_GET_LOCK(elem))
 163 #define GST_LIVE_TRYLOCK(elem)                g_mutex_trylock(GST_LIVE_GET_LOCK(elem))
 164 #define GST_LIVE_UNLOCK(elem)                 g_mutex_unlock(GST_LIVE_GET_LOCK(elem))
 165 #define GST_LIVE_GET_COND(elem)               (&GST_BASE_SRC_CAST(elem)->live_cond)
 166 #define GST_LIVE_WAIT(elem)                   g_cond_wait (GST_LIVE_GET_COND (elem), GST_LIVE_GET_LOCK (elem))
 167 #define GST_LIVE_WAIT_UNTIL(elem, end_time)   g_cond_timed_wait (GST_LIVE_GET_COND (elem), GST_LIVE_GET_LOCK (elem), end_time)
 168 #define GST_LIVE_SIGNAL(elem)                 g_cond_signal (GST_LIVE_GET_COND (elem));
 169 #define GST_LIVE_BROADCAST(elem)              g_cond_broadcast (GST_LIVE_GET_COND (elem));
 170
 171
 172 #define GST_ASYNC_GET_COND(elem)              (&GST_BASE_SRC_CAST(elem)->priv->async_cond)
 173 #define GST_ASYNC_WAIT(elem)                  g_cond_wait (GST_ASYNC_GET_COND (elem), GST_OBJECT_GET_LOCK (elem))
 174 #define GST_ASYNC_SIGNAL(elem)                g_cond_signal (GST_ASYNC_GET_COND (elem));
 175
 176 #define CLEAR_PENDING_EOS(bsrc) \
 177   G_STMT_START { \
 178     g_atomic_int_set (&bsrc->priv->has_pending_eos, FALSE); \
 179     gst_event_replace (&bsrc->priv->pending_eos, NULL); \
 180   } G_STMT_END
 181
 182
 183 /* BaseSrc signals and args */
 184 enum
 185 {
 186   /* FILL ME */
 187   LAST_SIGNAL
 188 };
 189
 190 #define DEFAULT_BLOCKSIZE       4096
 191 #define DEFAULT_NUM_BUFFERS     -1
 192 #define DEFAULT_TYPEFIND        FALSE
 193 #define DEFAULT_DO_TIMESTAMP    FALSE
 194
 195 enum
 196 {
 197   PROP_0,
 198   PROP_BLOCKSIZE,
 199   PROP_NUM_BUFFERS,
 200   PROP_TYPEFIND,
 201   PROP_DO_TIMESTAMP
 202 };
 203
 204 #define GST_BASE_SRC_GET_PRIVATE(obj)  \
 205    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BASE_SRC, GstBaseSrcPrivate))
 206
 207 struct _GstBaseSrcPrivate
 208 {
 209   gboolean discont;
 210   gboolean flushing;
 211
 212   GstFlowReturn start_result;
 213   gboolean async;
 214
 215   /* if a stream-start event should be sent */
 216   gboolean stream_start_pending;
 217
 218   /* if segment should be sent and a
 219    * seqnum if it was originated by a seek */
 220   gboolean segment_pending;
 221   guint32 segment_seqnum;
 222
 223   /* if EOS is pending (atomic) */
 224   GstEvent *pending_eos;
 225   gint has_pending_eos;
 226
 227   /* if the eos was caused by a forced eos from the application */
 228   gboolean forced_eos;
 229
 230   /* startup latency is the time it takes between going to PLAYING and producing
 231    * the first BUFFER with running_time 0. This value is included in the latency
 232    * reporting. */
 233   GstClockTime latency;
 234   /* timestamp offset, this is the offset add to the values of gst_times for
 235    * pseudo live sources */
 236   GstClockTimeDiff ts_offset;
 237
 238   gboolean do_timestamp;
 239   volatile gint dynamic_size;
 240   volatile gint automatic_eos;
 241
 242   /* stream sequence number */
 243   guint32 seqnum;
 244
 245   /* pending events (TAG, CUSTOM_BOTH, CUSTOM_DOWNSTREAM) to be
 246    * pushed in the data stream */
 247   GList *pending_events;
 248   volatile gint have_events;
 249
 250   /* QoS *//* with LOCK */
 251   gboolean qos_enabled;
 252   gdouble proportion;
 253   GstClockTime earliest_time;
 254
 255   GstBufferPool *pool;
 256   GstAllocator *allocator;
 257   GstAllocationParams params;
 258
 259   GCond async_cond;
 260 };
 261
 262 static GstElementClass *parent_class = NULL;
 263
 264 static void gst_base_src_class_init (GstBaseSrcClass * klass);
 265 static void gst_base_src_init (GstBaseSrc * src, gpointer g_class);
 266 static void gst_base_src_finalize (GObject * object);
 267
 268
 269 GType
 270 gst_base_src_get_type (void)
 271 {
 272   static volatile gsize base_src_type = 0;
 273
 274   if (g_once_init_enter (&base_src_type)) {
 275     GType _type;
 276     static const GTypeInfo base_src_info = {
 277       sizeof (GstBaseSrcClass),
 278       NULL,
 279       NULL,
 280       (GClassInitFunc) gst_base_src_class_init,
 281       NULL,
 282       NULL,
 283       sizeof (GstBaseSrc),
 284       0,
 285       (GInstanceInitFunc) gst_base_src_init,
 286     };
 287
 288     _type = g_type_register_static (GST_TYPE_ELEMENT,
 289         "GstBaseSrc", &base_src_info, G_TYPE_FLAG_ABSTRACT);
 290     g_once_init_leave (&base_src_type, _type);
 291   }
 292   return base_src_type;
 293 }
 294
 295 static GstCaps *gst_base_src_default_get_caps (GstBaseSrc * bsrc,
 296     GstCaps * filter);
 297 static GstCaps *gst_base_src_default_fixate (GstBaseSrc * src, GstCaps * caps);
 298 static GstCaps *gst_base_src_fixate (GstBaseSrc * src, GstCaps * caps);
 299
 300 static gboolean gst_base_src_is_random_access (GstBaseSrc * src);
 301 static gboolean gst_base_src_activate_mode (GstPad * pad, GstObject * parent,
 302     GstPadMode mode, gboolean active);
 303 static void gst_base_src_set_property (GObject * object, guint prop_id,
 304     const GValue * value, GParamSpec * pspec);
 305 static void gst_base_src_get_property (GObject * object, guint prop_id,
 306     GValue * value, GParamSpec * pspec);
 307 static gboolean gst_base_src_event (GstPad * pad, GstObject * parent,
 308     GstEvent * event);
 309 static gboolean gst_base_src_send_event (GstElement * elem, GstEvent * event);
 310 static gboolean gst_base_src_default_event (GstBaseSrc * src, GstEvent * event);
 311
 312 static gboolean gst_base_src_query (GstPad * pad, GstObject * parent,
 313     GstQuery * query);
 314
 315 static gboolean gst_base_src_activate_pool (GstBaseSrc * basesrc,
 316     gboolean active);
 317 static gboolean gst_base_src_default_negotiate (GstBaseSrc * basesrc);
 318 static gboolean gst_base_src_default_do_seek (GstBaseSrc * src,
 319     GstSegment * segment);
 320 static gboolean gst_base_src_default_query (GstBaseSrc * src, GstQuery * query);
 321 static gboolean gst_base_src_default_prepare_seek_segment (GstBaseSrc * src,
 322     GstEvent * event, GstSegment * segment);
 323 static GstFlowReturn gst_base_src_default_create (GstBaseSrc * basesrc,
 324     guint64 offset, guint size, GstBuffer ** buf);
 325 static GstFlowReturn gst_base_src_default_alloc (GstBaseSrc * basesrc,
 326     guint64 offset, guint size, GstBuffer ** buf);
 327 static gboolean gst_base_src_decide_allocation_default (GstBaseSrc * basesrc,
 328     GstQuery * query);
 329
 330 static gboolean gst_base_src_set_flushing (GstBaseSrc * basesrc,
 331     gboolean flushing, gboolean live_play, gboolean * playing);
 332
 333 static gboolean gst_base_src_start (GstBaseSrc * basesrc);
 334 static gboolean gst_base_src_stop (GstBaseSrc * basesrc);
 335
 336 static GstStateChangeReturn gst_base_src_change_state (GstElement * element,
 337     GstStateChange transition);
 338
 339 static void gst_base_src_loop (GstPad * pad);
 340 static GstFlowReturn gst_base_src_getrange (GstPad * pad, GstObject * parent,
 341     guint64 offset, guint length, GstBuffer ** buf);
 342 static GstFlowReturn gst_base_src_get_range (GstBaseSrc * src, guint64 offset,
 343     guint length, GstBuffer ** buf);
 344 static gboolean gst_base_src_seekable (GstBaseSrc * src);
 345 static gboolean gst_base_src_negotiate (GstBaseSrc * basesrc);
 346 static gboolean gst_base_src_update_length (GstBaseSrc * src, guint64 offset,
 347     guint * length, gboolean force);
 348
 349 static void
 350 gst_base_src_class_init (GstBaseSrcClass * klass)
 351 {
 352   GObjectClass *gobject_class;
 353   GstElementClass *gstelement_class;
 354
 355   gobject_class = G_OBJECT_CLASS (klass);
 356   gstelement_class = GST_ELEMENT_CLASS (klass);
 357
 358   GST_DEBUG_CATEGORY_INIT (gst_base_src_debug, "basesrc", 0, "basesrc element");
 359
 360   g_type_class_add_private (klass, sizeof (GstBaseSrcPrivate));
 361
 362   parent_class = g_type_class_peek_parent (klass);
 363
 364   gobject_class->finalize = gst_base_src_finalize;
 365   gobject_class->set_property = gst_base_src_set_property;
 366   gobject_class->get_property = gst_base_src_get_property;
 367
 368   g_object_class_install_property (gobject_class, PROP_BLOCKSIZE,
 369       g_param_spec_uint ("blocksize", "Block size",
 370           "Size in bytes to read per buffer (-1 = default)", 0, G_MAXUINT,
 371           DEFAULT_BLOCKSIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 372   g_object_class_install_property (gobject_class, PROP_NUM_BUFFERS,
 373       g_param_spec_int ("num-buffers", "num-buffers",
 374           "Number of buffers to output before sending EOS (-1 = unlimited)",
 375           -1, G_MAXINT, DEFAULT_NUM_BUFFERS, G_PARAM_READWRITE |
 376           G_PARAM_STATIC_STRINGS));
 377   g_object_class_install_property (gobject_class, PROP_TYPEFIND,
 378       g_param_spec_boolean ("typefind", "Typefind",
 379           "Run typefind before negotiating", DEFAULT_TYPEFIND,
 380           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 381   g_object_class_install_property (gobject_class, PROP_DO_TIMESTAMP,
 382       g_param_spec_boolean ("do-timestamp", "Do timestamp",
 383           "Apply current stream time to buffers", DEFAULT_DO_TIMESTAMP,
 384           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 385
 386   gstelement_class->change_state =
 387       GST_DEBUG_FUNCPTR (gst_base_src_change_state);
 388   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_base_src_send_event);
 389
 390   klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_src_default_get_caps);
 391   klass->negotiate = GST_DEBUG_FUNCPTR (gst_base_src_default_negotiate);
 392   klass->fixate = GST_DEBUG_FUNCPTR (gst_base_src_default_fixate);
 393   klass->prepare_seek_segment =
 394       GST_DEBUG_FUNCPTR (gst_base_src_default_prepare_seek_segment);
 395   klass->do_seek = GST_DEBUG_FUNCPTR (gst_base_src_default_do_seek);
 396   klass->query = GST_DEBUG_FUNCPTR (gst_base_src_default_query);
 397   klass->event = GST_DEBUG_FUNCPTR (gst_base_src_default_event);
 398   klass->create = GST_DEBUG_FUNCPTR (gst_base_src_default_create);
 399   klass->alloc = GST_DEBUG_FUNCPTR (gst_base_src_default_alloc);
 400   klass->decide_allocation =
 401       GST_DEBUG_FUNCPTR (gst_base_src_decide_allocation_default);
 402
 403   /* Registering debug symbols for function pointers */
 404   GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_activate_mode);
 405   GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_event);
 406   GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_query);
 407   GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_getrange);
 408   GST_DEBUG_REGISTER_FUNCPTR (gst_base_src_fixate);
 409 }
 410
 411 static void
 412 gst_base_src_init (GstBaseSrc * basesrc, gpointer g_class)
 413 {
 414   GstPad *pad;
 415   GstPadTemplate *pad_template;
 416
 417   basesrc->priv = GST_BASE_SRC_GET_PRIVATE (basesrc);
 418
 419   basesrc->is_live = FALSE;
 420   g_mutex_init (&basesrc->live_lock);
 421   g_cond_init (&basesrc->live_cond);
 422   basesrc->num_buffers = DEFAULT_NUM_BUFFERS;
 423   basesrc->num_buffers_left = -1;
 424   basesrc->priv->automatic_eos = TRUE;
 425
 426   basesrc->can_activate_push = TRUE;
 427
 428   pad_template =
 429       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "src");
 430   g_return_if_fail (pad_template != NULL);
 431
 432   GST_DEBUG_OBJECT (basesrc, "creating src pad");
 433   pad = gst_pad_new_from_template (pad_template, "src");
 434
 435   GST_DEBUG_OBJECT (basesrc, "setting functions on src pad");
 436   gst_pad_set_activatemode_function (pad, gst_base_src_activate_mode);
 437   gst_pad_set_event_function (pad, gst_base_src_event);
 438   gst_pad_set_query_function (pad, gst_base_src_query);
 439   gst_pad_set_getrange_function (pad, gst_base_src_getrange);
 440
 441   /* hold pointer to pad */
 442   basesrc->srcpad = pad;
 443   GST_DEBUG_OBJECT (basesrc, "adding src pad");
 444   gst_element_add_pad (GST_ELEMENT (basesrc), pad);
 445
 446   basesrc->blocksize = DEFAULT_BLOCKSIZE;
 447   basesrc->clock_id = NULL;
 448   /* we operate in BYTES by default */
 449   gst_base_src_set_format (basesrc, GST_FORMAT_BYTES);
 450   basesrc->typefind = DEFAULT_TYPEFIND;
 451   basesrc->priv->do_timestamp = DEFAULT_DO_TIMESTAMP;
 452   g_atomic_int_set (&basesrc->priv->have_events, FALSE);
 453
 454   g_cond_init (&basesrc->priv->async_cond);
 455   basesrc->priv->start_result = GST_FLOW_FLUSHING;
 456   GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTED);
 457   GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTING);
 458   GST_OBJECT_FLAG_SET (basesrc, GST_ELEMENT_FLAG_SOURCE);
 459
 460   GST_DEBUG_OBJECT (basesrc, "init done");
 461 }
 462
 463 static void
 464 gst_base_src_finalize (GObject * object)
 465 {
 466   GstBaseSrc *basesrc;
 467   GstEvent **event_p;
 468
 469   basesrc = GST_BASE_SRC (object);
 470
 471   g_mutex_clear (&basesrc->live_lock);
 472   g_cond_clear (&basesrc->live_cond);
 473   g_cond_clear (&basesrc->priv->async_cond);
 474
 475   event_p = &basesrc->pending_seek;
 476   gst_event_replace (event_p, NULL);
 477
 478   if (basesrc->priv->pending_events) {
 479     g_list_foreach (basesrc->priv->pending_events, (GFunc) gst_event_unref,
 480         NULL);
 481     g_list_free (basesrc->priv->pending_events);
 482   }
 483
 484   G_OBJECT_CLASS (parent_class)->finalize (object);
 485 }
 486
 487 /**
 488  * gst_base_src_wait_playing:
 489  * @src: the src
 490  *
 491  * If the #GstBaseSrcClass.create() method performs its own synchronisation
 492  * against the clock it must unblock when going from PLAYING to the PAUSED state
 493  * and call this method before continuing to produce the remaining data.
 494  *
 495  * This function will block until a state change to PLAYING happens (in which
 496  * case this function returns %GST_FLOW_OK) or the processing must be stopped due
 497  * to a state change to READY or a FLUSH event (in which case this function
 498  * returns %GST_FLOW_FLUSHING).
 499  *
 500  * Returns: %GST_FLOW_OK if @src is PLAYING and processing can
 501  * continue. Any other return value should be returned from the create vmethod.
 502  */
 503 GstFlowReturn
 504 gst_base_src_wait_playing (GstBaseSrc * src)
 505 {
 506   g_return_val_if_fail (GST_IS_BASE_SRC (src), GST_FLOW_ERROR);
 507
 508   do {
 509     /* block until the state changes, or we get a flush, or something */
 510     GST_DEBUG_OBJECT (src, "live source waiting for running state");
 511     GST_LIVE_WAIT (src);
 512     GST_DEBUG_OBJECT (src, "live source unlocked");
 513     if (src->priv->flushing)
 514       goto flushing;
 515   } while (G_UNLIKELY (!src->live_running));
 516
 517   return GST_FLOW_OK;
 518
 519   /* ERRORS */
 520 flushing:
 521   {
 522     GST_DEBUG_OBJECT (src, "we are flushing");
 523     return GST_FLOW_FLUSHING;
 524   }
 525 }
 526
 527 /**
 528  * gst_base_src_set_live:
 529  * @src: base source instance
 530  * @live: new live-mode
 531  *
 532  * If the element listens to a live source, @live should
 533  * be set to %TRUE.
 534  *
 535  * A live source will not produce data in the PAUSED state and
 536  * will therefore not be able to participate in the PREROLL phase
 537  * of a pipeline. To signal this fact to the application and the
 538  * pipeline, the state change return value of the live source will
 539  * be GST_STATE_CHANGE_NO_PREROLL.
 540  */
 541 void
 542 gst_base_src_set_live (GstBaseSrc * src, gboolean live)
 543 {
 544   g_return_if_fail (GST_IS_BASE_SRC (src));
 545
 546   GST_OBJECT_LOCK (src);
 547   src->is_live = live;
 548   GST_OBJECT_UNLOCK (src);
 549 }
 550
 551 /**
 552  * gst_base_src_is_live:
 553  * @src: base source instance
 554  *
 555  * Check if an element is in live mode.
 556  *
 557  * Returns: %TRUE if element is in live mode.
 558  */
 559 gboolean
 560 gst_base_src_is_live (GstBaseSrc * src)
 561 {
 562   gboolean result;
 563
 564   g_return_val_if_fail (GST_IS_BASE_SRC (src), FALSE);
 565
 566   GST_OBJECT_LOCK (src);
 567   result = src->is_live;
 568   GST_OBJECT_UNLOCK (src);
 569
 570   return result;
 571 }
 572
 573 /**
 574  * gst_base_src_set_format:
 575  * @src: base source instance
 576  * @format: the format to use
 577  *
 578  * Sets the default format of the source. This will be the format used
 579  * for sending SEGMENT events and for performing seeks.
 580  *
 581  * If a format of GST_FORMAT_BYTES is set, the element will be able to
 582  * operate in pull mode if the #GstBaseSrcClass.is_seekable() returns %TRUE.
 583  *
 584  * This function must only be called in states < %GST_STATE_PAUSED.
 585  */
 586 void
 587 gst_base_src_set_format (GstBaseSrc * src, GstFormat format)
 588 {
 589   g_return_if_fail (GST_IS_BASE_SRC (src));
 590   g_return_if_fail (GST_STATE (src) <= GST_STATE_READY);
 591
 592   GST_OBJECT_LOCK (src);
 593   gst_segment_init (&src->segment, format);
 594   GST_OBJECT_UNLOCK (src);
 595 }
 596
 597 /**
 598  * gst_base_src_set_dynamic_size:
 599  * @src: base source instance
 600  * @dynamic: new dynamic size mode
 601  *
 602  * If not @dynamic, size is only updated when needed, such as when trying to
 603  * read past current tracked size.  Otherwise, size is checked for upon each
 604  * read.
 605  */
 606 void
 607 gst_base_src_set_dynamic_size (GstBaseSrc * src, gboolean dynamic)
 608 {
 609   g_return_if_fail (GST_IS_BASE_SRC (src));
 610
 611   g_atomic_int_set (&src->priv->dynamic_size, dynamic);
 612 }
 613
 614 /**
 615  * gst_base_src_set_automatic_eos:
 616  * @src: base source instance
 617  * @automatic_eos: automatic eos
 618  *
 619  * If @automatic_eos is %TRUE, @src will automatically go EOS if a buffer
 620  * after the total size is returned. By default this is %TRUE but sources
 621  * that can't return an authoritative size and only know that they're EOS
 622  * when trying to read more should set this to %FALSE.
 623  *
 624  * Since: 1.4
 625  */
 626 void
 627 gst_base_src_set_automatic_eos (GstBaseSrc * src, gboolean automatic_eos)
 628 {
 629   g_return_if_fail (GST_IS_BASE_SRC (src));
 630
 631   g_atomic_int_set (&src->priv->automatic_eos, automatic_eos);
 632 }
 633
 634 /**
 635  * gst_base_src_set_async:
 636  * @src: base source instance
 637  * @async: new async mode
 638  *
 639  * Configure async behaviour in @src, no state change will block. The open,
 640  * close, start, stop, play and pause virtual methods will be executed in a
 641  * different thread and are thus allowed to perform blocking operations. Any
 642  * blocking operation should be unblocked with the unlock vmethod.
 643  */
 644 void
 645 gst_base_src_set_async (GstBaseSrc * src, gboolean async)
 646 {
 647   g_return_if_fail (GST_IS_BASE_SRC (src));
 648
 649   GST_OBJECT_LOCK (src);
 650   src->priv->async = async;
 651   GST_OBJECT_UNLOCK (src);
 652 }
 653
 654 /**
 655  * gst_base_src_is_async:
 656  * @src: base source instance
 657  *
 658  * Get the current async behaviour of @src. See also gst_base_src_set_async().
 659  *
 660  * Returns: %TRUE if @src is operating in async mode.
 661  */
 662 gboolean
 663 gst_base_src_is_async (GstBaseSrc * src)
 664 {
 665   gboolean res;
 666
 667   g_return_val_if_fail (GST_IS_BASE_SRC (src), FALSE);
 668
 669   GST_OBJECT_LOCK (src);
 670   res = src->priv->async;
 671   GST_OBJECT_UNLOCK (src);
 672
 673   return res;
 674 }
 675
 676
 677 /**
 678  * gst_base_src_query_latency:
 679  * @src: the source
 680  * @live: (out) (allow-none): if the source is live
 681  * @min_latency: (out) (allow-none): the min latency of the source
 682  * @max_latency: (out) (allow-none): the max latency of the source
 683  *
 684  * Query the source for the latency parameters. @live will be %TRUE when @src is
 685  * configured as a live source. @min_latency and @max_latency will be set
 686  * to the difference between the running time and the timestamp of the first
 687  * buffer.
 688  *
 689  * This function is mostly used by subclasses.
 690  *
 691  * Returns: %TRUE if the query succeeded.
 692  */
 693 gboolean
 694 gst_base_src_query_latency (GstBaseSrc * src, gboolean * live,
 695     GstClockTime * min_latency, GstClockTime * max_latency)
 696 {
 697   GstClockTime min;
 698
 699   g_return_val_if_fail (GST_IS_BASE_SRC (src), FALSE);
 700
 701   GST_OBJECT_LOCK (src);
 702   if (live)
 703     *live = src->is_live;
 704
 705   /* if we have a startup latency, report this one, else report 0. Subclasses
 706    * are supposed to override the query function if they want something
 707    * else. */
 708   if (src->priv->latency != -1)
 709     min = src->priv->latency;
 710   else
 711     min = 0;
 712
 713   if (min_latency)
 714     *min_latency = min;
 715   if (max_latency)
 716     *max_latency = min;
 717
 718   GST_LOG_OBJECT (src, "latency: live %d, min %" GST_TIME_FORMAT
 719       ", max %" GST_TIME_FORMAT, src->is_live, GST_TIME_ARGS (min),
 720       GST_TIME_ARGS (min));
 721   GST_OBJECT_UNLOCK (src);
 722
 723   return TRUE;
 724 }
 725
 726 /**
 727  * gst_base_src_set_blocksize:
 728  * @src: the source
 729  * @blocksize: the new blocksize in bytes
 730  *
 731  * Set the number of bytes that @src will push out with each buffer. When
 732  * @blocksize is set to -1, a default length will be used.
 733  */
 734 void
 735 gst_base_src_set_blocksize (GstBaseSrc * src, guint blocksize)
 736 {
 737   g_return_if_fail (GST_IS_BASE_SRC (src));
 738
 739   GST_OBJECT_LOCK (src);
 740   src->blocksize = blocksize;
 741   GST_OBJECT_UNLOCK (src);
 742 }
 743
 744 /**
 745  * gst_base_src_get_blocksize:
 746  * @src: the source
 747  *
 748  * Get the number of bytes that @src will push out with each buffer.
 749  *
 750  * Returns: the number of bytes pushed with each buffer.
 751  */
 752 guint
 753 gst_base_src_get_blocksize (GstBaseSrc * src)
 754 {
 755   gint res;
 756
 757   g_return_val_if_fail (GST_IS_BASE_SRC (src), 0);
 758
 759   GST_OBJECT_LOCK (src);
 760   res = src->blocksize;
 761   GST_OBJECT_UNLOCK (src);
 762
 763   return res;
 764 }
 765
 766
 767 /**
 768  * gst_base_src_set_do_timestamp:
 769  * @src: the source
 770  * @timestamp: enable or disable timestamping
 771  *
 772  * Configure @src to automatically timestamp outgoing buffers based on the
 773  * current running_time of the pipeline. This property is mostly useful for live
 774  * sources.
 775  */
 776 void
 777 gst_base_src_set_do_timestamp (GstBaseSrc * src, gboolean timestamp)
 778 {
 779   g_return_if_fail (GST_IS_BASE_SRC (src));
 780
 781   GST_OBJECT_LOCK (src);
 782   src->priv->do_timestamp = timestamp;
 783   if (timestamp && src->segment.format != GST_FORMAT_TIME)
 784     gst_segment_init (&src->segment, GST_FORMAT_TIME);
 785   GST_OBJECT_UNLOCK (src);
 786 }
 787
 788 /**
 789  * gst_base_src_get_do_timestamp:
 790  * @src: the source
 791  *
 792  * Query if @src timestamps outgoing buffers based on the current running_time.
 793  *
 794  * Returns: %TRUE if the base class will automatically timestamp outgoing buffers.
 795  */
 796 gboolean
 797 gst_base_src_get_do_timestamp (GstBaseSrc * src)
 798 {
 799   gboolean res;
 800
 801   g_return_val_if_fail (GST_IS_BASE_SRC (src), FALSE);
 802
 803   GST_OBJECT_LOCK (src);
 804   res = src->priv->do_timestamp;
 805   GST_OBJECT_UNLOCK (src);
 806
 807   return res;
 808 }
 809
 810 /**
 811  * gst_base_src_new_seamless_segment:
 812  * @src: The source
 813  * @start: The new start value for the segment
 814  * @stop: Stop value for the new segment
 815  * @time: The new time value for the start of the new segment
 816  *
 817  * Prepare a new seamless segment for emission downstream. This function must
 818  * only be called by derived sub-classes, and only from the create() function,
 819  * as the stream-lock needs to be held.
 820  *
 821  * The format for the new segment will be the current format of the source, as
 822  * configured with gst_base_src_set_format()
 823  *
 824  * Returns: %TRUE if preparation of the seamless segment succeeded.
 825  */
 826 gboolean
 827 gst_base_src_new_seamless_segment (GstBaseSrc * src, gint64 start, gint64 stop,
 828     gint64 time)
 829 {
 830   gboolean res = TRUE;
 831
 832   GST_OBJECT_LOCK (src);
 833
 834   src->segment.base = gst_segment_to_running_time (&src->segment,
 835       src->segment.format, src->segment.position);
 836   src->segment.position = src->segment.start = start;
 837   src->segment.stop = stop;
 838   src->segment.time = time;
 839
 840   /* Mark pending segment. Will be sent before next data */
 841   src->priv->segment_pending = TRUE;
 842   src->priv->segment_seqnum = gst_util_seqnum_next ();
 843
 844   GST_DEBUG_OBJECT (src,
 845       "Starting new seamless segment. Start %" GST_TIME_FORMAT " stop %"
 846       GST_TIME_FORMAT " time %" GST_TIME_FORMAT " base %" GST_TIME_FORMAT,
 847       GST_TIME_ARGS (start), GST_TIME_ARGS (stop), GST_TIME_ARGS (time),
 848       GST_TIME_ARGS (src->segment.base));
 849
 850   GST_OBJECT_UNLOCK (src);
 851
 852   src->priv->discont = TRUE;
 853   src->running = TRUE;
 854
 855   return res;
 856 }
 857
 858 static gboolean
 859 gst_base_src_send_stream_start (GstBaseSrc * src)
 860 {
 861   gboolean ret = TRUE;
 862
 863   if (src->priv->stream_start_pending) {
 864     gchar *stream_id;
 865     GstEvent *event;
 866
 867     stream_id =
 868         gst_pad_create_stream_id (src->srcpad, GST_ELEMENT_CAST (src), NULL);
 869
 870     GST_DEBUG_OBJECT (src, "Pushing STREAM_START");
 871     event = gst_event_new_stream_start (stream_id);
 872     gst_event_set_group_id (event, gst_util_group_id_next ());
 873
 874     ret = gst_pad_push_event (src->srcpad, event);
 875     src->priv->stream_start_pending = FALSE;
 876     g_free (stream_id);
 877   }
 878
 879   return ret;
 880 }
 881
 882 /**
 883  * gst_base_src_set_caps:
 884  * @src: a #GstBaseSrc
 885  * @caps: (transfer none): a #GstCaps
 886  *
 887  * Set new caps on the basesrc source pad.
 888  *
 889  * Returns: %TRUE if the caps could be set
 890  */
 891 gboolean
 892 gst_base_src_set_caps (GstBaseSrc * src, GstCaps * caps)
 893 {
 894   GstBaseSrcClass *bclass;
 895   gboolean res = TRUE;
 896   GstCaps *current_caps;
 897
 898   bclass = GST_BASE_SRC_GET_CLASS (src);
 899
 900   gst_base_src_send_stream_start (src);
 901
 902   current_caps = gst_pad_get_current_caps (GST_BASE_SRC_PAD (src));
 903   if (current_caps && gst_caps_is_equal (current_caps, caps)) {
 904     GST_DEBUG_OBJECT (src, "New caps equal to old ones: %" GST_PTR_FORMAT,
 905         caps);
 906     res = TRUE;
 907   } else {
 908     if (bclass->set_caps)
 909       res = bclass->set_caps (src, caps);
 910
 911     if (res)
 912       res = gst_pad_push_event (src->srcpad, gst_event_new_caps (caps));
 913   }
 914
 915   if (current_caps)
 916     gst_caps_unref (current_caps);
 917
 918   return res;
 919 }
 920
 921 static GstCaps *
 922 gst_base_src_default_get_caps (GstBaseSrc * bsrc, GstCaps * filter)
 923 {
 924   GstCaps *caps = NULL;
 925   GstPadTemplate *pad_template;
 926   GstBaseSrcClass *bclass;
 927
 928   bclass = GST_BASE_SRC_GET_CLASS (bsrc);
 929
 930   pad_template =
 931       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "src");
 932
 933   if (pad_template != NULL) {
 934     caps = gst_pad_template_get_caps (pad_template);
 935
 936     if (filter) {
 937       GstCaps *intersection;
 938
 939       intersection =
 940           gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
 941       gst_caps_unref (caps);
 942       caps = intersection;
 943     }
 944   }
 945   return caps;
 946 }
 947
 948 static GstCaps *
 949 gst_base_src_default_fixate (GstBaseSrc * bsrc, GstCaps * caps)
 950 {
 951   GST_DEBUG_OBJECT (bsrc, "using default caps fixate function");
 952   return gst_caps_fixate (caps);
 953 }
 954
 955 static GstCaps *
 956 gst_base_src_fixate (GstBaseSrc * bsrc, GstCaps * caps)
 957 {
 958   GstBaseSrcClass *bclass;
 959
 960   bclass = GST_BASE_SRC_GET_CLASS (bsrc);
 961
 962   if (bclass->fixate)
 963     caps = bclass->fixate (bsrc, caps);
 964
 965   return caps;
 966 }
 967
 968 static gboolean
 969 gst_base_src_default_query (GstBaseSrc * src, GstQuery * query)
 970 {
 971   gboolean res;
 972
 973   switch (GST_QUERY_TYPE (query)) {
 974     case GST_QUERY_POSITION:
 975     {
 976       GstFormat format;
 977
 978       gst_query_parse_position (query, &format, NULL);
 979
 980       GST_DEBUG_OBJECT (src, "position query in format %s",
 981           gst_format_get_name (format));
 982
 983       switch (format) {
 984         case GST_FORMAT_PERCENT:
 985         {
 986           gint64 percent;
 987           gint64 position;
 988           gint64 duration;
 989
 990           GST_OBJECT_LOCK (src);
 991           position = src->segment.position;
 992           duration = src->segment.duration;
 993           GST_OBJECT_UNLOCK (src);
 994
 995           if (position != -1 && duration != -1) {
 996             if (position < duration)
 997               percent = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX, position,
 998                   duration);
 999             else
1000               percent = GST_FORMAT_PERCENT_MAX;
1001           } else
1002             percent = -1;
1003
1004           gst_query_set_position (query, GST_FORMAT_PERCENT, percent);
1005           res = TRUE;
1006           break;
1007         }
1008         default:
1009         {
1010           gint64 position;
1011           GstFormat seg_format;
1012
1013           GST_OBJECT_LOCK (src);
1014           position =
1015               gst_segment_to_stream_time (&src->segment, src->segment.format,
1016               src->segment.position);
1017           seg_format = src->segment.format;
1018           GST_OBJECT_UNLOCK (src);
1019
1020           if (position != -1) {
1021             /* convert to requested format */
1022             res =
1023                 gst_pad_query_convert (src->srcpad, seg_format,
1024                 position, format, &position);
1025           } else
1026             res = TRUE;
1027
1028           if (res)
1029             gst_query_set_position (query, format, position);
1030
1031           break;
1032         }
1033       }
1034       break;
1035     }
1036     case GST_QUERY_DURATION:
1037     {
1038       GstFormat format;
1039
1040       gst_query_parse_duration (query, &format, NULL);
1041
1042       GST_DEBUG_OBJECT (src, "duration query in format %s",
1043           gst_format_get_name (format));
1044
1045       switch (format) {
1046         case GST_FORMAT_PERCENT:
1047           gst_query_set_duration (query, GST_FORMAT_PERCENT,
1048               GST_FORMAT_PERCENT_MAX);
1049           res = TRUE;
1050           break;
1051         default:
1052         {
1053           gint64 duration;
1054           GstFormat seg_format;
1055           guint length = 0;
1056
1057           /* may have to refresh duration */
1058           gst_base_src_update_length (src, 0, &length,
1059               g_atomic_int_get (&src->priv->dynamic_size));
1060
1061           /* this is the duration as configured by the subclass. */
1062           GST_OBJECT_LOCK (src);
1063           duration = src->segment.duration;
1064           seg_format = src->segment.format;
1065           GST_OBJECT_UNLOCK (src);
1066
1067           GST_LOG_OBJECT (src, "duration %" G_GINT64_FORMAT ", format %s",
1068               duration, gst_format_get_name (seg_format));
1069
1070           if (duration != -1) {
1071             /* convert to requested format, if this fails, we have a duration
1072              * but we cannot answer the query, we must return FALSE. */
1073             res =
1074                 gst_pad_query_convert (src->srcpad, seg_format,
1075                 duration, format, &duration);
1076           } else {
1077             /* The subclass did not configure a duration, we assume that the
1078              * media has an unknown duration then and we return TRUE to report
1079              * this. Note that this is not the same as returning FALSE, which
1080              * means that we cannot report the duration at all. */
1081             res = TRUE;
1082           }
1083
1084           if (res)
1085             gst_query_set_duration (query, format, duration);
1086
1087           break;
1088         }
1089       }
1090       break;
1091     }
1092
1093     case GST_QUERY_SEEKING:
1094     {
1095       GstFormat format, seg_format;
1096       gint64 duration;
1097
1098       GST_OBJECT_LOCK (src);
1099       duration = src->segment.duration;
1100       seg_format = src->segment.format;
1101       GST_OBJECT_UNLOCK (src);
1102
1103       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
1104       if (format == seg_format) {
1105         gst_query_set_seeking (query, seg_format,
1106             gst_base_src_seekable (src), 0, duration);
1107         res = TRUE;
1108       } else {
1109         /* FIXME 2.0: return TRUE + seekable=FALSE for SEEKING query here */
1110         /* Don't reply to the query to make up for demuxers which don't
1111          * handle the SEEKING query yet. Players like Totem will fall back
1112          * to the duration when the SEEKING query isn't answered. */
1113         res = FALSE;
1114       }
1115       break;
1116     }
1117     case GST_QUERY_SEGMENT:
1118     {
1119       GstFormat format;
1120       gint64 start, stop;
1121
1122       GST_OBJECT_LOCK (src);
1123
1124       format = src->segment.format;
1125
1126       start =
1127           gst_segment_to_stream_time (&src->segment, format,
1128           src->segment.start);
1129       if ((stop = src->segment.stop) == -1)
1130         stop = src->segment.duration;
1131       else
1132         stop = gst_segment_to_stream_time (&src->segment, format, stop);
1133
1134       gst_query_set_segment (query, src->segment.rate, format, start, stop);
1135
1136       GST_OBJECT_UNLOCK (src);
1137       res = TRUE;
1138       break;
1139     }
1140
1141     case GST_QUERY_FORMATS:
1142     {
1143       gst_query_set_formats (query, 3, GST_FORMAT_DEFAULT,
1144           GST_FORMAT_BYTES, GST_FORMAT_PERCENT);
1145       res = TRUE;
1146       break;
1147     }
1148     case GST_QUERY_CONVERT:
1149     {
1150       GstFormat src_fmt, dest_fmt;
1151       gint64 src_val, dest_val;
1152
1153       gst_query_parse_convert (query, &src_fmt, &src_val, &dest_fmt, &dest_val);
1154
1155       /* we can only convert between equal formats... */
1156       if (src_fmt == dest_fmt) {
1157         dest_val = src_val;
1158         res = TRUE;
1159       } else
1160         res = FALSE;
1161
1162       gst_query_set_convert (query, src_fmt, src_val, dest_fmt, dest_val);
1163       break;
1164     }
1165     case GST_QUERY_LATENCY:
1166     {
1167       GstClockTime min, max;
1168       gboolean live;
1169
1170       /* Subclasses should override and implement something useful */
1171       res = gst_base_src_query_latency (src, &live, &min, &max);
1172
1173       GST_LOG_OBJECT (src, "report latency: live %d, min %" GST_TIME_FORMAT
1174           ", max %" GST_TIME_FORMAT, live, GST_TIME_ARGS (min),
1175           GST_TIME_ARGS (max));
1176
1177       gst_query_set_latency (query, live, min, max);
1178       break;
1179     }
1180     case GST_QUERY_JITTER:
1181     case GST_QUERY_RATE:
1182       res = FALSE;
1183       break;
1184     case GST_QUERY_BUFFERING:
1185     {
1186       GstFormat format, seg_format;
1187       gint64 start, stop, estimated;
1188
1189       gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1190
1191       GST_DEBUG_OBJECT (src, "buffering query in format %s",
1192           gst_format_get_name (format));
1193
1194       GST_OBJECT_LOCK (src);
1195       if (src->random_access) {
1196         estimated = 0;
1197         start = 0;
1198         if (format == GST_FORMAT_PERCENT)
1199           stop = GST_FORMAT_PERCENT_MAX;
1200         else
1201           stop = src->segment.duration;
1202       } else {
1203         estimated = -1;
1204         start = -1;
1205         stop = -1;
1206       }
1207       seg_format = src->segment.format;
1208       GST_OBJECT_UNLOCK (src);
1209
1210       /* convert to required format. When the conversion fails, we can't answer
1211        * the query. When the value is unknown, we can don't perform conversion
1212        * but report TRUE. */
1213       if (format != GST_FORMAT_PERCENT && stop != -1) {
1214         res = gst_pad_query_convert (src->srcpad, seg_format,
1215             stop, format, &stop);
1216       } else {
1217         res = TRUE;
1218       }
1219       if (res && format != GST_FORMAT_PERCENT && start != -1)
1220         res = gst_pad_query_convert (src->srcpad, seg_format,
1221             start, format, &start);
1222
1223       gst_query_set_buffering_range (query, format, start, stop, estimated);
1224       break;
1225     }
1226     case GST_QUERY_SCHEDULING:
1227     {
1228       gboolean random_access;
1229
1230       random_access = gst_base_src_is_random_access (src);
1231
1232       /* we can operate in getrange mode if the native format is bytes
1233        * and we are seekable, this condition is set in the random_access
1234        * flag and is set in the _start() method. */
1235       gst_query_set_scheduling (query, GST_SCHEDULING_FLAG_SEEKABLE, 1, -1, 0);
1236       if (random_access)
1237         gst_query_add_scheduling_mode (query, GST_PAD_MODE_PULL);
1238       gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
1239
1240       res = TRUE;
1241       break;
1242     }
1243     case GST_QUERY_CAPS:
1244     {
1245       GstBaseSrcClass *bclass;
1246       GstCaps *caps, *filter;
1247
1248       bclass = GST_BASE_SRC_GET_CLASS (src);
1249       if (bclass->get_caps) {
1250         gst_query_parse_caps (query, &filter);
1251         if ((caps = bclass->get_caps (src, filter))) {
1252           gst_query_set_caps_result (query, caps);
1253           gst_caps_unref (caps);
1254           res = TRUE;
1255         } else {
1256           res = FALSE;
1257         }
1258       } else
1259         res = FALSE;
1260       break;
1261     }
1262     case GST_QUERY_URI:{
1263       if (GST_IS_URI_HANDLER (src)) {
1264         gchar *uri = gst_uri_handler_get_uri (GST_URI_HANDLER (src));
1265
1266         if (uri != NULL) {
1267           gst_query_set_uri (query, uri);
1268           g_free (uri);
1269           res = TRUE;
1270         } else {
1271           res = FALSE;
1272         }
1273       } else {
1274         res = FALSE;
1275       }
1276       break;
1277     }
1278     default:
1279       res = FALSE;
1280       break;
1281   }
1282   GST_DEBUG_OBJECT (src, "query %s returns %d", GST_QUERY_TYPE_NAME (query),
1283       res);
1284
1285   return res;
1286 }
1287
1288 static gboolean
1289 gst_base_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
1290 {
1291   GstBaseSrc *src;
1292   GstBaseSrcClass *bclass;
1293   gboolean result = FALSE;
1294
1295   src = GST_BASE_SRC (parent);
1296   bclass = GST_BASE_SRC_GET_CLASS (src);
1297
1298   if (bclass->query)
1299     result = bclass->query (src, query);
1300
1301   return result;
1302 }
1303
1304 static gboolean
1305 gst_base_src_default_do_seek (GstBaseSrc * src, GstSegment * segment)
1306 {
1307   gboolean res = TRUE;
1308
1309   /* update our offset if the start/stop position was updated */
1310   if (segment->format == GST_FORMAT_BYTES) {
1311     segment->time = segment->start;
1312   } else if (segment->start == 0) {
1313     /* seek to start, we can implement a default for this. */
1314     segment->time = 0;
1315   } else {
1316     res = FALSE;
1317     GST_INFO_OBJECT (src, "Can't do a default seek");
1318   }
1319
1320   return res;
1321 }
1322
1323 static gboolean
1324 gst_base_src_do_seek (GstBaseSrc * src, GstSegment * segment)
1325 {
1326   GstBaseSrcClass *bclass;
1327   gboolean result = FALSE;
1328
1329   bclass = GST_BASE_SRC_GET_CLASS (src);
1330
1331   GST_INFO_OBJECT (src, "seeking: %" GST_SEGMENT_FORMAT, segment);
1332
1333   if (bclass->do_seek)
1334     result = bclass->do_seek (src, segment);
1335
1336   return result;
1337 }
1338
1339 #define SEEK_TYPE_IS_RELATIVE(t) (((t) != GST_SEEK_TYPE_NONE) && ((t) != GST_SEEK_TYPE_SET))
1340
1341 static gboolean
1342 gst_base_src_default_prepare_seek_segment (GstBaseSrc * src, GstEvent * event,
1343     GstSegment * segment)
1344 {
1345   /* By default, we try one of 2 things:
1346    *   - For absolute seek positions, convert the requested position to our
1347    *     configured processing format and place it in the output segment \
1348    *   - For relative seek positions, convert our current (input) values to the
1349    *     seek format, adjust by the relative seek offset and then convert back to
1350    *     the processing format
1351    */
1352   GstSeekType start_type, stop_type;
1353   gint64 start, stop;
1354   GstSeekFlags flags;
1355   GstFormat seek_format, dest_format;
1356   gdouble rate;
1357   gboolean update;
1358   gboolean res = TRUE;
1359
1360   gst_event_parse_seek (event, &rate, &seek_format, &flags,
1361       &start_type, &start, &stop_type, &stop);
1362   dest_format = segment->format;
1363
1364   if (seek_format == dest_format) {
1365     gst_segment_do_seek (segment, rate, seek_format, flags,
1366         start_type, start, stop_type, stop, &update);
1367     return TRUE;
1368   }
1369
1370   if (start_type != GST_SEEK_TYPE_NONE) {
1371     /* FIXME: Handle seek_end by converting the input segment vals */
1372     res =
1373         gst_pad_query_convert (src->srcpad, seek_format, start, dest_format,
1374         &start);
1375     start_type = GST_SEEK_TYPE_SET;
1376   }
1377
1378   if (res && stop_type != GST_SEEK_TYPE_NONE) {
1379     /* FIXME: Handle seek_end by converting the input segment vals */
1380     res =
1381         gst_pad_query_convert (src->srcpad, seek_format, stop, dest_format,
1382         &stop);
1383     stop_type = GST_SEEK_TYPE_SET;
1384   }
1385
1386   /* And finally, configure our output segment in the desired format */
1387   gst_segment_do_seek (segment, rate, dest_format, flags, start_type, start,
1388       stop_type, stop, &update);
1389
1390   if (!res)
1391     goto no_format;
1392
1393   return res;
1394
1395 no_format:
1396   {
1397     GST_DEBUG_OBJECT (src, "undefined format given, seek aborted.");
1398     return FALSE;
1399   }
1400 }
1401
1402 static gboolean
1403 gst_base_src_prepare_seek_segment (GstBaseSrc * src, GstEvent * event,
1404     GstSegment * seeksegment)
1405 {
1406   GstBaseSrcClass *bclass;
1407   gboolean result = FALSE;
1408
1409   bclass = GST_BASE_SRC_GET_CLASS (src);
1410
1411   if (bclass->prepare_seek_segment)
1412     result = bclass->prepare_seek_segment (src, event, seeksegment);
1413
1414   return result;
1415 }
1416
1417 static GstFlowReturn
1418 gst_base_src_default_alloc (GstBaseSrc * src, guint64 offset,
1419     guint size, GstBuffer ** buffer)
1420 {
1421   GstFlowReturn ret;
1422   GstBaseSrcPrivate *priv = src->priv;
1423
1424   if (priv->pool) {
1425     ret = gst_buffer_pool_acquire_buffer (priv->pool, buffer, NULL);
1426   } else if (size != -1) {
1427     *buffer = gst_buffer_new_allocate (priv->allocator, size, &priv->params);
1428     if (G_UNLIKELY (*buffer == NULL))
1429       goto alloc_failed;
1430
1431     ret = GST_FLOW_OK;
1432   } else {
1433     GST_WARNING_OBJECT (src, "Not trying to alloc %u bytes. Blocksize not set?",
1434         size);
1435     goto alloc_failed;
1436   }
1437   return ret;
1438
1439   /* ERRORS */
1440 alloc_failed:
1441   {
1442     GST_ERROR_OBJECT (src, "Failed to allocate %u bytes", size);
1443     return GST_FLOW_ERROR;
1444   }
1445 }
1446
1447 static GstFlowReturn
1448 gst_base_src_default_create (GstBaseSrc * src, guint64 offset,
1449     guint size, GstBuffer ** buffer)
1450 {
1451   GstBaseSrcClass *bclass;
1452   GstFlowReturn ret;
1453   GstBuffer *res_buf;
1454
1455   bclass = GST_BASE_SRC_GET_CLASS (src);
1456
1457   if (G_UNLIKELY (!bclass->alloc))
1458     goto no_function;
1459   if (G_UNLIKELY (!bclass->fill))
1460     goto no_function;
1461
1462   if (*buffer == NULL) {
1463     /* downstream did not provide us with a buffer to fill, allocate one
1464      * ourselves */
1465     ret = bclass->alloc (src, offset, size, &res_buf);
1466     if (G_UNLIKELY (ret != GST_FLOW_OK))
1467       goto alloc_failed;
1468   } else {
1469     res_buf = *buffer;
1470   }
1471
1472   if (G_LIKELY (size > 0)) {
1473     /* only call fill when there is a size */
1474     ret = bclass->fill (src, offset, size, res_buf);
1475     if (G_UNLIKELY (ret != GST_FLOW_OK))
1476       goto not_ok;
1477   }
1478
1479   *buffer = res_buf;
1480
1481   return GST_FLOW_OK;
1482
1483   /* ERRORS */
1484 no_function:
1485   {
1486     GST_DEBUG_OBJECT (src, "no fill or alloc function");
1487     return GST_FLOW_NOT_SUPPORTED;
1488   }
1489 alloc_failed:
1490   {
1491     GST_DEBUG_OBJECT (src, "Failed to allocate buffer of %u bytes", size);
1492     return ret;
1493   }
1494 not_ok:
1495   {
1496     GST_DEBUG_OBJECT (src, "fill returned %d (%s)", ret,
1497         gst_flow_get_name (ret));
1498     if (*buffer == NULL)
1499       gst_buffer_unref (res_buf);
1500     return ret;
1501   }
1502 }
1503
1504 /* this code implements the seeking. It is a good example
1505  * handling all cases.
1506  *
1507  * A seek updates the currently configured segment.start
1508  * and segment.stop values based on the SEEK_TYPE. If the
1509  * segment.start value is updated, a seek to this new position
1510  * should be performed.
1511  *
1512  * The seek can only be executed when we are not currently
1513  * streaming any data, to make sure that this is the case, we
1514  * acquire the STREAM_LOCK which is taken when we are in the
1515  * _loop() function or when a getrange() is called. Normally
1516  * we will not receive a seek if we are operating in pull mode
1517  * though. When we operate as a live source we might block on the live
1518  * cond, which does not release the STREAM_LOCK. Therefore we will try
1519  * to grab the LIVE_LOCK instead of the STREAM_LOCK to make sure it is
1520  * safe to perform the seek.
1521  *
1522  * When we are in the loop() function, we might be in the middle
1523  * of pushing a buffer, which might block in a sink. To make sure
1524  * that the push gets unblocked we push out a FLUSH_START event.
1525  * Our loop function will get a FLUSHING return value from
1526  * the push and will pause, effectively releasing the STREAM_LOCK.
1527  *
1528  * For a non-flushing seek, we pause the task, which might eventually
1529  * release the STREAM_LOCK. We say eventually because when the sink
1530  * blocks on the sample we might wait a very long time until the sink
1531  * unblocks the sample. In any case we acquire the STREAM_LOCK and
1532  * can continue the seek. A non-flushing seek is normally done in a
1533  * running pipeline to perform seamless playback, this means that the sink is
1534  * PLAYING and will return from its chain function.
1535  * In the case of a non-flushing seek we need to make sure that the
1536  * data we output after the seek is continuous with the previous data,
1537  * this is because a non-flushing seek does not reset the running-time
1538  * to 0. We do this by closing the currently running segment, ie. sending
1539  * a new_segment event with the stop position set to the last processed
1540  * position.
1541  *
1542  * After updating the segment.start/stop values, we prepare for
1543  * streaming again. We push out a FLUSH_STOP to make the peer pad
1544  * accept data again and we start our task again.
1545  *
1546  * A segment seek posts a message on the bus saying that the playback
1547  * of the segment started. We store the segment flag internally because
1548  * when we reach the segment.stop we have to post a segment.done
1549  * instead of EOS when doing a segment seek.
1550  */
1551 static gboolean
1552 gst_base_src_perform_seek (GstBaseSrc * src, GstEvent * event, gboolean unlock)
1553 {
1554   gboolean res = TRUE, tres;
1555   gdouble rate;
1556   GstFormat seek_format, dest_format;
1557   GstSeekFlags flags;
1558   GstSeekType start_type, stop_type;
1559   gint64 start, stop;
1560   gboolean flush, playing;
1561   gboolean update;
1562   gboolean relative_seek = FALSE;
1563   gboolean seekseg_configured = FALSE;
1564   GstSegment seeksegment;
1565   guint32 seqnum;
1566   GstEvent *tevent;
1567
1568   GST_DEBUG_OBJECT (src, "doing seek: %" GST_PTR_FORMAT, event);
1569
1570   GST_OBJECT_LOCK (src);
1571   dest_format = src->segment.format;
1572   GST_OBJECT_UNLOCK (src);
1573
1574   if (event) {
1575     gst_event_parse_seek (event, &rate, &seek_format, &flags,
1576         &start_type, &start, &stop_type, &stop);
1577
1578     relative_seek = SEEK_TYPE_IS_RELATIVE (start_type) ||
1579         SEEK_TYPE_IS_RELATIVE (stop_type);
1580
1581     if (dest_format != seek_format && !relative_seek) {
1582       /* If we have an ABSOLUTE position (SEEK_SET only), we can convert it
1583        * here before taking the stream lock, otherwise we must convert it later,
1584        * once we have the stream lock and can read the last configures segment
1585        * start and stop positions */
1586       gst_segment_init (&seeksegment, dest_format);
1587
1588       if (!gst_base_src_prepare_seek_segment (src, event, &seeksegment))
1589         goto prepare_failed;
1590
1591       seekseg_configured = TRUE;
1592     }
1593
1594     flush = flags & GST_SEEK_FLAG_FLUSH;
1595     seqnum = gst_event_get_seqnum (event);
1596   } else {
1597     flush = FALSE;
1598     /* get next seqnum */
1599     seqnum = gst_util_seqnum_next ();
1600   }
1601
1602   /* send flush start */
1603   if (flush) {
1604     tevent = gst_event_new_flush_start ();
1605     gst_event_set_seqnum (tevent, seqnum);
1606     gst_pad_push_event (src->srcpad, tevent);
1607   } else
1608     gst_pad_pause_task (src->srcpad);
1609
1610   /* unblock streaming thread. */
1611   if (unlock)
1612     gst_base_src_set_flushing (src, TRUE, FALSE, &playing);
1613
1614   /* grab streaming lock, this should eventually be possible, either
1615    * because the task is paused, our streaming thread stopped
1616    * or because our peer is flushing. */
1617   GST_PAD_STREAM_LOCK (src->srcpad);
1618   if (G_UNLIKELY (src->priv->seqnum == seqnum)) {
1619     /* we have seen this event before, issue a warning for now */
1620     GST_WARNING_OBJECT (src, "duplicate event found %" G_GUINT32_FORMAT,
1621         seqnum);
1622   } else {
1623     src->priv->seqnum = seqnum;
1624     GST_DEBUG_OBJECT (src, "seek with seqnum %" G_GUINT32_FORMAT, seqnum);
1625   }
1626
1627   if (unlock)
1628     gst_base_src_set_flushing (src, FALSE, playing, NULL);
1629
1630   /* If we configured the seeksegment above, don't overwrite it now. Otherwise
1631    * copy the current segment info into the temp segment that we can actually
1632    * attempt the seek with. We only update the real segment if the seek succeeds. */
1633   if (!seekseg_configured) {
1634     memcpy (&seeksegment, &src->segment, sizeof (GstSegment));
1635
1636     /* now configure the final seek segment */
1637     if (event) {
1638       if (seeksegment.format != seek_format) {
1639         /* OK, here's where we give the subclass a chance to convert the relative
1640          * seek into an absolute one in the processing format. We set up any
1641          * absolute seek above, before taking the stream lock. */
1642         if (!gst_base_src_prepare_seek_segment (src, event, &seeksegment)) {
1643           GST_DEBUG_OBJECT (src, "Preparing the seek failed after flushing. "
1644               "Aborting seek");
1645           res = FALSE;
1646         }
1647       } else {
1648         /* The seek format matches our processing format, no need to ask the
1649          * the subclass to configure the segment. */
1650         gst_segment_do_seek (&seeksegment, rate, seek_format, flags,
1651             start_type, start, stop_type, stop, &update);
1652       }
1653     }
1654     /* Else, no seek event passed, so we're just (re)starting the
1655        current segment. */
1656   }
1657
1658   if (res) {
1659     GST_DEBUG_OBJECT (src, "segment configured from %" G_GINT64_FORMAT
1660         " to %" G_GINT64_FORMAT ", position %" G_GINT64_FORMAT,
1661         seeksegment.start, seeksegment.stop, seeksegment.position);
1662
1663     /* do the seek, segment.position contains the new position. */
1664     res = gst_base_src_do_seek (src, &seeksegment);
1665   }
1666
1667   /* and prepare to continue streaming */
1668   if (flush) {
1669     tevent = gst_event_new_flush_stop (TRUE);
1670     gst_event_set_seqnum (tevent, seqnum);
1671     /* send flush stop, peer will accept data and events again. We
1672      * are not yet providing data as we still have the STREAM_LOCK. */
1673     gst_pad_push_event (src->srcpad, tevent);
1674   }
1675
1676   /* The subclass must have converted the segment to the processing format
1677    * by now */
1678   if (res && seeksegment.format != dest_format) {
1679     GST_DEBUG_OBJECT (src, "Subclass failed to prepare a seek segment "
1680         "in the correct format. Aborting seek.");
1681     res = FALSE;
1682   }
1683
1684   /* if the seek was successful, we update our real segment and push
1685    * out the new segment. */
1686   if (res) {
1687     GST_OBJECT_LOCK (src);
1688     memcpy (&src->segment, &seeksegment, sizeof (GstSegment));
1689     GST_OBJECT_UNLOCK (src);
1690
1691     if (seeksegment.flags & GST_SEGMENT_FLAG_SEGMENT) {
1692       GstMessage *message;
1693
1694       message = gst_message_new_segment_start (GST_OBJECT (src),
1695           seeksegment.format, seeksegment.position);
1696       gst_message_set_seqnum (message, seqnum);
1697
1698       gst_element_post_message (GST_ELEMENT (src), message);
1699     }
1700
1701     src->priv->segment_pending = TRUE;
1702     src->priv->segment_seqnum = seqnum;
1703   }
1704
1705   src->priv->discont = TRUE;
1706   src->running = TRUE;
1707   /* and restart the task in case it got paused explicitly or by
1708    * the FLUSH_START event we pushed out. */
1709   tres = gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
1710       src->srcpad, NULL);
1711   if (res && !tres)
1712     res = FALSE;
1713
1714   /* and release the lock again so we can continue streaming */
1715   GST_PAD_STREAM_UNLOCK (src->srcpad);
1716
1717   return res;
1718
1719   /* ERROR */
1720 prepare_failed:
1721   GST_DEBUG_OBJECT (src, "Preparing the seek failed before flushing. "
1722       "Aborting seek");
1723   return FALSE;
1724 }
1725
1726 /* all events send to this element directly. This is mainly done from the
1727  * application.
1728  */
1729 static gboolean
1730 gst_base_src_send_event (GstElement * element, GstEvent * event)
1731 {
1732   GstBaseSrc *src;
1733   gboolean result = FALSE;
1734   GstBaseSrcClass *bclass;
1735
1736   src = GST_BASE_SRC (element);
1737   bclass = GST_BASE_SRC_GET_CLASS (src);
1738
1739   GST_DEBUG_OBJECT (src, "handling event %p %" GST_PTR_FORMAT, event, event);
1740
1741   switch (GST_EVENT_TYPE (event)) {
1742       /* bidirectional events */
1743     case GST_EVENT_FLUSH_START:
1744       GST_DEBUG_OBJECT (src, "pushing flush-start event downstream");
1745       result = gst_pad_push_event (src->srcpad, event);
1746       /* also unblock the create function */
1747       gst_base_src_activate_pool (src, FALSE);
1748       /* unlock any subclasses, we need to do this before grabbing the
1749        * LIVE_LOCK since we hold this lock before going into ::create. We pass an
1750        * unlock to the params because of backwards compat (see seek handler)*/
1751       if (bclass->unlock)
1752         bclass->unlock (src);
1753
1754       /* the live lock is released when we are blocked, waiting for playing or
1755        * when we sync to the clock. */
1756       GST_LIVE_LOCK (src);
1757       src->priv->flushing = TRUE;
1758       /* clear pending EOS if any */
1759       if (g_atomic_int_get (&src->priv->has_pending_eos)) {
1760         GST_OBJECT_LOCK (src);
1761         CLEAR_PENDING_EOS (src);
1762         src->priv->forced_eos = FALSE;
1763         GST_OBJECT_UNLOCK (src);
1764       }
1765       if (bclass->unlock_stop)
1766         bclass->unlock_stop (src);
1767       if (src->clock_id)
1768         gst_clock_id_unschedule (src->clock_id);
1769       GST_DEBUG_OBJECT (src, "signal");
1770       GST_LIVE_SIGNAL (src);
1771       GST_LIVE_UNLOCK (src);
1772       event = NULL;
1773       break;
1774     case GST_EVENT_FLUSH_STOP:
1775     {
1776       gboolean start;
1777
1778       GST_LIVE_LOCK (src);
1779       src->priv->segment_pending = TRUE;
1780       src->priv->flushing = FALSE;
1781       GST_DEBUG_OBJECT (src, "pushing flush-stop event downstream");
1782       result = gst_pad_push_event (src->srcpad, event);
1783
1784       gst_base_src_activate_pool (src, TRUE);
1785
1786       GST_OBJECT_LOCK (src->srcpad);
1787       start = (GST_PAD_MODE (src->srcpad) == GST_PAD_MODE_PUSH);
1788       GST_OBJECT_UNLOCK (src->srcpad);
1789
1790       if (src->is_live) {
1791         if (!src->live_running)
1792           start = FALSE;
1793       }
1794
1795       if (start)
1796         gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
1797             src->srcpad, NULL);
1798       GST_LIVE_UNLOCK (src);
1799       event = NULL;
1800       break;
1801     }
1802
1803       /* downstream serialized events */
1804     case GST_EVENT_EOS:
1805     {
1806       /* queue EOS and make sure the task or pull function performs the EOS
1807        * actions.
1808        *
1809        * We have two possibilities:
1810        *
1811        *  - Before we are to enter the _create function, we check the has_pending_eos
1812        *    first and do EOS instead of entering it.
1813        *  - If we are in the _create function or we did not manage to set the
1814        *    flag fast enough and we are about to enter the _create function,
1815        *    we unlock it so that we exit with FLUSHING immediately. We then
1816        *    check the EOS flag and do the EOS logic.
1817        */
1818       GST_OBJECT_LOCK (src);
1819       g_atomic_int_set (&src->priv->has_pending_eos, TRUE);
1820       if (src->priv->pending_eos)
1821         gst_event_unref (src->priv->pending_eos);
1822       src->priv->pending_eos = event;
1823       event = NULL;
1824       GST_OBJECT_UNLOCK (src);
1825
1826       GST_DEBUG_OBJECT (src, "EOS marked, calling unlock");
1827
1828       /* unlock the _create function so that we can check the has_pending_eos flag
1829        * and we can do EOS. This will eventually release the LIVE_LOCK again so
1830        * that we can grab it and stop the unlock again. We don't take the stream
1831        * lock so that this operation is guaranteed to never block. */
1832       gst_base_src_activate_pool (src, FALSE);
1833       if (bclass->unlock)
1834         bclass->unlock (src);
1835
1836       GST_DEBUG_OBJECT (src, "unlock called, waiting for LIVE_LOCK");
1837
1838       GST_LIVE_LOCK (src);
1839       GST_DEBUG_OBJECT (src, "LIVE_LOCK acquired, calling unlock_stop");
1840       /* now stop the unlock of the streaming thread again. Grabbing the live
1841        * lock is enough because that protects the create function. */
1842       if (bclass->unlock_stop)
1843         bclass->unlock_stop (src);
1844       gst_base_src_activate_pool (src, TRUE);
1845       GST_LIVE_UNLOCK (src);
1846
1847       result = TRUE;
1848       break;
1849     }
1850     case GST_EVENT_SEGMENT:
1851       /* sending random SEGMENT downstream can break sync. */
1852       break;
1853     case GST_EVENT_TAG:
1854     case GST_EVENT_CUSTOM_DOWNSTREAM:
1855     case GST_EVENT_CUSTOM_BOTH:
1856     case GST_EVENT_PROTECTION:
1857       /* Insert TAG, CUSTOM_DOWNSTREAM, CUSTOM_BOTH, PROTECTION in the dataflow */
1858       GST_OBJECT_LOCK (src);
1859       src->priv->pending_events =
1860           g_list_append (src->priv->pending_events, event);
1861       g_atomic_int_set (&src->priv->have_events, TRUE);
1862       GST_OBJECT_UNLOCK (src);
1863       event = NULL;
1864       result = TRUE;
1865       break;
1866     case GST_EVENT_BUFFERSIZE:
1867       /* does not seem to make much sense currently */
1868       break;
1869
1870       /* upstream events */
1871     case GST_EVENT_QOS:
1872       /* elements should override send_event and do something */
1873       break;
1874     case GST_EVENT_SEEK:
1875     {
1876       gboolean started;
1877
1878       GST_OBJECT_LOCK (src->srcpad);
1879       if (GST_PAD_MODE (src->srcpad) == GST_PAD_MODE_PULL)
1880         goto wrong_mode;
1881       started = GST_PAD_MODE (src->srcpad) == GST_PAD_MODE_PUSH;
1882       GST_OBJECT_UNLOCK (src->srcpad);
1883
1884       if (started) {
1885         GST_DEBUG_OBJECT (src, "performing seek");
1886         /* when we are running in push mode, we can execute the
1887          * seek right now. */
1888         result = gst_base_src_perform_seek (src, event, TRUE);
1889       } else {
1890         GstEvent **event_p;
1891
1892         /* else we store the event and execute the seek when we
1893          * get activated */
1894         GST_OBJECT_LOCK (src);
1895         GST_DEBUG_OBJECT (src, "queueing seek");
1896         event_p = &src->pending_seek;
1897         gst_event_replace ((GstEvent **) event_p, event);
1898         GST_OBJECT_UNLOCK (src);
1899         /* assume the seek will work */
1900         result = TRUE;
1901       }
1902       break;
1903     }
1904     case GST_EVENT_NAVIGATION:
1905       /* could make sense for elements that do something with navigation events
1906        * but then they would need to override the send_event function */
1907       break;
1908     case GST_EVENT_LATENCY:
1909       /* does not seem to make sense currently */
1910       break;
1911
1912       /* custom events */
1913     case GST_EVENT_CUSTOM_UPSTREAM:
1914       /* override send_event if you want this */
1915       break;
1916     case GST_EVENT_CUSTOM_DOWNSTREAM_OOB:
1917     case GST_EVENT_CUSTOM_BOTH_OOB:
1918       /* insert a random custom event into the pipeline */
1919       GST_DEBUG_OBJECT (src, "pushing custom OOB event downstream");
1920       result = gst_pad_push_event (src->srcpad, event);
1921       /* we gave away the ref to the event in the push */
1922       event = NULL;
1923       break;
1924     default:
1925       break;
1926   }
1927 done:
1928   /* if we still have a ref to the event, unref it now */
1929   if (event)
1930     gst_event_unref (event);
1931
1932   return result;
1933
1934   /* ERRORS */
1935 wrong_mode:
1936   {
1937     GST_DEBUG_OBJECT (src, "cannot perform seek when operating in pull mode");
1938     GST_OBJECT_UNLOCK (src->srcpad);
1939     result = FALSE;
1940     goto done;
1941   }
1942 }
1943
1944 static gboolean
1945 gst_base_src_seekable (GstBaseSrc * src)
1946 {
1947   GstBaseSrcClass *bclass;
1948   bclass = GST_BASE_SRC_GET_CLASS (src);
1949   if (bclass->is_seekable)
1950     return bclass->is_seekable (src);
1951   else
1952     return FALSE;
1953 }
1954
1955 static void
1956 gst_base_src_update_qos (GstBaseSrc * src,
1957     gdouble proportion, GstClockTimeDiff diff, GstClockTime timestamp)
1958 {
1959   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, src,
1960       "qos: proportion: %lf, diff %" G_GINT64_FORMAT ", timestamp %"
1961       GST_TIME_FORMAT, proportion, diff, GST_TIME_ARGS (timestamp));
1962
1963   GST_OBJECT_LOCK (src);
1964   src->priv->proportion = proportion;
1965   src->priv->earliest_time = timestamp + diff;
1966   GST_OBJECT_UNLOCK (src);
1967 }
1968
1969
1970 static gboolean
1971 gst_base_src_default_event (GstBaseSrc * src, GstEvent * event)
1972 {
1973   gboolean result;
1974
1975   GST_DEBUG_OBJECT (src, "handle event %" GST_PTR_FORMAT, event);
1976
1977   switch (GST_EVENT_TYPE (event)) {
1978     case GST_EVENT_SEEK:
1979       /* is normally called when in push mode */
1980       if (!gst_base_src_seekable (src))
1981         goto not_seekable;
1982
1983       result = gst_base_src_perform_seek (src, event, TRUE);
1984       break;
1985     case GST_EVENT_FLUSH_START:
1986       /* cancel any blocking getrange, is normally called
1987        * when in pull mode. */
1988       result = gst_base_src_set_flushing (src, TRUE, FALSE, NULL);
1989       break;
1990     case GST_EVENT_FLUSH_STOP:
1991       result = gst_base_src_set_flushing (src, FALSE, TRUE, NULL);
1992       break;
1993     case GST_EVENT_QOS:
1994     {
1995       gdouble proportion;
1996       GstClockTimeDiff diff;
1997       GstClockTime timestamp;
1998
1999       gst_event_parse_qos (event, NULL, &proportion, &diff, &timestamp);
2000       gst_base_src_update_qos (src, proportion, diff, timestamp);
2001       result = TRUE;
2002       break;
2003     }
2004     case GST_EVENT_RECONFIGURE:
2005       result = TRUE;
2006       break;
2007     case GST_EVENT_LATENCY:
2008       result = TRUE;
2009       break;
2010     default:
2011       result = FALSE;
2012       break;
2013   }
2014   return result;
2015
2016   /* ERRORS */
2017 not_seekable:
2018   {
2019     GST_DEBUG_OBJECT (src, "is not seekable");
2020     return FALSE;
2021   }
2022 }
2023
2024 static gboolean
2025 gst_base_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
2026 {
2027   GstBaseSrc *src;
2028   GstBaseSrcClass *bclass;
2029   gboolean result = FALSE;
2030
2031   src = GST_BASE_SRC (parent);
2032   bclass = GST_BASE_SRC_GET_CLASS (src);
2033
2034   if (bclass->event) {
2035     if (!(result = bclass->event (src, event)))
2036       goto subclass_failed;
2037   }
2038
2039 done:
2040   gst_event_unref (event);
2041
2042   return result;
2043
2044   /* ERRORS */
2045 subclass_failed:
2046   {
2047     GST_DEBUG_OBJECT (src, "subclass refused event");
2048     goto done;
2049   }
2050 }
2051
2052 static void
2053 gst_base_src_set_property (GObject * object, guint prop_id,
2054     const GValue * value, GParamSpec * pspec)
2055 {
2056   GstBaseSrc *src;
2057
2058   src = GST_BASE_SRC (object);
2059
2060   switch (prop_id) {
2061     case PROP_BLOCKSIZE:
2062       gst_base_src_set_blocksize (src, g_value_get_uint (value));
2063       break;
2064     case PROP_NUM_BUFFERS:
2065       src->num_buffers = g_value_get_int (value);
2066       break;
2067     case PROP_TYPEFIND:
2068       src->typefind = g_value_get_boolean (value);
2069       break;
2070     case PROP_DO_TIMESTAMP:
2071       gst_base_src_set_do_timestamp (src, g_value_get_boolean (value));
2072       break;
2073     default:
2074       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2075       break;
2076   }
2077 }
2078
2079 static void
2080 gst_base_src_get_property (GObject * object, guint prop_id, GValue * value,
2081     GParamSpec * pspec)
2082 {
2083   GstBaseSrc *src;
2084
2085   src = GST_BASE_SRC (object);
2086
2087   switch (prop_id) {
2088     case PROP_BLOCKSIZE:
2089       g_value_set_uint (value, gst_base_src_get_blocksize (src));
2090       break;
2091     case PROP_NUM_BUFFERS:
2092       g_value_set_int (value, src->num_buffers);
2093       break;
2094     case PROP_TYPEFIND:
2095       g_value_set_boolean (value, src->typefind);
2096       break;
2097     case PROP_DO_TIMESTAMP:
2098       g_value_set_boolean (value, gst_base_src_get_do_timestamp (src));
2099       break;
2100     default:
2101       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2102       break;
2103   }
2104 }
2105
2106 /* with STREAM_LOCK and LOCK */
2107 static GstClockReturn
2108 gst_base_src_wait (GstBaseSrc * basesrc, GstClock * clock, GstClockTime time)
2109 {
2110   GstClockReturn ret;
2111   GstClockID id;
2112
2113   id = gst_clock_new_single_shot_id (clock, time);
2114
2115   basesrc->clock_id = id;
2116   /* release the live lock while waiting */
2117   GST_LIVE_UNLOCK (basesrc);
2118
2119   ret = gst_clock_id_wait (id, NULL);
2120
2121   GST_LIVE_LOCK (basesrc);
2122   gst_clock_id_unref (id);
2123   basesrc->clock_id = NULL;
2124
2125   return ret;
2126 }
2127
2128 /* perform synchronisation on a buffer.
2129  * with STREAM_LOCK.
2130  */
2131 static GstClockReturn
2132 gst_base_src_do_sync (GstBaseSrc * basesrc, GstBuffer * buffer)
2133 {
2134   GstClockReturn result;
2135   GstClockTime start, end;
2136   GstBaseSrcClass *bclass;
2137   GstClockTime base_time;
2138   GstClock *clock;
2139   GstClockTime now = GST_CLOCK_TIME_NONE, pts, dts, timestamp;
2140   gboolean do_timestamp, first, pseudo_live, is_live;
2141
2142   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
2143
2144   start = end = -1;
2145   if (bclass->get_times)
2146     bclass->get_times (basesrc, buffer, &start, &end);
2147
2148   /* get buffer timestamp */
2149   dts = GST_BUFFER_DTS (buffer);
2150   pts = GST_BUFFER_PTS (buffer);
2151
2152   if (GST_CLOCK_TIME_IS_VALID (dts))
2153     timestamp = dts;
2154   else
2155     timestamp = pts;
2156
2157   /* grab the lock to prepare for clocking and calculate the startup
2158    * latency. */
2159   GST_OBJECT_LOCK (basesrc);
2160
2161   is_live = basesrc->is_live;
2162   /* if we are asked to sync against the clock we are a pseudo live element */
2163   pseudo_live = (start != -1 && is_live);
2164   /* check for the first buffer */
2165   first = (basesrc->priv->latency == -1);
2166
2167   if (timestamp != -1 && pseudo_live) {
2168     GstClockTime latency;
2169
2170     /* we have a timestamp and a sync time, latency is the diff */
2171     if (timestamp <= start)
2172       latency = start - timestamp;
2173     else
2174       latency = 0;
2175
2176     if (first) {
2177       GST_DEBUG_OBJECT (basesrc, "pseudo_live with latency %" GST_TIME_FORMAT,
2178           GST_TIME_ARGS (latency));
2179       /* first time we calculate latency, just configure */
2180       basesrc->priv->latency = latency;
2181     } else {
2182       if (basesrc->priv->latency != latency) {
2183         /* we have a new latency, FIXME post latency message */
2184         basesrc->priv->latency = latency;
2185         GST_DEBUG_OBJECT (basesrc, "latency changed to %" GST_TIME_FORMAT,
2186             GST_TIME_ARGS (latency));
2187       }
2188     }
2189   } else if (first) {
2190     GST_DEBUG_OBJECT (basesrc, "no latency needed, live %d, sync %d",
2191         is_live, start != -1);
2192     basesrc->priv->latency = 0;
2193   }
2194
2195   /* get clock, if no clock, we can't sync or do timestamps */
2196   if ((clock = GST_ELEMENT_CLOCK (basesrc)) == NULL)
2197     goto no_clock;
2198   else
2199     gst_object_ref (clock);
2200
2201   base_time = GST_ELEMENT_CAST (basesrc)->base_time;
2202
2203   do_timestamp = basesrc->priv->do_timestamp;
2204   GST_OBJECT_UNLOCK (basesrc);
2205
2206   /* first buffer, calculate the timestamp offset */
2207   if (first) {
2208     GstClockTime running_time;
2209
2210     now = gst_clock_get_time (clock);
2211     running_time = now - base_time;
2212
2213     GST_LOG_OBJECT (basesrc,
2214         "startup PTS: %" GST_TIME_FORMAT ", DTS %" GST_TIME_FORMAT
2215         ", running_time %" GST_TIME_FORMAT, GST_TIME_ARGS (pts),
2216         GST_TIME_ARGS (dts), GST_TIME_ARGS (running_time));
2217
2218     if (pseudo_live && timestamp != -1) {
2219       /* live source and we need to sync, add startup latency to all timestamps
2220        * to get the real running_time. Live sources should always timestamp
2221        * according to the current running time. */
2222       basesrc->priv->ts_offset = GST_CLOCK_DIFF (timestamp, running_time);
2223
2224       GST_LOG_OBJECT (basesrc, "live with sync, ts_offset %" GST_TIME_FORMAT,
2225           GST_TIME_ARGS (basesrc->priv->ts_offset));
2226     } else {
2227       basesrc->priv->ts_offset = 0;
2228       GST_LOG_OBJECT (basesrc, "no timestamp offset needed");
2229     }
2230
2231     if (!GST_CLOCK_TIME_IS_VALID (dts)) {
2232       if (do_timestamp) {
2233         dts = running_time;
2234       } else if (!GST_CLOCK_TIME_IS_VALID (pts)) {
2235         if (GST_CLOCK_TIME_IS_VALID (basesrc->segment.start)) {
2236           dts = basesrc->segment.start;
2237         } else {
2238           dts = 0;
2239         }
2240       }
2241       GST_BUFFER_DTS (buffer) = dts;
2242
2243       GST_LOG_OBJECT (basesrc, "created DTS %" GST_TIME_FORMAT,
2244           GST_TIME_ARGS (dts));
2245     }
2246   } else {
2247     /* not the first buffer, the timestamp is the diff between the clock and
2248      * base_time */
2249     if (do_timestamp && !GST_CLOCK_TIME_IS_VALID (dts)) {
2250       now = gst_clock_get_time (clock);
2251
2252       dts = now - base_time;
2253       GST_BUFFER_DTS (buffer) = dts;
2254
2255       GST_LOG_OBJECT (basesrc, "created DTS %" GST_TIME_FORMAT,
2256           GST_TIME_ARGS (dts));
2257     }
2258   }
2259   if (!GST_CLOCK_TIME_IS_VALID (pts)) {
2260     if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT))
2261       pts = dts;
2262
2263     GST_BUFFER_PTS (buffer) = dts;
2264
2265     GST_LOG_OBJECT (basesrc, "created PTS %" GST_TIME_FORMAT,
2266         GST_TIME_ARGS (pts));
2267   }
2268
2269   /* if we don't have a buffer timestamp, we don't sync */
2270   if (!GST_CLOCK_TIME_IS_VALID (start))
2271     goto no_sync;
2272
2273   if (is_live) {
2274     /* for pseudo live sources, add our ts_offset to the timestamp */
2275     if (GST_CLOCK_TIME_IS_VALID (pts))
2276       GST_BUFFER_PTS (buffer) += basesrc->priv->ts_offset;
2277     if (GST_CLOCK_TIME_IS_VALID (dts))
2278       GST_BUFFER_DTS (buffer) += basesrc->priv->ts_offset;
2279     start += basesrc->priv->ts_offset;
2280   }
2281
2282   GST_LOG_OBJECT (basesrc,
2283       "waiting for clock, base time %" GST_TIME_FORMAT
2284       ", stream_start %" GST_TIME_FORMAT,
2285       GST_TIME_ARGS (base_time), GST_TIME_ARGS (start));
2286
2287   result = gst_base_src_wait (basesrc, clock, start + base_time);
2288
2289   gst_object_unref (clock);
2290
2291   GST_LOG_OBJECT (basesrc, "clock entry done: %d", result);
2292
2293   return result;
2294
2295   /* special cases */
2296 no_clock:
2297   {
2298     GST_DEBUG_OBJECT (basesrc, "we have no clock");
2299     GST_OBJECT_UNLOCK (basesrc);
2300     return GST_CLOCK_OK;
2301   }
2302 no_sync:
2303   {
2304     GST_DEBUG_OBJECT (basesrc, "no sync needed");
2305     gst_object_unref (clock);
2306     return GST_CLOCK_OK;
2307   }
2308 }
2309
2310 /* Called with STREAM_LOCK and LIVE_LOCK */
2311 static gboolean
2312 gst_base_src_update_length (GstBaseSrc * src, guint64 offset, guint * length,
2313     gboolean force)
2314 {
2315   guint64 size, maxsize;
2316   GstBaseSrcClass *bclass;
2317   gint64 stop;
2318
2319   /* only operate if we are working with bytes */
2320   if (src->segment.format != GST_FORMAT_BYTES)
2321     return TRUE;
2322
2323   bclass = GST_BASE_SRC_GET_CLASS (src);
2324
2325   stop = src->segment.stop;
2326   /* get total file size */
2327   size = src->segment.duration;
2328
2329   /* when not doing automatic EOS, just use the stop position. We don't use
2330    * the size to check for EOS */
2331   if (!g_atomic_int_get (&src->priv->automatic_eos))
2332     maxsize = stop;
2333   /* Otherwise, the max amount of bytes to read is the total
2334    * size or up to the segment.stop if present. */
2335   else if (stop != -1)
2336     maxsize = size != -1 ? MIN (size, stop) : stop;
2337   else
2338     maxsize = size;
2339
2340   GST_DEBUG_OBJECT (src,
2341       "reading offset %" G_GUINT64_FORMAT ", length %u, size %" G_GINT64_FORMAT
2342       ", segment.stop %" G_GINT64_FORMAT ", maxsize %" G_GINT64_FORMAT, offset,
2343       *length, size, stop, maxsize);
2344
2345   /* check size if we have one */
2346   if (maxsize != -1) {
2347     /* if we run past the end, check if the file became bigger and
2348      * retry.  Mind wrap when checking. */
2349     if (G_UNLIKELY (offset >= maxsize || offset + *length >= maxsize || force)) {
2350       /* see if length of the file changed */
2351       if (bclass->get_size)
2352         if (!bclass->get_size (src, &size))
2353           size = -1;
2354
2355       /* when not doing automatic EOS, just use the stop position. We don't use
2356        * the size to check for EOS */
2357       if (!g_atomic_int_get (&src->priv->automatic_eos))
2358         maxsize = stop;
2359       /* Otherwise, the max amount of bytes to read is the total
2360        * size or up to the segment.stop if present. */
2361       else if (stop != -1)
2362         maxsize = size != -1 ? MIN (size, stop) : stop;
2363       else
2364         maxsize = size;
2365
2366       if (maxsize != -1) {
2367         /* if we are at or past the end, EOS */
2368         if (G_UNLIKELY (offset >= maxsize))
2369           goto unexpected_length;
2370
2371         /* else we can clip to the end */
2372         if (G_UNLIKELY (offset + *length >= maxsize))
2373           *length = maxsize - offset;
2374       }
2375     }
2376   }
2377
2378   /* keep track of current duration. segment is in bytes, we checked
2379    * that above. */
2380   GST_OBJECT_LOCK (src);
2381   src->segment.duration = size;
2382   GST_OBJECT_UNLOCK (src);
2383
2384   return TRUE;
2385
2386   /* ERRORS */
2387 unexpected_length:
2388   {
2389     GST_WARNING_OBJECT (src, "processing at or past EOS");
2390     return FALSE;
2391   }
2392 }
2393
2394 /* must be called with LIVE_LOCK */
2395 static GstFlowReturn
2396 gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length,
2397     GstBuffer ** buf)
2398 {
2399   GstFlowReturn ret;
2400   GstBaseSrcClass *bclass;
2401   GstClockReturn status;
2402   GstBuffer *res_buf;
2403   GstBuffer *in_buf;
2404
2405   bclass = GST_BASE_SRC_GET_CLASS (src);
2406
2407 again:
2408   if (src->is_live) {
2409     if (G_UNLIKELY (!src->live_running)) {
2410       ret = gst_base_src_wait_playing (src);
2411       if (ret != GST_FLOW_OK)
2412         goto stopped;
2413     }
2414   }
2415
2416   if (G_UNLIKELY (!GST_BASE_SRC_IS_STARTED (src)
2417           && !GST_BASE_SRC_IS_STARTING (src)))
2418     goto not_started;
2419
2420   if (G_UNLIKELY (!bclass->create))
2421     goto no_function;
2422
2423   if (G_UNLIKELY (!gst_base_src_update_length (src, offset, &length, FALSE)))
2424     goto unexpected_length;
2425
2426   /* track position */
2427   GST_OBJECT_LOCK (src);
2428   if (src->segment.format == GST_FORMAT_BYTES)
2429     src->segment.position = offset;
2430   GST_OBJECT_UNLOCK (src);
2431
2432   /* normally we don't count buffers */
2433   if (G_UNLIKELY (src->num_buffers_left >= 0)) {
2434     if (src->num_buffers_left == 0)
2435       goto reached_num_buffers;
2436     else
2437       src->num_buffers_left--;
2438   }
2439
2440   /* don't enter the create function if a pending EOS event was set. For the
2441    * logic of the has_pending_eos, check the event function of this class. */
2442   if (G_UNLIKELY (g_atomic_int_get (&src->priv->has_pending_eos))) {
2443     src->priv->forced_eos = TRUE;
2444     goto eos;
2445   }
2446
2447   GST_DEBUG_OBJECT (src,
2448       "calling create offset %" G_GUINT64_FORMAT " length %u, time %"
2449       G_GINT64_FORMAT, offset, length, src->segment.time);
2450
2451   res_buf = in_buf = *buf;
2452
2453   ret = bclass->create (src, offset, length, &res_buf);
2454
2455   /* The create function could be unlocked because we have a pending EOS. It's
2456    * possible that we have a valid buffer from create that we need to
2457    * discard when the create function returned _OK. */
2458   if (G_UNLIKELY (g_atomic_int_get (&src->priv->has_pending_eos))) {
2459     if (ret == GST_FLOW_OK) {
2460       if (*buf == NULL)
2461         gst_buffer_unref (res_buf);
2462     }
2463     src->priv->forced_eos = TRUE;
2464     goto eos;
2465   }
2466
2467   if (G_UNLIKELY (ret != GST_FLOW_OK))
2468     goto not_ok;
2469
2470   /* fallback in case the create function didn't fill a provided buffer */
2471   if (in_buf != NULL && res_buf != in_buf) {
2472     GstMapInfo info;
2473     gsize copied_size;
2474
2475     GST_CAT_DEBUG_OBJECT (GST_CAT_PERFORMANCE, src, "create function didn't "
2476         "fill the provided buffer, copying");
2477
2478     if (!gst_buffer_map (in_buf, &info, GST_MAP_WRITE))
2479       goto map_failed;
2480
2481     copied_size = gst_buffer_extract (res_buf, 0, info.data, info.size);
2482     gst_buffer_unmap (in_buf, &info);
2483     gst_buffer_set_size (in_buf, copied_size);
2484
2485     gst_buffer_copy_into (in_buf, res_buf, GST_BUFFER_COPY_METADATA, 0, -1);
2486
2487     gst_buffer_unref (res_buf);
2488     res_buf = in_buf;
2489   }
2490
2491   /* no timestamp set and we are at offset 0, we can timestamp with 0 */
2492   if (offset == 0 && src->segment.time == 0
2493       && GST_BUFFER_DTS (res_buf) == -1 && !src->is_live) {
2494     GST_DEBUG_OBJECT (src, "setting first timestamp to 0");
2495     res_buf = gst_buffer_make_writable (res_buf);
2496     GST_BUFFER_DTS (res_buf) = 0;
2497   }
2498
2499   /* now sync before pushing the buffer */
2500   status = gst_base_src_do_sync (src, res_buf);
2501
2502   /* waiting for the clock could have made us flushing */
2503   if (G_UNLIKELY (src->priv->flushing))
2504     goto flushing;
2505
2506   switch (status) {
2507     case GST_CLOCK_EARLY:
2508       /* the buffer is too late. We currently don't drop the buffer. */
2509       GST_DEBUG_OBJECT (src, "buffer too late!, returning anyway");
2510       break;
2511     case GST_CLOCK_OK:
2512       /* buffer synchronised properly */
2513       GST_DEBUG_OBJECT (src, "buffer ok");
2514       break;
2515     case GST_CLOCK_UNSCHEDULED:
2516       /* this case is triggered when we were waiting for the clock and
2517        * it got unlocked because we did a state change. In any case, get rid of
2518        * the buffer. */
2519       if (*buf == NULL)
2520         gst_buffer_unref (res_buf);
2521
2522       if (!src->live_running) {
2523         /* We return FLUSHING when we are not running to stop the dataflow also
2524          * get rid of the produced buffer. */
2525         GST_DEBUG_OBJECT (src,
2526             "clock was unscheduled (%d), returning FLUSHING", status);
2527         ret = GST_FLOW_FLUSHING;
2528       } else {
2529         /* If we are running when this happens, we quickly switched between
2530          * pause and playing. We try to produce a new buffer */
2531         GST_DEBUG_OBJECT (src,
2532             "clock was unscheduled (%d), but we are running", status);
2533         goto again;
2534       }
2535       break;
2536     default:
2537       /* all other result values are unexpected and errors */
2538       GST_ELEMENT_ERROR (src, CORE, CLOCK,
2539           (_("Internal clock error.")),
2540           ("clock returned unexpected return value %d", status));
2541       if (*buf == NULL)
2542         gst_buffer_unref (res_buf);
2543       ret = GST_FLOW_ERROR;
2544       break;
2545   }
2546   if (G_LIKELY (ret == GST_FLOW_OK))
2547     *buf = res_buf;
2548
2549   return ret;
2550
2551   /* ERROR */
2552 stopped:
2553   {
2554     GST_DEBUG_OBJECT (src, "wait_playing returned %d (%s)", ret,
2555         gst_flow_get_name (ret));
2556     return ret;
2557   }
2558 not_ok:
2559   {
2560     GST_DEBUG_OBJECT (src, "create returned %d (%s)", ret,
2561         gst_flow_get_name (ret));
2562     return ret;
2563   }
2564 map_failed:
2565   {
2566     GST_ELEMENT_ERROR (src, RESOURCE, BUSY,
2567         (_("Failed to map buffer.")),
2568         ("failed to map result buffer in WRITE mode"));
2569     if (*buf == NULL)
2570       gst_buffer_unref (res_buf);
2571     return GST_FLOW_ERROR;
2572   }
2573 not_started:
2574   {
2575     GST_DEBUG_OBJECT (src, "getrange but not started");
2576     return GST_FLOW_FLUSHING;
2577   }
2578 no_function:
2579   {
2580     GST_DEBUG_OBJECT (src, "no create function");
2581     return GST_FLOW_NOT_SUPPORTED;
2582   }
2583 unexpected_length:
2584   {
2585     GST_DEBUG_OBJECT (src, "unexpected length %u (offset=%" G_GUINT64_FORMAT
2586         ", size=%" G_GINT64_FORMAT ")", length, offset, src->segment.duration);
2587     return GST_FLOW_EOS;
2588   }
2589 reached_num_buffers:
2590   {
2591     GST_DEBUG_OBJECT (src, "sent all buffers");
2592     return GST_FLOW_EOS;
2593   }
2594 flushing:
2595   {
2596     GST_DEBUG_OBJECT (src, "we are flushing");
2597     if (*buf == NULL)
2598       gst_buffer_unref (res_buf);
2599     return GST_FLOW_FLUSHING;
2600   }
2601 eos:
2602   {
2603     GST_DEBUG_OBJECT (src, "we are EOS");
2604     return GST_FLOW_EOS;
2605   }
2606 }
2607
2608 static GstFlowReturn
2609 gst_base_src_getrange (GstPad * pad, GstObject * parent, guint64 offset,
2610     guint length, GstBuffer ** buf)
2611 {
2612   GstBaseSrc *src;
2613   GstFlowReturn res;
2614
2615   src = GST_BASE_SRC_CAST (parent);
2616
2617   GST_LIVE_LOCK (src);
2618   if (G_UNLIKELY (src->priv->flushing))
2619     goto flushing;
2620
2621   res = gst_base_src_get_range (src, offset, length, buf);
2622
2623 done:
2624   GST_LIVE_UNLOCK (src);
2625
2626   return res;
2627
2628   /* ERRORS */
2629 flushing:
2630   {
2631     GST_DEBUG_OBJECT (src, "we are flushing");
2632     res = GST_FLOW_FLUSHING;
2633     goto done;
2634   }
2635 }
2636
2637 static gboolean
2638 gst_base_src_is_random_access (GstBaseSrc * src)
2639 {
2640   /* we need to start the basesrc to check random access */
2641   if (!GST_BASE_SRC_IS_STARTED (src)) {
2642     GST_LOG_OBJECT (src, "doing start/stop to check get_range support");
2643     if (G_LIKELY (gst_base_src_start (src))) {
2644       if (gst_base_src_start_wait (src) != GST_FLOW_OK)
2645         goto start_failed;
2646       gst_base_src_stop (src);
2647     }
2648   }
2649
2650   return src->random_access;
2651
2652   /* ERRORS */
2653 start_failed:
2654   {
2655     GST_DEBUG_OBJECT (src, "failed to start");
2656     return FALSE;
2657   }
2658 }
2659
2660 static void
2661 gst_base_src_loop (GstPad * pad)
2662 {
2663   GstBaseSrc *src;
2664   GstBuffer *buf = NULL;
2665   GstFlowReturn ret;
2666   gint64 position;
2667   gboolean eos;
2668   guint blocksize;
2669   GList *pending_events = NULL, *tmp;
2670
2671   eos = FALSE;
2672
2673   src = GST_BASE_SRC (GST_OBJECT_PARENT (pad));
2674
2675   /* Just leave immediately if we're flushing */
2676   GST_LIVE_LOCK (src);
2677   if (G_UNLIKELY (src->priv->flushing || GST_PAD_IS_FLUSHING (pad)))
2678     goto flushing;
2679   GST_LIVE_UNLOCK (src);
2680
2681   gst_base_src_send_stream_start (src);
2682
2683   /* The stream-start event could've caused something to flush us */
2684   GST_LIVE_LOCK (src);
2685   if (G_UNLIKELY (src->priv->flushing || GST_PAD_IS_FLUSHING (pad)))
2686     goto flushing;
2687   GST_LIVE_UNLOCK (src);
2688
2689   /* check if we need to renegotiate */
2690   if (gst_pad_check_reconfigure (pad)) {
2691     if (!gst_base_src_negotiate (src)) {
2692       gst_pad_mark_reconfigure (pad);
2693       if (GST_PAD_IS_FLUSHING (pad)) {
2694         GST_LIVE_LOCK (src);
2695         goto flushing;
2696       } else {
2697         goto negotiate_failed;
2698       }
2699     }
2700   }
2701
2702   GST_LIVE_LOCK (src);
2703
2704   if (G_UNLIKELY (src->priv->flushing || GST_PAD_IS_FLUSHING (pad)))
2705     goto flushing;
2706
2707   blocksize = src->blocksize;
2708
2709   /* if we operate in bytes, we can calculate an offset */
2710   if (src->segment.format == GST_FORMAT_BYTES) {
2711     position = src->segment.position;
2712     /* for negative rates, start with subtracting the blocksize */
2713     if (src->segment.rate < 0.0) {
2714       /* we cannot go below segment.start */
2715       if (position > src->segment.start + blocksize)
2716         position -= blocksize;
2717       else {
2718         /* last block, remainder up to segment.start */
2719         blocksize = position - src->segment.start;
2720         position = src->segment.start;
2721       }
2722     }
2723   } else
2724     position = -1;
2725
2726   GST_LOG_OBJECT (src, "next_ts %" GST_TIME_FORMAT " size %u",
2727       GST_TIME_ARGS (position), blocksize);
2728
2729   ret = gst_base_src_get_range (src, position, blocksize, &buf);
2730   if (G_UNLIKELY (ret != GST_FLOW_OK)) {
2731     GST_INFO_OBJECT (src, "pausing after gst_base_src_get_range() = %s",
2732         gst_flow_get_name (ret));
2733     GST_LIVE_UNLOCK (src);
2734     goto pause;
2735   }
2736   /* this should not happen */
2737   if (G_UNLIKELY (buf == NULL))
2738     goto null_buffer;
2739
2740   /* push events to close/start our segment before we push the buffer. */
2741   if (G_UNLIKELY (src->priv->segment_pending)) {
2742     GstEvent *seg_event = gst_event_new_segment (&src->segment);
2743
2744     gst_event_set_seqnum (seg_event, src->priv->segment_seqnum);
2745     src->priv->segment_seqnum = gst_util_seqnum_next ();
2746     gst_pad_push_event (pad, seg_event);
2747     src->priv->segment_pending = FALSE;
2748   }
2749
2750   if (g_atomic_int_get (&src->priv->have_events)) {
2751     GST_OBJECT_LOCK (src);
2752     /* take the events */
2753     pending_events = src->priv->pending_events;
2754     src->priv->pending_events = NULL;
2755     g_atomic_int_set (&src->priv->have_events, FALSE);
2756     GST_OBJECT_UNLOCK (src);
2757   }
2758
2759   /* Push out pending events if any */
2760   if (G_UNLIKELY (pending_events != NULL)) {
2761     for (tmp = pending_events; tmp; tmp = g_list_next (tmp)) {
2762       GstEvent *ev = (GstEvent *) tmp->data;
2763       gst_pad_push_event (pad, ev);
2764     }
2765     g_list_free (pending_events);
2766   }
2767
2768   /* figure out the new position */
2769   switch (src->segment.format) {
2770     case GST_FORMAT_BYTES:
2771     {
2772       guint bufsize = gst_buffer_get_size (buf);
2773
2774       /* we subtracted above for negative rates */
2775       if (src->segment.rate >= 0.0)
2776         position += bufsize;
2777       break;
2778     }
2779     case GST_FORMAT_TIME:
2780     {
2781       GstClockTime start, duration;
2782
2783       start = GST_BUFFER_TIMESTAMP (buf);
2784       duration = GST_BUFFER_DURATION (buf);
2785
2786       if (GST_CLOCK_TIME_IS_VALID (start))
2787         position = start;
2788       else
2789         position = src->segment.position;
2790
2791       if (GST_CLOCK_TIME_IS_VALID (duration)) {
2792         if (src->segment.rate >= 0.0)
2793           position += duration;
2794         else if (position > duration)
2795           position -= duration;
2796         else
2797           position = 0;
2798       }
2799       break;
2800     }
2801     case GST_FORMAT_DEFAULT:
2802       if (src->segment.rate >= 0.0)
2803         position = GST_BUFFER_OFFSET_END (buf);
2804       else
2805         position = GST_BUFFER_OFFSET (buf);
2806       break;
2807     default:
2808       position = -1;
2809       break;
2810   }
2811   if (position != -1) {
2812     if (src->segment.rate >= 0.0) {
2813       /* positive rate, check if we reached the stop */
2814       if (src->segment.stop != -1) {
2815         if (position >= src->segment.stop) {
2816           eos = TRUE;
2817           position = src->segment.stop;
2818         }
2819       }
2820     } else {
2821       /* negative rate, check if we reached the start. start is always set to
2822        * something different from -1 */
2823       if (position <= src->segment.start) {
2824         eos = TRUE;
2825         position = src->segment.start;
2826       }
2827       /* when going reverse, all buffers are DISCONT */
2828       src->priv->discont = TRUE;
2829     }
2830     GST_OBJECT_LOCK (src);
2831     src->segment.position = position;
2832     GST_OBJECT_UNLOCK (src);
2833   }
2834
2835   if (G_UNLIKELY (src->priv->discont)) {
2836     GST_INFO_OBJECT (src, "marking pending DISCONT");
2837     buf = gst_buffer_make_writable (buf);
2838     GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
2839     src->priv->discont = FALSE;
2840   }
2841   GST_LIVE_UNLOCK (src);
2842
2843   ret = gst_pad_push (pad, buf);
2844   if (G_UNLIKELY (ret != GST_FLOW_OK)) {
2845     if (ret == GST_FLOW_NOT_NEGOTIATED) {
2846       goto not_negotiated;
2847     }
2848     GST_INFO_OBJECT (src, "pausing after gst_pad_push() = %s",
2849         gst_flow_get_name (ret));
2850     goto pause;
2851   }
2852
2853   /* Segment pending means that a new segment was configured
2854    * during this loop run */
2855   if (G_UNLIKELY (eos && !src->priv->segment_pending)) {
2856     GST_INFO_OBJECT (src, "pausing after end of segment");
2857     ret = GST_FLOW_EOS;
2858     goto pause;
2859   }
2860
2861 done:
2862   return;
2863
2864   /* special cases */
2865 not_negotiated:
2866   {
2867     if (gst_pad_needs_reconfigure (pad)) {
2868       GST_DEBUG_OBJECT (src, "Retrying to renegotiate");
2869       return;
2870     }
2871     /* fallthrough when push returns NOT_NEGOTIATED and we don't have
2872      * a pending negotiation request on our srcpad */
2873   }
2874 negotiate_failed:
2875   {
2876     GST_DEBUG_OBJECT (src, "Not negotiated");
2877     ret = GST_FLOW_NOT_NEGOTIATED;
2878     goto pause;
2879   }
2880 flushing:
2881   {
2882     GST_DEBUG_OBJECT (src, "we are flushing");
2883     GST_LIVE_UNLOCK (src);
2884     ret = GST_FLOW_FLUSHING;
2885     goto pause;
2886   }
2887 pause:
2888   {
2889     const gchar *reason = gst_flow_get_name (ret);
2890     GstEvent *event;
2891
2892     GST_DEBUG_OBJECT (src, "pausing task, reason %s", reason);
2893     src->running = FALSE;
2894     gst_pad_pause_task (pad);
2895     if (ret == GST_FLOW_EOS) {
2896       gboolean flag_segment;
2897       GstFormat format;
2898       gint64 position;
2899
2900       flag_segment = (src->segment.flags & GST_SEGMENT_FLAG_SEGMENT) != 0;
2901       format = src->segment.format;
2902       position = src->segment.position;
2903
2904       /* perform EOS logic */
2905       if (src->priv->forced_eos) {
2906         g_assert (g_atomic_int_get (&src->priv->has_pending_eos));
2907         GST_OBJECT_LOCK (src);
2908         event = src->priv->pending_eos;
2909         src->priv->pending_eos = NULL;
2910         GST_OBJECT_UNLOCK (src);
2911
2912       } else if (flag_segment) {
2913         GstMessage *message;
2914
2915         message = gst_message_new_segment_done (GST_OBJECT_CAST (src),
2916             format, position);
2917         gst_message_set_seqnum (message, src->priv->seqnum);
2918         gst_element_post_message (GST_ELEMENT_CAST (src), message);
2919         event = gst_event_new_segment_done (format, position);
2920         gst_event_set_seqnum (event, src->priv->seqnum);
2921
2922       } else {
2923         event = gst_event_new_eos ();
2924         gst_event_set_seqnum (event, src->priv->seqnum);
2925       }
2926
2927       gst_pad_push_event (pad, event);
2928       src->priv->forced_eos = FALSE;
2929
2930     } else if (ret == GST_FLOW_NOT_LINKED || ret <= GST_FLOW_EOS) {
2931       event = gst_event_new_eos ();
2932       gst_event_set_seqnum (event, src->priv->seqnum);
2933       /* for fatal errors we post an error message, post the error
2934        * first so the app knows about the error first.
2935        * Also don't do this for FLUSHING because it happens
2936        * due to flushing and posting an error message because of
2937        * that is the wrong thing to do, e.g. when we're doing
2938        * a flushing seek. */
2939       GST_ELEMENT_FLOW_ERROR (src, ret);
2940       gst_pad_push_event (pad, event);
2941     }
2942     goto done;
2943   }
2944 null_buffer:
2945   {
2946     GST_ELEMENT_ERROR (src, STREAM, FAILED,
2947         (_("Internal data flow error.")), ("element returned NULL buffer"));
2948     GST_LIVE_UNLOCK (src);
2949     goto done;
2950   }
2951 }
2952
2953 static gboolean
2954 gst_base_src_set_allocation (GstBaseSrc * basesrc, GstBufferPool * pool,
2955     GstAllocator * allocator, GstAllocationParams * params)
2956 {
2957   GstAllocator *oldalloc;
2958   GstBufferPool *oldpool;
2959   GstBaseSrcPrivate *priv = basesrc->priv;
2960
2961   if (pool) {
2962     GST_DEBUG_OBJECT (basesrc, "activate pool");
2963     if (!gst_buffer_pool_set_active (pool, TRUE))
2964       goto activate_failed;
2965   }
2966
2967   GST_OBJECT_LOCK (basesrc);
2968   oldpool = priv->pool;
2969   priv->pool = pool;
2970
2971   oldalloc = priv->allocator;
2972   priv->allocator = allocator;
2973
2974   if (priv->pool)
2975     gst_object_ref (priv->pool);
2976   if (priv->allocator)
2977     gst_object_ref (priv->allocator);
2978
2979   if (params)
2980     priv->params = *params;
2981   else
2982     gst_allocation_params_init (&priv->params);
2983   GST_OBJECT_UNLOCK (basesrc);
2984
2985   if (oldpool) {
2986     /* only deactivate if the pool is not the one we're using */
2987     if (oldpool != pool) {
2988       GST_DEBUG_OBJECT (basesrc, "deactivate old pool");
2989       gst_buffer_pool_set_active (oldpool, FALSE);
2990     }
2991     gst_object_unref (oldpool);
2992   }
2993   if (oldalloc) {
2994     gst_object_unref (oldalloc);
2995   }
2996   return TRUE;
2997
2998   /* ERRORS */
2999 activate_failed:
3000   {
3001     GST_ERROR_OBJECT (basesrc, "failed to activate bufferpool.");
3002     return FALSE;
3003   }
3004 }
3005
3006 static gboolean
3007 gst_base_src_activate_pool (GstBaseSrc * basesrc, gboolean active)
3008 {
3009   GstBaseSrcPrivate *priv = basesrc->priv;
3010   GstBufferPool *pool;
3011   gboolean res = TRUE;
3012
3013   GST_OBJECT_LOCK (basesrc);
3014   if ((pool = priv->pool))
3015     pool = gst_object_ref (pool);
3016   GST_OBJECT_UNLOCK (basesrc);
3017
3018   if (pool) {
3019     res = gst_buffer_pool_set_active (pool, active);
3020     gst_object_unref (pool);
3021   }
3022   return res;
3023 }
3024
3025
3026 static gboolean
3027 gst_base_src_decide_allocation_default (GstBaseSrc * basesrc, GstQuery * query)
3028 {
3029   GstCaps *outcaps;
3030   GstBufferPool *pool;
3031   guint size, min, max;
3032   GstAllocator *allocator;
3033   GstAllocationParams params;
3034   GstStructure *config;
3035   gboolean update_allocator;
3036
3037   gst_query_parse_allocation (query, &outcaps, NULL);
3038
3039   /* we got configuration from our peer or the decide_allocation method,
3040    * parse them */
3041   if (gst_query_get_n_allocation_params (query) > 0) {
3042     /* try the allocator */
3043     gst_query_parse_nth_allocation_param (query, 0, &allocator, &params);
3044     update_allocator = TRUE;
3045   } else {
3046     allocator = NULL;
3047     gst_allocation_params_init (&params);
3048     update_allocator = FALSE;
3049   }
3050
3051   if (gst_query_get_n_allocation_pools (query) > 0) {
3052     gst_query_parse_nth_allocation_pool (query, 0, &pool, &size, &min, &max);
3053
3054     if (pool == NULL) {
3055       /* no pool, we can make our own */
3056       GST_DEBUG_OBJECT (basesrc, "no pool, making new pool");
3057       pool = gst_buffer_pool_new ();
3058     }
3059   } else {
3060     pool = NULL;
3061     size = min = max = 0;
3062   }
3063
3064   /* now configure */
3065   if (pool) {
3066     config = gst_buffer_pool_get_config (pool);
3067     gst_buffer_pool_config_set_params (config, outcaps, size, min, max);
3068     gst_buffer_pool_config_set_allocator (config, allocator, &params);
3069
3070     /* buffer pool may have to do some changes */
3071     if (!gst_buffer_pool_set_config (pool, config)) {
3072       config = gst_buffer_pool_get_config (pool);
3073
3074       /* If change are not acceptable, fallback to generic pool */
3075       if (!gst_buffer_pool_config_validate_params (config, outcaps, size, min,
3076               max)) {
3077         GST_DEBUG_OBJECT (basesrc, "unsupported pool, making new pool");
3078
3079         gst_object_unref (pool);
3080         pool = gst_buffer_pool_new ();
3081         gst_buffer_pool_config_set_params (config, outcaps, size, min, max);
3082         gst_buffer_pool_config_set_allocator (config, allocator, &params);
3083       }
3084
3085       if (!gst_buffer_pool_set_config (pool, config))
3086         goto config_failed;
3087     }
3088   }
3089
3090   if (update_allocator)
3091     gst_query_set_nth_allocation_param (query, 0, allocator, &params);
3092   else
3093     gst_query_add_allocation_param (query, allocator, &params);
3094   if (allocator)
3095     gst_object_unref (allocator);
3096
3097   if (pool) {
3098     gst_query_set_nth_allocation_pool (query, 0, pool, size, min, max);
3099     gst_object_unref (pool);
3100   }
3101
3102   return TRUE;
3103
3104 config_failed:
3105   GST_ELEMENT_ERROR (basesrc, RESOURCE, SETTINGS,
3106       ("Failed to configure the buffer pool"),
3107       ("Configuration is most likely invalid, please report this issue."));
3108   gst_object_unref (pool);
3109   return FALSE;
3110 }
3111
3112 static gboolean
3113 gst_base_src_prepare_allocation (GstBaseSrc * basesrc, GstCaps * caps)
3114 {
3115   GstBaseSrcClass *bclass;
3116   gboolean result = TRUE;
3117   GstQuery *query;
3118   GstBufferPool *pool = NULL;
3119   GstAllocator *allocator = NULL;
3120   GstAllocationParams params;
3121
3122   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3123
3124   /* make query and let peer pad answer, we don't really care if it worked or
3125    * not, if it failed, the allocation query would contain defaults and the
3126    * subclass would then set better values if needed */
3127   query = gst_query_new_allocation (caps, TRUE);
3128   if (!gst_pad_peer_query (basesrc->srcpad, query)) {
3129     /* not a problem, just debug a little */
3130     GST_DEBUG_OBJECT (basesrc, "peer ALLOCATION query failed");
3131   }
3132
3133   g_assert (bclass->decide_allocation != NULL);
3134   result = bclass->decide_allocation (basesrc, query);
3135
3136   GST_DEBUG_OBJECT (basesrc, "ALLOCATION (%d) params: %" GST_PTR_FORMAT, result,
3137       query);
3138
3139   if (!result)
3140     goto no_decide_allocation;
3141
3142   /* we got configuration from our peer or the decide_allocation method,
3143    * parse them */
3144   if (gst_query_get_n_allocation_params (query) > 0) {
3145     gst_query_parse_nth_allocation_param (query, 0, &allocator, &params);
3146   } else {
3147     allocator = NULL;
3148     gst_allocation_params_init (&params);
3149   }
3150
3151   if (gst_query_get_n_allocation_pools (query) > 0)
3152     gst_query_parse_nth_allocation_pool (query, 0, &pool, NULL, NULL, NULL);
3153
3154   result = gst_base_src_set_allocation (basesrc, pool, allocator, &params);
3155
3156   if (allocator)
3157     gst_object_unref (allocator);
3158   if (pool)
3159     gst_object_unref (pool);
3160
3161   gst_query_unref (query);
3162
3163   return result;
3164
3165   /* Errors */
3166 no_decide_allocation:
3167   {
3168     GST_WARNING_OBJECT (basesrc, "Subclass failed to decide allocation");
3169     gst_query_unref (query);
3170
3171     return result;
3172   }
3173 }
3174
3175 /* default negotiation code.
3176  *
3177  * Take intersection between src and sink pads, take first
3178  * caps and fixate.
3179  */
3180 static gboolean
3181 gst_base_src_default_negotiate (GstBaseSrc * basesrc)
3182 {
3183   GstCaps *thiscaps;
3184   GstCaps *caps = NULL;
3185   GstCaps *peercaps = NULL;
3186   gboolean result = FALSE;
3187
3188   /* first see what is possible on our source pad */
3189   thiscaps = gst_pad_query_caps (GST_BASE_SRC_PAD (basesrc), NULL);
3190   GST_DEBUG_OBJECT (basesrc, "caps of src: %" GST_PTR_FORMAT, thiscaps);
3191   /* nothing or anything is allowed, we're done */
3192   if (thiscaps == NULL || gst_caps_is_any (thiscaps))
3193     goto no_nego_needed;
3194
3195   if (G_UNLIKELY (gst_caps_is_empty (thiscaps)))
3196     goto no_caps;
3197
3198   /* get the peer caps */
3199   peercaps = gst_pad_peer_query_caps (GST_BASE_SRC_PAD (basesrc), thiscaps);
3200   GST_DEBUG_OBJECT (basesrc, "caps of peer: %" GST_PTR_FORMAT, peercaps);
3201   if (peercaps) {
3202     /* The result is already a subset of our caps */
3203     caps = peercaps;
3204     gst_caps_unref (thiscaps);
3205   } else {
3206     /* no peer, work with our own caps then */
3207     caps = thiscaps;
3208   }
3209   if (caps && !gst_caps_is_empty (caps)) {
3210     /* now fixate */
3211     GST_DEBUG_OBJECT (basesrc, "have caps: %" GST_PTR_FORMAT, caps);
3212     if (gst_caps_is_any (caps)) {
3213       GST_DEBUG_OBJECT (basesrc, "any caps, we stop");
3214       /* hmm, still anything, so element can do anything and
3215        * nego is not needed */
3216       result = TRUE;
3217     } else {
3218       caps = gst_base_src_fixate (basesrc, caps);
3219       GST_DEBUG_OBJECT (basesrc, "fixated to: %" GST_PTR_FORMAT, caps);
3220       if (gst_caps_is_fixed (caps)) {
3221         /* yay, fixed caps, use those then, it's possible that the subclass does
3222          * not accept this caps after all and we have to fail. */
3223         result = gst_base_src_set_caps (basesrc, caps);
3224       }
3225     }
3226     gst_caps_unref (caps);
3227   } else {
3228     if (caps)
3229       gst_caps_unref (caps);
3230     GST_DEBUG_OBJECT (basesrc, "no common caps");
3231   }
3232   return result;
3233
3234 no_nego_needed:
3235   {
3236     GST_DEBUG_OBJECT (basesrc, "no negotiation needed");
3237     if (thiscaps)
3238       gst_caps_unref (thiscaps);
3239     return TRUE;
3240   }
3241 no_caps:
3242   {
3243     GST_ELEMENT_ERROR (basesrc, STREAM, FORMAT,
3244         ("No supported formats found"),
3245         ("This element did not produce valid caps"));
3246     if (thiscaps)
3247       gst_caps_unref (thiscaps);
3248     return TRUE;
3249   }
3250 }
3251
3252 static gboolean
3253 gst_base_src_negotiate (GstBaseSrc * basesrc)
3254 {
3255   GstBaseSrcClass *bclass;
3256   gboolean result;
3257
3258   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3259
3260   GST_DEBUG_OBJECT (basesrc, "starting negotiation");
3261
3262   if (G_LIKELY (bclass->negotiate))
3263     result = bclass->negotiate (basesrc);
3264   else
3265     result = TRUE;
3266
3267   if (G_LIKELY (result)) {
3268     GstCaps *caps;
3269
3270     caps = gst_pad_get_current_caps (basesrc->srcpad);
3271
3272     result = gst_base_src_prepare_allocation (basesrc, caps);
3273
3274     if (caps)
3275       gst_caps_unref (caps);
3276   }
3277   return result;
3278 }
3279
3280 static gboolean
3281 gst_base_src_start (GstBaseSrc * basesrc)
3282 {
3283   GstBaseSrcClass *bclass;
3284   gboolean result;
3285
3286   GST_LIVE_LOCK (basesrc);
3287
3288   GST_OBJECT_LOCK (basesrc);
3289   if (GST_BASE_SRC_IS_STARTING (basesrc))
3290     goto was_starting;
3291   if (GST_BASE_SRC_IS_STARTED (basesrc))
3292     goto was_started;
3293
3294   basesrc->priv->start_result = GST_FLOW_FLUSHING;
3295   GST_OBJECT_FLAG_SET (basesrc, GST_BASE_SRC_FLAG_STARTING);
3296   gst_segment_init (&basesrc->segment, basesrc->segment.format);
3297   GST_OBJECT_UNLOCK (basesrc);
3298
3299   basesrc->num_buffers_left = basesrc->num_buffers;
3300   basesrc->running = FALSE;
3301   basesrc->priv->segment_pending = FALSE;
3302   basesrc->priv->segment_seqnum = gst_util_seqnum_next ();
3303   basesrc->priv->forced_eos = FALSE;
3304   GST_LIVE_UNLOCK (basesrc);
3305
3306   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3307   if (bclass->start)
3308     result = bclass->start (basesrc);
3309   else
3310     result = TRUE;
3311
3312   if (!result)
3313     goto could_not_start;
3314
3315   if (!gst_base_src_is_async (basesrc)) {
3316     gst_base_src_start_complete (basesrc, GST_FLOW_OK);
3317     /* not really waiting here, we call this to get the result
3318      * from the start_complete call */
3319     result = gst_base_src_start_wait (basesrc) == GST_FLOW_OK;
3320   }
3321
3322   return result;
3323
3324   /* ERROR */
3325 was_starting:
3326   {
3327     GST_DEBUG_OBJECT (basesrc, "was starting");
3328     GST_OBJECT_UNLOCK (basesrc);
3329     GST_LIVE_UNLOCK (basesrc);
3330     return TRUE;
3331   }
3332 was_started:
3333   {
3334     GST_DEBUG_OBJECT (basesrc, "was started");
3335     GST_OBJECT_UNLOCK (basesrc);
3336     GST_LIVE_UNLOCK (basesrc);
3337     return TRUE;
3338   }
3339 could_not_start:
3340   {
3341     GST_DEBUG_OBJECT (basesrc, "could not start");
3342     /* subclass is supposed to post a message but we post one as a fallback
3343      * just in case. We don't have to call _stop. */
3344     GST_ELEMENT_ERROR (basesrc, CORE, STATE_CHANGE, (NULL),
3345         ("Failed to start"));
3346     gst_base_src_start_complete (basesrc, GST_FLOW_ERROR);
3347     return FALSE;
3348   }
3349 }
3350
3351 /**
3352  * gst_base_src_start_complete:
3353  * @basesrc: base source instance
3354  * @ret: a #GstFlowReturn
3355  *
3356  * Complete an asynchronous start operation. When the subclass overrides the
3357  * start method, it should call gst_base_src_start_complete() when the start
3358  * operation completes either from the same thread or from an asynchronous
3359  * helper thread.
3360  */
3361 void
3362 gst_base_src_start_complete (GstBaseSrc * basesrc, GstFlowReturn ret)
3363 {
3364   gboolean have_size;
3365   guint64 size;
3366   gboolean seekable;
3367   GstFormat format;
3368   GstPadMode mode;
3369   GstEvent *event;
3370
3371   if (ret != GST_FLOW_OK)
3372     goto error;
3373
3374   GST_DEBUG_OBJECT (basesrc, "starting source");
3375   format = basesrc->segment.format;
3376
3377   /* figure out the size */
3378   have_size = FALSE;
3379   size = -1;
3380   if (format == GST_FORMAT_BYTES) {
3381     GstBaseSrcClass *bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3382
3383     if (bclass->get_size) {
3384       if (!(have_size = bclass->get_size (basesrc, &size)))
3385         size = -1;
3386     }
3387     GST_DEBUG_OBJECT (basesrc, "setting size %" G_GUINT64_FORMAT, size);
3388     /* only update the size when operating in bytes, subclass is supposed
3389      * to set duration in the start method for other formats */
3390     GST_OBJECT_LOCK (basesrc);
3391     basesrc->segment.duration = size;
3392     GST_OBJECT_UNLOCK (basesrc);
3393   }
3394
3395   GST_DEBUG_OBJECT (basesrc,
3396       "format: %s, have size: %d, size: %" G_GUINT64_FORMAT ", duration: %"
3397       G_GINT64_FORMAT, gst_format_get_name (format), have_size, size,
3398       basesrc->segment.duration);
3399
3400   seekable = gst_base_src_seekable (basesrc);
3401   GST_DEBUG_OBJECT (basesrc, "is seekable: %d", seekable);
3402
3403   /* update for random access flag */
3404   basesrc->random_access = seekable && format == GST_FORMAT_BYTES;
3405
3406   GST_DEBUG_OBJECT (basesrc, "is random_access: %d", basesrc->random_access);
3407
3408   /* stop flushing now but for live sources, still block in the LIVE lock when
3409    * we are not yet PLAYING */
3410   gst_base_src_set_flushing (basesrc, FALSE, FALSE, NULL);
3411
3412   gst_pad_mark_reconfigure (GST_BASE_SRC_PAD (basesrc));
3413
3414   GST_OBJECT_LOCK (basesrc->srcpad);
3415   mode = GST_PAD_MODE (basesrc->srcpad);
3416   GST_OBJECT_UNLOCK (basesrc->srcpad);
3417
3418   /* take the stream lock here, we only want to let the task run when we have
3419    * set the STARTED flag */
3420   GST_PAD_STREAM_LOCK (basesrc->srcpad);
3421   switch (mode) {
3422     case GST_PAD_MODE_PUSH:
3423       /* do initial seek, which will start the task */
3424       GST_OBJECT_LOCK (basesrc);
3425       event = basesrc->pending_seek;
3426       basesrc->pending_seek = NULL;
3427       GST_OBJECT_UNLOCK (basesrc);
3428
3429       /* The perform seek code will start the task when finished. We don't have to
3430        * unlock the streaming thread because it is not running yet */
3431       if (G_UNLIKELY (!gst_base_src_perform_seek (basesrc, event, FALSE)))
3432         goto seek_failed;
3433
3434       if (event)
3435         gst_event_unref (event);
3436       break;
3437     case GST_PAD_MODE_PULL:
3438       /* if not random_access, we cannot operate in pull mode for now */
3439       if (G_UNLIKELY (!basesrc->random_access))
3440         goto no_get_range;
3441       break;
3442     default:
3443       goto not_activated_yet;
3444       break;
3445   }
3446
3447   GST_OBJECT_LOCK (basesrc);
3448   GST_OBJECT_FLAG_SET (basesrc, GST_BASE_SRC_FLAG_STARTED);
3449   GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTING);
3450   basesrc->priv->start_result = ret;
3451   GST_ASYNC_SIGNAL (basesrc);
3452   GST_OBJECT_UNLOCK (basesrc);
3453
3454   GST_PAD_STREAM_UNLOCK (basesrc->srcpad);
3455
3456   return;
3457
3458 seek_failed:
3459   {
3460     GST_PAD_STREAM_UNLOCK (basesrc->srcpad);
3461     GST_ERROR_OBJECT (basesrc, "Failed to perform initial seek");
3462     gst_base_src_stop (basesrc);
3463     if (event)
3464       gst_event_unref (event);
3465     ret = GST_FLOW_ERROR;
3466     goto error;
3467   }
3468 no_get_range:
3469   {
3470     GST_PAD_STREAM_UNLOCK (basesrc->srcpad);
3471     gst_base_src_stop (basesrc);
3472     GST_ERROR_OBJECT (basesrc, "Cannot operate in pull mode, stopping");
3473     ret = GST_FLOW_ERROR;
3474     goto error;
3475   }
3476 not_activated_yet:
3477   {
3478     GST_PAD_STREAM_UNLOCK (basesrc->srcpad);
3479     gst_base_src_stop (basesrc);
3480     GST_WARNING_OBJECT (basesrc, "pad not activated yet");
3481     ret = GST_FLOW_ERROR;
3482     goto error;
3483   }
3484 error:
3485   {
3486     GST_OBJECT_LOCK (basesrc);
3487     basesrc->priv->start_result = ret;
3488     GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTING);
3489     GST_ASYNC_SIGNAL (basesrc);
3490     GST_OBJECT_UNLOCK (basesrc);
3491     return;
3492   }
3493 }
3494
3495 /**
3496  * gst_base_src_start_wait:
3497  * @basesrc: base source instance
3498  *
3499  * Wait until the start operation completes.
3500  *
3501  * Returns: a #GstFlowReturn.
3502  */
3503 GstFlowReturn
3504 gst_base_src_start_wait (GstBaseSrc * basesrc)
3505 {
3506   GstFlowReturn result;
3507
3508   GST_OBJECT_LOCK (basesrc);
3509   while (GST_BASE_SRC_IS_STARTING (basesrc)) {
3510     GST_ASYNC_WAIT (basesrc);
3511   }
3512   result = basesrc->priv->start_result;
3513   GST_OBJECT_UNLOCK (basesrc);
3514
3515   GST_DEBUG_OBJECT (basesrc, "got %s", gst_flow_get_name (result));
3516
3517   return result;
3518 }
3519
3520 static gboolean
3521 gst_base_src_stop (GstBaseSrc * basesrc)
3522 {
3523   GstBaseSrcClass *bclass;
3524   gboolean result = TRUE;
3525
3526   GST_DEBUG_OBJECT (basesrc, "stopping source");
3527
3528   /* flush all */
3529   gst_base_src_set_flushing (basesrc, TRUE, FALSE, NULL);
3530   /* stop the task */
3531   gst_pad_stop_task (basesrc->srcpad);
3532
3533   GST_OBJECT_LOCK (basesrc);
3534   if (!GST_BASE_SRC_IS_STARTED (basesrc) && !GST_BASE_SRC_IS_STARTING (basesrc))
3535     goto was_stopped;
3536
3537   GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTING);
3538   GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_FLAG_STARTED);
3539   basesrc->priv->start_result = GST_FLOW_FLUSHING;
3540   GST_ASYNC_SIGNAL (basesrc);
3541   GST_OBJECT_UNLOCK (basesrc);
3542
3543   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3544   if (bclass->stop)
3545     result = bclass->stop (basesrc);
3546
3547   gst_base_src_set_allocation (basesrc, NULL, NULL, NULL);
3548
3549   return result;
3550
3551 was_stopped:
3552   {
3553     GST_DEBUG_OBJECT (basesrc, "was stopped");
3554     GST_OBJECT_UNLOCK (basesrc);
3555     return TRUE;
3556   }
3557 }
3558
3559 /* start or stop flushing dataprocessing
3560  */
3561 static gboolean
3562 gst_base_src_set_flushing (GstBaseSrc * basesrc,
3563     gboolean flushing, gboolean live_play, gboolean * playing)
3564 {
3565   GstBaseSrcClass *bclass;
3566
3567   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3568
3569   GST_DEBUG_OBJECT (basesrc, "flushing %d, live_play %d", flushing, live_play);
3570
3571   if (flushing) {
3572     gst_base_src_activate_pool (basesrc, FALSE);
3573     /* unlock any subclasses, we need to do this before grabbing the
3574      * LIVE_LOCK since we hold this lock before going into ::create. We pass an
3575      * unlock to the params because of backwards compat (see seek handler)*/
3576     if (bclass->unlock)
3577       bclass->unlock (basesrc);
3578   }
3579
3580   /* the live lock is released when we are blocked, waiting for playing or
3581    * when we sync to the clock. */
3582   GST_LIVE_LOCK (basesrc);
3583   if (playing)
3584     *playing = basesrc->live_running;
3585   basesrc->priv->flushing = flushing;
3586   if (flushing) {
3587     /* if we are locked in the live lock, signal it to make it flush */
3588     basesrc->live_running = TRUE;
3589
3590     /* clear pending EOS if any */
3591     if (g_atomic_int_get (&basesrc->priv->has_pending_eos)) {
3592       GST_OBJECT_LOCK (basesrc);
3593       CLEAR_PENDING_EOS (basesrc);
3594       basesrc->priv->forced_eos = FALSE;
3595       GST_OBJECT_UNLOCK (basesrc);
3596     }
3597
3598     /* step 1, now that we have the LIVE lock, clear our unlock request */
3599     if (bclass->unlock_stop)
3600       bclass->unlock_stop (basesrc);
3601
3602     /* step 2, unblock clock sync (if any) or any other blocking thing */
3603     if (basesrc->clock_id)
3604       gst_clock_id_unschedule (basesrc->clock_id);
3605   } else {
3606     /* signal the live source that it can start playing */
3607     basesrc->live_running = live_play;
3608
3609     gst_base_src_activate_pool (basesrc, TRUE);
3610
3611     /* Drop all delayed events */
3612     GST_OBJECT_LOCK (basesrc);
3613     if (basesrc->priv->pending_events) {
3614       g_list_foreach (basesrc->priv->pending_events, (GFunc) gst_event_unref,
3615           NULL);
3616       g_list_free (basesrc->priv->pending_events);
3617       basesrc->priv->pending_events = NULL;
3618       g_atomic_int_set (&basesrc->priv->have_events, FALSE);
3619     }
3620     GST_OBJECT_UNLOCK (basesrc);
3621   }
3622   GST_LIVE_SIGNAL (basesrc);
3623   GST_LIVE_UNLOCK (basesrc);
3624
3625   return TRUE;
3626 }
3627
3628 /* the purpose of this function is to make sure that a live source blocks in the
3629  * LIVE lock or leaves the LIVE lock and continues playing. */
3630 static gboolean
3631 gst_base_src_set_playing (GstBaseSrc * basesrc, gboolean live_play)
3632 {
3633   GstBaseSrcClass *bclass;
3634
3635   bclass = GST_BASE_SRC_GET_CLASS (basesrc);
3636
3637   /* unlock subclasses locked in ::create, we only do this when we stop playing. */
3638   if (!live_play) {
3639     GST_DEBUG_OBJECT (basesrc, "unlock");
3640     if (bclass->unlock)
3641       bclass->unlock (basesrc);
3642   }
3643
3644   /* we are now able to grab the LIVE lock, when we get it, we can be
3645    * waiting for PLAYING while blocked in the LIVE cond or we can be waiting
3646    * for the clock. */
3647   GST_LIVE_LOCK (basesrc);
3648   GST_DEBUG_OBJECT (basesrc, "unschedule clock");
3649
3650   /* unblock clock sync (if any) */
3651   if (basesrc->clock_id)
3652     gst_clock_id_unschedule (basesrc->clock_id);
3653
3654   /* configure what to do when we get to the LIVE lock. */
3655   GST_DEBUG_OBJECT (basesrc, "live running %d", live_play);
3656   basesrc->live_running = live_play;
3657
3658   if (live_play) {
3659     gboolean start;
3660
3661     /* clear our unlock request when going to PLAYING */
3662     GST_DEBUG_OBJECT (basesrc, "unlock stop");
3663     if (bclass->unlock_stop)
3664       bclass->unlock_stop (basesrc);
3665
3666     /* for live sources we restart the timestamp correction */
3667     basesrc->priv->latency = -1;
3668     /* have to restart the task in case it stopped because of the unlock when
3669      * we went to PAUSED. Only do this if we operating in push mode. */
3670     GST_OBJECT_LOCK (basesrc->srcpad);
3671     start = (GST_PAD_MODE (basesrc->srcpad) == GST_PAD_MODE_PUSH);
3672     GST_OBJECT_UNLOCK (basesrc->srcpad);
3673     if (start)
3674       gst_pad_start_task (basesrc->srcpad, (GstTaskFunction) gst_base_src_loop,
3675           basesrc->srcpad, NULL);
3676     GST_DEBUG_OBJECT (basesrc, "signal");
3677     GST_LIVE_SIGNAL (basesrc);
3678   }
3679   GST_LIVE_UNLOCK (basesrc);
3680
3681   return TRUE;
3682 }
3683
3684 static gboolean
3685 gst_base_src_activate_push (GstPad * pad, GstObject * parent, gboolean active)
3686 {
3687   GstBaseSrc *basesrc;
3688
3689   basesrc = GST_BASE_SRC (parent);
3690
3691   /* prepare subclass first */
3692   if (active) {
3693     GST_DEBUG_OBJECT (basesrc, "Activating in push mode");
3694
3695     if (G_UNLIKELY (!basesrc->can_activate_push))
3696       goto no_push_activation;
3697
3698     if (G_UNLIKELY (!gst_base_src_start (basesrc)))
3699       goto error_start;
3700   } else {
3701     GST_DEBUG_OBJECT (basesrc, "Deactivating in push mode");
3702     /* now we can stop the source */
3703     if (G_UNLIKELY (!gst_base_src_stop (basesrc)))
3704       goto error_stop;
3705   }
3706   return TRUE;
3707
3708   /* ERRORS */
3709 no_push_activation:
3710   {
3711     GST_WARNING_OBJECT (basesrc, "Subclass disabled push-mode activation");
3712     return FALSE;
3713   }
3714 error_start:
3715   {
3716     GST_WARNING_OBJECT (basesrc, "Failed to start in push mode");
3717     return FALSE;
3718   }
3719 error_stop:
3720   {
3721     GST_DEBUG_OBJECT (basesrc, "Failed to stop in push mode");
3722     return FALSE;
3723   }
3724 }
3725
3726 static gboolean
3727 gst_base_src_activate_pull (GstPad * pad, GstObject * parent, gboolean active)
3728 {
3729   GstBaseSrc *basesrc;
3730
3731   basesrc = GST_BASE_SRC (parent);
3732
3733   /* prepare subclass first */
3734   if (active) {
3735     GST_DEBUG_OBJECT (basesrc, "Activating in pull mode");
3736     if (G_UNLIKELY (!gst_base_src_start (basesrc)))
3737       goto error_start;
3738   } else {
3739     GST_DEBUG_OBJECT (basesrc, "Deactivating in pull mode");
3740     if (G_UNLIKELY (!gst_base_src_stop (basesrc)))
3741       goto error_stop;
3742   }
3743   return TRUE;
3744
3745   /* ERRORS */
3746 error_start:
3747   {
3748     GST_ERROR_OBJECT (basesrc, "Failed to start in pull mode");
3749     return FALSE;
3750   }
3751 error_stop:
3752   {
3753     GST_ERROR_OBJECT (basesrc, "Failed to stop in pull mode");
3754     return FALSE;
3755   }
3756 }
3757
3758 static gboolean
3759 gst_base_src_activate_mode (GstPad * pad, GstObject * parent,
3760     GstPadMode mode, gboolean active)
3761 {
3762   gboolean res;
3763   GstBaseSrc *src = GST_BASE_SRC (parent);
3764
3765   src->priv->stream_start_pending = FALSE;
3766
3767   GST_DEBUG_OBJECT (pad, "activating in mode %d", mode);
3768
3769   switch (mode) {
3770     case GST_PAD_MODE_PULL:
3771       res = gst_base_src_activate_pull (pad, parent, active);
3772       break;
3773     case GST_PAD_MODE_PUSH:
3774       src->priv->stream_start_pending = active;
3775       res = gst_base_src_activate_push (pad, parent, active);
3776       break;
3777     default:
3778       GST_LOG_OBJECT (pad, "unknown activation mode %d", mode);
3779       res = FALSE;
3780       break;
3781   }
3782   return res;
3783 }
3784
3785
3786 static GstStateChangeReturn
3787 gst_base_src_change_state (GstElement * element, GstStateChange transition)
3788 {
3789   GstBaseSrc *basesrc;
3790   GstStateChangeReturn result;
3791   gboolean no_preroll = FALSE;
3792
3793   basesrc = GST_BASE_SRC (element);
3794
3795   switch (transition) {
3796     case GST_STATE_CHANGE_NULL_TO_READY:
3797       break;
3798     case GST_STATE_CHANGE_READY_TO_PAUSED:
3799       no_preroll = gst_base_src_is_live (basesrc);
3800       break;
3801     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
3802       GST_DEBUG_OBJECT (basesrc, "PAUSED->PLAYING");
3803       if (gst_base_src_is_live (basesrc)) {
3804         /* now we can start playback */
3805         gst_base_src_set_playing (basesrc, TRUE);
3806       }
3807       break;
3808     default:
3809       break;
3810   }
3811
3812   if ((result =
3813           GST_ELEMENT_CLASS (parent_class)->change_state (element,
3814               transition)) == GST_STATE_CHANGE_FAILURE)
3815     goto failure;
3816
3817   switch (transition) {
3818     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
3819       GST_DEBUG_OBJECT (basesrc, "PLAYING->PAUSED");
3820       if (gst_base_src_is_live (basesrc)) {
3821         /* make sure we block in the live lock in PAUSED */
3822         gst_base_src_set_playing (basesrc, FALSE);
3823         no_preroll = TRUE;
3824       }
3825       break;
3826     case GST_STATE_CHANGE_PAUSED_TO_READY:
3827     {
3828       /* we don't need to unblock anything here, the pad deactivation code
3829        * already did this */
3830       if (g_atomic_int_get (&basesrc->priv->has_pending_eos)) {
3831         GST_OBJECT_LOCK (basesrc);
3832         CLEAR_PENDING_EOS (basesrc);
3833         GST_OBJECT_UNLOCK (basesrc);
3834       }
3835       gst_event_replace (&basesrc->pending_seek, NULL);
3836       break;
3837     }
3838     case GST_STATE_CHANGE_READY_TO_NULL:
3839       break;
3840     default:
3841       break;
3842   }
3843
3844   if (no_preroll && result == GST_STATE_CHANGE_SUCCESS)
3845     result = GST_STATE_CHANGE_NO_PREROLL;
3846
3847   return result;
3848
3849   /* ERRORS */
3850 failure:
3851   {
3852     GST_DEBUG_OBJECT (basesrc, "parent failed state change");
3853     return result;
3854   }
3855 }
3856
3857 /**
3858  * gst_base_src_get_buffer_pool:
3859  * @src: a #GstBaseSrc
3860  *
3861  * Returns: (transfer full): the instance of the #GstBufferPool used
3862  * by the src; unref it after usage.
3863  */
3864 GstBufferPool *
3865 gst_base_src_get_buffer_pool (GstBaseSrc * src)
3866 {
3867   g_return_val_if_fail (GST_IS_BASE_SRC (src), NULL);
3868
3869   if (src->priv->pool)
3870     return gst_object_ref (src->priv->pool);
3871
3872   return NULL;
3873 }
3874
3875 /**
3876  * gst_base_src_get_allocator:
3877  * @src: a #GstBaseSrc
3878  * @allocator: (out) (allow-none) (transfer full): the #GstAllocator
3879  * used
3880  * @params: (out) (allow-none) (transfer full): the
3881  * #GstAllocationParams of @allocator
3882  *
3883  * Lets #GstBaseSrc sub-classes to know the memory @allocator
3884  * used by the base class and its @params.
3885  *
3886  * Unref the @allocator after usage.
3887  */
3888 void
3889 gst_base_src_get_allocator (GstBaseSrc * src,
3890     GstAllocator ** allocator, GstAllocationParams * params)
3891 {
3892   g_return_if_fail (GST_IS_BASE_SRC (src));
3893
3894   if (allocator)
3895     *allocator = src->priv->allocator ?
3896         gst_object_ref (src->priv->allocator) : NULL;
3897
3898   if (params)
3899     *params = src->priv->params;
3900 }