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