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