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  *
  25  * The downloadbuffer element provides on-disk buffering and caching of, typically,
  26  * a network file. temp-template should be set to a value such as
  27  * /tmp/gstreamer-XXXXXX and the element will allocate a random free filename and
  28  * buffer the data in the file.
  29  *
  30  * With max-size-bytes and max-size-time you can configure the buffering limits.
  31  * The downloadbuffer element will try to read-ahead these amounts of data. When
  32  * the amount of read-ahead data drops below low-percent of the configured max,
  33  * the element will start emiting BUFFERING messages until high-percent of max is
  34  * reached again.
  35  *
  36  * The downloadbuffer provides push and pull based scheduling on its source pad
  37  * and will efficiently seek in the upstream element when needed.
  38  *
  39  * The temp-location property will be used to notify the application of the
  40  * allocated filename.
  41  *
  42  * When the downloadbuffer has completely downloaded the media, it will
  43  * post an application message named  <classname>&quot;GstCacheDownloadComplete&quot;</classname>
  44  * with the following information:
  45  * <itemizedlist>
  46  * <listitem>
  47  *   <para>
  48  *   G_TYPE_STRING
  49  *   <classname>&quot;location&quot;</classname>:
  50  *   the location of the completely downloaded file.
  51  *   </para>
  52  * </listitem>
  53  * </itemizedlist>
  54  */
  55
  56 #ifdef HAVE_CONFIG_H
  57 #include "config.h"
  58 #endif
  59
  60 #include "gstdownloadbuffer.h"
  61
  62 #include <glib/gstdio.h>
  63
  64 #include "gst/gst-i18n-lib.h"
  65 #include "gst/glib-compat-private.h"
  66
  67 #include <string.h>
  68
  69 #ifdef G_OS_WIN32
  70 #include <io.h>                 /* lseek, open, close, read */
  71 #undef lseek
  72 #define lseek _lseeki64
  73 #undef off_t
  74 #define off_t guint64
  75 #else
  76 #include <unistd.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 void 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 | G_PARAM_STATIC_STRINGS));
 229   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 230       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 231           "Max. amount of data to buffer (in ns, 0=disable)", 0, G_MAXUINT64,
 232           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 233
 234   g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
 235       g_param_spec_int ("low-percent", "Low percent",
 236           "Low threshold for buffering to start. Only used if use-buffering is True",
 237           0, 100, DEFAULT_LOW_PERCENT,
 238           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 239   g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
 240       g_param_spec_int ("high-percent", "High percent",
 241           "High threshold for buffering to finish. Only used if use-buffering is True",
 242           0, 100, DEFAULT_HIGH_PERCENT,
 243           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 244
 245   g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
 246       g_param_spec_string ("temp-template", "Temporary File Template",
 247           "File template to store temporary files in, should contain directory "
 248           "and XXXXXX. (NULL == disabled)",
 249           NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 250
 251   g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
 252       g_param_spec_string ("temp-location", "Temporary File Location",
 253           "Location to store temporary files in (Only read this property, "
 254           "use temp-template to configure the name template)",
 255           NULL, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 256
 257   /**
 258    * GstDownloadBuffer:temp-remove
 259    *
 260    * When temp-template is set, remove the temporary file when going to READY.
 261    */
 262   g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
 263       g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
 264           "Remove the temp-location after use",
 265           DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 266
 267   /* set several parent class virtual functions */
 268   gobject_class->finalize = gst_download_buffer_finalize;
 269
 270   gst_element_class_add_pad_template (gstelement_class,
 271       gst_static_pad_template_get (&srctemplate));
 272   gst_element_class_add_pad_template (gstelement_class,
 273       gst_static_pad_template_get (&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_time_level (GstDownloadBuffer * dlbuf)
 394 {
 395   if (dlbuf->byte_in_rate > 0.0) {
 396     dlbuf->cur_level.time =
 397         dlbuf->cur_level.bytes / dlbuf->byte_in_rate * GST_SECOND;
 398   }
 399   GST_DEBUG ("levels: bytes %u/%u, time %" GST_TIME_FORMAT "/%" GST_TIME_FORMAT,
 400       dlbuf->cur_level.bytes, dlbuf->max_level.bytes,
 401       GST_TIME_ARGS (dlbuf->cur_level.time),
 402       GST_TIME_ARGS (dlbuf->max_level.time));
 403   /* update the buffering */
 404   update_buffering (dlbuf);
 405 }
 406
 407 static void
 408 update_levels (GstDownloadBuffer * dlbuf, guint bytes)
 409 {
 410   dlbuf->cur_level.bytes = bytes;
 411   update_time_level (dlbuf);
 412 }
 413
 414 static void
 415 update_in_rates (GstDownloadBuffer * dlbuf)
 416 {
 417   gdouble elapsed, period;
 418   gdouble byte_in_rate;
 419
 420   if (!dlbuf->in_timer_started) {
 421     dlbuf->in_timer_started = TRUE;
 422     g_timer_start (dlbuf->in_timer);
 423     return;
 424   }
 425
 426   elapsed = g_timer_elapsed (dlbuf->in_timer, NULL);
 427
 428   /* recalc after each interval. */
 429   if (dlbuf->last_in_elapsed + RATE_INTERVAL < elapsed) {
 430     period = elapsed - dlbuf->last_in_elapsed;
 431
 432     GST_DEBUG_OBJECT (dlbuf,
 433         "rates: period %f, in %" G_GUINT64_FORMAT ", global period %f",
 434         period, dlbuf->bytes_in, dlbuf->byte_in_period);
 435
 436     byte_in_rate = dlbuf->bytes_in / period;
 437
 438     if (dlbuf->byte_in_rate == 0.0)
 439       dlbuf->byte_in_rate = byte_in_rate;
 440     else
 441       dlbuf->byte_in_rate = AVG_IN (dlbuf->byte_in_rate, byte_in_rate,
 442           (double) dlbuf->byte_in_period, period);
 443
 444     /* another data point, cap at 16 for long time running average */
 445     if (dlbuf->byte_in_period < 16 * RATE_INTERVAL)
 446       dlbuf->byte_in_period += period;
 447
 448     /* reset the values to calculate rate over the next interval */
 449     dlbuf->last_in_elapsed = elapsed;
 450     dlbuf->bytes_in = 0;
 451     GST_DEBUG_OBJECT (dlbuf, "rates: in %f", dlbuf->byte_in_rate);
 452   }
 453 }
 454
 455 static void
 456 update_out_rates (GstDownloadBuffer * dlbuf)
 457 {
 458   gdouble elapsed, period;
 459   gdouble byte_out_rate;
 460
 461   if (!dlbuf->out_timer_started) {
 462     dlbuf->out_timer_started = TRUE;
 463     g_timer_start (dlbuf->out_timer);
 464     return;
 465   }
 466
 467   elapsed = g_timer_elapsed (dlbuf->out_timer, NULL);
 468
 469   /* recalc after each interval. */
 470   if (dlbuf->last_out_elapsed + RATE_INTERVAL < elapsed) {
 471     period = elapsed - dlbuf->last_out_elapsed;
 472
 473     GST_DEBUG_OBJECT (dlbuf,
 474         "rates: period %f, out %" G_GUINT64_FORMAT, period, dlbuf->bytes_out);
 475
 476     byte_out_rate = dlbuf->bytes_out / period;
 477
 478     if (dlbuf->byte_out_rate == 0.0)
 479       dlbuf->byte_out_rate = byte_out_rate;
 480     else
 481       dlbuf->byte_out_rate = AVG_OUT (dlbuf->byte_out_rate, byte_out_rate);
 482
 483     /* reset the values to calculate rate over the next interval */
 484     dlbuf->last_out_elapsed = elapsed;
 485     dlbuf->bytes_out = 0;
 486     GST_DEBUG_OBJECT (dlbuf, "rates: out %f", dlbuf->byte_out_rate);
 487   }
 488 }
 489
 490 static gboolean
 491 get_buffering_percent (GstDownloadBuffer * dlbuf, gboolean * is_buffering,
 492     gint * percent)
 493 {
 494   gint perc;
 495
 496   if (dlbuf->high_percent <= 0) {
 497     if (percent)
 498       *percent = 100;
 499     if (is_buffering)
 500       *is_buffering = FALSE;
 501     return FALSE;
 502   }
 503
 504   /* Ensure the variables used to calculate buffering state are up-to-date. */
 505   update_in_rates (dlbuf);
 506   update_out_rates (dlbuf);
 507
 508   /* figure out the percent we are filled, we take the max of all formats. */
 509   if (dlbuf->max_level.bytes > 0) {
 510     if (dlbuf->cur_level.bytes >= dlbuf->max_level.bytes)
 511       perc = 100;
 512     else
 513       perc = dlbuf->cur_level.bytes * 100 / dlbuf->max_level.bytes;
 514   } else
 515     perc = 0;
 516
 517   if (dlbuf->max_level.time > 0) {
 518     if (dlbuf->cur_level.time >= dlbuf->max_level.time)
 519       perc = 100;
 520     else
 521       perc = MAX (perc, dlbuf->cur_level.time * 100 / dlbuf->max_level.time);
 522   } else
 523     perc = MAX (0, perc);
 524
 525   if (is_buffering)
 526     *is_buffering = dlbuf->is_buffering;
 527
 528   /* scale to high percent so that it becomes the 100% mark */
 529   perc = perc * 100 / dlbuf->high_percent;
 530   /* clip */
 531   if (perc > 100)
 532     perc = 100;
 533
 534   if (percent)
 535     *percent = perc;
 536
 537   GST_DEBUG_OBJECT (dlbuf, "buffering %d, percent %d", dlbuf->is_buffering,
 538       perc);
 539
 540   return TRUE;
 541 }
 542
 543 static void
 544 get_buffering_stats (GstDownloadBuffer * dlbuf, gint percent,
 545     GstBufferingMode * mode, gint * avg_in, gint * avg_out,
 546     gint64 * buffering_left)
 547 {
 548   if (mode)
 549     *mode = GST_BUFFERING_DOWNLOAD;
 550
 551   if (avg_in)
 552     *avg_in = dlbuf->byte_in_rate;
 553   if (avg_out)
 554     *avg_out = dlbuf->byte_out_rate;
 555
 556   if (buffering_left) {
 557     guint64 max, cur;
 558
 559     *buffering_left = (percent == 100 ? 0 : -1);
 560
 561     max = dlbuf->max_level.time;
 562     cur = dlbuf->cur_level.time;
 563
 564     if (percent != 100 && max > cur)
 565       *buffering_left = (max - cur) / 1000000;
 566   }
 567 }
 568
 569 static void
 570 update_buffering (GstDownloadBuffer * dlbuf)
 571 {
 572   gint percent;
 573   gboolean post = FALSE;
 574
 575   if (!get_buffering_percent (dlbuf, NULL, &percent))
 576     return;
 577
 578   if (dlbuf->is_buffering) {
 579     post = TRUE;
 580     /* if we were buffering see if we reached the high watermark */
 581     if (percent >= dlbuf->high_percent)
 582       dlbuf->is_buffering = FALSE;
 583   } else {
 584     /* we were not buffering, check if we need to start buffering if we drop
 585      * below the low threshold */
 586     if (percent < dlbuf->low_percent) {
 587       dlbuf->is_buffering = TRUE;
 588       post = TRUE;
 589     }
 590   }
 591
 592   if (post) {
 593     if (percent == dlbuf->buffering_percent)
 594       post = FALSE;
 595     else
 596       dlbuf->buffering_percent = percent;
 597   }
 598
 599   if (post) {
 600     GstMessage *message;
 601     GstBufferingMode mode;
 602     gint avg_in, avg_out;
 603     gint64 buffering_left;
 604
 605     get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
 606         &buffering_left);
 607
 608     message = gst_message_new_buffering (GST_OBJECT_CAST (dlbuf),
 609         (gint) percent);
 610     gst_message_set_buffering_stats (message, mode,
 611         avg_in, avg_out, buffering_left);
 612
 613     gst_element_post_message (GST_ELEMENT_CAST (dlbuf), message);
 614   }
 615 }
 616
 617 static gboolean
 618 perform_seek_to_offset (GstDownloadBuffer * dlbuf, guint64 offset)
 619 {
 620   GstEvent *event;
 621   gboolean res;
 622
 623   if (dlbuf->seeking)
 624     return TRUE;
 625
 626   /* until we receive the FLUSH_STOP from this seek, we skip data */
 627   dlbuf->seeking = TRUE;
 628   dlbuf->write_pos = offset;
 629   dlbuf->filling = FALSE;
 630   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
 631
 632   GST_DEBUG_OBJECT (dlbuf, "Seeking to %" G_GUINT64_FORMAT, offset);
 633
 634   event =
 635       gst_event_new_seek (1.0, GST_FORMAT_BYTES,
 636       GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
 637       GST_SEEK_TYPE_NONE, -1);
 638
 639   res = gst_pad_push_event (dlbuf->sinkpad, event);
 640   GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
 641
 642   return res;
 643 }
 644
 645 /* get the threshold for when we decide to seek rather than wait */
 646 static guint64
 647 get_seek_threshold (GstDownloadBuffer * dlbuf)
 648 {
 649   guint64 threshold;
 650
 651   /* FIXME, find a good threshold based on the incoming rate. */
 652   threshold = 1024 * 512;
 653
 654   return threshold;
 655 }
 656
 657 /* called with DOWNLOAD_BUFFER_MUTEX */
 658 static void
 659 gst_download_buffer_update_upstream_size (GstDownloadBuffer * dlbuf)
 660 {
 661   gint64 upstream_size = 0;
 662
 663   if (gst_pad_peer_query_duration (dlbuf->sinkpad, GST_FORMAT_BYTES,
 664           &upstream_size)) {
 665     GST_INFO_OBJECT (dlbuf, "upstream size: %" G_GINT64_FORMAT, upstream_size);
 666     dlbuf->upstream_size = upstream_size;
 667   }
 668 }
 669
 670 /* called with DOWNLOAD_BUFFER_MUTEX */
 671 static GstFlowReturn
 672 gst_download_buffer_wait_for_data (GstDownloadBuffer * dlbuf, guint64 offset,
 673     guint length)
 674 {
 675   gsize start, stop;
 676   guint64 wanted;
 677   gboolean started;
 678
 679   GST_DEBUG_OBJECT (dlbuf, "wait for %" G_GUINT64_FORMAT ", length %u",
 680       offset, length);
 681
 682   wanted = offset + length;
 683
 684   /* pause the timer while we wait. The fact that we are waiting does not mean
 685    * the byterate on the output pad is lower */
 686   if ((started = dlbuf->out_timer_started))
 687     g_timer_stop (dlbuf->out_timer);
 688
 689   /* check range before us */
 690   if (gst_sparse_file_get_range_before (dlbuf->file, offset, &start, &stop)) {
 691     GST_DEBUG_OBJECT (dlbuf,
 692         "range before %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT, start, stop);
 693     if (start <= offset && offset < stop) {
 694       GST_DEBUG_OBJECT (dlbuf, "we have the offset");
 695       /* we have the range, continue it */
 696       offset = stop;
 697     } else {
 698       guint64 threshold, dist;
 699
 700       /* there is a range before us, check how far away it is */
 701       threshold = get_seek_threshold (dlbuf);
 702       dist = offset - stop;
 703
 704       if (dist <= threshold) {
 705         GST_DEBUG_OBJECT (dlbuf, "not too far");
 706         /* not far away, continue it */
 707         offset = stop;
 708       }
 709     }
 710   }
 711
 712   if (dlbuf->write_pos != offset)
 713     perform_seek_to_offset (dlbuf, offset);
 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   gst_buffer_map (buf, &info, GST_MAP_WRITE);
 790
 791   GST_DEBUG_OBJECT (dlbuf, "Reading %u bytes from %" G_GUINT64_FORMAT, length,
 792       offset);
 793
 794   dlbuf->read_pos = offset;
 795
 796   do {
 797     res =
 798         gst_sparse_file_read (dlbuf->file, offset, info.data, length,
 799         &remaining, &error);
 800     if (G_UNLIKELY (res == 0)) {
 801       switch (error->code) {
 802         case GST_SPARSE_FILE_IO_ERROR_WOULD_BLOCK:
 803           /* we don't have the requested data in the file, decide what to
 804            * do next. */
 805           ret = gst_download_buffer_wait_for_data (dlbuf, offset, length);
 806           if (ret != GST_FLOW_OK)
 807             goto out_flushing;
 808           break;
 809         default:
 810           goto read_error;
 811       }
 812       g_clear_error (&error);
 813     }
 814   } while (res == 0);
 815
 816   gst_buffer_unmap (buf, &info);
 817   gst_buffer_resize (buf, 0, res);
 818
 819   dlbuf->bytes_out += res;
 820   dlbuf->read_pos += res;
 821
 822   GST_DEBUG_OBJECT (dlbuf,
 823       "Read %" G_GSIZE_FORMAT " bytes, remaining %" G_GSIZE_FORMAT, res,
 824       remaining);
 825
 826   if (dlbuf->read_pos + remaining == dlbuf->upstream_size)
 827     update_levels (dlbuf, dlbuf->max_level.bytes);
 828   else
 829     update_levels (dlbuf, remaining);
 830
 831   GST_BUFFER_OFFSET (buf) = offset;
 832   GST_BUFFER_OFFSET_END (buf) = offset + res;
 833
 834   *buffer = buf;
 835
 836   return ret;
 837
 838   /* ERRORS */
 839 hit_eos:
 840   {
 841     GST_DEBUG_OBJECT (dlbuf, "EOS hit");
 842     return GST_FLOW_EOS;
 843   }
 844 out_flushing:
 845   {
 846     GST_DEBUG_OBJECT (dlbuf, "we are flushing");
 847     gst_buffer_unmap (buf, &info);
 848     if (*buffer == NULL)
 849       gst_buffer_unref (buf);
 850     return GST_FLOW_FLUSHING;
 851   }
 852 read_error:
 853   {
 854     GST_DEBUG_OBJECT (dlbuf, "we have a read error: %s", error->message);
 855     g_clear_error (&error);
 856     gst_buffer_unmap (buf, &info);
 857     if (*buffer == NULL)
 858       gst_buffer_unref (buf);
 859     return ret;
 860   }
 861 }
 862
 863 /* must be called with MUTEX_LOCK. Will briefly release the lock when notifying
 864  * the temp filename. */
 865 static gboolean
 866 gst_download_buffer_open_temp_location_file (GstDownloadBuffer * dlbuf)
 867 {
 868   gint fd = -1;
 869   gchar *name = NULL;
 870
 871   if (dlbuf->file)
 872     goto already_opened;
 873
 874   GST_DEBUG_OBJECT (dlbuf, "opening temp file %s", dlbuf->temp_template);
 875
 876   /* If temp_template was set, allocate a filename and open that filen */
 877
 878   /* nothing to do */
 879   if (dlbuf->temp_template == NULL)
 880     goto no_directory;
 881
 882   /* make copy of the template, we don't want to change this */
 883   name = g_strdup (dlbuf->temp_template);
 884   fd = g_mkstemp (name);
 885   if (fd == -1)
 886     goto mkstemp_failed;
 887
 888   /* open the file for update/writing */
 889   dlbuf->file = gst_sparse_file_new ();
 890   /* error creating file */
 891   if (!gst_sparse_file_set_fd (dlbuf->file, fd))
 892     goto open_failed;
 893
 894   g_free (dlbuf->temp_location);
 895   dlbuf->temp_location = name;
 896   dlbuf->temp_fd = fd;
 897   reset_positions (dlbuf);
 898
 899   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
 900
 901   /* we can't emit the notify with the lock */
 902   g_object_notify (G_OBJECT (dlbuf), "temp-location");
 903
 904   GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
 905
 906   GST_DEBUG_OBJECT (dlbuf, "opened temp file %s", dlbuf->temp_template);
 907
 908   return TRUE;
 909
 910   /* ERRORS */
 911 already_opened:
 912   {
 913     GST_DEBUG_OBJECT (dlbuf, "temp file was already open");
 914     return TRUE;
 915   }
 916 no_directory:
 917   {
 918     GST_ELEMENT_ERROR (dlbuf, RESOURCE, NOT_FOUND,
 919         (_("No Temp directory specified.")), (NULL));
 920     return FALSE;
 921   }
 922 mkstemp_failed:
 923   {
 924     GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
 925         (_("Could not create temp file \"%s\"."), dlbuf->temp_template),
 926         GST_ERROR_SYSTEM);
 927     g_free (name);
 928     return FALSE;
 929   }
 930 open_failed:
 931   {
 932     GST_ELEMENT_ERROR (dlbuf, RESOURCE, OPEN_READ,
 933         (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
 934     g_free (name);
 935     if (fd != -1)
 936       close (fd);
 937     return FALSE;
 938   }
 939 }
 940
 941 static void
 942 gst_download_buffer_close_temp_location_file (GstDownloadBuffer * dlbuf)
 943 {
 944   /* nothing to do */
 945   if (dlbuf->file == NULL)
 946     return;
 947
 948   GST_DEBUG_OBJECT (dlbuf, "closing sparse file");
 949
 950   if (dlbuf->temp_remove) {
 951     if (remove (dlbuf->temp_location) < 0) {
 952       GST_WARNING_OBJECT (dlbuf, "Failed to remove temporary file %s: %s",
 953           dlbuf->temp_location, g_strerror (errno));
 954     }
 955   }
 956   gst_sparse_file_free (dlbuf->file);
 957   close (dlbuf->temp_fd);
 958   dlbuf->file = NULL;
 959 }
 960
 961 static void
 962 gst_download_buffer_flush_temp_file (GstDownloadBuffer * dlbuf)
 963 {
 964   if (dlbuf->file == NULL)
 965     return;
 966
 967   GST_DEBUG_OBJECT (dlbuf, "flushing temp file");
 968
 969   gst_sparse_file_clear (dlbuf->file);
 970 }
 971
 972 static void
 973 gst_download_buffer_locked_flush (GstDownloadBuffer * dlbuf, gboolean full,
 974     gboolean clear_temp)
 975 {
 976   if (clear_temp)
 977     gst_download_buffer_flush_temp_file (dlbuf);
 978   reset_positions (dlbuf);
 979   gst_event_replace (&dlbuf->stream_start_event, NULL);
 980   gst_event_replace (&dlbuf->segment_event, NULL);
 981 }
 982
 983 static gboolean
 984 gst_download_buffer_handle_sink_event (GstPad * pad, GstObject * parent,
 985     GstEvent * event)
 986 {
 987   gboolean ret = TRUE;
 988   GstDownloadBuffer *dlbuf;
 989
 990   dlbuf = GST_DOWNLOAD_BUFFER (parent);
 991
 992   switch (GST_EVENT_TYPE (event)) {
 993     case GST_EVENT_FLUSH_START:
 994     {
 995       GST_LOG_OBJECT (dlbuf, "received flush start event");
 996       if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
 997         /* forward event */
 998         ret = gst_pad_push_event (dlbuf->srcpad, event);
 999
1000         /* now unblock the chain function */
1001         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1002         dlbuf->srcresult = GST_FLOW_FLUSHING;
1003         dlbuf->sinkresult = GST_FLOW_FLUSHING;
1004         /* unblock the loop and chain functions */
1005         GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1006         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1007
1008         /* make sure it pauses, this should happen since we sent
1009          * flush_start downstream. */
1010         gst_pad_pause_task (dlbuf->srcpad);
1011         GST_LOG_OBJECT (dlbuf, "loop stopped");
1012       } else {
1013         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1014         /* flush the sink pad */
1015         dlbuf->sinkresult = GST_FLOW_FLUSHING;
1016         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1017
1018         gst_event_unref (event);
1019       }
1020       break;
1021     }
1022     case GST_EVENT_FLUSH_STOP:
1023     {
1024       GST_LOG_OBJECT (dlbuf, "received flush stop event");
1025
1026       if (GST_PAD_MODE (dlbuf->srcpad) == GST_PAD_MODE_PUSH) {
1027         /* forward event */
1028         ret = gst_pad_push_event (dlbuf->srcpad, event);
1029
1030         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1031         gst_download_buffer_locked_flush (dlbuf, FALSE, TRUE);
1032         dlbuf->srcresult = GST_FLOW_OK;
1033         dlbuf->sinkresult = GST_FLOW_OK;
1034         dlbuf->unexpected = FALSE;
1035         dlbuf->seeking = FALSE;
1036         /* reset rate counters */
1037         reset_rate_timer (dlbuf);
1038         gst_pad_start_task (dlbuf->srcpad,
1039             (GstTaskFunction) gst_download_buffer_loop, dlbuf->srcpad, NULL);
1040         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1041       } else {
1042         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1043         dlbuf->unexpected = FALSE;
1044         dlbuf->sinkresult = GST_FLOW_OK;
1045         dlbuf->seeking = FALSE;
1046         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1047
1048         gst_event_unref (event);
1049       }
1050       break;
1051     }
1052     default:
1053       if (GST_EVENT_IS_SERIALIZED (event)) {
1054         /* serialized events go in the buffer */
1055         GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult,
1056             out_flushing);
1057         switch (GST_EVENT_TYPE (event)) {
1058           case GST_EVENT_EOS:
1059             GST_DEBUG_OBJECT (dlbuf, "we have EOS");
1060             /* Zero the thresholds, this makes sure the dlbuf is completely
1061              * filled and we can read all data from the dlbuf. */
1062             /* update the buffering status */
1063             update_levels (dlbuf, dlbuf->max_level.bytes);
1064             /* wakeup the waiter and let it recheck */
1065             GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1066             break;
1067           case GST_EVENT_SEGMENT:
1068             gst_event_replace (&dlbuf->segment_event, event);
1069             /* a new segment allows us to accept more buffers if we got EOS
1070              * from downstream */
1071             dlbuf->unexpected = FALSE;
1072             break;
1073           case GST_EVENT_STREAM_START:
1074             gst_event_replace (&dlbuf->stream_start_event, event);
1075             break;
1076           default:
1077             break;
1078         }
1079         gst_event_unref (event);
1080         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1081       } else {
1082         /* non-serialized events are passed upstream. */
1083         ret = gst_pad_push_event (dlbuf->srcpad, event);
1084       }
1085       break;
1086   }
1087   return ret;
1088
1089   /* ERRORS */
1090 out_flushing:
1091   {
1092     GST_DEBUG_OBJECT (dlbuf, "refusing event, we are flushing");
1093     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1094     gst_event_unref (event);
1095     return FALSE;
1096   }
1097 }
1098
1099 static gboolean
1100 gst_download_buffer_handle_sink_query (GstPad * pad, GstObject * parent,
1101     GstQuery * query)
1102 {
1103   GstDownloadBuffer *dlbuf;
1104   gboolean res;
1105
1106   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1107
1108   switch (GST_QUERY_TYPE (query)) {
1109     default:
1110       if (GST_QUERY_IS_SERIALIZED (query)) {
1111         GST_LOG_OBJECT (dlbuf, "received query %p", query);
1112         GST_DEBUG_OBJECT (dlbuf, "refusing query, we are not using the dlbuf");
1113         res = FALSE;
1114       } else {
1115         res = gst_pad_query_default (pad, parent, query);
1116       }
1117       break;
1118   }
1119   return res;
1120 }
1121
1122 static GstFlowReturn
1123 gst_download_buffer_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
1124 {
1125   GstDownloadBuffer *dlbuf;
1126   GstMapInfo info;
1127   guint64 offset;
1128   gsize res, available;
1129   GError *error = NULL;
1130
1131   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1132
1133   GST_LOG_OBJECT (dlbuf, "received buffer %p of "
1134       "size %" G_GSIZE_FORMAT ", time %" GST_TIME_FORMAT ", duration %"
1135       GST_TIME_FORMAT, buffer, gst_buffer_get_size (buffer),
1136       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1137       GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1138
1139   /* we have to lock the dlbuf since we span threads */
1140   GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->sinkresult, out_flushing);
1141   /* when we received unexpected from downstream, refuse more buffers */
1142   if (dlbuf->unexpected)
1143     goto out_eos;
1144
1145   /* while we didn't receive the newsegment, we're seeking and we skip data */
1146   if (dlbuf->seeking)
1147     goto out_seeking;
1148
1149   /* put buffer in dlbuf now */
1150   offset = dlbuf->write_pos;
1151
1152   /* sanity check */
1153   if (GST_BUFFER_OFFSET_IS_VALID (buffer) &&
1154       GST_BUFFER_OFFSET (buffer) != offset) {
1155     GST_WARNING_OBJECT (dlbuf, "buffer offset does not match current writing "
1156         "position! %" G_GINT64_FORMAT " != %" G_GINT64_FORMAT,
1157         GST_BUFFER_OFFSET (buffer), offset);
1158   }
1159
1160   gst_buffer_map (buffer, &info, GST_MAP_READ);
1161
1162   GST_DEBUG_OBJECT (dlbuf, "Writing %" G_GSIZE_FORMAT " bytes to %"
1163       G_GUINT64_FORMAT, info.size, offset);
1164
1165   res =
1166       gst_sparse_file_write (dlbuf->file, offset, info.data, info.size,
1167       &available, &error);
1168   if (res == 0)
1169     goto write_error;
1170
1171   gst_buffer_unmap (buffer, &info);
1172   gst_buffer_unref (buffer);
1173
1174   dlbuf->write_pos = offset + info.size;
1175   dlbuf->bytes_in += info.size;
1176
1177   GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, dlbuf->write_pos + available);
1178
1179   /* we hit the end, see what to do */
1180   if (dlbuf->write_pos + available == dlbuf->upstream_size) {
1181     gsize start, stop;
1182
1183     /* we have everything up to the end, find a region to fill */
1184     if (gst_sparse_file_get_range_after (dlbuf->file, 0, &start, &stop)) {
1185       if (stop < dlbuf->upstream_size) {
1186         /* a hole to fill, seek to its end */
1187         perform_seek_to_offset (dlbuf, stop);
1188       } else {
1189         /* we filled all the holes, we are done */
1190         goto completed;
1191       }
1192     }
1193   } else {
1194     /* see if we need to skip this region or just read it again. The idea
1195      * is that when the region is not big, we want to avoid a seek and just
1196      * let it reread */
1197     guint64 threshold = get_seek_threshold (dlbuf);
1198
1199     if (available > threshold) {
1200       /* further than threshold, it's better to skip than to reread */
1201       perform_seek_to_offset (dlbuf, dlbuf->write_pos + available);
1202     }
1203   }
1204   if (dlbuf->filling) {
1205     if (dlbuf->write_pos > dlbuf->read_pos)
1206       update_levels (dlbuf, dlbuf->write_pos - dlbuf->read_pos);
1207     else
1208       update_levels (dlbuf, 0);
1209   }
1210
1211   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1212
1213   return GST_FLOW_OK;
1214
1215   /* ERRORS */
1216 out_flushing:
1217   {
1218     GstFlowReturn ret = dlbuf->sinkresult;
1219     GST_LOG_OBJECT (dlbuf,
1220         "exit because task paused, reason: %s", gst_flow_get_name (ret));
1221     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1222     gst_buffer_unref (buffer);
1223     return ret;
1224   }
1225 out_eos:
1226   {
1227     GST_LOG_OBJECT (dlbuf, "exit because we received EOS");
1228     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1229     gst_buffer_unref (buffer);
1230     return GST_FLOW_EOS;
1231   }
1232 out_seeking:
1233   {
1234     GST_LOG_OBJECT (dlbuf, "exit because we are seeking");
1235     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1236     gst_buffer_unref (buffer);
1237     return GST_FLOW_OK;
1238   }
1239 write_error:
1240   {
1241     gst_buffer_unmap (buffer, &info);
1242     gst_buffer_unref (buffer);
1243     GST_ELEMENT_ERROR (dlbuf, RESOURCE, WRITE,
1244         (_("Error while writing to download file.")), ("%s", error->message));
1245     g_clear_error (&error);
1246     return GST_FLOW_ERROR;
1247   }
1248 completed:
1249   {
1250     GST_LOG_OBJECT (dlbuf, "we completed the download");
1251     dlbuf->write_pos = dlbuf->upstream_size;
1252     dlbuf->filling = FALSE;
1253     update_levels (dlbuf, dlbuf->max_level.bytes);
1254     gst_element_post_message (GST_ELEMENT_CAST (dlbuf),
1255         gst_message_new_element (GST_OBJECT_CAST (dlbuf),
1256             gst_structure_new ("GstCacheDownloadComplete",
1257                 "location", G_TYPE_STRING, dlbuf->temp_location, NULL)));
1258     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1259
1260     return GST_FLOW_EOS;
1261   }
1262 }
1263
1264 /* called repeatedly with @pad as the source pad. This function should push out
1265  * data to the peer element. */
1266 static void
1267 gst_download_buffer_loop (GstPad * pad)
1268 {
1269   GstDownloadBuffer *dlbuf;
1270   GstFlowReturn ret;
1271   GstBuffer *buffer = NULL;
1272
1273   dlbuf = GST_DOWNLOAD_BUFFER (GST_PAD_PARENT (pad));
1274
1275   /* have to lock for thread-safety */
1276   GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1277
1278   if (dlbuf->stream_start_event != NULL) {
1279     gst_pad_push_event (dlbuf->srcpad, dlbuf->stream_start_event);
1280     dlbuf->stream_start_event = NULL;
1281   }
1282   if (dlbuf->segment_event != NULL) {
1283     gst_pad_push_event (dlbuf->srcpad, dlbuf->segment_event);
1284     dlbuf->segment_event = NULL;
1285   }
1286
1287   ret = gst_download_buffer_read_buffer (dlbuf, -1, -1, &buffer);
1288   if (ret != GST_FLOW_OK)
1289     goto out_flushing;
1290
1291   g_atomic_int_set (&dlbuf->downstream_may_block, 1);
1292   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1293
1294   ret = gst_pad_push (dlbuf->srcpad, buffer);
1295   g_atomic_int_set (&dlbuf->downstream_may_block, 0);
1296
1297   /* need to check for srcresult here as well */
1298   GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1299   dlbuf->srcresult = ret;
1300   dlbuf->sinkresult = ret;
1301   if (ret != GST_FLOW_OK)
1302     goto out_flushing;
1303   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1304
1305   return;
1306
1307   /* ERRORS */
1308 out_flushing:
1309   {
1310     GstFlowReturn ret = dlbuf->srcresult;
1311
1312     gst_pad_pause_task (dlbuf->srcpad);
1313     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1314     GST_LOG_OBJECT (dlbuf, "pause task, reason:  %s", gst_flow_get_name (ret));
1315     /* let app know about us giving up if upstream is not expected to do so */
1316     if (ret == GST_FLOW_EOS) {
1317       /* FIXME perform EOS logic, this is really a basesrc operating on a
1318        * file. */
1319       gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1320     } else if ((ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS)) {
1321       GST_ELEMENT_ERROR (dlbuf, STREAM, FAILED,
1322           (_("Internal data flow error.")),
1323           ("streaming task paused, reason %s (%d)",
1324               gst_flow_get_name (ret), ret));
1325       gst_pad_push_event (dlbuf->srcpad, gst_event_new_eos ());
1326     }
1327     return;
1328   }
1329 }
1330
1331 static gboolean
1332 gst_download_buffer_handle_src_event (GstPad * pad, GstObject * parent,
1333     GstEvent * event)
1334 {
1335   gboolean res = TRUE;
1336   GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (parent);
1337
1338 #ifndef GST_DISABLE_GST_DEBUG
1339   GST_DEBUG_OBJECT (dlbuf, "got event %p (%s)",
1340       event, GST_EVENT_TYPE_NAME (event));
1341 #endif
1342
1343   switch (GST_EVENT_TYPE (event)) {
1344     case GST_EVENT_FLUSH_START:
1345       /* now unblock the getrange function */
1346       GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1347       GST_DEBUG_OBJECT (dlbuf, "flushing");
1348       dlbuf->srcresult = GST_FLOW_FLUSHING;
1349       GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1350       GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1351
1352       /* when using a temp file, we eat the event */
1353       res = TRUE;
1354       gst_event_unref (event);
1355       break;
1356     case GST_EVENT_FLUSH_STOP:
1357       /* now unblock the getrange function */
1358       GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1359       dlbuf->srcresult = GST_FLOW_OK;
1360       GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1361
1362       /* when using a temp file, we eat the event */
1363       res = TRUE;
1364       gst_event_unref (event);
1365       break;
1366     case GST_EVENT_RECONFIGURE:
1367       GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1368       /* assume downstream is linked now and try to push again */
1369       if (dlbuf->srcresult == GST_FLOW_NOT_LINKED) {
1370         dlbuf->srcresult = GST_FLOW_OK;
1371         dlbuf->sinkresult = GST_FLOW_OK;
1372         if (GST_PAD_MODE (pad) == GST_PAD_MODE_PUSH) {
1373           gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1374               pad, NULL);
1375         }
1376       }
1377       GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1378
1379       res = gst_pad_push_event (dlbuf->sinkpad, event);
1380       break;
1381     default:
1382       res = gst_pad_push_event (dlbuf->sinkpad, event);
1383       break;
1384   }
1385
1386   return res;
1387 }
1388
1389 static gboolean
1390 gst_download_buffer_handle_src_query (GstPad * pad, GstObject * parent,
1391     GstQuery * query)
1392 {
1393   GstDownloadBuffer *dlbuf;
1394
1395   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1396
1397   switch (GST_QUERY_TYPE (query)) {
1398     case GST_QUERY_POSITION:
1399     {
1400       gint64 peer_pos;
1401       GstFormat format;
1402
1403       if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1404         goto peer_failed;
1405
1406       /* get peer position */
1407       gst_query_parse_position (query, &format, &peer_pos);
1408
1409       /* FIXME: this code assumes that there's no discont in the dlbuf */
1410       switch (format) {
1411         case GST_FORMAT_BYTES:
1412           peer_pos -= dlbuf->cur_level.bytes;
1413           break;
1414         case GST_FORMAT_TIME:
1415           peer_pos -= dlbuf->cur_level.time;
1416           break;
1417         default:
1418           GST_WARNING_OBJECT (dlbuf, "dropping query in %s format, don't "
1419               "know how to adjust value", gst_format_get_name (format));
1420           return FALSE;
1421       }
1422       /* set updated position */
1423       gst_query_set_position (query, format, peer_pos);
1424       break;
1425     }
1426     case GST_QUERY_DURATION:
1427     {
1428       GST_DEBUG_OBJECT (dlbuf, "doing peer query");
1429
1430       if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1431         goto peer_failed;
1432
1433       GST_DEBUG_OBJECT (dlbuf, "peer query success");
1434       break;
1435     }
1436     case GST_QUERY_BUFFERING:
1437     {
1438       gint percent;
1439       gboolean is_buffering;
1440       GstBufferingMode mode;
1441       gint avg_in, avg_out;
1442       gint64 buffering_left;
1443
1444       GST_DEBUG_OBJECT (dlbuf, "query buffering");
1445
1446       get_buffering_percent (dlbuf, &is_buffering, &percent);
1447       gst_query_set_buffering_percent (query, is_buffering, percent);
1448
1449       get_buffering_stats (dlbuf, percent, &mode, &avg_in, &avg_out,
1450           &buffering_left);
1451       gst_query_set_buffering_stats (query, mode, avg_in, avg_out,
1452           buffering_left);
1453
1454       {
1455         /* add ranges for download and ringbuffer buffering */
1456         GstFormat format;
1457         gint64 start, stop;
1458         guint64 write_pos;
1459         gint64 estimated_total;
1460         gint64 duration;
1461         gsize offset, range_start, range_stop;
1462
1463         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1464         write_pos = dlbuf->write_pos;
1465
1466         /* get duration of upstream in bytes */
1467         gst_download_buffer_update_upstream_size (dlbuf);
1468         duration = dlbuf->upstream_size;
1469
1470         GST_DEBUG_OBJECT (dlbuf, "percent %d, duration %" G_GINT64_FORMAT
1471             ", writing %" G_GINT64_FORMAT, percent, duration, write_pos);
1472
1473         gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1474
1475         /* fill out the buffered ranges */
1476         start = offset = 0;
1477         stop = -1;
1478         estimated_total = -1;
1479         while (gst_sparse_file_get_range_after (dlbuf->file, offset,
1480                 &range_start, &range_stop)) {
1481           gboolean current_range;
1482
1483           GST_DEBUG_OBJECT (dlbuf,
1484               "range starting at %" G_GSIZE_FORMAT " and finishing at %"
1485               G_GSIZE_FORMAT, range_start, range_stop);
1486
1487           offset = range_stop;
1488
1489           /* find the range we are currently downloading, we'll remember it
1490            * after we convert to the target format */
1491           if (range_start <= write_pos && range_stop >= write_pos) {
1492             current_range = TRUE;
1493             /* calculate remaining and total download time */
1494             if (duration >= range_stop && avg_in > 0.0)
1495               estimated_total = ((duration - range_stop) * 1000) / avg_in;
1496           } else
1497             current_range = FALSE;
1498
1499           switch (format) {
1500             case GST_FORMAT_PERCENT:
1501               /* get our available data relative to the duration */
1502               if (duration == -1) {
1503                 range_start = 0;
1504                 range_stop = 0;
1505               } else {
1506                 range_start = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1507                     range_start, duration);
1508                 range_stop = gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
1509                     range_stop, duration);
1510               }
1511               break;
1512             case GST_FORMAT_BYTES:
1513               break;
1514             default:
1515               range_start = -1;
1516               range_stop = -1;
1517               break;
1518           }
1519
1520           if (current_range) {
1521             /* we are currently downloading this range */
1522             start = range_start;
1523             stop = range_stop;
1524           }
1525           GST_DEBUG_OBJECT (dlbuf,
1526               "range to format: %" G_GSIZE_FORMAT " - %" G_GSIZE_FORMAT,
1527               range_start, range_stop);
1528           if (range_start == range_stop)
1529             continue;
1530           gst_query_add_buffering_range (query, range_start, range_stop);
1531         }
1532
1533         GST_DEBUG_OBJECT (dlbuf, "estimated-total %" G_GINT64_FORMAT,
1534             estimated_total);
1535
1536         gst_query_set_buffering_range (query, format, start, stop,
1537             estimated_total);
1538
1539         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1540       }
1541       break;
1542     }
1543     case GST_QUERY_SCHEDULING:
1544     {
1545       GstSchedulingFlags flags = 0;
1546
1547       if (!gst_pad_peer_query (dlbuf->sinkpad, query))
1548         goto peer_failed;
1549
1550       gst_query_parse_scheduling (query, &flags, NULL, NULL, NULL);
1551
1552       /* we can operate in pull mode when we are using a tempfile */
1553       flags |= GST_SCHEDULING_FLAG_SEEKABLE;
1554       gst_query_set_scheduling (query, flags, 0, -1, 0);
1555       gst_query_add_scheduling_mode (query, GST_PAD_MODE_PULL);
1556       gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
1557       break;
1558     }
1559     default:
1560       /* peer handled other queries */
1561       if (!gst_pad_query_default (pad, parent, query))
1562         goto peer_failed;
1563       break;
1564   }
1565
1566   return TRUE;
1567
1568   /* ERRORS */
1569 peer_failed:
1570   {
1571     GST_DEBUG_OBJECT (dlbuf, "failed peer query");
1572     return FALSE;
1573   }
1574 }
1575
1576 static gboolean
1577 gst_download_buffer_handle_query (GstElement * element, GstQuery * query)
1578 {
1579   GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (element);
1580
1581   /* simply forward to the srcpad query function */
1582   return gst_download_buffer_handle_src_query (dlbuf->srcpad,
1583       GST_OBJECT_CAST (element), query);
1584 }
1585
1586 static GstFlowReturn
1587 gst_download_buffer_get_range (GstPad * pad, GstObject * parent, guint64 offset,
1588     guint length, GstBuffer ** buffer)
1589 {
1590   GstDownloadBuffer *dlbuf;
1591   GstFlowReturn ret;
1592
1593   dlbuf = GST_DOWNLOAD_BUFFER_CAST (parent);
1594
1595   GST_DOWNLOAD_BUFFER_MUTEX_LOCK_CHECK (dlbuf, dlbuf->srcresult, out_flushing);
1596   /* FIXME - function will block when the range is not yet available */
1597   ret = gst_download_buffer_read_buffer (dlbuf, offset, length, buffer);
1598   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1599
1600   return ret;
1601
1602   /* ERRORS */
1603 out_flushing:
1604   {
1605     ret = dlbuf->srcresult;
1606
1607     GST_DEBUG_OBJECT (dlbuf, "we are flushing");
1608     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1609     return ret;
1610   }
1611 }
1612
1613 /* sink currently only operates in push mode */
1614 static gboolean
1615 gst_download_buffer_sink_activate_mode (GstPad * pad, GstObject * parent,
1616     GstPadMode mode, gboolean active)
1617 {
1618   gboolean result;
1619   GstDownloadBuffer *dlbuf;
1620
1621   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1622
1623   switch (mode) {
1624     case GST_PAD_MODE_PUSH:
1625       if (active) {
1626         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1627         GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1628         dlbuf->srcresult = GST_FLOW_OK;
1629         dlbuf->sinkresult = GST_FLOW_OK;
1630         dlbuf->unexpected = FALSE;
1631         reset_rate_timer (dlbuf);
1632         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1633       } else {
1634         /* unblock chain function */
1635         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1636         GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1637         dlbuf->srcresult = GST_FLOW_FLUSHING;
1638         dlbuf->sinkresult = GST_FLOW_FLUSHING;
1639         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1640
1641         /* wait until it is unblocked and clean up */
1642         GST_PAD_STREAM_LOCK (pad);
1643         GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1644         gst_download_buffer_locked_flush (dlbuf, TRUE, FALSE);
1645         GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1646         GST_PAD_STREAM_UNLOCK (pad);
1647       }
1648       result = TRUE;
1649       break;
1650     default:
1651       result = FALSE;
1652       break;
1653   }
1654   return result;
1655 }
1656
1657 /* src operating in push mode, we start a task on the source pad that pushes out
1658  * buffers from the dlbuf */
1659 static gboolean
1660 gst_download_buffer_src_activate_push (GstPad * pad, GstObject * parent,
1661     gboolean active)
1662 {
1663   gboolean result = FALSE;
1664   GstDownloadBuffer *dlbuf;
1665
1666   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1667
1668   if (active) {
1669     GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1670     GST_DEBUG_OBJECT (dlbuf, "activating push mode");
1671     dlbuf->srcresult = GST_FLOW_OK;
1672     dlbuf->sinkresult = GST_FLOW_OK;
1673     dlbuf->unexpected = FALSE;
1674     result =
1675         gst_pad_start_task (pad, (GstTaskFunction) gst_download_buffer_loop,
1676         pad, NULL);
1677     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1678   } else {
1679     /* unblock loop function */
1680     GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1681     GST_DEBUG_OBJECT (dlbuf, "deactivating push mode");
1682     dlbuf->srcresult = GST_FLOW_FLUSHING;
1683     dlbuf->sinkresult = GST_FLOW_FLUSHING;
1684     /* the item add signal will unblock */
1685     GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1686     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1687
1688     /* step 2, make sure streaming finishes */
1689     result = gst_pad_stop_task (pad);
1690   }
1691
1692   return result;
1693 }
1694
1695 /* pull mode, downstream will call our getrange function */
1696 static gboolean
1697 gst_download_buffer_src_activate_pull (GstPad * pad, GstObject * parent,
1698     gboolean active)
1699 {
1700   gboolean result;
1701   GstDownloadBuffer *dlbuf;
1702
1703   dlbuf = GST_DOWNLOAD_BUFFER (parent);
1704
1705   if (active) {
1706     GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1707     /* open the temp file now */
1708     result = gst_download_buffer_open_temp_location_file (dlbuf);
1709     GST_DEBUG_OBJECT (dlbuf, "activating pull mode");
1710     dlbuf->srcresult = GST_FLOW_OK;
1711     dlbuf->sinkresult = GST_FLOW_OK;
1712     dlbuf->unexpected = FALSE;
1713     dlbuf->upstream_size = 0;
1714     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1715   } else {
1716     GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1717     GST_DEBUG_OBJECT (dlbuf, "deactivating pull mode");
1718     dlbuf->srcresult = GST_FLOW_FLUSHING;
1719     dlbuf->sinkresult = GST_FLOW_FLUSHING;
1720     /* this will unlock getrange */
1721     GST_DOWNLOAD_BUFFER_SIGNAL_ADD (dlbuf, -1);
1722     result = TRUE;
1723     GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1724   }
1725
1726   return result;
1727 }
1728
1729 static gboolean
1730 gst_download_buffer_src_activate_mode (GstPad * pad, GstObject * parent,
1731     GstPadMode mode, gboolean active)
1732 {
1733   gboolean res;
1734
1735   switch (mode) {
1736     case GST_PAD_MODE_PULL:
1737       res = gst_download_buffer_src_activate_pull (pad, parent, active);
1738       break;
1739     case GST_PAD_MODE_PUSH:
1740       res = gst_download_buffer_src_activate_push (pad, parent, active);
1741       break;
1742     default:
1743       GST_LOG_OBJECT (pad, "unknown activation mode %d", mode);
1744       res = FALSE;
1745       break;
1746   }
1747   return res;
1748 }
1749
1750 static GstStateChangeReturn
1751 gst_download_buffer_change_state (GstElement * element,
1752     GstStateChange transition)
1753 {
1754   GstDownloadBuffer *dlbuf;
1755   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
1756
1757   dlbuf = GST_DOWNLOAD_BUFFER (element);
1758
1759   switch (transition) {
1760     case GST_STATE_CHANGE_NULL_TO_READY:
1761       break;
1762     case GST_STATE_CHANGE_READY_TO_PAUSED:
1763       GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1764       if (!gst_download_buffer_open_temp_location_file (dlbuf))
1765         ret = GST_STATE_CHANGE_FAILURE;
1766       gst_event_replace (&dlbuf->stream_start_event, NULL);
1767       gst_event_replace (&dlbuf->segment_event, NULL);
1768       GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1769       break;
1770     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1771       break;
1772     default:
1773       break;
1774   }
1775
1776   if (ret == GST_STATE_CHANGE_FAILURE)
1777     return ret;
1778
1779   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1780
1781   if (ret == GST_STATE_CHANGE_FAILURE)
1782     return ret;
1783
1784   switch (transition) {
1785     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1786       break;
1787     case GST_STATE_CHANGE_PAUSED_TO_READY:
1788       GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1789       gst_download_buffer_close_temp_location_file (dlbuf);
1790       gst_event_replace (&dlbuf->stream_start_event, NULL);
1791       gst_event_replace (&dlbuf->segment_event, NULL);
1792       GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1793       break;
1794     case GST_STATE_CHANGE_READY_TO_NULL:
1795       break;
1796     default:
1797       break;
1798   }
1799
1800   return ret;
1801 }
1802
1803 #define CAPACITY_CHANGE(elem) \
1804   update_buffering (elem);
1805
1806 static void
1807 gst_download_buffer_set_temp_template (GstDownloadBuffer * dlbuf,
1808     const gchar * template)
1809 {
1810   GstState state;
1811
1812   /* the element must be stopped in order to do this */
1813   GST_OBJECT_LOCK (dlbuf);
1814   state = GST_STATE (dlbuf);
1815   if (state != GST_STATE_READY && state != GST_STATE_NULL)
1816     goto wrong_state;
1817   GST_OBJECT_UNLOCK (dlbuf);
1818
1819   /* set new location */
1820   g_free (dlbuf->temp_template);
1821   dlbuf->temp_template = g_strdup (template);
1822
1823   return;
1824
1825 /* ERROR */
1826 wrong_state:
1827   {
1828     GST_WARNING_OBJECT (dlbuf, "setting temp-template property in wrong state");
1829     GST_OBJECT_UNLOCK (dlbuf);
1830   }
1831 }
1832
1833 static void
1834 gst_download_buffer_set_property (GObject * object,
1835     guint prop_id, const GValue * value, GParamSpec * pspec)
1836 {
1837   GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1838
1839   /* someone could change levels here, and since this
1840    * affects the get/put funcs, we need to lock for safety. */
1841   GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1842
1843   switch (prop_id) {
1844     case PROP_MAX_SIZE_BYTES:
1845       dlbuf->max_level.bytes = g_value_get_uint (value);
1846       CAPACITY_CHANGE (dlbuf);
1847       break;
1848     case PROP_MAX_SIZE_TIME:
1849       dlbuf->max_level.time = g_value_get_uint64 (value);
1850       CAPACITY_CHANGE (dlbuf);
1851       break;
1852     case PROP_LOW_PERCENT:
1853       dlbuf->low_percent = g_value_get_int (value);
1854       break;
1855     case PROP_HIGH_PERCENT:
1856       dlbuf->high_percent = g_value_get_int (value);
1857       break;
1858     case PROP_TEMP_TEMPLATE:
1859       gst_download_buffer_set_temp_template (dlbuf, g_value_get_string (value));
1860       break;
1861     case PROP_TEMP_REMOVE:
1862       dlbuf->temp_remove = g_value_get_boolean (value);
1863       break;
1864     default:
1865       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1866       break;
1867   }
1868
1869   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1870 }
1871
1872 static void
1873 gst_download_buffer_get_property (GObject * object,
1874     guint prop_id, GValue * value, GParamSpec * pspec)
1875 {
1876   GstDownloadBuffer *dlbuf = GST_DOWNLOAD_BUFFER (object);
1877
1878   GST_DOWNLOAD_BUFFER_MUTEX_LOCK (dlbuf);
1879
1880   switch (prop_id) {
1881     case PROP_MAX_SIZE_BYTES:
1882       g_value_set_uint (value, dlbuf->max_level.bytes);
1883       break;
1884     case PROP_MAX_SIZE_TIME:
1885       g_value_set_uint64 (value, dlbuf->max_level.time);
1886       break;
1887     case PROP_LOW_PERCENT:
1888       g_value_set_int (value, dlbuf->low_percent);
1889       break;
1890     case PROP_HIGH_PERCENT:
1891       g_value_set_int (value, dlbuf->high_percent);
1892       break;
1893     case PROP_TEMP_TEMPLATE:
1894       g_value_set_string (value, dlbuf->temp_template);
1895       break;
1896     case PROP_TEMP_LOCATION:
1897       g_value_set_string (value, dlbuf->temp_location);
1898       break;
1899     case PROP_TEMP_REMOVE:
1900       g_value_set_boolean (value, dlbuf->temp_remove);
1901       break;
1902     default:
1903       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1904       break;
1905   }
1906
1907   GST_DOWNLOAD_BUFFER_MUTEX_UNLOCK (dlbuf);
1908 }