libs/gst/base/gstbasesrc.c

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