2 * Copyright (C) 2005 Wim Taymans <wim@fluendo.com>
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.
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.
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.
24 * @short_description: Base class for sink elements
25 * @see_also: #GstBaseTransformc, #GstBaseSource
27 * This class is for elements that do output operations.
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>
43 #include "gstbasesink.h"
44 #include <gst/gstmarshal.h>
46 GST_DEBUG_CATEGORY_STATIC (gst_base_sink_debug);
47 #define GST_CAT_DEFAULT gst_base_sink_debug
49 /* BaseSink signals and properties */
57 #define DEFAULT_SIZE 1024
58 #define DEFAULT_CAN_ACTIVATE_PULL FALSE /* fixme: enable me */
59 #define DEFAULT_CAN_ACTIVATE_PUSH TRUE
64 PROP_PREROLL_QUEUE_LEN
67 static GstElementClass *parent_class = NULL;
69 static void gst_base_sink_base_init (gpointer g_class);
70 static void gst_base_sink_class_init (GstBaseSinkClass * klass);
71 static void gst_base_sink_init (GstBaseSink * trans, gpointer g_class);
72 static void gst_base_sink_finalize (GObject * object);
75 gst_base_sink_get_type (void)
77 static GType base_sink_type = 0;
79 if (!base_sink_type) {
80 static const GTypeInfo base_sink_info = {
81 sizeof (GstBaseSinkClass),
82 (GBaseInitFunc) gst_base_sink_base_init,
84 (GClassInitFunc) gst_base_sink_class_init,
89 (GInstanceInitFunc) gst_base_sink_init,
92 base_sink_type = g_type_register_static (GST_TYPE_ELEMENT,
93 "GstBaseSink", &base_sink_info, G_TYPE_FLAG_ABSTRACT);
95 return base_sink_type;
98 static void gst_base_sink_set_clock (GstElement * element, GstClock * clock);
100 static void gst_base_sink_set_property (GObject * object, guint prop_id,
101 const GValue * value, GParamSpec * pspec);
102 static void gst_base_sink_get_property (GObject * object, guint prop_id,
103 GValue * value, GParamSpec * pspec);
105 static GstCaps *gst_base_sink_get_caps (GstBaseSink * sink);
106 static gboolean gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps);
107 static GstFlowReturn gst_base_sink_buffer_alloc (GstBaseSink * sink,
108 guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
109 static void gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
110 GstClockTime * start, GstClockTime * end);
112 static GstStateChangeReturn gst_base_sink_change_state (GstElement * element,
113 GstStateChange transition);
115 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
116 static void gst_base_sink_loop (GstPad * pad);
117 static gboolean gst_base_sink_activate (GstPad * pad);
118 static gboolean gst_base_sink_activate_push (GstPad * pad, gboolean active);
119 static gboolean gst_base_sink_activate_pull (GstPad * pad, gboolean active);
120 static gboolean gst_base_sink_event (GstPad * pad, GstEvent * event);
121 static inline GstFlowReturn gst_base_sink_handle_buffer (GstBaseSink * basesink,
123 static inline gboolean gst_base_sink_handle_event (GstBaseSink * basesink,
127 gst_base_sink_base_init (gpointer g_class)
129 GST_DEBUG_CATEGORY_INIT (gst_base_sink_debug, "basesink", 0,
134 gst_base_sink_class_init (GstBaseSinkClass * klass)
136 GObjectClass *gobject_class;
137 GstElementClass *gstelement_class;
139 gobject_class = (GObjectClass *) klass;
140 gstelement_class = (GstElementClass *) klass;
142 parent_class = g_type_class_ref (GST_TYPE_ELEMENT);
144 gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_sink_finalize);
145 gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_base_sink_set_property);
146 gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_base_sink_get_property);
148 /* FIXME, this next value should be configured using an event from the
149 * upstream element */
150 g_object_class_install_property (G_OBJECT_CLASS (klass),
151 PROP_PREROLL_QUEUE_LEN,
152 g_param_spec_uint ("preroll-queue-len", "preroll-queue-len",
153 "Number of buffers to queue during preroll", 0, G_MAXUINT, 0,
154 G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
156 gstelement_class->set_clock = GST_DEBUG_FUNCPTR (gst_base_sink_set_clock);
157 gstelement_class->change_state =
158 GST_DEBUG_FUNCPTR (gst_base_sink_change_state);
160 klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_sink_get_caps);
161 klass->set_caps = GST_DEBUG_FUNCPTR (gst_base_sink_set_caps);
162 klass->buffer_alloc = GST_DEBUG_FUNCPTR (gst_base_sink_buffer_alloc);
163 klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
167 gst_base_sink_pad_getcaps (GstPad * pad)
169 GstBaseSinkClass *bclass;
171 GstCaps *caps = NULL;
173 bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
174 bclass = GST_BASE_SINK_GET_CLASS (bsink);
175 if (bclass->get_caps)
176 caps = bclass->get_caps (bsink);
179 GstPadTemplate *pad_template;
182 gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "sink");
183 if (pad_template != NULL) {
184 caps = gst_caps_ref (gst_pad_template_get_caps (pad_template));
187 gst_object_unref (bsink);
193 gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps)
195 GstBaseSinkClass *bclass;
197 gboolean res = FALSE;
199 bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
200 bclass = GST_BASE_SINK_GET_CLASS (bsink);
202 if (bclass->set_caps)
203 res = bclass->set_caps (bsink, caps);
205 gst_object_unref (bsink);
211 gst_base_sink_pad_buffer_alloc (GstPad * pad, guint64 offset, guint size,
212 GstCaps * caps, GstBuffer ** buf)
214 GstBaseSinkClass *bclass;
216 GstFlowReturn result = GST_FLOW_OK;
218 bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
219 bclass = GST_BASE_SINK_GET_CLASS (bsink);
221 if (bclass->buffer_alloc)
222 result = bclass->buffer_alloc (bsink, offset, size, caps, buf);
224 *buf = NULL; /* fallback in gstpad.c will allocate generic buffer */
226 gst_object_unref (bsink);
232 gst_base_sink_init (GstBaseSink * basesink, gpointer g_class)
234 GstPadTemplate *pad_template;
237 gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "sink");
238 g_return_if_fail (pad_template != NULL);
240 basesink->sinkpad = gst_pad_new_from_template (pad_template, "sink");
242 gst_pad_set_getcaps_function (basesink->sinkpad,
243 GST_DEBUG_FUNCPTR (gst_base_sink_pad_getcaps));
244 gst_pad_set_setcaps_function (basesink->sinkpad,
245 GST_DEBUG_FUNCPTR (gst_base_sink_pad_setcaps));
246 gst_pad_set_bufferalloc_function (basesink->sinkpad,
247 GST_DEBUG_FUNCPTR (gst_base_sink_pad_buffer_alloc));
248 gst_pad_set_activate_function (basesink->sinkpad,
249 GST_DEBUG_FUNCPTR (gst_base_sink_activate));
250 gst_pad_set_activatepush_function (basesink->sinkpad,
251 GST_DEBUG_FUNCPTR (gst_base_sink_activate_push));
252 gst_pad_set_activatepull_function (basesink->sinkpad,
253 GST_DEBUG_FUNCPTR (gst_base_sink_activate_pull));
254 gst_pad_set_event_function (basesink->sinkpad,
255 GST_DEBUG_FUNCPTR (gst_base_sink_event));
256 gst_pad_set_chain_function (basesink->sinkpad,
257 GST_DEBUG_FUNCPTR (gst_base_sink_chain));
258 gst_element_add_pad (GST_ELEMENT (basesink), basesink->sinkpad);
260 basesink->pad_mode = GST_ACTIVATE_NONE;
261 GST_PAD_TASK (basesink->sinkpad) = NULL;
262 basesink->preroll_queue = g_queue_new ();
264 basesink->can_activate_push = DEFAULT_CAN_ACTIVATE_PUSH;
265 basesink->can_activate_pull = DEFAULT_CAN_ACTIVATE_PULL;
267 GST_FLAG_SET (basesink, GST_ELEMENT_IS_SINK);
271 gst_base_sink_finalize (GObject * object)
273 GstBaseSink *basesink;
275 basesink = GST_BASE_SINK (object);
277 g_queue_free (basesink->preroll_queue);
279 G_OBJECT_CLASS (parent_class)->finalize (object);
283 gst_base_sink_set_clock (GstElement * element, GstClock * clock)
287 sink = GST_BASE_SINK (element);
293 gst_base_sink_set_property (GObject * object, guint prop_id,
294 const GValue * value, GParamSpec * pspec)
296 GstBaseSink *sink = GST_BASE_SINK (object);
299 case PROP_PREROLL_QUEUE_LEN:
300 /* preroll lock necessary to serialize with finish_preroll */
301 GST_PREROLL_LOCK (sink->sinkpad);
302 sink->preroll_queue_max_len = g_value_get_uint (value);
303 GST_PREROLL_UNLOCK (sink->sinkpad);
306 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
312 gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
315 GstBaseSink *sink = GST_BASE_SINK (object);
319 case PROP_PREROLL_QUEUE_LEN:
320 g_value_set_uint (value, sink->preroll_queue_max_len);
323 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
330 gst_base_sink_get_caps (GstBaseSink * sink)
336 gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
342 gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
343 GstCaps * caps, GstBuffer ** buf)
349 /* with PREROLL_LOCK */
351 gst_base_sink_preroll_queue_empty (GstBaseSink * basesink, GstPad * pad)
354 GQueue *q = basesink->preroll_queue;
360 GST_DEBUG ("emptying queue");
361 while ((obj = g_queue_pop_head (q))) {
364 is_buffer = GST_IS_BUFFER (obj);
366 basesink->preroll_queued--;
367 basesink->buffers_queued--;
369 switch (GST_EVENT_TYPE (obj)) {
371 basesink->preroll_queued--;
376 basesink->events_queued--;
378 /* we release the preroll lock while pushing so that we
379 * can still flush it while blocking on the clock or
380 * inside the element. */
381 GST_PREROLL_UNLOCK (pad);
384 GST_DEBUG ("popped buffer %p", obj);
385 ret = gst_base_sink_handle_buffer (basesink, GST_BUFFER (obj));
387 GST_DEBUG ("popped event %p", obj);
388 gst_base_sink_handle_event (basesink, GST_EVENT (obj));
392 GST_PREROLL_LOCK (pad);
394 GST_DEBUG ("queue empty");
399 /* with PREROLL_LOCK */
401 gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
404 GQueue *q = basesink->preroll_queue;
406 GST_DEBUG ("flushing queue %p", basesink);
408 while ((obj = g_queue_pop_head (q))) {
409 GST_DEBUG ("popped %p", obj);
410 gst_mini_object_unref (obj);
413 /* we can't have EOS anymore now */
414 basesink->eos = FALSE;
415 basesink->preroll_queued = 0;
416 basesink->buffers_queued = 0;
417 basesink->events_queued = 0;
418 basesink->have_preroll = FALSE;
419 /* and signal any waiters now */
420 GST_PREROLL_SIGNAL (pad);
423 /* with STREAM_LOCK */
425 gst_base_sink_handle_object (GstBaseSink * basesink, GstPad * pad,
431 GST_PREROLL_LOCK (pad);
432 /* push object on the queue */
433 GST_DEBUG_OBJECT (basesink, "push %p on preroll_queue", obj);
434 g_queue_push_tail (basesink->preroll_queue, obj);
436 have_event = GST_IS_EVENT (obj);
438 GstEvent *event = GST_EVENT (obj);
440 switch (GST_EVENT_TYPE (obj)) {
442 basesink->preroll_queued++;
443 basesink->eos = TRUE;
445 case GST_EVENT_NEWSEGMENT:
448 gint64 segment_start;
451 /* the newsegment event is needed to bring the buffer timestamps to the
452 * stream time and to drop samples outside of the playback segment. */
453 gst_event_parse_newsegment (event, &basesink->segment_rate, &format,
454 &segment_start, &segment_stop, &basesink->segment_base);
456 basesink->have_newsegment = TRUE;
458 if (format != GST_FORMAT_TIME) {
459 GST_DEBUG_OBJECT (basesink,
460 "received non time %d NEW_SEGMENT %" G_GINT64_FORMAT
461 " -- %" G_GINT64_FORMAT ", base %" G_GINT64_FORMAT,
462 format, basesink->segment_start, basesink->segment_stop,
463 basesink->segment_base);
465 /* this means this sink will not be able to clip or drop samples
466 * and timestamps have to start from 0. */
467 basesink->segment_start = -1;
468 basesink->segment_stop = -1;
469 basesink->segment_base = -1;
470 goto done_newsegment;
472 /* check if we really have a new segment or the previous one is
474 if (basesink->segment_start != segment_start) {
475 /* the new segment has to be aligned with the old segment.
476 * We first update the accumulated time of the previous
477 * segment. the accumulated time is used when syncing to the
478 * clock. A flush event sets the accumulated time back to 0
480 if (GST_CLOCK_TIME_IS_VALID (basesink->segment_stop)) {
481 basesink->segment_accum +=
482 basesink->segment_stop - basesink->segment_start;
483 } else if (GST_CLOCK_TIME_IS_VALID (basesink->current_end)) {
484 /* else use last seen timestamp as segment stop */
485 basesink->segment_accum +=
486 basesink->current_end - basesink->segment_start;
488 basesink->segment_accum = 0;
492 basesink->segment_start = segment_start;
493 basesink->segment_stop = segment_stop;
495 GST_DEBUG_OBJECT (basesink,
496 "received DISCONT %" GST_TIME_FORMAT " -- %"
497 GST_TIME_FORMAT ", base %" GST_TIME_FORMAT ", accum %"
499 GST_TIME_ARGS (basesink->segment_start),
500 GST_TIME_ARGS (basesink->segment_stop),
501 GST_TIME_ARGS (basesink->segment_base),
502 GST_TIME_ARGS (basesink->segment_accum));
509 basesink->events_queued++;
511 GstBuffer *buf = GST_BUFFER (obj);
513 if (!basesink->have_newsegment) {
514 GST_ELEMENT_WARNING (basesink, STREAM, STOPPED,
515 ("Received buffer without a new-segment. Cannot sync to clock."),
516 ("Received buffer without a new-segment. Cannot sync to clock."));
517 basesink->have_newsegment = TRUE;
518 /* this means this sink will not be able to sync to the clock */
519 basesink->segment_start = -1;
520 basesink->segment_stop = -1;
523 /* check if the buffer needs to be dropped */
525 GstClockTime start = -1, end = -1;
527 /* we don't use the subclassed method as it may not return
528 * valid values for our purpose here */
529 gst_base_sink_get_times (basesink, buf, &start, &end);
531 GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
532 ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start),
533 GST_TIME_ARGS (end));
535 /* need to drop if the timestamp is not between segment_start and
536 * segment_stop. we check if the complete sample is outside of the
537 * range since the sink might be able to clip the sample. */
538 if (GST_CLOCK_TIME_IS_VALID (end) &&
539 GST_CLOCK_TIME_IS_VALID (basesink->segment_start)) {
540 if (end <= basesink->segment_start) {
541 GST_DEBUG ("buffer end %" GST_TIME_FORMAT " <= segment start %"
542 GST_TIME_FORMAT ", dropping buffer", GST_TIME_ARGS (end),
543 GST_TIME_ARGS (basesink->segment_start));
547 if (GST_CLOCK_TIME_IS_VALID (start) &&
548 GST_CLOCK_TIME_IS_VALID (basesink->segment_stop)) {
549 if (basesink->segment_stop <= start) {
550 GST_DEBUG ("buffer start %" GST_TIME_FORMAT " >= segment stop %"
551 GST_TIME_FORMAT ", dropping buffer", GST_TIME_ARGS (start),
552 GST_TIME_ARGS (basesink->segment_stop));
557 basesink->preroll_queued++;
558 basesink->buffers_queued++;
560 GST_DEBUG ("now %d preroll, %d buffers, %d events on queue",
561 basesink->preroll_queued,
562 basesink->buffers_queued, basesink->events_queued);
564 if (basesink->playing_async)
567 /* check if we are prerolling */
568 if (!basesink->need_preroll)
571 /* there is a buffer queued */
572 if (basesink->buffers_queued == 1) {
573 GST_DEBUG ("do preroll %p", obj);
575 /* if it's a buffer, we need to call the preroll method */
576 if (GST_IS_BUFFER (obj)) {
577 GstBaseSinkClass *bclass;
580 bclass = GST_BASE_SINK_GET_CLASS (basesink);
583 bclass->preroll (basesink, GST_BUFFER (obj))) != GST_FLOW_OK)
587 length = basesink->preroll_queued;
588 GST_DEBUG ("prerolled length %d", length);
593 basesink->have_preroll = TRUE;
594 /* we are prerolling */
595 GST_PREROLL_UNLOCK (pad);
597 /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
598 * inside the STREAM_LOCK */
599 t = GST_STREAM_UNLOCK_FULL (pad);
600 GST_DEBUG ("released stream lock %d times", t);
602 GST_WARNING ("STREAM_LOCK should have been locked !!");
603 g_warning ("STREAM_LOCK should have been locked !!");
606 /* now we commit our state */
607 GST_STATE_LOCK (basesink);
608 GST_DEBUG_OBJECT (basesink, "commit state");
609 gst_element_commit_state (GST_ELEMENT (basesink));
610 GST_STATE_UNLOCK (basesink);
612 /* reacquire stream lock, pad could be flushing now */
613 /* FIXME in glib, if t==0, the lock is still taken... hmmm */
615 GST_STREAM_LOCK_FULL (pad, t);
617 /* and wait if needed */
618 GST_PREROLL_LOCK (pad);
621 if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
625 /* it is possible that the application set the state to PLAYING
626 * now in which case we don't need to block anymore. */
627 if (!basesink->need_preroll)
630 length = basesink->preroll_queued;
632 g_assert (length == 1);
635 /* see if we need to block now. We cannot block on events, only
636 * on buffers, the reason is that events can be sent from the
637 * application thread and we don't want to block there. */
638 if (length > basesink->preroll_queue_max_len && !have_event) {
639 /* block until the state changes, or we get a flush, or something */
640 GST_DEBUG_OBJECT (basesink, "waiting to finish preroll");
641 GST_PREROLL_WAIT (pad);
642 GST_DEBUG_OBJECT (basesink, "done preroll");
645 if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
649 GST_PREROLL_UNLOCK (pad);
657 GST_DEBUG ("no preroll needed");
658 /* maybe it was another sink that blocked in preroll, need to check for
660 basesink->have_preroll = FALSE;
661 ret = gst_base_sink_preroll_queue_empty (basesink, pad);
662 GST_PREROLL_UNLOCK (pad);
670 buf = GST_BUFFER (g_queue_pop_tail (basesink->preroll_queue));
672 gst_buffer_unref (buf);
673 GST_PREROLL_UNLOCK (pad);
682 basesink->have_preroll = FALSE;
683 basesink->playing_async = FALSE;
685 /* handle buffer first */
686 ret = gst_base_sink_preroll_queue_empty (basesink, pad);
688 /* unroll locks, commit state, reacquire stream lock */
689 GST_PREROLL_UNLOCK (pad);
690 t = GST_STREAM_UNLOCK_FULL (pad);
691 GST_DEBUG ("released stream lock %d times", t);
693 GST_WARNING ("STREAM_LOCK should have been locked !!");
694 g_warning ("STREAM_LOCK should have been locked !!");
696 GST_STATE_LOCK (basesink);
697 GST_DEBUG_OBJECT (basesink, "commit state");
698 gst_element_commit_state (GST_ELEMENT (basesink));
699 GST_STATE_UNLOCK (basesink);
701 GST_STREAM_LOCK_FULL (pad, t);
708 gst_base_sink_preroll_queue_flush (basesink, pad);
709 GST_PREROLL_UNLOCK (pad);
710 GST_DEBUG ("pad is flushing");
711 return GST_FLOW_WRONG_STATE;
717 GST_DEBUG ("preroll failed");
718 gst_base_sink_preroll_queue_flush (basesink, pad);
719 GST_PREROLL_UNLOCK (pad);
721 /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
722 * inside the STREAM_LOCK */
723 t = GST_STREAM_UNLOCK_FULL (pad);
724 GST_DEBUG ("released stream lock %d times", t);
726 GST_WARNING ("STREAM_LOCK should have been locked !!");
727 g_warning ("STREAM_LOCK should have been locked !!");
730 /* now we abort our state */
731 GST_STATE_LOCK (basesink);
732 GST_DEBUG_OBJECT (basesink, "abort state");
733 gst_element_abort_state (GST_ELEMENT (basesink));
734 GST_STATE_UNLOCK (basesink);
736 /* reacquire stream lock, pad could be flushing now */
738 GST_STREAM_LOCK_FULL (pad, t);
740 return GST_FLOW_ERROR;
745 gst_base_sink_event (GstPad * pad, GstEvent * event)
747 GstBaseSink *basesink;
748 gboolean result = TRUE;
749 GstBaseSinkClass *bclass;
751 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
753 bclass = GST_BASE_SINK_GET_CLASS (basesink);
755 GST_DEBUG ("event %p", event);
757 switch (GST_EVENT_TYPE (event)) {
762 GST_STREAM_LOCK (pad);
763 /* EOS also finishes the preroll */
765 gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
766 GST_STREAM_UNLOCK (pad);
769 case GST_EVENT_NEWSEGMENT:
773 GST_STREAM_LOCK (pad);
775 gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
776 GST_STREAM_UNLOCK (pad);
779 case GST_EVENT_FLUSH_START:
780 /* make sure we are not blocked on the clock also clear any pending
783 bclass->event (basesink, event);
785 GST_PREROLL_LOCK (pad);
786 /* we need preroll after the flush */
787 basesink->need_preroll = TRUE;
788 /* unlock from a possible state change/preroll */
789 gst_base_sink_preroll_queue_flush (basesink, pad);
792 if (basesink->clock_id) {
793 gst_clock_id_unschedule (basesink->clock_id);
795 GST_UNLOCK (basesink);
796 GST_PREROLL_UNLOCK (pad);
798 /* and we need to commit our state again on the next
799 * prerolled buffer */
800 GST_STATE_LOCK (basesink);
801 GST_STREAM_LOCK (pad);
802 gst_element_lost_state (GST_ELEMENT (basesink));
803 GST_STREAM_UNLOCK (pad);
804 GST_STATE_UNLOCK (basesink);
805 GST_DEBUG ("event unref %p %p", basesink, event);
806 gst_event_unref (event);
808 case GST_EVENT_FLUSH_STOP:
810 bclass->event (basesink, event);
812 /* now we are completely unblocked and the _chain method
814 GST_STREAM_LOCK (pad);
815 /* we need new segment info after the flush. */
816 basesink->segment_start = -1;
817 basesink->segment_stop = -1;
818 basesink->current_start = -1;
819 basesink->current_end = -1;
820 GST_DEBUG ("reset accum %" GST_TIME_FORMAT,
821 GST_TIME_ARGS (basesink->segment_accum));
822 basesink->segment_accum = 0;
823 GST_STREAM_UNLOCK (pad);
825 GST_DEBUG ("event unref %p %p", basesink, event);
826 gst_event_unref (event);
829 gst_event_unref (event);
832 gst_object_unref (basesink);
837 /* default implementation to calculate the start and end
838 * timestamps on a buffer, subclasses can override
841 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
842 GstClockTime * start, GstClockTime * end)
844 GstClockTime timestamp, duration;
846 timestamp = GST_BUFFER_TIMESTAMP (buffer);
847 if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
849 /* get duration to calculate end time */
850 duration = GST_BUFFER_DURATION (buffer);
851 if (GST_CLOCK_TIME_IS_VALID (duration)) {
852 *end = timestamp + duration;
858 /* perform synchronisation on a buffer
860 * 1) check if we have a clock, if not, do nothing
861 * 2) calculate the start and end time of the buffer
862 * 3) create a single shot notification to wait on
863 * the clock, save the entry so we can unlock it
864 * 4) wait on the clock, this blocks
865 * 5) unref the clockid again
868 gst_base_sink_do_sync (GstBaseSink * basesink, GstBuffer * buffer)
870 gboolean result = TRUE;
871 GstClockTime start, end;
872 GstClockTimeDiff stream_start, stream_end;
873 GstBaseSinkClass *bclass;
874 gboolean start_valid, end_valid;
876 bclass = GST_BASE_SINK_GET_CLASS (basesink);
879 if (bclass->get_times)
880 bclass->get_times (basesink, buffer, &start, &end);
882 start_valid = GST_CLOCK_TIME_IS_VALID (start);
883 end_valid = GST_CLOCK_TIME_IS_VALID (start);
885 GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
886 ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
888 /* if we don't have a timestamp, we don't sync */
892 /* save last times seen. */
893 basesink->current_start = start;
895 basesink->current_end = end;
897 basesink->current_end = start;
899 if (GST_CLOCK_TIME_IS_VALID (basesink->segment_stop)) {
900 /* check if not outside of the segment range, start is
901 * always valid here. */
902 if (start > basesink->segment_stop)
906 /* bring timestamp to stream time using last segment offset. */
907 if (GST_CLOCK_TIME_IS_VALID (basesink->segment_start)) {
908 /* check if not outside of the segment range */
909 if (end_valid && end < basesink->segment_start)
912 stream_start = (gint64) start - basesink->segment_start;
913 stream_end = (gint64) end - basesink->segment_start;
915 stream_start = (gint64) start;
916 stream_end = (gint64) end;
919 stream_start += basesink->segment_accum;
921 stream_end += basesink->segment_accum;
923 /* now do clocking */
924 if (basesink->clock) {
926 GstClockTime base_time;
929 base_time = GST_ELEMENT (basesink)->base_time;
931 GST_LOG_OBJECT (basesink,
932 "waiting for clock, base time %" GST_TIME_FORMAT,
933 GST_TIME_ARGS (base_time));
935 /* save clock id so that we can unlock it if needed */
936 basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
937 stream_start + base_time);
938 /* also save end_time of this buffer so that we can wait
941 basesink->end_time = stream_end + base_time;
943 basesink->end_time = GST_CLOCK_TIME_NONE;
944 GST_UNLOCK (basesink);
946 ret = gst_clock_id_wait (basesink->clock_id, NULL);
949 if (basesink->clock_id) {
950 gst_clock_id_unref (basesink->clock_id);
951 basesink->clock_id = NULL;
953 GST_UNLOCK (basesink);
955 GST_LOG_OBJECT (basesink, "clock entry done: %d", ret);
956 if (ret == GST_CLOCK_UNSCHEDULED)
965 GST_LOG_OBJECT (basesink, "buffer skipped, not in segment");
973 * 2) render the event
976 static inline gboolean
977 gst_base_sink_handle_event (GstBaseSink * basesink, GstEvent * event)
979 GstBaseSinkClass *bclass;
982 switch (GST_EVENT_TYPE (event)) {
985 if (basesink->clock) {
986 /* wait for last buffer to finish if we have a valid end time */
987 if (GST_CLOCK_TIME_IS_VALID (basesink->end_time)) {
988 basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
990 GST_UNLOCK (basesink);
992 gst_clock_id_wait (basesink->clock_id, NULL);
995 if (basesink->clock_id) {
996 gst_clock_id_unref (basesink->clock_id);
997 basesink->clock_id = NULL;
999 basesink->end_time = GST_CLOCK_TIME_NONE;
1002 GST_UNLOCK (basesink);
1008 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1010 ret = bclass->event (basesink, event);
1014 switch (GST_EVENT_TYPE (event)) {
1016 GST_PREROLL_LOCK (basesink->sinkpad);
1017 /* if we are still EOS, we can post the EOS message */
1018 if (basesink->eos) {
1019 /* ok, now we can post the message */
1020 GST_DEBUG_OBJECT (basesink, "Now posting EOS");
1021 gst_element_post_message (GST_ELEMENT (basesink),
1022 gst_message_new_eos (GST_OBJECT (basesink)));
1024 GST_PREROLL_UNLOCK (basesink->sinkpad);
1030 GST_DEBUG ("event unref %p %p", basesink, event);
1031 gst_event_unref (event);
1038 * 1) first sync on the buffer
1039 * 2) render the buffer
1040 * 3) unref the buffer
1042 static inline GstFlowReturn
1043 gst_base_sink_handle_buffer (GstBaseSink * basesink, GstBuffer * buf)
1045 GstFlowReturn ret = GST_FLOW_OK;
1048 render = gst_base_sink_do_sync (basesink, buf);
1051 GstBaseSinkClass *bclass;
1053 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1055 ret = bclass->render (basesink, buf);
1058 GST_DEBUG ("buffer unref after render %p", basesink, buf);
1059 gst_buffer_unref (buf);
1064 static GstFlowReturn
1065 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
1067 GstBaseSink *basesink;
1068 GstFlowReturn result;
1070 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1072 if (!(basesink->pad_mode == GST_ACTIVATE_PUSH)) {
1074 g_warning ("Push on pad %s:%s, but it was not activated in push mode",
1075 GST_DEBUG_PAD_NAME (pad));
1077 result = GST_FLOW_UNEXPECTED;
1081 result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
1084 gst_object_unref (basesink);
1090 gst_base_sink_loop (GstPad * pad)
1092 GstBaseSink *basesink;
1093 GstBuffer *buf = NULL;
1094 GstFlowReturn result;
1096 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1098 g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
1100 result = gst_pad_pull_range (pad, basesink->offset, DEFAULT_SIZE, &buf);
1101 if (result != GST_FLOW_OK)
1104 result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
1105 if (result != GST_FLOW_OK)
1108 gst_object_unref (basesink);
1115 gst_base_sink_event (pad, gst_event_new_eos ());
1116 gst_object_unref (basesink);
1117 gst_pad_pause_task (pad);
1123 gst_base_sink_deactivate (GstBaseSink * basesink, GstPad * pad)
1125 gboolean result = FALSE;
1126 GstBaseSinkClass *bclass;
1128 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1130 /* step 1, unblock clock sync (if any) or any other blocking thing */
1131 GST_PREROLL_LOCK (pad);
1132 GST_LOCK (basesink);
1133 if (basesink->clock_id) {
1134 gst_clock_id_unschedule (basesink->clock_id);
1136 GST_UNLOCK (basesink);
1138 /* unlock any subclasses */
1140 bclass->unlock (basesink);
1142 /* flush out the data thread if it's locked in finish_preroll */
1143 basesink->need_preroll = FALSE;
1144 gst_base_sink_preroll_queue_flush (basesink, pad);
1145 GST_PREROLL_SIGNAL (pad);
1146 GST_PREROLL_UNLOCK (pad);
1148 /* step 2, make sure streaming finishes */
1149 result = gst_pad_stop_task (pad);
1155 gst_base_sink_activate (GstPad * pad)
1157 gboolean result = FALSE;
1158 GstBaseSink *basesink;
1160 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1162 GST_DEBUG_OBJECT (basesink, "Trying pull mode first");
1164 if (basesink->can_activate_pull && gst_pad_check_pull_range (pad)
1165 && gst_pad_activate_pull (pad, TRUE)) {
1166 GST_DEBUG_OBJECT (basesink, "Success activating pull mode");
1169 GST_DEBUG_OBJECT (basesink, "Falling back to push mode");
1170 if (gst_pad_activate_push (pad, TRUE)) {
1171 GST_DEBUG_OBJECT (basesink, "Success activating push mode");
1177 GST_WARNING_OBJECT (basesink, "Could not activate pad in either mode");
1180 gst_object_unref (basesink);
1186 gst_base_sink_activate_push (GstPad * pad, gboolean active)
1189 GstBaseSink *basesink;
1191 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1194 if (!basesink->can_activate_push) {
1196 basesink->pad_mode = GST_ACTIVATE_NONE;
1199 basesink->pad_mode = GST_ACTIVATE_PUSH;
1202 if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH)) {
1203 g_warning ("Internal GStreamer activation error!!!");
1206 result = gst_base_sink_deactivate (basesink, pad);
1207 basesink->pad_mode = GST_ACTIVATE_NONE;
1211 gst_object_unref (basesink);
1216 /* this won't get called until we implement an activate function */
1218 gst_base_sink_activate_pull (GstPad * pad, gboolean active)
1220 gboolean result = FALSE;
1221 GstBaseSink *basesink;
1223 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1226 if (!basesink->can_activate_pull) {
1228 basesink->pad_mode = GST_ACTIVATE_NONE;
1230 GstPad *peer = gst_pad_get_peer (pad);
1232 if (G_UNLIKELY (peer == NULL)) {
1233 g_warning ("Trying to activate pad in pull mode, but no peer");
1236 if (gst_pad_activate_pull (peer, TRUE)) {
1237 basesink->have_newsegment = TRUE;
1238 basesink->segment_start = basesink->segment_stop = 0;
1240 gst_pad_start_task (pad, (GstTaskFunction) gst_base_sink_loop,
1243 GST_DEBUG_OBJECT (pad, "Failed to activate peer in pull mode");
1246 gst_object_unref (peer);
1249 basesink->pad_mode = result ? GST_ACTIVATE_PULL : GST_ACTIVATE_NONE;
1252 if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PULL)) {
1253 g_warning ("Internal GStreamer activation error!!!");
1256 basesink->have_newsegment = FALSE;
1257 result = gst_base_sink_deactivate (basesink, pad);
1258 basesink->pad_mode = GST_ACTIVATE_NONE;
1262 gst_object_unref (basesink);
1267 static GstStateChangeReturn
1268 gst_base_sink_change_state (GstElement * element, GstStateChange transition)
1270 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1271 GstBaseSink *basesink = GST_BASE_SINK (element);
1272 GstBaseSinkClass *bclass;
1274 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1276 switch (transition) {
1277 case GST_STATE_CHANGE_NULL_TO_READY:
1279 if (!bclass->start (basesink))
1282 case GST_STATE_CHANGE_READY_TO_PAUSED:
1283 /* need to complete preroll before this state change completes, there
1284 * is no data flow in READY so we can safely assume we need to preroll. */
1285 basesink->offset = 0;
1286 GST_PREROLL_LOCK (basesink->sinkpad);
1287 basesink->have_preroll = FALSE;
1288 basesink->need_preroll = TRUE;
1289 GST_PREROLL_UNLOCK (basesink->sinkpad);
1290 basesink->have_newsegment = FALSE;
1291 basesink->segment_rate = 1.0;
1292 basesink->segment_start = 0;
1293 basesink->segment_stop = 0;
1294 ret = GST_STATE_CHANGE_ASYNC;
1296 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1298 GST_PREROLL_LOCK (basesink->sinkpad);
1299 /* if we have EOS, we should empty the queue now as there will
1300 * be no more data received in the chain function.
1301 * FIXME, this could block the state change function too long when
1302 * we are pushing and syncing the buffers, better start a new
1303 * thread to do this. */
1304 if (basesink->eos) {
1305 gst_base_sink_preroll_queue_empty (basesink, basesink->sinkpad);
1306 } else if (!basesink->have_preroll) {
1307 /* don't need preroll, but do queue a commit_state */
1308 basesink->need_preroll = FALSE;
1309 basesink->playing_async = TRUE;
1310 ret = GST_STATE_CHANGE_ASYNC;
1311 /* we know it's not waiting, no need to signal */
1313 /* don't need the preroll anymore */
1314 basesink->need_preroll = FALSE;
1315 /* now let it play */
1316 GST_PREROLL_SIGNAL (basesink->sinkpad);
1318 GST_PREROLL_UNLOCK (basesink->sinkpad);
1326 GstStateChangeReturn bret;
1328 bret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1330 if (bret != GST_STATE_CHANGE_SUCCESS)
1331 goto activate_failed;
1334 switch (transition) {
1335 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1337 GstBaseSinkClass *bclass;
1339 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1341 GST_PREROLL_LOCK (basesink->sinkpad);
1342 GST_LOCK (basesink);
1343 /* unlock clock wait if any */
1344 if (basesink->clock_id) {
1345 gst_clock_id_unschedule (basesink->clock_id);
1347 GST_UNLOCK (basesink);
1349 basesink->playing_async = FALSE;
1351 /* unlock any subclasses */
1353 bclass->unlock (basesink);
1355 /* if we don't have a preroll buffer and we have not received EOS,
1356 * we need to wait for a preroll */
1357 GST_DEBUG ("have_preroll: %d, EOS: %d", basesink->have_preroll,
1359 if (!basesink->have_preroll && !basesink->eos) {
1360 basesink->need_preroll = TRUE;
1361 ret = GST_STATE_CHANGE_ASYNC;
1363 GST_PREROLL_UNLOCK (basesink->sinkpad);
1366 case GST_STATE_CHANGE_PAUSED_TO_READY:
1368 case GST_STATE_CHANGE_READY_TO_NULL:
1370 if (!bclass->stop (basesink)) {
1371 GST_WARNING ("failed to stop");
1383 GST_DEBUG ("failed to start");
1384 return GST_STATE_CHANGE_FAILURE;
1388 GST_DEBUG ("element failed to change states -- activation problem?");
1389 return GST_STATE_CHANGE_FAILURE;