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