libs/gst/base/gstbasesrc.c

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