1 /* GStreamer
2 * Copyright (C) 2014 Wim Taymans <wim.taymans@gmail.com>
3 *
4 * gstdownloadbuffer.c:
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Library General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Library General Public License for more details.
15 *
16 * You should have received a copy of the GNU Library General Public
17 * License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
20 */
21
22 /**
23 * SECTION:element-downloadbuffer
24 * @title: downloadbuffer
25 *
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.
30 *
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
35 * reached again.
36 *
37 * The downloadbuffer provides push and pull based scheduling on its source pad
38 * and will efficiently seek in the upstream element when needed.
39 *
40 * The temp-location property will be used to notify the application of the
41 * allocated filename.
42 *
43 * When the downloadbuffer has completely downloaded the media, it will
44 * post an application message named `GstCacheDownloadComplete`
45 * with the following information:
46 *
47 * * G_TYPE_STRING `location`: the location of the completely downloaded file.
48 *
49 */
50
51 #ifdef HAVE_CONFIG_H
52 #include "config.h"
53 #endif
54
55 #include "gstdownloadbuffer.h"
56 #include "gstcoreelementselements.h"
57
58 #include <glib/gstdio.h>
59
60 #include "gst/gst-i18n-lib.h"
61 #include "gst/glib-compat-private.h"
62
63 #include <string.h>
64
65 #ifdef G_OS_WIN32
66 #include <io.h> /* lseek, open, close, read */
67 #undef lseek
68 #define lseek _lseeki64
69 #else
70 #include <unistd.h>
71 #endif
72
73 #ifdef __BIONIC__
74 #include <fcntl.h>
75 #endif
76
77 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
78 GST_PAD_SINK,
79 GST_PAD_ALWAYS,
80 GST_STATIC_CAPS_ANY);
81
82 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
83 GST_PAD_SRC,
84 GST_PAD_ALWAYS,
85 GST_STATIC_CAPS_ANY);
86
87 GST_DEBUG_CATEGORY_STATIC (downloadbuffer_debug);
88 #define GST_CAT_DEFAULT (downloadbuffer_debug)
89
90 enum
91 {
92 LAST_SIGNAL
93 };
94
95 /* other defines */
96 #define DEFAULT_BUFFER_SIZE 4096
97
98 /* default property values */
99 #define DEFAULT_MAX_SIZE_BYTES (2 * 1024 * 1024) /* 2 MB */
100 #define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND /* 2 seconds */
101 #define DEFAULT_LOW_PERCENT 10
102 #define DEFAULT_HIGH_PERCENT 99
103 #define DEFAULT_TEMP_REMOVE TRUE
104
105 enum
106 {
107 PROP_0,
108 PROP_MAX_SIZE_BYTES,
109 PROP_MAX_SIZE_TIME,
110 PROP_LOW_PERCENT,
111 PROP_HIGH_PERCENT,
112 PROP_TEMP_TEMPLATE,
113 PROP_TEMP_LOCATION,
114 PROP_TEMP_REMOVE,
115 PROP_LAST
116 };
117
118 #define GST_DOWNLOAD_BUFFER_CLEAR_LEVEL(l) G_STMT_START { \
119 l.bytes = 0; \
120 l.time = 0; \
121 } G_STMT_END
122
123 #define STATUS(elem, pad, msg) \
124 GST_LOG_OBJECT (elem, "(%s:%s) " msg ": %u of %u " \
125 "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
126 " ns", \
127 GST_DEBUG_PAD_NAME (pad), \
128 elem->cur_level.bytes, \
129 elem->max_level.bytes, \
130 elem->cur_level.time, \
131 elem->max_level.time)
132
133 #define GST_DOWNLOAD_BUFFER_MUTEX_LOCK(q) G_STMT_START { \
134 g_mutex_lock (&q->qlock); \
135 } G_STMT_END
136
137 #define GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK(q,res,label) G_STMT_START { \
138 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (q); \
139 if (res != GST_FLOW_OK) \
140 goto label; \
141 } G_STMT_END
142
143 #define GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK(q) G_STMT_START { \
144 g_mutex_unlock (&q->qlock); \
145 } G_STMT_END
146
147 #define GST_DOWNLOAD_BUFFER_WAIT_ADD_CHECK(q, res, o, label) G_STMT_START { \
148 STATUS (q, q->srcpad, "wait for ADD"); \
149 q->waiting_add = TRUE; \
150 q->waiting_offset = o; \
151 g_cond_wait (&q->item_add, &q->qlock); \
152 q->waiting_add = FALSE; \
153 if (res != GST_FLOW_OK) { \
154 STATUS (q, q->srcpad, "received ADD wakeup"); \
155 goto label; \
156 } \
157 STATUS (q, q->srcpad, "received ADD"); \
158 } G_STMT_END
159
160 #define GST_DOWNLOAD_BUFFER_SIGNAL_ADD(q, o) G_STMT_START { \
161 if (q->waiting_add && q->waiting_offset <= o) { \
162 STATUS (q, q->sinkpad, "signal ADD"); \
163 g_cond_signal (&q->item_add); \
164 } \
165 } G_STMT_END
166
167 #define _do_init \
168 GST_DEBUG_CATEGORY_INIT (downloadbuffer_debug, "downloadbuffer", 0, \
169 "downloadbuffer element");
170
171 #define gst_download_buffer_parent_class parent_class
172 G_DEFINE_TYPE_WITH_CODE (GstDownloadBuffer, gst_download_buffer,
173 GST_TYPE_ELEMENT, _do_init);
174 GST_ELEMENT_REGISTER_DEFINE (downloadbuffer, "downloadbuffer", GST_RANK_NONE,
175 GST_TYPE_DOWNLOAD_BUFFER);
176
177 static GstMessage *update_buffering (GstDownloadBuffer * dlbuf);
178
179 static void gst_download_buffer_finalize (GObject * object);
180
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);
185
186 static GstFlowReturn gst_download_buffer_chain (GstPad * pad,
187 GstObject * parent, GstBuffer * buffer);
188 static void gst_download_buffer_loop (GstPad * pad);
189
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);
194
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,
200 GstQuery * query);
201
202 static GstFlowReturn gst_download_buffer_get_range (GstPad * pad,
203 GstObject * parent, guint64 offset, guint length, GstBuffer ** buffer);
204
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);
211
212 /* static guint gst_download_buffer_signals[LAST_SIGNAL] = { 0 }; */
213
214 static void
gst_download_buffer_class_init(GstDownloadBufferClass * klass)215 gst_download_buffer_class_init (GstDownloadBufferClass * klass)
216 {
217 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
218 GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
219
220 gobject_class->set_property = gst_download_buffer_set_property;
221 gobject_class->get_property = gst_download_buffer_get_property;
222
223 /* properties */
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));
235
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. "
239 "Emits GST_MESSAGE_BUFFERING with a value of 0%",
240 0, 100, DEFAULT_LOW_PERCENT,
241 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
242 g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
243 g_param_spec_int ("high-percent", "High percent",
244 "High threshold for buffering to finish. "
245 "Emits GST_MESSAGE_BUFFERING with a value of 100%",
246 0, 100, DEFAULT_HIGH_PERCENT,
247 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
248
249 g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
250 g_param_spec_string ("temp-template", "Temporary File Template",
251 "File template to store temporary files in, should contain directory "
252 "and XXXXXX. (NULL == disabled)",
253 NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
254
255 g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
256 g_param_spec_string ("temp-location", "Temporary File Location",
257 "Location to store temporary files in (Only read this property, "
258 "use temp-template to configure the name template)",
259 NULL, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
260
261 /**
262 * GstDownloadBuffer:temp-remove
263 *
264 * When temp-template is set, remove the temporary file when going to READY.
265 */
266 g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
267 g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
268 "Remove the temp-location after use",
269 DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
270
271 /* set several parent class virtual functions */
272 gobject_class->finalize = gst_download_buffer_finalize;
273
274 gst_element_class_add_static_pad_template (gstelement_class, &srctemplate);
275 gst_element_class_add_static_pad_template (gstelement_class, &sinktemplate);
276
277 gst_element_class_set_static_metadata (gstelement_class, "DownloadBuffer",
278 "Generic", "Download Buffer element",
279 "Wim Taymans <wim.taymans@gmail.com>");
280
281 gstelement_class->change_state =
282 GST_DEBUG_FUNCPTR (gst_download_buffer_change_state);
283 gstelement_class->query =
284 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_query);
285 }
286
287 static void
gst_download_buffer_init(GstDownloadBuffer * dlbuf)288 gst_download_buffer_init (GstDownloadBuffer * dlbuf)
289 {
290 dlbuf->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
291
292 gst_pad_set_chain_function (dlbuf->sinkpad,
293 GST_DEBUG_FUNCPTR (gst_download_buffer_chain));
294 gst_pad_set_activatemode_function (dlbuf->sinkpad,
295 GST_DEBUG_FUNCPTR (gst_download_buffer_sink_activate_mode));
296 gst_pad_set_event_function (dlbuf->sinkpad,
297 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_sink_event));
298 gst_pad_set_query_function (dlbuf->sinkpad,
299 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_sink_query));
300 GST_PAD_SET_PROXY_CAPS (dlbuf->sinkpad);
301 gst_element_add_pad (GST_ELEMENT (dlbuf), dlbuf->sinkpad);
302
303 dlbuf->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
304
305 gst_pad_set_activatemode_function (dlbuf->srcpad,
306 GST_DEBUG_FUNCPTR (gst_download_buffer_src_activate_mode));
307 gst_pad_set_getrange_function (dlbuf->srcpad,
308 GST_DEBUG_FUNCPTR (gst_download_buffer_get_range));
309 gst_pad_set_event_function (dlbuf->srcpad,
310 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_src_event));
311 gst_pad_set_query_function (dlbuf->srcpad,
312 GST_DEBUG_FUNCPTR (gst_download_buffer_handle_src_query));
313 GST_PAD_SET_PROXY_CAPS (dlbuf->srcpad);
314 gst_element_add_pad (GST_ELEMENT (dlbuf), dlbuf->srcpad);
315
316 /* levels */
317 GST_DOWNLOAD_BUFFER_CLEAR_LEVEL (dlbuf->cur_level);
318 dlbuf->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;
319 dlbuf->max_level.time = DEFAULT_MAX_SIZE_TIME;
320 dlbuf->low_percent = DEFAULT_LOW_PERCENT;
321 dlbuf->high_percent = DEFAULT_HIGH_PERCENT;
322
323 dlbuf->srcresult = GST_FLOW_FLUSHING;
324 dlbuf->sinkresult = GST_FLOW_FLUSHING;
325 dlbuf->in_timer = g_timer_new ();
326 dlbuf->out_timer = g_timer_new ();
327
328 g_mutex_init (&dlbuf->qlock);
329 dlbuf->waiting_add = FALSE;
330 g_cond_init (&dlbuf->item_add);
331
332 /* tempfile related */
333 dlbuf->temp_template = NULL;
334 dlbuf->temp_location = NULL;
335 dlbuf->temp_remove = DEFAULT_TEMP_REMOVE;
336 }
337
338 /* called only once, as opposed to dispose */
339 static void
gst_download_buffer_finalize(GObject * object)340 gst_download_buffer_finalize (GObject * object)
341 {
342 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
343
344 g_mutex_clear (&dlbuf->qlock);
345 g_cond_clear (&dlbuf->item_add);
346 g_timer_destroy (dlbuf->in_timer);
347 g_timer_destroy (dlbuf->out_timer);
348
349 /* temp_file path cleanup */
350 g_free (dlbuf->temp_template);
351 g_free (dlbuf->temp_location);
352
353 G_OBJECT_CLASS (parent_class)->finalize (object);
354 }
355
356 static void
reset_positions(GstDownloadBuffer * dlbuf)357 reset_positions (GstDownloadBuffer * dlbuf)
358 {
359 dlbuf->write_pos = 0;
360 dlbuf->read_pos = 0;
361 dlbuf->filling = TRUE;
362 dlbuf->buffering_percent = 0;
363 dlbuf->is_buffering = TRUE;
364 dlbuf->seeking = FALSE;
365 GST_DOWNLOAD_BUFFER_CLEAR_LEVEL (dlbuf->cur_level);
366 }
367
368 static void
reset_rate_timer(GstDownloadBuffer * dlbuf)369 reset_rate_timer (GstDownloadBuffer * dlbuf)
370 {
371 dlbuf->bytes_in = 0;
372 dlbuf->bytes_out = 0;
373 dlbuf->byte_in_rate = 0.0;
374 dlbuf->byte_in_period = 0;
375 dlbuf->byte_out_rate = 0.0;
376 dlbuf->last_in_elapsed = 0.0;
377 dlbuf->last_out_elapsed = 0.0;
378 dlbuf->in_timer_started = FALSE;
379 dlbuf->out_timer_started = FALSE;
380 }
381
382 /* the interval in seconds to recalculate the rate */
383 #define RATE_INTERVAL 0.2
384 /* Tuning for rate estimation. We use a large window for the input rate because
385 * it should be stable when connected to a network. The output rate is less
386 * stable (the elements preroll, queues behind a demuxer fill, ...) and should
387 * therefore adapt more quickly.
388 * However, initial input rate may be subject to a burst, and should therefore
389 * initially also adapt more quickly to changes, and only later on give higher
390 * weight to previous values. */
391 #define AVG_IN(avg,val,w1,w2) ((avg) * (w1) + (val) * (w2)) / ((w1) + (w2))
392 #define AVG_OUT(avg,val) ((avg) * 3.0 + (val)) / 4.0
393
394 static void
update_levels(GstDownloadBuffer * dlbuf,guint bytes)395 update_levels (GstDownloadBuffer * dlbuf, guint bytes)
396 {
397 dlbuf->cur_level.bytes = bytes;
398
399 if (dlbuf->byte_in_rate > 0.0) {
400 dlbuf->cur_level.time =
401 dlbuf->cur_level.bytes / dlbuf->byte_in_rate * GST_SECOND;
402 }
403
404 GST_DEBUG ("levels: bytes %u/%u, time %" GST_TIME_FORMAT "/%" GST_TIME_FORMAT,
405 dlbuf->cur_level.bytes, dlbuf->max_level.bytes,
406 GST_TIME_ARGS (dlbuf->cur_level.time),
407 GST_TIME_ARGS (dlbuf->max_level.time));
408 }
409
410 static void
update_in_rates(GstDownloadBuffer * dlbuf)411 update_in_rates (GstDownloadBuffer * dlbuf)
412 {
413 gdouble elapsed, period;
414 gdouble byte_in_rate;
415
416 if (!dlbuf->in_timer_started) {
417 dlbuf->in_timer_started = TRUE;
418 g_timer_start (dlbuf->in_timer);
419 return;
420 }
421
422 elapsed = g_timer_elapsed (dlbuf->in_timer, NULL);
423
424 /* recalc after each interval. */
425 if (dlbuf->last_in_elapsed + RATE_INTERVAL < elapsed) {
426 period = elapsed - dlbuf->last_in_elapsed;
427
428 GST_DEBUG_OBJECT (dlbuf,
429 "rates: period %f, in %" G_GUINT64_FORMAT ", global period %f",
430 period, dlbuf->bytes_in, dlbuf->byte_in_period);
431
432 byte_in_rate = dlbuf->bytes_in / period;
433
434 if (dlbuf->byte_in_rate == 0.0)
435 dlbuf->byte_in_rate = byte_in_rate;
436 else
437 dlbuf->byte_in_rate = AVG_IN (dlbuf->byte_in_rate, byte_in_rate,
438 (double) dlbuf->byte_in_period, period);
439
440 /* another data point, cap at 16 for long time running average */
441 if (dlbuf->byte_in_period < 16 * RATE_INTERVAL)
442 dlbuf->byte_in_period += period;
443
444 /* reset the values to calculate rate over the next interval */
445 dlbuf->last_in_elapsed = elapsed;
446 dlbuf->bytes_in = 0;
447 GST_DEBUG_OBJECT (dlbuf, "rates: in %f", dlbuf->byte_in_rate);
448 }
449 }
450
451 static void
update_out_rates(GstDownloadBuffer * dlbuf)452 update_out_rates (GstDownloadBuffer * dlbuf)
453 {
454 gdouble elapsed, period;
455 gdouble byte_out_rate;
456
457 if (!dlbuf->out_timer_started) {
458 dlbuf->out_timer_started = TRUE;
459 g_timer_start (dlbuf->out_timer);
460 return;
461 }
462
463 elapsed = g_timer_elapsed (dlbuf->out_timer, NULL);
464
465 /* recalc after each interval. */
466 if (dlbuf->last_out_elapsed + RATE_INTERVAL < elapsed) {
467 period = elapsed - dlbuf->last_out_elapsed;
468
469 GST_DEBUG_OBJECT (dlbuf,
470 "rates: period %f, out %" G_GUINT64_FORMAT, period, dlbuf->bytes_out);
471
472 byte_out_rate = dlbuf->bytes_out / period;
473
474 if (dlbuf->byte_out_rate == 0.0)
475 dlbuf->byte_out_rate = byte_out_rate;
476 else
477 dlbuf->byte_out_rate = AVG_OUT (dlbuf->byte_out_rate, byte_out_rate);
478
479 /* reset the values to calculate rate over the next interval */
480 dlbuf->last_out_elapsed = elapsed;
481 dlbuf->bytes_out = 0;
482 GST_DEBUG_OBJECT (dlbuf, "rates: out %f", dlbuf->byte_out_rate);
483 }
484 }
485
486 static gboolean
get_buffering_percent(GstDownloadBuffer * dlbuf,gboolean * is_buffering,gint * percent)487 get_buffering_percent (GstDownloadBuffer * dlbuf, gboolean * is_buffering,
488 gint * percent)
489 {
490 gint perc;
491
492 if (dlbuf->high_percent <= 0) {
493 if (percent)
494 *percent = 100;
495 if (is_buffering)
496 *is_buffering = FALSE;
497 return FALSE;
498 }
499
500 /* Ensure the variables used to calculate buffering state are up-to-date. */
501 update_in_rates (dlbuf);
502 update_out_rates (dlbuf);
503
504 /* figure out the percent we are filled, we take the max of all formats. */
505 if (dlbuf->max_level.bytes > 0) {
506 if (dlbuf->cur_level.bytes >= dlbuf->max_level.bytes)
507 perc = 100;
508 else
509 perc = dlbuf->cur_level.bytes * 100 / dlbuf->max_level.bytes;
510 } else
511 perc = 0;
512
513 if (dlbuf->max_level.time > 0) {
514 if (dlbuf->cur_level.time >= dlbuf->max_level.time)
515 perc = 100;
516 else
517 perc = MAX (perc, dlbuf->cur_level.time * 100 / dlbuf->max_level.time);
518 } else
519 perc = MAX (0, perc);
520
521 if (is_buffering)
522 *is_buffering = dlbuf->is_buffering;
523
524 /* scale to high percent so that it becomes the 100% mark */
525 perc = perc * 100 / dlbuf->high_percent;
526 /* clip */
527 if (perc > 100)
528 perc = 100;
529
530 if (percent)
531 *percent = perc;
532
533 GST_DEBUG_OBJECT (dlbuf, "buffering %d, percent %d", dlbuf->is_buffering,
534 perc);
535
536 return TRUE;
537 }
538
539 static void
get_buffering_stats(GstDownloadBuffer * dlbuf,gint percent,GstBufferingMode * mode,gint * avg_in,gint * avg_out,gint64 * buffering_left)540 get_buffering_stats (GstDownloadBuffer * dlbuf, gint percent,
541 GstBufferingMode * mode, gint * avg_in, gint * avg_out,
542 gint64 * buffering_left)
543 {
544 if (mode)
545 *mode = GST_BUFFERING_DOWNLOAD;
546
547 if (avg_in)
548 *avg_in = dlbuf->byte_in_rate;
549 if (avg_out)
550 *avg_out = dlbuf->byte_out_rate;
551
552 if (buffering_left) {
553 guint64 max, cur;
554
555 *buffering_left = (percent == 100 ? 0 : -1);
556
557 max = dlbuf->max_level.time;
558 cur = dlbuf->cur_level.time;
559
560 if (percent != 100 && max > cur)
561 *buffering_left = (max - cur) / 1000000;
562 }
563 }
564
565 static GstMessage *
update_buffering(GstDownloadBuffer * dlbuf)566 update_buffering (GstDownloadBuffer * dlbuf)
567 {
568 gint percent;
569 gboolean post = FALSE;
570 GstMessage *message = NULL;
571
572 if (!get_buffering_percent (dlbuf, NULL, &percent))
573 return NULL;
574
575 if (dlbuf->is_buffering) {
576 post = TRUE;
577 /* if we were buffering see if we reached the high watermark */
578 if (percent >= dlbuf->high_percent)
579 dlbuf->is_buffering = FALSE;
580 } else {
581 /* we were not buffering, check if we need to start buffering if we drop
582 * below the low threshold */
583 if (percent < dlbuf->low_percent) {
584 dlbuf->is_buffering = TRUE;
585 post = TRUE;
586 }
587 }
588
589 if (post) {
590 if (percent == dlbuf->buffering_percent)
591 post = FALSE;
592 else
593 dlbuf->buffering_percent = percent;
594 }
595
596 if (post) {
597 GstBufferingMode mode;
598 gint avg_in, avg_out;
599 gint64 buffering_left;
600
601 get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
602 &buffering_left);
603
604 message = gst_message_new_buffering (GST_OBJECT_CAST (dlbuf),
605 (gint) percent);
606 gst_message_set_buffering_stats (message, mode,
607 avg_in, avg_out, buffering_left);
608 }
609
610 return message;
611 }
612
613 static gboolean
perform_seek_to_offset(GstDownloadBuffer * dlbuf,guint64 offset)614 perform_seek_to_offset (GstDownloadBuffer * dlbuf, guint64 offset)
615 {
616 GstEvent *event;
617 gboolean res;
618
619 if (dlbuf->seeking)
620 return TRUE;
621
622 /* until we receive the FLUSH_STOP from this seek, we skip data */
623 dlbuf->seeking = TRUE;
624 dlbuf->write_pos = offset;
625 dlbuf->filling = FALSE;
626 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
627
628 GST_DEBUG_OBJECT (dlbuf, "Seeking to %" G_GUINT64_FORMAT, offset);
629
630 event =
631 gst_event_new_seek (1.0, GST_FORMAT_BYTES,
632 GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
633 GST_SEEK_TYPE_NONE, -1);
634
635 res = gst_pad_push_event (dlbuf->sinkpad, event);
636 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
637
638 return res;
639 }
640
641 /* get the threshold for when we decide to seek rather than wait */
642 static guint64
get_seek_threshold(GstDownloadBuffer * dlbuf)643 get_seek_threshold (GstDownloadBuffer * dlbuf)
644 {
645 guint64 threshold;
646
647 /* FIXME, find a good threshold based on the incoming rate. */
648 threshold = 1024 * 512;
649
650 return threshold;
651 }
652
653 /* called with DOWNLOAD_BUFFER_MUTEX */
654 static void
gst_download_buffer_update_upstream_size(GstDownloadBuffer * dlbuf)655 gst_download_buffer_update_upstream_size (GstDownloadBuffer * dlbuf)
656 {
657 gint64 upstream_size = 0;
658
659 if (gst_pad_peer_query_duration (dlbuf->sinkpad, GST_FORMAT_BYTES,
660 &upstream_size)) {
661 GST_INFO_OBJECT (dlbuf, "upstream size: %" G_GINT64_FORMAT, upstream_size);
662 dlbuf->upstream_size = upstream_size;
663 }
664 }
665
666 /* called with DOWNLOAD_BUFFER_MUTEX */
667 static GstFlowReturn
gst_download_buffer_wait_for_data(GstDownloadBuffer * dlbuf,guint64 offset,guint length)668 gst_download_buffer_wait_for_data (GstDownloadBuffer * dlbuf, guint64 offset,
669 guint length)
670 {
671 gsize start, stop;
672 guint64 wanted;
673 gboolean started;
674
675 GST_DEBUG_OBJECT (dlbuf, "wait for %" G_GUINT64_FORMAT ", length %u",
676 offset, length);
677
678 wanted = offset + length;
679
680 /* pause the timer while we wait. The fact that we are waiting does not mean
681 * the byterate on the output pad is lower */
682 if ((started = dlbuf->out_timer_started))
683 g_timer_stop (dlbuf->out_timer);
684
685 /* check range before us */
686 if (gst_sparse_file_get_range_before (dlbuf->file, offset, &start, &stop)) {
687 GST_DEBUG_OBJECT (dlbuf,
688 "range before %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT, start, stop);
689 if (start <= offset && offset < stop) {
690 GST_DEBUG_OBJECT (dlbuf, "we have the offset");
691 /* we have the range, continue it */
692 offset = stop;
693 } else {
694 guint64 threshold, dist;
695
696 /* there is a range before us, check how far away it is */
697 threshold = get_seek_threshold (dlbuf);
698 dist = offset - stop;
699
700 if (dist <= threshold) {
701 GST_DEBUG_OBJECT (dlbuf, "not too far");
702 /* not far away, continue it */
703 offset = stop;
704 }
705 }
706 }
707
708 if (dlbuf->write_pos != offset) {
709 perform_seek_to_offset (dlbuf, offset);
710
711 /* perform_seek_to_offset() releases the lock, so we may have been flushed
712 * during the call. */
713 if (dlbuf->srcresult == GST_FLOW_FLUSHING)
714 goto out_flushing;
715 }
716
717 dlbuf->filling = TRUE;
718 if (dlbuf->write_pos > dlbuf->read_pos)
719 update_levels (dlbuf, dlbuf->write_pos - dlbuf->read_pos);
720 else
721 update_levels (dlbuf, 0);
722
723 /* now wait for more data */
724 GST_DEBUG_OBJECT (dlbuf, "waiting for more data");
725 GST_DOWNLOAD_BUFFER_WAIT_ADD_CHECK (dlbuf, dlbuf->srcresult, wanted,
726 out_flushing);
727 GST_DEBUG_OBJECT (dlbuf, "got more data");
728
729 /* and continue if we were running before */
730 if (started)
731 g_timer_continue (dlbuf->out_timer);
732
733 return GST_FLOW_OK;
734
735 out_flushing:
736 {
737 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
738 return GST_FLOW_FLUSHING;
739 }
740 }
741
742 /* called with DOWNLOAD_BUFFER_MUTEX */
743 static gboolean
check_upstream_size(GstDownloadBuffer * dlbuf,gsize offset,guint * length)744 check_upstream_size (GstDownloadBuffer * dlbuf, gsize offset, guint * length)
745 {
746 gsize stop = offset + *length;
747 /* catch any reads beyond the size of the file here to make sure cache
748 * doesn't send seek events beyond the size of the file upstream, since
749 * that would confuse elements such as souphttpsrc and/or http servers.
750 * Demuxers often just loop until EOS at the end of the file to figure out
751 * when they've read all the end-headers or index chunks. */
752 if (G_UNLIKELY (dlbuf->upstream_size == -1 || stop >= dlbuf->upstream_size)) {
753 gst_download_buffer_update_upstream_size (dlbuf);
754 }
755
756 if (dlbuf->upstream_size != -1) {
757 if (offset >= dlbuf->upstream_size)
758 return FALSE;
759
760 if (G_UNLIKELY (stop > dlbuf->upstream_size)) {
761 *length = dlbuf->upstream_size - offset;
762 GST_DEBUG_OBJECT (dlbuf, "adjusting length downto %u", *length);
763 }
764 }
765 return TRUE;
766 }
767
768 /* called with DOWNLOAD_BUFFER_MUTEX */
769 static GstFlowReturn
gst_download_buffer_read_buffer(GstDownloadBuffer * dlbuf,guint64 offset,guint length,GstBuffer ** buffer)770 gst_download_buffer_read_buffer (GstDownloadBuffer * dlbuf, guint64 offset,
771 guint length, GstBuffer ** buffer)
772 {
773 GstBuffer *buf;
774 GstMapInfo info;
775 GstFlowReturn ret = GST_FLOW_OK;
776 gsize res, remaining;
777 GError *error = NULL;
778
779 length = (length == -1) ? DEFAULT_BUFFER_SIZE : length;
780 offset = (offset == -1) ? dlbuf->read_pos : offset;
781
782 if (!check_upstream_size (dlbuf, offset, &length))
783 goto hit_eos;
784
785 /* allocate the output buffer of the requested size */
786 if (*buffer == NULL)
787 buf = gst_buffer_new_allocate (NULL, length, NULL);
788 else
789 buf = *buffer;
790
791 if (!gst_buffer_map (buf, &info, GST_MAP_WRITE))
792 goto map_failed;
793
794 GST_DEBUG_OBJECT (dlbuf, "Reading %u bytes from %" G_GUINT64_FORMAT, length,
795 offset);
796
797 dlbuf->read_pos = offset;
798
799 do {
800 res =
801 gst_sparse_file_read (dlbuf->file, offset, info.data, length,
802 &remaining, &error);
803 if (G_UNLIKELY (res == 0)) {
804 switch (error->code) {
805 case GST_SPARSE_FILE_IO_ERROR_WOULD_BLOCK:
806 /* we don't have the requested data in the file, decide what to
807 * do next. */
808 ret = gst_download_buffer_wait_for_data (dlbuf, offset, length);
809 if (ret != GST_FLOW_OK)
810 goto out_flushing;
811 break;
812 default:
813 goto read_error;
814 }
815 g_clear_error (&error);
816 }
817 } while (res == 0);
818
819 gst_buffer_unmap (buf, &info);
820 gst_buffer_resize (buf, 0, res);
821
822 dlbuf->bytes_out += res;
823 dlbuf->read_pos += res;
824
825 GST_DEBUG_OBJECT (dlbuf,
826 "Read %" G_GSIZE_FORMAT " bytes, remaining %" G_GSIZE_FORMAT, res,
827 remaining);
828
829 if (dlbuf->read_pos + remaining == dlbuf->upstream_size)
830 update_levels (dlbuf, dlbuf->max_level.bytes);
831 else
832 update_levels (dlbuf, remaining);
833
834 GST_BUFFER_OFFSET (buf) = offset;
835 GST_BUFFER_OFFSET_END (buf) = offset + res;
836
837 *buffer = buf;
838
839 return ret;
840
841 /* ERRORS */
842 hit_eos:
843 {
844 GST_DEBUG_OBJECT (dlbuf, "EOS hit");
845 return GST_FLOW_EOS;
846 }
847 map_failed:
848 {
849 GST_ELEMENT_ERROR (dlbuf, RESOURCE, BUSY,
850 (_("Failed to map buffer.")), ("failed to map buffer in WRITE mode"));
851 if (*buffer == NULL)
852 gst_buffer_unref (buf);
853 return GST_FLOW_ERROR;
854 }
855 out_flushing:
856 {
857 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
858 g_clear_error (&error);
859 gst_buffer_unmap (buf, &info);
860 if (*buffer == NULL)
861 gst_buffer_unref (buf);
862 return GST_FLOW_FLUSHING;
863 }
864 read_error:
865 {
866 GST_DEBUG_OBJECT (dlbuf, "we have a read error: %s", error->message);
867 g_clear_error (&error);
868 gst_buffer_unmap (buf, &info);
869 if (*buffer == NULL)
870 gst_buffer_unref (buf);
871 return GST_FLOW_ERROR;
872 }
873 }
874
875 /* must be called with MUTEX_LOCK. Will briefly release the lock when notifying
876 * the temp filename. */
877 static gboolean
gst_download_buffer_open_temp_location_file(GstDownloadBuffer * dlbuf)878 gst_download_buffer_open_temp_location_file (GstDownloadBuffer * dlbuf)
879 {
880 gint fd = -1;
881 gchar *name = NULL;
882
883 if (dlbuf->file)
884 goto already_opened;
885
886 GST_DEBUG_OBJECT (dlbuf, "opening temp file %s", dlbuf->temp_template);
887
888 /* If temp_template was set, allocate a filename and open that file */
889
890 /* nothing to do */
891 if (dlbuf->temp_template == NULL)
892 goto no_directory;
893
894 /* make copy of the template, we don't want to change this */
895 name = g_strdup (dlbuf->temp_template);
896 #ifdef __BIONIC__
897 fd = g_mkstemp_full (name, O_RDWR | O_LARGEFILE, S_IRUSR | S_IWUSR);
898 #else
899 fd = g_mkstemp (name);
900 #endif
901 if (fd == -1)
902 goto mkstemp_failed;
903
904 /* open the file for update/writing */
905 dlbuf->file = gst_sparse_file_new ();
906 /* error creating file */
907 if (!gst_sparse_file_set_fd (dlbuf->file, fd))
908 goto open_failed;
909
910 g_free (dlbuf->temp_location);
911 dlbuf->temp_location = name;
912 dlbuf->temp_fd = fd;
913 reset_positions (dlbuf);
914
915 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
916
917 /* we can't emit the notify with the lock */
918 g_object_notify (G_OBJECT (dlbuf), "temp-location");
919
920 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
921
922 GST_DEBUG_OBJECT (dlbuf, "opened temp file %s", dlbuf->temp_template);
923
924 return TRUE;
925
926 /* ERRORS */
927 already_opened:
928 {
929 GST_DEBUG_OBJECT (dlbuf, "temp file was already open");
930 return TRUE;
931 }
932 no_directory:
933 {
934 GST_ELEMENT_ERROR (dlbuf, RESOURCE, NOT_FOUND,
935 (_("No Temp directory specified.")), (NULL));
936 return FALSE;
937 }
938 mkstemp_failed:
939 {
940 GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
941 (_("Could not create temp file \"%s\"."), dlbuf->temp_template),
942 GST_ERROR_SYSTEM);
943 g_free (name);
944 return FALSE;
945 }
946 open_failed:
947 {
948 GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
949 (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
950 g_free (name);
951 if (fd != -1)
952 close (fd);
953 return FALSE;
954 }
955 }
956
957 static void
gst_download_buffer_close_temp_location_file(GstDownloadBuffer * dlbuf)958 gst_download_buffer_close_temp_location_file (GstDownloadBuffer * dlbuf)
959 {
960 /* nothing to do */
961 if (dlbuf->file == NULL)
962 return;
963
964 GST_DEBUG_OBJECT (dlbuf, "closing sparse file");
965
966 gst_sparse_file_free (dlbuf->file);
967 dlbuf->file = NULL;
968 /* fd was closed by gst_sparse_file_free's fclose() */
969 dlbuf->temp_fd = -1;
970
971 if (dlbuf->temp_remove) {
972 if (remove (dlbuf->temp_location) < 0) {
973 GST_WARNING_OBJECT (dlbuf, "Failed to remove temporary file %s: %s",
974 dlbuf->temp_location, g_strerror (errno));
975 }
976 }
977 }
978
979 static void
gst_download_buffer_flush_temp_file(GstDownloadBuffer * dlbuf)980 gst_download_buffer_flush_temp_file (GstDownloadBuffer * dlbuf)
981 {
982 if (dlbuf->file == NULL)
983 return;
984
985 GST_DEBUG_OBJECT (dlbuf, "flushing temp file");
986
987 gst_sparse_file_clear (dlbuf->file);
988 }
989
990 static void
gst_download_buffer_locked_flush(GstDownloadBuffer * dlbuf,gboolean full,gboolean clear_temp)991 gst_download_buffer_locked_flush (GstDownloadBuffer * dlbuf, gboolean full,
992 gboolean clear_temp)
993 {
994 if (clear_temp)
995 gst_download_buffer_flush_temp_file (dlbuf);
996 reset_positions (dlbuf);
997 gst_event_replace (&dlbuf->stream_start_event, NULL);
998 gst_event_replace (&dlbuf->segment_event, NULL);
999 }
1000
1001 static gboolean
gst_download_buffer_handle_sink_event(GstPad * pad,GstObject * parent,GstEvent * event)1002 gst_download_buffer_handle_sink_event (GstPad * pad, GstObject * parent,
1003 GstEvent * event)
1004 {
1005 gboolean ret = TRUE;
1006 GstDownloadBuffer *dlbuf;
1007
1008 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1009
1010 switch (GST_EVENT_TYPE (event)) {
1011 case GST_EVENT_FLUSH_START:
1012 {
1013 GST_LOG_OBJECT (dlbuf, "received flush start event");
1014 if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
1015 /* forward event */
1016 ret = gst_pad_push_event (dlbuf->srcpad, event);
1017
1018 /* now unblock the chain function */
1019 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1020 dlbuf->srcresult = GST_FLOW_FLUSHING;
1021 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1022 /* unblock the loop and chain functions */
1023 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1024 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1025
1026 /* make sure it pauses, this should happen since we sent
1027 * flush_start downstream. */
1028 gst_pad_pause_task (dlbuf->srcpad);
1029 GST_LOG_OBJECT (dlbuf, "loop stopped");
1030 } else {
1031 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1032 /* flush the sink pad */
1033 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1034 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1035
1036 gst_event_unref (event);
1037 }
1038 break;
1039 }
1040 case GST_EVENT_FLUSH_STOP:
1041 {
1042 GST_LOG_OBJECT (dlbuf, "received flush stop event");
1043
1044 if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
1045 /* forward event */
1046 ret = gst_pad_push_event (dlbuf->srcpad, event);
1047
1048 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1049 gst_download_buffer_locked_flush (dlbuf, FALSE, TRUE);
1050 dlbuf->srcresult = GST_FLOW_OK;
1051 dlbuf->sinkresult = GST_FLOW_OK;
1052 dlbuf->unexpected = FALSE;
1053 dlbuf->seeking = FALSE;
1054 /* reset rate counters */
1055 reset_rate_timer (dlbuf);
1056 gst_pad_start_task (dlbuf->srcpad,
1057 (GstTaskFunction) gst_download_buffer_loop, dlbuf->srcpad, NULL);
1058 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1059 } else {
1060 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1061 dlbuf->unexpected = FALSE;
1062 dlbuf->sinkresult = GST_FLOW_OK;
1063 dlbuf->seeking = FALSE;
1064 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1065
1066 gst_event_unref (event);
1067 }
1068 break;
1069 }
1070 default:
1071 if (GST_EVENT_IS_SERIALIZED (event)) {
1072 GstMessage *msg = NULL;
1073
1074 /* serialized events go in the buffer */
1075 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult,
1076 out_flushing);
1077 switch (GST_EVENT_TYPE (event)) {
1078 case GST_EVENT_EOS:
1079 GST_DEBUG_OBJECT (dlbuf, "we have EOS");
1080 /* Zero the thresholds, this makes sure the dlbuf is completely
1081 * filled and we can read all data from the dlbuf. */
1082 /* update the buffering status */
1083 update_levels (dlbuf, dlbuf->max_level.bytes);
1084 /* update the buffering */
1085 msg = update_buffering (dlbuf);
1086 /* wakeup the waiter and let it recheck */
1087 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1088 break;
1089 case GST_EVENT_SEGMENT:
1090 gst_event_replace (&dlbuf->segment_event, event);
1091 /* a new segment allows us to accept more buffers if we got EOS
1092 * from downstream */
1093 dlbuf->unexpected = FALSE;
1094 break;
1095 case GST_EVENT_STREAM_START:
1096 gst_event_replace (&dlbuf->stream_start_event, event);
1097 break;
1098 default:
1099 break;
1100 }
1101 gst_event_unref (event);
1102 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1103 if (msg != NULL)
1104 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1105 } else {
1106 /* non-serialized events are passed upstream. */
1107 ret = gst_pad_push_event (dlbuf->srcpad, event);
1108 }
1109 break;
1110 }
1111 return ret;
1112
1113 /* ERRORS */
1114 out_flushing:
1115 {
1116 GST_DEBUG_OBJECT (dlbuf, "refusing event, we are flushing");
1117 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1118 gst_event_unref (event);
1119 return FALSE;
1120 }
1121 }
1122
1123 static gboolean
gst_download_buffer_handle_sink_query(GstPad * pad,GstObject * parent,GstQuery * query)1124 gst_download_buffer_handle_sink_query (GstPad * pad, GstObject * parent,
1125 GstQuery * query)
1126 {
1127 GstDownloadBuffer *dlbuf;
1128 gboolean res;
1129
1130 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1131
1132 switch (GST_QUERY_TYPE (query)) {
1133 default:
1134 if (GST_QUERY_IS_SERIALIZED (query)) {
1135 GST_DEBUG_OBJECT (dlbuf, "refusing serialized query %p", query);
1136 res = FALSE;
1137 } else {
1138 res = gst_pad_query_default (pad, parent, query);
1139 }
1140 break;
1141 }
1142 return res;
1143 }
1144
1145 static GstFlowReturn
gst_download_buffer_chain(GstPad * pad,GstObject * parent,GstBuffer * buffer)1146 gst_download_buffer_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
1147 {
1148 GstDownloadBuffer *dlbuf;
1149 GstMapInfo info;
1150 guint64 offset;
1151 gsize res, available;
1152 GError *error = NULL;
1153 GstMessage *msg = NULL;
1154
1155 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1156
1157 GST_LOG_OBJECT (dlbuf, "received buffer %p of "
1158 "size %" G_GSIZE_FORMAT ", time %" GST_TIME_FORMAT ", duration %"
1159 GST_TIME_FORMAT, buffer, gst_buffer_get_size (buffer),
1160 GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1161 GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1162
1163 /* we have to lock the dlbuf since we span threads */
1164 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult, out_flushing);
1165 /* when we received unexpected from downstream, refuse more buffers */
1166 if (dlbuf->unexpected)
1167 goto out_eos;
1168
1169 /* while we didn't receive the newsegment, we're seeking and we skip data */
1170 if (dlbuf->seeking)
1171 goto out_seeking;
1172
1173 /* put buffer in dlbuf now */
1174 offset = dlbuf->write_pos;
1175
1176 /* sanity check */
1177 if (GST_BUFFER_OFFSET_IS_VALID (buffer) &&
1178 GST_BUFFER_OFFSET (buffer) != offset) {
1179 GST_WARNING_OBJECT (dlbuf, "buffer offset does not match current writing "
1180 "position! %" G_GINT64_FORMAT " != %" G_GINT64_FORMAT,
1181 GST_BUFFER_OFFSET (buffer), offset);
1182 }
1183
1184 if (!gst_buffer_map (buffer, &info, GST_MAP_READ))
1185 goto map_error;
1186
1187 GST_DEBUG_OBJECT (dlbuf, "Writing %" G_GSIZE_FORMAT " bytes to %"
1188 G_GUINT64_FORMAT, info.size, offset);
1189
1190 res =
1191 gst_sparse_file_write (dlbuf->file, offset, info.data, info.size,
1192 &available, &error);
1193 if (res == 0)
1194 goto write_error;
1195
1196 gst_buffer_unmap (buffer, &info);
1197 gst_buffer_unref (buffer);
1198
1199 dlbuf->write_pos = offset + info.size;
1200 dlbuf->bytes_in += info.size;
1201
1202 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, dlbuf->write_pos + available);
1203
1204 /* we hit the end, see what to do */
1205 if (dlbuf->write_pos + available == dlbuf->upstream_size) {
1206 gsize start, stop;
1207
1208 /* we have everything up to the end, find a region to fill */
1209 if (gst_sparse_file_get_range_after (dlbuf->file, 0, &start, &stop)) {
1210 if (stop < dlbuf->upstream_size) {
1211 /* a hole to fill, seek to its end */
1212 perform_seek_to_offset (dlbuf, stop);
1213 } else {
1214 /* we filled all the holes, we are done */
1215 goto completed;
1216 }
1217 }
1218 } else {
1219 /* see if we need to skip this region or just read it again. The idea
1220 * is that when the region is not big, we want to avoid a seek and just
1221 * let it reread */
1222 guint64 threshold = get_seek_threshold (dlbuf);
1223
1224 if (available > threshold) {
1225 /* further than threshold, it's better to skip than to reread */
1226 perform_seek_to_offset (dlbuf, dlbuf->write_pos + available);
1227 }
1228 }
1229 if (dlbuf->filling) {
1230 if (dlbuf->write_pos > dlbuf->read_pos)
1231 update_levels (dlbuf, dlbuf->write_pos - dlbuf->read_pos);
1232 else
1233 update_levels (dlbuf, 0);
1234 }
1235
1236 /* update the buffering */
1237 msg = update_buffering (dlbuf);
1238
1239 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1240
1241 if (msg != NULL)
1242 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1243
1244 return GST_FLOW_OK;
1245
1246 /* ERRORS */
1247 out_flushing:
1248 {
1249 GstFlowReturn ret = dlbuf->sinkresult;
1250 GST_LOG_OBJECT (dlbuf,
1251 "exit because task paused, reason: %s", gst_flow_get_name (ret));
1252 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1253 gst_buffer_unref (buffer);
1254 return ret;
1255 }
1256 out_eos:
1257 {
1258 GST_LOG_OBJECT (dlbuf, "exit because we received EOS");
1259 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1260 gst_buffer_unref (buffer);
1261 return GST_FLOW_EOS;
1262 }
1263 out_seeking:
1264 {
1265 GST_LOG_OBJECT (dlbuf, "exit because we are seeking");
1266 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1267 gst_buffer_unref (buffer);
1268 return GST_FLOW_OK;
1269 }
1270 map_error:
1271 {
1272 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1273 gst_buffer_unref (buffer);
1274 GST_ELEMENT_ERROR (dlbuf, RESOURCE, BUSY,
1275 (_("Failed to map buffer.")), ("failed to map buffer in READ mode"));
1276 return GST_FLOW_ERROR;
1277 }
1278 write_error:
1279 {
1280 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1281 gst_buffer_unmap (buffer, &info);
1282 gst_buffer_unref (buffer);
1283 GST_ELEMENT_ERROR (dlbuf, RESOURCE, WRITE,
1284 (_("Error while writing to download file.")), ("%s", error->message));
1285 g_clear_error (&error);
1286 return GST_FLOW_ERROR;
1287 }
1288 completed:
1289 {
1290 GstMessage *complete_message;
1291
1292 GST_LOG_OBJECT (dlbuf, "we completed the download");
1293 dlbuf->write_pos = dlbuf->upstream_size;
1294 dlbuf->filling = FALSE;
1295 update_levels (dlbuf, dlbuf->max_level.bytes);
1296 msg = update_buffering (dlbuf);
1297 complete_message = gst_message_new_element (GST_OBJECT_CAST (dlbuf),
1298 gst_structure_new ("GstCacheDownloadComplete",
1299 "location", G_TYPE_STRING, dlbuf->temp_location, NULL));
1300 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1301
1302 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), complete_message);
1303 if (msg != NULL)
1304 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1305
1306 return GST_FLOW_EOS;
1307 }
1308 }
1309
1310 /* called repeatedly with @pad as the source pad. This function should push out
1311 * data to the peer element. */
1312 static void
gst_download_buffer_loop(GstPad * pad)1313 gst_download_buffer_loop (GstPad * pad)
1314 {
1315 GstDownloadBuffer *dlbuf;
1316 GstFlowReturn ret;
1317 GstBuffer *buffer = NULL;
1318 GstMessage *msg = NULL;
1319
1320 dlbuf = GST_DOWNLOAD_BUFFER (GST_PAD_PARENT (pad));
1321
1322 /* have to lock for thread-safety */
1323 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1324
1325 ret = gst_download_buffer_read_buffer (dlbuf, -1, -1, &buffer);
1326 if (ret != GST_FLOW_OK)
1327 goto out_flushing;
1328
1329 if (dlbuf->stream_start_event != NULL) {
1330 gst_pad_push_event (dlbuf->srcpad, dlbuf->stream_start_event);
1331 dlbuf->stream_start_event = NULL;
1332 }
1333 if (dlbuf->segment_event != NULL) {
1334 gst_pad_push_event (dlbuf->srcpad, dlbuf->segment_event);
1335 dlbuf->segment_event = NULL;
1336 }
1337
1338
1339 /* update the buffering */
1340 msg = update_buffering (dlbuf);
1341
1342 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1343
1344 if (msg != NULL)
1345 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1346
1347 ret = gst_pad_push (dlbuf->srcpad, buffer);
1348
1349 /* need to check for srcresult here as well */
1350 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1351 dlbuf->srcresult = ret;
1352 dlbuf->sinkresult = ret;
1353 if (ret != GST_FLOW_OK)
1354 goto out_flushing;
1355 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1356
1357 return;
1358
1359 /* ERRORS */
1360 out_flushing:
1361 {
1362 GstFlowReturn ret = dlbuf->srcresult;
1363
1364 gst_pad_pause_task (dlbuf->srcpad);
1365 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1366 GST_LOG_OBJECT (dlbuf, "pause task, reason: %s", gst_flow_get_name (ret));
1367 /* let app know about us giving up if upstream is not expected to do so */
1368 if (ret == GST_FLOW_EOS) {
1369 /* FIXME perform EOS logic, this is really a basesrc operating on a
1370 * file. */
1371 gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1372 } else if ((ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS)) {
1373 GST_ELEMENT_FLOW_ERROR (dlbuf, ret);
1374 gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1375 }
1376 return;
1377 }
1378 }
1379
1380 static gboolean
gst_download_buffer_handle_src_event(GstPad * pad,GstObject * parent,GstEvent * event)1381 gst_download_buffer_handle_src_event (GstPad * pad, GstObject * parent,
1382 GstEvent * event)
1383 {
1384 gboolean res = TRUE;
1385 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (parent);
1386
1387 #ifndef GST_DISABLE_GST_DEBUG
1388 GST_DEBUG_OBJECT (dlbuf, "got event %p (%s)",
1389 event, GST_EVENT_TYPE_NAME (event));
1390 #endif
1391
1392 switch (GST_EVENT_TYPE (event)) {
1393 case GST_EVENT_FLUSH_START:
1394 /* now unblock the getrange function */
1395 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1396 GST_DEBUG_OBJECT (dlbuf, "flushing");
1397 dlbuf->srcresult = GST_FLOW_FLUSHING;
1398 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1399 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1400
1401 /* when using a temp file, we eat the event */
1402 res = TRUE;
1403 gst_event_unref (event);
1404 break;
1405 case GST_EVENT_FLUSH_STOP:
1406 /* now unblock the getrange function */
1407 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1408 dlbuf->srcresult = GST_FLOW_OK;
1409 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1410
1411 /* when using a temp file, we eat the event */
1412 res = TRUE;
1413 gst_event_unref (event);
1414 break;
1415 case GST_EVENT_RECONFIGURE:
1416 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1417 /* assume downstream is linked now and try to push again */
1418 if (dlbuf->srcresult == GST_FLOW_NOT_LINKED) {
1419 dlbuf->srcresult = GST_FLOW_OK;
1420 dlbuf->sinkresult = GST_FLOW_OK;
1421 if (GST_PAD_MODE (pad) == GST_PAD_MODE_PUSH) {
1422 gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1423 pad, NULL);
1424 }
1425 }
1426 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1427
1428 res = gst_pad_push_event (dlbuf->sinkpad, event);
1429 break;
1430 default:
1431 res = gst_pad_push_event (dlbuf->sinkpad, event);
1432 break;
1433 }
1434
1435 return res;
1436 }
1437
1438 static gboolean
gst_download_buffer_handle_src_query(GstPad * pad,GstObject * parent,GstQuery * query)1439 gst_download_buffer_handle_src_query (GstPad * pad, GstObject * parent,
1440 GstQuery * query)
1441 {
1442 GstDownloadBuffer *dlbuf;
1443
1444 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1445
1446 switch (GST_QUERY_TYPE (query)) {
1447 case GST_QUERY_POSITION:
1448 {
1449 gint64 peer_pos;
1450 GstFormat format;
1451
1452 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1453 goto peer_failed;
1454
1455 /* get peer position */
1456 gst_query_parse_position (query, &format, &peer_pos);
1457
1458 /* FIXME: this code assumes that there's no discont in the dlbuf */
1459 switch (format) {
1460 case GST_FORMAT_BYTES:
1461 peer_pos -= dlbuf->cur_level.bytes;
1462 if (peer_pos < 0) /* Clamp result to 0 */
1463 peer_pos = 0;
1464 break;
1465 case GST_FORMAT_TIME:
1466 peer_pos -= dlbuf->cur_level.time;
1467 if (peer_pos < 0) /* Clamp result to 0 */
1468 peer_pos = 0;
1469 break;
1470 default:
1471 GST_WARNING_OBJECT (dlbuf, "dropping query in %s format, don't "
1472 "know how to adjust value", gst_format_get_name (format));
1473 return FALSE;
1474 }
1475 /* set updated position */
1476 gst_query_set_position (query, format, peer_pos);
1477 break;
1478 }
1479 case GST_QUERY_DURATION:
1480 {
1481 GST_DEBUG_OBJECT (dlbuf, "doing peer query");
1482
1483 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1484 goto peer_failed;
1485
1486 GST_DEBUG_OBJECT (dlbuf, "peer query success");
1487 break;
1488 }
1489 case GST_QUERY_BUFFERING:
1490 {
1491 gint percent;
1492 gboolean is_buffering;
1493 GstBufferingMode mode;
1494 gint avg_in, avg_out;
1495 gint64 buffering_left;
1496
1497 GST_DEBUG_OBJECT (dlbuf, "query buffering");
1498
1499 get_buffering_percent (dlbuf, &is_buffering, &percent);
1500 gst_query_set_buffering_percent (query, is_buffering, percent);
1501
1502 get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
1503 &buffering_left);
1504 gst_query_set_buffering_stats (query, mode, avg_in, avg_out,
1505 buffering_left);
1506
1507 {
1508 /* add ranges for download and ringbuffer buffering */
1509 GstFormat format;
1510 gint64 start, stop;
1511 guint64 write_pos;
1512 gint64 estimated_total;
1513 gint64 duration;
1514 gsize offset, range_start, range_stop;
1515
1516 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1517 write_pos = dlbuf->write_pos;
1518
1519 /* get duration of upstream in bytes */
1520 gst_download_buffer_update_upstream_size (dlbuf);
1521 duration = dlbuf->upstream_size;
1522
1523 GST_DEBUG_OBJECT (dlbuf, "percent %d, duration %" G_GINT64_FORMAT
1524 ", writing %" G_GINT64_FORMAT, percent, duration, write_pos);
1525
1526 gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1527
1528 /* fill out the buffered ranges */
1529 start = offset = 0;
1530 stop = -1;
1531 estimated_total = -1;
1532 while (gst_sparse_file_get_range_after (dlbuf->file, offset,
1533 &range_start, &range_stop)) {
1534 gboolean current_range;
1535
1536 GST_DEBUG_OBJECT (dlbuf,
1537 "range starting at %" G_GSIZE_FORMAT " and finishing at %"
1538 G_GSIZE_FORMAT, range_start, range_stop);
1539
1540 offset = range_stop;
1541
1542 /* find the range we are currently downloading, we'll remember it
1543 * after we convert to the target format */
1544 if (range_start <= write_pos && range_stop >= write_pos) {
1545 current_range = TRUE;
1546 /* calculate remaining and total download time */
1547 if (duration >= range_stop && avg_in > 0.0)
1548 estimated_total = ((duration - range_stop) * 1000) / avg_in;
1549 } else
1550 current_range = FALSE;
1551
1552 switch (format) {
1553 case GST_FORMAT_PERCENT:
1554 /* get our available data relative to the duration */
1555 if (duration == -1) {
1556 range_start = 0;
1557 range_stop = 0;
1558 } else {
1559 range_start = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1560 range_start, duration);
1561 range_stop = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1562 range_stop, duration);
1563 }
1564 break;
1565 case GST_FORMAT_BYTES:
1566 break;
1567 default:
1568 range_start = -1;
1569 range_stop = -1;
1570 break;
1571 }
1572
1573 if (current_range) {
1574 /* we are currently downloading this range */
1575 start = range_start;
1576 stop = range_stop;
1577 }
1578 GST_DEBUG_OBJECT (dlbuf,
1579 "range to format: %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT,
1580 range_start, range_stop);
1581 if (range_start == range_stop)
1582 continue;
1583 gst_query_add_buffering_range (query, range_start, range_stop);
1584 }
1585
1586 GST_DEBUG_OBJECT (dlbuf, "estimated-total %" G_GINT64_FORMAT,
1587 estimated_total);
1588
1589 gst_query_set_buffering_range (query, format, start, stop,
1590 estimated_total);
1591
1592 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1593 }
1594 break;
1595 }
1596 case GST_QUERY_SCHEDULING:
1597 {
1598 GstSchedulingFlags flags = 0;
1599
1600 if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1601 goto peer_failed;
1602
1603 gst_query_parse_scheduling (query, &flags, NULL, NULL, NULL);
1604
1605 /* we can operate in pull mode when we are using a tempfile */
1606 flags |= GST_SCHEDULING_FLAG_SEEKABLE;
1607 gst_query_set_scheduling (query, flags, 0, -1, 0);
1608 gst_query_add_scheduling_mode (query, GST_PAD_MODE_PULL);
1609 gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
1610 break;
1611 }
1612 default:
1613 /* peer handled other queries */
1614 if (!gst_pad_query_default (pad, parent, query))
1615 goto peer_failed;
1616 break;
1617 }
1618
1619 return TRUE;
1620
1621 /* ERRORS */
1622 peer_failed:
1623 {
1624 GST_DEBUG_OBJECT (dlbuf, "failed peer query");
1625 return FALSE;
1626 }
1627 }
1628
1629 static gboolean
gst_download_buffer_handle_query(GstElement * element,GstQuery * query)1630 gst_download_buffer_handle_query (GstElement * element, GstQuery * query)
1631 {
1632 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (element);
1633
1634 /* simply forward to the srcpad query function */
1635 return gst_download_buffer_handle_src_query (dlbuf->srcpad,
1636 GST_OBJECT_CAST (element), query);
1637 }
1638
1639 static GstFlowReturn
gst_download_buffer_get_range(GstPad * pad,GstObject * parent,guint64 offset,guint length,GstBuffer ** buffer)1640 gst_download_buffer_get_range (GstPad * pad, GstObject * parent, guint64 offset,
1641 guint length, GstBuffer ** buffer)
1642 {
1643 GstDownloadBuffer *dlbuf;
1644 GstFlowReturn ret;
1645 GstMessage *msg = NULL;
1646
1647 dlbuf = GST_DOWNLOAD_BUFFER_CAST (parent);
1648
1649 GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1650 /* FIXME - function will block when the range is not yet available */
1651 ret = gst_download_buffer_read_buffer (dlbuf, offset, length, buffer);
1652 /* update the buffering */
1653 msg = update_buffering (dlbuf);
1654
1655 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1656
1657 if (msg != NULL)
1658 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1659
1660 return ret;
1661
1662 /* ERRORS */
1663 out_flushing:
1664 {
1665 ret = dlbuf->srcresult;
1666
1667 GST_DEBUG_OBJECT (dlbuf, "we are flushing");
1668 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1669 return ret;
1670 }
1671 }
1672
1673 /* sink currently only operates in push mode */
1674 static gboolean
gst_download_buffer_sink_activate_mode(GstPad * pad,GstObject * parent,GstPadMode mode,gboolean active)1675 gst_download_buffer_sink_activate_mode (GstPad * pad, GstObject * parent,
1676 GstPadMode mode, gboolean active)
1677 {
1678 gboolean result;
1679 GstDownloadBuffer *dlbuf;
1680
1681 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1682
1683 switch (mode) {
1684 case GST_PAD_MODE_PUSH:
1685 if (active) {
1686 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1687 GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1688 dlbuf->srcresult = GST_FLOW_OK;
1689 dlbuf->sinkresult = GST_FLOW_OK;
1690 dlbuf->unexpected = FALSE;
1691 reset_rate_timer (dlbuf);
1692 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1693 } else {
1694 /* unblock chain function */
1695 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1696 GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1697 dlbuf->srcresult = GST_FLOW_FLUSHING;
1698 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1699 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1700
1701 /* wait until it is unblocked and clean up */
1702 GST_PAD_STREAM_LOCK (pad);
1703 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1704 gst_download_buffer_locked_flush (dlbuf, TRUE, FALSE);
1705 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1706 GST_PAD_STREAM_UNLOCK (pad);
1707 }
1708 result = TRUE;
1709 break;
1710 default:
1711 result = FALSE;
1712 break;
1713 }
1714 return result;
1715 }
1716
1717 /* src operating in push mode, we start a task on the source pad that pushes out
1718 * buffers from the dlbuf */
1719 static gboolean
gst_download_buffer_src_activate_push(GstPad * pad,GstObject * parent,gboolean active)1720 gst_download_buffer_src_activate_push (GstPad * pad, GstObject * parent,
1721 gboolean active)
1722 {
1723 gboolean result = FALSE;
1724 GstDownloadBuffer *dlbuf;
1725
1726 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1727
1728 if (active) {
1729 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1730 GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1731 dlbuf->srcresult = GST_FLOW_OK;
1732 dlbuf->sinkresult = GST_FLOW_OK;
1733 dlbuf->unexpected = FALSE;
1734 result =
1735 gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1736 pad, NULL);
1737 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1738 } else {
1739 /* unblock loop function */
1740 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1741 GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1742 dlbuf->srcresult = GST_FLOW_FLUSHING;
1743 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1744 /* the item add signal will unblock */
1745 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1746 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1747
1748 /* step 2, make sure streaming finishes */
1749 result = gst_pad_stop_task (pad);
1750 }
1751
1752 return result;
1753 }
1754
1755 /* pull mode, downstream will call our getrange function */
1756 static gboolean
gst_download_buffer_src_activate_pull(GstPad * pad,GstObject * parent,gboolean active)1757 gst_download_buffer_src_activate_pull (GstPad * pad, GstObject * parent,
1758 gboolean active)
1759 {
1760 gboolean result;
1761 GstDownloadBuffer *dlbuf;
1762
1763 dlbuf = GST_DOWNLOAD_BUFFER (parent);
1764
1765 if (active) {
1766 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1767 /* open the temp file now */
1768 result = gst_download_buffer_open_temp_location_file (dlbuf);
1769 GST_DEBUG_OBJECT (dlbuf, "activating pull mode");
1770 dlbuf->srcresult = GST_FLOW_OK;
1771 dlbuf->sinkresult = GST_FLOW_OK;
1772 dlbuf->unexpected = FALSE;
1773 dlbuf->upstream_size = 0;
1774 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1775 } else {
1776 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1777 GST_DEBUG_OBJECT (dlbuf, "deactivating pull mode");
1778 dlbuf->srcresult = GST_FLOW_FLUSHING;
1779 dlbuf->sinkresult = GST_FLOW_FLUSHING;
1780 /* this will unlock getrange */
1781 GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1782 result = TRUE;
1783 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1784 }
1785
1786 return result;
1787 }
1788
1789 static gboolean
gst_download_buffer_src_activate_mode(GstPad * pad,GstObject * parent,GstPadMode mode,gboolean active)1790 gst_download_buffer_src_activate_mode (GstPad * pad, GstObject * parent,
1791 GstPadMode mode, gboolean active)
1792 {
1793 gboolean res;
1794
1795 switch (mode) {
1796 case GST_PAD_MODE_PULL:
1797 res = gst_download_buffer_src_activate_pull (pad, parent, active);
1798 break;
1799 case GST_PAD_MODE_PUSH:
1800 res = gst_download_buffer_src_activate_push (pad, parent, active);
1801 break;
1802 default:
1803 GST_LOG_OBJECT (pad, "unknown activation mode %d", mode);
1804 res = FALSE;
1805 break;
1806 }
1807 return res;
1808 }
1809
1810 static GstStateChangeReturn
gst_download_buffer_change_state(GstElement * element,GstStateChange transition)1811 gst_download_buffer_change_state (GstElement * element,
1812 GstStateChange transition)
1813 {
1814 GstDownloadBuffer *dlbuf;
1815 GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1816
1817 dlbuf = GST_DOWNLOAD_BUFFER (element);
1818
1819 switch (transition) {
1820 case GST_STATE_CHANGE_NULL_TO_READY:
1821 break;
1822 case GST_STATE_CHANGE_READY_TO_PAUSED:
1823 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1824 if (!gst_download_buffer_open_temp_location_file (dlbuf))
1825 ret = GST_STATE_CHANGE_FAILURE;
1826 gst_event_replace (&dlbuf->stream_start_event, NULL);
1827 gst_event_replace (&dlbuf->segment_event, NULL);
1828 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1829 break;
1830 case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1831 break;
1832 default:
1833 break;
1834 }
1835
1836 if (ret == GST_STATE_CHANGE_FAILURE)
1837 return ret;
1838
1839 ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1840
1841 if (ret == GST_STATE_CHANGE_FAILURE)
1842 return ret;
1843
1844 switch (transition) {
1845 case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1846 break;
1847 case GST_STATE_CHANGE_PAUSED_TO_READY:
1848 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1849 gst_download_buffer_close_temp_location_file (dlbuf);
1850 gst_event_replace (&dlbuf->stream_start_event, NULL);
1851 gst_event_replace (&dlbuf->segment_event, NULL);
1852 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1853 break;
1854 case GST_STATE_CHANGE_READY_TO_NULL:
1855 break;
1856 default:
1857 break;
1858 }
1859
1860 return ret;
1861 }
1862
1863 #define CAPACITY_CHANGE(elem) \
1864 update_buffering (elem);
1865
1866 static void
gst_download_buffer_set_temp_template(GstDownloadBuffer * dlbuf,const gchar * template)1867 gst_download_buffer_set_temp_template (GstDownloadBuffer * dlbuf,
1868 const gchar * template)
1869 {
1870 GstState state;
1871
1872 /* the element must be stopped in order to do this */
1873 GST_OBJECT_LOCK (dlbuf);
1874 state = GST_STATE (dlbuf);
1875 if (state != GST_STATE_READY && state != GST_STATE_NULL)
1876 goto wrong_state;
1877 GST_OBJECT_UNLOCK (dlbuf);
1878
1879 /* set new location */
1880 g_free (dlbuf->temp_template);
1881 dlbuf->temp_template = g_strdup (template);
1882
1883 return;
1884
1885 /* ERROR */
1886 wrong_state:
1887 {
1888 GST_WARNING_OBJECT (dlbuf, "setting temp-template property in wrong state");
1889 GST_OBJECT_UNLOCK (dlbuf);
1890 }
1891 }
1892
1893 static void
gst_download_buffer_set_property(GObject * object,guint prop_id,const GValue * value,GParamSpec * pspec)1894 gst_download_buffer_set_property (GObject * object,
1895 guint prop_id, const GValue * value, GParamSpec * pspec)
1896 {
1897 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1898 GstMessage *msg = NULL;
1899
1900 /* someone could change levels here, and since this
1901 * affects the get/put funcs, we need to lock for safety. */
1902 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1903
1904 switch (prop_id) {
1905 case PROP_MAX_SIZE_BYTES:
1906 dlbuf->max_level.bytes = g_value_get_uint (value);
1907 msg = CAPACITY_CHANGE (dlbuf);
1908 break;
1909 case PROP_MAX_SIZE_TIME:
1910 dlbuf->max_level.time = g_value_get_uint64 (value);
1911 msg = CAPACITY_CHANGE (dlbuf);
1912 break;
1913 case PROP_LOW_PERCENT:
1914 dlbuf->low_percent = g_value_get_int (value);
1915 break;
1916 case PROP_HIGH_PERCENT:
1917 dlbuf->high_percent = g_value_get_int (value);
1918 break;
1919 case PROP_TEMP_TEMPLATE:
1920 gst_download_buffer_set_temp_template (dlbuf, g_value_get_string (value));
1921 break;
1922 case PROP_TEMP_REMOVE:
1923 dlbuf->temp_remove = g_value_get_boolean (value);
1924 break;
1925 default:
1926 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1927 break;
1928 }
1929
1930 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1931
1932 if (msg != NULL)
1933 gst_element_post_message (GST_ELEMENT_CAST (dlbuf), msg);
1934
1935 }
1936
1937 static void
gst_download_buffer_get_property(GObject * object,guint prop_id,GValue * value,GParamSpec * pspec)1938 gst_download_buffer_get_property (GObject * object,
1939 guint prop_id, GValue * value, GParamSpec * pspec)
1940 {
1941 GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1942
1943 GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1944
1945 switch (prop_id) {
1946 case PROP_MAX_SIZE_BYTES:
1947 g_value_set_uint (value, dlbuf->max_level.bytes);
1948 break;
1949 case PROP_MAX_SIZE_TIME:
1950 g_value_set_uint64 (value, dlbuf->max_level.time);
1951 break;
1952 case PROP_LOW_PERCENT:
1953 g_value_set_int (value, dlbuf->low_percent);
1954 break;
1955 case PROP_HIGH_PERCENT:
1956 g_value_set_int (value, dlbuf->high_percent);
1957 break;
1958 case PROP_TEMP_TEMPLATE:
1959 g_value_set_string (value, dlbuf->temp_template);
1960 break;
1961 case PROP_TEMP_LOCATION:
1962 g_value_set_string (value, dlbuf->temp_location);
1963 break;
1964 case PROP_TEMP_REMOVE:
1965 g_value_set_boolean (value, dlbuf->temp_remove);
1966 break;
1967 default:
1968 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1969 break;
1970 }
1971
1972 GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1973 }
1974