2 * Copyright (C) 2014 Wim Taymans <wim.taymans@gmail.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., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
23 * SECTION:element-downloadbuffer
24 * @title: downloadbuffer
26 * The downloadbuffer element provides on-disk buffering and caching of, typically,
27 * a network file. temp-template should be set to a value such as
28 * /tmp/gstreamer-XXXXXX and the element will allocate a random free filename and
29 * buffer the data in the file.
31 * With max-size-bytes and max-size-time you can configure the buffering limits.
32 * The downloadbuffer element will try to read-ahead these amounts of data. When
33 * the amount of read-ahead data drops below low-percent of the configured max,
34 * the element will start emitting BUFFERING messages until high-percent of max is
37 * The downloadbuffer provides push and pull based scheduling on its source pad
38 * and will efficiently seek in the upstream element when needed.
40 * The temp-location property will be used to notify the application of the
43 * When the downloadbuffer has completely downloaded the media, it will
44 * post an application message named <classname>"GstCacheDownloadComplete"</classname>
45 * with the following information:
49 * <classname>"location"</classname>:
50 * the location of the completely downloaded file.
58 #include "gstdownloadbuffer.h"
60 #include <glib/gstdio.h>
62 #include "gst/gst-i18n-lib.h"
63 #include "gst/glib-compat-private.h"
68 #include <io.h> /* lseek, open, close, read */
70 #define lseek _lseeki64
79 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
84 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
89 GST_DEBUG_CATEGORY_STATIC (downloadbuffer_debug);
90 #define GST_CAT_DEFAULT (downloadbuffer_debug)
98 #define DEFAULT_BUFFER_SIZE 4096
100 /* default property values */
101 #define DEFAULT_MAX_SIZE_BYTES (2 * 1024 * 1024) /* 2 MB */
102 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND /* 2 seconds */
103 #define DEFAULT_LOW_PERCENT 10
104 #define DEFAULT_HIGH_PERCENT 99
105 #define DEFAULT_TEMP_REMOVE TRUE
120 #define GST_DOWNLOAD_BUFFER_CLEAR_LEVEL(l) G_STMT_START { \
125 #define STATUS(elem, pad, msg) \
126 GST_LOG_OBJECT (elem, "(%s:%s) " msg ": %u of %u " \
127 "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
129 GST_DEBUG_PAD_NAME (pad), \
130 elem->cur_level.bytes, \
131 elem->max_level.bytes, \
132 elem->cur_level.time, \
133 elem->max_level.time)
135 #define GST_DOWNLOAD_BUFFER_MUTEX_LOCK(q) G_STMT_START { \
136 g_mutex_lock (&q->qlock); \
139 #define GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK(q,res,label) G_STMT_START { \
140 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (q); \
141 if (res != GST_FLOW_OK) \
145 #define GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK(q) G_STMT_START { \
146 g_mutex_unlock (&q->qlock); \
149 #define GST_DOWNLOAD_BUFFER_WAIT_ADD_CHECK(q, res, o, label) G_STMT_START { \
150 STATUS (q, q->srcpad, "wait for ADD"); \
151 q->waiting_add = TRUE; \
152 q->waiting_offset = o; \
153 g_cond_wait (&q->item_add, &q->qlock); \
154 q->waiting_add = FALSE; \
155 if (res != GST_FLOW_OK) { \
156 STATUS (q, q->srcpad, "received ADD wakeup"); \
159 STATUS (q, q->srcpad, "received ADD"); \
162 #define GST_DOWNLOAD_BUFFER_SIGNAL_ADD(q, o) G_STMT_START { \
163 if (q->waiting_add && q->waiting_offset <= o) { \
164 STATUS (q, q->sinkpad, "signal ADD"); \
165 g_cond_signal (&q->item_add); \
170 GST_DEBUG_CATEGORY_INIT (downloadbuffer_debug, "downloadbuffer", 0, \
171 "downloadbuffer element");
173 #define gst_download_buffer_parent_class parent_class
174 G_DEFINE_TYPE_WITH_CODE (GstDownloadBuffer, gst_download_buffer,
175 GST_TYPE_ELEMENT, _do_init);
177 static GstMessage *update_buffering (GstDownloadBuffer * dlbuf);
179 static void gst_download_buffer_finalize (GObject * object);
181 static void gst_download_buffer_set_property (GObject * object,
182 guint prop_id, const GValue * value, GParamSpec * pspec);
183 static void gst_download_buffer_get_property (GObject * object,
184 guint prop_id, GValue * value, GParamSpec * pspec);
186 static GstFlowReturn gst_download_buffer_chain (GstPad * pad,
187 GstObject * parent, GstBuffer * buffer);
188 static void gst_download_buffer_loop (GstPad * pad);
190 static gboolean gst_download_buffer_handle_sink_event (GstPad * pad,
191 GstObject * parent, GstEvent * event);
192 static gboolean gst_download_buffer_handle_sink_query (GstPad * pad,
193 GstObject * parent, GstQuery * query);
195 static gboolean gst_download_buffer_handle_src_event (GstPad * pad,
196 GstObject * parent, GstEvent * event);
197 static gboolean gst_download_buffer_handle_src_query (GstPad * pad,
198 GstObject * parent, GstQuery * query);
199 static gboolean gst_download_buffer_handle_query (GstElement * element,
202 static GstFlowReturn gst_download_buffer_get_range (GstPad * pad,
203 GstObject * parent, guint64 offset, guint length, GstBuffer ** buffer);
205 static gboolean gst_download_buffer_src_activate_mode (GstPad * pad,
206 GstObject * parent, GstPadMode mode, gboolean active);
207 static gboolean gst_download_buffer_sink_activate_mode (GstPad * pad,
208 GstObject * parent, GstPadMode mode, gboolean active);
209 static GstStateChangeReturn gst_download_buffer_change_state (GstElement *
210 element, GstStateChange transition);
212 /* static guint gst_download_buffer_signals[LAST_SIGNAL] = { 0 }; */
215 gst_download_buffer_class_init (GstDownloadBufferClass * klass)
217 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
218 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
220 gobject_class->set_property = gst_download_buffer_set_property;
221 gobject_class->get_property = gst_download_buffer_get_property;
224 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
225 g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
226 "Max. amount of data to buffer (bytes, 0=disable)",
227 0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
228 G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
229 G_PARAM_STATIC_STRINGS));
230 g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
231 g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
232 "Max. amount of data to buffer (in ns, 0=disable)", 0, G_MAXUINT64,
233 DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
234 G_PARAM_STATIC_STRINGS));
236 g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
237 g_param_spec_int ("low-percent", "Low percent",
238 "Low threshold for buffering to start. Only used if use-buffering is True",
239 0, 100, DEFAULT_LOW_PERCENT,
240 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
241 g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
242 g_param_spec_int ("high-percent", "High percent",
243 "High threshold for buffering to finish. Only used if use-buffering is True",
244 0, 100, DEFAULT_HIGH_PERCENT,
245 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
247 g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
248 g_param_spec_string ("temp-template", "Temporary File Template",
249 "File template to store temporary files in, should contain directory "
250 "and XXXXXX. (NULL == disabled)",
251 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
253 g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
254 g_param_spec_string ("temp-location", "Temporary File Location",
255 "Location to store temporary files in (Only read this property, "
256 "use temp-template to configure the name template)",
257 NULL, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
260 * GstDownloadBuffer:temp-remove
262 * When temp-template is set, remove the temporary file when going to READY.
264 g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
265 g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
266 "Remove the temp-location after use",
267 DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
269 /* set several parent class virtual functions */
270 gobject_class->finalize = gst_download_buffer_finalize;
272 gst_element_class_add_static_pad_template (gstelement_class, &srctemplate);
273 gst_element_class_add_static_pad_template (gstelement_class, &sinktemplate);
275 gst_element_class_set_static_metadata (gstelement_class, "DownloadBuffer",
276 "Generic", "Download Buffer element",
277 "Wim Taymans <wim.taymans@gmail.com>");
279 gstelement_class->change_state =
280 GST_DEBUG_FUNCPTR (gst_download_buffer_change_state);
281 gstelement_class->query =
282 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_query);
286 gst_download_buffer_init (GstDownloadBuffer * dlbuf)
288 dlbuf->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
290 gst_pad_set_chain_function (dlbuf->sinkpad,
291 GST_DEBUG_FUNCPTR (gst_download_buffer_chain));
292 gst_pad_set_activatemode_function (dlbuf->sinkpad,
293 GST_DEBUG_FUNCPTR (gst_download_buffer_sink_activate_mode));
294 gst_pad_set_event_function (dlbuf->sinkpad,
295 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_sink_event));
296 gst_pad_set_query_function (dlbuf->sinkpad,
297 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_sink_query));
298 GST_PAD_SET_PROXY_CAPS (dlbuf->sinkpad);
299 gst_element_add_pad (GST_ELEMENT (dlbuf), dlbuf->sinkpad);
301 dlbuf->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
303 gst_pad_set_activatemode_function (dlbuf->srcpad,
304 GST_DEBUG_FUNCPTR (gst_download_buffer_src_activate_mode));
305 gst_pad_set_getrange_function (dlbuf->srcpad,
306 GST_DEBUG_FUNCPTR (gst_download_buffer_get_range));
307 gst_pad_set_event_function (dlbuf->srcpad,
308 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_src_event));
309 gst_pad_set_query_function (dlbuf->srcpad,
310 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_src_query));
311 GST_PAD_SET_PROXY_CAPS (dlbuf->srcpad);
312 gst_element_add_pad (GST_ELEMENT (dlbuf), dlbuf->srcpad);
315 GST_DOWNLOAD_BUFFER_CLEAR_LEVEL (dlbuf->cur_level);
316 dlbuf->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;
317 dlbuf->max_level.time = DEFAULT_MAX_SIZE_TIME;
318 dlbuf->low_percent = DEFAULT_LOW_PERCENT;
319 dlbuf->high_percent = DEFAULT_HIGH_PERCENT;
321 dlbuf->srcresult = GST_FLOW_FLUSHING;
322 dlbuf->sinkresult = GST_FLOW_FLUSHING;
323 dlbuf->in_timer = g_timer_new ();
324 dlbuf->out_timer = g_timer_new ();
326 g_mutex_init (&dlbuf->qlock);
327 dlbuf->waiting_add = FALSE;
328 g_cond_init (&dlbuf->item_add);
330 /* tempfile related */
331 dlbuf->temp_template = NULL;
332 dlbuf->temp_location = NULL;
333 dlbuf->temp_remove = DEFAULT_TEMP_REMOVE;
336 /* called only once, as opposed to dispose */
338 gst_download_buffer_finalize (GObject * object)
340 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
342 g_mutex_clear (&dlbuf->qlock);
343 g_cond_clear (&dlbuf->item_add);
344 g_timer_destroy (dlbuf->in_timer);
345 g_timer_destroy (dlbuf->out_timer);
347 /* temp_file path cleanup */
348 g_free (dlbuf->temp_template);
349 g_free (dlbuf->temp_location);
351 G_OBJECT_CLASS (parent_class)->finalize (object);
355 reset_positions (GstDownloadBuffer * dlbuf)
357 dlbuf->write_pos = 0;
359 dlbuf->filling = TRUE;
360 dlbuf->buffering_percent = 0;
361 dlbuf->is_buffering = TRUE;
362 dlbuf->seeking = FALSE;
363 GST_DOWNLOAD_BUFFER_CLEAR_LEVEL (dlbuf->cur_level);
367 reset_rate_timer (GstDownloadBuffer * dlbuf)
370 dlbuf->bytes_out = 0;
371 dlbuf->byte_in_rate = 0.0;
372 dlbuf->byte_in_period = 0;
373 dlbuf->byte_out_rate = 0.0;
374 dlbuf->last_in_elapsed = 0.0;
375 dlbuf->last_out_elapsed = 0.0;
376 dlbuf->in_timer_started = FALSE;
377 dlbuf->out_timer_started = FALSE;
380 /* the interval in seconds to recalculate the rate */
381 #define RATE_INTERVAL 0.2
382 /* Tuning for rate estimation. We use a large window for the input rate because
383 * it should be stable when connected to a network. The output rate is less
384 * stable (the elements preroll, queues behind a demuxer fill, ...) and should
385 * therefore adapt more quickly.
386 * However, initial input rate may be subject to a burst, and should therefore
387 * initially also adapt more quickly to changes, and only later on give higher
388 * weight to previous values. */
389 #define AVG_IN(avg,val,w1,w2) ((avg) * (w1) + (val) * (w2)) / ((w1) + (w2))
390 #define AVG_OUT(avg,val) ((avg) * 3.0 + (val)) / 4.0
393 update_levels (GstDownloadBuffer * dlbuf, guint bytes)
395 dlbuf->cur_level.bytes = bytes;
397 if (dlbuf->byte_in_rate > 0.0) {
398 dlbuf->cur_level.time =
399 dlbuf->cur_level.bytes / dlbuf->byte_in_rate * GST_SECOND;
402 GST_DEBUG ("levels: bytes %u/%u, time %" GST_TIME_FORMAT "/%" GST_TIME_FORMAT,
403 dlbuf->cur_level.bytes, dlbuf->max_level.bytes,
404 GST_TIME_ARGS (dlbuf->cur_level.time),
405 GST_TIME_ARGS (dlbuf->max_level.time));
409 update_in_rates (GstDownloadBuffer * dlbuf)
411 gdouble elapsed, period;
412 gdouble byte_in_rate;
414 if (!dlbuf->in_timer_started) {
415 dlbuf->in_timer_started = TRUE;
416 g_timer_start (dlbuf->in_timer);
420 elapsed = g_timer_elapsed (dlbuf->in_timer, NULL);
422 /* recalc after each interval. */
423 if (dlbuf->last_in_elapsed + RATE_INTERVAL < elapsed) {
424 period = elapsed - dlbuf->last_in_elapsed;
426 GST_DEBUG_OBJECT (dlbuf,
427 "rates: period %f, in %" G_GUINT64_FORMAT ", global period %f",
428 period, dlbuf->bytes_in, dlbuf->byte_in_period);
430 byte_in_rate = dlbuf->bytes_in / period;
432 if (dlbuf->byte_in_rate == 0.0)
433 dlbuf->byte_in_rate = byte_in_rate;
435 dlbuf->byte_in_rate = AVG_IN (dlbuf->byte_in_rate, byte_in_rate,
436 (double) dlbuf->byte_in_period, period);
438 /* another data point, cap at 16 for long time running average */
439 if (dlbuf->byte_in_period < 16 * RATE_INTERVAL)
440 dlbuf->byte_in_period += period;
442 /* reset the values to calculate rate over the next interval */
443 dlbuf->last_in_elapsed = elapsed;
445 GST_DEBUG_OBJECT (dlbuf, "rates: in %f", dlbuf->byte_in_rate);
450 update_out_rates (GstDownloadBuffer * dlbuf)
452 gdouble elapsed, period;
453 gdouble byte_out_rate;
455 if (!dlbuf->out_timer_started) {
456 dlbuf->out_timer_started = TRUE;
457 g_timer_start (dlbuf->out_timer);
461 elapsed = g_timer_elapsed (dlbuf->out_timer, NULL);
463 /* recalc after each interval. */
464 if (dlbuf->last_out_elapsed + RATE_INTERVAL < elapsed) {
465 period = elapsed - dlbuf->last_out_elapsed;
467 GST_DEBUG_OBJECT (dlbuf,
468 "rates: period %f, out %" G_GUINT64_FORMAT, period, dlbuf->bytes_out);
470 byte_out_rate = dlbuf->bytes_out / period;
472 if (dlbuf->byte_out_rate == 0.0)
473 dlbuf->byte_out_rate = byte_out_rate;
475 dlbuf->byte_out_rate = AVG_OUT (dlbuf->byte_out_rate, byte_out_rate);
477 /* reset the values to calculate rate over the next interval */
478 dlbuf->last_out_elapsed = elapsed;
479 dlbuf->bytes_out = 0;
480 GST_DEBUG_OBJECT (dlbuf, "rates: out %f", dlbuf->byte_out_rate);
485 get_buffering_percent (GstDownloadBuffer * dlbuf, gboolean * is_buffering,
490 if (dlbuf->high_percent <= 0) {
494 *is_buffering = FALSE;
498 /* Ensure the variables used to calculate buffering state are up-to-date. */
499 update_in_rates (dlbuf);
500 update_out_rates (dlbuf);
502 /* figure out the percent we are filled, we take the max of all formats. */
503 if (dlbuf->max_level.bytes > 0) {
504 if (dlbuf->cur_level.bytes >= dlbuf->max_level.bytes)
507 perc = dlbuf->cur_level.bytes * 100 / dlbuf->max_level.bytes;
511 if (dlbuf->max_level.time > 0) {
512 if (dlbuf->cur_level.time >= dlbuf->max_level.time)
515 perc = MAX (perc, dlbuf->cur_level.time * 100 / dlbuf->max_level.time);
517 perc = MAX (0, perc);
520 *is_buffering = dlbuf->is_buffering;
522 /* scale to high percent so that it becomes the 100% mark */
523 perc = perc * 100 / dlbuf->high_percent;
531 GST_DEBUG_OBJECT (dlbuf, "buffering %d, percent %d", dlbuf->is_buffering,
538 get_buffering_stats (GstDownloadBuffer * dlbuf, gint percent,
539 GstBufferingMode * mode, gint * avg_in, gint * avg_out,
540 gint64 * buffering_left)
543 *mode = GST_BUFFERING_DOWNLOAD;
546 *avg_in = dlbuf->byte_in_rate;
548 *avg_out = dlbuf->byte_out_rate;
550 if (buffering_left) {
553 *buffering_left = (percent == 100 ? 0 : -1);
555 max = dlbuf->max_level.time;
556 cur = dlbuf->cur_level.time;
558 if (percent != 100 && max > cur)
559 *buffering_left = (max - cur) / 1000000;
564 update_buffering (GstDownloadBuffer * dlbuf)
567 gboolean post = FALSE;
568 GstMessage *message = NULL;
570 if (!get_buffering_percent (dlbuf, NULL, &percent))
573 if (dlbuf->is_buffering) {
575 /* if we were buffering see if we reached the high watermark */
576 if (percent >= dlbuf->high_percent)
577 dlbuf->is_buffering = FALSE;
579 /* we were not buffering, check if we need to start buffering if we drop
580 * below the low threshold */
581 if (percent < dlbuf->low_percent) {
582 dlbuf->is_buffering = TRUE;
588 if (percent == dlbuf->buffering_percent)
591 dlbuf->buffering_percent = percent;
595 GstBufferingMode mode;
596 gint avg_in, avg_out;
597 gint64 buffering_left;
599 get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
602 message = gst_message_new_buffering (GST_OBJECT_CAST (dlbuf),
604 gst_message_set_buffering_stats (message, mode,
605 avg_in, avg_out, buffering_left);
612 perform_seek_to_offset (GstDownloadBuffer * dlbuf, guint64 offset)
620 /* until we receive the FLUSH_STOP from this seek, we skip data */
621 dlbuf->seeking = TRUE;
622 dlbuf->write_pos = offset;
623 dlbuf->filling = FALSE;
624 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
626 GST_DEBUG_OBJECT (dlbuf, "Seeking to %" G_GUINT64_FORMAT, offset);
629 gst_event_new_seek (1.0, GST_FORMAT_BYTES,
630 GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
631 GST_SEEK_TYPE_NONE, -1);
633 res = gst_pad_push_event (dlbuf->sinkpad, event);
634 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
639 /* get the threshold for when we decide to seek rather than wait */
641 get_seek_threshold (GstDownloadBuffer * dlbuf)
645 /* FIXME, find a good threshold based on the incoming rate. */
646 threshold = 1024 * 512;
651 /* called with DOWNLOAD_BUFFER_MUTEX */
653 gst_download_buffer_update_upstream_size (GstDownloadBuffer * dlbuf)
655 gint64 upstream_size = 0;
657 if (gst_pad_peer_query_duration (dlbuf->sinkpad, GST_FORMAT_BYTES,
659 GST_INFO_OBJECT (dlbuf, "upstream size: %" G_GINT64_FORMAT, upstream_size);
660 dlbuf->upstream_size = upstream_size;
664 /* called with DOWNLOAD_BUFFER_MUTEX */
666 gst_download_buffer_wait_for_data (GstDownloadBuffer * dlbuf, guint64 offset,
673 GST_DEBUG_OBJECT (dlbuf, "wait for %" G_GUINT64_FORMAT ", length %u",
676 wanted = offset + length;
678 /* pause the timer while we wait. The fact that we are waiting does not mean
679 * the byterate on the output pad is lower */
680 if ((started = dlbuf->out_timer_started))
681 g_timer_stop (dlbuf->out_timer);
683 /* check range before us */
684 if (gst_sparse_file_get_range_before (dlbuf->file, offset, &start, &stop)) {
685 GST_DEBUG_OBJECT (dlbuf,
686 "range before %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT, start, stop);
687 if (start <= offset && offset < stop) {
688 GST_DEBUG_OBJECT (dlbuf, "we have the offset");
689 /* we have the range, continue it */
692 guint64 threshold, dist;
694 /* there is a range before us, check how far away it is */
695 threshold = get_seek_threshold (dlbuf);
696 dist = offset - stop;
698 if (dist <= threshold) {
699 GST_DEBUG_OBJECT (dlbuf, "not too far");
700 /* not far away, continue it */
706 if (dlbuf->write_pos != offset)
707 perform_seek_to_offset (dlbuf, offset);
709 dlbuf->filling = TRUE;
710 if (dlbuf->write_pos > dlbuf->read_pos)
711 update_levels (dlbuf, dlbuf->write_pos - dlbuf->read_pos);
713 update_levels (dlbuf, 0);
715 /* now wait for more data */
716 GST_DEBUG_OBJECT (dlbuf, "waiting for more data");
717 GST_DOWNLOAD_BUFFER_WAIT_ADD_CHECK (dlbuf, dlbuf->srcresult, wanted,
719 GST_DEBUG_OBJECT (dlbuf, "got more data");
721 /* and continue if we were running before */
723 g_timer_continue (dlbuf->out_timer);
729 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
730 return GST_FLOW_FLUSHING;
734 /* called with DOWNLOAD_BUFFER_MUTEX */
736 check_upstream_size (GstDownloadBuffer * dlbuf, gsize offset, guint * length)
738 gsize stop = offset + *length;
739 /* catch any reads beyond the size of the file here to make sure cache
740 * doesn't send seek events beyond the size of the file upstream, since
741 * that would confuse elements such as souphttpsrc and/or http servers.
742 * Demuxers often just loop until EOS at the end of the file to figure out
743 * when they've read all the end-headers or index chunks. */
744 if (G_UNLIKELY (dlbuf->upstream_size == -1 || stop >= dlbuf->upstream_size)) {
745 gst_download_buffer_update_upstream_size (dlbuf);
748 if (dlbuf->upstream_size != -1) {
749 if (offset >= dlbuf->upstream_size)
752 if (G_UNLIKELY (stop > dlbuf->upstream_size)) {
753 *length = dlbuf->upstream_size - offset;
754 GST_DEBUG_OBJECT (dlbuf, "adjusting length downto %u", *length);
760 /* called with DOWNLOAD_BUFFER_MUTEX */
762 gst_download_buffer_read_buffer (GstDownloadBuffer * dlbuf, guint64 offset,
763 guint length, GstBuffer ** buffer)
767 GstFlowReturn ret = GST_FLOW_OK;
768 gsize res, remaining;
769 GError *error = NULL;
771 length = (length == -1) ? DEFAULT_BUFFER_SIZE : length;
772 offset = (offset == -1) ? dlbuf->read_pos : offset;
774 if (!check_upstream_size (dlbuf, offset, &length))
777 /* allocate the output buffer of the requested size */
779 buf = gst_buffer_new_allocate (NULL, length, NULL);
783 gst_buffer_map (buf, &info, GST_MAP_WRITE);
785 GST_DEBUG_OBJECT (dlbuf, "Reading %u bytes from %" G_GUINT64_FORMAT, length,
788 dlbuf->read_pos = offset;
792 gst_sparse_file_read (dlbuf->file, offset, info.data, length,
794 if (G_UNLIKELY (res == 0)) {
795 switch (error->code) {
796 case GST_SPARSE_FILE_IO_ERROR_WOULD_BLOCK:
797 /* we don't have the requested data in the file, decide what to
799 ret = gst_download_buffer_wait_for_data (dlbuf, offset, length);
800 if (ret != GST_FLOW_OK)
806 g_clear_error (&error);
810 gst_buffer_unmap (buf, &info);
811 gst_buffer_resize (buf, 0, res);
813 dlbuf->bytes_out += res;
814 dlbuf->read_pos += res;
816 GST_DEBUG_OBJECT (dlbuf,
817 "Read %" G_GSIZE_FORMAT " bytes, remaining %" G_GSIZE_FORMAT, res,
820 if (dlbuf->read_pos + remaining == dlbuf->upstream_size)
821 update_levels (dlbuf, dlbuf->max_level.bytes);
823 update_levels (dlbuf, remaining);
825 GST_BUFFER_OFFSET (buf) = offset;
826 GST_BUFFER_OFFSET_END (buf) = offset + res;
835 GST_DEBUG_OBJECT (dlbuf, "EOS hit");
840 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
841 g_clear_error (&error);
842 gst_buffer_unmap (buf, &info);
844 gst_buffer_unref (buf);
845 return GST_FLOW_FLUSHING;
849 GST_DEBUG_OBJECT (dlbuf, "we have a read error: %s", error->message);
850 g_clear_error (&error);
851 gst_buffer_unmap (buf, &info);
853 gst_buffer_unref (buf);
858 /* must be called with MUTEX_LOCK. Will briefly release the lock when notifying
859 * the temp filename. */
861 gst_download_buffer_open_temp_location_file (GstDownloadBuffer * dlbuf)
869 GST_DEBUG_OBJECT (dlbuf, "opening temp file %s", dlbuf->temp_template);
871 /* If temp_template was set, allocate a filename and open that file */
874 if (dlbuf->temp_template == NULL)
877 /* make copy of the template, we don't want to change this */
878 name = g_strdup (dlbuf->temp_template);
880 fd = g_mkstemp_full (name, O_RDWR | O_LARGEFILE, S_IRUSR | S_IWUSR);
882 fd = g_mkstemp (name);
887 /* open the file for update/writing */
888 dlbuf->file = gst_sparse_file_new ();
889 /* error creating file */
890 if (!gst_sparse_file_set_fd (dlbuf->file, fd))
893 g_free (dlbuf->temp_location);
894 dlbuf->temp_location = name;
896 reset_positions (dlbuf);
898 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
900 /* we can't emit the notify with the lock */
901 g_object_notify (G_OBJECT (dlbuf), "temp-location");
903 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
905 GST_DEBUG_OBJECT (dlbuf, "opened temp file %s", dlbuf->temp_template);
912 GST_DEBUG_OBJECT (dlbuf, "temp file was already open");
917 GST_ELEMENT_ERROR (dlbuf, RESOURCE, NOT_FOUND,
918 (_("No Temp directory specified.")), (NULL));
923 GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
924 (_("Could not create temp file \"%s\"."), dlbuf->temp_template),
931 GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
932 (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
941 gst_download_buffer_close_temp_location_file (GstDownloadBuffer * dlbuf)
944 if (dlbuf->file == NULL)
947 GST_DEBUG_OBJECT (dlbuf, "closing sparse file");
949 if (dlbuf->temp_remove) {
950 if (remove (dlbuf->temp_location) < 0) {
951 GST_WARNING_OBJECT (dlbuf, "Failed to remove temporary file %s: %s",
952 dlbuf->temp_location, g_strerror (errno));
955 gst_sparse_file_free (dlbuf->file);
956 close (dlbuf->temp_fd);
961 gst_download_buffer_flush_temp_file (GstDownloadBuffer * dlbuf)
963 if (dlbuf->file == NULL)
966 GST_DEBUG_OBJECT (dlbuf, "flushing temp file");
968 gst_sparse_file_clear (dlbuf->file);
972 gst_download_buffer_locked_flush (GstDownloadBuffer * dlbuf, gboolean full,
976 gst_download_buffer_flush_temp_file (dlbuf);
977 reset_positions (dlbuf);
978 gst_event_replace (&dlbuf->stream_start_event, NULL);
979 gst_event_replace (&dlbuf->segment_event, NULL);
983 gst_download_buffer_handle_sink_event (GstPad * pad, GstObject * parent,
987 GstDownloadBuffer *dlbuf;
989 dlbuf = GST_DOWNLOAD_BUFFER (parent);
991 switch (GST_EVENT_TYPE (event)) {
992 case GST_EVENT_FLUSH_START:
994 GST_LOG_OBJECT (dlbuf, "received flush start event");
995 if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
997 ret = gst_pad_push_event (dlbuf->srcpad, event);
999 /* now unblock the chain function */
1000 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1001 dlbuf->srcresult = GST_FLOW_FLUSHING;
1002 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1003 /* unblock the loop and chain functions */
1004 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1005 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1007 /* make sure it pauses, this should happen since we sent
1008 * flush_start downstream. */
1009 gst_pad_pause_task (dlbuf->srcpad);
1010 GST_LOG_OBJECT (dlbuf, "loop stopped");
1012 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1013 /* flush the sink pad */
1014 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1015 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1017 gst_event_unref (event);
1021 case GST_EVENT_FLUSH_STOP:
1023 GST_LOG_OBJECT (dlbuf, "received flush stop event");
1025 if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
1027 ret = gst_pad_push_event (dlbuf->srcpad, event);
1029 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1030 gst_download_buffer_locked_flush (dlbuf, FALSE, TRUE);
1031 dlbuf->srcresult = GST_FLOW_OK;
1032 dlbuf->sinkresult = GST_FLOW_OK;
1033 dlbuf->unexpected = FALSE;
1034 dlbuf->seeking = FALSE;
1035 /* reset rate counters */
1036 reset_rate_timer (dlbuf);
1037 gst_pad_start_task (dlbuf->srcpad,
1038 (GstTaskFunction) gst_download_buffer_loop, dlbuf->srcpad, NULL);
1039 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1041 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1042 dlbuf->unexpected = FALSE;
1043 dlbuf->sinkresult = GST_FLOW_OK;
1044 dlbuf->seeking = FALSE;
1045 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1047 gst_event_unref (event);
1052 if (GST_EVENT_IS_SERIALIZED (event)) {
1053 GstMessage *msg = NULL;
1055 /* serialized events go in the buffer */
1056 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult,
1058 switch (GST_EVENT_TYPE (event)) {
1060 GST_DEBUG_OBJECT (dlbuf, "we have EOS");
1061 /* Zero the thresholds, this makes sure the dlbuf is completely
1062 * filled and we can read all data from the dlbuf. */
1063 /* update the buffering status */
1064 update_levels (dlbuf, dlbuf->max_level.bytes);
1065 /* update the buffering */
1066 msg = update_buffering (dlbuf);
1067 /* wakeup the waiter and let it recheck */
1068 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1070 case GST_EVENT_SEGMENT:
1071 gst_event_replace (&dlbuf->segment_event, event);
1072 /* a new segment allows us to accept more buffers if we got EOS
1073 * from downstream */
1074 dlbuf->unexpected = FALSE;
1076 case GST_EVENT_STREAM_START:
1077 gst_event_replace (&dlbuf->stream_start_event, event);
1082 gst_event_unref (event);
1083 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1085 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1087 /* non-serialized events are passed upstream. */
1088 ret = gst_pad_push_event (dlbuf->srcpad, event);
1097 GST_DEBUG_OBJECT (dlbuf, "refusing event, we are flushing");
1098 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1099 gst_event_unref (event);
1105 gst_download_buffer_handle_sink_query (GstPad * pad, GstObject * parent,
1108 GstDownloadBuffer *dlbuf;
1111 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1113 switch (GST_QUERY_TYPE (query)) {
1115 if (GST_QUERY_IS_SERIALIZED (query)) {
1116 GST_LOG_OBJECT (dlbuf, "received query %p", query);
1117 GST_DEBUG_OBJECT (dlbuf, "refusing query, we are not using the dlbuf");
1120 res = gst_pad_query_default (pad, parent, query);
1127 static GstFlowReturn
1128 gst_download_buffer_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
1130 GstDownloadBuffer *dlbuf;
1133 gsize res, available;
1134 GError *error = NULL;
1135 GstMessage *msg = NULL;
1137 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1139 GST_LOG_OBJECT (dlbuf, "received buffer %p of "
1140 "size %" G_GSIZE_FORMAT ", time %" GST_TIME_FORMAT ", duration %"
1141 GST_TIME_FORMAT, buffer, gst_buffer_get_size (buffer),
1142 GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1143 GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1145 /* we have to lock the dlbuf since we span threads */
1146 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult, out_flushing);
1147 /* when we received unexpected from downstream, refuse more buffers */
1148 if (dlbuf->unexpected)
1151 /* while we didn't receive the newsegment, we're seeking and we skip data */
1155 /* put buffer in dlbuf now */
1156 offset = dlbuf->write_pos;
1159 if (GST_BUFFER_OFFSET_IS_VALID (buffer) &&
1160 GST_BUFFER_OFFSET (buffer) != offset) {
1161 GST_WARNING_OBJECT (dlbuf, "buffer offset does not match current writing "
1162 "position! %" G_GINT64_FORMAT " != %" G_GINT64_FORMAT,
1163 GST_BUFFER_OFFSET (buffer), offset);
1166 gst_buffer_map (buffer, &info, GST_MAP_READ);
1168 GST_DEBUG_OBJECT (dlbuf, "Writing %" G_GSIZE_FORMAT " bytes to %"
1169 G_GUINT64_FORMAT, info.size, offset);
1172 gst_sparse_file_write (dlbuf->file, offset, info.data, info.size,
1173 &available, &error);
1177 gst_buffer_unmap (buffer, &info);
1178 gst_buffer_unref (buffer);
1180 dlbuf->write_pos = offset + info.size;
1181 dlbuf->bytes_in += info.size;
1183 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, dlbuf->write_pos + available);
1185 /* we hit the end, see what to do */
1186 if (dlbuf->write_pos + available == dlbuf->upstream_size) {
1189 /* we have everything up to the end, find a region to fill */
1190 if (gst_sparse_file_get_range_after (dlbuf->file, 0, &start, &stop)) {
1191 if (stop < dlbuf->upstream_size) {
1192 /* a hole to fill, seek to its end */
1193 perform_seek_to_offset (dlbuf, stop);
1195 /* we filled all the holes, we are done */
1200 /* see if we need to skip this region or just read it again. The idea
1201 * is that when the region is not big, we want to avoid a seek and just
1203 guint64 threshold = get_seek_threshold (dlbuf);
1205 if (available > threshold) {
1206 /* further than threshold, it's better to skip than to reread */
1207 perform_seek_to_offset (dlbuf, dlbuf->write_pos + available);
1210 if (dlbuf->filling) {
1211 if (dlbuf->write_pos > dlbuf->read_pos)
1212 update_levels (dlbuf, dlbuf->write_pos - dlbuf->read_pos);
1214 update_levels (dlbuf, 0);
1217 /* update the buffering */
1218 msg = update_buffering (dlbuf);
1220 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1223 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1230 GstFlowReturn ret = dlbuf->sinkresult;
1231 GST_LOG_OBJECT (dlbuf,
1232 "exit because task paused, reason: %s", gst_flow_get_name (ret));
1233 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1234 gst_buffer_unref (buffer);
1239 GST_LOG_OBJECT (dlbuf, "exit because we received EOS");
1240 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1241 gst_buffer_unref (buffer);
1242 return GST_FLOW_EOS;
1246 GST_LOG_OBJECT (dlbuf, "exit because we are seeking");
1247 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1248 gst_buffer_unref (buffer);
1253 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1254 gst_buffer_unmap (buffer, &info);
1255 gst_buffer_unref (buffer);
1256 GST_ELEMENT_ERROR (dlbuf, RESOURCE, WRITE,
1257 (_("Error while writing to download file.")), ("%s", error->message));
1258 g_clear_error (&error);
1259 return GST_FLOW_ERROR;
1263 GST_LOG_OBJECT (dlbuf, "we completed the download");
1264 dlbuf->write_pos = dlbuf->upstream_size;
1265 dlbuf->filling = FALSE;
1266 update_levels (dlbuf, dlbuf->max_level.bytes);
1267 msg = update_buffering (dlbuf);
1269 gst_element_post_message (GST_ELEMENT_CAST (dlbuf),
1270 gst_message_new_element (GST_OBJECT_CAST (dlbuf),
1271 gst_structure_new ("GstCacheDownloadComplete",
1272 "location", G_TYPE_STRING, dlbuf->temp_location, NULL)));
1274 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1277 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1279 return GST_FLOW_EOS;
1283 /* called repeatedly with @pad as the source pad. This function should push out
1284 * data to the peer element. */
1286 gst_download_buffer_loop (GstPad * pad)
1288 GstDownloadBuffer *dlbuf;
1290 GstBuffer *buffer = NULL;
1291 GstMessage *msg = NULL;
1293 dlbuf = GST_DOWNLOAD_BUFFER (GST_PAD_PARENT (pad));
1295 /* have to lock for thread-safety */
1296 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1298 if (dlbuf->stream_start_event != NULL) {
1299 gst_pad_push_event (dlbuf->srcpad, dlbuf->stream_start_event);
1300 dlbuf->stream_start_event = NULL;
1302 if (dlbuf->segment_event != NULL) {
1303 gst_pad_push_event (dlbuf->srcpad, dlbuf->segment_event);
1304 dlbuf->segment_event = NULL;
1307 ret = gst_download_buffer_read_buffer (dlbuf, -1, -1, &buffer);
1308 if (ret != GST_FLOW_OK)
1311 /* update the buffering */
1312 msg = update_buffering (dlbuf);
1314 g_atomic_int_set (&dlbuf->downstream_may_block, 1);
1315 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1318 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1320 ret = gst_pad_push (dlbuf->srcpad, buffer);
1321 g_atomic_int_set (&dlbuf->downstream_may_block, 0);
1323 /* need to check for srcresult here as well */
1324 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1325 dlbuf->srcresult = ret;
1326 dlbuf->sinkresult = ret;
1327 if (ret != GST_FLOW_OK)
1329 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1336 GstFlowReturn ret = dlbuf->srcresult;
1338 gst_pad_pause_task (dlbuf->srcpad);
1339 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1340 GST_LOG_OBJECT (dlbuf, "pause task, reason: %s", gst_flow_get_name (ret));
1341 /* let app know about us giving up if upstream is not expected to do so */
1342 if (ret == GST_FLOW_EOS) {
1343 /* FIXME perform EOS logic, this is really a basesrc operating on a
1345 gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1346 } else if ((ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS)) {
1347 GST_ELEMENT_FLOW_ERROR (dlbuf, ret);
1348 gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1355 gst_download_buffer_handle_src_event (GstPad * pad, GstObject * parent,
1358 gboolean res = TRUE;
1359 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (parent);
1361 #ifndef GST_DISABLE_GST_DEBUG
1362 GST_DEBUG_OBJECT (dlbuf, "got event %p (%s)",
1363 event, GST_EVENT_TYPE_NAME (event));
1366 switch (GST_EVENT_TYPE (event)) {
1367 case GST_EVENT_FLUSH_START:
1368 /* now unblock the getrange function */
1369 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1370 GST_DEBUG_OBJECT (dlbuf, "flushing");
1371 dlbuf->srcresult = GST_FLOW_FLUSHING;
1372 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1373 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1375 /* when using a temp file, we eat the event */
1377 gst_event_unref (event);
1379 case GST_EVENT_FLUSH_STOP:
1380 /* now unblock the getrange function */
1381 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1382 dlbuf->srcresult = GST_FLOW_OK;
1383 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1385 /* when using a temp file, we eat the event */
1387 gst_event_unref (event);
1389 case GST_EVENT_RECONFIGURE:
1390 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1391 /* assume downstream is linked now and try to push again */
1392 if (dlbuf->srcresult == GST_FLOW_NOT_LINKED) {
1393 dlbuf->srcresult = GST_FLOW_OK;
1394 dlbuf->sinkresult = GST_FLOW_OK;
1395 if (GST_PAD_MODE (pad) == GST_PAD_MODE_PUSH) {
1396 gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1400 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1402 res = gst_pad_push_event (dlbuf->sinkpad, event);
1405 res = gst_pad_push_event (dlbuf->sinkpad, event);
1413 gst_download_buffer_handle_src_query (GstPad * pad, GstObject * parent,
1416 GstDownloadBuffer *dlbuf;
1418 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1420 switch (GST_QUERY_TYPE (query)) {
1421 case GST_QUERY_POSITION:
1426 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1429 /* get peer position */
1430 gst_query_parse_position (query, &format, &peer_pos);
1432 /* FIXME: this code assumes that there's no discont in the dlbuf */
1434 case GST_FORMAT_BYTES:
1435 peer_pos -= dlbuf->cur_level.bytes;
1436 if (peer_pos < 0) /* Clamp result to 0 */
1439 case GST_FORMAT_TIME:
1440 peer_pos -= dlbuf->cur_level.time;
1441 if (peer_pos < 0) /* Clamp result to 0 */
1445 GST_WARNING_OBJECT (dlbuf, "dropping query in %s format, don't "
1446 "know how to adjust value", gst_format_get_name (format));
1449 /* set updated position */
1450 gst_query_set_position (query, format, peer_pos);
1453 case GST_QUERY_DURATION:
1455 GST_DEBUG_OBJECT (dlbuf, "doing peer query");
1457 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1460 GST_DEBUG_OBJECT (dlbuf, "peer query success");
1463 case GST_QUERY_BUFFERING:
1466 gboolean is_buffering;
1467 GstBufferingMode mode;
1468 gint avg_in, avg_out;
1469 gint64 buffering_left;
1471 GST_DEBUG_OBJECT (dlbuf, "query buffering");
1473 get_buffering_percent (dlbuf, &is_buffering, &percent);
1474 gst_query_set_buffering_percent (query, is_buffering, percent);
1476 get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
1478 gst_query_set_buffering_stats (query, mode, avg_in, avg_out,
1482 /* add ranges for download and ringbuffer buffering */
1486 gint64 estimated_total;
1488 gsize offset, range_start, range_stop;
1490 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1491 write_pos = dlbuf->write_pos;
1493 /* get duration of upstream in bytes */
1494 gst_download_buffer_update_upstream_size (dlbuf);
1495 duration = dlbuf->upstream_size;
1497 GST_DEBUG_OBJECT (dlbuf, "percent %d, duration %" G_GINT64_FORMAT
1498 ", writing %" G_GINT64_FORMAT, percent, duration, write_pos);
1500 gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1502 /* fill out the buffered ranges */
1505 estimated_total = -1;
1506 while (gst_sparse_file_get_range_after (dlbuf->file, offset,
1507 &range_start, &range_stop)) {
1508 gboolean current_range;
1510 GST_DEBUG_OBJECT (dlbuf,
1511 "range starting at %" G_GSIZE_FORMAT " and finishing at %"
1512 G_GSIZE_FORMAT, range_start, range_stop);
1514 offset = range_stop;
1516 /* find the range we are currently downloading, we'll remember it
1517 * after we convert to the target format */
1518 if (range_start <= write_pos && range_stop >= write_pos) {
1519 current_range = TRUE;
1520 /* calculate remaining and total download time */
1521 if (duration >= range_stop && avg_in > 0.0)
1522 estimated_total = ((duration - range_stop) * 1000) / avg_in;
1524 current_range = FALSE;
1527 case GST_FORMAT_PERCENT:
1528 /* get our available data relative to the duration */
1529 if (duration == -1) {
1533 range_start = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1534 range_start, duration);
1535 range_stop = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1536 range_stop, duration);
1539 case GST_FORMAT_BYTES:
1547 if (current_range) {
1548 /* we are currently downloading this range */
1549 start = range_start;
1552 GST_DEBUG_OBJECT (dlbuf,
1553 "range to format: %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT,
1554 range_start, range_stop);
1555 if (range_start == range_stop)
1557 gst_query_add_buffering_range (query, range_start, range_stop);
1560 GST_DEBUG_OBJECT (dlbuf, "estimated-total %" G_GINT64_FORMAT,
1563 gst_query_set_buffering_range (query, format, start, stop,
1566 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1570 case GST_QUERY_SCHEDULING:
1572 GstSchedulingFlags flags = 0;
1574 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1577 gst_query_parse_scheduling (query, &flags, NULL, NULL, NULL);
1579 /* we can operate in pull mode when we are using a tempfile */
1580 flags |= GST_SCHEDULING_FLAG_SEEKABLE;
1581 gst_query_set_scheduling (query, flags, 0, -1, 0);
1582 gst_query_add_scheduling_mode (query, GST_PAD_MODE_PULL);
1583 gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
1587 /* peer handled other queries */
1588 if (!gst_pad_query_default (pad, parent, query))
1598 GST_DEBUG_OBJECT (dlbuf, "failed peer query");
1604 gst_download_buffer_handle_query (GstElement * element, GstQuery * query)
1606 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (element);
1608 /* simply forward to the srcpad query function */
1609 return gst_download_buffer_handle_src_query (dlbuf->srcpad,
1610 GST_OBJECT_CAST (element), query);
1613 static GstFlowReturn
1614 gst_download_buffer_get_range (GstPad * pad, GstObject * parent, guint64 offset,
1615 guint length, GstBuffer ** buffer)
1617 GstDownloadBuffer *dlbuf;
1619 GstMessage *msg = NULL;
1621 dlbuf = GST_DOWNLOAD_BUFFER_CAST (parent);
1623 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1624 /* FIXME - function will block when the range is not yet available */
1625 ret = gst_download_buffer_read_buffer (dlbuf, offset, length, buffer);
1626 /* update the buffering */
1627 msg = update_buffering (dlbuf);
1629 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1632 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1639 ret = dlbuf->srcresult;
1641 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
1642 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1647 /* sink currently only operates in push mode */
1649 gst_download_buffer_sink_activate_mode (GstPad * pad, GstObject * parent,
1650 GstPadMode mode, gboolean active)
1653 GstDownloadBuffer *dlbuf;
1655 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1658 case GST_PAD_MODE_PUSH:
1660 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1661 GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1662 dlbuf->srcresult = GST_FLOW_OK;
1663 dlbuf->sinkresult = GST_FLOW_OK;
1664 dlbuf->unexpected = FALSE;
1665 reset_rate_timer (dlbuf);
1666 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1668 /* unblock chain function */
1669 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1670 GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1671 dlbuf->srcresult = GST_FLOW_FLUSHING;
1672 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1673 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1675 /* wait until it is unblocked and clean up */
1676 GST_PAD_STREAM_LOCK (pad);
1677 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1678 gst_download_buffer_locked_flush (dlbuf, TRUE, FALSE);
1679 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1680 GST_PAD_STREAM_UNLOCK (pad);
1691 /* src operating in push mode, we start a task on the source pad that pushes out
1692 * buffers from the dlbuf */
1694 gst_download_buffer_src_activate_push (GstPad * pad, GstObject * parent,
1697 gboolean result = FALSE;
1698 GstDownloadBuffer *dlbuf;
1700 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1703 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1704 GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1705 dlbuf->srcresult = GST_FLOW_OK;
1706 dlbuf->sinkresult = GST_FLOW_OK;
1707 dlbuf->unexpected = FALSE;
1709 gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1711 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1713 /* unblock loop function */
1714 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1715 GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1716 dlbuf->srcresult = GST_FLOW_FLUSHING;
1717 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1718 /* the item add signal will unblock */
1719 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1720 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1722 /* step 2, make sure streaming finishes */
1723 result = gst_pad_stop_task (pad);
1729 /* pull mode, downstream will call our getrange function */
1731 gst_download_buffer_src_activate_pull (GstPad * pad, GstObject * parent,
1735 GstDownloadBuffer *dlbuf;
1737 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1740 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1741 /* open the temp file now */
1742 result = gst_download_buffer_open_temp_location_file (dlbuf);
1743 GST_DEBUG_OBJECT (dlbuf, "activating pull mode");
1744 dlbuf->srcresult = GST_FLOW_OK;
1745 dlbuf->sinkresult = GST_FLOW_OK;
1746 dlbuf->unexpected = FALSE;
1747 dlbuf->upstream_size = 0;
1748 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1750 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1751 GST_DEBUG_OBJECT (dlbuf, "deactivating pull mode");
1752 dlbuf->srcresult = GST_FLOW_FLUSHING;
1753 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1754 /* this will unlock getrange */
1755 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1757 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1764 gst_download_buffer_src_activate_mode (GstPad * pad, GstObject * parent,
1765 GstPadMode mode, gboolean active)
1770 case GST_PAD_MODE_PULL:
1771 res = gst_download_buffer_src_activate_pull (pad, parent, active);
1773 case GST_PAD_MODE_PUSH:
1774 res = gst_download_buffer_src_activate_push (pad, parent, active);
1777 GST_LOG_OBJECT (pad, "unknown activation mode %d", mode);
1784 static GstStateChangeReturn
1785 gst_download_buffer_change_state (GstElement * element,
1786 GstStateChange transition)
1788 GstDownloadBuffer *dlbuf;
1789 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1791 dlbuf = GST_DOWNLOAD_BUFFER (element);
1793 switch (transition) {
1794 case GST_STATE_CHANGE_NULL_TO_READY:
1796 case GST_STATE_CHANGE_READY_TO_PAUSED:
1797 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1798 if (!gst_download_buffer_open_temp_location_file (dlbuf))
1799 ret = GST_STATE_CHANGE_FAILURE;
1800 gst_event_replace (&dlbuf->stream_start_event, NULL);
1801 gst_event_replace (&dlbuf->segment_event, NULL);
1802 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1804 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1810 if (ret == GST_STATE_CHANGE_FAILURE)
1813 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1815 if (ret == GST_STATE_CHANGE_FAILURE)
1818 switch (transition) {
1819 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1821 case GST_STATE_CHANGE_PAUSED_TO_READY:
1822 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1823 gst_download_buffer_close_temp_location_file (dlbuf);
1824 gst_event_replace (&dlbuf->stream_start_event, NULL);
1825 gst_event_replace (&dlbuf->segment_event, NULL);
1826 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1828 case GST_STATE_CHANGE_READY_TO_NULL:
1837 #define CAPACITY_CHANGE(elem) \
1838 update_buffering (elem);
1841 gst_download_buffer_set_temp_template (GstDownloadBuffer * dlbuf,
1842 const gchar * template)
1846 /* the element must be stopped in order to do this */
1847 GST_OBJECT_LOCK (dlbuf);
1848 state = GST_STATE (dlbuf);
1849 if (state != GST_STATE_READY && state != GST_STATE_NULL)
1851 GST_OBJECT_UNLOCK (dlbuf);
1853 /* set new location */
1854 g_free (dlbuf->temp_template);
1855 dlbuf->temp_template = g_strdup (template);
1862 GST_WARNING_OBJECT (dlbuf, "setting temp-template property in wrong state");
1863 GST_OBJECT_UNLOCK (dlbuf);
1868 gst_download_buffer_set_property (GObject * object,
1869 guint prop_id, const GValue * value, GParamSpec * pspec)
1871 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1872 GstMessage *msg = NULL;
1874 /* someone could change levels here, and since this
1875 * affects the get/put funcs, we need to lock for safety. */
1876 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1879 case PROP_MAX_SIZE_BYTES:
1880 dlbuf->max_level.bytes = g_value_get_uint (value);
1881 msg = CAPACITY_CHANGE (dlbuf);
1883 case PROP_MAX_SIZE_TIME:
1884 dlbuf->max_level.time = g_value_get_uint64 (value);
1885 msg = CAPACITY_CHANGE (dlbuf);
1887 case PROP_LOW_PERCENT:
1888 dlbuf->low_percent = g_value_get_int (value);
1890 case PROP_HIGH_PERCENT:
1891 dlbuf->high_percent = g_value_get_int (value);
1893 case PROP_TEMP_TEMPLATE:
1894 gst_download_buffer_set_temp_template (dlbuf, g_value_get_string (value));
1896 case PROP_TEMP_REMOVE:
1897 dlbuf->temp_remove = g_value_get_boolean (value);
1900 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1904 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1907 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1912 gst_download_buffer_get_property (GObject * object,
1913 guint prop_id, GValue * value, GParamSpec * pspec)
1915 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1917 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1920 case PROP_MAX_SIZE_BYTES:
1921 g_value_set_uint (value, dlbuf->max_level.bytes);
1923 case PROP_MAX_SIZE_TIME:
1924 g_value_set_uint64 (value, dlbuf->max_level.time);
1926 case PROP_LOW_PERCENT:
1927 g_value_set_int (value, dlbuf->low_percent);
1929 case PROP_HIGH_PERCENT:
1930 g_value_set_int (value, dlbuf->high_percent);
1932 case PROP_TEMP_TEMPLATE:
1933 g_value_set_string (value, dlbuf->temp_template);
1935 case PROP_TEMP_LOCATION:
1936 g_value_set_string (value, dlbuf->temp_location);
1938 case PROP_TEMP_REMOVE:
1939 g_value_set_boolean (value, dlbuf->temp_remove);
1942 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1946 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);