14b40215f4f16660fd6a64ebcfe1f33fbcecd12d
[glsdk/gstreamer0-10.git] / libs / gst / base / gstbasesink.c
1 /* GStreamer
2  * Copyright (C) 2005-2007 Wim Taymans <wim.taymans@gmail.com>
3  *
4  * gstbasesink.c: Base class for sink elements
5  *
6  * This library is free software; you can redistribute it and/or
7  * modify it under the terms of the GNU Library General Public
8  * License as published by the Free Software Foundation; either
9  * version 2 of the License, or (at your option) any later version.
10  *
11  * This library is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14  * Library General Public License for more details.
15  *
16  * You should have received a copy of the GNU Library General Public
17  * License along with this library; if not, write to the
18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
19  * Boston, MA 02111-1307, USA.
20  */
22 /**
23  * SECTION:gstbasesink
24  * @short_description: Base class for sink elements
25  * @see_also: #GstBaseTransform, #GstBaseSrc
26  *
27  * #GstBaseSink is the base class for sink elements in GStreamer, such as
28  * xvimagesink or filesink. It is a layer on top of #GstElement that provides a
29  * simplified interface to plugin writers. #GstBaseSink handles many details
30  * for you, for example: preroll, clock synchronization, state changes,
31  * activation in push or pull mode, and queries.
32  *
33  * In most cases, when writing sink elements, there is no need to implement
34  * class methods from #GstElement or to set functions on pads, because the
35  * #GstBaseSink infrastructure should be sufficient.
36  *
37  * #GstBaseSink provides support for exactly one sink pad, which should be
38  * named "sink". A sink implementation (subclass of #GstBaseSink) should
39  * install a pad template in its base_init function, like so:
40  * |[
41  * static void
42  * my_element_base_init (gpointer g_class)
43  * {
44  *   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
45  *
46  *   // sinktemplate should be a #GstStaticPadTemplate with direction
47  *   // #GST_PAD_SINK and name "sink"
48  *   gst_element_class_add_pad_template (gstelement_class,
49  *       gst_static_pad_template_get (&amp;sinktemplate));
50  *   // see #GstElementDetails
51  *   gst_element_class_set_details (gstelement_class, &amp;details);
52  * }
53  * ]|
54  *
55  * #GstBaseSink will handle the prerolling correctly. This means that it will
56  * return #GST_STATE_CHANGE_ASYNC from a state change to PAUSED until the first
57  * buffer arrives in this element. The base class will call the
58  * #GstBaseSinkClass.preroll() vmethod with this preroll buffer and will then
59  * commit the state change to the next asynchronously pending state.
60  *
61  * When the element is set to PLAYING, #GstBaseSink will synchronise on the
62  * clock using the times returned from #GstBaseSinkClass.get_times(). If this
63  * function returns #GST_CLOCK_TIME_NONE for the start time, no synchronisation
64  * will be done. Synchronisation can be disabled entirely by setting the object
65  * #GstBaseSink:sync property to %FALSE.
66  *
67  * After synchronisation the virtual method #GstBaseSinkClass.render() will be
68  * called. Subclasses should minimally implement this method.
69  *
70  * Since 0.10.3 subclasses that synchronise on the clock in the
71  * #GstBaseSinkClass.render() method are supported as well. These classes
72  * typically receive a buffer in the render method and can then potentially
73  * block on the clock while rendering. A typical example is an audiosink.
74  * Since 0.10.11 these subclasses can use gst_base_sink_wait_preroll() to
75  * perform the blocking wait.
76  *
77  * Upon receiving the EOS event in the PLAYING state, #GstBaseSink will wait
78  * for the clock to reach the time indicated by the stop time of the last
79  * #GstBaseSinkClass.get_times() call before posting an EOS message. When the
80  * element receives EOS in PAUSED, preroll completes, the event is queued and an
81  * EOS message is posted when going to PLAYING.
82  *
83  * #GstBaseSink will internally use the #GST_EVENT_NEWSEGMENT events to schedule
84  * synchronisation and clipping of buffers. Buffers that fall completely outside
85  * of the current segment are dropped. Buffers that fall partially in the
86  * segment are rendered (and prerolled). Subclasses should do any subbuffer
87  * clipping themselves when needed.
88  *
89  * #GstBaseSink will by default report the current playback position in
90  * #GST_FORMAT_TIME based on the current clock time and segment information.
91  * If no clock has been set on the element, the query will be forwarded
92  * upstream.
93  *
94  * The #GstBaseSinkClass.set_caps() function will be called when the subclass
95  * should configure itself to process a specific media type.
96  *
97  * The #GstBaseSinkClass.start() and #GstBaseSinkClass.stop() virtual methods
98  * will be called when resources should be allocated. Any 
99  * #GstBaseSinkClass.preroll(), #GstBaseSinkClass.render() and
100  * #GstBaseSinkClass.set_caps() function will be called between the
101  * #GstBaseSinkClass.start() and #GstBaseSinkClass.stop() calls.
102  *
103  * The #GstBaseSinkClass.event() virtual method will be called when an event is
104  * received by #GstBaseSink. Normally this method should only be overriden by
105  * very specific elements (such as file sinks) which need to handle the
106  * newsegment event specially.
107  *
108  * #GstBaseSink provides an overridable #GstBaseSinkClass.buffer_alloc()
109  * function that can be used by sinks that want to do reverse negotiation or to
110  * provide custom buffers (hardware buffers for example) to upstream elements.
111  *
112  * The #GstBaseSinkClass.unlock() method is called when the elements should
113  * unblock any blocking operations they perform in the
114  * #GstBaseSinkClass.render() method. This is mostly useful when the
115  * #GstBaseSinkClass.render() method performs a blocking write on a file
116  * descriptor, for example.
117  *
118  * The #GstBaseSink:max-lateness property affects how the sink deals with
119  * buffers that arrive too late in the sink. A buffer arrives too late in the
120  * sink when the presentation time (as a combination of the last segment, buffer
121  * timestamp and element base_time) plus the duration is before the current
122  * time of the clock.
123  * If the frame is later than max-lateness, the sink will drop the buffer
124  * without calling the render method.
125  * This feature is disabled if sync is disabled, the
126  * #GstBaseSinkClass.get_times() method does not return a valid start time or
127  * max-lateness is set to -1 (the default).
128  * Subclasses can use gst_base_sink_set_max_lateness() to configure the
129  * max-lateness value.
130  *
131  * The #GstBaseSink:qos property will enable the quality-of-service features of
132  * the basesink which gather statistics about the real-time performance of the
133  * clock synchronisation. For each buffer received in the sink, statistics are
134  * gathered and a QOS event is sent upstream with these numbers. This
135  * information can then be used by upstream elements to reduce their processing
136  * rate, for example.
137  *
138  * Since 0.10.15 the #GstBaseSink:async property can be used to instruct the
139  * sink to never perform an ASYNC state change. This feature is mostly usable
140  * when dealing with non-synchronized streams or sparse streams.
141  *
142  * Last reviewed on 2007-08-29 (0.10.15)
143  */
145 #ifdef HAVE_CONFIG_H
146 #  include "config.h"
147 #endif
149 #include "gstbasesink.h"
150 #include <gst/gstmarshal.h>
151 #include <gst/gst_private.h>
152 #include <gst/gst-i18n-lib.h>
154 GST_DEBUG_CATEGORY_STATIC (gst_base_sink_debug);
155 #define GST_CAT_DEFAULT gst_base_sink_debug
157 #define GST_BASE_SINK_GET_PRIVATE(obj)  \
158    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BASE_SINK, GstBaseSinkPrivate))
160 #define GST_FLOW_STEP GST_FLOW_CUSTOM_ERROR
162 typedef struct
164   gboolean valid;               /* if this info is valid */
165   guint32 seqnum;               /* the seqnum of the STEP event */
166   GstFormat format;             /* the format of the amount */
167   guint64 amount;               /* the total amount of data to skip */
168   guint64 position;             /* the position in the stepped data */
169   guint64 duration;             /* the duration in time of the skipped data */
170   guint64 start;                /* running_time of the start */
171   gdouble rate;                 /* rate of skipping */
172   gdouble start_rate;           /* rate before skipping */
173   guint64 start_start;          /* start position skipping */
174   guint64 start_stop;           /* stop position skipping */
175   gboolean flush;               /* if this was a flushing step */
176   gboolean intermediate;        /* if this is an intermediate step */
177   gboolean need_preroll;        /* if we need preroll after this step */
178 } GstStepInfo;
180 /* FIXME, some stuff in ABI.data and other in Private...
181  * Make up your mind please.
182  */
183 struct _GstBaseSinkPrivate
185   gint qos_enabled;             /* ATOMIC */
186   gboolean async_enabled;
187   GstClockTimeDiff ts_offset;
188   GstClockTime render_delay;
190   /* start, stop of current buffer, stream time, used to report position */
191   GstClockTime current_sstart;
192   GstClockTime current_sstop;
194   /* start, stop and jitter of current buffer, running time */
195   GstClockTime current_rstart;
196   GstClockTime current_rstop;
197   GstClockTimeDiff current_jitter;
199   /* EOS sync time in running time */
200   GstClockTime eos_rtime;
202   /* last buffer that arrived in time, running time */
203   GstClockTime last_in_time;
204   /* when the last buffer left the sink, running time */
205   GstClockTime last_left;
207   /* running averages go here these are done on running time */
208   GstClockTime avg_pt;
209   GstClockTime avg_duration;
210   gdouble avg_rate;
212   /* these are done on system time. avg_jitter and avg_render are
213    * compared to eachother to see if the rendering time takes a
214    * huge amount of the processing, If so we are flooded with
215    * buffers. */
216   GstClockTime last_left_systime;
217   GstClockTime avg_jitter;
218   GstClockTime start, stop;
219   GstClockTime avg_render;
221   /* number of rendered and dropped frames */
222   guint64 rendered;
223   guint64 dropped;
225   /* latency stuff */
226   GstClockTime latency;
228   /* if we already commited the state */
229   gboolean commited;
231   /* when we received EOS */
232   gboolean received_eos;
234   /* when we are prerolled and able to report latency */
235   gboolean have_latency;
237   /* the last buffer we prerolled or rendered. Useful for making snapshots */
238   GstBuffer *last_buffer;
240   /* caps for pull based scheduling */
241   GstCaps *pull_caps;
243   /* blocksize for pulling */
244   guint blocksize;
246   gboolean discont;
248   /* seqnum of the stream */
249   guint32 seqnum;
251   gboolean call_preroll;
252   gboolean step_unlock;
254   /* we have a pending and a current step operation */
255   GstStepInfo current_step;
256   GstStepInfo pending_step;
257 };
259 #define DO_RUNNING_AVG(avg,val,size) (((val) + ((size)-1) * (avg)) / (size))
261 /* generic running average, this has a neutral window size */
262 #define UPDATE_RUNNING_AVG(avg,val)   DO_RUNNING_AVG(avg,val,8)
264 /* the windows for these running averages are experimentally obtained.
265  * possitive values get averaged more while negative values use a small
266  * window so we can react faster to badness. */
267 #define UPDATE_RUNNING_AVG_P(avg,val) DO_RUNNING_AVG(avg,val,16)
268 #define UPDATE_RUNNING_AVG_N(avg,val) DO_RUNNING_AVG(avg,val,4)
270 /* BaseSink properties */
272 #define DEFAULT_CAN_ACTIVATE_PULL FALSE /* fixme: enable me */
273 #define DEFAULT_CAN_ACTIVATE_PUSH TRUE
275 #define DEFAULT_PREROLL_QUEUE_LEN   0
276 #define DEFAULT_SYNC                TRUE
277 #define DEFAULT_MAX_LATENESS        -1
278 #define DEFAULT_QOS                 FALSE
279 #define DEFAULT_ASYNC               TRUE
280 #define DEFAULT_TS_OFFSET           0
281 #define DEFAULT_BLOCKSIZE           4096
282 #define DEFAULT_RENDER_DELAY        0
284 enum
286   PROP_0,
287   PROP_PREROLL_QUEUE_LEN,
288   PROP_SYNC,
289   PROP_MAX_LATENESS,
290   PROP_QOS,
291   PROP_ASYNC,
292   PROP_TS_OFFSET,
293   PROP_LAST_BUFFER,
294   PROP_BLOCKSIZE,
295   PROP_RENDER_DELAY,
296   PROP_LAST
297 };
299 static GstElementClass *parent_class = NULL;
301 static void gst_base_sink_class_init (GstBaseSinkClass * klass);
302 static void gst_base_sink_init (GstBaseSink * trans, gpointer g_class);
303 static void gst_base_sink_finalize (GObject * object);
305 GType
306 gst_base_sink_get_type (void)
308   static volatile gsize base_sink_type = 0;
310   if (g_once_init_enter (&base_sink_type)) {
311     GType _type;
312     static const GTypeInfo base_sink_info = {
313       sizeof (GstBaseSinkClass),
314       NULL,
315       NULL,
316       (GClassInitFunc) gst_base_sink_class_init,
317       NULL,
318       NULL,
319       sizeof (GstBaseSink),
320       0,
321       (GInstanceInitFunc) gst_base_sink_init,
322     };
324     _type = g_type_register_static (GST_TYPE_ELEMENT,
325         "GstBaseSink", &base_sink_info, G_TYPE_FLAG_ABSTRACT);
326     g_once_init_leave (&base_sink_type, _type);
327   }
328   return base_sink_type;
331 static void gst_base_sink_set_property (GObject * object, guint prop_id,
332     const GValue * value, GParamSpec * pspec);
333 static void gst_base_sink_get_property (GObject * object, guint prop_id,
334     GValue * value, GParamSpec * pspec);
336 static gboolean gst_base_sink_send_event (GstElement * element,
337     GstEvent * event);
338 static gboolean gst_base_sink_query (GstElement * element, GstQuery * query);
340 static GstCaps *gst_base_sink_get_caps (GstBaseSink * sink);
341 static gboolean gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps);
342 static GstFlowReturn gst_base_sink_buffer_alloc (GstBaseSink * sink,
343     guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
344 static void gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
345     GstClockTime * start, GstClockTime * end);
346 static gboolean gst_base_sink_set_flushing (GstBaseSink * basesink,
347     GstPad * pad, gboolean flushing);
348 static gboolean gst_base_sink_default_activate_pull (GstBaseSink * basesink,
349     gboolean active);
350 static gboolean gst_base_sink_default_do_seek (GstBaseSink * sink,
351     GstSegment * segment);
352 static gboolean gst_base_sink_default_prepare_seek_segment (GstBaseSink * sink,
353     GstEvent * event, GstSegment * segment);
355 static GstStateChangeReturn gst_base_sink_change_state (GstElement * element,
356     GstStateChange transition);
358 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
359 static GstFlowReturn gst_base_sink_chain_list (GstPad * pad,
360     GstBufferList * list);
362 static void gst_base_sink_loop (GstPad * pad);
363 static gboolean gst_base_sink_pad_activate (GstPad * pad);
364 static gboolean gst_base_sink_pad_activate_push (GstPad * pad, gboolean active);
365 static gboolean gst_base_sink_pad_activate_pull (GstPad * pad, gboolean active);
366 static gboolean gst_base_sink_event (GstPad * pad, GstEvent * event);
367 static gboolean gst_base_sink_peer_query (GstBaseSink * sink, GstQuery * query);
369 static gboolean gst_base_sink_negotiate_pull (GstBaseSink * basesink);
370 static GstCaps *gst_base_sink_pad_getcaps (GstPad * pad);
371 static gboolean gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps);
372 static void gst_base_sink_pad_fixate (GstPad * pad, GstCaps * caps);
373 static GstFlowReturn gst_base_sink_pad_buffer_alloc (GstPad * pad,
374     guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
377 /* check if an object was too late */
378 static gboolean gst_base_sink_is_too_late (GstBaseSink * basesink,
379     GstMiniObject * obj, GstClockTime start, GstClockTime stop,
380     GstClockReturn status, GstClockTimeDiff jitter);
381 static GstFlowReturn gst_base_sink_preroll_object (GstBaseSink * basesink,
382     gboolean is_list, GstMiniObject * obj);
384 static void
385 gst_base_sink_class_init (GstBaseSinkClass * klass)
387   GObjectClass *gobject_class;
388   GstElementClass *gstelement_class;
390   gobject_class = G_OBJECT_CLASS (klass);
391   gstelement_class = GST_ELEMENT_CLASS (klass);
393   GST_DEBUG_CATEGORY_INIT (gst_base_sink_debug, "basesink", 0,
394       "basesink element");
396   g_type_class_add_private (klass, sizeof (GstBaseSinkPrivate));
398   parent_class = g_type_class_peek_parent (klass);
400   gobject_class->finalize = gst_base_sink_finalize;
401   gobject_class->set_property = gst_base_sink_set_property;
402   gobject_class->get_property = gst_base_sink_get_property;
404   /* FIXME, this next value should be configured using an event from the
405    * upstream element, ie, the BUFFER_SIZE event. */
406   g_object_class_install_property (gobject_class, PROP_PREROLL_QUEUE_LEN,
407       g_param_spec_uint ("preroll-queue-len", "Preroll queue length",
408           "Number of buffers to queue during preroll", 0, G_MAXUINT,
409           DEFAULT_PREROLL_QUEUE_LEN,
410           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
412   g_object_class_install_property (gobject_class, PROP_SYNC,
413       g_param_spec_boolean ("sync", "Sync", "Sync on the clock", DEFAULT_SYNC,
414           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
416   g_object_class_install_property (gobject_class, PROP_MAX_LATENESS,
417       g_param_spec_int64 ("max-lateness", "Max Lateness",
418           "Maximum number of nanoseconds that a buffer can be late before it "
419           "is dropped (-1 unlimited)", -1, G_MAXINT64, DEFAULT_MAX_LATENESS,
420           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
422   g_object_class_install_property (gobject_class, PROP_QOS,
423       g_param_spec_boolean ("qos", "Qos",
424           "Generate Quality-of-Service events upstream", DEFAULT_QOS,
425           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
426   /**
427    * GstBaseSink:async
428    *
429    * If set to #TRUE, the basesink will perform asynchronous state changes.
430    * When set to #FALSE, the sink will not signal the parent when it prerolls.
431    * Use this option when dealing with sparse streams or when synchronisation is
432    * not required.
433    *
434    * Since: 0.10.15
435    */
436   g_object_class_install_property (gobject_class, PROP_ASYNC,
437       g_param_spec_boolean ("async", "Async",
438           "Go asynchronously to PAUSED", DEFAULT_ASYNC,
439           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
440   /**
441    * GstBaseSink:ts-offset
442    *
443    * Controls the final synchronisation, a negative value will render the buffer
444    * earlier while a positive value delays playback. This property can be
445    * used to fix synchronisation in bad files.
446    *
447    * Since: 0.10.15
448    */
449   g_object_class_install_property (gobject_class, PROP_TS_OFFSET,
450       g_param_spec_int64 ("ts-offset", "TS Offset",
451           "Timestamp offset in nanoseconds", G_MININT64, G_MAXINT64,
452           DEFAULT_TS_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
453   /**
454    * GstBaseSink:last-buffer
455    *
456    * The last buffer that arrived in the sink and was used for preroll or for
457    * rendering. This property can be used to generate thumbnails. This property
458    * can be NULL when the sink has not yet received a bufer.
459    *
460    * Since: 0.10.15
461    */
462   g_object_class_install_property (gobject_class, PROP_LAST_BUFFER,
463       gst_param_spec_mini_object ("last-buffer", "Last Buffer",
464           "The last buffer received in the sink", GST_TYPE_BUFFER,
465           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
466   /**
467    * GstBaseSink:blocksize
468    *
469    * The amount of bytes to pull when operating in pull mode.
470    *
471    * Since: 0.10.22
472    */
473   g_object_class_install_property (gobject_class, PROP_BLOCKSIZE,
474       g_param_spec_uint ("blocksize", "Block size",
475           "Size in bytes to pull per buffer (0 = default)", 0, G_MAXUINT,
476           DEFAULT_BLOCKSIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
477   /**
478    * GstBaseSink:render-delay
479    *
480    * The additional delay between synchronisation and actual rendering of the
481    * media. This property will add additional latency to the device in order to
482    * make other sinks compensate for the delay.
483    *
484    * Since: 0.10.22
485    */
486   g_object_class_install_property (gobject_class, PROP_RENDER_DELAY,
487       g_param_spec_uint64 ("render-delay", "Render Delay",
488           "Additional render delay of the sink in nanoseconds", 0, G_MAXUINT64,
489           DEFAULT_RENDER_DELAY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
491   gstelement_class->change_state =
492       GST_DEBUG_FUNCPTR (gst_base_sink_change_state);
493   gstelement_class->send_event = GST_DEBUG_FUNCPTR (gst_base_sink_send_event);
494   gstelement_class->query = GST_DEBUG_FUNCPTR (gst_base_sink_query);
496   klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_sink_get_caps);
497   klass->set_caps = GST_DEBUG_FUNCPTR (gst_base_sink_set_caps);
498   klass->buffer_alloc = GST_DEBUG_FUNCPTR (gst_base_sink_buffer_alloc);
499   klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
500   klass->activate_pull =
501       GST_DEBUG_FUNCPTR (gst_base_sink_default_activate_pull);
503   /* Registering debug symbols for function pointers */
504   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_getcaps);
505   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_setcaps);
506   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_fixate);
507   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_buffer_alloc);
508   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_activate);
509   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_activate_push);
510   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_pad_activate_pull);
511   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_event);
512   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_chain);
513   GST_DEBUG_REGISTER_FUNCPTR (gst_base_sink_chain_list);
516 static GstCaps *
517 gst_base_sink_pad_getcaps (GstPad * pad)
519   GstBaseSinkClass *bclass;
520   GstBaseSink *bsink;
521   GstCaps *caps = NULL;
523   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
524   bclass = GST_BASE_SINK_GET_CLASS (bsink);
526   if (bsink->pad_mode == GST_ACTIVATE_PULL) {
527     /* if we are operating in pull mode we only accept the negotiated caps */
528     GST_OBJECT_LOCK (pad);
529     if ((caps = GST_PAD_CAPS (pad)))
530       gst_caps_ref (caps);
531     GST_OBJECT_UNLOCK (pad);
532   }
533   if (caps == NULL) {
534     if (bclass->get_caps)
535       caps = bclass->get_caps (bsink);
537     if (caps == NULL) {
538       GstPadTemplate *pad_template;
540       pad_template =
541           gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass),
542           "sink");
543       if (pad_template != NULL) {
544         caps = gst_caps_ref (gst_pad_template_get_caps (pad_template));
545       }
546     }
547   }
548   gst_object_unref (bsink);
550   return caps;
553 static gboolean
554 gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps)
556   GstBaseSinkClass *bclass;
557   GstBaseSink *bsink;
558   gboolean res = TRUE;
560   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
561   bclass = GST_BASE_SINK_GET_CLASS (bsink);
563   if (res && bclass->set_caps)
564     res = bclass->set_caps (bsink, caps);
566   gst_object_unref (bsink);
568   return res;
571 static void
572 gst_base_sink_pad_fixate (GstPad * pad, GstCaps * caps)
574   GstBaseSinkClass *bclass;
575   GstBaseSink *bsink;
577   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
578   bclass = GST_BASE_SINK_GET_CLASS (bsink);
580   if (bclass->fixate)
581     bclass->fixate (bsink, caps);
583   gst_object_unref (bsink);
586 static GstFlowReturn
587 gst_base_sink_pad_buffer_alloc (GstPad * pad, guint64 offset, guint size,
588     GstCaps * caps, GstBuffer ** buf)
590   GstBaseSinkClass *bclass;
591   GstBaseSink *bsink;
592   GstFlowReturn result = GST_FLOW_OK;
594   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
595   bclass = GST_BASE_SINK_GET_CLASS (bsink);
597   if (bclass->buffer_alloc)
598     result = bclass->buffer_alloc (bsink, offset, size, caps, buf);
599   else
600     *buf = NULL;                /* fallback in gstpad.c will allocate generic buffer */
602   gst_object_unref (bsink);
604   return result;
607 static void
608 gst_base_sink_init (GstBaseSink * basesink, gpointer g_class)
610   GstPadTemplate *pad_template;
611   GstBaseSinkPrivate *priv;
613   basesink->priv = priv = GST_BASE_SINK_GET_PRIVATE (basesink);
615   pad_template =
616       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "sink");
617   g_return_if_fail (pad_template != NULL);
619   basesink->sinkpad = gst_pad_new_from_template (pad_template, "sink");
621   gst_pad_set_getcaps_function (basesink->sinkpad, gst_base_sink_pad_getcaps);
622   gst_pad_set_setcaps_function (basesink->sinkpad, gst_base_sink_pad_setcaps);
623   gst_pad_set_fixatecaps_function (basesink->sinkpad, gst_base_sink_pad_fixate);
624   gst_pad_set_bufferalloc_function (basesink->sinkpad,
625       gst_base_sink_pad_buffer_alloc);
626   gst_pad_set_activate_function (basesink->sinkpad, gst_base_sink_pad_activate);
627   gst_pad_set_activatepush_function (basesink->sinkpad,
628       gst_base_sink_pad_activate_push);
629   gst_pad_set_activatepull_function (basesink->sinkpad,
630       gst_base_sink_pad_activate_pull);
631   gst_pad_set_event_function (basesink->sinkpad, gst_base_sink_event);
632   gst_pad_set_chain_function (basesink->sinkpad, gst_base_sink_chain);
633   gst_pad_set_chain_list_function (basesink->sinkpad, gst_base_sink_chain_list);
634   gst_element_add_pad (GST_ELEMENT_CAST (basesink), basesink->sinkpad);
636   basesink->pad_mode = GST_ACTIVATE_NONE;
637   basesink->preroll_queue = g_queue_new ();
638   basesink->abidata.ABI.clip_segment = gst_segment_new ();
639   priv->have_latency = FALSE;
641   basesink->can_activate_push = DEFAULT_CAN_ACTIVATE_PUSH;
642   basesink->can_activate_pull = DEFAULT_CAN_ACTIVATE_PULL;
644   basesink->sync = DEFAULT_SYNC;
645   basesink->abidata.ABI.max_lateness = DEFAULT_MAX_LATENESS;
646   g_atomic_int_set (&priv->qos_enabled, DEFAULT_QOS);
647   priv->async_enabled = DEFAULT_ASYNC;
648   priv->ts_offset = DEFAULT_TS_OFFSET;
649   priv->render_delay = DEFAULT_RENDER_DELAY;
650   priv->blocksize = DEFAULT_BLOCKSIZE;
652   GST_OBJECT_FLAG_SET (basesink, GST_ELEMENT_IS_SINK);
655 static void
656 gst_base_sink_finalize (GObject * object)
658   GstBaseSink *basesink;
660   basesink = GST_BASE_SINK (object);
662   g_queue_free (basesink->preroll_queue);
663   gst_segment_free (basesink->abidata.ABI.clip_segment);
665   G_OBJECT_CLASS (parent_class)->finalize (object);
668 /**
669  * gst_base_sink_set_sync:
670  * @sink: the sink
671  * @sync: the new sync value.
672  *
673  * Configures @sink to synchronize on the clock or not. When
674  * @sync is FALSE, incomming samples will be played as fast as
675  * possible. If @sync is TRUE, the timestamps of the incomming
676  * buffers will be used to schedule the exact render time of its
677  * contents.
678  *
679  * Since: 0.10.4
680  */
681 void
682 gst_base_sink_set_sync (GstBaseSink * sink, gboolean sync)
684   g_return_if_fail (GST_IS_BASE_SINK (sink));
686   GST_OBJECT_LOCK (sink);
687   sink->sync = sync;
688   GST_OBJECT_UNLOCK (sink);
691 /**
692  * gst_base_sink_get_sync:
693  * @sink: the sink
694  *
695  * Checks if @sink is currently configured to synchronize against the
696  * clock.
697  *
698  * Returns: TRUE if the sink is configured to synchronize against the clock.
699  *
700  * Since: 0.10.4
701  */
702 gboolean
703 gst_base_sink_get_sync (GstBaseSink * sink)
705   gboolean res;
707   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
709   GST_OBJECT_LOCK (sink);
710   res = sink->sync;
711   GST_OBJECT_UNLOCK (sink);
713   return res;
716 /**
717  * gst_base_sink_set_max_lateness:
718  * @sink: the sink
719  * @max_lateness: the new max lateness value.
720  *
721  * Sets the new max lateness value to @max_lateness. This value is
722  * used to decide if a buffer should be dropped or not based on the
723  * buffer timestamp and the current clock time. A value of -1 means
724  * an unlimited time.
725  *
726  * Since: 0.10.4
727  */
728 void
729 gst_base_sink_set_max_lateness (GstBaseSink * sink, gint64 max_lateness)
731   g_return_if_fail (GST_IS_BASE_SINK (sink));
733   GST_OBJECT_LOCK (sink);
734   sink->abidata.ABI.max_lateness = max_lateness;
735   GST_OBJECT_UNLOCK (sink);
738 /**
739  * gst_base_sink_get_max_lateness:
740  * @sink: the sink
741  *
742  * Gets the max lateness value. See gst_base_sink_set_max_lateness for
743  * more details.
744  *
745  * Returns: The maximum time in nanoseconds that a buffer can be late
746  * before it is dropped and not rendered. A value of -1 means an
747  * unlimited time.
748  *
749  * Since: 0.10.4
750  */
751 gint64
752 gst_base_sink_get_max_lateness (GstBaseSink * sink)
754   gint64 res;
756   g_return_val_if_fail (GST_IS_BASE_SINK (sink), -1);
758   GST_OBJECT_LOCK (sink);
759   res = sink->abidata.ABI.max_lateness;
760   GST_OBJECT_UNLOCK (sink);
762   return res;
765 /**
766  * gst_base_sink_set_qos_enabled:
767  * @sink: the sink
768  * @enabled: the new qos value.
769  *
770  * Configures @sink to send Quality-of-Service events upstream.
771  *
772  * Since: 0.10.5
773  */
774 void
775 gst_base_sink_set_qos_enabled (GstBaseSink * sink, gboolean enabled)
777   g_return_if_fail (GST_IS_BASE_SINK (sink));
779   g_atomic_int_set (&sink->priv->qos_enabled, enabled);
782 /**
783  * gst_base_sink_is_qos_enabled:
784  * @sink: the sink
785  *
786  * Checks if @sink is currently configured to send Quality-of-Service events
787  * upstream.
788  *
789  * Returns: TRUE if the sink is configured to perform Quality-of-Service.
790  *
791  * Since: 0.10.5
792  */
793 gboolean
794 gst_base_sink_is_qos_enabled (GstBaseSink * sink)
796   gboolean res;
798   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
800   res = g_atomic_int_get (&sink->priv->qos_enabled);
802   return res;
805 /**
806  * gst_base_sink_set_async_enabled:
807  * @sink: the sink
808  * @enabled: the new async value.
809  *
810  * Configures @sink to perform all state changes asynchronusly. When async is
811  * disabled, the sink will immediatly go to PAUSED instead of waiting for a
812  * preroll buffer. This feature is usefull if the sink does not synchronize
813  * against the clock or when it is dealing with sparse streams.
814  *
815  * Since: 0.10.15
816  */
817 void
818 gst_base_sink_set_async_enabled (GstBaseSink * sink, gboolean enabled)
820   g_return_if_fail (GST_IS_BASE_SINK (sink));
822   GST_PAD_PREROLL_LOCK (sink->sinkpad);
823   sink->priv->async_enabled = enabled;
824   GST_LOG_OBJECT (sink, "set async enabled to %d", enabled);
825   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
828 /**
829  * gst_base_sink_is_async_enabled:
830  * @sink: the sink
831  *
832  * Checks if @sink is currently configured to perform asynchronous state
833  * changes to PAUSED.
834  *
835  * Returns: TRUE if the sink is configured to perform asynchronous state
836  * changes.
837  *
838  * Since: 0.10.15
839  */
840 gboolean
841 gst_base_sink_is_async_enabled (GstBaseSink * sink)
843   gboolean res;
845   g_return_val_if_fail (GST_IS_BASE_SINK (sink), FALSE);
847   GST_PAD_PREROLL_LOCK (sink->sinkpad);
848   res = sink->priv->async_enabled;
849   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
851   return res;
854 /**
855  * gst_base_sink_set_ts_offset:
856  * @sink: the sink
857  * @offset: the new offset
858  *
859  * Adjust the synchronisation of @sink with @offset. A negative value will
860  * render buffers earlier than their timestamp. A positive value will delay
861  * rendering. This function can be used to fix playback of badly timestamped
862  * buffers.
863  *
864  * Since: 0.10.15
865  */
866 void
867 gst_base_sink_set_ts_offset (GstBaseSink * sink, GstClockTimeDiff offset)
869   g_return_if_fail (GST_IS_BASE_SINK (sink));
871   GST_OBJECT_LOCK (sink);
872   sink->priv->ts_offset = offset;
873   GST_LOG_OBJECT (sink, "set time offset to %" G_GINT64_FORMAT, offset);
874   GST_OBJECT_UNLOCK (sink);
877 /**
878  * gst_base_sink_get_ts_offset:
879  * @sink: the sink
880  *
881  * Get the synchronisation offset of @sink.
882  *
883  * Returns: The synchronisation offset.
884  *
885  * Since: 0.10.15
886  */
887 GstClockTimeDiff
888 gst_base_sink_get_ts_offset (GstBaseSink * sink)
890   GstClockTimeDiff res;
892   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
894   GST_OBJECT_LOCK (sink);
895   res = sink->priv->ts_offset;
896   GST_OBJECT_UNLOCK (sink);
898   return res;
901 /**
902  * gst_base_sink_get_last_buffer:
903  * @sink: the sink
904  *
905  * Get the last buffer that arrived in the sink and was used for preroll or for
906  * rendering. This property can be used to generate thumbnails.
907  *
908  * The #GstCaps on the buffer can be used to determine the type of the buffer.
909  *
910  * Returns: a #GstBuffer. gst_buffer_unref() after usage. This function returns
911  * NULL when no buffer has arrived in the sink yet or when the sink is not in
912  * PAUSED or PLAYING.
913  *
914  * Since: 0.10.15
915  */
916 GstBuffer *
917 gst_base_sink_get_last_buffer (GstBaseSink * sink)
919   GstBuffer *res;
921   g_return_val_if_fail (GST_IS_BASE_SINK (sink), NULL);
923   GST_OBJECT_LOCK (sink);
924   if ((res = sink->priv->last_buffer))
925     gst_buffer_ref (res);
926   GST_OBJECT_UNLOCK (sink);
928   return res;
931 static void
932 gst_base_sink_set_last_buffer (GstBaseSink * sink, GstBuffer * buffer)
934   GstBuffer *old;
936   GST_OBJECT_LOCK (sink);
937   old = sink->priv->last_buffer;
938   if (G_LIKELY (old != buffer)) {
939     GST_DEBUG_OBJECT (sink, "setting last buffer to %p", buffer);
940     if (G_LIKELY (buffer))
941       gst_buffer_ref (buffer);
942     sink->priv->last_buffer = buffer;
943   } else {
944     old = NULL;
945   }
946   GST_OBJECT_UNLOCK (sink);
948   /* avoid unreffing with the lock because cleanup code might want to take the
949    * lock too */
950   if (G_LIKELY (old))
951     gst_buffer_unref (old);
954 /**
955  * gst_base_sink_get_latency:
956  * @sink: the sink
957  *
958  * Get the currently configured latency.
959  *
960  * Returns: The configured latency.
961  *
962  * Since: 0.10.12
963  */
964 GstClockTime
965 gst_base_sink_get_latency (GstBaseSink * sink)
967   GstClockTime res;
969   GST_OBJECT_LOCK (sink);
970   res = sink->priv->latency;
971   GST_OBJECT_UNLOCK (sink);
973   return res;
976 /**
977  * gst_base_sink_query_latency:
978  * @sink: the sink
979  * @live: if the sink is live
980  * @upstream_live: if an upstream element is live
981  * @min_latency: the min latency of the upstream elements
982  * @max_latency: the max latency of the upstream elements
983  *
984  * Query the sink for the latency parameters. The latency will be queried from
985  * the upstream elements. @live will be TRUE if @sink is configured to
986  * synchronize against the clock. @upstream_live will be TRUE if an upstream
987  * element is live.
988  *
989  * If both @live and @upstream_live are TRUE, the sink will want to compensate
990  * for the latency introduced by the upstream elements by setting the
991  * @min_latency to a strictly possitive value.
992  *
993  * This function is mostly used by subclasses.
994  *
995  * Returns: TRUE if the query succeeded.
996  *
997  * Since: 0.10.12
998  */
999 gboolean
1000 gst_base_sink_query_latency (GstBaseSink * sink, gboolean * live,
1001     gboolean * upstream_live, GstClockTime * min_latency,
1002     GstClockTime * max_latency)
1004   gboolean l, us_live, res, have_latency;
1005   GstClockTime min, max, render_delay;
1006   GstQuery *query;
1007   GstClockTime us_min, us_max;
1009   /* we are live when we sync to the clock */
1010   GST_OBJECT_LOCK (sink);
1011   l = sink->sync;
1012   have_latency = sink->priv->have_latency;
1013   render_delay = sink->priv->render_delay;
1014   GST_OBJECT_UNLOCK (sink);
1016   /* assume no latency */
1017   min = 0;
1018   max = -1;
1019   us_live = FALSE;
1021   if (have_latency) {
1022     GST_DEBUG_OBJECT (sink, "we are ready for LATENCY query");
1023     /* we are ready for a latency query this is when we preroll or when we are
1024      * not async. */
1025     query = gst_query_new_latency ();
1027     /* ask the peer for the latency */
1028     if ((res = gst_base_sink_peer_query (sink, query))) {
1029       /* get upstream min and max latency */
1030       gst_query_parse_latency (query, &us_live, &us_min, &us_max);
1032       if (us_live) {
1033         /* upstream live, use its latency, subclasses should use these
1034          * values to create the complete latency. */
1035         min = us_min;
1036         max = us_max;
1037       }
1038       if (l) {
1039         /* we need to add the render delay if we are live */
1040         if (min != -1)
1041           min += render_delay;
1042         if (max != -1)
1043           max += render_delay;
1044       }
1045     }
1046     gst_query_unref (query);
1047   } else {
1048     GST_DEBUG_OBJECT (sink, "we are not yet ready for LATENCY query");
1049     res = FALSE;
1050   }
1052   /* not live, we tried to do the query, if it failed we return TRUE anyway */
1053   if (!res) {
1054     if (!l) {
1055       res = TRUE;
1056       GST_DEBUG_OBJECT (sink, "latency query failed but we are not live");
1057     } else {
1058       GST_DEBUG_OBJECT (sink, "latency query failed and we are live");
1059     }
1060   }
1062   if (res) {
1063     GST_DEBUG_OBJECT (sink, "latency query: live: %d, have_latency %d,"
1064         " upstream: %d, min %" GST_TIME_FORMAT ", max %" GST_TIME_FORMAT, l,
1065         have_latency, us_live, GST_TIME_ARGS (min), GST_TIME_ARGS (max));
1067     if (live)
1068       *live = l;
1069     if (upstream_live)
1070       *upstream_live = us_live;
1071     if (min_latency)
1072       *min_latency = min;
1073     if (max_latency)
1074       *max_latency = max;
1075   }
1076   return res;
1079 /**
1080  * gst_base_sink_set_render_delay:
1081  * @sink: a #GstBaseSink
1082  * @delay: the new delay
1083  *
1084  * Set the render delay in @sink to @delay. The render delay is the time
1085  * between actual rendering of a buffer and its synchronisation time. Some
1086  * devices might delay media rendering which can be compensated for with this
1087  * function.
1088  *
1089  * After calling this function, this sink will report additional latency and
1090  * other sinks will adjust their latency to delay the rendering of their media.
1091  *
1092  * This function is usually called by subclasses.
1093  *
1094  * Since: 0.10.21
1095  */
1096 void
1097 gst_base_sink_set_render_delay (GstBaseSink * sink, GstClockTime delay)
1099   GstClockTime old_render_delay;
1101   g_return_if_fail (GST_IS_BASE_SINK (sink));
1103   GST_OBJECT_LOCK (sink);
1104   old_render_delay = sink->priv->render_delay;
1105   sink->priv->render_delay = delay;
1106   GST_LOG_OBJECT (sink, "set render delay to %" GST_TIME_FORMAT,
1107       GST_TIME_ARGS (delay));
1108   GST_OBJECT_UNLOCK (sink);
1110   if (delay != old_render_delay) {
1111     GST_DEBUG_OBJECT (sink, "posting latency changed");
1112     gst_element_post_message (GST_ELEMENT_CAST (sink),
1113         gst_message_new_latency (GST_OBJECT_CAST (sink)));
1114   }
1117 /**
1118  * gst_base_sink_get_render_delay:
1119  * @sink: a #GstBaseSink
1120  *
1121  * Get the render delay of @sink. see gst_base_sink_set_render_delay() for more
1122  * information about the render delay.
1123  *
1124  * Returns: the render delay of @sink.
1125  *
1126  * Since: 0.10.21
1127  */
1128 GstClockTime
1129 gst_base_sink_get_render_delay (GstBaseSink * sink)
1131   GstClockTimeDiff res;
1133   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
1135   GST_OBJECT_LOCK (sink);
1136   res = sink->priv->render_delay;
1137   GST_OBJECT_UNLOCK (sink);
1139   return res;
1142 /**
1143  * gst_base_sink_set_blocksize:
1144  * @sink: a #GstBaseSink
1145  * @blocksize: the blocksize in bytes
1146  *
1147  * Set the number of bytes that the sink will pull when it is operating in pull
1148  * mode.
1149  *
1150  * Since: 0.10.22
1151  */
1152 void
1153 gst_base_sink_set_blocksize (GstBaseSink * sink, guint blocksize)
1155   g_return_if_fail (GST_IS_BASE_SINK (sink));
1157   GST_OBJECT_LOCK (sink);
1158   sink->priv->blocksize = blocksize;
1159   GST_LOG_OBJECT (sink, "set blocksize to %u", blocksize);
1160   GST_OBJECT_UNLOCK (sink);
1163 /**
1164  * gst_base_sink_get_blocksize:
1165  * @sink: a #GstBaseSink
1166  *
1167  * Get the number of bytes that the sink will pull when it is operating in pull
1168  * mode.
1169  *
1170  * Returns: the number of bytes @sink will pull in pull mode.
1171  *
1172  * Since: 0.10.22
1173  */
1174 guint
1175 gst_base_sink_get_blocksize (GstBaseSink * sink)
1177   guint res;
1179   g_return_val_if_fail (GST_IS_BASE_SINK (sink), 0);
1181   GST_OBJECT_LOCK (sink);
1182   res = sink->priv->blocksize;
1183   GST_OBJECT_UNLOCK (sink);
1185   return res;
1188 static void
1189 gst_base_sink_set_property (GObject * object, guint prop_id,
1190     const GValue * value, GParamSpec * pspec)
1192   GstBaseSink *sink = GST_BASE_SINK (object);
1194   switch (prop_id) {
1195     case PROP_PREROLL_QUEUE_LEN:
1196       /* preroll lock necessary to serialize with finish_preroll */
1197       GST_PAD_PREROLL_LOCK (sink->sinkpad);
1198       sink->preroll_queue_max_len = g_value_get_uint (value);
1199       GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
1200       break;
1201     case PROP_SYNC:
1202       gst_base_sink_set_sync (sink, g_value_get_boolean (value));
1203       break;
1204     case PROP_MAX_LATENESS:
1205       gst_base_sink_set_max_lateness (sink, g_value_get_int64 (value));
1206       break;
1207     case PROP_QOS:
1208       gst_base_sink_set_qos_enabled (sink, g_value_get_boolean (value));
1209       break;
1210     case PROP_ASYNC:
1211       gst_base_sink_set_async_enabled (sink, g_value_get_boolean (value));
1212       break;
1213     case PROP_TS_OFFSET:
1214       gst_base_sink_set_ts_offset (sink, g_value_get_int64 (value));
1215       break;
1216     case PROP_BLOCKSIZE:
1217       gst_base_sink_set_blocksize (sink, g_value_get_uint (value));
1218       break;
1219     case PROP_RENDER_DELAY:
1220       gst_base_sink_set_render_delay (sink, g_value_get_uint64 (value));
1221       break;
1222     default:
1223       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1224       break;
1225   }
1228 static void
1229 gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
1230     GParamSpec * pspec)
1232   GstBaseSink *sink = GST_BASE_SINK (object);
1234   switch (prop_id) {
1235     case PROP_PREROLL_QUEUE_LEN:
1236       GST_PAD_PREROLL_LOCK (sink->sinkpad);
1237       g_value_set_uint (value, sink->preroll_queue_max_len);
1238       GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
1239       break;
1240     case PROP_SYNC:
1241       g_value_set_boolean (value, gst_base_sink_get_sync (sink));
1242       break;
1243     case PROP_MAX_LATENESS:
1244       g_value_set_int64 (value, gst_base_sink_get_max_lateness (sink));
1245       break;
1246     case PROP_QOS:
1247       g_value_set_boolean (value, gst_base_sink_is_qos_enabled (sink));
1248       break;
1249     case PROP_ASYNC:
1250       g_value_set_boolean (value, gst_base_sink_is_async_enabled (sink));
1251       break;
1252     case PROP_TS_OFFSET:
1253       g_value_set_int64 (value, gst_base_sink_get_ts_offset (sink));
1254       break;
1255     case PROP_LAST_BUFFER:
1256       gst_value_take_buffer (value, gst_base_sink_get_last_buffer (sink));
1257       break;
1258     case PROP_BLOCKSIZE:
1259       g_value_set_uint (value, gst_base_sink_get_blocksize (sink));
1260       break;
1261     case PROP_RENDER_DELAY:
1262       g_value_set_uint64 (value, gst_base_sink_get_render_delay (sink));
1263       break;
1264     default:
1265       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1266       break;
1267   }
1271 static GstCaps *
1272 gst_base_sink_get_caps (GstBaseSink * sink)
1274   return NULL;
1277 static gboolean
1278 gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
1280   return TRUE;
1283 static GstFlowReturn
1284 gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
1285     GstCaps * caps, GstBuffer ** buf)
1287   *buf = NULL;
1288   return GST_FLOW_OK;
1291 /* with PREROLL_LOCK, STREAM_LOCK */
1292 static void
1293 gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
1295   GstMiniObject *obj;
1297   GST_DEBUG_OBJECT (basesink, "flushing queue %p", basesink);
1298   while ((obj = g_queue_pop_head (basesink->preroll_queue))) {
1299     GST_DEBUG_OBJECT (basesink, "popped %p", obj);
1300     gst_mini_object_unref (obj);
1301   }
1302   /* we can't have EOS anymore now */
1303   basesink->eos = FALSE;
1304   basesink->priv->received_eos = FALSE;
1305   basesink->have_preroll = FALSE;
1306   basesink->priv->step_unlock = FALSE;
1307   basesink->eos_queued = FALSE;
1308   basesink->preroll_queued = 0;
1309   basesink->buffers_queued = 0;
1310   basesink->events_queued = 0;
1311   /* can't report latency anymore until we preroll again */
1312   if (basesink->priv->async_enabled) {
1313     GST_OBJECT_LOCK (basesink);
1314     basesink->priv->have_latency = FALSE;
1315     GST_OBJECT_UNLOCK (basesink);
1316   }
1317   /* and signal any waiters now */
1318   GST_PAD_PREROLL_SIGNAL (pad);
1321 /* with STREAM_LOCK, configures given segment with the event information. */
1322 static void
1323 gst_base_sink_configure_segment (GstBaseSink * basesink, GstPad * pad,
1324     GstEvent * event, GstSegment * segment)
1326   gboolean update;
1327   gdouble rate, arate;
1328   GstFormat format;
1329   gint64 start;
1330   gint64 stop;
1331   gint64 time;
1333   /* the newsegment event is needed to bring the buffer timestamps to the
1334    * stream time and to drop samples outside of the playback segment. */
1335   gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format,
1336       &start, &stop, &time);
1338   /* The segment is protected with both the STREAM_LOCK and the OBJECT_LOCK.
1339    * We protect with the OBJECT_LOCK so that we can use the values to
1340    * safely answer a POSITION query. */
1341   GST_OBJECT_LOCK (basesink);
1342   gst_segment_set_newsegment_full (segment, update, rate, arate, format, start,
1343       stop, time);
1345   if (format == GST_FORMAT_TIME) {
1346     GST_DEBUG_OBJECT (basesink,
1347         "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
1348         "format GST_FORMAT_TIME, "
1349         "%" GST_TIME_FORMAT " -- %" GST_TIME_FORMAT
1350         ", time %" GST_TIME_FORMAT ", accum %" GST_TIME_FORMAT,
1351         update, rate, arate, GST_TIME_ARGS (segment->start),
1352         GST_TIME_ARGS (segment->stop), GST_TIME_ARGS (segment->time),
1353         GST_TIME_ARGS (segment->accum));
1354   } else {
1355     GST_DEBUG_OBJECT (basesink,
1356         "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
1357         "format %d, "
1358         "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
1359         G_GINT64_FORMAT ", accum %" G_GINT64_FORMAT, update, rate, arate,
1360         segment->format, segment->start, segment->stop, segment->time,
1361         segment->accum);
1362   }
1363   GST_OBJECT_UNLOCK (basesink);
1366 /* with PREROLL_LOCK, STREAM_LOCK */
1367 static gboolean
1368 gst_base_sink_commit_state (GstBaseSink * basesink)
1370   /* commit state and proceed to next pending state */
1371   GstState current, next, pending, post_pending;
1372   gboolean post_paused = FALSE;
1373   gboolean post_async_done = FALSE;
1374   gboolean post_playing = FALSE;
1376   /* we are certainly not playing async anymore now */
1377   basesink->playing_async = FALSE;
1379   GST_OBJECT_LOCK (basesink);
1380   current = GST_STATE (basesink);
1381   next = GST_STATE_NEXT (basesink);
1382   pending = GST_STATE_PENDING (basesink);
1383   post_pending = pending;
1385   switch (pending) {
1386     case GST_STATE_PLAYING:
1387     {
1388       GstBaseSinkClass *bclass;
1389       GstStateChangeReturn ret;
1391       bclass = GST_BASE_SINK_GET_CLASS (basesink);
1393       GST_DEBUG_OBJECT (basesink, "commiting state to PLAYING");
1395       basesink->need_preroll = FALSE;
1396       post_async_done = TRUE;
1397       basesink->priv->commited = TRUE;
1398       post_playing = TRUE;
1399       /* post PAUSED too when we were READY */
1400       if (current == GST_STATE_READY) {
1401         post_paused = TRUE;
1402       }
1404       /* make sure we notify the subclass of async playing */
1405       if (bclass->async_play) {
1406         GST_WARNING_OBJECT (basesink, "deprecated async_play");
1407         ret = bclass->async_play (basesink);
1408         if (ret == GST_STATE_CHANGE_FAILURE)
1409           goto async_failed;
1410       }
1411       break;
1412     }
1413     case GST_STATE_PAUSED:
1414       GST_DEBUG_OBJECT (basesink, "commiting state to PAUSED");
1415       post_paused = TRUE;
1416       post_async_done = TRUE;
1417       basesink->priv->commited = TRUE;
1418       post_pending = GST_STATE_VOID_PENDING;
1419       break;
1420     case GST_STATE_READY:
1421     case GST_STATE_NULL:
1422       goto stopping;
1423     case GST_STATE_VOID_PENDING:
1424       goto nothing_pending;
1425     default:
1426       break;
1427   }
1429   /* we can report latency queries now */
1430   basesink->priv->have_latency = TRUE;
1432   GST_STATE (basesink) = pending;
1433   GST_STATE_NEXT (basesink) = GST_STATE_VOID_PENDING;
1434   GST_STATE_PENDING (basesink) = GST_STATE_VOID_PENDING;
1435   GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_SUCCESS;
1436   GST_OBJECT_UNLOCK (basesink);
1438   if (post_paused) {
1439     GST_DEBUG_OBJECT (basesink, "posting PAUSED state change message");
1440     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1441         gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
1442             current, next, post_pending));
1443   }
1444   if (post_async_done) {
1445     GST_DEBUG_OBJECT (basesink, "posting async-done message");
1446     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1447         gst_message_new_async_done (GST_OBJECT_CAST (basesink)));
1448   }
1449   if (post_playing) {
1450     GST_DEBUG_OBJECT (basesink, "posting PLAYING state change message");
1451     gst_element_post_message (GST_ELEMENT_CAST (basesink),
1452         gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
1453             next, pending, GST_STATE_VOID_PENDING));
1454   }
1456   GST_STATE_BROADCAST (basesink);
1458   return TRUE;
1460 nothing_pending:
1461   {
1462     /* Depending on the state, set our vars. We get in this situation when the
1463      * state change function got a change to update the state vars before the
1464      * streaming thread did. This is fine but we need to make sure that we
1465      * update the need_preroll var since it was TRUE when we got here and might
1466      * become FALSE if we got to PLAYING. */
1467     GST_DEBUG_OBJECT (basesink, "nothing to commit, now in %s",
1468         gst_element_state_get_name (current));
1469     switch (current) {
1470       case GST_STATE_PLAYING:
1471         basesink->need_preroll = FALSE;
1472         break;
1473       case GST_STATE_PAUSED:
1474         basesink->need_preroll = TRUE;
1475         break;
1476       default:
1477         basesink->need_preroll = FALSE;
1478         basesink->flushing = TRUE;
1479         break;
1480     }
1481     /* we can report latency queries now */
1482     basesink->priv->have_latency = TRUE;
1483     GST_OBJECT_UNLOCK (basesink);
1484     return TRUE;
1485   }
1486 stopping:
1487   {
1488     /* app is going to READY */
1489     GST_DEBUG_OBJECT (basesink, "stopping");
1490     basesink->need_preroll = FALSE;
1491     basesink->flushing = TRUE;
1492     GST_OBJECT_UNLOCK (basesink);
1493     return FALSE;
1494   }
1495 async_failed:
1496   {
1497     GST_DEBUG_OBJECT (basesink, "async commit failed");
1498     GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_FAILURE;
1499     GST_OBJECT_UNLOCK (basesink);
1500     return FALSE;
1501   }
1504 static void
1505 start_stepping (GstBaseSink * sink, GstSegment * segment,
1506     GstStepInfo * pending, GstStepInfo * current)
1508   gint64 end;
1509   GstMessage *message;
1511   GST_DEBUG_OBJECT (sink, "update pending step");
1513   GST_OBJECT_LOCK (sink);
1514   memcpy (current, pending, sizeof (GstStepInfo));
1515   pending->valid = FALSE;
1516   GST_OBJECT_UNLOCK (sink);
1518   /* post message first */
1519   message =
1520       gst_message_new_step_start (GST_OBJECT (sink), TRUE, current->format,
1521       current->amount, current->rate, current->flush, current->intermediate);
1522   gst_message_set_seqnum (message, current->seqnum);
1523   gst_element_post_message (GST_ELEMENT (sink), message);
1525   /* get the running time of where we paused and remember it */
1526   current->start = gst_element_get_start_time (GST_ELEMENT_CAST (sink));
1527   gst_segment_set_running_time (segment, GST_FORMAT_TIME, current->start);
1529   /* set the new rate for the remainder of the segment */
1530   current->start_rate = segment->rate;
1531   segment->rate *= current->rate;
1532   segment->abs_rate = ABS (segment->rate);
1534   /* save values */
1535   if (segment->rate > 0.0)
1536     current->start_stop = segment->stop;
1537   else
1538     current->start_start = segment->start;
1540   if (current->format == GST_FORMAT_TIME) {
1541     end = current->start + current->amount;
1542     if (!current->flush) {
1543       /* update the segment clipping regions for non-flushing seeks */
1544       if (segment->rate > 0.0) {
1545         segment->stop = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1546         segment->last_stop = segment->stop;
1547       } else {
1548         gint64 position;
1550         position = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1551         segment->time = position;
1552         segment->start = position;
1553         segment->last_stop = position;
1554       }
1555     }
1556   }
1558   GST_DEBUG_OBJECT (sink,
1559       "segment now rate %lf, applied rate %lf, "
1560       "format GST_FORMAT_TIME, "
1561       "%" GST_TIME_FORMAT " -- %" GST_TIME_FORMAT
1562       ", time %" GST_TIME_FORMAT ", accum %" GST_TIME_FORMAT,
1563       segment->rate, segment->applied_rate, GST_TIME_ARGS (segment->start),
1564       GST_TIME_ARGS (segment->stop), GST_TIME_ARGS (segment->time),
1565       GST_TIME_ARGS (segment->accum));
1567   GST_DEBUG_OBJECT (sink, "step started at running_time %" GST_TIME_FORMAT,
1568       GST_TIME_ARGS (current->start));
1570   if (current->amount == -1) {
1571     GST_DEBUG_OBJECT (sink, "step amount == -1, stop stepping");
1572     current->valid = FALSE;
1573   } else {
1574     GST_DEBUG_OBJECT (sink, "step amount: %" G_GUINT64_FORMAT ", format: %s, "
1575         "rate: %f", current->amount, gst_format_get_name (current->format),
1576         current->rate);
1577   }
1580 static void
1581 stop_stepping (GstBaseSink * sink, GstSegment * segment,
1582     GstStepInfo * current, gint64 rstart, gint64 rstop, gboolean eos)
1584   gint64 stop, position;
1585   GstMessage *message;
1587   GST_DEBUG_OBJECT (sink, "step complete");
1589   if (segment->rate > 0.0)
1590     stop = rstart;
1591   else
1592     stop = rstop;
1594   GST_DEBUG_OBJECT (sink,
1595       "step stop at running_time %" GST_TIME_FORMAT, GST_TIME_ARGS (stop));
1597   if (stop == -1)
1598     current->duration = current->position;
1599   else
1600     current->duration = stop - current->start;
1602   GST_DEBUG_OBJECT (sink, "step elapsed running_time %" GST_TIME_FORMAT,
1603       GST_TIME_ARGS (current->duration));
1605   position = current->start + current->duration;
1607   /* now move the segment to the new running time */
1608   gst_segment_set_running_time (segment, GST_FORMAT_TIME, position);
1610   if (current->flush) {
1611     /* and remove the accumulated time we flushed, start time did not change */
1612     segment->accum = current->start;
1613   } else {
1614     /* start time is now the stepped position */
1615     gst_element_set_start_time (GST_ELEMENT_CAST (sink), position);
1616   }
1618   /* restore the previous rate */
1619   segment->rate = current->start_rate;
1620   segment->abs_rate = ABS (segment->rate);
1622   if (segment->rate > 0.0)
1623     segment->stop = current->start_stop;
1624   else
1625     segment->start = current->start_start;
1627   /* the clip segment is used for position report in paused... */
1628   memcpy (sink->abidata.ABI.clip_segment, segment, sizeof (GstSegment));
1630   /* post the step done when we know the stepped duration in TIME */
1631   message =
1632       gst_message_new_step_done (GST_OBJECT_CAST (sink), current->format,
1633       current->amount, current->rate, current->flush, current->intermediate,
1634       current->duration, eos);
1635   gst_message_set_seqnum (message, current->seqnum);
1636   gst_element_post_message (GST_ELEMENT_CAST (sink), message);
1638   if (!current->intermediate)
1639     sink->need_preroll = current->need_preroll;
1641   /* and the current step info finished and becomes invalid */
1642   current->valid = FALSE;
1645 static gboolean
1646 handle_stepping (GstBaseSink * sink, GstSegment * segment,
1647     GstStepInfo * current, gint64 * cstart, gint64 * cstop, gint64 * rstart,
1648     gint64 * rstop)
1650   gboolean step_end = FALSE;
1652   /* see if we need to skip this buffer because of stepping */
1653   switch (current->format) {
1654     case GST_FORMAT_TIME:
1655     {
1656       guint64 end;
1657       gint64 first, last;
1659       if (segment->rate > 0.0) {
1660         if (segment->stop == *cstop)
1661           *rstop = *rstart + current->amount;
1663         first = *rstart;
1664         last = *rstop;
1665       } else {
1666         if (segment->start == *cstart)
1667           *rstart = *rstop + current->amount;
1669         first = *rstop;
1670         last = *rstart;
1671       }
1673       end = current->start + current->amount;
1674       current->position = first - current->start;
1676       if (G_UNLIKELY (segment->abs_rate != 1.0))
1677         current->position /= segment->abs_rate;
1679       GST_DEBUG_OBJECT (sink,
1680           "buffer: %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
1681           GST_TIME_ARGS (first), GST_TIME_ARGS (last));
1682       GST_DEBUG_OBJECT (sink,
1683           "got time step %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT "/%"
1684           GST_TIME_FORMAT, GST_TIME_ARGS (current->position),
1685           GST_TIME_ARGS (last - current->start),
1686           GST_TIME_ARGS (current->amount));
1688       if ((current->flush && current->position >= current->amount)
1689           || last >= end) {
1690         GST_DEBUG_OBJECT (sink, "step ended, we need clipping");
1691         step_end = TRUE;
1692         if (segment->rate > 0.0) {
1693           *rstart = end;
1694           *cstart = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1695         } else {
1696           *rstop = end;
1697           *cstop = gst_segment_to_position (segment, GST_FORMAT_TIME, end);
1698         }
1699       }
1700       GST_DEBUG_OBJECT (sink,
1701           "cstart %" GST_TIME_FORMAT ", rstart %" GST_TIME_FORMAT,
1702           GST_TIME_ARGS (*cstart), GST_TIME_ARGS (*rstart));
1703       GST_DEBUG_OBJECT (sink,
1704           "cstop %" GST_TIME_FORMAT ", rstop %" GST_TIME_FORMAT,
1705           GST_TIME_ARGS (*cstop), GST_TIME_ARGS (*rstop));
1706       break;
1707     }
1708     case GST_FORMAT_BUFFERS:
1709       GST_DEBUG_OBJECT (sink,
1710           "got default step %" G_GUINT64_FORMAT "/%" G_GUINT64_FORMAT,
1711           current->position, current->amount);
1713       if (current->position < current->amount) {
1714         current->position++;
1715       } else {
1716         step_end = TRUE;
1717       }
1718       break;
1719     case GST_FORMAT_DEFAULT:
1720     default:
1721       GST_DEBUG_OBJECT (sink,
1722           "got unknown step %" G_GUINT64_FORMAT "/%" G_GUINT64_FORMAT,
1723           current->position, current->amount);
1724       break;
1725   }
1726   return step_end;
1729 /* with STREAM_LOCK, PREROLL_LOCK
1730  *
1731  * Returns TRUE if the object needs synchronisation and takes therefore
1732  * part in prerolling.
1733  *
1734  * rsstart/rsstop contain the start/stop in stream time.
1735  * rrstart/rrstop contain the start/stop in running time.
1736  */
1737 static gboolean
1738 gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
1739     GstClockTime * rsstart, GstClockTime * rsstop,
1740     GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync,
1741     gboolean * stepped, GstSegment * segment, GstStepInfo * step,
1742     gboolean * step_end)
1744   GstBaseSinkClass *bclass;
1745   GstBuffer *buffer;
1746   GstClockTime start, stop;     /* raw start/stop timestamps */
1747   gint64 cstart, cstop;         /* clipped raw timestamps */
1748   gint64 rstart, rstop;         /* clipped timestamps converted to running time */
1749   GstClockTime sstart, sstop;   /* clipped timestamps converted to stream time */
1750   GstFormat format;
1751   GstBaseSinkPrivate *priv;
1752   gboolean eos;
1754   priv = basesink->priv;
1756   /* start with nothing */
1757   start = stop = GST_CLOCK_TIME_NONE;
1759   if (G_UNLIKELY (GST_IS_EVENT (obj))) {
1760     GstEvent *event = GST_EVENT_CAST (obj);
1762     switch (GST_EVENT_TYPE (event)) {
1763         /* EOS event needs syncing */
1764       case GST_EVENT_EOS:
1765       {
1766         if (basesink->segment.rate >= 0.0) {
1767           sstart = sstop = priv->current_sstop;
1768           if (!GST_CLOCK_TIME_IS_VALID (sstart)) {
1769             /* we have not seen a buffer yet, use the segment values */
1770             sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
1771                 basesink->segment.format, basesink->segment.stop);
1772           }
1773         } else {
1774           sstart = sstop = priv->current_sstart;
1775           if (!GST_CLOCK_TIME_IS_VALID (sstart)) {
1776             /* we have not seen a buffer yet, use the segment values */
1777             sstart = sstop = gst_segment_to_stream_time (&basesink->segment,
1778                 basesink->segment.format, basesink->segment.start);
1779           }
1780         }
1782         rstart = rstop = priv->eos_rtime;
1783         *do_sync = rstart != -1;
1784         GST_DEBUG_OBJECT (basesink, "sync times for EOS %" GST_TIME_FORMAT,
1785             GST_TIME_ARGS (rstart));
1786         /* if we are stepping, we end now */
1787         *step_end = step->valid;
1788         eos = TRUE;
1789         goto eos_done;
1790       }
1791       default:
1792         /* other events do not need syncing */
1793         /* FIXME, maybe NEWSEGMENT might need synchronisation
1794          * since the POSITION query depends on accumulated times and
1795          * we cannot accumulate the current segment before the previous
1796          * one completed.
1797          */
1798         return FALSE;
1799     }
1800   }
1802   eos = FALSE;
1804   /* else do buffer sync code */
1805   buffer = GST_BUFFER_CAST (obj);
1807   bclass = GST_BASE_SINK_GET_CLASS (basesink);
1809   /* just get the times to see if we need syncing, if the start returns -1 we
1810    * don't sync. */
1811   if (bclass->get_times)
1812     bclass->get_times (basesink, buffer, &start, &stop);
1814   if (!GST_CLOCK_TIME_IS_VALID (start)) {
1815     /* we don't need to sync but we still want to get the timestamps for
1816      * tracking the position */
1817     gst_base_sink_get_times (basesink, buffer, &start, &stop);
1818     *do_sync = FALSE;
1819   } else {
1820     *do_sync = TRUE;
1821   }
1823   GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
1824       ", stop: %" GST_TIME_FORMAT ", do_sync %d", GST_TIME_ARGS (start),
1825       GST_TIME_ARGS (stop), *do_sync);
1827   /* collect segment and format for code clarity */
1828   format = segment->format;
1830   /* no timestamp clipping if we did not get a TIME segment format */
1831   if (G_UNLIKELY (format != GST_FORMAT_TIME)) {
1832     cstart = start;
1833     cstop = stop;
1834     /* do running and stream time in TIME format */
1835     format = GST_FORMAT_TIME;
1836     GST_LOG_OBJECT (basesink, "not time format, don't clip");
1837     goto do_times;
1838   }
1840   /* clip, only when we know about time */
1841   if (G_UNLIKELY (!gst_segment_clip (segment, GST_FORMAT_TIME,
1842               (gint64) start, (gint64) stop, &cstart, &cstop))) {
1843     if (step->valid) {
1844       GST_DEBUG_OBJECT (basesink, "step out of segment");
1845       /* when we are stepping, pretend we're at the end of the segment */
1846       if (segment->rate > 0.0) {
1847         cstart = segment->stop;
1848         cstop = segment->stop;
1849       } else {
1850         cstart = segment->start;
1851         cstop = segment->start;
1852       }
1853       goto do_times;
1854     }
1855     goto out_of_segment;
1856   }
1858   if (G_UNLIKELY (start != cstart || stop != cstop)) {
1859     GST_DEBUG_OBJECT (basesink, "clipped to: start %" GST_TIME_FORMAT
1860         ", stop: %" GST_TIME_FORMAT, GST_TIME_ARGS (cstart),
1861         GST_TIME_ARGS (cstop));
1862   }
1864   /* set last stop position */
1865   if (G_LIKELY (cstop != GST_CLOCK_TIME_NONE))
1866     gst_segment_set_last_stop (segment, GST_FORMAT_TIME, cstop);
1867   else
1868     gst_segment_set_last_stop (segment, GST_FORMAT_TIME, cstart);
1870 do_times:
1871   rstart = gst_segment_to_running_time (segment, format, cstart);
1872   rstop = gst_segment_to_running_time (segment, format, cstop);
1874   if (G_UNLIKELY (step->valid)) {
1875     if (!(*step_end = handle_stepping (basesink, segment, step, &cstart, &cstop,
1876                 &rstart, &rstop))) {
1877       /* step is still busy, we discard data when we are flushing */
1878       *stepped = step->flush;
1879       GST_DEBUG_OBJECT (basesink, "stepping busy");
1880     }
1881   }
1882   /* this can produce wrong values if we accumulated non-TIME segments. If this happens,
1883    * upstream is behaving very badly */
1884   sstart = gst_segment_to_stream_time (segment, format, cstart);
1885   sstop = gst_segment_to_stream_time (segment, format, cstop);
1887 eos_done:
1888   /* eos_done label only called when doing EOS, we also stop stepping then */
1889   if (*step_end && step->flush) {
1890     GST_DEBUG_OBJECT (basesink, "flushing step ended");
1891     stop_stepping (basesink, segment, step, rstart, rstop, eos);
1892     *step_end = FALSE;
1893   }
1895   /* save times */
1896   *rsstart = sstart;
1897   *rsstop = sstop;
1898   *rrstart = rstart;
1899   *rrstop = rstop;
1901   /* buffers and EOS always need syncing and preroll */
1902   return TRUE;
1904   /* special cases */
1905 out_of_segment:
1906   {
1907     /* we usually clip in the chain function already but stepping could cause
1908      * the segment to be updated later. we return FALSE so that we don't try
1909      * to sync on it. */
1910     GST_LOG_OBJECT (basesink, "buffer skipped, not in segment");
1911     return FALSE;
1912   }
1915 /* with STREAM_LOCK, PREROLL_LOCK, LOCK
1916  * adjust a timestamp with the latency and timestamp offset */
1917 static GstClockTime
1918 gst_base_sink_adjust_time (GstBaseSink * basesink, GstClockTime time)
1920   GstClockTimeDiff ts_offset;
1922   /* don't do anything funny with invalid timestamps */
1923   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time)))
1924     return time;
1926   time += basesink->priv->latency;
1928   /* apply offset, be carefull for underflows */
1929   ts_offset = basesink->priv->ts_offset;
1930   if (ts_offset < 0) {
1931     ts_offset = -ts_offset;
1932     if (ts_offset < time)
1933       time -= ts_offset;
1934     else
1935       time = 0;
1936   } else
1937     time += ts_offset;
1939   return time;
1942 /**
1943  * gst_base_sink_wait_clock:
1944  * @sink: the sink
1945  * @time: the running_time to be reached
1946  * @jitter: the jitter to be filled with time diff (can be NULL)
1947  *
1948  * This function will block until @time is reached. It is usually called by
1949  * subclasses that use their own internal synchronisation.
1950  *
1951  * If @time is not valid, no sycnhronisation is done and #GST_CLOCK_BADTIME is
1952  * returned. Likewise, if synchronisation is disabled in the element or there
1953  * is no clock, no synchronisation is done and #GST_CLOCK_BADTIME is returned.
1954  *
1955  * This function should only be called with the PREROLL_LOCK held, like when
1956  * receiving an EOS event in the #GstBaseSinkClass.event() vmethod or when
1957  * receiving a buffer in
1958  * the #GstBaseSinkClass.render() vmethod.
1959  *
1960  * The @time argument should be the running_time of when this method should
1961  * return and is not adjusted with any latency or offset configured in the
1962  * sink.
1963  *
1964  * Since 0.10.20
1965  *
1966  * Returns: #GstClockReturn
1967  */
1968 GstClockReturn
1969 gst_base_sink_wait_clock (GstBaseSink * sink, GstClockTime time,
1970     GstClockTimeDiff * jitter)
1972   GstClockID id;
1973   GstClockReturn ret;
1974   GstClock *clock;
1975   GstClockTime base_time;
1977   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time)))
1978     goto invalid_time;
1980   GST_OBJECT_LOCK (sink);
1981   if (G_UNLIKELY (!sink->sync))
1982     goto no_sync;
1984   if (G_UNLIKELY ((clock = GST_ELEMENT_CLOCK (sink)) == NULL))
1985     goto no_clock;
1987   base_time = GST_ELEMENT_CAST (sink)->base_time;
1988   GST_LOG_OBJECT (sink,
1989       "time %" GST_TIME_FORMAT ", base_time %" GST_TIME_FORMAT,
1990       GST_TIME_ARGS (time), GST_TIME_ARGS (base_time));
1992   /* add base_time to running_time to get the time against the clock */
1993   time += base_time;
1995   id = gst_clock_new_single_shot_id (clock, time);
1996   GST_OBJECT_UNLOCK (sink);
1998   /* A blocking wait is performed on the clock. We save the ClockID
1999    * so we can unlock the entry at any time. While we are blocking, we
2000    * release the PREROLL_LOCK so that other threads can interrupt the
2001    * entry. */
2002   sink->clock_id = id;
2003   /* release the preroll lock while waiting */
2004   GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
2006   ret = gst_clock_id_wait (id, jitter);
2008   GST_PAD_PREROLL_LOCK (sink->sinkpad);
2009   gst_clock_id_unref (id);
2010   sink->clock_id = NULL;
2012   return ret;
2014   /* no syncing needed */
2015 invalid_time:
2016   {
2017     GST_DEBUG_OBJECT (sink, "time not valid, no sync needed");
2018     return GST_CLOCK_BADTIME;
2019   }
2020 no_sync:
2021   {
2022     GST_DEBUG_OBJECT (sink, "sync disabled");
2023     GST_OBJECT_UNLOCK (sink);
2024     return GST_CLOCK_BADTIME;
2025   }
2026 no_clock:
2027   {
2028     GST_DEBUG_OBJECT (sink, "no clock, can't sync");
2029     GST_OBJECT_UNLOCK (sink);
2030     return GST_CLOCK_BADTIME;
2031   }
2034 /**
2035  * gst_base_sink_wait_preroll:
2036  * @sink: the sink
2037  *
2038  * If the #GstBaseSinkClass.render() method performs its own synchronisation
2039  * against the clock it must unblock when going from PLAYING to the PAUSED state
2040  * and call this method before continuing to render the remaining data.
2041  *
2042  * This function will block until a state change to PLAYING happens (in which
2043  * case this function returns #GST_FLOW_OK) or the processing must be stopped due
2044  * to a state change to READY or a FLUSH event (in which case this function
2045  * returns #GST_FLOW_WRONG_STATE).
2046  *
2047  * This function should only be called with the PREROLL_LOCK held, like in the
2048  * render function.
2049  *
2050  * Since: 0.10.11
2051  *
2052  * Returns: #GST_FLOW_OK if the preroll completed and processing can
2053  * continue. Any other return value should be returned from the render vmethod.
2054  */
2055 GstFlowReturn
2056 gst_base_sink_wait_preroll (GstBaseSink * sink)
2058   sink->have_preroll = TRUE;
2059   GST_DEBUG_OBJECT (sink, "waiting in preroll for flush or PLAYING");
2060   /* block until the state changes, or we get a flush, or something */
2061   GST_PAD_PREROLL_WAIT (sink->sinkpad);
2062   sink->have_preroll = FALSE;
2063   if (G_UNLIKELY (sink->flushing))
2064     goto stopping;
2065   if (G_UNLIKELY (sink->priv->step_unlock))
2066     goto step_unlocked;
2067   GST_DEBUG_OBJECT (sink, "continue after preroll");
2069   return GST_FLOW_OK;
2071   /* ERRORS */
2072 stopping:
2073   {
2074     GST_DEBUG_OBJECT (sink, "preroll interrupted because of flush");
2075     return GST_FLOW_WRONG_STATE;
2076   }
2077 step_unlocked:
2078   {
2079     sink->priv->step_unlock = FALSE;
2080     GST_DEBUG_OBJECT (sink, "preroll interrupted because of step");
2081     return GST_FLOW_STEP;
2082   }
2085 /**
2086  * gst_base_sink_do_preroll:
2087  * @sink: the sink
2088  * @obj: the object that caused the preroll
2089  *
2090  * If the @sink spawns its own thread for pulling buffers from upstream it
2091  * should call this method after it has pulled a buffer. If the element needed
2092  * to preroll, this function will perform the preroll and will then block
2093  * until the element state is changed.
2094  *
2095  * This function should be called with the PREROLL_LOCK held.
2096  *
2097  * Since 0.10.22
2098  *
2099  * Returns: #GST_FLOW_OK if the preroll completed and processing can
2100  * continue. Any other return value should be returned from the render vmethod.
2101  */
2102 GstFlowReturn
2103 gst_base_sink_do_preroll (GstBaseSink * sink, GstMiniObject * obj)
2105   GstFlowReturn ret;
2107   while (G_UNLIKELY (sink->need_preroll)) {
2108     GST_DEBUG_OBJECT (sink, "prerolling object %p", obj);
2110     ret = gst_base_sink_preroll_object (sink, FALSE, obj);
2111     if (ret != GST_FLOW_OK)
2112       goto preroll_failed;
2114     /* need to recheck here because the commit state could have
2115      * made us not need the preroll anymore */
2116     if (G_LIKELY (sink->need_preroll)) {
2117       /* block until the state changes, or we get a flush, or something */
2118       ret = gst_base_sink_wait_preroll (sink);
2119       if ((ret != GST_FLOW_OK) && (ret != GST_FLOW_STEP))
2120         goto preroll_failed;
2121     }
2122   }
2123   return GST_FLOW_OK;
2125   /* ERRORS */
2126 preroll_failed:
2127   {
2128     GST_DEBUG_OBJECT (sink, "preroll failed %d", ret);
2129     return ret;
2130   }
2133 /**
2134  * gst_base_sink_wait_eos:
2135  * @sink: the sink
2136  * @time: the running_time to be reached
2137  * @jitter: the jitter to be filled with time diff (can be NULL)
2138  *
2139  * This function will block until @time is reached. It is usually called by
2140  * subclasses that use their own internal synchronisation but want to let the
2141  * EOS be handled by the base class.
2142  *
2143  * This function should only be called with the PREROLL_LOCK held, like when
2144  * receiving an EOS event in the ::event vmethod.
2145  *
2146  * The @time argument should be the running_time of when the EOS should happen
2147  * and will be adjusted with any latency and offset configured in the sink.
2148  *
2149  * Since 0.10.15
2150  *
2151  * Returns: #GstFlowReturn
2152  */
2153 GstFlowReturn
2154 gst_base_sink_wait_eos (GstBaseSink * sink, GstClockTime time,
2155     GstClockTimeDiff * jitter)
2157   GstClockReturn status;
2158   GstFlowReturn ret;
2160   do {
2161     GstClockTime stime;
2163     GST_DEBUG_OBJECT (sink, "checking preroll");
2165     /* first wait for the playing state before we can continue */
2166     if (G_UNLIKELY (sink->need_preroll)) {
2167       ret = gst_base_sink_wait_preroll (sink);
2168       if ((ret != GST_FLOW_OK) && (ret != GST_FLOW_STEP))
2169         goto flushing;
2170     }
2172     /* preroll done, we can sync since we are in PLAYING now. */
2173     GST_DEBUG_OBJECT (sink, "possibly waiting for clock to reach %"
2174         GST_TIME_FORMAT, GST_TIME_ARGS (time));
2176     /* compensate for latency and ts_offset. We don't adjust for render delay
2177      * because we don't interact with the device on EOS normally. */
2178     stime = gst_base_sink_adjust_time (sink, time);
2180     /* wait for the clock, this can be interrupted because we got shut down or
2181      * we PAUSED. */
2182     status = gst_base_sink_wait_clock (sink, stime, jitter);
2184     GST_DEBUG_OBJECT (sink, "clock returned %d", status);
2186     /* invalid time, no clock or sync disabled, just continue then */
2187     if (status == GST_CLOCK_BADTIME)
2188       break;
2190     /* waiting could have been interrupted and we can be flushing now */
2191     if (G_UNLIKELY (sink->flushing))
2192       goto flushing;
2194     /* retry if we got unscheduled, which means we did not reach the timeout
2195      * yet. if some other error occures, we continue. */
2196   } while (status == GST_CLOCK_UNSCHEDULED);
2198   GST_DEBUG_OBJECT (sink, "end of stream");
2200   return GST_FLOW_OK;
2202   /* ERRORS */
2203 flushing:
2204   {
2205     GST_DEBUG_OBJECT (sink, "we are flushing");
2206     return GST_FLOW_WRONG_STATE;
2207   }
2210 /* with STREAM_LOCK, PREROLL_LOCK
2211  *
2212  * Make sure we are in PLAYING and synchronize an object to the clock.
2213  *
2214  * If we need preroll, we are not in PLAYING. We try to commit the state
2215  * if needed and then block if we still are not PLAYING.
2216  *
2217  * We start waiting on the clock in PLAYING. If we got interrupted, we
2218  * immediatly try to re-preroll.
2219  *
2220  * Some objects do not need synchronisation (most events) and so this function
2221  * immediatly returns GST_FLOW_OK.
2222  *
2223  * for objects that arrive later than max-lateness to be synchronized to the
2224  * clock have the @late boolean set to TRUE.
2225  *
2226  * This function keeps a running average of the jitter (the diff between the
2227  * clock time and the requested sync time). The jitter is negative for
2228  * objects that arrive in time and positive for late buffers.
2229  *
2230  * does not take ownership of obj.
2231  */
2232 static GstFlowReturn
2233 gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
2234     GstMiniObject * obj, gboolean * late, gboolean * step_end)
2236   GstClockTimeDiff jitter;
2237   gboolean syncable;
2238   GstClockReturn status = GST_CLOCK_OK;
2239   GstClockTime rstart, rstop, sstart, sstop, stime;
2240   gboolean do_sync;
2241   GstBaseSinkPrivate *priv;
2242   GstFlowReturn ret;
2243   GstStepInfo *current, *pending;
2244   gboolean stepped;
2246   priv = basesink->priv;
2248 do_step:
2249   sstart = sstop = rstart = rstop = GST_CLOCK_TIME_NONE;
2250   do_sync = TRUE;
2251   stepped = FALSE;
2253   priv->current_rstart = GST_CLOCK_TIME_NONE;
2255   /* get stepping info */
2256   current = &priv->current_step;
2257   pending = &priv->pending_step;
2259   /* get timing information for this object against the render segment */
2260   syncable = gst_base_sink_get_sync_times (basesink, obj,
2261       &sstart, &sstop, &rstart, &rstop, &do_sync, &stepped, &basesink->segment,
2262       current, step_end);
2264   if (G_UNLIKELY (stepped))
2265     goto step_skipped;
2267   /* a syncable object needs to participate in preroll and
2268    * clocking. All buffers and EOS are syncable. */
2269   if (G_UNLIKELY (!syncable))
2270     goto not_syncable;
2272   /* store timing info for current object */
2273   priv->current_rstart = rstart;
2274   priv->current_rstop = (GST_CLOCK_TIME_IS_VALID (rstop) ? rstop : rstart);
2276   /* save sync time for eos when the previous object needed sync */
2277   priv->eos_rtime = (do_sync ? priv->current_rstop : GST_CLOCK_TIME_NONE);
2279 again:
2280   /* first do preroll, this makes sure we commit our state
2281    * to PAUSED and can continue to PLAYING. We cannot perform
2282    * any clock sync in PAUSED because there is no clock. */
2283   ret = gst_base_sink_do_preroll (basesink, obj);
2284   if (G_UNLIKELY (ret != GST_FLOW_OK))
2285     goto preroll_failed;
2287   /* update the segment with a pending step if the current one is invalid and we
2288    * have a new pending one. We only accept new step updates after a preroll */
2289   if (G_UNLIKELY (pending->valid && !current->valid)) {
2290     start_stepping (basesink, &basesink->segment, pending, current);
2291     goto do_step;
2292   }
2294   /* After rendering we store the position of the last buffer so that we can use
2295    * it to report the position. We need to take the lock here. */
2296   GST_OBJECT_LOCK (basesink);
2297   priv->current_sstart = sstart;
2298   priv->current_sstop = (GST_CLOCK_TIME_IS_VALID (sstop) ? sstop : sstart);
2299   GST_OBJECT_UNLOCK (basesink);
2301   if (!do_sync)
2302     goto done;
2304   /* adjust for latency */
2305   stime = gst_base_sink_adjust_time (basesink, rstart);
2307   /* adjust for render-delay, avoid underflows */
2308   if (GST_CLOCK_TIME_IS_VALID (stime)) {
2309     if (stime > priv->render_delay)
2310       stime -= priv->render_delay;
2311     else
2312       stime = 0;
2313   }
2315   /* preroll done, we can sync since we are in PLAYING now. */
2316   GST_DEBUG_OBJECT (basesink, "possibly waiting for clock to reach %"
2317       GST_TIME_FORMAT ", adjusted %" GST_TIME_FORMAT,
2318       GST_TIME_ARGS (rstart), GST_TIME_ARGS (stime));
2320   /* This function will return immediatly if start == -1, no clock
2321    * or sync is disabled with GST_CLOCK_BADTIME. */
2322   status = gst_base_sink_wait_clock (basesink, stime, &jitter);
2324   GST_DEBUG_OBJECT (basesink, "clock returned %d", status);
2326   /* invalid time, no clock or sync disabled, just render */
2327   if (status == GST_CLOCK_BADTIME)
2328     goto done;
2330   /* waiting could have been interrupted and we can be flushing now */
2331   if (G_UNLIKELY (basesink->flushing))
2332     goto flushing;
2334   /* check for unlocked by a state change, we are not flushing so
2335    * we can try to preroll on the current buffer. */
2336   if (G_UNLIKELY (status == GST_CLOCK_UNSCHEDULED)) {
2337     GST_DEBUG_OBJECT (basesink, "unscheduled, waiting some more");
2338     priv->call_preroll = TRUE;
2339     goto again;
2340   }
2342   /* successful syncing done, record observation */
2343   priv->current_jitter = jitter;
2345   /* check if the object should be dropped */
2346   *late = gst_base_sink_is_too_late (basesink, obj, rstart, rstop,
2347       status, jitter);
2349 done:
2350   return GST_FLOW_OK;
2352   /* ERRORS */
2353 step_skipped:
2354   {
2355     GST_DEBUG_OBJECT (basesink, "skipped stepped object %p", obj);
2356     *late = TRUE;
2357     return GST_FLOW_OK;
2358   }
2359 not_syncable:
2360   {
2361     GST_DEBUG_OBJECT (basesink, "non syncable object %p", obj);
2362     return GST_FLOW_OK;
2363   }
2364 flushing:
2365   {
2366     GST_DEBUG_OBJECT (basesink, "we are flushing");
2367     return GST_FLOW_WRONG_STATE;
2368   }
2369 preroll_failed:
2370   {
2371     GST_DEBUG_OBJECT (basesink, "preroll failed");
2372     *step_end = FALSE;
2373     return ret;
2374   }
2377 static gboolean
2378 gst_base_sink_send_qos (GstBaseSink * basesink,
2379     gdouble proportion, GstClockTime time, GstClockTimeDiff diff)
2381   GstEvent *event;
2382   gboolean res;
2384   /* generate Quality-of-Service event */
2385   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
2386       "qos: proportion: %lf, diff %" G_GINT64_FORMAT ", timestamp %"
2387       GST_TIME_FORMAT, proportion, diff, GST_TIME_ARGS (time));
2389   event = gst_event_new_qos (proportion, diff, time);
2391   /* send upstream */
2392   res = gst_pad_push_event (basesink->sinkpad, event);
2394   return res;
2397 static void
2398 gst_base_sink_perform_qos (GstBaseSink * sink, gboolean dropped)
2400   GstBaseSinkPrivate *priv;
2401   GstClockTime start, stop;
2402   GstClockTimeDiff jitter;
2403   GstClockTime pt, entered, left;
2404   GstClockTime duration;
2405   gdouble rate;
2407   priv = sink->priv;
2409   start = priv->current_rstart;
2411   if (priv->current_step.valid)
2412     return;
2414   /* if Quality-of-Service disabled, do nothing */
2415   if (!g_atomic_int_get (&priv->qos_enabled) ||
2416       !GST_CLOCK_TIME_IS_VALID (start))
2417     return;
2419   stop = priv->current_rstop;
2420   jitter = priv->current_jitter;
2422   if (jitter < 0) {
2423     /* this is the time the buffer entered the sink */
2424     if (start < -jitter)
2425       entered = 0;
2426     else
2427       entered = start + jitter;
2428     left = start;
2429   } else {
2430     /* this is the time the buffer entered the sink */
2431     entered = start + jitter;
2432     /* this is the time the buffer left the sink */
2433     left = start + jitter;
2434   }
2436   /* calculate duration of the buffer */
2437   if (GST_CLOCK_TIME_IS_VALID (stop))
2438     duration = stop - start;
2439   else
2440     duration = GST_CLOCK_TIME_NONE;
2442   /* if we have the time when the last buffer left us, calculate
2443    * processing time */
2444   if (GST_CLOCK_TIME_IS_VALID (priv->last_left)) {
2445     if (entered > priv->last_left) {
2446       pt = entered - priv->last_left;
2447     } else {
2448       pt = 0;
2449     }
2450   } else {
2451     pt = priv->avg_pt;
2452   }
2454   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "start: %" GST_TIME_FORMAT
2455       ", entered %" GST_TIME_FORMAT ", left %" GST_TIME_FORMAT ", pt: %"
2456       GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT ",jitter %"
2457       G_GINT64_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (entered),
2458       GST_TIME_ARGS (left), GST_TIME_ARGS (pt), GST_TIME_ARGS (duration),
2459       jitter);
2461   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "avg_duration: %" GST_TIME_FORMAT
2462       ", avg_pt: %" GST_TIME_FORMAT ", avg_rate: %g",
2463       GST_TIME_ARGS (priv->avg_duration), GST_TIME_ARGS (priv->avg_pt),
2464       priv->avg_rate);
2466   /* collect running averages. for first observations, we copy the
2467    * values */
2468   if (!GST_CLOCK_TIME_IS_VALID (priv->avg_duration))
2469     priv->avg_duration = duration;
2470   else
2471     priv->avg_duration = UPDATE_RUNNING_AVG (priv->avg_duration, duration);
2473   if (!GST_CLOCK_TIME_IS_VALID (priv->avg_pt))
2474     priv->avg_pt = pt;
2475   else
2476     priv->avg_pt = UPDATE_RUNNING_AVG (priv->avg_pt, pt);
2478   if (priv->avg_duration != 0)
2479     rate =
2480         gst_guint64_to_gdouble (priv->avg_pt) /
2481         gst_guint64_to_gdouble (priv->avg_duration);
2482   else
2483     rate = 0.0;
2485   if (GST_CLOCK_TIME_IS_VALID (priv->last_left)) {
2486     if (dropped || priv->avg_rate < 0.0) {
2487       priv->avg_rate = rate;
2488     } else {
2489       if (rate > 1.0)
2490         priv->avg_rate = UPDATE_RUNNING_AVG_N (priv->avg_rate, rate);
2491       else
2492         priv->avg_rate = UPDATE_RUNNING_AVG_P (priv->avg_rate, rate);
2493     }
2494   }
2496   GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink,
2497       "updated: avg_duration: %" GST_TIME_FORMAT ", avg_pt: %" GST_TIME_FORMAT
2498       ", avg_rate: %g", GST_TIME_ARGS (priv->avg_duration),
2499       GST_TIME_ARGS (priv->avg_pt), priv->avg_rate);
2502   if (priv->avg_rate >= 0.0) {
2503     /* if we have a valid rate, start sending QoS messages */
2504     if (priv->current_jitter < 0) {
2505       /* make sure we never go below 0 when adding the jitter to the
2506        * timestamp. */
2507       if (priv->current_rstart < -priv->current_jitter)
2508         priv->current_jitter = -priv->current_rstart;
2509     }
2510     gst_base_sink_send_qos (sink, priv->avg_rate, priv->current_rstart,
2511         priv->current_jitter);
2512   }
2514   /* record when this buffer will leave us */
2515   priv->last_left = left;
2518 /* reset all qos measuring */
2519 static void
2520 gst_base_sink_reset_qos (GstBaseSink * sink)
2522   GstBaseSinkPrivate *priv;
2524   priv = sink->priv;
2526   priv->last_in_time = GST_CLOCK_TIME_NONE;
2527   priv->last_left = GST_CLOCK_TIME_NONE;
2528   priv->avg_duration = GST_CLOCK_TIME_NONE;
2529   priv->avg_pt = GST_CLOCK_TIME_NONE;
2530   priv->avg_rate = -1.0;
2531   priv->avg_render = GST_CLOCK_TIME_NONE;
2532   priv->rendered = 0;
2533   priv->dropped = 0;
2537 /* Checks if the object was scheduled too late.
2538  *
2539  * start/stop contain the raw timestamp start and stop values
2540  * of the object.
2541  *
2542  * status and jitter contain the return values from the clock wait.
2543  *
2544  * returns TRUE if the buffer was too late.
2545  */
2546 static gboolean
2547 gst_base_sink_is_too_late (GstBaseSink * basesink, GstMiniObject * obj,
2548     GstClockTime start, GstClockTime stop,
2549     GstClockReturn status, GstClockTimeDiff jitter)
2551   gboolean late;
2552   gint64 max_lateness;
2553   GstBaseSinkPrivate *priv;
2555   priv = basesink->priv;
2557   late = FALSE;
2559   /* only for objects that were too late */
2560   if (G_LIKELY (status != GST_CLOCK_EARLY))
2561     goto in_time;
2563   max_lateness = basesink->abidata.ABI.max_lateness;
2565   /* check if frame dropping is enabled */
2566   if (max_lateness == -1)
2567     goto no_drop;
2569   /* only check for buffers */
2570   if (G_UNLIKELY (!GST_IS_BUFFER (obj)))
2571     goto not_buffer;
2573   /* can't do check if we don't have a timestamp */
2574   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (start)))
2575     goto no_timestamp;
2577   /* we can add a valid stop time */
2578   if (GST_CLOCK_TIME_IS_VALID (stop))
2579     max_lateness += stop;
2580   else
2581     max_lateness += start;
2583   /* if the jitter bigger than duration and lateness we are too late */
2584   if ((late = start + jitter > max_lateness)) {
2585     GST_CAT_DEBUG_OBJECT (GST_CAT_PERFORMANCE, basesink,
2586         "buffer is too late %" GST_TIME_FORMAT
2587         " > %" GST_TIME_FORMAT, GST_TIME_ARGS (start + jitter),
2588         GST_TIME_ARGS (max_lateness));
2589     /* !!emergency!!, if we did not receive anything valid for more than a
2590      * second, render it anyway so the user sees something */
2591     if (GST_CLOCK_TIME_IS_VALID (priv->last_in_time) &&
2592         start - priv->last_in_time > GST_SECOND) {
2593       late = FALSE;
2594       GST_ELEMENT_WARNING (basesink, CORE, CLOCK,
2595           (_("A lot of buffers are being dropped.")),
2596           ("There may be a timestamping problem, or this computer is too slow."));
2597       GST_CAT_DEBUG_OBJECT (GST_CAT_PERFORMANCE, basesink,
2598           "**emergency** last buffer at %" GST_TIME_FORMAT " > GST_SECOND",
2599           GST_TIME_ARGS (priv->last_in_time));
2600     }
2601   }
2603 done:
2604   if (!late) {
2605     priv->last_in_time = start;
2606   }
2607   return late;
2609   /* all is fine */
2610 in_time:
2611   {
2612     GST_DEBUG_OBJECT (basesink, "object was scheduled in time");
2613     goto done;
2614   }
2615 no_drop:
2616   {
2617     GST_DEBUG_OBJECT (basesink, "frame dropping disabled");
2618     goto done;
2619   }
2620 not_buffer:
2621   {
2622     GST_DEBUG_OBJECT (basesink, "object is not a buffer");
2623     return FALSE;
2624   }
2625 no_timestamp:
2626   {
2627     GST_DEBUG_OBJECT (basesink, "buffer has no timestamp");
2628     return FALSE;
2629   }
2632 /* called before and after calling the render vmethod. It keeps track of how
2633  * much time was spent in the render method and is used to check if we are
2634  * flooded */
2635 static void
2636 gst_base_sink_do_render_stats (GstBaseSink * basesink, gboolean start)
2638   GstBaseSinkPrivate *priv;
2640   priv = basesink->priv;
2642   if (start) {
2643     priv->start = gst_util_get_timestamp ();
2644   } else {
2645     GstClockTime elapsed;
2647     priv->stop = gst_util_get_timestamp ();
2649     elapsed = GST_CLOCK_DIFF (priv->start, priv->stop);
2651     if (!GST_CLOCK_TIME_IS_VALID (priv->avg_render))
2652       priv->avg_render = elapsed;
2653     else
2654       priv->avg_render = UPDATE_RUNNING_AVG (priv->avg_render, elapsed);
2656     GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
2657         "avg_render: %" GST_TIME_FORMAT, GST_TIME_ARGS (priv->avg_render));
2658   }
2661 /* with STREAM_LOCK, PREROLL_LOCK,
2662  *
2663  * Synchronize the object on the clock and then render it.
2664  *
2665  * takes ownership of obj.
2666  */
2667 static GstFlowReturn
2668 gst_base_sink_render_object (GstBaseSink * basesink, GstPad * pad,
2669     gboolean is_list, gpointer obj)
2671   GstFlowReturn ret;
2672   GstBaseSinkClass *bclass;
2673   gboolean late, step_end;
2674   gpointer sync_obj;
2676   GstBaseSinkPrivate *priv;
2678   priv = basesink->priv;
2680   if (is_list) {
2681     /*
2682      * If buffer list, use the first group buffer within the list
2683      * for syncing
2684      */
2685     sync_obj = gst_buffer_list_get (GST_BUFFER_LIST_CAST (obj), 0, 0);
2686     g_assert (NULL != sync_obj);
2687   } else {
2688     sync_obj = obj;
2689   }
2691 again:
2692   late = FALSE;
2693   step_end = FALSE;
2695   /* synchronize this object, non syncable objects return OK
2696    * immediatly. */
2697   ret = gst_base_sink_do_sync (basesink, pad, sync_obj, &late, &step_end);
2698   if (G_UNLIKELY (ret != GST_FLOW_OK))
2699     goto sync_failed;
2701   /* and now render, event or buffer/buffer list. */
2702   if (G_LIKELY (is_list || GST_IS_BUFFER (obj))) {
2703     /* drop late buffers unconditionally, let's hope it's unlikely */
2704     if (G_UNLIKELY (late))
2705       goto dropped;
2707     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2709     if (G_LIKELY ((is_list && bclass->render_list) ||
2710             (!is_list && bclass->render))) {
2711       gint do_qos;
2713       /* read once, to get same value before and after */
2714       do_qos = g_atomic_int_get (&priv->qos_enabled);
2716       GST_DEBUG_OBJECT (basesink, "rendering object %p", obj);
2718       /* record rendering time for QoS and stats */
2719       if (do_qos)
2720         gst_base_sink_do_render_stats (basesink, TRUE);
2722       if (!is_list) {
2723         GstBuffer *buf;
2725         /* For buffer lists do not set last buffer. Creating buffer
2726          * with meaningful data can be done only with memcpy which will
2727          * significantly affect performance */
2728         buf = GST_BUFFER_CAST (obj);
2729         gst_base_sink_set_last_buffer (basesink, buf);
2731         ret = bclass->render (basesink, buf);
2732       } else {
2733         GstBufferList *buflist;
2735         buflist = GST_BUFFER_LIST_CAST (obj);
2737         ret = bclass->render_list (basesink, buflist);
2738       }
2740       if (do_qos)
2741         gst_base_sink_do_render_stats (basesink, FALSE);
2743       if (ret == GST_FLOW_STEP)
2744         goto again;
2746       if (G_UNLIKELY (basesink->flushing))
2747         goto flushing;
2749       priv->rendered++;
2750     }
2751   } else {
2752     GstEvent *event = GST_EVENT_CAST (obj);
2753     gboolean event_res = TRUE;
2754     GstEventType type;
2756     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2758     type = GST_EVENT_TYPE (event);
2760     GST_DEBUG_OBJECT (basesink, "rendering event %p, type %s", obj,
2761         gst_event_type_get_name (type));
2763     if (bclass->event)
2764       event_res = bclass->event (basesink, event);
2766     /* when we get here we could be flushing again when the event handler calls
2767      * _wait_eos(). We have to ignore this object in that case. */
2768     if (G_UNLIKELY (basesink->flushing))
2769       goto flushing;
2771     if (G_LIKELY (event_res)) {
2772       guint32 seqnum;
2774       seqnum = basesink->priv->seqnum = gst_event_get_seqnum (event);
2775       GST_DEBUG_OBJECT (basesink, "Got seqnum #%" G_GUINT32_FORMAT, seqnum);
2777       switch (type) {
2778         case GST_EVENT_EOS:
2779         {
2780           GstMessage *message;
2782           /* the EOS event is completely handled so we mark
2783            * ourselves as being in the EOS state. eos is also
2784            * protected by the object lock so we can read it when
2785            * answering the POSITION query. */
2786           GST_OBJECT_LOCK (basesink);
2787           basesink->eos = TRUE;
2788           GST_OBJECT_UNLOCK (basesink);
2790           /* ok, now we can post the message */
2791           GST_DEBUG_OBJECT (basesink, "Now posting EOS");
2793           message = gst_message_new_eos (GST_OBJECT_CAST (basesink));
2794           gst_message_set_seqnum (message, seqnum);
2795           gst_element_post_message (GST_ELEMENT_CAST (basesink), message);
2796           break;
2797         }
2798         case GST_EVENT_NEWSEGMENT:
2799           /* configure the segment */
2800           gst_base_sink_configure_segment (basesink, pad, event,
2801               &basesink->segment);
2802           break;
2803         case GST_EVENT_SINK_MESSAGE:{
2804           GstMessage *msg = NULL;
2806           gst_event_parse_sink_message (event, &msg);
2808           if (msg)
2809             gst_element_post_message (GST_ELEMENT_CAST (basesink), msg);
2810         }
2811         default:
2812           break;
2813       }
2814     }
2815   }
2817 done:
2818   if (step_end) {
2819     /* the step ended, check if we need to activate a new step */
2820     GST_DEBUG_OBJECT (basesink, "step ended");
2821     stop_stepping (basesink, &basesink->segment, &priv->current_step,
2822         priv->current_rstart, priv->current_rstop, basesink->eos);
2823     goto again;
2824   }
2826   gst_base_sink_perform_qos (basesink, late);
2828   GST_DEBUG_OBJECT (basesink, "object unref after render %p", obj);
2829   gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
2830   return ret;
2832   /* ERRORS */
2833 sync_failed:
2834   {
2835     GST_DEBUG_OBJECT (basesink, "do_sync returned %s", gst_flow_get_name (ret));
2836     goto done;
2837   }
2838 dropped:
2839   {
2840     priv->dropped++;
2841     GST_DEBUG_OBJECT (basesink, "buffer late, dropping");
2842     goto done;
2843   }
2844 flushing:
2845   {
2846     GST_DEBUG_OBJECT (basesink, "we are flushing, ignore object");
2847     gst_mini_object_unref (obj);
2848     return GST_FLOW_WRONG_STATE;
2849   }
2852 /* with STREAM_LOCK, PREROLL_LOCK
2853  *
2854  * Perform preroll on the given object. For buffers this means
2855  * calling the preroll subclass method.
2856  * If that succeeds, the state will be commited.
2857  *
2858  * function does not take ownership of obj.
2859  */
2860 static GstFlowReturn
2861 gst_base_sink_preroll_object (GstBaseSink * basesink, gboolean is_list,
2862     GstMiniObject * obj)
2864   GstFlowReturn ret;
2866   GST_DEBUG_OBJECT (basesink, "prerolling object %p", obj);
2868   /* if it's a buffer, we need to call the preroll method */
2869   if (G_LIKELY (is_list || GST_IS_BUFFER (obj)) && basesink->priv->call_preroll) {
2870     GstBaseSinkClass *bclass;
2871     GstBuffer *buf;
2872     GstClockTime timestamp;
2874     if (is_list) {
2875       buf = gst_buffer_list_get (GST_BUFFER_LIST_CAST (obj), 0, 0);
2876       g_assert (NULL != buf);
2877     } else {
2878       buf = GST_BUFFER_CAST (obj);
2879     }
2881     timestamp = GST_BUFFER_TIMESTAMP (buf);
2883     GST_DEBUG_OBJECT (basesink, "preroll buffer %" GST_TIME_FORMAT,
2884         GST_TIME_ARGS (timestamp));
2886     /*
2887      * For buffer lists do not set last buffer. Creating buffer
2888      * with meaningful data can be done only with memcpy which will
2889      * significantly affect performance
2890      */
2891     if (!is_list) {
2892       gst_base_sink_set_last_buffer (basesink, buf);
2893     }
2895     bclass = GST_BASE_SINK_GET_CLASS (basesink);
2896     if (bclass->preroll)
2897       if ((ret = bclass->preroll (basesink, buf)) != GST_FLOW_OK)
2898         goto preroll_failed;
2900     basesink->priv->call_preroll = FALSE;
2901   }
2903   /* commit state */
2904   if (G_LIKELY (basesink->playing_async)) {
2905     if (G_UNLIKELY (!gst_base_sink_commit_state (basesink)))
2906       goto stopping;
2907   }
2909   return GST_FLOW_OK;
2911   /* ERRORS */
2912 preroll_failed:
2913   {
2914     GST_DEBUG_OBJECT (basesink, "preroll failed, abort state");
2915     gst_element_abort_state (GST_ELEMENT_CAST (basesink));
2916     return ret;
2917   }
2918 stopping:
2919   {
2920     GST_DEBUG_OBJECT (basesink, "stopping while commiting state");
2921     return GST_FLOW_WRONG_STATE;
2922   }
2925 /* with STREAM_LOCK, PREROLL_LOCK
2926  *
2927  * Queue an object for rendering.
2928  * The first prerollable object queued will complete the preroll. If the
2929  * preroll queue if filled, we render all the objects in the queue.
2930  *
2931  * This function takes ownership of the object.
2932  */
2933 static GstFlowReturn
2934 gst_base_sink_queue_object_unlocked (GstBaseSink * basesink, GstPad * pad,
2935     gboolean is_list, gpointer obj, gboolean prerollable)
2937   GstFlowReturn ret = GST_FLOW_OK;
2938   gint length;
2939   GQueue *q;
2941   if (G_UNLIKELY (basesink->need_preroll)) {
2942     if (G_LIKELY (prerollable))
2943       basesink->preroll_queued++;
2945     length = basesink->preroll_queued;
2947     GST_DEBUG_OBJECT (basesink, "now %d prerolled items", length);
2949     /* first prerollable item needs to finish the preroll */
2950     if (length == 1) {
2951       ret = gst_base_sink_preroll_object (basesink, is_list, obj);
2952       if (G_UNLIKELY (ret != GST_FLOW_OK))
2953         goto preroll_failed;
2954     }
2955     /* need to recheck if we need preroll, commmit state during preroll
2956      * could have made us not need more preroll. */
2957     if (G_UNLIKELY (basesink->need_preroll)) {
2958       /* see if we can render now, if we can't add the object to the preroll
2959        * queue. */
2960       if (G_UNLIKELY (length <= basesink->preroll_queue_max_len))
2961         goto more_preroll;
2962     }
2963   }
2964   /* we can start rendering (or blocking) the queued object
2965    * if any. */
2966   q = basesink->preroll_queue;
2967   while (G_UNLIKELY (!g_queue_is_empty (q))) {
2968     GstMiniObject *o;
2970     o = g_queue_pop_head (q);
2971     GST_DEBUG_OBJECT (basesink, "rendering queued object %p", o);
2973     /* do something with the return value */
2974     ret = gst_base_sink_render_object (basesink, pad, FALSE, o);
2975     if (ret != GST_FLOW_OK)
2976       goto dequeue_failed;
2977   }
2979   /* now render the object */
2980   ret = gst_base_sink_render_object (basesink, pad, is_list, obj);
2981   basesink->preroll_queued = 0;
2983   return ret;
2985   /* special cases */
2986 preroll_failed:
2987   {
2988     GST_DEBUG_OBJECT (basesink, "preroll failed, reason %s",
2989         gst_flow_get_name (ret));
2990     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
2991     return ret;
2992   }
2993 more_preroll:
2994   {
2995     /* add object to the queue and return */
2996     GST_DEBUG_OBJECT (basesink, "need more preroll data %d <= %d",
2997         length, basesink->preroll_queue_max_len);
2998     g_queue_push_tail (basesink->preroll_queue, obj);
2999     return GST_FLOW_OK;
3000   }
3001 dequeue_failed:
3002   {
3003     GST_DEBUG_OBJECT (basesink, "rendering queued objects failed, reason %s",
3004         gst_flow_get_name (ret));
3005     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
3006     return ret;
3007   }
3010 /* with STREAM_LOCK
3011  *
3012  * This function grabs the PREROLL_LOCK and adds the object to
3013  * the queue.
3014  *
3015  * This function takes ownership of obj.
3016  */
3017 static GstFlowReturn
3018 gst_base_sink_queue_object (GstBaseSink * basesink, GstPad * pad,
3019     GstMiniObject * obj, gboolean prerollable)
3021   GstFlowReturn ret;
3023   GST_PAD_PREROLL_LOCK (pad);
3024   if (G_UNLIKELY (basesink->flushing))
3025     goto flushing;
3027   if (G_UNLIKELY (basesink->priv->received_eos))
3028     goto was_eos;
3030   ret =
3031       gst_base_sink_queue_object_unlocked (basesink, pad, FALSE, obj,
3032       prerollable);
3033   GST_PAD_PREROLL_UNLOCK (pad);
3035   return ret;
3037   /* ERRORS */
3038 flushing:
3039   {
3040     GST_DEBUG_OBJECT (basesink, "sink is flushing");
3041     GST_PAD_PREROLL_UNLOCK (pad);
3042     gst_mini_object_unref (obj);
3043     return GST_FLOW_WRONG_STATE;
3044   }
3045 was_eos:
3046   {
3047     GST_DEBUG_OBJECT (basesink,
3048         "we are EOS, dropping object, return UNEXPECTED");
3049     GST_PAD_PREROLL_UNLOCK (pad);
3050     gst_mini_object_unref (obj);
3051     return GST_FLOW_UNEXPECTED;
3052   }
3055 static void
3056 gst_base_sink_flush_start (GstBaseSink * basesink, GstPad * pad)
3058   /* make sure we are not blocked on the clock also clear any pending
3059    * eos state. */
3060   gst_base_sink_set_flushing (basesink, pad, TRUE);
3062   /* we grab the stream lock but that is not needed since setting the
3063    * sink to flushing would make sure no state commit is being done
3064    * anymore */
3065   GST_PAD_STREAM_LOCK (pad);
3066   gst_base_sink_reset_qos (basesink);
3067   if (basesink->priv->async_enabled) {
3068     /* and we need to commit our state again on the next
3069      * prerolled buffer */
3070     basesink->playing_async = TRUE;
3071     gst_element_lost_state (GST_ELEMENT_CAST (basesink));
3072   } else {
3073     basesink->priv->have_latency = TRUE;
3074     basesink->need_preroll = FALSE;
3075   }
3076   gst_base_sink_set_last_buffer (basesink, NULL);
3077   GST_PAD_STREAM_UNLOCK (pad);
3080 static void
3081 gst_base_sink_flush_stop (GstBaseSink * basesink, GstPad * pad)
3083   /* unset flushing so we can accept new data, this also flushes out any EOS
3084    * event. */
3085   gst_base_sink_set_flushing (basesink, pad, FALSE);
3087   /* for position reporting */
3088   GST_OBJECT_LOCK (basesink);
3089   basesink->priv->current_sstart = GST_CLOCK_TIME_NONE;
3090   basesink->priv->current_sstop = GST_CLOCK_TIME_NONE;
3091   basesink->priv->eos_rtime = GST_CLOCK_TIME_NONE;
3092   basesink->priv->call_preroll = TRUE;
3093   basesink->priv->current_step.valid = FALSE;
3094   basesink->priv->pending_step.valid = FALSE;
3095   if (basesink->pad_mode == GST_ACTIVATE_PUSH) {
3096     /* we need new segment info after the flush. */
3097     basesink->have_newsegment = FALSE;
3098     gst_segment_init (&basesink->segment, GST_FORMAT_UNDEFINED);
3099     gst_segment_init (basesink->abidata.ABI.clip_segment, GST_FORMAT_UNDEFINED);
3100   }
3101   GST_OBJECT_UNLOCK (basesink);
3104 static gboolean
3105 gst_base_sink_event (GstPad * pad, GstEvent * event)
3107   GstBaseSink *basesink;
3108   gboolean result = TRUE;
3109   GstBaseSinkClass *bclass;
3111   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3113   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3115   GST_DEBUG_OBJECT (basesink, "event %p (%s)", event,
3116       GST_EVENT_TYPE_NAME (event));
3118   switch (GST_EVENT_TYPE (event)) {
3119     case GST_EVENT_EOS:
3120     {
3121       GstFlowReturn ret;
3123       GST_PAD_PREROLL_LOCK (pad);
3124       if (G_UNLIKELY (basesink->flushing))
3125         goto flushing;
3127       if (G_UNLIKELY (basesink->priv->received_eos)) {
3128         /* we can't accept anything when we are EOS */
3129         result = FALSE;
3130         gst_event_unref (event);
3131       } else {
3132         /* we set the received EOS flag here so that we can use it when testing if
3133          * we are prerolled and to refuse more buffers. */
3134         basesink->priv->received_eos = TRUE;
3136         /* EOS is a prerollable object, we call the unlocked version because it
3137          * does not check the received_eos flag. */
3138         ret = gst_base_sink_queue_object_unlocked (basesink, pad,
3139             FALSE, GST_MINI_OBJECT_CAST (event), TRUE);
3140         if (G_UNLIKELY (ret != GST_FLOW_OK))
3141           result = FALSE;
3142       }
3143       GST_PAD_PREROLL_UNLOCK (pad);
3144       break;
3145     }
3146     case GST_EVENT_NEWSEGMENT:
3147     {
3148       GstFlowReturn ret;
3149       gboolean update;
3151       GST_DEBUG_OBJECT (basesink, "newsegment %p", event);
3153       GST_PAD_PREROLL_LOCK (pad);
3154       if (G_UNLIKELY (basesink->flushing))
3155         goto flushing;
3157       gst_event_parse_new_segment_full (event, &update, NULL, NULL, NULL, NULL,
3158           NULL, NULL);
3160       if (G_UNLIKELY (basesink->priv->received_eos && !update)) {
3161         /* we can't accept anything when we are EOS */
3162         result = FALSE;
3163         gst_event_unref (event);
3164       } else {
3165         /* the new segment is a non prerollable item and does not block anything,
3166          * we need to configure the current clipping segment and insert the event
3167          * in the queue to serialize it with the buffers for rendering. */
3168         gst_base_sink_configure_segment (basesink, pad, event,
3169             basesink->abidata.ABI.clip_segment);
3171         ret =
3172             gst_base_sink_queue_object_unlocked (basesink, pad,
3173             FALSE, GST_MINI_OBJECT_CAST (event), FALSE);
3174         if (G_UNLIKELY (ret != GST_FLOW_OK))
3175           result = FALSE;
3176         else {
3177           GST_OBJECT_LOCK (basesink);
3178           basesink->have_newsegment = TRUE;
3179           GST_OBJECT_UNLOCK (basesink);
3180         }
3181       }
3182       GST_PAD_PREROLL_UNLOCK (pad);
3183       break;
3184     }
3185     case GST_EVENT_FLUSH_START:
3186       if (bclass->event)
3187         bclass->event (basesink, event);
3189       GST_DEBUG_OBJECT (basesink, "flush-start %p", event);
3191       gst_base_sink_flush_start (basesink, pad);
3193       gst_event_unref (event);
3194       break;
3195     case GST_EVENT_FLUSH_STOP:
3196       if (bclass->event)
3197         bclass->event (basesink, event);
3199       GST_DEBUG_OBJECT (basesink, "flush-stop %p", event);
3201       gst_base_sink_flush_stop (basesink, pad);
3203       gst_event_unref (event);
3204       break;
3205     default:
3206       /* other events are sent to queue or subclass depending on if they
3207        * are serialized. */
3208       if (GST_EVENT_IS_SERIALIZED (event)) {
3209         gst_base_sink_queue_object (basesink, pad,
3210             GST_MINI_OBJECT_CAST (event), FALSE);
3211       } else {
3212         if (bclass->event)
3213           bclass->event (basesink, event);
3214         gst_event_unref (event);
3215       }
3216       break;
3217   }
3218 done:
3219   gst_object_unref (basesink);
3221   return result;
3223   /* ERRORS */
3224 flushing:
3225   {
3226     GST_DEBUG_OBJECT (basesink, "we are flushing");
3227     GST_PAD_PREROLL_UNLOCK (pad);
3228     result = FALSE;
3229     gst_event_unref (event);
3230     goto done;
3231   }
3234 /* default implementation to calculate the start and end
3235  * timestamps on a buffer, subclasses can override
3236  */
3237 static void
3238 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
3239     GstClockTime * start, GstClockTime * end)
3241   GstClockTime timestamp, duration;
3243   timestamp = GST_BUFFER_TIMESTAMP (buffer);
3244   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
3246     /* get duration to calculate end time */
3247     duration = GST_BUFFER_DURATION (buffer);
3248     if (GST_CLOCK_TIME_IS_VALID (duration)) {
3249       *end = timestamp + duration;
3250     }
3251     *start = timestamp;
3252   }
3255 /* must be called with PREROLL_LOCK */
3256 static gboolean
3257 gst_base_sink_needs_preroll (GstBaseSink * basesink)
3259   gboolean is_prerolled, res;
3261   /* we have 2 cases where the PREROLL_LOCK is released:
3262    *  1) we are blocking in the PREROLL_LOCK and thus are prerolled.
3263    *  2) we are syncing on the clock
3264    */
3265   is_prerolled = basesink->have_preroll || basesink->priv->received_eos;
3266   res = !is_prerolled;
3268   GST_DEBUG_OBJECT (basesink, "have_preroll: %d, EOS: %d => needs preroll: %d",
3269       basesink->have_preroll, basesink->priv->received_eos, res);
3271   return res;
3274 /* with STREAM_LOCK, PREROLL_LOCK
3275  *
3276  * Takes a buffer and compare the timestamps with the last segment.
3277  * If the buffer falls outside of the segment boundaries, drop it.
3278  * Else queue the buffer for preroll and rendering.
3279  *
3280  * This function takes ownership of the buffer.
3281  */
3282 static GstFlowReturn
3283 gst_base_sink_chain_unlocked (GstBaseSink * basesink, GstPad * pad,
3284     gboolean is_list, gpointer obj)
3286   GstBaseSinkClass *bclass;
3287   GstFlowReturn result;
3288   GstClockTime start = GST_CLOCK_TIME_NONE, end = GST_CLOCK_TIME_NONE;
3289   GstSegment *clip_segment;
3290   GstBuffer *time_buf;
3292   if (G_UNLIKELY (basesink->flushing))
3293     goto flushing;
3295   if (G_UNLIKELY (basesink->priv->received_eos))
3296     goto was_eos;
3298   if (is_list) {
3299     time_buf = gst_buffer_list_get (GST_BUFFER_LIST_CAST (obj), 0, 0);
3300     g_assert (NULL != time_buf);
3301   } else {
3302     time_buf = GST_BUFFER_CAST (obj);
3303   }
3305   /* for code clarity */
3306   clip_segment = basesink->abidata.ABI.clip_segment;
3308   if (G_UNLIKELY (!basesink->have_newsegment)) {
3309     gboolean sync;
3311     sync = gst_base_sink_get_sync (basesink);
3312     if (sync) {
3313       GST_ELEMENT_WARNING (basesink, STREAM, FAILED,
3314           (_("Internal data flow problem.")),
3315           ("Received buffer without a new-segment. Assuming timestamps start from 0."));
3316     }
3318     /* this means this sink will assume timestamps start from 0 */
3319     GST_OBJECT_LOCK (basesink);
3320     clip_segment->start = 0;
3321     clip_segment->stop = -1;
3322     basesink->segment.start = 0;
3323     basesink->segment.stop = -1;
3324     basesink->have_newsegment = TRUE;
3325     GST_OBJECT_UNLOCK (basesink);
3326   }
3328   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3330   /* check if the buffer needs to be dropped, we first ask the subclass for the
3331    * start and end */
3332   if (bclass->get_times)
3333     bclass->get_times (basesink, time_buf, &start, &end);
3335   if (!GST_CLOCK_TIME_IS_VALID (start)) {
3336     /* if the subclass does not want sync, we use our own values so that we at
3337      * least clip the buffer to the segment */
3338     gst_base_sink_get_times (basesink, time_buf, &start, &end);
3339   }
3341   GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
3342       ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
3344   /* a dropped buffer does not participate in anything */
3345   if (GST_CLOCK_TIME_IS_VALID (start) &&
3346       (clip_segment->format == GST_FORMAT_TIME)) {
3347     if (G_UNLIKELY (!gst_segment_clip (clip_segment,
3348                 GST_FORMAT_TIME, (gint64) start, (gint64) end, NULL, NULL)))
3349       goto out_of_segment;
3350   }
3352   /* now we can process the buffer in the queue, this function takes ownership
3353    * of the buffer */
3354   result = gst_base_sink_queue_object_unlocked (basesink, pad,
3355       is_list, obj, TRUE);
3356   return result;
3358   /* ERRORS */
3359 flushing:
3360   {
3361     GST_DEBUG_OBJECT (basesink, "sink is flushing");
3362     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
3363     return GST_FLOW_WRONG_STATE;
3364   }
3365 was_eos:
3366   {
3367     GST_DEBUG_OBJECT (basesink,
3368         "we are EOS, dropping object, return UNEXPECTED");
3369     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
3370     return GST_FLOW_UNEXPECTED;
3371   }
3372 out_of_segment:
3373   {
3374     GST_DEBUG_OBJECT (basesink, "dropping buffer, out of clipping segment");
3375     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
3376     return GST_FLOW_OK;
3377   }
3380 /* with STREAM_LOCK
3381  */
3382 static GstFlowReturn
3383 gst_base_sink_chain_main (GstBaseSink * basesink, GstPad * pad,
3384     gboolean is_list, gpointer obj)
3386   GstFlowReturn result;
3388   if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH))
3389     goto wrong_mode;
3391   GST_PAD_PREROLL_LOCK (pad);
3392   result = gst_base_sink_chain_unlocked (basesink, pad, is_list, obj);
3393   GST_PAD_PREROLL_UNLOCK (pad);
3395 done:
3396   return result;
3398   /* ERRORS */
3399 wrong_mode:
3400   {
3401     GST_OBJECT_LOCK (pad);
3402     GST_WARNING_OBJECT (basesink,
3403         "Push on pad %s:%s, but it was not activated in push mode",
3404         GST_DEBUG_PAD_NAME (pad));
3405     GST_OBJECT_UNLOCK (pad);
3406     gst_mini_object_unref (GST_MINI_OBJECT_CAST (obj));
3407     /* we don't post an error message this will signal to the peer
3408      * pushing that EOS is reached. */
3409     result = GST_FLOW_UNEXPECTED;
3410     goto done;
3411   }
3414 static GstFlowReturn
3415 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
3417   GstBaseSink *basesink;
3419   basesink = GST_BASE_SINK (GST_OBJECT_PARENT (pad));
3421   return gst_base_sink_chain_main (basesink, pad, FALSE, buf);
3424 static GstFlowReturn
3425 gst_base_sink_chain_list (GstPad * pad, GstBufferList * list)
3427   GstBaseSink *basesink;
3428   GstBaseSinkClass *bclass;
3429   GstFlowReturn result;
3431   basesink = GST_BASE_SINK (GST_OBJECT_PARENT (pad));
3432   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3434   if (G_LIKELY (bclass->render_list)) {
3435     result = gst_base_sink_chain_main (basesink, pad, TRUE, list);
3436   } else {
3437     GstBufferListIterator *it;
3438     GstBuffer *group;
3440     GST_INFO_OBJECT (pad, "chaining each group in list as a merged buffer");
3442     it = gst_buffer_list_iterate (list);
3444     if (gst_buffer_list_iterator_next_group (it)) {
3445       do {
3446         group = gst_buffer_list_iterator_merge_group (it);
3447         if (group == NULL) {
3448           group = gst_buffer_new ();
3449           GST_CAT_INFO_OBJECT (GST_CAT_SCHEDULING, pad, "chaining empty group");
3450         } else {
3451           GST_CAT_INFO_OBJECT (GST_CAT_SCHEDULING, pad, "chaining group");
3452         }
3453         result = gst_base_sink_chain_main (basesink, pad, FALSE, group);
3454       } while (result == GST_FLOW_OK
3455           && gst_buffer_list_iterator_next_group (it));
3456     } else {
3457       GST_CAT_INFO_OBJECT (GST_CAT_SCHEDULING, pad, "chaining empty group");
3458       result =
3459           gst_base_sink_chain_main (basesink, pad, FALSE, gst_buffer_new ());
3460     }
3461     gst_buffer_list_iterator_free (it);
3462     gst_buffer_list_unref (list);
3463   }
3464   return result;
3468 static gboolean
3469 gst_base_sink_default_do_seek (GstBaseSink * sink, GstSegment * segment)
3471   gboolean res = TRUE;
3473   /* update our offset if the start/stop position was updated */
3474   if (segment->format == GST_FORMAT_BYTES) {
3475     segment->time = segment->start;
3476   } else if (segment->start == 0) {
3477     /* seek to start, we can implement a default for this. */
3478     segment->time = 0;
3479   } else {
3480     res = FALSE;
3481     GST_INFO_OBJECT (sink, "Can't do a default seek");
3482   }
3484   return res;
3487 #define SEEK_TYPE_IS_RELATIVE(t) (((t) != GST_SEEK_TYPE_NONE) && ((t) != GST_SEEK_TYPE_SET))
3489 static gboolean
3490 gst_base_sink_default_prepare_seek_segment (GstBaseSink * sink,
3491     GstEvent * event, GstSegment * segment)
3493   /* By default, we try one of 2 things:
3494    *   - For absolute seek positions, convert the requested position to our
3495    *     configured processing format and place it in the output segment \
3496    *   - For relative seek positions, convert our current (input) values to the
3497    *     seek format, adjust by the relative seek offset and then convert back to
3498    *     the processing format
3499    */
3500   GstSeekType cur_type, stop_type;
3501   gint64 cur, stop;
3502   GstSeekFlags flags;
3503   GstFormat seek_format, dest_format;
3504   gdouble rate;
3505   gboolean update;
3506   gboolean res = TRUE;
3508   gst_event_parse_seek (event, &rate, &seek_format, &flags,
3509       &cur_type, &cur, &stop_type, &stop);
3510   dest_format = segment->format;
3512   if (seek_format == dest_format) {
3513     gst_segment_set_seek (segment, rate, seek_format, flags,
3514         cur_type, cur, stop_type, stop, &update);
3515     return TRUE;
3516   }
3518   if (cur_type != GST_SEEK_TYPE_NONE) {
3519     /* FIXME: Handle seek_cur & seek_end by converting the input segment vals */
3520     res =
3521         gst_pad_query_convert (sink->sinkpad, seek_format, cur, &dest_format,
3522         &cur);
3523     cur_type = GST_SEEK_TYPE_SET;
3524   }
3526   if (res && stop_type != GST_SEEK_TYPE_NONE) {
3527     /* FIXME: Handle seek_cur & seek_end by converting the input segment vals */
3528     res =
3529         gst_pad_query_convert (sink->sinkpad, seek_format, stop, &dest_format,
3530         &stop);
3531     stop_type = GST_SEEK_TYPE_SET;
3532   }
3534   /* And finally, configure our output segment in the desired format */
3535   gst_segment_set_seek (segment, rate, dest_format, flags, cur_type, cur,
3536       stop_type, stop, &update);
3538   if (!res)
3539     goto no_format;
3541   return res;
3543 no_format:
3544   {
3545     GST_DEBUG_OBJECT (sink, "undefined format given, seek aborted.");
3546     return FALSE;
3547   }
3550 /* perform a seek, only executed in pull mode */
3551 static gboolean
3552 gst_base_sink_perform_seek (GstBaseSink * sink, GstPad * pad, GstEvent * event)
3554   gboolean flush;
3555   gdouble rate;
3556   GstFormat seek_format, dest_format;
3557   GstSeekFlags flags;
3558   GstSeekType cur_type, stop_type;
3559   gboolean seekseg_configured = FALSE;
3560   gint64 cur, stop;
3561   gboolean update, res = TRUE;
3562   GstSegment seeksegment;
3564   dest_format = sink->segment.format;
3566   if (event) {
3567     GST_DEBUG_OBJECT (sink, "performing seek with event %p", event);
3568     gst_event_parse_seek (event, &rate, &seek_format, &flags,
3569         &cur_type, &cur, &stop_type, &stop);
3571     flush = flags & GST_SEEK_FLAG_FLUSH;
3572   } else {
3573     GST_DEBUG_OBJECT (sink, "performing seek without event");
3574     flush = FALSE;
3575   }
3577   if (flush) {
3578     GST_DEBUG_OBJECT (sink, "flushing upstream");
3579     gst_pad_push_event (pad, gst_event_new_flush_start ());
3580     gst_base_sink_flush_start (sink, pad);
3581   } else {
3582     GST_DEBUG_OBJECT (sink, "pausing pulling thread");
3583   }
3585   GST_PAD_STREAM_LOCK (pad);
3587   /* If we configured the seeksegment above, don't overwrite it now. Otherwise
3588    * copy the current segment info into the temp segment that we can actually
3589    * attempt the seek with. We only update the real segment if the seek suceeds. */
3590   if (!seekseg_configured) {
3591     memcpy (&seeksegment, &sink->segment, sizeof (GstSegment));
3593     /* now configure the final seek segment */
3594     if (event) {
3595       if (sink->segment.format != seek_format) {
3596         /* OK, here's where we give the subclass a chance to convert the relative
3597          * seek into an absolute one in the processing format. We set up any
3598          * absolute seek above, before taking the stream lock. */
3599         if (!gst_base_sink_default_prepare_seek_segment (sink, event,
3600                 &seeksegment)) {
3601           GST_DEBUG_OBJECT (sink,
3602               "Preparing the seek failed after flushing. " "Aborting seek");
3603           res = FALSE;
3604         }
3605       } else {
3606         /* The seek format matches our processing format, no need to ask the
3607          * the subclass to configure the segment. */
3608         gst_segment_set_seek (&seeksegment, rate, seek_format, flags,
3609             cur_type, cur, stop_type, stop, &update);
3610       }
3611     }
3612     /* Else, no seek event passed, so we're just (re)starting the
3613        current segment. */
3614   }
3616   if (res) {
3617     GST_DEBUG_OBJECT (sink, "segment configured from %" G_GINT64_FORMAT
3618         " to %" G_GINT64_FORMAT ", position %" G_GINT64_FORMAT,
3619         seeksegment.start, seeksegment.stop, seeksegment.last_stop);
3621     /* do the seek, segment.last_stop contains the new position. */
3622     res = gst_base_sink_default_do_seek (sink, &seeksegment);
3623   }
3626   if (flush) {
3627     GST_DEBUG_OBJECT (sink, "stop flushing upstream");
3628     gst_pad_push_event (pad, gst_event_new_flush_stop ());
3629     gst_base_sink_flush_stop (sink, pad);
3630   } else if (res && sink->abidata.ABI.running) {
3631     /* we are running the current segment and doing a non-flushing seek,
3632      * close the segment first based on the last_stop. */
3633     GST_DEBUG_OBJECT (sink, "closing running segment %" G_GINT64_FORMAT
3634         " to %" G_GINT64_FORMAT, sink->segment.start, sink->segment.last_stop);
3635   }
3637   /* The subclass must have converted the segment to the processing format
3638    * by now */
3639   if (res && seeksegment.format != dest_format) {
3640     GST_DEBUG_OBJECT (sink, "Subclass failed to prepare a seek segment "
3641         "in the correct format. Aborting seek.");
3642     res = FALSE;
3643   }
3645   /* if successfull seek, we update our real segment and push
3646    * out the new segment. */
3647   if (res) {
3648     memcpy (&sink->segment, &seeksegment, sizeof (GstSegment));
3650     if (sink->segment.flags & GST_SEEK_FLAG_SEGMENT) {
3651       gst_element_post_message (GST_ELEMENT (sink),
3652           gst_message_new_segment_start (GST_OBJECT (sink),
3653               sink->segment.format, sink->segment.last_stop));
3654     }
3655   }
3657   sink->priv->discont = TRUE;
3658   sink->abidata.ABI.running = TRUE;
3660   GST_PAD_STREAM_UNLOCK (pad);
3662   return res;
3665 static void
3666 set_step_info (GstBaseSink * sink, GstStepInfo * current, GstStepInfo * pending,
3667     guint seqnum, GstFormat format, guint64 amount, gdouble rate,
3668     gboolean flush, gboolean intermediate)
3670   GST_OBJECT_LOCK (sink);
3671   pending->seqnum = seqnum;
3672   pending->format = format;
3673   pending->amount = amount;
3674   pending->position = 0;
3675   pending->rate = rate;
3676   pending->flush = flush;
3677   pending->intermediate = intermediate;
3678   pending->valid = TRUE;
3679   /* flush invalidates the current stepping segment */
3680   if (flush)
3681     current->valid = FALSE;
3682   GST_OBJECT_UNLOCK (sink);
3685 static gboolean
3686 gst_base_sink_perform_step (GstBaseSink * sink, GstPad * pad, GstEvent * event)
3688   GstBaseSinkPrivate *priv;
3689   GstBaseSinkClass *bclass;
3690   gboolean flush, intermediate;
3691   gdouble rate;
3692   GstFormat format;
3693   guint64 amount;
3694   guint seqnum;
3695   GstStepInfo *pending, *current;
3696   GstMessage *message;
3698   bclass = GST_BASE_SINK_GET_CLASS (sink);
3699   priv = sink->priv;
3701   GST_DEBUG_OBJECT (sink, "performing step with event %p", event);
3703   gst_event_parse_step (event, &format, &amount, &rate, &flush, &intermediate);
3704   seqnum = gst_event_get_seqnum (event);
3706   pending = &priv->pending_step;
3707   current = &priv->current_step;
3709   /* post message first */
3710   message = gst_message_new_step_start (GST_OBJECT (sink), FALSE, format,
3711       amount, rate, flush, intermediate);
3712   gst_message_set_seqnum (message, seqnum);
3713   gst_element_post_message (GST_ELEMENT (sink), message);
3715   if (flush) {
3716     /* we need to call ::unlock before locking PREROLL_LOCK
3717      * since we lock it before going into ::render */
3718     if (bclass->unlock)
3719       bclass->unlock (sink);
3721     GST_PAD_PREROLL_LOCK (sink->sinkpad);
3722     /* now that we have the PREROLL lock, clear our unlock request */
3723     if (bclass->unlock_stop)
3724       bclass->unlock_stop (sink);
3726     /* update the stepinfo and make it valid */
3727     set_step_info (sink, current, pending, seqnum, format, amount, rate, flush,
3728         intermediate);
3730     if (sink->priv->async_enabled) {
3731       /* and we need to commit our state again on the next
3732        * prerolled buffer */
3733       sink->playing_async = TRUE;
3734       priv->pending_step.need_preroll = TRUE;
3735       sink->need_preroll = FALSE;
3736       gst_element_lost_state_full (GST_ELEMENT_CAST (sink), FALSE);
3737     } else {
3738       sink->priv->have_latency = TRUE;
3739       sink->need_preroll = FALSE;
3740     }
3741     priv->current_sstart = GST_CLOCK_TIME_NONE;
3742     priv->current_sstop = GST_CLOCK_TIME_NONE;
3743     priv->eos_rtime = GST_CLOCK_TIME_NONE;
3744     priv->call_preroll = TRUE;
3745     gst_base_sink_set_last_buffer (sink, NULL);
3746     gst_base_sink_reset_qos (sink);
3748     if (sink->clock_id) {
3749       gst_clock_id_unschedule (sink->clock_id);
3750     }
3752     if (sink->have_preroll) {
3753       GST_DEBUG_OBJECT (sink, "signal waiter");
3754       priv->step_unlock = TRUE;
3755       GST_PAD_PREROLL_SIGNAL (sink->sinkpad);
3756     }
3757     GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
3758   } else {
3759     /* update the stepinfo and make it valid */
3760     set_step_info (sink, current, pending, seqnum, format, amount, rate, flush,
3761         intermediate);
3762   }
3764   return TRUE;
3767 /* with STREAM_LOCK
3768  */
3769 static void
3770 gst_base_sink_loop (GstPad * pad)
3772   GstBaseSink *basesink;
3773   GstBuffer *buf = NULL;
3774   GstFlowReturn result;
3775   guint blocksize;
3776   guint64 offset;
3778   basesink = GST_BASE_SINK (GST_OBJECT_PARENT (pad));
3780   g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
3782   if ((blocksize = basesink->priv->blocksize) == 0)
3783     blocksize = -1;
3785   offset = basesink->segment.last_stop;
3787   GST_DEBUG_OBJECT (basesink, "pulling %" G_GUINT64_FORMAT ", %u",
3788       offset, blocksize);
3790   result = gst_pad_pull_range (pad, offset, blocksize, &buf);
3791   if (G_UNLIKELY (result != GST_FLOW_OK))
3792     goto paused;
3794   if (G_UNLIKELY (buf == NULL))
3795     goto no_buffer;
3797   offset += GST_BUFFER_SIZE (buf);
3799   gst_segment_set_last_stop (&basesink->segment, GST_FORMAT_BYTES, offset);
3801   GST_PAD_PREROLL_LOCK (pad);
3802   result = gst_base_sink_chain_unlocked (basesink, pad, FALSE, buf);
3803   GST_PAD_PREROLL_UNLOCK (pad);
3804   if (G_UNLIKELY (result != GST_FLOW_OK))
3805     goto paused;
3807   return;
3809   /* ERRORS */
3810 paused:
3811   {
3812     GST_LOG_OBJECT (basesink, "pausing task, reason %s",
3813         gst_flow_get_name (result));
3814     gst_pad_pause_task (pad);
3815     /* fatal errors and NOT_LINKED cause EOS */
3816     if (GST_FLOW_IS_FATAL (result) || result == GST_FLOW_NOT_LINKED) {
3817       if (result == GST_FLOW_UNEXPECTED) {
3818         /* perform EOS logic */
3819         if (basesink->segment.flags & GST_SEEK_FLAG_SEGMENT) {
3820           gst_element_post_message (GST_ELEMENT_CAST (basesink),
3821               gst_message_new_segment_done (GST_OBJECT_CAST (basesink),
3822                   basesink->segment.format, basesink->segment.last_stop));
3823         } else {
3824           gst_base_sink_event (pad, gst_event_new_eos ());
3825         }
3826       } else {
3827         /* for fatal errors we post an error message, post the error
3828          * first so the app knows about the error first. */
3829         GST_ELEMENT_ERROR (basesink, STREAM, FAILED,
3830             (_("Internal data stream error.")),
3831             ("stream stopped, reason %s", gst_flow_get_name (result)));
3832         gst_base_sink_event (pad, gst_event_new_eos ());
3833       }
3834     }
3835     return;
3836   }
3837 no_buffer:
3838   {
3839     GST_LOG_OBJECT (basesink, "no buffer, pausing");
3840     GST_ELEMENT_ERROR (basesink, STREAM, FAILED,
3841         (_("Internal data flow error.")), ("element returned NULL buffer"));
3842     result = GST_FLOW_ERROR;
3843     goto paused;
3844   }
3847 static gboolean
3848 gst_base_sink_set_flushing (GstBaseSink * basesink, GstPad * pad,
3849     gboolean flushing)
3851   GstBaseSinkClass *bclass;
3853   bclass = GST_BASE_SINK_GET_CLASS (basesink);
3855   if (flushing) {
3856     /* unlock any subclasses, we need to do this before grabbing the
3857      * PREROLL_LOCK since we hold this lock before going into ::render. */
3858     if (bclass->unlock)
3859       bclass->unlock (basesink);
3860   }
3862   GST_PAD_PREROLL_LOCK (pad);
3863   basesink->flushing = flushing;
3864   if (flushing) {
3865     /* step 1, now that we have the PREROLL lock, clear our unlock request */
3866     if (bclass->unlock_stop)
3867       bclass->unlock_stop (basesink);
3869     /* set need_preroll before we unblock the clock. If the clock is unblocked
3870      * before timing out, we can reuse the buffer for preroll. */
3871     basesink->need_preroll = TRUE;
3873     /* step 2, unblock clock sync (if any) or any other blocking thing */
3874     if (basesink->clock_id) {
3875       gst_clock_id_unschedule (basesink->clock_id);
3876     }
3878     /* flush out the data thread if it's locked in finish_preroll, this will
3879      * also flush out the EOS state */
3880     GST_DEBUG_OBJECT (basesink,
3881         "flushing out data thread, need preroll to TRUE");
3882     gst_base_sink_preroll_queue_flush (basesink, pad);
3883   }
3884   GST_PAD_PREROLL_UNLOCK (pad);
3886   return TRUE;
3889 static gboolean
3890 gst_base_sink_default_activate_pull (GstBaseSink * basesink, gboolean active)
3892   gboolean result;
3894   if (active) {
3895     /* start task */
3896     result = gst_pad_start_task (basesink->sinkpad,
3897         (GstTaskFunction) gst_base_sink_loop, basesink->sinkpad);
3898   } else {
3899     /* step 2, make sure streaming finishes */
3900     result = gst_pad_stop_task (basesink->sinkpad);
3901   }
3903   return result;
3906 static gboolean
3907 gst_base_sink_pad_activate (GstPad * pad)
3909   gboolean result = FALSE;
3910   GstBaseSink *basesink;
3912   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3914   GST_DEBUG_OBJECT (basesink, "Trying pull mode first");
3916   gst_base_sink_set_flushing (basesink, pad, FALSE);
3918   /* we need to have the pull mode enabled */
3919   if (!basesink->can_activate_pull) {
3920     GST_DEBUG_OBJECT (basesink, "pull mode disabled");
3921     goto fallback;
3922   }
3924   /* check if downstreams supports pull mode at all */
3925   if (!gst_pad_check_pull_range (pad)) {
3926     GST_DEBUG_OBJECT (basesink, "pull mode not supported");
3927     goto fallback;
3928   }
3930   /* set the pad mode before starting the task so that it's in the
3931    * correct state for the new thread. also the sink set_caps and get_caps
3932    * function checks this */
3933   basesink->pad_mode = GST_ACTIVATE_PULL;
3935   /* we first try to negotiate a format so that when we try to activate
3936    * downstream, it knows about our format */
3937   if (!gst_base_sink_negotiate_pull (basesink)) {
3938     GST_DEBUG_OBJECT (basesink, "failed to negotiate in pull mode");
3939     goto fallback;
3940   }
3942   /* ok activate now */
3943   if (!gst_pad_activate_pull (pad, TRUE)) {
3944     /* clear any pending caps */
3945     GST_OBJECT_LOCK (basesink);
3946     gst_caps_replace (&basesink->priv->pull_caps, NULL);
3947     GST_OBJECT_UNLOCK (basesink);
3948     GST_DEBUG_OBJECT (basesink, "failed to activate in pull mode");
3949     goto fallback;
3950   }
3952   GST_DEBUG_OBJECT (basesink, "Success activating pull mode");
3953   result = TRUE;
3954   goto done;
3956   /* push mode fallback */
3957 fallback:
3958   GST_DEBUG_OBJECT (basesink, "Falling back to push mode");
3959   if ((result = gst_pad_activate_push (pad, TRUE))) {
3960     GST_DEBUG_OBJECT (basesink, "Success activating push mode");
3961   }
3963 done:
3964   if (!result) {
3965     GST_WARNING_OBJECT (basesink, "Could not activate pad in either mode");
3966     gst_base_sink_set_flushing (basesink, pad, TRUE);
3967   }
3969   gst_object_unref (basesink);
3971   return result;
3974 static gboolean
3975 gst_base_sink_pad_activate_push (GstPad * pad, gboolean active)
3977   gboolean result;
3978   GstBaseSink *basesink;
3980   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
3982   if (active) {
3983     if (!basesink->can_activate_push) {
3984       result = FALSE;
3985       basesink->pad_mode = GST_ACTIVATE_NONE;
3986     } else {
3987       result = TRUE;
3988       basesink->pad_mode = GST_ACTIVATE_PUSH;
3989     }
3990   } else {
3991     if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH)) {
3992       g_warning ("Internal GStreamer activation error!!!");
3993       result = FALSE;
3994     } else {
3995       gst_base_sink_set_flushing (basesink, pad, TRUE);
3996       result = TRUE;
3997       basesink->pad_mode = GST_ACTIVATE_NONE;
3998     }
3999   }
4001   gst_object_unref (basesink);
4003   return result;
4006 static gboolean
4007 gst_base_sink_negotiate_pull (GstBaseSink * basesink)
4009   GstCaps *caps;
4010   gboolean result;
4012   result = FALSE;
4014   /* this returns the intersection between our caps and the peer caps. If there
4015    * is no peer, it returns NULL and we can't operate in pull mode so we can
4016    * fail the negotiation. */
4017   caps = gst_pad_get_allowed_caps (GST_BASE_SINK_PAD (basesink));
4018   if (caps == NULL || gst_caps_is_empty (caps))
4019     goto no_caps_possible;
4021   GST_DEBUG_OBJECT (basesink, "allowed caps: %" GST_PTR_FORMAT, caps);
4023   caps = gst_caps_make_writable (caps);
4024   /* get the first (prefered) format */
4025   gst_caps_truncate (caps);
4026   /* try to fixate */
4027   gst_pad_fixate_caps (GST_BASE_SINK_PAD (basesink), caps);
4029   GST_DEBUG_OBJECT (basesink, "fixated to: %" GST_PTR_FORMAT, caps);
4031   if (gst_caps_is_any (caps)) {
4032     GST_DEBUG_OBJECT (basesink, "caps were ANY after fixating, "
4033         "allowing pull()");
4034     /* neither side has template caps in this case, so they are prepared for
4035        pull() without setcaps() */
4036     result = TRUE;
4037   } else if (gst_caps_is_fixed (caps)) {
4038     if (!gst_pad_set_caps (GST_BASE_SINK_PAD (basesink), caps))
4039       goto could_not_set_caps;
4041     GST_OBJECT_LOCK (basesink);
4042     gst_caps_replace (&basesink->priv->pull_caps, caps);
4043     GST_OBJECT_UNLOCK (basesink);
4045     result = TRUE;
4046   }
4048   gst_caps_unref (caps);
4050   return result;
4052 no_caps_possible:
4053   {
4054     GST_INFO_OBJECT (basesink, "Pipeline could not agree on caps");
4055     GST_DEBUG_OBJECT (basesink, "get_allowed_caps() returned EMPTY");
4056     if (caps)
4057       gst_caps_unref (caps);
4058     return FALSE;
4059   }
4060 could_not_set_caps:
4061   {
4062     GST_INFO_OBJECT (basesink, "Could not set caps: %" GST_PTR_FORMAT, caps);
4063     gst_caps_unref (caps);
4064     return FALSE;
4065   }
4068 /* this won't get called until we implement an activate function */
4069 static gboolean
4070 gst_base_sink_pad_activate_pull (GstPad * pad, gboolean active)
4072   gboolean result = FALSE;
4073   GstBaseSink *basesink;
4074   GstBaseSinkClass *bclass;
4076   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
4077   bclass = GST_BASE_SINK_GET_CLASS (basesink);
4079   if (active) {
4080     GstFormat format;
4081     gint64 duration;
4083     /* we mark we have a newsegment here because pull based
4084      * mode works just fine without having a newsegment before the
4085      * first buffer */
4086     format = GST_FORMAT_BYTES;
4088     gst_segment_init (&basesink->segment, format);
4089     gst_segment_init (basesink->abidata.ABI.clip_segment, format);
4090     GST_OBJECT_LOCK (basesink);
4091     basesink->have_newsegment = TRUE;
4092     GST_OBJECT_UNLOCK (basesink);
4094     /* get the peer duration in bytes */
4095     result = gst_pad_query_peer_duration (pad, &format, &duration);
4096     if (result) {
4097       GST_DEBUG_OBJECT (basesink,
4098           "setting duration in bytes to %" G_GINT64_FORMAT, duration);
4099       gst_segment_set_duration (basesink->abidata.ABI.clip_segment, format,
4100           duration);
4101       gst_segment_set_duration (&basesink->segment, format, duration);
4102     } else {
4103       GST_DEBUG_OBJECT (basesink, "unknown duration");
4104     }
4106     if (bclass->activate_pull)
4107       result = bclass->activate_pull (basesink, TRUE);
4108     else
4109       result = FALSE;
4111     if (!result)
4112       goto activate_failed;
4114   } else {
4115     if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PULL)) {
4116       g_warning ("Internal GStreamer activation error!!!");
4117       result = FALSE;
4118     } else {
4119       result = gst_base_sink_set_flushing (basesink, pad, TRUE);
4120       if (bclass->activate_pull)
4121         result &= bclass->activate_pull (basesink, FALSE);
4122       basesink->pad_mode = GST_ACTIVATE_NONE;
4123       /* clear any pending caps */
4124       GST_OBJECT_LOCK (basesink);
4125       gst_caps_replace (&basesink->priv->pull_caps, NULL);
4126       GST_OBJECT_UNLOCK (basesink);
4127     }
4128   }
4129   gst_object_unref (basesink);
4131   return result;
4133   /* ERRORS */
4134 activate_failed:
4135   {
4136     /* reset, as starting the thread failed */
4137     basesink->pad_mode = GST_ACTIVATE_NONE;
4139     GST_ERROR_OBJECT (basesink, "subclass failed to activate in pull mode");
4140     return FALSE;
4141   }
4144 /* send an event to our sinkpad peer. */
4145 static gboolean
4146 gst_base_sink_send_event (GstElement * element, GstEvent * event)
4148   GstPad *pad;
4149   GstBaseSink *basesink = GST_BASE_SINK (element);
4150   gboolean forward, result = TRUE;
4151   GstActivateMode mode;
4153   GST_OBJECT_LOCK (element);
4154   /* get the pad and the scheduling mode */
4155   pad = gst_object_ref (basesink->sinkpad);
4156   mode = basesink->pad_mode;
4157   GST_OBJECT_UNLOCK (element);
4159   /* only push UPSTREAM events upstream */
4160   forward = GST_EVENT_IS_UPSTREAM (event);
4162   switch (GST_EVENT_TYPE (event)) {
4163     case GST_EVENT_LATENCY:
4164     {
4165       GstClockTime latency;
4167       gst_event_parse_latency (event, &latency);
4169       /* store the latency. We use this to adjust the running_time before syncing
4170        * it to the clock. */
4171       GST_OBJECT_LOCK (element);
4172       basesink->priv->latency = latency;
4173       if (!basesink->priv->have_latency)
4174         forward = FALSE;
4175       GST_OBJECT_UNLOCK (element);
4176       GST_DEBUG_OBJECT (basesink, "latency set to %" GST_TIME_FORMAT,
4177           GST_TIME_ARGS (latency));
4179       /* We forward this event so that all elements know about the global pipeline
4180        * latency. This is interesting for an element when it wants to figure out
4181        * when a particular piece of data will be rendered. */
4182       break;
4183     }
4184     case GST_EVENT_SEEK:
4185       /* in pull mode we will execute the seek */
4186       if (mode == GST_ACTIVATE_PULL)
4187         result = gst_base_sink_perform_seek (basesink, pad, event);
4188       break;
4189     case GST_EVENT_STEP:
4190       result = gst_base_sink_perform_step (basesink, pad, event);
4191       forward = FALSE;
4192       break;
4193     default:
4194       break;
4195   }
4197   if (forward) {
4198     result = gst_pad_push_event (pad, event);
4199   } else {
4200     /* not forwarded, unref the event */
4201     gst_event_unref (event);
4202   }
4204   gst_object_unref (pad);
4205   return result;
4208 static gboolean
4209 gst_base_sink_peer_query (GstBaseSink * sink, GstQuery * query)
4211   GstPad *peer;
4212   gboolean res = FALSE;
4214   if ((peer = gst_pad_get_peer (sink->sinkpad))) {
4215     res = gst_pad_query (peer, query);
4216     gst_object_unref (peer);
4217   }
4218   return res;
4221 /* get the end position of the last seen object, this is used
4222  * for EOS and for making sure that we don't report a position we
4223  * have not reached yet. With LOCK. */
4224 static gboolean
4225 gst_base_sink_get_position_last (GstBaseSink * basesink, GstFormat format,
4226     gint64 * cur, gboolean * upstream)
4228   GstFormat oformat;
4229   GstSegment *segment;
4230   gboolean ret = TRUE;
4232   segment = &basesink->segment;
4233   oformat = segment->format;