subprojects/gstreamer/plugins/elements/gstdownloadbuffer.c

   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
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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 *
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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 }