libs/gst/base/gstbasesrc.c

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