1 /* GStreamer
2 * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
3 * 2003 Colin Walters <cwalters@gnome.org>
4 * 2000,2005,2007 Wim Taymans <wim.taymans@gmail.com>
5 * 2007 Thiago Sousa Santos <thiagoss@lcc.ufcg.edu.br>
6 *
7 * gstqueue2.c:
8 *
9 * This library is free software; you can redistribute it and/or
10 * modify it under the terms of the GNU Library General Public
11 * License as published by the Free Software Foundation; either
12 * version 2 of the License, or (at your option) any later version.
13 *
14 * This library is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * Library General Public License for more details.
18 *
19 * You should have received a copy of the GNU Library General Public
20 * License along with this library; if not, write to the
21 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
22 * Boston, MA 02111-1307, USA.
23 */
25 /**
26 * SECTION:element-queue2
27 *
28 * Data is queued until one of the limits specified by the
29 * #GstQueue2:max-size-buffers, #GstQueue2:max-size-bytes and/or
30 * #GstQueue2:max-size-time properties has been reached. Any attempt to push
31 * more buffers into the queue will block the pushing thread until more space
32 * becomes available.
33 *
34 * The queue will create a new thread on the source pad to decouple the
35 * processing on sink and source pad.
36 *
37 * You can query how many buffers are queued by reading the
38 * #GstQueue2:current-level-buffers property.
39 *
40 * The default queue size limits are 100 buffers, 2MB of data, or
41 * two seconds worth of data, whichever is reached first.
42 *
43 * If you set temp-tmpl to a value such as /tmp/gstreamer-XXXXXX, the element
44 * will allocate a random free filename and buffer data in the file.
45 * By using this, it will buffer the entire stream data on the file independently
46 * of the queue size limits, they will only be used for buffering statistics.
47 *
48 * Since 0.10.24, setting the temp-location property with a filename is deprecated
49 * because it's impossible to securely open a temporary file in this way. The
50 * property will still be used to notify the application of the allocated
51 * filename, though.
52 *
53 * Last reviewed on 2009-07-10 (0.10.24)
54 */
56 #ifdef HAVE_CONFIG_H
57 #include "config.h"
58 #endif
60 #include "gstqueue2.h"
62 #include <glib/gstdio.h>
64 #include "gst/gst-i18n-lib.h"
66 #ifdef G_OS_WIN32
67 #include <io.h> /* lseek, open, close, read */
68 #undef lseek
69 #define lseek _lseeki64
70 #undef off_t
71 #define off_t guint64
72 #else
73 #include <unistd.h>
74 #endif
76 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
77 GST_PAD_SINK,
78 GST_PAD_ALWAYS,
79 GST_STATIC_CAPS_ANY);
81 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
82 GST_PAD_SRC,
83 GST_PAD_ALWAYS,
84 GST_STATIC_CAPS_ANY);
86 GST_DEBUG_CATEGORY_STATIC (queue_debug);
87 #define GST_CAT_DEFAULT (queue_debug)
88 GST_DEBUG_CATEGORY_STATIC (queue_dataflow);
90 enum
91 {
92 LAST_SIGNAL
93 };
95 /* default property values */
96 #define DEFAULT_MAX_SIZE_BUFFERS 100 /* 100 buffers */
97 #define DEFAULT_MAX_SIZE_BYTES (2 * 1024 * 1024) /* 2 MB */
98 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND /* 2 seconds */
99 #define DEFAULT_USE_BUFFERING FALSE
100 #define DEFAULT_USE_RATE_ESTIMATE TRUE
101 #define DEFAULT_LOW_PERCENT 10
102 #define DEFAULT_HIGH_PERCENT 99
103 #define DEFAULT_TEMP_REMOVE TRUE
105 /* other defines */
106 #define DEFAULT_BUFFER_SIZE 4096
107 #define QUEUE_IS_USING_TEMP_FILE(queue) ((queue)->temp_location_set || (queue)->temp_template != NULL)
109 enum
110 {
111 PROP_0,
112 PROP_CUR_LEVEL_BUFFERS,
113 PROP_CUR_LEVEL_BYTES,
114 PROP_CUR_LEVEL_TIME,
115 PROP_MAX_SIZE_BUFFERS,
116 PROP_MAX_SIZE_BYTES,
117 PROP_MAX_SIZE_TIME,
118 PROP_USE_BUFFERING,
119 PROP_USE_RATE_ESTIMATE,
120 PROP_LOW_PERCENT,
121 PROP_HIGH_PERCENT,
122 PROP_TEMP_TEMPLATE,
123 PROP_TEMP_LOCATION,
124 PROP_TEMP_REMOVE,
125 PROP_LAST
126 };
128 #define GST_QUEUE2_CLEAR_LEVEL(l) G_STMT_START { \
129 l.buffers = 0; \
130 l.bytes = 0; \
131 l.time = 0; \
132 l.rate_time = 0; \
133 } G_STMT_END
135 #define STATUS(queue, pad, msg) \
136 GST_CAT_LOG_OBJECT (queue_dataflow, queue, \
137 "(%s:%s) " msg ": %u of %u buffers, %u of %u " \
138 "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
139 " ns, %"G_GUINT64_FORMAT" items", \
140 GST_DEBUG_PAD_NAME (pad), \
141 queue->cur_level.buffers, \
142 queue->max_level.buffers, \
143 queue->cur_level.bytes, \
144 queue->max_level.bytes, \
145 queue->cur_level.time, \
146 queue->max_level.time, \
147 (guint64) (QUEUE_IS_USING_TEMP_FILE(queue) ? \
148 queue->current->writing_pos - queue->current->max_reading_pos : \
149 queue->queue->length))
151 #define GST_QUEUE2_MUTEX_LOCK(q) G_STMT_START { \
152 g_mutex_lock (q->qlock); \
153 } G_STMT_END
155 #define GST_QUEUE2_MUTEX_LOCK_CHECK(q,res,label) G_STMT_START { \
156 GST_QUEUE2_MUTEX_LOCK (q); \
157 if (res != GST_FLOW_OK) \
158 goto label; \
159 } G_STMT_END
161 #define GST_QUEUE2_MUTEX_UNLOCK(q) G_STMT_START { \
162 g_mutex_unlock (q->qlock); \
163 } G_STMT_END
165 #define GST_QUEUE2_WAIT_DEL_CHECK(q, res, label) G_STMT_START { \
166 STATUS (queue, q->sinkpad, "wait for DEL"); \
167 q->waiting_del = TRUE; \
168 g_cond_wait (q->item_del, queue->qlock); \
169 q->waiting_del = FALSE; \
170 if (res != GST_FLOW_OK) { \
171 STATUS (queue, q->srcpad, "received DEL wakeup"); \
172 goto label; \
173 } \
174 STATUS (queue, q->sinkpad, "received DEL"); \
175 } G_STMT_END
177 #define GST_QUEUE2_WAIT_ADD_CHECK(q, res, label) G_STMT_START { \
178 STATUS (queue, q->srcpad, "wait for ADD"); \
179 q->waiting_add = TRUE; \
180 g_cond_wait (q->item_add, q->qlock); \
181 q->waiting_add = FALSE; \
182 if (res != GST_FLOW_OK) { \
183 STATUS (queue, q->srcpad, "received ADD wakeup"); \
184 goto label; \
185 } \
186 STATUS (queue, q->srcpad, "received ADD"); \
187 } G_STMT_END
189 #define GST_QUEUE2_SIGNAL_DEL(q) G_STMT_START { \
190 if (q->waiting_del) { \
191 STATUS (q, q->srcpad, "signal DEL"); \
192 g_cond_signal (q->item_del); \
193 } \
194 } G_STMT_END
196 #define GST_QUEUE2_SIGNAL_ADD(q) G_STMT_START { \
197 if (q->waiting_add) { \
198 STATUS (q, q->sinkpad, "signal ADD"); \
199 g_cond_signal (q->item_add); \
200 } \
201 } G_STMT_END
203 #define _do_init(bla) \
204 GST_DEBUG_CATEGORY_INIT (queue_debug, "queue2", 0, "queue element"); \
205 GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue2_dataflow", 0, \
206 "dataflow inside the queue element");
208 GST_BOILERPLATE_FULL (GstQueue2, gst_queue2, GstElement, GST_TYPE_ELEMENT,
209 _do_init);
211 static void gst_queue2_finalize (GObject * object);
213 static void gst_queue2_set_property (GObject * object,
214 guint prop_id, const GValue * value, GParamSpec * pspec);
215 static void gst_queue2_get_property (GObject * object,
216 guint prop_id, GValue * value, GParamSpec * pspec);
218 static GstFlowReturn gst_queue2_chain (GstPad * pad, GstBuffer * buffer);
219 static GstFlowReturn gst_queue2_bufferalloc (GstPad * pad, guint64 offset,
220 guint size, GstCaps * caps, GstBuffer ** buf);
221 static GstFlowReturn gst_queue2_push_one (GstQueue2 * queue);
222 static void gst_queue2_loop (GstPad * pad);
224 static gboolean gst_queue2_handle_sink_event (GstPad * pad, GstEvent * event);
226 static gboolean gst_queue2_handle_src_event (GstPad * pad, GstEvent * event);
227 static gboolean gst_queue2_handle_src_query (GstPad * pad, GstQuery * query);
229 static GstCaps *gst_queue2_getcaps (GstPad * pad);
230 static gboolean gst_queue2_acceptcaps (GstPad * pad, GstCaps * caps);
232 static GstFlowReturn gst_queue2_get_range (GstPad * pad, guint64 offset,
233 guint length, GstBuffer ** buffer);
234 static gboolean gst_queue2_src_checkgetrange_function (GstPad * pad);
236 static gboolean gst_queue2_src_activate_pull (GstPad * pad, gboolean active);
237 static gboolean gst_queue2_src_activate_push (GstPad * pad, gboolean active);
238 static gboolean gst_queue2_sink_activate_push (GstPad * pad, gboolean active);
239 static GstStateChangeReturn gst_queue2_change_state (GstElement * element,
240 GstStateChange transition);
242 static gboolean gst_queue2_is_empty (GstQueue2 * queue);
243 static gboolean gst_queue2_is_filled (GstQueue2 * queue);
245 /* static guint gst_queue2_signals[LAST_SIGNAL] = { 0 }; */
247 static void
248 gst_queue2_base_init (gpointer g_class)
249 {
250 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
252 gst_element_class_add_pad_template (gstelement_class,
253 gst_static_pad_template_get (&srctemplate));
254 gst_element_class_add_pad_template (gstelement_class,
255 gst_static_pad_template_get (&sinktemplate));
257 gst_element_class_set_details_simple (gstelement_class, "Queue 2",
258 "Generic",
259 "Simple data queue",
260 "Erik Walthinsen <omega@cse.ogi.edu>, "
261 "Wim Taymans <wim.taymans@gmail.com>");
262 }
264 static void
265 gst_queue2_class_init (GstQueue2Class * klass)
266 {
267 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
268 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
270 parent_class = g_type_class_peek_parent (klass);
272 gobject_class->set_property = gst_queue2_set_property;
273 gobject_class->get_property = gst_queue2_get_property;
275 /* properties */
276 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,
277 g_param_spec_uint ("current-level-bytes", "Current level (kB)",
278 "Current amount of data in the queue (bytes)",
279 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
280 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BUFFERS,
281 g_param_spec_uint ("current-level-buffers", "Current level (buffers)",
282 "Current number of buffers in the queue",
283 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
284 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,
285 g_param_spec_uint64 ("current-level-time", "Current level (ns)",
286 "Current amount of data in the queue (in ns)",
287 0, G_MAXUINT64, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
289 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
290 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
291 "Max. amount of data in the queue (bytes, 0=disable)",
292 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
293 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
294 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BUFFERS,
295 g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
296 "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
297 DEFAULT_MAX_SIZE_BUFFERS,
298 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
299 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
300 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
301 "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
302 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
304 g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,
305 g_param_spec_boolean ("use-buffering", "Use buffering",
306 "Emit GST_MESSAGE_BUFFERING based on low-/high-percent thresholds",
307 DEFAULT_USE_BUFFERING, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
308 g_object_class_install_property (gobject_class, PROP_USE_RATE_ESTIMATE,
309 g_param_spec_boolean ("use-rate-estimate", "Use Rate Estimate",
310 "Estimate the bitrate of the stream to calculate time level",
311 DEFAULT_USE_RATE_ESTIMATE,
312 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
313 g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
314 g_param_spec_int ("low-percent", "Low percent",
315 "Low threshold for buffering to start", 0, 100, DEFAULT_LOW_PERCENT,
316 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
317 g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
318 g_param_spec_int ("high-percent", "High percent",
319 "High threshold for buffering to finish", 0, 100,
320 DEFAULT_HIGH_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
322 g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
323 g_param_spec_string ("temp-template", "Temporary File Template",
324 "File template to store temporary files in, should contain directory "
325 "and XXXXXX. (NULL == disabled)",
326 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
328 g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
329 g_param_spec_string ("temp-location", "Temporary File Location",
330 "Location to store temporary files in (Deprecated: Only read this "
331 "property, use temp-template to configure the name template)",
332 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
334 /**
335 * GstQueue2:temp-remove
336 *
337 * When temp-template is set, remove the temporary file when going to READY.
338 *
339 * Since: 0.10.26
340 */
341 g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
342 g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
343 "Remove the temp-location after use",
344 DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
346 /* set several parent class virtual functions */
347 gobject_class->finalize = gst_queue2_finalize;
349 gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_queue2_change_state);
350 }
352 static void
353 gst_queue2_init (GstQueue2 * queue, GstQueue2Class * g_class)
354 {
355 queue->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
357 gst_pad_set_chain_function (queue->sinkpad,
358 GST_DEBUG_FUNCPTR (gst_queue2_chain));
359 gst_pad_set_activatepush_function (queue->sinkpad,
360 GST_DEBUG_FUNCPTR (gst_queue2_sink_activate_push));
361 gst_pad_set_event_function (queue->sinkpad,
362 GST_DEBUG_FUNCPTR (gst_queue2_handle_sink_event));
363 gst_pad_set_getcaps_function (queue->sinkpad,
364 GST_DEBUG_FUNCPTR (gst_queue2_getcaps));
365 gst_pad_set_acceptcaps_function (queue->sinkpad,
366 GST_DEBUG_FUNCPTR (gst_queue2_acceptcaps));
367 gst_pad_set_bufferalloc_function (queue->sinkpad,
368 GST_DEBUG_FUNCPTR (gst_queue2_bufferalloc));
369 gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
371 queue->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
373 gst_pad_set_activatepull_function (queue->srcpad,
374 GST_DEBUG_FUNCPTR (gst_queue2_src_activate_pull));
375 gst_pad_set_activatepush_function (queue->srcpad,
376 GST_DEBUG_FUNCPTR (gst_queue2_src_activate_push));
377 gst_pad_set_getrange_function (queue->srcpad,
378 GST_DEBUG_FUNCPTR (gst_queue2_get_range));
379 gst_pad_set_checkgetrange_function (queue->srcpad,
380 GST_DEBUG_FUNCPTR (gst_queue2_src_checkgetrange_function));
381 gst_pad_set_getcaps_function (queue->srcpad,
382 GST_DEBUG_FUNCPTR (gst_queue2_getcaps));
383 gst_pad_set_acceptcaps_function (queue->srcpad,
384 GST_DEBUG_FUNCPTR (gst_queue2_acceptcaps));
385 gst_pad_set_event_function (queue->srcpad,
386 GST_DEBUG_FUNCPTR (gst_queue2_handle_src_event));
387 gst_pad_set_query_function (queue->srcpad,
388 GST_DEBUG_FUNCPTR (gst_queue2_handle_src_query));
389 gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
391 /* levels */
392 GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
393 queue->max_level.buffers = DEFAULT_MAX_SIZE_BUFFERS;
394 queue->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;
395 queue->max_level.time = DEFAULT_MAX_SIZE_TIME;
396 queue->max_level.rate_time = DEFAULT_MAX_SIZE_TIME;
397 queue->use_buffering = DEFAULT_USE_BUFFERING;
398 queue->use_rate_estimate = DEFAULT_USE_RATE_ESTIMATE;
399 queue->low_percent = DEFAULT_LOW_PERCENT;
400 queue->high_percent = DEFAULT_HIGH_PERCENT;
402 gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
403 gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
405 queue->srcresult = GST_FLOW_WRONG_STATE;
406 queue->sinkresult = GST_FLOW_WRONG_STATE;
407 queue->is_eos = FALSE;
408 queue->in_timer = g_timer_new ();
409 queue->out_timer = g_timer_new ();
411 queue->qlock = g_mutex_new ();
412 queue->waiting_add = FALSE;
413 queue->item_add = g_cond_new ();
414 queue->waiting_del = FALSE;
415 queue->item_del = g_cond_new ();
416 queue->queue = g_queue_new ();
418 /* tempfile related */
419 queue->temp_template = NULL;
420 queue->temp_location = NULL;
421 queue->temp_location_set = FALSE;
422 queue->temp_remove = DEFAULT_TEMP_REMOVE;
424 GST_DEBUG_OBJECT (queue,
425 "initialized queue's not_empty & not_full conditions");
426 }
428 /* called only once, as opposed to dispose */
429 static void
430 gst_queue2_finalize (GObject * object)
431 {
432 GstQueue2 *queue = GST_QUEUE2 (object);
434 GST_DEBUG_OBJECT (queue, "finalizing queue");
436 while (!g_queue_is_empty (queue->queue)) {
437 GstMiniObject *data = g_queue_pop_head (queue->queue);
439 gst_mini_object_unref (data);
440 }
442 g_queue_free (queue->queue);
443 g_mutex_free (queue->qlock);
444 g_cond_free (queue->item_add);
445 g_cond_free (queue->item_del);
446 g_timer_destroy (queue->in_timer);
447 g_timer_destroy (queue->out_timer);
449 /* temp_file path cleanup */
450 g_free (queue->temp_template);
451 g_free (queue->temp_location);
453 G_OBJECT_CLASS (parent_class)->finalize (object);
454 }
456 static void
457 debug_ranges (GstQueue2 * queue)
458 {
459 GstQueue2Range *walk;
461 for (walk = queue->ranges; walk; walk = walk->next) {
462 GST_DEBUG_OBJECT (queue, "range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT,
463 walk->offset, walk->writing_pos);
464 }
465 }
467 /* clear all the downloaded ranges */
468 static void
469 clean_ranges (GstQueue2 * queue)
470 {
471 GST_DEBUG_OBJECT (queue, "clean queue ranges");
473 g_slice_free_chain (GstQueue2Range, queue->ranges, next);
474 queue->ranges = NULL;
475 queue->current = NULL;
476 }
478 /* find a range that contains @offset or NULL when nothing does */
479 static GstQueue2Range *
480 find_range (GstQueue2 * queue, guint64 offset, guint64 length)
481 {
482 GstQueue2Range *range = NULL;
483 GstQueue2Range *walk;
485 /* first do a quick check for the current range */
486 for (walk = queue->ranges; walk; walk = walk->next) {
487 if (offset >= walk->offset && offset <= walk->writing_pos) {
488 /* we can reuse an existing range */
489 range = walk;
490 break;
491 }
492 }
493 return range;
494 }
496 /* make a new range for @offset or reuse an existing range */
497 static GstQueue2Range *
498 add_range (GstQueue2 * queue, guint64 offset)
499 {
500 GstQueue2Range *range, *prev, *next;
502 GST_DEBUG_OBJECT (queue, "find range for %" G_GUINT64_FORMAT, offset);
504 if ((range = find_range (queue, offset, 0))) {
505 GST_DEBUG_OBJECT (queue,
506 "reusing range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, range->offset,
507 range->writing_pos);
508 range->writing_pos = offset;
509 } else {
510 GST_DEBUG_OBJECT (queue,
511 "new range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, offset, offset);
513 range = g_slice_new0 (GstQueue2Range);
514 range->offset = offset;
515 range->writing_pos = offset;
516 range->reading_pos = offset;
517 range->max_reading_pos = offset;
519 /* insert sorted */
520 prev = NULL;
521 next = queue->ranges;
522 while (next) {
523 if (next->offset > offset) {
524 /* insert before next */
525 GST_DEBUG_OBJECT (queue,
526 "insert before range %p, offset %" G_GUINT64_FORMAT, next,
527 next->offset);
528 break;
529 }
530 /* try next */
531 prev = next;
532 next = next->next;
533 }
534 range->next = next;
535 if (prev)
536 prev->next = range;
537 else
538 queue->ranges = range;
539 }
540 debug_ranges (queue);
542 return range;
543 }
546 /* clear and init the download ranges for offset 0 */
547 static void
548 init_ranges (GstQueue2 * queue)
549 {
550 GST_DEBUG_OBJECT (queue, "init queue ranges");
552 /* get rid of all the current ranges */
553 clean_ranges (queue);
554 /* make a range for offset 0 */
555 queue->current = add_range (queue, 0);
556 }
558 static gboolean
559 gst_queue2_acceptcaps (GstPad * pad, GstCaps * caps)
560 {
561 GstQueue2 *queue;
562 GstPad *otherpad;
563 gboolean result;
565 queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
567 otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
568 result = gst_pad_peer_accept_caps (otherpad, caps);
570 return result;
571 }
573 static GstCaps *
574 gst_queue2_getcaps (GstPad * pad)
575 {
576 GstQueue2 *queue;
577 GstPad *otherpad;
578 GstCaps *result;
580 queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
582 otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
583 result = gst_pad_peer_get_caps (otherpad);
584 if (result == NULL)
585 result = gst_caps_new_any ();
587 return result;
588 }
590 static GstFlowReturn
591 gst_queue2_bufferalloc (GstPad * pad, guint64 offset, guint size,
592 GstCaps * caps, GstBuffer ** buf)
593 {
594 GstQueue2 *queue;
595 GstFlowReturn result;
597 queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
599 /* Forward to src pad, without setting caps on the src pad */
600 result = gst_pad_alloc_buffer (queue->srcpad, offset, size, caps, buf);
602 return result;
603 }
605 /* calculate the diff between running time on the sink and src of the queue.
606 * This is the total amount of time in the queue. */
607 static void
608 update_time_level (GstQueue2 * queue)
609 {
610 gint64 sink_time, src_time;
612 sink_time =
613 gst_segment_to_running_time (&queue->sink_segment, GST_FORMAT_TIME,
614 queue->sink_segment.last_stop);
616 src_time = gst_segment_to_running_time (&queue->src_segment, GST_FORMAT_TIME,
617 queue->src_segment.last_stop);
619 GST_DEBUG_OBJECT (queue, "sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT,
620 GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
622 if (sink_time >= src_time)
623 queue->cur_level.time = sink_time - src_time;
624 else
625 queue->cur_level.time = 0;
626 }
628 /* take a NEWSEGMENT event and apply the values to segment, updating the time
629 * level of queue. */
630 static void
631 apply_segment (GstQueue2 * queue, GstEvent * event, GstSegment * segment)
632 {
633 gboolean update;
634 GstFormat format;
635 gdouble rate, arate;
636 gint64 start, stop, time;
638 gst_event_parse_new_segment_full (event, &update, &rate, &arate,
639 &format, &start, &stop, &time);
641 GST_DEBUG_OBJECT (queue,
642 "received NEWSEGMENT update %d, rate %lf, applied rate %lf, "
643 "format %d, "
644 "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
645 G_GINT64_FORMAT, update, rate, arate, format, start, stop, time);
647 if (format == GST_FORMAT_BYTES) {
648 }
650 /* now configure the values, we use these to track timestamps on the
651 * sinkpad. */
652 if (format != GST_FORMAT_TIME) {
653 /* non-time format, pretent the current time segment is closed with a
654 * 0 start and unknown stop time. */
655 update = FALSE;
656 format = GST_FORMAT_TIME;
657 start = 0;
658 stop = -1;
659 time = 0;
660 }
661 gst_segment_set_newsegment_full (segment, update,
662 rate, arate, format, start, stop, time);
664 GST_DEBUG_OBJECT (queue,
665 "configured NEWSEGMENT %" GST_SEGMENT_FORMAT, segment);
667 /* segment can update the time level of the queue */
668 update_time_level (queue);
669 }
671 /* take a buffer and update segment, updating the time level of the queue. */
672 static void
673 apply_buffer (GstQueue2 * queue, GstBuffer * buffer, GstSegment * segment)
674 {
675 GstClockTime duration, timestamp;
677 timestamp = GST_BUFFER_TIMESTAMP (buffer);
678 duration = GST_BUFFER_DURATION (buffer);
680 /* if no timestamp is set, assume it's continuous with the previous
681 * time */
682 if (timestamp == GST_CLOCK_TIME_NONE)
683 timestamp = segment->last_stop;
685 /* add duration */
686 if (duration != GST_CLOCK_TIME_NONE)
687 timestamp += duration;
689 GST_DEBUG_OBJECT (queue, "last_stop updated to %" GST_TIME_FORMAT,
690 GST_TIME_ARGS (timestamp));
692 gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
694 /* calc diff with other end */
695 update_time_level (queue);
696 }
698 static void
699 update_buffering (GstQueue2 * queue)
700 {
701 gint64 percent;
702 gboolean post = FALSE;
704 if (!queue->use_buffering || queue->high_percent <= 0)
705 return;
707 #define GET_PERCENT(format) ((queue->max_level.format) > 0 ? \
708 (queue->cur_level.format) * 100 / (queue->max_level.format) : 0)
710 if (queue->is_eos) {
711 /* on EOS we are always 100% full, we set the var here so that it we can
712 * reuse the logic below to stop buffering */
713 percent = 100;
714 } else {
715 /* figure out the percent we are filled, we take the max of all formats. */
716 percent = GET_PERCENT (bytes);
717 percent = MAX (percent, GET_PERCENT (time));
718 percent = MAX (percent, GET_PERCENT (buffers));
720 /* also apply the rate estimate when we need to */
721 if (queue->use_rate_estimate)
722 percent = MAX (percent, GET_PERCENT (rate_time));
723 }
725 if (queue->is_buffering) {
726 post = TRUE;
727 /* if we were buffering see if we reached the high watermark */
728 if (percent >= queue->high_percent)
729 queue->is_buffering = FALSE;
730 } else {
731 /* we were not buffering, check if we need to start buffering if we drop
732 * below the low threshold */
733 if (percent < queue->low_percent) {
734 queue->is_buffering = TRUE;
735 queue->buffering_iteration++;
736 post = TRUE;
737 }
738 }
739 if (post) {
740 GstMessage *message;
741 GstBufferingMode mode;
742 gint64 buffering_left = -1;
744 /* scale to high percent so that it becomes the 100% mark */
745 percent = percent * 100 / queue->high_percent;
746 /* clip */
747 if (percent > 100)
748 percent = 100;
750 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
751 GstFormat fmt = GST_FORMAT_BYTES;
752 gint64 duration;
754 mode = GST_BUFFERING_DOWNLOAD;
755 if (queue->byte_in_rate > 0) {
756 if (gst_pad_query_peer_duration (queue->sinkpad, &fmt, &duration))
757 buffering_left =
758 (gdouble) ((duration -
759 queue->current->writing_pos) * 1000) / queue->byte_in_rate;
760 } else {
761 buffering_left = G_MAXINT64;
762 }
763 } else {
764 mode = GST_BUFFERING_STREAM;
765 }
767 GST_DEBUG_OBJECT (queue, "buffering %d percent", (gint) percent);
768 message = gst_message_new_buffering (GST_OBJECT_CAST (queue),
769 (gint) percent);
770 gst_message_set_buffering_stats (message, mode,
771 queue->byte_in_rate, queue->byte_out_rate, buffering_left);
773 gst_element_post_message (GST_ELEMENT_CAST (queue), message);
775 } else {
776 GST_DEBUG_OBJECT (queue, "filled %d percent", (gint) percent);
777 }
779 #undef GET_PERCENT
780 }
782 static void
783 reset_rate_timer (GstQueue2 * queue)
784 {
785 queue->bytes_in = 0;
786 queue->bytes_out = 0;
787 queue->byte_in_rate = 0.0;
788 queue->byte_out_rate = 0.0;
789 queue->last_in_elapsed = 0.0;
790 queue->last_out_elapsed = 0.0;
791 queue->in_timer_started = FALSE;
792 queue->out_timer_started = FALSE;
793 }
795 /* the interval in seconds to recalculate the rate */
796 #define RATE_INTERVAL 0.2
797 /* Tuning for rate estimation. We use a large window for the input rate because
798 * it should be stable when connected to a network. The output rate is less
799 * stable (the elements preroll, queues behind a demuxer fill, ...) and should
800 * therefore adapt more quickly. */
801 #define AVG_IN(avg,val) ((avg) * 15.0 + (val)) / 16.0
802 #define AVG_OUT(avg,val) ((avg) * 3.0 + (val)) / 4.0
804 static void
805 update_in_rates (GstQueue2 * queue)
806 {
807 gdouble elapsed, period;
808 gdouble byte_in_rate;
810 if (!queue->in_timer_started) {
811 queue->in_timer_started = TRUE;
812 g_timer_start (queue->in_timer);
813 return;
814 }
816 elapsed = g_timer_elapsed (queue->in_timer, NULL);
818 /* recalc after each interval. */
819 if (queue->last_in_elapsed + RATE_INTERVAL < elapsed) {
820 period = elapsed - queue->last_in_elapsed;
822 GST_DEBUG_OBJECT (queue,
823 "rates: period %f, in %" G_GUINT64_FORMAT, period, queue->bytes_in);
825 byte_in_rate = queue->bytes_in / period;
827 if (queue->byte_in_rate == 0.0)
828 queue->byte_in_rate = byte_in_rate;
829 else
830 queue->byte_in_rate = AVG_IN (queue->byte_in_rate, byte_in_rate);
832 /* reset the values to calculate rate over the next interval */
833 queue->last_in_elapsed = elapsed;
834 queue->bytes_in = 0;
835 }
837 if (queue->byte_in_rate > 0.0) {
838 queue->cur_level.rate_time =
839 queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
840 }
841 GST_DEBUG_OBJECT (queue, "rates: in %f, time %" GST_TIME_FORMAT,
842 queue->byte_in_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
843 }
845 static void
846 update_out_rates (GstQueue2 * queue)
847 {
848 gdouble elapsed, period;
849 gdouble byte_out_rate;
851 if (!queue->out_timer_started) {
852 queue->out_timer_started = TRUE;
853 g_timer_start (queue->out_timer);
854 return;
855 }
857 elapsed = g_timer_elapsed (queue->out_timer, NULL);
859 /* recalc after each interval. */
860 if (queue->last_out_elapsed + RATE_INTERVAL < elapsed) {
861 period = elapsed - queue->last_out_elapsed;
863 GST_DEBUG_OBJECT (queue,
864 "rates: period %f, out %" G_GUINT64_FORMAT, period, queue->bytes_out);
866 byte_out_rate = queue->bytes_out / period;
868 if (queue->byte_out_rate == 0.0)
869 queue->byte_out_rate = byte_out_rate;
870 else
871 queue->byte_out_rate = AVG_OUT (queue->byte_out_rate, byte_out_rate);
873 /* reset the values to calculate rate over the next interval */
874 queue->last_out_elapsed = elapsed;
875 queue->bytes_out = 0;
876 }
877 if (queue->byte_in_rate > 0.0) {
878 queue->cur_level.rate_time =
879 queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
880 }
881 GST_DEBUG_OBJECT (queue, "rates: out %f, time %" GST_TIME_FORMAT,
882 queue->byte_out_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
883 }
885 static void
886 gst_queue2_write_buffer_to_file (GstQueue2 * queue, GstBuffer * buffer)
887 {
888 guint size;
889 guint8 *data;
890 int ret;
891 guint64 writing_pos, max_reading_pos;
892 GstQueue2Range *next;
894 writing_pos = queue->current->writing_pos;
895 max_reading_pos = queue->current->max_reading_pos;
897 #ifdef HAVE_FSEEKO
898 fseeko (queue->temp_file, (off_t) writing_pos, SEEK_SET);
899 #elif defined (G_OS_UNIX) || defined (G_OS_WIN32)
900 lseek (fileno (queue->temp_file), (off_t) writing_pos, SEEK_SET);
901 #else
902 fseek (queue->temp_file, writing_pos, SEEK_SET);
903 #endif
905 data = GST_BUFFER_DATA (buffer);
906 size = GST_BUFFER_SIZE (buffer);
908 ret = fwrite (data, 1, size, queue->temp_file);
909 if (ret < size) {
910 /* FIXME do something useful here */
911 GST_ERROR_OBJECT (queue, "fwrite returned error");
912 }
913 writing_pos += size;
915 if (writing_pos > max_reading_pos)
916 queue->cur_level.bytes = writing_pos - max_reading_pos;
917 else
918 queue->cur_level.bytes = 0;
920 /* try to merge with next range */
921 while ((next = queue->current->next)) {
922 GST_DEBUG_OBJECT (queue,
923 "cheking %" G_GUINT64_FORMAT " < %" G_GUINT64_FORMAT, writing_pos,
924 next->offset);
925 if (writing_pos < next->offset)
926 break;
928 GST_DEBUG_OBJECT (queue, "merging ranges %" G_GUINT64_FORMAT,
929 next->writing_pos);
930 /* we ran over the offset of the next group */
931 queue->current->writing_pos = writing_pos = next->writing_pos;
933 /* remove the group */
934 queue->current->next = next->next;
935 g_slice_free (GstQueue2Range, next);
937 debug_ranges (queue);
938 }
939 queue->current->writing_pos = writing_pos;
940 }
942 static gboolean
943 perform_seek_to_offset (GstQueue2 * queue, guint64 offset)
944 {
945 GstEvent *event;
946 gboolean res;
948 GST_DEBUG_OBJECT (queue, "Seeking to %" G_GUINT64_FORMAT, offset);
950 event =
951 gst_event_new_seek (1.0, GST_FORMAT_BYTES,
952 GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
953 GST_SEEK_TYPE_NONE, -1);
955 GST_QUEUE2_MUTEX_UNLOCK (queue);
956 res = gst_pad_push_event (queue->sinkpad, event);
957 GST_QUEUE2_MUTEX_LOCK (queue);
959 if (res) {
960 queue->current = add_range (queue, offset);
961 }
962 return res;
963 }
965 /* see if there is enough data in the file to read a full buffer */
966 static gboolean
967 gst_queue2_have_data (GstQueue2 * queue, guint64 offset, guint length)
968 {
969 GstQueue2Range *range;
971 GST_DEBUG_OBJECT (queue, "offset %" G_GUINT64_FORMAT ", len %u", offset,
972 length);
974 if ((range = find_range (queue, offset, length))) {
975 if (queue->current != range) {
976 GST_DEBUG_OBJECT (queue, "switching ranges");
977 perform_seek_to_offset (queue, range->writing_pos);
978 }
979 /* we have a range for offset */
980 GST_DEBUG_OBJECT (queue,
981 "we have a range %p, offset %" G_GUINT64_FORMAT ", writing_pos %"
982 G_GUINT64_FORMAT, range, range->offset, range->writing_pos);
984 if (queue->is_eos)
985 return TRUE;
987 if (offset + length < range->writing_pos)
988 return TRUE;
990 } else {
991 /* we don't have the range, see how far away we are, FIXME, find a good
992 * threshold based on the incomming rate. */
993 if (queue->current && offset < queue->current->writing_pos + 200000)
994 return FALSE;
996 /* too far away, do a seek */
997 perform_seek_to_offset (queue, offset);
998 }
1000 return FALSE;
1001 }
1003 static GstFlowReturn
1004 gst_queue2_create_read (GstQueue2 * queue, guint64 offset, guint length,
1005 GstBuffer ** buffer)
1006 {
1007 size_t res;
1008 GstBuffer *buf;
1009 guint64 reading_pos, max_reading_pos, writing_pos;
1011 /* check if we have enough data at @offset. If there is not enough data, we
1012 * block and wait. */
1013 while (!gst_queue2_have_data (queue, offset, length)) {
1014 GST_QUEUE2_WAIT_ADD_CHECK (queue, queue->srcresult, out_flushing);
1015 }
1017 #ifdef HAVE_FSEEKO
1018 if (fseeko (queue->temp_file, (off_t) offset, SEEK_SET) != 0)
1019 goto seek_failed;
1020 #elif defined (G_OS_UNIX) || defined (G_OS_WIN32)
1021 if (lseek (fileno (queue->temp_file), (off_t) offset,
1022 SEEK_SET) == (off_t) - 1)
1023 goto seek_failed;
1024 #else
1025 if (fseek (queue->temp_file, (long) offset, SEEK_SET) != 0)
1026 goto seek_failed;
1027 #endif
1029 buf = gst_buffer_new_and_alloc (length);
1031 /* this should not block */
1032 GST_LOG_OBJECT (queue, "Reading %d bytes", length);
1033 res = fread (GST_BUFFER_DATA (buf), 1, length, queue->temp_file);
1034 GST_LOG_OBJECT (queue, "read %" G_GSIZE_FORMAT " bytes", res);
1036 if (G_UNLIKELY (res == 0)) {
1037 /* check for errors or EOF */
1038 if (ferror (queue->temp_file))
1039 goto could_not_read;
1040 if (feof (queue->temp_file) && length > 0)
1041 goto eos;
1042 }
1044 length = res;
1046 GST_BUFFER_SIZE (buf) = length;
1047 GST_BUFFER_OFFSET (buf) = offset;
1048 GST_BUFFER_OFFSET_END (buf) = offset + length;
1050 *buffer = buf;
1052 reading_pos = queue->current->reading_pos;
1053 writing_pos = queue->current->writing_pos;
1054 max_reading_pos = queue->current->max_reading_pos;
1056 reading_pos = offset + length;
1057 max_reading_pos = MAX (max_reading_pos, reading_pos);
1059 if (writing_pos > max_reading_pos)
1060 queue->cur_level.bytes = writing_pos - max_reading_pos;
1061 else
1062 queue->cur_level.bytes = 0;
1064 queue->current->reading_pos = reading_pos;
1065 queue->current->max_reading_pos = max_reading_pos;
1067 return GST_FLOW_OK;
1069 /* ERRORS */
1070 out_flushing:
1071 {
1072 GST_DEBUG_OBJECT (queue, "we are flushing");
1073 return GST_FLOW_WRONG_STATE;
1074 }
1075 seek_failed:
1076 {
1077 GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
1078 return GST_FLOW_ERROR;
1079 }
1080 could_not_read:
1081 {
1082 GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
1083 gst_buffer_unref (buf);
1084 return GST_FLOW_ERROR;
1085 }
1086 eos:
1087 {
1088 GST_DEBUG ("non-regular file hits EOS");
1089 gst_buffer_unref (buf);
1090 return GST_FLOW_UNEXPECTED;
1091 }
1092 }
1094 /* should be called with QUEUE_LOCK */
1095 static GstMiniObject *
1096 gst_queue2_read_item_from_file (GstQueue2 * queue)
1097 {
1098 GstMiniObject *item;
1100 if (queue->starting_segment != NULL) {
1101 item = GST_MINI_OBJECT_CAST (queue->starting_segment);
1102 queue->starting_segment = NULL;
1103 } else {
1104 GstFlowReturn ret;
1105 GstBuffer *buffer;
1106 guint64 reading_pos;
1108 reading_pos = queue->current->reading_pos;
1110 ret =
1111 gst_queue2_create_read (queue, reading_pos, DEFAULT_BUFFER_SIZE,
1112 &buffer);
1113 switch (ret) {
1114 case GST_FLOW_OK:
1115 item = GST_MINI_OBJECT_CAST (buffer);
1116 break;
1117 case GST_FLOW_UNEXPECTED:
1118 item = GST_MINI_OBJECT_CAST (gst_event_new_eos ());
1119 break;
1120 default:
1121 item = NULL;
1122 break;
1123 }
1124 }
1125 return item;
1126 }
1128 static gboolean
1129 gst_queue2_open_temp_location_file (GstQueue2 * queue)
1130 {
1131 gint fd = -1;
1132 gchar *name = NULL;
1134 if (queue->temp_file)
1135 goto already_opened;
1137 GST_DEBUG_OBJECT (queue, "opening temp file %s", queue->temp_template);
1139 /* we have two cases:
1140 * - temp_location was set to something !NULL (Deprecated). in this case we
1141 * open the specified filename.
1142 * - temp_template was set, allocate a filename and open that filename
1143 */
1144 if (!queue->temp_location_set) {
1145 /* nothing to do */
1146 if (queue->temp_template == NULL)
1147 goto no_directory;
1149 /* make copy of the template, we don't want to change this */
1150 name = g_strdup (queue->temp_template);
1151 fd = g_mkstemp (name);
1152 if (fd == -1)
1153 goto mkstemp_failed;
1155 /* open the file for update/writing */
1156 queue->temp_file = fdopen (fd, "wb+");
1157 /* error creating file */
1158 if (queue->temp_file == NULL)
1159 goto open_failed;
1161 g_free (queue->temp_location);
1162 queue->temp_location = name;
1164 g_object_notify (G_OBJECT (queue), "temp-location");
1165 } else {
1166 /* open the file for update/writing, this is deprecated but we still need to
1167 * support it for API/ABI compatibility */
1168 queue->temp_file = g_fopen (queue->temp_location, "wb+");
1169 /* error creating file */
1170 if (queue->temp_file == NULL)
1171 goto open_failed;
1172 }
1173 GST_DEBUG_OBJECT (queue, "opened temp file %s", queue->temp_template);
1175 init_ranges (queue);
1177 return TRUE;
1179 /* ERRORS */
1180 already_opened:
1181 {
1182 GST_DEBUG_OBJECT (queue, "temp file was already open");
1183 return TRUE;
1184 }
1185 no_directory:
1186 {
1187 GST_ELEMENT_ERROR (queue, RESOURCE, NOT_FOUND,
1188 (_("No Temp directory specified.")), (NULL));
1189 return FALSE;
1190 }
1191 mkstemp_failed:
1192 {
1193 GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1194 (_("Could not create temp file \"%s\"."), queue->temp_template),
1195 GST_ERROR_SYSTEM);
1196 g_free (name);
1197 return FALSE;
1198 }
1199 open_failed:
1200 {
1201 GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1202 (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
1203 g_free (name);
1204 if (fd != -1)
1205 close (fd);
1206 return FALSE;
1207 }
1208 }
1210 static void
1211 gst_queue2_close_temp_location_file (GstQueue2 * queue)
1212 {
1213 /* nothing to do */
1214 if (queue->temp_file == NULL)
1215 return;
1217 GST_DEBUG_OBJECT (queue, "closing temp file");
1219 fflush (queue->temp_file);
1220 fclose (queue->temp_file);
1222 if (queue->temp_remove)
1223 remove (queue->temp_location);
1225 queue->temp_file = NULL;
1226 clean_ranges (queue);
1227 }
1229 static void
1230 gst_queue2_flush_temp_file (GstQueue2 * queue)
1231 {
1232 if (queue->temp_file == NULL)
1233 return;
1235 GST_DEBUG_OBJECT (queue, "flushing temp file");
1237 queue->temp_file = g_freopen (queue->temp_location, "wb+", queue->temp_file);
1239 init_ranges (queue);
1240 }
1242 static void
1243 gst_queue2_locked_flush (GstQueue2 * queue)
1244 {
1245 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1246 gst_queue2_flush_temp_file (queue);
1247 } else {
1248 while (!g_queue_is_empty (queue->queue)) {
1249 GstMiniObject *data = g_queue_pop_head (queue->queue);
1251 /* Then lose another reference because we are supposed to destroy that
1252 data when flushing */
1253 gst_mini_object_unref (data);
1254 }
1255 }
1256 GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
1257 gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
1258 gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
1259 if (queue->starting_segment != NULL)
1260 gst_event_unref (queue->starting_segment);
1261 queue->starting_segment = NULL;
1262 queue->segment_event_received = FALSE;
1264 /* we deleted a lot of something */
1265 GST_QUEUE2_SIGNAL_DEL (queue);
1266 }
1268 /* enqueue an item an update the level stats */
1269 static void
1270 gst_queue2_locked_enqueue (GstQueue2 * queue, gpointer item)
1271 {
1272 if (GST_IS_BUFFER (item)) {
1273 GstBuffer *buffer;
1274 guint size;
1276 buffer = GST_BUFFER_CAST (item);
1277 size = GST_BUFFER_SIZE (buffer);
1279 /* add buffer to the statistics */
1280 queue->cur_level.buffers++;
1281 queue->cur_level.bytes += size;
1282 queue->bytes_in += size;
1284 /* apply new buffer to segment stats */
1285 apply_buffer (queue, buffer, &queue->sink_segment);
1286 /* update the byterate stats */
1287 update_in_rates (queue);
1289 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1290 gst_queue2_write_buffer_to_file (queue, buffer);
1291 }
1293 } else if (GST_IS_EVENT (item)) {
1294 GstEvent *event;
1296 event = GST_EVENT_CAST (item);
1298 switch (GST_EVENT_TYPE (event)) {
1299 case GST_EVENT_EOS:
1300 /* Zero the thresholds, this makes sure the queue is completely
1301 * filled and we can read all data from the queue. */
1302 GST_DEBUG_OBJECT (queue, "we have EOS");
1303 queue->is_eos = TRUE;
1304 break;
1305 case GST_EVENT_NEWSEGMENT:
1306 apply_segment (queue, event, &queue->sink_segment);
1307 /* This is our first new segment, we hold it
1308 * as we can't save it on the temp file */
1309 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1310 if (queue->segment_event_received)
1311 goto unexpected_event;
1313 queue->segment_event_received = TRUE;
1314 if (queue->starting_segment != NULL)
1315 gst_event_unref (queue->starting_segment);
1316 queue->starting_segment = event;
1317 item = NULL;
1318 }
1319 /* a new segment allows us to accept more buffers if we got UNEXPECTED
1320 * from downstream */
1321 queue->unexpected = FALSE;
1322 break;
1323 default:
1324 if (QUEUE_IS_USING_TEMP_FILE (queue))
1325 goto unexpected_event;
1326 break;
1327 }
1328 } else {
1329 g_warning ("Unexpected item %p added in queue %s (refcounting problem?)",
1330 item, GST_OBJECT_NAME (queue));
1331 /* we can't really unref since we don't know what it is */
1332 item = NULL;
1333 }
1335 if (item) {
1336 /* update the buffering status */
1337 update_buffering (queue);
1339 if (!QUEUE_IS_USING_TEMP_FILE (queue))
1340 g_queue_push_tail (queue->queue, item);
1341 else
1342 gst_mini_object_unref (GST_MINI_OBJECT_CAST (item));
1344 GST_QUEUE2_SIGNAL_ADD (queue);
1345 }
1347 return;
1349 /* ERRORS */
1350 unexpected_event:
1351 {
1352 g_warning
1353 ("Unexpected event of kind %s can't be added in temp file of queue %s ",
1354 gst_event_type_get_name (GST_EVENT_TYPE (item)),
1355 GST_OBJECT_NAME (queue));
1356 gst_event_unref (GST_EVENT_CAST (item));
1357 return;
1358 }
1359 }
1361 /* dequeue an item from the queue and update level stats */
1362 static GstMiniObject *
1363 gst_queue2_locked_dequeue (GstQueue2 * queue)
1364 {
1365 GstMiniObject *item;
1367 if (QUEUE_IS_USING_TEMP_FILE (queue))
1368 item = gst_queue2_read_item_from_file (queue);
1369 else
1370 item = g_queue_pop_head (queue->queue);
1372 if (item == NULL)
1373 goto no_item;
1375 if (GST_IS_BUFFER (item)) {
1376 GstBuffer *buffer;
1377 guint size;
1379 buffer = GST_BUFFER_CAST (item);
1380 size = GST_BUFFER_SIZE (buffer);
1382 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1383 "retrieved buffer %p from queue", buffer);
1385 queue->cur_level.buffers--;
1386 queue->cur_level.bytes -= size;
1387 queue->bytes_out += size;
1388 apply_buffer (queue, buffer, &queue->src_segment);
1389 /* update the byterate stats */
1390 update_out_rates (queue);
1391 /* update the buffering */
1392 update_buffering (queue);
1394 } else if (GST_IS_EVENT (item)) {
1395 GstEvent *event = GST_EVENT_CAST (item);
1397 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1398 "retrieved event %p from queue", event);
1400 switch (GST_EVENT_TYPE (event)) {
1401 case GST_EVENT_EOS:
1402 /* queue is empty now that we dequeued the EOS */
1403 GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
1404 break;
1405 case GST_EVENT_NEWSEGMENT:
1406 apply_segment (queue, event, &queue->src_segment);
1407 break;
1408 default:
1409 break;
1410 }
1411 } else {
1412 g_warning
1413 ("Unexpected item %p dequeued from queue %s (refcounting problem?)",
1414 item, GST_OBJECT_NAME (queue));
1415 item = NULL;
1416 }
1417 GST_QUEUE2_SIGNAL_DEL (queue);
1419 return item;
1421 /* ERRORS */
1422 no_item:
1423 {
1424 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "the queue is empty");
1425 return NULL;
1426 }
1427 }
1429 static gboolean
1430 gst_queue2_handle_sink_event (GstPad * pad, GstEvent * event)
1431 {
1432 GstQueue2 *queue;
1434 queue = GST_QUEUE2 (GST_OBJECT_PARENT (pad));
1436 switch (GST_EVENT_TYPE (event)) {
1437 case GST_EVENT_FLUSH_START:
1438 {
1439 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush start event");
1440 if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1441 /* forward event */
1442 gst_pad_push_event (queue->srcpad, event);
1444 /* now unblock the chain function */
1445 GST_QUEUE2_MUTEX_LOCK (queue);
1446 queue->srcresult = GST_FLOW_WRONG_STATE;
1447 queue->sinkresult = GST_FLOW_WRONG_STATE;
1448 /* unblock the loop and chain functions */
1449 GST_QUEUE2_SIGNAL_ADD (queue);
1450 GST_QUEUE2_SIGNAL_DEL (queue);
1451 GST_QUEUE2_MUTEX_UNLOCK (queue);
1453 /* make sure it pauses, this should happen since we sent
1454 * flush_start downstream. */
1455 gst_pad_pause_task (queue->srcpad);
1456 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "loop stopped");
1457 }
1458 goto done;
1459 }
1460 case GST_EVENT_FLUSH_STOP:
1461 {
1462 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush stop event");
1464 if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1465 /* forward event */
1466 gst_pad_push_event (queue->srcpad, event);
1468 GST_QUEUE2_MUTEX_LOCK (queue);
1469 gst_queue2_locked_flush (queue);
1470 queue->srcresult = GST_FLOW_OK;
1471 queue->sinkresult = GST_FLOW_OK;
1472 queue->is_eos = FALSE;
1473 queue->unexpected = FALSE;
1474 /* reset rate counters */
1475 reset_rate_timer (queue);
1476 gst_pad_start_task (queue->srcpad, (GstTaskFunction) gst_queue2_loop,
1477 queue->srcpad);
1478 GST_QUEUE2_MUTEX_UNLOCK (queue);
1479 } else {
1480 GST_QUEUE2_MUTEX_LOCK (queue);
1481 queue->segment_event_received = FALSE;
1482 queue->is_eos = FALSE;
1483 queue->unexpected = FALSE;
1484 GST_QUEUE2_MUTEX_UNLOCK (queue);
1485 }
1486 goto done;
1487 }
1488 default:
1489 if (GST_EVENT_IS_SERIALIZED (event)) {
1490 /* serialized events go in the queue */
1491 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->sinkresult, out_flushing);
1492 /* refuse more events on EOS */
1493 if (queue->is_eos)
1494 goto out_eos;
1495 gst_queue2_locked_enqueue (queue, event);
1496 GST_QUEUE2_MUTEX_UNLOCK (queue);
1497 } else {
1498 /* non-serialized events are passed upstream. */
1499 gst_pad_push_event (queue->srcpad, event);
1500 }
1501 break;
1502 }
1503 done:
1504 return TRUE;
1506 /* ERRORS */
1507 out_flushing:
1508 {
1509 GST_DEBUG_OBJECT (queue, "refusing event, we are flushing");
1510 GST_QUEUE2_MUTEX_UNLOCK (queue);
1511 gst_event_unref (event);
1512 return FALSE;
1513 }
1514 out_eos:
1515 {
1516 GST_DEBUG_OBJECT (queue, "refusing event, we are EOS");
1517 GST_QUEUE2_MUTEX_UNLOCK (queue);
1518 gst_event_unref (event);
1519 return FALSE;
1520 }
1521 }
1523 static gboolean
1524 gst_queue2_is_empty (GstQueue2 * queue)
1525 {
1526 /* never empty on EOS */
1527 if (queue->is_eos)
1528 return FALSE;
1530 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1531 return queue->current->writing_pos == queue->current->max_reading_pos;
1532 } else {
1533 if (queue->queue->length == 0)
1534 return TRUE;
1535 }
1537 return FALSE;
1538 }
1540 static gboolean
1541 gst_queue2_is_filled (GstQueue2 * queue)
1542 {
1543 gboolean res;
1545 /* always filled on EOS */
1546 if (queue->is_eos)
1547 return TRUE;
1549 /* if using file, we're never filled if we don't have EOS */
1550 if (QUEUE_IS_USING_TEMP_FILE (queue))
1551 return FALSE;
1553 /* we are never filled when we have no buffers at all */
1554 if (queue->cur_level.buffers == 0)
1555 return FALSE;
1557 #define CHECK_FILLED(format) ((queue->max_level.format) > 0 && \
1558 (queue->cur_level.format) >= (queue->max_level.format))
1560 /* we are filled if one of the current levels exceeds the max */
1561 res = CHECK_FILLED (buffers) || CHECK_FILLED (bytes) || CHECK_FILLED (time);
1563 /* if we need to, use the rate estimate to check against the max time we are
1564 * allowed to queue */
1565 if (queue->use_rate_estimate)
1566 res |= CHECK_FILLED (rate_time);
1568 #undef CHECK_FILLED
1569 return res;
1570 }
1572 static GstFlowReturn
1573 gst_queue2_chain (GstPad * pad, GstBuffer * buffer)
1574 {
1575 GstQueue2 *queue;
1577 queue = GST_QUEUE2 (GST_OBJECT_PARENT (pad));
1579 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1580 "received buffer %p of size %d, time %" GST_TIME_FORMAT ", duration %"
1581 GST_TIME_FORMAT, buffer, GST_BUFFER_SIZE (buffer),
1582 GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1583 GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1585 /* we have to lock the queue since we span threads */
1586 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->sinkresult, out_flushing);
1587 /* when we received EOS, we refuse more data */
1588 if (queue->is_eos)
1589 goto out_eos;
1590 /* when we received unexpected from downstream, refuse more buffers */
1591 if (queue->unexpected)
1592 goto out_unexpected;
1594 /* We make space available if we're "full" according to whatever
1595 * the user defined as "full". */
1596 if (gst_queue2_is_filled (queue)) {
1597 gboolean started;
1599 /* pause the timer while we wait. The fact that we are waiting does not mean
1600 * the byterate on the input pad is lower */
1601 if ((started = queue->in_timer_started))
1602 g_timer_stop (queue->in_timer);
1604 GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
1605 "queue is full, waiting for free space");
1606 do {
1607 /* Wait for space to be available, we could be unlocked because of a flush. */
1608 GST_QUEUE2_WAIT_DEL_CHECK (queue, queue->sinkresult, out_flushing);
1609 }
1610 while (gst_queue2_is_filled (queue));
1612 /* and continue if we were running before */
1613 if (started)
1614 g_timer_continue (queue->in_timer);
1615 }
1617 /* put buffer in queue now */
1618 gst_queue2_locked_enqueue (queue, buffer);
1619 GST_QUEUE2_MUTEX_UNLOCK (queue);
1621 return GST_FLOW_OK;
1623 /* special conditions */
1624 out_flushing:
1625 {
1626 GstFlowReturn ret = queue->sinkresult;
1628 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1629 "exit because task paused, reason: %s", gst_flow_get_name (ret));
1630 GST_QUEUE2_MUTEX_UNLOCK (queue);
1631 gst_buffer_unref (buffer);
1633 return ret;
1634 }
1635 out_eos:
1636 {
1637 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we received EOS");
1638 GST_QUEUE2_MUTEX_UNLOCK (queue);
1639 gst_buffer_unref (buffer);
1641 return GST_FLOW_UNEXPECTED;
1642 }
1643 out_unexpected:
1644 {
1645 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1646 "exit because we received UNEXPECTED");
1647 GST_QUEUE2_MUTEX_UNLOCK (queue);
1648 gst_buffer_unref (buffer);
1650 return GST_FLOW_UNEXPECTED;
1651 }
1652 }
1654 /* dequeue an item from the queue an push it downstream. This functions returns
1655 * the result of the push. */
1656 static GstFlowReturn
1657 gst_queue2_push_one (GstQueue2 * queue)
1658 {
1659 GstFlowReturn result = GST_FLOW_OK;
1660 GstMiniObject *data;
1662 data = gst_queue2_locked_dequeue (queue);
1663 if (data == NULL)
1664 goto no_item;
1666 next:
1667 if (GST_IS_BUFFER (data)) {
1668 GstBuffer *buffer;
1669 GstCaps *caps;
1671 buffer = GST_BUFFER_CAST (data);
1672 caps = GST_BUFFER_CAPS (buffer);
1674 GST_QUEUE2_MUTEX_UNLOCK (queue);
1676 /* set caps before pushing the buffer so that core does not try to do
1677 * something fancy to check if this is possible. */
1678 if (caps && caps != GST_PAD_CAPS (queue->srcpad))
1679 gst_pad_set_caps (queue->srcpad, caps);
1681 result = gst_pad_push (queue->srcpad, buffer);
1683 /* need to check for srcresult here as well */
1684 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
1685 if (result == GST_FLOW_UNEXPECTED) {
1686 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1687 "got UNEXPECTED from downstream");
1688 /* stop pushing buffers, we dequeue all items until we see an item that we
1689 * can push again, which is EOS or NEWSEGMENT. If there is nothing in the
1690 * queue we can push, we set a flag to make the sinkpad refuse more
1691 * buffers with an UNEXPECTED return value until we receive something
1692 * pushable again or we get flushed. */
1693 while ((data = gst_queue2_locked_dequeue (queue))) {
1694 if (GST_IS_BUFFER (data)) {
1695 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1696 "dropping UNEXPECTED buffer %p", data);
1697 gst_buffer_unref (GST_BUFFER_CAST (data));
1698 } else if (GST_IS_EVENT (data)) {
1699 GstEvent *event = GST_EVENT_CAST (data);
1700 GstEventType type = GST_EVENT_TYPE (event);
1702 if (type == GST_EVENT_EOS || type == GST_EVENT_NEWSEGMENT) {
1703 /* we found a pushable item in the queue, push it out */
1704 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1705 "pushing pushable event %s after UNEXPECTED",
1706 GST_EVENT_TYPE_NAME (event));
1707 goto next;
1708 }
1709 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1710 "dropping UNEXPECTED event %p", event);
1711 gst_event_unref (event);
1712 }
1713 }
1714 /* no more items in the queue. Set the unexpected flag so that upstream
1715 * make us refuse any more buffers on the sinkpad. Since we will still
1716 * accept EOS and NEWSEGMENT we return _FLOW_OK to the caller so that the
1717 * task function does not shut down. */
1718 queue->unexpected = TRUE;
1719 result = GST_FLOW_OK;
1720 }
1721 } else if (GST_IS_EVENT (data)) {
1722 GstEvent *event = GST_EVENT_CAST (data);
1723 GstEventType type = GST_EVENT_TYPE (event);
1725 GST_QUEUE2_MUTEX_UNLOCK (queue);
1727 gst_pad_push_event (queue->srcpad, event);
1729 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
1730 /* if we're EOS, return UNEXPECTED so that the task pauses. */
1731 if (type == GST_EVENT_EOS) {
1732 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1733 "pushed EOS event %p, return UNEXPECTED", event);
1734 result = GST_FLOW_UNEXPECTED;
1735 }
1736 }
1737 return result;
1739 /* ERRORS */
1740 no_item:
1741 {
1742 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1743 "exit because we have no item in the queue");
1744 return GST_FLOW_ERROR;
1745 }
1746 out_flushing:
1747 {
1748 GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are flushing");
1749 return GST_FLOW_WRONG_STATE;
1750 }
1751 }
1753 /* called repeadedly with @pad as the source pad. This function should push out
1754 * data to the peer element. */
1755 static void
1756 gst_queue2_loop (GstPad * pad)
1757 {
1758 GstQueue2 *queue;
1759 GstFlowReturn ret;
1761 queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1763 /* have to lock for thread-safety */
1764 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
1766 if (gst_queue2_is_empty (queue)) {
1767 gboolean started;
1769 /* pause the timer while we wait. The fact that we are waiting does not mean
1770 * the byterate on the output pad is lower */
1771 if ((started = queue->out_timer_started))
1772 g_timer_stop (queue->out_timer);
1774 GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
1775 "queue is empty, waiting for new data");
1776 do {
1777 /* Wait for data to be available, we could be unlocked because of a flush. */
1778 GST_QUEUE2_WAIT_ADD_CHECK (queue, queue->srcresult, out_flushing);
1779 }
1780 while (gst_queue2_is_empty (queue));
1782 /* and continue if we were running before */
1783 if (started)
1784 g_timer_continue (queue->out_timer);
1785 }
1786 ret = gst_queue2_push_one (queue);
1787 queue->srcresult = ret;
1788 queue->sinkresult = ret;
1789 if (ret != GST_FLOW_OK)
1790 goto out_flushing;
1792 GST_QUEUE2_MUTEX_UNLOCK (queue);
1794 return;
1796 /* ERRORS */
1797 out_flushing:
1798 {
1799 gboolean eos = queue->is_eos;
1800 GstFlowReturn ret = queue->srcresult;
1802 gst_pad_pause_task (queue->srcpad);
1803 GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1804 "pause task, reason: %s", gst_flow_get_name (queue->srcresult));
1805 GST_QUEUE2_MUTEX_UNLOCK (queue);
1806 /* let app know about us giving up if upstream is not expected to do so */
1807 /* UNEXPECTED is already taken care of elsewhere */
1808 if (eos && (GST_FLOW_IS_FATAL (ret) || ret == GST_FLOW_NOT_LINKED) &&
1809 (ret != GST_FLOW_UNEXPECTED)) {
1810 GST_ELEMENT_ERROR (queue, STREAM, FAILED,
1811 (_("Internal data flow error.")),
1812 ("streaming task paused, reason %s (%d)",
1813 gst_flow_get_name (ret), ret));
1814 gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
1815 }
1816 return;
1817 }
1818 }
1820 static gboolean
1821 gst_queue2_handle_src_event (GstPad * pad, GstEvent * event)
1822 {
1823 gboolean res = TRUE;
1824 GstQueue2 *queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1826 #ifndef GST_DISABLE_GST_DEBUG
1827 GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "got event %p (%s)",
1828 event, GST_EVENT_TYPE_NAME (event));
1829 #endif
1831 switch (GST_EVENT_TYPE (event)) {
1832 case GST_EVENT_FLUSH_START:
1833 if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1834 /* just forward upstream */
1835 res = gst_pad_push_event (queue->sinkpad, event);
1836 } else {
1837 /* now unblock the getrange function */
1838 GST_QUEUE2_MUTEX_LOCK (queue);
1839 GST_DEBUG_OBJECT (queue, "flushing");
1840 queue->srcresult = GST_FLOW_WRONG_STATE;
1841 GST_QUEUE2_SIGNAL_ADD (queue);
1842 GST_QUEUE2_MUTEX_UNLOCK (queue);
1844 /* when using a temp file, we eat the event */
1845 res = TRUE;
1846 gst_event_unref (event);
1847 }
1848 break;
1849 case GST_EVENT_FLUSH_STOP:
1850 if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1851 /* just forward upstream */
1852 res = gst_pad_push_event (queue->sinkpad, event);
1853 } else {
1854 /* now unblock the getrange function */
1855 GST_QUEUE2_MUTEX_LOCK (queue);
1856 queue->srcresult = GST_FLOW_OK;
1857 GST_QUEUE2_MUTEX_UNLOCK (queue);
1859 /* when using a temp file, we eat the event */
1860 res = TRUE;
1861 gst_event_unref (event);
1862 }
1863 break;
1864 default:
1865 res = gst_pad_push_event (queue->sinkpad, event);
1866 break;
1867 }
1869 return res;
1870 }
1872 static gboolean
1873 gst_queue2_peer_query (GstQueue2 * queue, GstPad * pad, GstQuery * query)
1874 {
1875 gboolean ret = FALSE;
1876 GstPad *peer;
1878 if ((peer = gst_pad_get_peer (pad))) {
1879 ret = gst_pad_query (peer, query);
1880 gst_object_unref (peer);
1881 }
1882 return ret;
1883 }
1885 static gboolean
1886 gst_queue2_handle_src_query (GstPad * pad, GstQuery * query)
1887 {
1888 GstQueue2 *queue;
1890 queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1892 switch (GST_QUERY_TYPE (query)) {
1893 case GST_QUERY_POSITION:
1894 {
1895 gint64 peer_pos;
1896 GstFormat format;
1898 if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1899 goto peer_failed;
1901 /* get peer position */
1902 gst_query_parse_position (query, &format, &peer_pos);
1904 /* FIXME: this code assumes that there's no discont in the queue */
1905 switch (format) {
1906 case GST_FORMAT_BYTES:
1907 peer_pos -= queue->cur_level.bytes;
1908 break;
1909 case GST_FORMAT_TIME:
1910 peer_pos -= queue->cur_level.time;
1911 break;
1912 default:
1913 GST_WARNING_OBJECT (queue, "dropping query in %s format, don't "
1914 "know how to adjust value", gst_format_get_name (format));
1915 return FALSE;
1916 }
1917 /* set updated position */
1918 gst_query_set_position (query, format, peer_pos);
1919 break;
1920 }
1921 case GST_QUERY_DURATION:
1922 {
1923 GST_DEBUG_OBJECT (queue, "doing peer query");
1925 if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1926 goto peer_failed;
1928 GST_DEBUG_OBJECT (queue, "peer query success");
1929 break;
1930 }
1931 case GST_QUERY_BUFFERING:
1932 {
1933 GstFormat format;
1935 GST_DEBUG_OBJECT (queue, "query buffering");
1937 if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1938 /* no temp file, just forward to the peer */
1939 if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1940 goto peer_failed;
1941 GST_DEBUG_OBJECT (queue, "buffering forwarded to peer");
1942 } else {
1943 gint64 start, stop;
1944 guint64 writing_pos;
1945 gint64 estimated_total, buffering_left;
1946 GstFormat peer_fmt;
1947 gint64 duration;
1948 gboolean peer_res, is_buffering;
1949 gdouble byte_in_rate, byte_out_rate;
1951 /* we need a current download region */
1952 if (queue->current == NULL)
1953 return FALSE;
1955 writing_pos = queue->current->writing_pos;
1956 byte_in_rate = queue->byte_in_rate;
1957 byte_out_rate = queue->byte_out_rate;
1958 is_buffering = queue->is_buffering;
1960 /* get duration of upstream in bytes */
1961 peer_fmt = GST_FORMAT_BYTES;
1962 peer_res = gst_pad_query_peer_duration (queue->sinkpad, &peer_fmt,
1963 &duration);
1965 /* calculate remaining and total download time */
1966 if (peer_res && byte_in_rate > 0.0) {
1967 estimated_total = (duration * 1000) / byte_in_rate;
1968 buffering_left = ((duration - writing_pos) * 1000) / byte_in_rate;
1969 } else {
1970 estimated_total = -1;
1971 buffering_left = -1;
1972 }
1974 gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1976 switch (format) {
1977 case GST_FORMAT_PERCENT:
1978 /* we need duration */
1979 if (!peer_res)
1980 goto peer_failed;
1982 GST_DEBUG_OBJECT (queue, "duration %" G_GINT64_FORMAT ", writing %"
1983 G_GINT64_FORMAT, duration, writing_pos);
1985 start = 0;
1986 /* get our available data relative to the duration */
1987 if (duration != -1)
1988 stop = GST_FORMAT_PERCENT_MAX * writing_pos / duration;
1989 else
1990 stop = -1;
1991 break;
1992 case GST_FORMAT_BYTES:
1993 start = 0;
1994 stop = writing_pos;
1995 break;
1996 default:
1997 start = -1;
1998 stop = -1;
1999 break;
2000 }
2001 /* FIXME, percent buffering is not right */
2002 gst_query_set_buffering_percent (query, is_buffering, 100);
2003 gst_query_set_buffering_range (query, format, start, stop,
2004 estimated_total);
2005 gst_query_set_buffering_stats (query, GST_BUFFERING_DOWNLOAD,
2006 byte_in_rate, byte_out_rate, buffering_left);
2007 }
2008 break;
2009 }
2010 default:
2011 /* peer handled other queries */
2012 if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
2013 goto peer_failed;
2014 break;
2015 }
2017 return TRUE;
2019 /* ERRORS */
2020 peer_failed:
2021 {
2022 GST_DEBUG_OBJECT (queue, "failed peer query");
2023 return FALSE;
2024 }
2025 }
2027 static GstFlowReturn
2028 gst_queue2_get_range (GstPad * pad, guint64 offset, guint length,
2029 GstBuffer ** buffer)
2030 {
2031 GstQueue2 *queue;
2032 GstFlowReturn ret;
2034 queue = GST_QUEUE2_CAST (gst_pad_get_parent (pad));
2036 GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2037 length = (length == -1) ? DEFAULT_BUFFER_SIZE : length;
2038 offset = (offset == -1) ? queue->current->reading_pos : offset;
2040 /* function will block when the range is not yet available */
2041 ret = gst_queue2_create_read (queue, offset, length, buffer);
2042 GST_QUEUE2_MUTEX_UNLOCK (queue);
2044 gst_object_unref (queue);
2046 return ret;
2048 /* ERRORS */
2049 out_flushing:
2050 {
2051 ret = queue->srcresult;
2053 GST_DEBUG_OBJECT (queue, "we are flushing");
2054 GST_QUEUE2_MUTEX_UNLOCK (queue);
2055 return ret;
2056 }
2057 }
2059 static gboolean
2060 gst_queue2_src_checkgetrange_function (GstPad * pad)
2061 {
2062 GstQueue2 *queue;
2063 gboolean ret;
2065 queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2067 /* we can operate in pull mode when we are using a tempfile */
2068 ret = QUEUE_IS_USING_TEMP_FILE (queue);
2070 gst_object_unref (GST_OBJECT (queue));
2072 return ret;
2073 }
2075 /* sink currently only operates in push mode */
2076 static gboolean
2077 gst_queue2_sink_activate_push (GstPad * pad, gboolean active)
2078 {
2079 gboolean result = TRUE;
2080 GstQueue2 *queue;
2082 queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2084 if (active) {
2085 GST_QUEUE2_MUTEX_LOCK (queue);
2086 GST_DEBUG_OBJECT (queue, "activating push mode");
2087 queue->srcresult = GST_FLOW_OK;
2088 queue->sinkresult = GST_FLOW_OK;
2089 queue->is_eos = FALSE;
2090 queue->unexpected = FALSE;
2091 reset_rate_timer (queue);
2092 GST_QUEUE2_MUTEX_UNLOCK (queue);
2093 } else {
2094 /* unblock chain function */
2095 GST_QUEUE2_MUTEX_LOCK (queue);
2096 GST_DEBUG_OBJECT (queue, "deactivating push mode");
2097 queue->srcresult = GST_FLOW_WRONG_STATE;
2098 queue->sinkresult = GST_FLOW_WRONG_STATE;
2099 gst_queue2_locked_flush (queue);
2100 GST_QUEUE2_MUTEX_UNLOCK (queue);
2101 }
2103 gst_object_unref (queue);
2105 return result;
2106 }
2108 /* src operating in push mode, we start a task on the source pad that pushes out
2109 * buffers from the queue */
2110 static gboolean
2111 gst_queue2_src_activate_push (GstPad * pad, gboolean active)
2112 {
2113 gboolean result = FALSE;
2114 GstQueue2 *queue;
2116 queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2118 if (active) {
2119 GST_QUEUE2_MUTEX_LOCK (queue);
2120 GST_DEBUG_OBJECT (queue, "activating push mode");
2121 queue->srcresult = GST_FLOW_OK;
2122 queue->sinkresult = GST_FLOW_OK;
2123 queue->is_eos = FALSE;
2124 queue->unexpected = FALSE;
2125 result = gst_pad_start_task (pad, (GstTaskFunction) gst_queue2_loop, pad);
2126 GST_QUEUE2_MUTEX_UNLOCK (queue);
2127 } else {
2128 /* unblock loop function */
2129 GST_QUEUE2_MUTEX_LOCK (queue);
2130 GST_DEBUG_OBJECT (queue, "deactivating push mode");
2131 queue->srcresult = GST_FLOW_WRONG_STATE;
2132 queue->sinkresult = GST_FLOW_WRONG_STATE;
2133 /* the item add signal will unblock */
2134 GST_QUEUE2_SIGNAL_ADD (queue);
2135 GST_QUEUE2_MUTEX_UNLOCK (queue);
2137 /* step 2, make sure streaming finishes */
2138 result = gst_pad_stop_task (pad);
2139 }
2141 gst_object_unref (queue);
2143 return result;
2144 }
2146 /* pull mode, downstream will call our getrange function */
2147 static gboolean
2148 gst_queue2_src_activate_pull (GstPad * pad, gboolean active)
2149 {
2150 gboolean result;
2151 GstQueue2 *queue;
2153 queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2155 if (active) {
2156 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
2157 /* open the temp file now */
2158 result = gst_queue2_open_temp_location_file (queue);
2160 GST_QUEUE2_MUTEX_LOCK (queue);
2161 GST_DEBUG_OBJECT (queue, "activating pull mode");
2162 queue->srcresult = GST_FLOW_OK;
2163 queue->sinkresult = GST_FLOW_OK;
2164 queue->is_eos = FALSE;
2165 queue->unexpected = FALSE;
2166 GST_QUEUE2_MUTEX_UNLOCK (queue);
2167 } else {
2168 GST_QUEUE2_MUTEX_LOCK (queue);
2169 GST_DEBUG_OBJECT (queue, "no temp file, cannot activate pull mode");
2170 /* this is not allowed, we cannot operate in pull mode without a temp
2171 * file. */
2172 queue->srcresult = GST_FLOW_WRONG_STATE;
2173 queue->sinkresult = GST_FLOW_WRONG_STATE;
2174 result = FALSE;
2175 GST_QUEUE2_MUTEX_UNLOCK (queue);
2176 }
2177 } else {
2178 GST_QUEUE2_MUTEX_LOCK (queue);
2179 GST_DEBUG_OBJECT (queue, "deactivating pull mode");
2180 queue->srcresult = GST_FLOW_WRONG_STATE;
2181 queue->sinkresult = GST_FLOW_WRONG_STATE;
2182 /* this will unlock getrange */
2183 GST_QUEUE2_SIGNAL_ADD (queue);
2184 result = TRUE;
2185 GST_QUEUE2_MUTEX_UNLOCK (queue);
2186 }
2187 gst_object_unref (queue);
2189 return result;
2190 }
2192 static GstStateChangeReturn
2193 gst_queue2_change_state (GstElement * element, GstStateChange transition)
2194 {
2195 GstQueue2 *queue;
2196 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
2198 queue = GST_QUEUE2 (element);
2200 switch (transition) {
2201 case GST_STATE_CHANGE_NULL_TO_READY:
2202 break;
2203 case GST_STATE_CHANGE_READY_TO_PAUSED:
2204 if (QUEUE_IS_USING_TEMP_FILE (queue)) {
2205 if (!gst_queue2_open_temp_location_file (queue))
2206 ret = GST_STATE_CHANGE_FAILURE;
2207 }
2208 queue->segment_event_received = FALSE;
2209 queue->starting_segment = NULL;
2210 break;
2211 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
2212 break;
2213 default:
2214 break;
2215 }
2217 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2219 switch (transition) {
2220 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2221 break;
2222 case GST_STATE_CHANGE_PAUSED_TO_READY:
2223 if (QUEUE_IS_USING_TEMP_FILE (queue))
2224 gst_queue2_close_temp_location_file (queue);
2225 if (queue->starting_segment != NULL) {
2226 gst_event_unref (queue->starting_segment);
2227 queue->starting_segment = NULL;
2228 }
2229 break;
2230 case GST_STATE_CHANGE_READY_TO_NULL:
2231 break;
2232 default:
2233 break;
2234 }
2236 return ret;
2237 }
2239 /* changing the capacity of the queue must wake up
2240 * the _chain function, it might have more room now
2241 * to store the buffer/event in the queue */
2242 #define QUEUE_CAPACITY_CHANGE(q)\
2243 GST_QUEUE2_SIGNAL_DEL (queue);
2245 /* Changing the minimum required fill level must
2246 * wake up the _loop function as it might now
2247 * be able to preceed.
2248 */
2249 #define QUEUE_THRESHOLD_CHANGE(q)\
2250 GST_QUEUE2_SIGNAL_ADD (queue);
2252 static void
2253 gst_queue2_set_temp_template (GstQueue2 * queue, const gchar * template)
2254 {
2255 GstState state;
2257 /* the element must be stopped in order to do this */
2258 GST_OBJECT_LOCK (queue);
2259 state = GST_STATE (queue);
2260 if (state != GST_STATE_READY && state != GST_STATE_NULL)
2261 goto wrong_state;
2262 GST_OBJECT_UNLOCK (queue);
2264 /* set new location */
2265 g_free (queue->temp_template);
2266 queue->temp_template = g_strdup (template);
2268 return;
2270 /* ERROR */
2271 wrong_state:
2272 {
2273 GST_WARNING_OBJECT (queue, "setting temp-template property in wrong state");
2274 GST_OBJECT_UNLOCK (queue);
2275 }
2276 }
2278 static void
2279 gst_queue2_set_property (GObject * object,
2280 guint prop_id, const GValue * value, GParamSpec * pspec)
2281 {
2282 GstQueue2 *queue = GST_QUEUE2 (object);
2284 /* someone could change levels here, and since this
2285 * affects the get/put funcs, we need to lock for safety. */
2286 GST_QUEUE2_MUTEX_LOCK (queue);
2288 switch (prop_id) {
2289 case PROP_MAX_SIZE_BYTES:
2290 queue->max_level.bytes = g_value_get_uint (value);
2291 QUEUE_CAPACITY_CHANGE (queue);
2292 break;
2293 case PROP_MAX_SIZE_BUFFERS:
2294 queue->max_level.buffers = g_value_get_uint (value);
2295 QUEUE_CAPACITY_CHANGE (queue);
2296 break;
2297 case PROP_MAX_SIZE_TIME:
2298 queue->max_level.time = g_value_get_uint64 (value);
2299 /* set rate_time to the same value. We use an extra field in the level
2300 * structure so that we can easily access and compare it */
2301 queue->max_level.rate_time = queue->max_level.time;
2302 QUEUE_CAPACITY_CHANGE (queue);
2303 break;
2304 case PROP_USE_BUFFERING:
2305 queue->use_buffering = g_value_get_boolean (value);
2306 break;
2307 case PROP_USE_RATE_ESTIMATE:
2308 queue->use_rate_estimate = g_value_get_boolean (value);
2309 break;
2310 case PROP_LOW_PERCENT:
2311 queue->low_percent = g_value_get_int (value);
2312 break;
2313 case PROP_HIGH_PERCENT:
2314 queue->high_percent = g_value_get_int (value);
2315 break;
2316 case PROP_TEMP_TEMPLATE:
2317 gst_queue2_set_temp_template (queue, g_value_get_string (value));
2318 break;
2319 case PROP_TEMP_LOCATION:
2320 g_free (queue->temp_location);
2321 queue->temp_location = g_value_dup_string (value);
2322 /* you can set the property back to NULL to make it use the temp-tmpl
2323 * property. */
2324 queue->temp_location_set = queue->temp_location != NULL;
2325 break;
2326 case PROP_TEMP_REMOVE:
2327 queue->temp_remove = g_value_get_boolean (value);
2328 break;
2329 default:
2330 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2331 break;
2332 }
2334 GST_QUEUE2_MUTEX_UNLOCK (queue);
2335 }
2337 static void
2338 gst_queue2_get_property (GObject * object,
2339 guint prop_id, GValue * value, GParamSpec * pspec)
2340 {
2341 GstQueue2 *queue = GST_QUEUE2 (object);
2343 GST_QUEUE2_MUTEX_LOCK (queue);
2345 switch (prop_id) {
2346 case PROP_CUR_LEVEL_BYTES:
2347 g_value_set_uint (value, queue->cur_level.bytes);
2348 break;
2349 case PROP_CUR_LEVEL_BUFFERS:
2350 g_value_set_uint (value, queue->cur_level.buffers);
2351 break;
2352 case PROP_CUR_LEVEL_TIME:
2353 g_value_set_uint64 (value, queue->cur_level.time);
2354 break;
2355 case PROP_MAX_SIZE_BYTES:
2356 g_value_set_uint (value, queue->max_level.bytes);
2357 break;
2358 case PROP_MAX_SIZE_BUFFERS:
2359 g_value_set_uint (value, queue->max_level.buffers);
2360 break;
2361 case PROP_MAX_SIZE_TIME:
2362 g_value_set_uint64 (value, queue->max_level.time);
2363 break;
2364 case PROP_USE_BUFFERING:
2365 g_value_set_boolean (value, queue->use_buffering);
2366 break;
2367 case PROP_USE_RATE_ESTIMATE:
2368 g_value_set_boolean (value, queue->use_rate_estimate);
2369 break;
2370 case PROP_LOW_PERCENT:
2371 g_value_set_int (value, queue->low_percent);
2372 break;
2373 case PROP_HIGH_PERCENT:
2374 g_value_set_int (value, queue->high_percent);
2375 break;
2376 case PROP_TEMP_TEMPLATE:
2377 g_value_set_string (value, queue->temp_template);
2378 break;
2379 case PROP_TEMP_LOCATION:
2380 g_value_set_string (value, queue->temp_location);
2381 break;
2382 case PROP_TEMP_REMOVE:
2383 g_value_set_boolean (value, queue->temp_remove);
2384 break;
2385 default:
2386 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2387 break;
2388 }
2390 GST_QUEUE2_MUTEX_UNLOCK (queue);
2391 }