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