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_OBJECT (basesink, "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_OBJECT (basesink, "popped buffer %p", obj);
385 ret = gst_base_sink_handle_buffer (basesink, GST_BUFFER (obj));
387 GST_DEBUG_OBJECT (basesink, "popped event %p", obj);
388 gst_base_sink_handle_event (basesink, GST_EVENT (obj));
392 GST_PREROLL_LOCK (pad);
394 GST_DEBUG_OBJECT (basesink, "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_OBJECT (basesink, "flushing queue %p", basesink);
408 while ((obj = g_queue_pop_head (q))) {
409 GST_DEBUG_OBJECT (basesink, "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_OBJECT (basesink,
542 "buffer end %" GST_TIME_FORMAT " <= segment start %"
543 GST_TIME_FORMAT ", dropping buffer", GST_TIME_ARGS (end),
544 GST_TIME_ARGS (basesink->segment_start));
548 if (GST_CLOCK_TIME_IS_VALID (start) &&
549 GST_CLOCK_TIME_IS_VALID (basesink->segment_stop)) {
550 if (basesink->segment_stop <= start) {
551 GST_DEBUG_OBJECT (basesink,
552 "buffer start %" GST_TIME_FORMAT " >= segment stop %"
553 GST_TIME_FORMAT ", dropping buffer", GST_TIME_ARGS (start),
554 GST_TIME_ARGS (basesink->segment_stop));
559 basesink->preroll_queued++;
560 basesink->buffers_queued++;
562 GST_DEBUG_OBJECT (basesink,
563 "now %d preroll, %d buffers, %d events on queue",
564 basesink->preroll_queued,
565 basesink->buffers_queued, basesink->events_queued);
567 if (basesink->playing_async)
570 /* check if we are prerolling */
571 if (!basesink->need_preroll)
574 /* there is a buffer queued */
575 if (basesink->buffers_queued == 1) {
576 GST_DEBUG_OBJECT (basesink, "do preroll %p", obj);
578 /* if it's a buffer, we need to call the preroll method */
579 if (GST_IS_BUFFER (obj)) {
580 GstBaseSinkClass *bclass;
583 bclass = GST_BASE_SINK_GET_CLASS (basesink);
586 bclass->preroll (basesink, GST_BUFFER (obj))) != GST_FLOW_OK)
590 length = basesink->preroll_queued;
591 GST_DEBUG_OBJECT (basesink, "prerolled length %d", length);
596 basesink->have_preroll = TRUE;
597 /* we are prerolling */
598 GST_PREROLL_UNLOCK (pad);
600 /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
601 * inside the STREAM_LOCK */
602 t = GST_STREAM_UNLOCK_FULL (pad);
603 GST_DEBUG_OBJECT (basesink, "released stream lock %d times", t);
605 GST_WARNING ("STREAM_LOCK should have been locked !!");
606 g_warning ("STREAM_LOCK should have been locked !!");
609 /* now we commit our state */
610 GST_STATE_LOCK (basesink);
611 GST_DEBUG_OBJECT (basesink, "commit state");
612 gst_element_commit_state (GST_ELEMENT (basesink));
613 GST_STATE_UNLOCK (basesink);
615 /* reacquire stream lock, pad could be flushing now */
616 /* FIXME in glib, if t==0, the lock is still taken... hmmm */
618 GST_STREAM_LOCK_FULL (pad, t);
620 /* and wait if needed */
621 GST_PREROLL_LOCK (pad);
624 if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
628 /* it is possible that the application set the state to PLAYING
629 * now in which case we don't need to block anymore. */
630 if (!basesink->need_preroll)
633 length = basesink->preroll_queued;
635 g_assert (length == 1);
638 /* see if we need to block now. We cannot block on events, only
639 * on buffers, the reason is that events can be sent from the
640 * application thread and we don't want to block there. */
641 if (length > basesink->preroll_queue_max_len && !have_event) {
642 /* block until the state changes, or we get a flush, or something */
643 GST_DEBUG_OBJECT (basesink, "waiting to finish preroll");
644 GST_PREROLL_WAIT (pad);
645 GST_DEBUG_OBJECT (basesink, "done preroll");
648 if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
652 GST_PREROLL_UNLOCK (pad);
660 GST_DEBUG_OBJECT (basesink, "no preroll needed");
661 /* maybe it was another sink that blocked in preroll, need to check for
663 basesink->have_preroll = FALSE;
664 ret = gst_base_sink_preroll_queue_empty (basesink, pad);
665 GST_PREROLL_UNLOCK (pad);
673 buf = GST_BUFFER (g_queue_pop_tail (basesink->preroll_queue));
675 gst_buffer_unref (buf);
676 GST_PREROLL_UNLOCK (pad);
685 basesink->have_preroll = FALSE;
686 basesink->playing_async = FALSE;
688 /* handle buffer first */
689 ret = gst_base_sink_preroll_queue_empty (basesink, pad);
691 /* unroll locks, commit state, reacquire stream lock */
692 GST_PREROLL_UNLOCK (pad);
693 t = GST_STREAM_UNLOCK_FULL (pad);
694 GST_DEBUG_OBJECT (basesink, "released stream lock %d times", t);
696 GST_WARNING ("STREAM_LOCK should have been locked !!");
697 g_warning ("STREAM_LOCK should have been locked !!");
699 GST_STATE_LOCK (basesink);
700 GST_DEBUG_OBJECT (basesink, "commit state");
701 gst_element_commit_state (GST_ELEMENT (basesink));
702 GST_STATE_UNLOCK (basesink);
704 GST_STREAM_LOCK_FULL (pad, t);
711 gst_base_sink_preroll_queue_flush (basesink, pad);
712 GST_PREROLL_UNLOCK (pad);
713 GST_DEBUG_OBJECT (basesink, "pad is flushing");
714 return GST_FLOW_WRONG_STATE;
720 GST_DEBUG_OBJECT (basesink, "preroll failed");
721 gst_base_sink_preroll_queue_flush (basesink, pad);
722 GST_PREROLL_UNLOCK (pad);
724 /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
725 * inside the STREAM_LOCK */
726 t = GST_STREAM_UNLOCK_FULL (pad);
727 GST_DEBUG_OBJECT (basesink, "released stream lock %d times", t);
729 GST_WARNING ("STREAM_LOCK should have been locked !!");
730 g_warning ("STREAM_LOCK should have been locked !!");
733 /* now we abort our state */
734 GST_STATE_LOCK (basesink);
735 GST_DEBUG_OBJECT (basesink, "abort state");
736 gst_element_abort_state (GST_ELEMENT (basesink));
737 GST_STATE_UNLOCK (basesink);
739 /* reacquire stream lock, pad could be flushing now */
741 GST_STREAM_LOCK_FULL (pad, t);
743 return GST_FLOW_ERROR;
748 gst_base_sink_event (GstPad * pad, GstEvent * event)
750 GstBaseSink *basesink;
751 gboolean result = TRUE;
752 GstBaseSinkClass *bclass;
754 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
756 bclass = GST_BASE_SINK_GET_CLASS (basesink);
758 GST_DEBUG_OBJECT (basesink, "event %p", event);
760 switch (GST_EVENT_TYPE (event)) {
765 GST_STREAM_LOCK (pad);
766 /* EOS also finishes the preroll */
768 gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
769 GST_STREAM_UNLOCK (pad);
772 case GST_EVENT_NEWSEGMENT:
776 GST_STREAM_LOCK (pad);
778 gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
779 GST_STREAM_UNLOCK (pad);
782 case GST_EVENT_FLUSH_START:
783 /* make sure we are not blocked on the clock also clear any pending
786 bclass->event (basesink, event);
788 GST_PREROLL_LOCK (pad);
789 /* we need preroll after the flush */
790 GST_DEBUG_OBJECT (basesink, "flushing, need preroll after flush");
791 basesink->need_preroll = TRUE;
792 /* unlock from a possible state change/preroll */
793 gst_base_sink_preroll_queue_flush (basesink, pad);
796 if (basesink->clock_id) {
797 gst_clock_id_unschedule (basesink->clock_id);
799 GST_UNLOCK (basesink);
800 GST_PREROLL_UNLOCK (pad);
802 /* and we need to commit our state again on the next
803 * prerolled buffer */
804 GST_STATE_LOCK (basesink);
805 GST_STREAM_LOCK (pad);
806 gst_element_lost_state (GST_ELEMENT (basesink));
807 GST_STREAM_UNLOCK (pad);
808 GST_STATE_UNLOCK (basesink);
809 GST_DEBUG_OBJECT (basesink, "event unref %p %p", basesink, event);
810 gst_event_unref (event);
812 case GST_EVENT_FLUSH_STOP:
814 bclass->event (basesink, event);
816 /* now we are completely unblocked and the _chain method
818 GST_STREAM_LOCK (pad);
819 /* we need new segment info after the flush. */
820 basesink->segment_start = -1;
821 basesink->segment_stop = -1;
822 basesink->current_start = -1;
823 basesink->current_end = -1;
824 GST_DEBUG_OBJECT (basesink, "reset accum %" GST_TIME_FORMAT,
825 GST_TIME_ARGS (basesink->segment_accum));
826 basesink->segment_accum = 0;
827 GST_STREAM_UNLOCK (pad);
829 GST_DEBUG_OBJECT (basesink, "event unref %p %p", basesink, event);
830 gst_event_unref (event);
833 gst_event_unref (event);
836 gst_object_unref (basesink);
841 /* default implementation to calculate the start and end
842 * timestamps on a buffer, subclasses can override
845 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
846 GstClockTime * start, GstClockTime * end)
848 GstClockTime timestamp, duration;
850 timestamp = GST_BUFFER_TIMESTAMP (buffer);
851 if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
853 /* get duration to calculate end time */
854 duration = GST_BUFFER_DURATION (buffer);
855 if (GST_CLOCK_TIME_IS_VALID (duration)) {
856 *end = timestamp + duration;
862 /* perform synchronisation on a buffer
864 * 1) check if we have a clock, if not, do nothing
865 * 2) calculate the start and end time of the buffer
866 * 3) create a single shot notification to wait on
867 * the clock, save the entry so we can unlock it
868 * 4) wait on the clock, this blocks
869 * 5) unref the clockid again
872 gst_base_sink_do_sync (GstBaseSink * basesink, GstBuffer * buffer)
874 gboolean result = TRUE;
875 GstClockTime start, end;
876 GstClockTimeDiff stream_start, stream_end;
877 GstBaseSinkClass *bclass;
878 gboolean start_valid, end_valid;
880 bclass = GST_BASE_SINK_GET_CLASS (basesink);
883 if (bclass->get_times)
884 bclass->get_times (basesink, buffer, &start, &end);
886 start_valid = GST_CLOCK_TIME_IS_VALID (start);
887 end_valid = GST_CLOCK_TIME_IS_VALID (start);
889 GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
890 ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
892 /* if we don't have a timestamp, we don't sync */
896 /* save last times seen. */
897 basesink->current_start = start;
899 basesink->current_end = end;
901 basesink->current_end = start;
903 if (GST_CLOCK_TIME_IS_VALID (basesink->segment_stop)) {
904 /* check if not outside of the segment range, start is
905 * always valid here. */
906 if (start > basesink->segment_stop)
910 /* bring timestamp to stream time using last segment offset. */
911 if (GST_CLOCK_TIME_IS_VALID (basesink->segment_start)) {
912 /* check if not outside of the segment range */
913 if (end_valid && end < basesink->segment_start)
916 stream_start = (gint64) start - basesink->segment_start;
917 stream_end = (gint64) end - basesink->segment_start;
919 stream_start = (gint64) start;
920 stream_end = (gint64) end;
923 stream_start += basesink->segment_accum;
925 stream_end += basesink->segment_accum;
927 /* now do clocking */
928 if (basesink->clock) {
930 GstClockTime base_time;
933 base_time = GST_ELEMENT (basesink)->base_time;
935 GST_LOG_OBJECT (basesink,
936 "waiting for clock, base time %" GST_TIME_FORMAT,
937 GST_TIME_ARGS (base_time));
939 /* save clock id so that we can unlock it if needed */
940 basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
941 stream_start + base_time);
942 /* also save end_time of this buffer so that we can wait
945 basesink->end_time = stream_end + base_time;
947 basesink->end_time = GST_CLOCK_TIME_NONE;
948 GST_UNLOCK (basesink);
950 ret = gst_clock_id_wait (basesink->clock_id, NULL);
953 if (basesink->clock_id) {
954 gst_clock_id_unref (basesink->clock_id);
955 basesink->clock_id = NULL;
957 GST_UNLOCK (basesink);
959 GST_LOG_OBJECT (basesink, "clock entry done: %d", ret);
960 if (ret == GST_CLOCK_UNSCHEDULED)
969 GST_LOG_OBJECT (basesink, "buffer skipped, not in segment");
977 * 2) render the event
980 static inline gboolean
981 gst_base_sink_handle_event (GstBaseSink * basesink, GstEvent * event)
983 GstBaseSinkClass *bclass;
986 switch (GST_EVENT_TYPE (event)) {
989 if (basesink->clock) {
990 /* wait for last buffer to finish if we have a valid end time */
991 if (GST_CLOCK_TIME_IS_VALID (basesink->end_time)) {
992 basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
994 GST_UNLOCK (basesink);
996 gst_clock_id_wait (basesink->clock_id, NULL);
999 if (basesink->clock_id) {
1000 gst_clock_id_unref (basesink->clock_id);
1001 basesink->clock_id = NULL;
1003 basesink->end_time = GST_CLOCK_TIME_NONE;
1006 GST_UNLOCK (basesink);
1012 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1014 ret = bclass->event (basesink, event);
1018 switch (GST_EVENT_TYPE (event)) {
1020 GST_PREROLL_LOCK (basesink->sinkpad);
1021 /* if we are still EOS, we can post the EOS message */
1022 if (basesink->eos) {
1023 /* ok, now we can post the message */
1024 GST_DEBUG_OBJECT (basesink, "Now posting EOS");
1025 gst_element_post_message (GST_ELEMENT (basesink),
1026 gst_message_new_eos (GST_OBJECT (basesink)));
1028 GST_PREROLL_UNLOCK (basesink->sinkpad);
1034 GST_DEBUG_OBJECT (basesink, "event unref %p %p", basesink, event);
1035 gst_event_unref (event);
1042 * 1) first sync on the buffer
1043 * 2) render the buffer
1044 * 3) unref the buffer
1046 static inline GstFlowReturn
1047 gst_base_sink_handle_buffer (GstBaseSink * basesink, GstBuffer * buf)
1049 GstFlowReturn ret = GST_FLOW_OK;
1052 render = gst_base_sink_do_sync (basesink, buf);
1055 GstBaseSinkClass *bclass;
1057 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1059 ret = bclass->render (basesink, buf);
1062 GST_DEBUG_OBJECT (basesink, "buffer unref after render %p", basesink, buf);
1063 gst_buffer_unref (buf);
1068 static GstFlowReturn
1069 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
1071 GstBaseSink *basesink;
1072 GstFlowReturn result;
1074 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1076 if (!(basesink->pad_mode == GST_ACTIVATE_PUSH)) {
1078 g_warning ("Push on pad %s:%s, but it was not activated in push mode",
1079 GST_DEBUG_PAD_NAME (pad));
1081 result = GST_FLOW_UNEXPECTED;
1085 result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
1088 gst_object_unref (basesink);
1094 gst_base_sink_loop (GstPad * pad)
1096 GstBaseSink *basesink;
1097 GstBuffer *buf = NULL;
1098 GstFlowReturn result;
1100 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1102 g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
1104 result = gst_pad_pull_range (pad, basesink->offset, DEFAULT_SIZE, &buf);
1105 if (result != GST_FLOW_OK)
1108 result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
1109 if (result != GST_FLOW_OK)
1112 gst_object_unref (basesink);
1119 gst_base_sink_event (pad, gst_event_new_eos ());
1120 gst_object_unref (basesink);
1121 gst_pad_pause_task (pad);
1127 gst_base_sink_deactivate (GstBaseSink * basesink, GstPad * pad)
1129 gboolean result = FALSE;
1130 GstBaseSinkClass *bclass;
1132 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1134 /* step 1, unblock clock sync (if any) or any other blocking thing */
1135 GST_PREROLL_LOCK (pad);
1136 GST_LOCK (basesink);
1137 if (basesink->clock_id) {
1138 gst_clock_id_unschedule (basesink->clock_id);
1140 GST_UNLOCK (basesink);
1142 /* unlock any subclasses */
1144 bclass->unlock (basesink);
1146 /* flush out the data thread if it's locked in finish_preroll */
1147 GST_DEBUG_OBJECT (basesink,
1148 "flushing out data thread, need preroll to FALSE");
1149 basesink->need_preroll = FALSE;
1150 gst_base_sink_preroll_queue_flush (basesink, pad);
1151 GST_PREROLL_SIGNAL (pad);
1152 GST_PREROLL_UNLOCK (pad);
1154 /* step 2, make sure streaming finishes */
1155 result = gst_pad_stop_task (pad);
1161 gst_base_sink_activate (GstPad * pad)
1163 gboolean result = FALSE;
1164 GstBaseSink *basesink;
1166 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1168 GST_DEBUG_OBJECT (basesink, "Trying pull mode first");
1170 if (basesink->can_activate_pull && gst_pad_check_pull_range (pad)
1171 && gst_pad_activate_pull (pad, TRUE)) {
1172 GST_DEBUG_OBJECT (basesink, "Success activating pull mode");
1175 GST_DEBUG_OBJECT (basesink, "Falling back to push mode");
1176 if (gst_pad_activate_push (pad, TRUE)) {
1177 GST_DEBUG_OBJECT (basesink, "Success activating push mode");
1183 GST_WARNING_OBJECT (basesink, "Could not activate pad in either mode");
1186 gst_object_unref (basesink);
1192 gst_base_sink_activate_push (GstPad * pad, gboolean active)
1195 GstBaseSink *basesink;
1197 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1200 if (!basesink->can_activate_push) {
1202 basesink->pad_mode = GST_ACTIVATE_NONE;
1205 basesink->pad_mode = GST_ACTIVATE_PUSH;
1208 if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PUSH)) {
1209 g_warning ("Internal GStreamer activation error!!!");
1212 result = gst_base_sink_deactivate (basesink, pad);
1213 basesink->pad_mode = GST_ACTIVATE_NONE;
1217 gst_object_unref (basesink);
1222 /* this won't get called until we implement an activate function */
1224 gst_base_sink_activate_pull (GstPad * pad, gboolean active)
1226 gboolean result = FALSE;
1227 GstBaseSink *basesink;
1229 basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1232 if (!basesink->can_activate_pull) {
1234 basesink->pad_mode = GST_ACTIVATE_NONE;
1236 GstPad *peer = gst_pad_get_peer (pad);
1238 if (G_UNLIKELY (peer == NULL)) {
1239 g_warning ("Trying to activate pad in pull mode, but no peer");
1241 basesink->pad_mode = GST_ACTIVATE_NONE;
1243 if (gst_pad_activate_pull (peer, TRUE)) {
1244 basesink->have_newsegment = TRUE;
1245 basesink->segment_start = basesink->segment_stop = 0;
1247 /* set the pad mode before starting the task so that it's in the
1248 correct state for the new thread... */
1249 basesink->pad_mode = GST_ACTIVATE_PULL;
1251 gst_pad_start_task (pad, (GstTaskFunction) gst_base_sink_loop,
1253 /* but if starting the thread fails, set it back */
1255 basesink->pad_mode = GST_ACTIVATE_NONE;
1257 GST_DEBUG_OBJECT (pad, "Failed to activate peer in pull mode");
1259 basesink->pad_mode = GST_ACTIVATE_NONE;
1261 gst_object_unref (peer);
1265 if (G_UNLIKELY (basesink->pad_mode != GST_ACTIVATE_PULL)) {
1266 g_warning ("Internal GStreamer activation error!!!");
1269 basesink->have_newsegment = FALSE;
1270 result = gst_base_sink_deactivate (basesink, pad);
1271 basesink->pad_mode = GST_ACTIVATE_NONE;
1275 gst_object_unref (basesink);
1280 static GstStateChangeReturn
1281 gst_base_sink_change_state (GstElement * element, GstStateChange transition)
1283 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1284 GstBaseSink *basesink = GST_BASE_SINK (element);
1285 GstBaseSinkClass *bclass;
1287 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1289 switch (transition) {
1290 case GST_STATE_CHANGE_NULL_TO_READY:
1292 if (!bclass->start (basesink))
1295 case GST_STATE_CHANGE_READY_TO_PAUSED:
1296 /* need to complete preroll before this state change completes, there
1297 * is no data flow in READY so we can safely assume we need to preroll. */
1298 basesink->offset = 0;
1299 GST_PREROLL_LOCK (basesink->sinkpad);
1300 basesink->have_preroll = FALSE;
1301 GST_DEBUG_OBJECT (basesink, "READY to PAUSED, need preroll to FALSE");
1302 basesink->need_preroll = TRUE;
1303 GST_PREROLL_UNLOCK (basesink->sinkpad);
1304 basesink->have_newsegment = FALSE;
1305 basesink->segment_rate = 1.0;
1306 basesink->segment_start = 0;
1307 basesink->segment_stop = 0;
1308 ret = GST_STATE_CHANGE_ASYNC;
1310 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1312 GST_PREROLL_LOCK (basesink->sinkpad);
1313 /* if we have EOS, we should empty the queue now as there will
1314 * be no more data received in the chain function.
1315 * FIXME, this could block the state change function too long when
1316 * we are pushing and syncing the buffers, better start a new
1317 * thread to do this. */
1318 if (basesink->eos) {
1319 gst_base_sink_preroll_queue_empty (basesink, basesink->sinkpad);
1320 } else if (!basesink->have_preroll) {
1321 /* don't need preroll, but do queue a commit_state */
1322 GST_DEBUG_OBJECT (basesink,
1323 "PAUSED to PLAYING, !eos, !have_preroll, need preroll to FALSE");
1324 basesink->need_preroll = FALSE;
1325 basesink->playing_async = TRUE;
1326 ret = GST_STATE_CHANGE_ASYNC;
1327 /* we know it's not waiting, no need to signal */
1329 /* don't need the preroll anymore */
1330 basesink->need_preroll = FALSE;
1331 GST_DEBUG_OBJECT (basesink,
1332 "PAUSED to PLAYING, !eos, have_preroll, need preroll to FALSE");
1333 /* now let it play */
1334 GST_PREROLL_SIGNAL (basesink->sinkpad);
1336 GST_PREROLL_UNLOCK (basesink->sinkpad);
1344 GstStateChangeReturn bret;
1346 bret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1348 if (bret != GST_STATE_CHANGE_SUCCESS)
1349 goto activate_failed;
1352 switch (transition) {
1353 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1355 GstBaseSinkClass *bclass;
1357 bclass = GST_BASE_SINK_GET_CLASS (basesink);
1359 GST_PREROLL_LOCK (basesink->sinkpad);
1360 GST_LOCK (basesink);
1361 /* unlock clock wait if any */
1362 if (basesink->clock_id) {
1363 gst_clock_id_unschedule (basesink->clock_id);
1365 GST_UNLOCK (basesink);
1367 basesink->playing_async = FALSE;
1369 /* unlock any subclasses */
1371 bclass->unlock (basesink);
1373 /* if we don't have a preroll buffer and we have not received EOS,
1374 * we need to wait for a preroll */
1375 GST_DEBUG_OBJECT (basesink, "have_preroll: %d, EOS: %d",
1376 basesink->have_preroll, basesink->eos);
1377 if (!basesink->have_preroll && !basesink->eos) {
1378 GST_DEBUG_OBJECT (basesink, "PLAYING to PAUSED, need preroll to TRUE");
1379 basesink->need_preroll = TRUE;
1380 ret = GST_STATE_CHANGE_ASYNC;
1382 GST_PREROLL_UNLOCK (basesink->sinkpad);
1385 case GST_STATE_CHANGE_PAUSED_TO_READY:
1387 case GST_STATE_CHANGE_READY_TO_NULL:
1389 if (!bclass->stop (basesink)) {
1390 GST_WARNING ("failed to stop");
1402 GST_DEBUG_OBJECT (basesink, "failed to start");
1403 return GST_STATE_CHANGE_FAILURE;
1407 GST_DEBUG_OBJECT (basesink,
1408 "element failed to change states -- activation problem?");
1409 return GST_STATE_CHANGE_FAILURE;