libs/gst/base/gstbasesrc.c

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