2fabac9f27b8a94508406c0e6dead5414464100a
[platform/upstream/gstreamer.git] / libs / gst / base / gstbasesink.c
1 /* GStreamer
2  * Copyright (C) 2005 Wim Taymans <wim@fluendo.com>
3  *
4  * gstbasesink.c: 
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  */
21
22 /**
23  * SECTION:gstbasesink
24  * @short_description: Base class for sink elements
25  * @see_also: #GstBaseTransformc, #GstBaseSource
26  *
27  * This class is for elements that do output operations.
28  *
29  * <itemizedlist>
30  *   <listitem><para>one sinkpad</para></listitem>
31  *   <listitem><para>handles state changes</para></listitem>
32  *   <listitem><para>pull/push mode</para></listitem>
33  *   <listitem><para>handles seeking/query</para></listitem>
34  *   <listitem><para>handles preroll</para></listitem>
35  *   <listitem><para>EOS handling</para></listitem>
36  * </itemizedlist>
37  */
38
39 #ifdef HAVE_CONFIG_H
40 #  include "config.h"
41 #endif
42
43 #include "gstbasesink.h"
44 #include <gst/gstmarshal.h>
45
46 GST_DEBUG_CATEGORY_STATIC (gst_base_sink_debug);
47 #define GST_CAT_DEFAULT gst_base_sink_debug
48
49 /* BaseSink signals and properties */
50 enum
51 {
52   /* FILL ME */
53   SIGNAL_HANDOFF,
54   LAST_SIGNAL
55 };
56
57 /* FIXME, need to figure out a better way to handle the pull mode */
58 #define DEFAULT_SIZE 1024
59 #define DEFAULT_HAS_LOOP FALSE
60 #define DEFAULT_HAS_CHAIN TRUE
61
62 enum
63 {
64   PROP_0,
65   PROP_HAS_LOOP,
66   PROP_HAS_CHAIN,
67   PROP_PREROLL_QUEUE_LEN
68 };
69
70 static GstElementClass *parent_class = NULL;
71
72 static void gst_base_sink_base_init (gpointer g_class);
73 static void gst_base_sink_class_init (GstBaseSinkClass * klass);
74 static void gst_base_sink_init (GstBaseSink * trans, gpointer g_class);
75 static void gst_base_sink_finalize (GObject * object);
76
77 GType
78 gst_base_sink_get_type (void)
79 {
80   static GType base_sink_type = 0;
81
82   if (!base_sink_type) {
83     static const GTypeInfo base_sink_info = {
84       sizeof (GstBaseSinkClass),
85       (GBaseInitFunc) gst_base_sink_base_init,
86       NULL,
87       (GClassInitFunc) gst_base_sink_class_init,
88       NULL,
89       NULL,
90       sizeof (GstBaseSink),
91       0,
92       (GInstanceInitFunc) gst_base_sink_init,
93     };
94
95     base_sink_type = g_type_register_static (GST_TYPE_ELEMENT,
96         "GstBaseSink", &base_sink_info, G_TYPE_FLAG_ABSTRACT);
97   }
98   return base_sink_type;
99 }
100
101 static void gst_base_sink_set_clock (GstElement * element, GstClock * clock);
102
103 static void gst_base_sink_set_property (GObject * object, guint prop_id,
104     const GValue * value, GParamSpec * pspec);
105 static void gst_base_sink_get_property (GObject * object, guint prop_id,
106     GValue * value, GParamSpec * pspec);
107
108 static GstCaps *gst_base_sink_get_caps (GstBaseSink * sink);
109 static gboolean gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps);
110 static GstFlowReturn gst_base_sink_buffer_alloc (GstBaseSink * sink,
111     guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
112 static void gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
113     GstClockTime * start, GstClockTime * end);
114
115 static GstElementStateReturn gst_base_sink_change_state (GstElement * element);
116
117 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
118 static void gst_base_sink_loop (GstPad * pad);
119 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
120 static gboolean gst_base_sink_activate_push (GstPad * pad, gboolean active);
121 static gboolean gst_base_sink_activate_pull (GstPad * pad, gboolean active);
122 static gboolean gst_base_sink_event (GstPad * pad, GstEvent * event);
123 static inline GstFlowReturn gst_base_sink_handle_buffer (GstBaseSink * basesink,
124     GstBuffer * buf);
125 static inline gboolean gst_base_sink_handle_event (GstBaseSink * basesink,
126     GstEvent * event);
127
128 static void
129 gst_base_sink_base_init (gpointer g_class)
130 {
131   GST_DEBUG_CATEGORY_INIT (gst_base_sink_debug, "basesink", 0,
132       "basesink element");
133 }
134
135 static void
136 gst_base_sink_class_init (GstBaseSinkClass * klass)
137 {
138   GObjectClass *gobject_class;
139   GstElementClass *gstelement_class;
140
141   gobject_class = (GObjectClass *) klass;
142   gstelement_class = (GstElementClass *) klass;
143
144   parent_class = g_type_class_ref (GST_TYPE_ELEMENT);
145
146   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_sink_finalize);
147   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_base_sink_set_property);
148   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_base_sink_get_property);
149
150   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_LOOP,
151       g_param_spec_boolean ("has-loop", "has-loop",
152           "Enable loop-based operation", DEFAULT_HAS_LOOP,
153           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
154   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_CHAIN,
155       g_param_spec_boolean ("has-chain", "has-chain",
156           "Enable chain-based operation", DEFAULT_HAS_CHAIN,
157           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
158   /* FIXME, this next value should be configured using an event from the
159    * upstream element */
160   g_object_class_install_property (G_OBJECT_CLASS (klass),
161       PROP_PREROLL_QUEUE_LEN,
162       g_param_spec_uint ("preroll-queue-len", "preroll-queue-len",
163           "Number of buffers to queue during preroll", 0, G_MAXUINT, 0,
164           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
165
166   gstelement_class->set_clock = GST_DEBUG_FUNCPTR (gst_base_sink_set_clock);
167   gstelement_class->change_state =
168       GST_DEBUG_FUNCPTR (gst_base_sink_change_state);
169
170   klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_sink_get_caps);
171   klass->set_caps = GST_DEBUG_FUNCPTR (gst_base_sink_set_caps);
172   klass->buffer_alloc = GST_DEBUG_FUNCPTR (gst_base_sink_buffer_alloc);
173   klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
174 }
175
176 static GstCaps *
177 gst_base_sink_pad_getcaps (GstPad * pad)
178 {
179   GstBaseSinkClass *bclass;
180   GstBaseSink *bsink;
181   GstCaps *caps = NULL;
182
183   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
184   bclass = GST_BASE_SINK_GET_CLASS (bsink);
185   if (bclass->get_caps)
186     caps = bclass->get_caps (bsink);
187
188   if (caps == NULL) {
189     GstPadTemplate *pad_template;
190
191     pad_template =
192         gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "sink");
193     if (pad_template != NULL) {
194       caps = gst_caps_ref (gst_pad_template_get_caps (pad_template));
195     }
196   }
197   gst_object_unref (bsink);
198
199   return caps;
200 }
201
202 static gboolean
203 gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps)
204 {
205   GstBaseSinkClass *bclass;
206   GstBaseSink *bsink;
207   gboolean res = FALSE;
208
209   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
210   bclass = GST_BASE_SINK_GET_CLASS (bsink);
211
212   if (bclass->set_caps)
213     res = bclass->set_caps (bsink, caps);
214
215   gst_object_unref (bsink);
216
217   return res;
218 }
219
220 static GstFlowReturn
221 gst_base_sink_pad_buffer_alloc (GstPad * pad, guint64 offset, guint size,
222     GstCaps * caps, GstBuffer ** buf)
223 {
224   GstBaseSinkClass *bclass;
225   GstBaseSink *bsink;
226   GstFlowReturn result = GST_FLOW_OK;
227
228   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
229   bclass = GST_BASE_SINK_GET_CLASS (bsink);
230
231   if (bclass->buffer_alloc)
232     result = bclass->buffer_alloc (bsink, offset, size, caps, buf);
233   else
234     *buf = NULL;
235
236   gst_object_unref (bsink);
237
238   return result;
239 }
240
241 static void
242 gst_base_sink_init (GstBaseSink * basesink, gpointer g_class)
243 {
244   GstPadTemplate *pad_template;
245
246   pad_template =
247       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "sink");
248   g_return_if_fail (pad_template != NULL);
249
250   basesink->sinkpad = gst_pad_new_from_template (pad_template, "sink");
251
252   gst_pad_set_getcaps_function (basesink->sinkpad,
253       GST_DEBUG_FUNCPTR (gst_base_sink_pad_getcaps));
254   gst_pad_set_setcaps_function (basesink->sinkpad,
255       GST_DEBUG_FUNCPTR (gst_base_sink_pad_setcaps));
256   gst_pad_set_bufferalloc_function (basesink->sinkpad,
257       GST_DEBUG_FUNCPTR (gst_base_sink_pad_buffer_alloc));
258   gst_element_add_pad (GST_ELEMENT (basesink), basesink->sinkpad);
259
260   basesink->pad_mode = GST_ACTIVATE_NONE;
261   GST_PAD_TASK (basesink->sinkpad) = NULL;
262   basesink->preroll_queue = g_queue_new ();
263
264   GST_FLAG_SET (basesink, GST_ELEMENT_IS_SINK);
265 }
266
267 static void
268 gst_base_sink_finalize (GObject * object)
269 {
270   GstBaseSink *basesink;
271
272   basesink = GST_BASE_SINK (object);
273
274   g_queue_free (basesink->preroll_queue);
275
276   G_OBJECT_CLASS (parent_class)->finalize (object);
277 }
278
279 static void
280 gst_base_sink_set_pad_functions (GstBaseSink * this, GstPad * pad)
281 {
282   gst_pad_set_activatepush_function (pad,
283       GST_DEBUG_FUNCPTR (gst_base_sink_activate_push));
284   gst_pad_set_activatepull_function (pad,
285       GST_DEBUG_FUNCPTR (gst_base_sink_activate_pull));
286   gst_pad_set_event_function (pad, GST_DEBUG_FUNCPTR (gst_base_sink_event));
287
288   if (this->has_chain)
289     gst_pad_set_chain_function (pad, GST_DEBUG_FUNCPTR (gst_base_sink_chain));
290   else
291     gst_pad_set_chain_function (pad, NULL);
292 }
293
294 static void
295 gst_base_sink_set_all_pad_functions (GstBaseSink * this)
296 {
297   GList *l;
298
299   for (l = GST_ELEMENT_PADS (this); l; l = l->next)
300     gst_base_sink_set_pad_functions (this, (GstPad *) l->data);
301 }
302
303 static void
304 gst_base_sink_set_clock (GstElement * element, GstClock * clock)
305 {
306   GstBaseSink *sink;
307
308   sink = GST_BASE_SINK (element);
309
310   sink->clock = clock;
311 }
312
313 static void
314 gst_base_sink_set_property (GObject * object, guint prop_id,
315     const GValue * value, GParamSpec * pspec)
316 {
317   GstBaseSink *sink;
318
319   sink = GST_BASE_SINK (object);
320
321   switch (prop_id) {
322     case PROP_HAS_LOOP:
323       GST_LOCK (sink);
324       sink->has_loop = g_value_get_boolean (value);
325       gst_base_sink_set_all_pad_functions (sink);
326       GST_UNLOCK (sink);
327       break;
328     case PROP_HAS_CHAIN:
329       GST_LOCK (sink);
330       sink->has_chain = g_value_get_boolean (value);
331       gst_base_sink_set_all_pad_functions (sink);
332       GST_UNLOCK (sink);
333       break;
334     case PROP_PREROLL_QUEUE_LEN:
335       /* preroll lock necessary to serialize with finish_preroll */
336       GST_PREROLL_LOCK (sink->sinkpad);
337       sink->preroll_queue_max_len = g_value_get_uint (value);
338       GST_PREROLL_UNLOCK (sink->sinkpad);
339       break;
340     default:
341       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
342       break;
343   }
344 }
345
346 static void
347 gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
348     GParamSpec * pspec)
349 {
350   GstBaseSink *sink;
351
352   sink = GST_BASE_SINK (object);
353
354   GST_LOCK (sink);
355   switch (prop_id) {
356     case PROP_HAS_LOOP:
357       g_value_set_boolean (value, sink->has_loop);
358       break;
359     case PROP_HAS_CHAIN:
360       g_value_set_boolean (value, sink->has_chain);
361       break;
362     case PROP_PREROLL_QUEUE_LEN:
363       g_value_set_uint (value, sink->preroll_queue_max_len);
364       break;
365     default:
366       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
367       break;
368   }
369   GST_UNLOCK (sink);
370 }
371
372 static GstCaps *
373 gst_base_sink_get_caps (GstBaseSink * sink)
374 {
375   return NULL;
376 }
377
378 static gboolean
379 gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
380 {
381   return TRUE;
382 }
383
384 static GstFlowReturn
385 gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
386     GstCaps * caps, GstBuffer ** buf)
387 {
388   *buf = NULL;
389   return GST_FLOW_OK;
390 }
391
392 /* with PREROLL_LOCK */
393 static GstFlowReturn
394 gst_base_sink_preroll_queue_empty (GstBaseSink * basesink, GstPad * pad)
395 {
396   GstMiniObject *obj;
397   GQueue *q = basesink->preroll_queue;
398   GstFlowReturn ret;
399
400   ret = GST_FLOW_OK;
401
402   if (q) {
403     GST_DEBUG ("emptying queue");
404     while ((obj = g_queue_pop_head (q))) {
405       gboolean is_buffer;
406
407       is_buffer = GST_IS_BUFFER (obj);
408       if (is_buffer) {
409         basesink->preroll_queued--;
410         basesink->buffers_queued--;
411       } else {
412         switch (GST_EVENT_TYPE (obj)) {
413           case GST_EVENT_EOS:
414             basesink->preroll_queued--;
415             break;
416           default:
417             break;
418         }
419         basesink->events_queued--;
420       }
421       /* we release the preroll lock while pushing so that we
422        * can still flush it while blocking on the clock or
423        * inside the element. */
424       GST_PREROLL_UNLOCK (pad);
425
426       if (is_buffer) {
427         GST_DEBUG ("popped buffer %p", obj);
428         ret = gst_base_sink_handle_buffer (basesink, GST_BUFFER (obj));
429       } else {
430         GST_DEBUG ("popped event %p", obj);
431         gst_base_sink_handle_event (basesink, GST_EVENT (obj));
432         ret = GST_FLOW_OK;
433       }
434
435       GST_PREROLL_LOCK (pad);
436     }
437     GST_DEBUG ("queue empty");
438   }
439   return ret;
440 }
441
442 /* with PREROLL_LOCK */
443 static void
444 gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
445 {
446   GstMiniObject *obj;
447   GQueue *q = basesink->preroll_queue;
448
449   GST_DEBUG ("flushing queue %p", basesink);
450   if (q) {
451     while ((obj = g_queue_pop_head (q))) {
452       GST_DEBUG ("popped %p", obj);
453       gst_mini_object_unref (obj);
454     }
455   }
456   /* we can't have EOS anymore now */
457   basesink->eos = FALSE;
458   basesink->preroll_queued = 0;
459   basesink->buffers_queued = 0;
460   basesink->events_queued = 0;
461   basesink->have_preroll = FALSE;
462   /* and signal any waiters now */
463   GST_PREROLL_SIGNAL (pad);
464 }
465
466 /* with STREAM_LOCK */
467 static GstFlowReturn
468 gst_base_sink_handle_object (GstBaseSink * basesink, GstPad * pad,
469     GstMiniObject * obj)
470 {
471   gint length;
472   gboolean have_event;
473
474   GST_PREROLL_LOCK (pad);
475   /* push object on the queue */
476   GST_DEBUG ("push on queue %p", basesink, obj);
477   g_queue_push_tail (basesink->preroll_queue, obj);
478
479   have_event = GST_IS_EVENT (obj);
480   if (have_event) {
481     GstEvent *event = GST_EVENT (obj);
482
483     switch (GST_EVENT_TYPE (obj)) {
484       case GST_EVENT_EOS:
485         basesink->preroll_queued++;
486         basesink->eos = TRUE;
487         break;
488       case GST_EVENT_NEWSEGMENT:
489       {
490         GstFormat format;
491         gdouble rate;
492
493         /* the newsegment event is needed to bring the buffer timestamps to the
494          * stream time */
495         gst_event_parse_newsegment (event, &rate, &format,
496             &basesink->discont_start, &basesink->discont_stop, NULL);
497
498         if (format != GST_FORMAT_TIME) {
499           /* this means this sink will not be able to sync to the clock */
500           basesink->discont_start = 0;
501           basesink->discont_stop = 0;
502         }
503         basesink->have_discont = TRUE;
504
505         GST_DEBUG ("received DISCONT %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
506             GST_TIME_ARGS (basesink->discont_start),
507             GST_TIME_ARGS (basesink->discont_stop));
508         break;
509       }
510       default:
511         break;
512     }
513     basesink->events_queued++;
514   } else {
515     if (!basesink->have_discont) {
516       GST_ELEMENT_ERROR (basesink, STREAM, STOPPED,
517           ("received buffer without a discont"),
518           ("received buffer without a discont"));
519     }
520     basesink->preroll_queued++;
521     basesink->buffers_queued++;
522   }
523   GST_DEBUG ("now %d preroll, %d buffers, %d events on queue",
524       basesink->preroll_queued,
525       basesink->buffers_queued, basesink->events_queued);
526
527   /* check if we are prerolling */
528   if (!basesink->need_preroll)
529     goto no_preroll;
530
531   /* there is a buffer queued */
532   if (basesink->buffers_queued == 1) {
533     GST_DEBUG ("do preroll %p", obj);
534
535     /* if it's a buffer, we need to call the preroll method */
536     if (GST_IS_BUFFER (obj)) {
537       GstBaseSinkClass *bclass;
538       GstFlowReturn pres;
539
540       bclass = GST_BASE_SINK_GET_CLASS (basesink);
541       if (bclass->preroll)
542         if ((pres =
543                 bclass->preroll (basesink, GST_BUFFER (obj))) != GST_FLOW_OK)
544           goto preroll_failed;
545     }
546   }
547   length = basesink->preroll_queued;
548   GST_DEBUG ("prerolled length %d", length);
549
550   if (length == 1) {
551     gint t;
552
553     basesink->have_preroll = TRUE;
554     /* we are prerolling */
555     GST_PREROLL_UNLOCK (pad);
556
557     /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
558      * inside the STREAM_LOCK */
559     t = GST_STREAM_UNLOCK_FULL (pad);
560     GST_DEBUG ("released stream lock %d times", t);
561     if (t <= 0) {
562       GST_WARNING ("STREAM_LOCK should have been locked !!");
563       g_warning ("STREAM_LOCK should have been locked !!");
564     }
565
566     /* now we commit our state */
567     GST_STATE_LOCK (basesink);
568     GST_DEBUG ("commit state %p >", basesink);
569     gst_element_commit_state (GST_ELEMENT (basesink));
570     GST_STATE_UNLOCK (basesink);
571
572     /* reacquire stream lock, pad could be flushing now */
573     /* FIXME in glib, if t==0, the lock is still taken... hmmm */
574     if (t > 0)
575       GST_STREAM_LOCK_FULL (pad, t);
576
577     /* and wait if needed */
578     GST_PREROLL_LOCK (pad);
579
580     GST_LOCK (pad);
581     if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
582       goto flushing;
583     GST_UNLOCK (pad);
584
585     /* it is possible that the application set the state to PLAYING
586      * now in which case we don't need to block anymore. */
587     if (!basesink->need_preroll)
588       goto no_preroll;
589
590     length = basesink->preroll_queued;
591
592     g_assert (length == 1);
593   }
594
595   /* see if we need to block now. We cannot block on events, only
596    * on buffers, the reason is that events can be sent from the
597    * application thread and we don't want to block there. */
598   if (length > basesink->preroll_queue_max_len && !have_event) {
599     /* block until the state changes, or we get a flush, or something */
600     GST_DEBUG ("element %s waiting to finish preroll",
601         GST_ELEMENT_NAME (basesink));
602     GST_PREROLL_WAIT (pad);
603     GST_DEBUG ("done preroll");
604   }
605   GST_LOCK (pad);
606   if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
607     goto flushing;
608   GST_UNLOCK (pad);
609
610   GST_PREROLL_UNLOCK (pad);
611
612   return GST_FLOW_OK;
613
614 no_preroll:
615   {
616     GstFlowReturn ret;
617
618     GST_DEBUG ("no preroll needed");
619     /* maybe it was another sink that blocked in preroll, need to check for
620        buffers to drain */
621     basesink->have_preroll = FALSE;
622     ret = gst_base_sink_preroll_queue_empty (basesink, pad);
623     GST_PREROLL_UNLOCK (pad);
624
625     return ret;
626   }
627 flushing:
628   {
629     GST_UNLOCK (pad);
630     gst_base_sink_preroll_queue_flush (basesink, pad);
631     GST_PREROLL_UNLOCK (pad);
632     GST_DEBUG ("pad is flushing");
633     return GST_FLOW_WRONG_STATE;
634   }
635 preroll_failed:
636   {
637     gint t;
638
639     GST_DEBUG ("preroll failed");
640     gst_base_sink_preroll_queue_flush (basesink, pad);
641     GST_PREROLL_UNLOCK (pad);
642
643     /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
644      * inside the STREAM_LOCK */
645     t = GST_STREAM_UNLOCK_FULL (pad);
646     GST_DEBUG ("released stream lock %d times", t);
647     if (t <= 0) {
648       GST_WARNING ("STREAM_LOCK should have been locked !!");
649       g_warning ("STREAM_LOCK should have been locked !!");
650     }
651
652     /* now we abort our state */
653     GST_STATE_LOCK (basesink);
654     GST_DEBUG ("abort state %p >", basesink);
655     gst_element_abort_state (GST_ELEMENT (basesink));
656     GST_STATE_UNLOCK (basesink);
657
658     /* reacquire stream lock, pad could be flushing now */
659     if (t > 0)
660       GST_STREAM_LOCK_FULL (pad, t);
661
662     return GST_FLOW_ERROR;
663   }
664 }
665
666 static gboolean
667 gst_base_sink_event (GstPad * pad, GstEvent * event)
668 {
669   GstBaseSink *basesink;
670   gboolean result = TRUE;
671   GstBaseSinkClass *bclass;
672
673   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
674
675   bclass = GST_BASE_SINK_GET_CLASS (basesink);
676
677   GST_DEBUG ("event %p", event);
678
679   switch (GST_EVENT_TYPE (event)) {
680     case GST_EVENT_EOS:
681     {
682       GstFlowReturn ret;
683
684       GST_STREAM_LOCK (pad);
685       /* EOS also finishes the preroll */
686       ret =
687           gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
688       GST_STREAM_UNLOCK (pad);
689       break;
690     }
691     case GST_EVENT_NEWSEGMENT:
692     {
693       GstFlowReturn ret;
694
695       GST_STREAM_LOCK (pad);
696       ret =
697           gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
698       GST_STREAM_UNLOCK (pad);
699       break;
700     }
701     case GST_EVENT_FLUSH_START:
702       /* make sure we are not blocked on the clock also clear any pending
703        * eos state. */
704       if (bclass->event)
705         bclass->event (basesink, event);
706
707       GST_PREROLL_LOCK (pad);
708       /* we need preroll after the flush */
709       basesink->need_preroll = TRUE;
710       /* unlock from a possible state change/preroll */
711       gst_base_sink_preroll_queue_flush (basesink, pad);
712
713       GST_LOCK (basesink);
714       if (basesink->clock_id) {
715         gst_clock_id_unschedule (basesink->clock_id);
716       }
717       GST_UNLOCK (basesink);
718       GST_PREROLL_UNLOCK (pad);
719
720       /* and we need to commit our state again on the next
721        * prerolled buffer */
722       GST_STATE_LOCK (basesink);
723       GST_STREAM_LOCK (pad);
724       gst_element_lost_state (GST_ELEMENT (basesink));
725       GST_STREAM_UNLOCK (pad);
726       GST_STATE_UNLOCK (basesink);
727       GST_DEBUG ("event unref %p %p", basesink, event);
728       gst_event_unref (event);
729       break;
730     case GST_EVENT_FLUSH_STOP:
731       if (bclass->event)
732         bclass->event (basesink, event);
733
734       /* now we are completely unblocked and the _chain method
735        * will return */
736       GST_STREAM_LOCK (pad);
737       GST_STREAM_UNLOCK (pad);
738
739       GST_DEBUG ("event unref %p %p", basesink, event);
740       gst_event_unref (event);
741       break;
742     default:
743       gst_event_unref (event);
744       break;
745   }
746   gst_object_unref (basesink);
747
748   return result;
749 }
750
751 /* default implementation to calculate the start and end
752  * timestamps on a buffer, subclasses can override
753  */
754 static void
755 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
756     GstClockTime * start, GstClockTime * end)
757 {
758   GstClockTime timestamp, duration;
759
760   timestamp = GST_BUFFER_TIMESTAMP (buffer);
761   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
762     GstClockTimeDiff diff;
763
764     /* bring timestamp to stream time using last
765      * discont offset. */
766     if ((diff = timestamp - basesink->discont_start) < 0)
767       goto too_late;
768
769     /* get duration to calculate end time */
770     duration = GST_BUFFER_DURATION (buffer);
771     if (GST_CLOCK_TIME_IS_VALID (duration)) {
772       *end = diff + duration;
773     }
774     *start = diff;
775   }
776   return;
777
778 too_late:
779   {
780     *start = GST_CLOCK_TIME_NONE;
781     *end = GST_CLOCK_TIME_NONE;
782   }
783 }
784
785 /* perform synchronisation on a buffer
786  * 
787  * 1) check if we have a clock, if not, do nothing
788  * 2) calculate the start and end time of the buffer
789  * 3) create a single shot notification to wait on
790  *    the clock, save the entry so we can unlock it
791  * 4) wait on the clock, this blocks
792  * 5) unref the clockid again
793  */
794 static gboolean
795 gst_base_sink_do_sync (GstBaseSink * basesink, GstBuffer * buffer)
796 {
797   gboolean result = TRUE;
798
799   if (basesink->clock) {
800     GstClockTime start, end;
801     GstBaseSinkClass *bclass;
802
803     bclass = GST_BASE_SINK_GET_CLASS (basesink);
804     start = end = -1;
805     if (bclass->get_times)
806       bclass->get_times (basesink, buffer, &start, &end);
807
808     GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
809         ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
810
811     if (GST_CLOCK_TIME_IS_VALID (start)) {
812       GstClockReturn ret;
813       GstClockTime base_time;
814
815       GST_LOCK (basesink);
816       base_time = GST_ELEMENT (basesink)->base_time;
817
818       GST_LOG_OBJECT (basesink,
819           "waiting for clock, base time %" GST_TIME_FORMAT,
820           GST_TIME_ARGS (base_time));
821       /* save clock id so that we can unlock it if needed */
822       basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
823           start + base_time);
824       basesink->end_time = end;
825       GST_UNLOCK (basesink);
826
827       ret = gst_clock_id_wait (basesink->clock_id, NULL);
828
829       GST_LOCK (basesink);
830       if (basesink->clock_id) {
831         gst_clock_id_unref (basesink->clock_id);
832         basesink->clock_id = NULL;
833       }
834       GST_UNLOCK (basesink);
835
836       GST_LOG_OBJECT (basesink, "clock entry done: %d", ret);
837       if (ret == GST_CLOCK_UNSCHEDULED)
838         result = FALSE;
839     }
840   }
841   return result;
842 }
843
844
845 /* handle an event
846  *
847  * 2) render the event
848  * 3) unref the event
849  */
850 static inline gboolean
851 gst_base_sink_handle_event (GstBaseSink * basesink, GstEvent * event)
852 {
853   GstBaseSinkClass *bclass;
854   gboolean ret;
855
856   switch (GST_EVENT_TYPE (event)) {
857     case GST_EVENT_EOS:
858       GST_LOCK (basesink);
859       if (basesink->clock) {
860         /* wait for last buffer to finish if we have a valid end time */
861         if (GST_CLOCK_TIME_IS_VALID (basesink->end_time)) {
862           basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
863               basesink->end_time + GST_ELEMENT (basesink)->base_time);
864           GST_UNLOCK (basesink);
865
866           gst_clock_id_wait (basesink->clock_id, NULL);
867
868           GST_LOCK (basesink);
869           if (basesink->clock_id) {
870             gst_clock_id_unref (basesink->clock_id);
871             basesink->clock_id = NULL;
872           }
873           basesink->end_time = GST_CLOCK_TIME_NONE;
874         }
875       }
876       GST_UNLOCK (basesink);
877       break;
878     default:
879       break;
880   }
881
882   bclass = GST_BASE_SINK_GET_CLASS (basesink);
883   if (bclass->event)
884     ret = bclass->event (basesink, event);
885   else
886     ret = TRUE;
887
888   switch (GST_EVENT_TYPE (event)) {
889     case GST_EVENT_EOS:
890       GST_PREROLL_LOCK (basesink->sinkpad);
891       /* if we are still EOS, we can post the EOS message */
892       if (basesink->eos) {
893         /* ok, now we can post the message */
894         GST_DEBUG_OBJECT (basesink, "Now posting EOS");
895         gst_element_post_message (GST_ELEMENT (basesink),
896             gst_message_new_eos (GST_OBJECT (basesink)));
897       }
898       GST_PREROLL_UNLOCK (basesink->sinkpad);
899       break;
900     default:
901       break;
902   }
903
904   GST_DEBUG ("event unref %p %p", basesink, event);
905   gst_event_unref (event);
906
907   return ret;
908 }
909
910 /* handle a buffer
911  *
912  * 1) first sync on the buffer
913  * 2) render the buffer
914  * 3) unref the buffer
915  */
916 static inline GstFlowReturn
917 gst_base_sink_handle_buffer (GstBaseSink * basesink, GstBuffer * buf)
918 {
919   GstBaseSinkClass *bclass;
920   GstFlowReturn ret;
921
922   gst_base_sink_do_sync (basesink, buf);
923
924   bclass = GST_BASE_SINK_GET_CLASS (basesink);
925   if (bclass->render)
926     ret = bclass->render (basesink, buf);
927   else
928     ret = GST_FLOW_OK;
929
930   GST_DEBUG ("buffer unref after render %p", basesink, buf);
931   gst_buffer_unref (buf);
932
933   return ret;
934 }
935
936 static GstFlowReturn
937 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
938 {
939   GstBaseSink *basesink;
940   GstFlowReturn result;
941
942   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
943
944   result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
945
946   gst_object_unref (basesink);
947
948   return result;
949 }
950
951 /* FIXME, not all sinks can operate in pull mode 
952  */
953 static void
954 gst_base_sink_loop (GstPad * pad)
955 {
956   GstBaseSink *basesink;
957   GstBuffer *buf = NULL;
958   GstFlowReturn result;
959
960   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
961
962   g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
963
964   result = gst_pad_pull_range (pad, basesink->offset, DEFAULT_SIZE, &buf);
965   if (result != GST_FLOW_OK)
966     goto paused;
967
968   result = gst_base_sink_chain (pad, buf);
969   if (result != GST_FLOW_OK)
970     goto paused;
971
972   gst_object_unref (basesink);
973
974   /* default */
975   return;
976
977 paused:
978   {
979     gst_object_unref (basesink);
980     gst_pad_pause_task (pad);
981     return;
982   }
983 }
984
985 static gboolean
986 gst_base_sink_deactivate (GstBaseSink * basesink, GstPad * pad)
987 {
988   gboolean result = FALSE;
989   GstBaseSinkClass *bclass;
990
991   bclass = GST_BASE_SINK_GET_CLASS (basesink);
992
993   /* step 1, unblock clock sync (if any) or any other blocking thing */
994   GST_PREROLL_LOCK (pad);
995   GST_LOCK (basesink);
996   if (basesink->clock_id) {
997     gst_clock_id_unschedule (basesink->clock_id);
998   }
999   GST_UNLOCK (basesink);
1000
1001   /* unlock any subclasses */
1002   if (bclass->unlock)
1003     bclass->unlock (basesink);
1004
1005   /* flush out the data thread if it's locked in finish_preroll */
1006   basesink->need_preroll = FALSE;
1007   gst_base_sink_preroll_queue_flush (basesink, pad);
1008   GST_PREROLL_SIGNAL (pad);
1009   GST_PREROLL_UNLOCK (pad);
1010
1011   /* step 2, make sure streaming finishes */
1012   result = gst_pad_stop_task (pad);
1013
1014   return result;
1015 }
1016
1017 static gboolean
1018 gst_base_sink_activate_push (GstPad * pad, gboolean active)
1019 {
1020   gboolean result = FALSE;
1021   GstBaseSink *basesink;
1022
1023   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1024
1025   if (active) {
1026     if (!basesink->has_chain)
1027       goto done;
1028     result = TRUE;
1029   } else {
1030     result = gst_base_sink_deactivate (basesink, pad);
1031   }
1032   basesink->pad_mode = GST_ACTIVATE_PUSH;
1033
1034 done:
1035   gst_object_unref (basesink);
1036
1037   return result;
1038 }
1039
1040 /* this won't get called until we implement an activate function */
1041 static gboolean
1042 gst_base_sink_activate_pull (GstPad * pad, gboolean active)
1043 {
1044   gboolean result = FALSE;
1045   GstBaseSink *basesink;
1046
1047   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1048
1049   if (active) {
1050     /* if we have a scheduler we can start the task */
1051     if (!basesink->has_loop)
1052       goto done;
1053     result =
1054         gst_pad_start_task (pad, (GstTaskFunction) gst_base_sink_loop, pad);
1055   } else {
1056     result = gst_base_sink_deactivate (basesink, pad);
1057   }
1058 done:
1059   gst_object_unref (basesink);
1060
1061   return result;
1062 }
1063
1064 static GstElementStateReturn
1065 gst_base_sink_change_state (GstElement * element)
1066 {
1067   GstElementStateReturn ret = GST_STATE_SUCCESS;
1068   GstBaseSink *basesink = GST_BASE_SINK (element);
1069   GstElementState transition = GST_STATE_TRANSITION (element);
1070   GstBaseSinkClass *bclass;
1071
1072   bclass = GST_BASE_SINK_GET_CLASS (basesink);
1073
1074   switch (transition) {
1075     case GST_STATE_NULL_TO_READY:
1076       if (bclass->start)
1077         if (!bclass->start (basesink))
1078           goto start_failed;
1079       break;
1080     case GST_STATE_READY_TO_PAUSED:
1081       /* need to complete preroll before this state change completes, there
1082        * is no data flow in READY so we can safely assume we need to preroll. */
1083       basesink->offset = 0;
1084       GST_PREROLL_LOCK (basesink->sinkpad);
1085       basesink->have_preroll = FALSE;
1086       basesink->need_preroll = TRUE;
1087       GST_PREROLL_UNLOCK (basesink->sinkpad);
1088       basesink->have_discont = FALSE;
1089       basesink->discont_start = 0;
1090       basesink->discont_stop = 0;
1091       ret = GST_STATE_ASYNC;
1092       break;
1093     case GST_STATE_PAUSED_TO_PLAYING:
1094     {
1095       GST_PREROLL_LOCK (basesink->sinkpad);
1096       /* if we have EOS, we should empty the queue now as there will
1097        * be no more data received in the chain function.
1098        * FIXME, this could block the state change function too long when
1099        * we are pushing and syncing the buffers, better start a new
1100        * thread to do this. */
1101       if (basesink->eos) {
1102         gst_base_sink_preroll_queue_empty (basesink, basesink->sinkpad);
1103       }
1104       /* don't need the preroll anymore */
1105       basesink->need_preroll = FALSE;
1106       /* now let it play */
1107       GST_PREROLL_SIGNAL (basesink->sinkpad);
1108       GST_PREROLL_UNLOCK (basesink->sinkpad);
1109       break;
1110     }
1111     default:
1112       break;
1113   }
1114
1115   GST_ELEMENT_CLASS (parent_class)->change_state (element);
1116
1117   switch (transition) {
1118     case GST_STATE_PLAYING_TO_PAUSED:
1119     {
1120       GstBaseSinkClass *bclass;
1121
1122       bclass = GST_BASE_SINK_GET_CLASS (basesink);
1123
1124       GST_PREROLL_LOCK (basesink->sinkpad);
1125       GST_LOCK (basesink);
1126       /* unlock clock wait if any */
1127       if (basesink->clock_id) {
1128         gst_clock_id_unschedule (basesink->clock_id);
1129       }
1130       GST_UNLOCK (basesink);
1131
1132       /* unlock any subclasses */
1133       if (bclass->unlock)
1134         bclass->unlock (basesink);
1135
1136       /* if we don't have a preroll buffer and we have not received EOS,
1137        * we need to wait for a preroll */
1138       GST_DEBUG ("have_preroll: %d, EOS: %d", basesink->have_preroll,
1139           basesink->eos);
1140       if (!basesink->have_preroll && !basesink->eos) {
1141         basesink->need_preroll = TRUE;
1142         ret = GST_STATE_ASYNC;
1143       }
1144       GST_PREROLL_UNLOCK (basesink->sinkpad);
1145       break;
1146     }
1147     case GST_STATE_PAUSED_TO_READY:
1148       break;
1149     case GST_STATE_READY_TO_NULL:
1150       if (bclass->stop)
1151         if (!bclass->stop (basesink)) {
1152           GST_WARNING ("failed to stop");
1153         }
1154       break;
1155     default:
1156       break;
1157   }
1158
1159   return ret;
1160
1161   /* ERRORS */
1162 start_failed:
1163   {
1164     GST_DEBUG ("failed to start");
1165     return GST_STATE_FAILURE;
1166   }
1167 }