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  <classname>&quot;GstCacheDownloadComplete&quot;</classname>
  45  * with the following information:
  46  *
  47  * *
  48  *   G_TYPE_STRING
  49  *   <classname>&quot;location&quot;</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
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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 *
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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
 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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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 }