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