plugins/elements/gstqueue2.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2003 Colin Walters <cwalters@gnome.org>
   4  *                    2000,2005,2007 Wim Taymans <wim.taymans@gmail.com>
   5  *                    2007 Thiago Sousa Santos <thiagoss@lcc.ufcg.edu.br>
   6  *                 SA 2010 ST-Ericsson <benjamin.gaignard@stericsson.com>
   7  *
   8  * gstqueue2.c:
   9  *
  10  * This library is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Library General Public
  12  * License as published by the Free Software Foundation; either
  13  * version 2 of the License, or (at your option) any later version.
  14  *
  15  * This library is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18  * Library General Public License for more details.
  19  *
  20  * You should have received a copy of the GNU Library General Public
  21  * License along with this library; if not, write to the
  22  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  23  * Boston, MA 02110-1301, USA.
  24  */
  25
  26 /**
  27  * SECTION:element-queue2
  28  *
  29  * Data is queued until one of the limits specified by the
  30  * #GstQueue2:max-size-buffers, #GstQueue2:max-size-bytes and/or
  31  * #GstQueue2:max-size-time properties has been reached. Any attempt to push
  32  * more buffers into the queue will block the pushing thread until more space
  33  * becomes available.
  34  *
  35  * The queue will create a new thread on the source pad to decouple the
  36  * processing on sink and source pad.
  37  *
  38  * You can query how many buffers are queued by reading the
  39  * #GstQueue2:current-level-buffers property.
  40  *
  41  * The default queue size limits are 100 buffers, 2MB of data, or
  42  * two seconds worth of data, whichever is reached first.
  43  *
  44  * If you set temp-template to a value such as /tmp/gstreamer-XXXXXX, the element
  45  * will allocate a random free filename and buffer data in the file.
  46  * By using this, it will buffer the entire stream data on the file independently
  47  * of the queue size limits, they will only be used for buffering statistics.
  48  *
  49  * The temp-location property will be used to notify the application of the
  50  * allocated filename.
  51  */
  52
  53 #ifdef HAVE_CONFIG_H
  54 #include "config.h"
  55 #endif
  56
  57 #include "gstqueue2.h"
  58
  59 #include <glib/gstdio.h>
  60
  61 #include "gst/gst-i18n-lib.h"
  62 #include "gst/glib-compat-private.h"
  63
  64 #include <string.h>
  65
  66 #ifdef G_OS_WIN32
  67 #include <io.h>                 /* lseek, open, close, read */
  68 #undef lseek
  69 #define lseek _lseeki64
  70 #undef off_t
  71 #define off_t guint64
  72 #else
  73 #include <unistd.h>
  74 #endif
  75
  76 #ifdef __BIONIC__               /* Android */
  77 #undef lseek
  78 #define lseek lseek64
  79 #undef off_t
  80 #define off_t guint64
  81 #include <fcntl.h>
  82 #endif
  83
  84 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  85     GST_PAD_SINK,
  86     GST_PAD_ALWAYS,
  87     GST_STATIC_CAPS_ANY);
  88
  89 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  90     GST_PAD_SRC,
  91     GST_PAD_ALWAYS,
  92     GST_STATIC_CAPS_ANY);
  93
  94 GST_DEBUG_CATEGORY_STATIC (queue_debug);
  95 #define GST_CAT_DEFAULT (queue_debug)
  96 GST_DEBUG_CATEGORY_STATIC (queue_dataflow);
  97
  98 enum
  99 {
 100   LAST_SIGNAL
 101 };
 102
 103 /* other defines */
 104 #define DEFAULT_BUFFER_SIZE 4096
 105 #define QUEUE_IS_USING_TEMP_FILE(queue) ((queue)->temp_template != NULL)
 106 #define QUEUE_IS_USING_RING_BUFFER(queue) ((queue)->ring_buffer_max_size != 0)  /* for consistency with the above macro */
 107 #define QUEUE_IS_USING_QUEUE(queue) (!QUEUE_IS_USING_TEMP_FILE(queue) && !QUEUE_IS_USING_RING_BUFFER (queue))
 108
 109 #define QUEUE_MAX_BYTES(queue) MIN((queue)->max_level.bytes, (queue)->ring_buffer_max_size)
 110
 111 /* default property values */
 112 #define DEFAULT_MAX_SIZE_BUFFERS   100  /* 100 buffers */
 113 #define DEFAULT_MAX_SIZE_BYTES     (2 * 1024 * 1024)    /* 2 MB */
 114 #define DEFAULT_MAX_SIZE_TIME      2 * GST_SECOND       /* 2 seconds */
 115 #define DEFAULT_USE_BUFFERING      FALSE
 116 #define DEFAULT_USE_TAGS_BITRATE   FALSE
 117 #define DEFAULT_USE_RATE_ESTIMATE  TRUE
 118 #define DEFAULT_LOW_PERCENT        10
 119 #define DEFAULT_HIGH_PERCENT       99
 120 #define DEFAULT_TEMP_REMOVE        TRUE
 121 #define DEFAULT_RING_BUFFER_MAX_SIZE 0
 122
 123 enum
 124 {
 125   PROP_0,
 126   PROP_CUR_LEVEL_BUFFERS,
 127   PROP_CUR_LEVEL_BYTES,
 128   PROP_CUR_LEVEL_TIME,
 129   PROP_MAX_SIZE_BUFFERS,
 130   PROP_MAX_SIZE_BYTES,
 131   PROP_MAX_SIZE_TIME,
 132   PROP_USE_BUFFERING,
 133   PROP_USE_TAGS_BITRATE,
 134   PROP_USE_RATE_ESTIMATE,
 135   PROP_LOW_PERCENT,
 136   PROP_HIGH_PERCENT,
 137   PROP_TEMP_TEMPLATE,
 138   PROP_TEMP_LOCATION,
 139   PROP_TEMP_REMOVE,
 140   PROP_RING_BUFFER_MAX_SIZE,
 141   PROP_AVG_IN_RATE,
 142   PROP_LAST
 143 };
 144
 145 #define GST_QUEUE2_CLEAR_LEVEL(l) G_STMT_START {         \
 146   l.buffers = 0;                                        \
 147   l.bytes = 0;                                          \
 148   l.time = 0;                                           \
 149   l.rate_time = 0;                                      \
 150 } G_STMT_END
 151
 152 #define STATUS(queue, pad, msg) \
 153   GST_CAT_LOG_OBJECT (queue_dataflow, queue, \
 154                       "(%s:%s) " msg ": %u of %u buffers, %u of %u " \
 155                       "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
 156                       " ns, %"G_GUINT64_FORMAT" items", \
 157                       GST_DEBUG_PAD_NAME (pad), \
 158                       queue->cur_level.buffers, \
 159                       queue->max_level.buffers, \
 160                       queue->cur_level.bytes, \
 161                       queue->max_level.bytes, \
 162                       queue->cur_level.time, \
 163                       queue->max_level.time, \
 164                       (guint64) (!QUEUE_IS_USING_QUEUE(queue) ? \
 165                         queue->current->writing_pos - queue->current->max_reading_pos : \
 166                         queue->queue.length))
 167
 168 #define GST_QUEUE2_MUTEX_LOCK(q) G_STMT_START {                          \
 169   g_mutex_lock (&q->qlock);                                              \
 170 } G_STMT_END
 171
 172 #define GST_QUEUE2_MUTEX_LOCK_CHECK(q,res,label) G_STMT_START {         \
 173   GST_QUEUE2_MUTEX_LOCK (q);                                            \
 174   if (res != GST_FLOW_OK)                                               \
 175     goto label;                                                         \
 176 } G_STMT_END
 177
 178 #define GST_QUEUE2_MUTEX_UNLOCK(q) G_STMT_START {                        \
 179   g_mutex_unlock (&q->qlock);                                            \
 180 } G_STMT_END
 181
 182 #define GST_QUEUE2_WAIT_DEL_CHECK(q, res, label) G_STMT_START {         \
 183   STATUS (queue, q->sinkpad, "wait for DEL");                           \
 184   q->waiting_del = TRUE;                                                \
 185   g_cond_wait (&q->item_del, &queue->qlock);                              \
 186   q->waiting_del = FALSE;                                               \
 187   if (res != GST_FLOW_OK) {                                             \
 188     STATUS (queue, q->srcpad, "received DEL wakeup");                   \
 189     goto label;                                                         \
 190   }                                                                     \
 191   STATUS (queue, q->sinkpad, "received DEL");                           \
 192 } G_STMT_END
 193
 194 #define GST_QUEUE2_WAIT_ADD_CHECK(q, res, label) G_STMT_START {         \
 195   STATUS (queue, q->srcpad, "wait for ADD");                            \
 196   q->waiting_add = TRUE;                                                \
 197   g_cond_wait (&q->item_add, &q->qlock);                                  \
 198   q->waiting_add = FALSE;                                               \
 199   if (res != GST_FLOW_OK) {                                             \
 200     STATUS (queue, q->srcpad, "received ADD wakeup");                   \
 201     goto label;                                                         \
 202   }                                                                     \
 203   STATUS (queue, q->srcpad, "received ADD");                            \
 204 } G_STMT_END
 205
 206 #define GST_QUEUE2_SIGNAL_DEL(q) G_STMT_START {                          \
 207   if (q->waiting_del) {                                                 \
 208     STATUS (q, q->srcpad, "signal DEL");                                \
 209     g_cond_signal (&q->item_del);                                        \
 210   }                                                                     \
 211 } G_STMT_END
 212
 213 #define GST_QUEUE2_SIGNAL_ADD(q) G_STMT_START {                          \
 214   if (q->waiting_add) {                                                 \
 215     STATUS (q, q->sinkpad, "signal ADD");                               \
 216     g_cond_signal (&q->item_add);                                        \
 217   }                                                                     \
 218 } G_STMT_END
 219
 220 #define SET_PERCENT(q, perc) G_STMT_START {                              \
 221   if (perc != q->buffering_percent) {                                    \
 222     q->buffering_percent = perc;                                         \
 223     q->percent_changed = TRUE;                                           \
 224     GST_DEBUG_OBJECT (q, "buffering %d percent", perc);                  \
 225     get_buffering_stats (q, perc, &q->mode, &q->avg_in, &q->avg_out,     \
 226         &q->buffering_left);                                             \
 227   }                                                                      \
 228 } G_STMT_END
 229
 230 #define _do_init \
 231     GST_DEBUG_CATEGORY_INIT (queue_debug, "queue2", 0, "queue element"); \
 232     GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue2_dataflow", 0, \
 233         "dataflow inside the queue element");
 234 #define gst_queue2_parent_class parent_class
 235 G_DEFINE_TYPE_WITH_CODE (GstQueue2, gst_queue2, GST_TYPE_ELEMENT, _do_init);
 236
 237 static void gst_queue2_finalize (GObject * object);
 238
 239 static void gst_queue2_set_property (GObject * object,
 240     guint prop_id, const GValue * value, GParamSpec * pspec);
 241 static void gst_queue2_get_property (GObject * object,
 242     guint prop_id, GValue * value, GParamSpec * pspec);
 243
 244 static GstFlowReturn gst_queue2_chain (GstPad * pad, GstObject * parent,
 245     GstBuffer * buffer);
 246 static GstFlowReturn gst_queue2_chain_list (GstPad * pad, GstObject * parent,
 247     GstBufferList * buffer_list);
 248 static GstFlowReturn gst_queue2_push_one (GstQueue2 * queue);
 249 static void gst_queue2_loop (GstPad * pad);
 250
 251 static gboolean gst_queue2_handle_sink_event (GstPad * pad, GstObject * parent,
 252     GstEvent * event);
 253 static gboolean gst_queue2_handle_sink_query (GstPad * pad, GstObject * parent,
 254     GstQuery * query);
 255
 256 static gboolean gst_queue2_handle_src_event (GstPad * pad, GstObject * parent,
 257     GstEvent * event);
 258 static gboolean gst_queue2_handle_src_query (GstPad * pad, GstObject * parent,
 259     GstQuery * query);
 260 static gboolean gst_queue2_handle_query (GstElement * element,
 261     GstQuery * query);
 262
 263 static GstFlowReturn gst_queue2_get_range (GstPad * pad, GstObject * parent,
 264     guint64 offset, guint length, GstBuffer ** buffer);
 265
 266 static gboolean gst_queue2_src_activate_mode (GstPad * pad, GstObject * parent,
 267     GstPadMode mode, gboolean active);
 268 static gboolean gst_queue2_sink_activate_mode (GstPad * pad, GstObject * parent,
 269     GstPadMode mode, gboolean active);
 270 static GstStateChangeReturn gst_queue2_change_state (GstElement * element,
 271     GstStateChange transition);
 272
 273 static gboolean gst_queue2_is_empty (GstQueue2 * queue);
 274 static gboolean gst_queue2_is_filled (GstQueue2 * queue);
 275
 276 static void update_cur_level (GstQueue2 * queue, GstQueue2Range * range);
 277 static void update_in_rates (GstQueue2 * queue);
 278 static void gst_queue2_post_buffering (GstQueue2 * queue);
 279
 280 typedef enum
 281 {
 282   GST_QUEUE2_ITEM_TYPE_UNKNOWN = 0,
 283   GST_QUEUE2_ITEM_TYPE_BUFFER,
 284   GST_QUEUE2_ITEM_TYPE_BUFFER_LIST,
 285   GST_QUEUE2_ITEM_TYPE_EVENT,
 286   GST_QUEUE2_ITEM_TYPE_QUERY
 287 } GstQueue2ItemType;
 288
 289 typedef struct
 290 {
 291   GstQueue2ItemType type;
 292   GstMiniObject *item;
 293 } GstQueue2Item;
 294
 295 /* static guint gst_queue2_signals[LAST_SIGNAL] = { 0 }; */
 296
 297 static void
 298 gst_queue2_class_init (GstQueue2Class * klass)
 299 {
 300   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 301   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 302
 303   gobject_class->set_property = gst_queue2_set_property;
 304   gobject_class->get_property = gst_queue2_get_property;
 305
 306   /* properties */
 307   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,
 308       g_param_spec_uint ("current-level-bytes", "Current level (kB)",
 309           "Current amount of data in the queue (bytes)",
 310           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 311   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BUFFERS,
 312       g_param_spec_uint ("current-level-buffers", "Current level (buffers)",
 313           "Current number of buffers in the queue",
 314           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 315   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,
 316       g_param_spec_uint64 ("current-level-time", "Current level (ns)",
 317           "Current amount of data in the queue (in ns)",
 318           0, G_MAXUINT64, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 319
 320   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
 321       g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
 322           "Max. amount of data in the queue (bytes, 0=disable)",
 323           0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
 324           G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
 325           G_PARAM_STATIC_STRINGS));
 326   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BUFFERS,
 327       g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
 328           "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
 329           DEFAULT_MAX_SIZE_BUFFERS,
 330           G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
 331           G_PARAM_STATIC_STRINGS));
 332   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 333       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 334           "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
 335           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
 336           G_PARAM_STATIC_STRINGS));
 337
 338   g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,
 339       g_param_spec_boolean ("use-buffering", "Use buffering",
 340           "Emit GST_MESSAGE_BUFFERING based on low-/high-percent thresholds",
 341           DEFAULT_USE_BUFFERING, G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
 342           G_PARAM_STATIC_STRINGS));
 343   g_object_class_install_property (gobject_class, PROP_USE_TAGS_BITRATE,
 344       g_param_spec_boolean ("use-tags-bitrate", "Use bitrate from tags",
 345           "Use a bitrate from upstream tags to estimate buffer duration if not provided",
 346           DEFAULT_USE_TAGS_BITRATE,
 347           G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING |
 348           G_PARAM_STATIC_STRINGS));
 349   g_object_class_install_property (gobject_class, PROP_USE_RATE_ESTIMATE,
 350       g_param_spec_boolean ("use-rate-estimate", "Use Rate Estimate",
 351           "Estimate the bitrate of the stream to calculate time level",
 352           DEFAULT_USE_RATE_ESTIMATE,
 353           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 354   g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
 355       g_param_spec_int ("low-percent", "Low percent",
 356           "Low threshold for buffering to start. Only used if use-buffering is True",
 357           0, 100, DEFAULT_LOW_PERCENT,
 358           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 359   g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
 360       g_param_spec_int ("high-percent", "High percent",
 361           "High threshold for buffering to finish. Only used if use-buffering is True",
 362           0, 100, DEFAULT_HIGH_PERCENT,
 363           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 364
 365   g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
 366       g_param_spec_string ("temp-template", "Temporary File Template",
 367           "File template to store temporary files in, should contain directory "
 368           "and XXXXXX. (NULL == disabled)",
 369           NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 370
 371   g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
 372       g_param_spec_string ("temp-location", "Temporary File Location",
 373           "Location to store temporary files in (Only read this property, "
 374           "use temp-template to configure the name template)",
 375           NULL, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 376
 377   /**
 378    * GstQueue2:temp-remove
 379    *
 380    * When temp-template is set, remove the temporary file when going to READY.
 381    */
 382   g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
 383       g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
 384           "Remove the temp-location after use",
 385           DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 386
 387   /**
 388    * GstQueue2:ring-buffer-max-size
 389    *
 390    * The maximum size of the ring buffer in bytes. If set to 0, the ring
 391    * buffer is disabled. Default 0.
 392    */
 393   g_object_class_install_property (gobject_class, PROP_RING_BUFFER_MAX_SIZE,
 394       g_param_spec_uint64 ("ring-buffer-max-size",
 395           "Max. ring buffer size (bytes)",
 396           "Max. amount of data in the ring buffer (bytes, 0 = disabled)",
 397           0, G_MAXUINT64, DEFAULT_RING_BUFFER_MAX_SIZE,
 398           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 399
 400   /**
 401    * GstQueue2:avg-in-rate
 402    *
 403    * The average input data rate.
 404    */
 405   g_object_class_install_property (gobject_class, PROP_AVG_IN_RATE,
 406       g_param_spec_int64 ("avg-in-rate", "Input data rate (bytes/s)",
 407           "Average input data rate (bytes/s)",
 408           0, G_MAXINT64, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 409
 410   /* set several parent class virtual functions */
 411   gobject_class->finalize = gst_queue2_finalize;
 412
 413   gst_element_class_add_pad_template (gstelement_class,
 414       gst_static_pad_template_get (&srctemplate));
 415   gst_element_class_add_pad_template (gstelement_class,
 416       gst_static_pad_template_get (&sinktemplate));
 417
 418   gst_element_class_set_static_metadata (gstelement_class, "Queue 2",
 419       "Generic",
 420       "Simple data queue",
 421       "Erik Walthinsen <omega@cse.ogi.edu>, "
 422       "Wim Taymans <wim.taymans@gmail.com>");
 423
 424   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_queue2_change_state);
 425   gstelement_class->query = GST_DEBUG_FUNCPTR (gst_queue2_handle_query);
 426 }
 427
 428 static void
 429 gst_queue2_init (GstQueue2 * queue)
 430 {
 431   queue->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 432
 433   gst_pad_set_chain_function (queue->sinkpad,
 434       GST_DEBUG_FUNCPTR (gst_queue2_chain));
 435   gst_pad_set_chain_list_function (queue->sinkpad,
 436       GST_DEBUG_FUNCPTR (gst_queue2_chain_list));
 437   gst_pad_set_activatemode_function (queue->sinkpad,
 438       GST_DEBUG_FUNCPTR (gst_queue2_sink_activate_mode));
 439   gst_pad_set_event_function (queue->sinkpad,
 440       GST_DEBUG_FUNCPTR (gst_queue2_handle_sink_event));
 441   gst_pad_set_query_function (queue->sinkpad,
 442       GST_DEBUG_FUNCPTR (gst_queue2_handle_sink_query));
 443   GST_PAD_SET_PROXY_CAPS (queue->sinkpad);
 444   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 445
 446   queue->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
 447
 448   gst_pad_set_activatemode_function (queue->srcpad,
 449       GST_DEBUG_FUNCPTR (gst_queue2_src_activate_mode));
 450   gst_pad_set_getrange_function (queue->srcpad,
 451       GST_DEBUG_FUNCPTR (gst_queue2_get_range));
 452   gst_pad_set_event_function (queue->srcpad,
 453       GST_DEBUG_FUNCPTR (gst_queue2_handle_src_event));
 454   gst_pad_set_query_function (queue->srcpad,
 455       GST_DEBUG_FUNCPTR (gst_queue2_handle_src_query));
 456   GST_PAD_SET_PROXY_CAPS (queue->srcpad);
 457   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 458
 459   /* levels */
 460   GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
 461   queue->max_level.buffers = DEFAULT_MAX_SIZE_BUFFERS;
 462   queue->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;
 463   queue->max_level.time = DEFAULT_MAX_SIZE_TIME;
 464   queue->max_level.rate_time = DEFAULT_MAX_SIZE_TIME;
 465   queue->use_buffering = DEFAULT_USE_BUFFERING;
 466   queue->use_rate_estimate = DEFAULT_USE_RATE_ESTIMATE;
 467   queue->low_percent = DEFAULT_LOW_PERCENT;
 468   queue->high_percent = DEFAULT_HIGH_PERCENT;
 469
 470   gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
 471   gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
 472
 473   queue->sinktime = GST_CLOCK_TIME_NONE;
 474   queue->srctime = GST_CLOCK_TIME_NONE;
 475   queue->sink_tainted = TRUE;
 476   queue->src_tainted = TRUE;
 477
 478   queue->srcresult = GST_FLOW_FLUSHING;
 479   queue->sinkresult = GST_FLOW_FLUSHING;
 480   queue->is_eos = FALSE;
 481   queue->in_timer = g_timer_new ();
 482   queue->out_timer = g_timer_new ();
 483
 484   g_mutex_init (&queue->qlock);
 485   queue->waiting_add = FALSE;
 486   g_cond_init (&queue->item_add);
 487   queue->waiting_del = FALSE;
 488   g_cond_init (&queue->item_del);
 489   g_queue_init (&queue->queue);
 490
 491   g_cond_init (&queue->query_handled);
 492   queue->last_query = FALSE;
 493
 494   g_mutex_init (&queue->buffering_post_lock);
 495   queue->buffering_percent = 100;
 496
 497   /* tempfile related */
 498   queue->temp_template = NULL;
 499   queue->temp_location = NULL;
 500   queue->temp_remove = DEFAULT_TEMP_REMOVE;
 501
 502   queue->ring_buffer = NULL;
 503   queue->ring_buffer_max_size = DEFAULT_RING_BUFFER_MAX_SIZE;
 504
 505   GST_DEBUG_OBJECT (queue,
 506       "initialized queue's not_empty & not_full conditions");
 507 }
 508
 509 /* called only once, as opposed to dispose */
 510 static void
 511 gst_queue2_finalize (GObject * object)
 512 {
 513   GstQueue2 *queue = GST_QUEUE2 (object);
 514
 515   GST_DEBUG_OBJECT (queue, "finalizing queue");
 516
 517   while (!g_queue_is_empty (&queue->queue)) {
 518     GstQueue2Item *qitem = g_queue_pop_head (&queue->queue);
 519
 520     if (qitem->type != GST_QUEUE2_ITEM_TYPE_QUERY)
 521       gst_mini_object_unref (qitem->item);
 522     g_slice_free (GstQueue2Item, qitem);
 523   }
 524
 525   queue->last_query = FALSE;
 526   g_queue_clear (&queue->queue);
 527   g_mutex_clear (&queue->qlock);
 528   g_mutex_clear (&queue->buffering_post_lock);
 529   g_cond_clear (&queue->item_add);
 530   g_cond_clear (&queue->item_del);
 531   g_cond_clear (&queue->query_handled);
 532   g_timer_destroy (queue->in_timer);
 533   g_timer_destroy (queue->out_timer);
 534
 535   /* temp_file path cleanup  */
 536   g_free (queue->temp_template);
 537   g_free (queue->temp_location);
 538
 539   G_OBJECT_CLASS (parent_class)->finalize (object);
 540 }
 541
 542 static void
 543 debug_ranges (GstQueue2 * queue)
 544 {
 545   GstQueue2Range *walk;
 546
 547   for (walk = queue->ranges; walk; walk = walk->next) {
 548     GST_DEBUG_OBJECT (queue,
 549         "range [%" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT "] (rb [%"
 550         G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT "]), reading %" G_GUINT64_FORMAT
 551         " current range? %s", walk->offset, walk->writing_pos, walk->rb_offset,
 552         walk->rb_writing_pos, walk->reading_pos,
 553         walk == queue->current ? "**y**" : "  n  ");
 554   }
 555 }
 556
 557 /* clear all the downloaded ranges */
 558 static void
 559 clean_ranges (GstQueue2 * queue)
 560 {
 561   GST_DEBUG_OBJECT (queue, "clean queue ranges");
 562
 563   g_slice_free_chain (GstQueue2Range, queue->ranges, next);
 564   queue->ranges = NULL;
 565   queue->current = NULL;
 566 }
 567
 568 /* find a range that contains @offset or NULL when nothing does */
 569 static GstQueue2Range *
 570 find_range (GstQueue2 * queue, guint64 offset)
 571 {
 572   GstQueue2Range *range = NULL;
 573   GstQueue2Range *walk;
 574
 575   /* first do a quick check for the current range */
 576   for (walk = queue->ranges; walk; walk = walk->next) {
 577     if (offset >= walk->offset && offset <= walk->writing_pos) {
 578       /* we can reuse an existing range */
 579       range = walk;
 580       break;
 581     }
 582   }
 583   if (range) {
 584     GST_DEBUG_OBJECT (queue,
 585         "found range for %" G_GUINT64_FORMAT ": [%" G_GUINT64_FORMAT "-%"
 586         G_GUINT64_FORMAT "]", offset, range->offset, range->writing_pos);
 587   } else {
 588     GST_DEBUG_OBJECT (queue, "no range for %" G_GUINT64_FORMAT, offset);
 589   }
 590   return range;
 591 }
 592
 593 static void
 594 update_cur_level (GstQueue2 * queue, GstQueue2Range * range)
 595 {
 596   guint64 max_reading_pos, writing_pos;
 597
 598   writing_pos = range->writing_pos;
 599   max_reading_pos = range->max_reading_pos;
 600
 601   if (writing_pos > max_reading_pos)
 602     queue->cur_level.bytes = writing_pos - max_reading_pos;
 603   else
 604     queue->cur_level.bytes = 0;
 605 }
 606
 607 /* make a new range for @offset or reuse an existing range */
 608 static GstQueue2Range *
 609 add_range (GstQueue2 * queue, guint64 offset, gboolean update_existing)
 610 {
 611   GstQueue2Range *range, *prev, *next;
 612
 613   GST_DEBUG_OBJECT (queue, "find range for %" G_GUINT64_FORMAT, offset);
 614
 615   if ((range = find_range (queue, offset))) {
 616     GST_DEBUG_OBJECT (queue,
 617         "reusing range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, range->offset,
 618         range->writing_pos);
 619     if (update_existing && range->writing_pos != offset) {
 620       GST_DEBUG_OBJECT (queue, "updating range writing position to "
 621           "%" G_GUINT64_FORMAT, offset);
 622       range->writing_pos = offset;
 623     }
 624   } else {
 625     GST_DEBUG_OBJECT (queue,
 626         "new range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, offset, offset);
 627
 628     range = g_slice_new0 (GstQueue2Range);
 629     range->offset = offset;
 630     /* we want to write to the next location in the ring buffer */
 631     range->rb_offset = queue->current ? queue->current->rb_writing_pos : 0;
 632     range->writing_pos = offset;
 633     range->rb_writing_pos = range->rb_offset;
 634     range->reading_pos = offset;
 635     range->max_reading_pos = offset;
 636
 637     /* insert sorted */
 638     prev = NULL;
 639     next = queue->ranges;
 640     while (next) {
 641       if (next->offset > offset) {
 642         /* insert before next */
 643         GST_DEBUG_OBJECT (queue,
 644             "insert before range %p, offset %" G_GUINT64_FORMAT, next,
 645             next->offset);
 646         break;
 647       }
 648       /* try next */
 649       prev = next;
 650       next = next->next;
 651     }
 652     range->next = next;
 653     if (prev)
 654       prev->next = range;
 655     else
 656       queue->ranges = range;
 657   }
 658   debug_ranges (queue);
 659
 660   /* update the stats for this range */
 661   update_cur_level (queue, range);
 662
 663   return range;
 664 }
 665
 666
 667 /* clear and init the download ranges for offset 0 */
 668 static void
 669 init_ranges (GstQueue2 * queue)
 670 {
 671   GST_DEBUG_OBJECT (queue, "init queue ranges");
 672
 673   /* get rid of all the current ranges */
 674   clean_ranges (queue);
 675   /* make a range for offset 0 */
 676   queue->current = add_range (queue, 0, TRUE);
 677 }
 678
 679 /* calculate the diff between running time on the sink and src of the queue.
 680  * This is the total amount of time in the queue. */
 681 static void
 682 update_time_level (GstQueue2 * queue)
 683 {
 684   if (queue->sink_tainted) {
 685     queue->sinktime =
 686         gst_segment_to_running_time (&queue->sink_segment, GST_FORMAT_TIME,
 687         queue->sink_segment.position);
 688     queue->sink_tainted = FALSE;
 689   }
 690
 691   if (queue->src_tainted) {
 692     queue->srctime =
 693         gst_segment_to_running_time (&queue->src_segment, GST_FORMAT_TIME,
 694         queue->src_segment.position);
 695     queue->src_tainted = FALSE;
 696   }
 697
 698   GST_DEBUG_OBJECT (queue, "sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT,
 699       GST_TIME_ARGS (queue->sinktime), GST_TIME_ARGS (queue->srctime));
 700
 701   if (queue->sinktime != GST_CLOCK_TIME_NONE
 702       && queue->srctime != GST_CLOCK_TIME_NONE
 703       && queue->sinktime >= queue->srctime)
 704     queue->cur_level.time = queue->sinktime - queue->srctime;
 705   else
 706     queue->cur_level.time = 0;
 707 }
 708
 709 /* take a SEGMENT event and apply the values to segment, updating the time
 710  * level of queue. */
 711 static void
 712 apply_segment (GstQueue2 * queue, GstEvent * event, GstSegment * segment,
 713     gboolean is_sink)
 714 {
 715   gst_event_copy_segment (event, segment);
 716
 717   if (segment->format == GST_FORMAT_BYTES) {
 718     if (!QUEUE_IS_USING_QUEUE (queue) && is_sink) {
 719       /* start is where we'll be getting from and as such writing next */
 720       queue->current = add_range (queue, segment->start, TRUE);
 721     }
 722   }
 723
 724   /* now configure the values, we use these to track timestamps on the
 725    * sinkpad. */
 726   if (segment->format != GST_FORMAT_TIME) {
 727     /* non-time format, pretend the current time segment is closed with a
 728      * 0 start and unknown stop time. */
 729     segment->format = GST_FORMAT_TIME;
 730     segment->start = 0;
 731     segment->stop = -1;
 732     segment->time = 0;
 733   }
 734
 735   GST_DEBUG_OBJECT (queue, "configured SEGMENT %" GST_SEGMENT_FORMAT, segment);
 736
 737   if (is_sink)
 738     queue->sink_tainted = TRUE;
 739   else
 740     queue->src_tainted = TRUE;
 741
 742   /* segment can update the time level of the queue */
 743   update_time_level (queue);
 744 }
 745
 746 static void
 747 apply_gap (GstQueue2 * queue, GstEvent * event,
 748     GstSegment * segment, gboolean is_sink)
 749 {
 750   GstClockTime timestamp;
 751   GstClockTime duration;
 752
 753   gst_event_parse_gap (event, &timestamp, &duration);
 754
 755   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
 756
 757     if (GST_CLOCK_TIME_IS_VALID (duration)) {
 758       timestamp += duration;
 759     }
 760
 761     segment->position = timestamp;
 762
 763     if (is_sink)
 764       queue->sink_tainted = TRUE;
 765     else
 766       queue->src_tainted = TRUE;
 767
 768     /* calc diff with other end */
 769     update_time_level (queue);
 770   }
 771 }
 772
 773 /* take a buffer and update segment, updating the time level of the queue. */
 774 static void
 775 apply_buffer (GstQueue2 * queue, GstBuffer * buffer, GstSegment * segment,
 776     guint64 size, gboolean is_sink)
 777 {
 778   GstClockTime duration, timestamp;
 779
 780   timestamp = GST_BUFFER_DTS_OR_PTS (buffer);
 781   duration = GST_BUFFER_DURATION (buffer);
 782
 783   /* If we have no duration, pick one from the bitrate if we can */
 784   if (duration == GST_CLOCK_TIME_NONE && queue->use_tags_bitrate) {
 785     guint bitrate =
 786         is_sink ? queue->sink_tags_bitrate : queue->src_tags_bitrate;
 787     if (bitrate)
 788       duration = gst_util_uint64_scale (size, 8 * GST_SECOND, bitrate);
 789   }
 790
 791   /* if no timestamp is set, assume it's continuous with the previous
 792    * time */
 793   if (timestamp == GST_CLOCK_TIME_NONE)
 794     timestamp = segment->position;
 795
 796   /* add duration */
 797   if (duration != GST_CLOCK_TIME_NONE)
 798     timestamp += duration;
 799
 800   GST_DEBUG_OBJECT (queue, "position updated to %" GST_TIME_FORMAT,
 801       GST_TIME_ARGS (timestamp));
 802
 803   segment->position = timestamp;
 804
 805   if (is_sink)
 806     queue->sink_tainted = TRUE;
 807   else
 808     queue->src_tainted = TRUE;
 809
 810   /* calc diff with other end */
 811   update_time_level (queue);
 812 }
 813
 814 struct BufListData
 815 {
 816   GstClockTime timestamp;
 817   guint bitrate;
 818 };
 819
 820 static gboolean
 821 buffer_list_apply_time (GstBuffer ** buf, guint idx, gpointer data)
 822 {
 823   struct BufListData *bld = data;
 824   GstClockTime *timestamp = &bld->timestamp;
 825   GstClockTime btime;
 826
 827   GST_TRACE ("buffer %u has pts %" GST_TIME_FORMAT " dts %" GST_TIME_FORMAT
 828       " duration %" GST_TIME_FORMAT, idx,
 829       GST_TIME_ARGS (GST_BUFFER_PTS (*buf)),
 830       GST_TIME_ARGS (GST_BUFFER_DTS (*buf)),
 831       GST_TIME_ARGS (GST_BUFFER_DURATION (*buf)));
 832
 833   btime = GST_BUFFER_DTS_OR_PTS (*buf);
 834   if (GST_CLOCK_TIME_IS_VALID (btime))
 835     *timestamp = btime;
 836
 837   if (GST_BUFFER_DURATION_IS_VALID (*buf))
 838     *timestamp += GST_BUFFER_DURATION (*buf);
 839   else if (bld->bitrate != 0) {
 840     guint64 size = gst_buffer_get_size (*buf);
 841
 842     /* If we have no duration, pick one from the bitrate if we can */
 843     *timestamp += gst_util_uint64_scale (bld->bitrate, 8 * GST_SECOND, size);
 844   }
 845
 846
 847   GST_TRACE ("ts now %" GST_TIME_FORMAT, GST_TIME_ARGS (*timestamp));
 848   return TRUE;
 849 }
 850
 851 /* take a buffer list and update segment, updating the time level of the queue */
 852 static void
 853 apply_buffer_list (GstQueue2 * queue, GstBufferList * buffer_list,
 854     GstSegment * segment, gboolean is_sink)
 855 {
 856   struct BufListData bld;
 857
 858   /* if no timestamp is set, assume it's continuous with the previous time */
 859   bld.timestamp = segment->position;
 860
 861   if (queue->use_tags_bitrate) {
 862     if (is_sink)
 863       bld.bitrate = queue->sink_tags_bitrate;
 864     else
 865       bld.bitrate = queue->src_tags_bitrate;
 866   } else
 867     bld.bitrate = 0;
 868
 869   gst_buffer_list_foreach (buffer_list, buffer_list_apply_time, &bld);
 870
 871   GST_DEBUG_OBJECT (queue, "last_stop updated to %" GST_TIME_FORMAT,
 872       GST_TIME_ARGS (bld.timestamp));
 873
 874   segment->position = bld.timestamp;
 875
 876   if (is_sink)
 877     queue->sink_tainted = TRUE;
 878   else
 879     queue->src_tainted = TRUE;
 880
 881   /* calc diff with other end */
 882   update_time_level (queue);
 883 }
 884
 885 static inline gint
 886 get_percent (guint64 cur_level, guint64 max_level, guint64 alt_max)
 887 {
 888   guint64 p;
 889
 890   if (max_level == 0)
 891     return 0;
 892
 893   if (alt_max > 0)
 894     p = gst_util_uint64_scale (cur_level, 100, MIN (max_level, alt_max));
 895   else
 896     p = gst_util_uint64_scale (cur_level, 100, max_level);
 897
 898   return MIN (p, 100);
 899 }
 900
 901 static gboolean
 902 get_buffering_percent (GstQueue2 * queue, gboolean * is_buffering,
 903     gint * percent)
 904 {
 905   gint perc, perc2;
 906
 907   if (queue->high_percent <= 0) {
 908     if (percent)
 909       *percent = 100;
 910     if (is_buffering)
 911       *is_buffering = FALSE;
 912     return FALSE;
 913   }
 914 #define GET_PERCENT(format,alt_max) \
 915     get_percent(queue->cur_level.format,queue->max_level.format,(alt_max))
 916
 917   if (queue->is_eos) {
 918     /* on EOS we are always 100% full, we set the var here so that it we can
 919      * reuse the logic below to stop buffering */
 920     perc = 100;
 921     GST_LOG_OBJECT (queue, "we are EOS");
 922   } else {
 923     GST_LOG_OBJECT (queue,
 924         "Cur level bytes/time/buffers %u/%" GST_TIME_FORMAT "/%u",
 925         queue->cur_level.bytes, GST_TIME_ARGS (queue->cur_level.time),
 926         queue->cur_level.buffers);
 927
 928     /* figure out the percent we are filled, we take the max of all formats. */
 929     if (!QUEUE_IS_USING_RING_BUFFER (queue)) {
 930       perc = GET_PERCENT (bytes, 0);
 931     } else {
 932       guint64 rb_size = queue->ring_buffer_max_size;
 933       perc = GET_PERCENT (bytes, rb_size);
 934     }
 935
 936     perc2 = GET_PERCENT (time, 0);
 937     perc = MAX (perc, perc2);
 938
 939     perc2 = GET_PERCENT (buffers, 0);
 940     perc = MAX (perc, perc2);
 941
 942     /* also apply the rate estimate when we need to */
 943     if (queue->use_rate_estimate) {
 944       perc2 = GET_PERCENT (rate_time, 0);
 945       perc = MAX (perc, perc2);
 946     }
 947
 948     /* Don't get to 0% unless we're really empty */
 949     if (queue->cur_level.bytes > 0)
 950       perc = MAX (1, perc);
 951   }
 952 #undef GET_PERCENT
 953
 954   if (is_buffering)
 955     *is_buffering = queue->is_buffering;
 956
 957   /* scale to high percent so that it becomes the 100% mark */
 958   perc = perc * 100 / queue->high_percent;
 959   /* clip */
 960   if (perc > 100)
 961     perc = 100;
 962
 963   if (percent)
 964     *percent = perc;
 965
 966   GST_DEBUG_OBJECT (queue, "buffering %d, percent %d", queue->is_buffering,
 967       perc);
 968
 969   return TRUE;
 970 }
 971
 972 static void
 973 get_buffering_stats (GstQueue2 * queue, gint percent, GstBufferingMode * mode,
 974     gint * avg_in, gint * avg_out, gint64 * buffering_left)
 975 {
 976   if (mode) {
 977     if (!QUEUE_IS_USING_QUEUE (queue)) {
 978       if (QUEUE_IS_USING_RING_BUFFER (queue))
 979         *mode = GST_BUFFERING_TIMESHIFT;
 980       else
 981         *mode = GST_BUFFERING_DOWNLOAD;
 982     } else {
 983       *mode = GST_BUFFERING_STREAM;
 984     }
 985   }
 986
 987   if (avg_in)
 988     *avg_in = queue->byte_in_rate;
 989   if (avg_out)
 990     *avg_out = queue->byte_out_rate;
 991
 992   if (buffering_left) {
 993     *buffering_left = (percent == 100 ? 0 : -1);
 994
 995     if (queue->use_rate_estimate) {
 996       guint64 max, cur;
 997
 998       max = queue->max_level.rate_time;
 999       cur = queue->cur_level.rate_time;
1000
1001       if (percent != 100 && max > cur)
1002         *buffering_left = (max - cur) / 1000000;
1003     }
1004   }
1005 }
1006
1007 static void
1008 gst_queue2_post_buffering (GstQueue2 * queue)
1009 {
1010   GstMessage *msg = NULL;
1011
1012   g_mutex_lock (&queue->buffering_post_lock);
1013   GST_QUEUE2_MUTEX_LOCK (queue);
1014   if (queue->percent_changed) {
1015     gint percent = queue->buffering_percent;
1016
1017     queue->percent_changed = FALSE;
1018
1019     GST_DEBUG_OBJECT (queue, "Going to post buffering: %d%%", percent);
1020     msg = gst_message_new_buffering (GST_OBJECT_CAST (queue), percent);
1021
1022     gst_message_set_buffering_stats (msg, queue->mode, queue->avg_in,
1023         queue->avg_out, queue->buffering_left);
1024   }
1025   GST_QUEUE2_MUTEX_UNLOCK (queue);
1026
1027   if (msg != NULL)
1028     gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
1029
1030   g_mutex_unlock (&queue->buffering_post_lock);
1031 }
1032
1033 static void
1034 update_buffering (GstQueue2 * queue)
1035 {
1036   gint percent;
1037
1038   /* Ensure the variables used to calculate buffering state are up-to-date. */
1039   if (queue->current)
1040     update_cur_level (queue, queue->current);
1041   update_in_rates (queue);
1042
1043   if (!get_buffering_percent (queue, NULL, &percent))
1044     return;
1045
1046   if (queue->is_buffering) {
1047     /* if we were buffering see if we reached the high watermark */
1048     if (percent >= 100)
1049       queue->is_buffering = FALSE;
1050
1051     SET_PERCENT (queue, percent);
1052   } else {
1053     /* we were not buffering, check if we need to start buffering if we drop
1054      * below the low threshold */
1055     if (percent < queue->low_percent) {
1056       queue->is_buffering = TRUE;
1057       SET_PERCENT (queue, percent);
1058     }
1059   }
1060 }
1061
1062 static void
1063 reset_rate_timer (GstQueue2 * queue)
1064 {
1065   queue->bytes_in = 0;
1066   queue->bytes_out = 0;
1067   queue->byte_in_rate = 0.0;
1068   queue->byte_in_period = 0;
1069   queue->byte_out_rate = 0.0;
1070   queue->last_update_in_rates_elapsed = 0.0;
1071   queue->last_in_elapsed = 0.0;
1072   queue->last_out_elapsed = 0.0;
1073   queue->in_timer_started = FALSE;
1074   queue->out_timer_started = FALSE;
1075 }
1076
1077 /* the interval in seconds to recalculate the rate */
1078 #define RATE_INTERVAL    0.2
1079 /* Tuning for rate estimation. We use a large window for the input rate because
1080  * it should be stable when connected to a network. The output rate is less
1081  * stable (the elements preroll, queues behind a demuxer fill, ...) and should
1082  * therefore adapt more quickly.
1083  * However, initial input rate may be subject to a burst, and should therefore
1084  * initially also adapt more quickly to changes, and only later on give higher
1085  * weight to previous values. */
1086 #define AVG_IN(avg,val,w1,w2)  ((avg) * (w1) + (val) * (w2)) / ((w1) + (w2))
1087 #define AVG_OUT(avg,val) ((avg) * 3.0 + (val)) / 4.0
1088
1089 static void
1090 update_in_rates (GstQueue2 * queue)
1091 {
1092   gdouble elapsed, period;
1093   gdouble byte_in_rate;
1094
1095   if (!queue->in_timer_started) {
1096     queue->in_timer_started = TRUE;
1097     g_timer_start (queue->in_timer);
1098     return;
1099   }
1100
1101   queue->last_update_in_rates_elapsed = elapsed =
1102       g_timer_elapsed (queue->in_timer, NULL);
1103
1104   /* recalc after each interval. */
1105   if (queue->last_in_elapsed + RATE_INTERVAL < elapsed) {
1106     period = elapsed - queue->last_in_elapsed;
1107
1108     GST_DEBUG_OBJECT (queue,
1109         "rates: period %f, in %" G_GUINT64_FORMAT ", global period %f",
1110         period, queue->bytes_in, queue->byte_in_period);
1111
1112     byte_in_rate = queue->bytes_in / period;
1113
1114     if (queue->byte_in_rate == 0.0)
1115       queue->byte_in_rate = byte_in_rate;
1116     else
1117       queue->byte_in_rate = AVG_IN (queue->byte_in_rate, byte_in_rate,
1118           (double) queue->byte_in_period, period);
1119
1120     /* another data point, cap at 16 for long time running average */
1121     if (queue->byte_in_period < 16 * RATE_INTERVAL)
1122       queue->byte_in_period += period;
1123
1124     /* reset the values to calculate rate over the next interval */
1125     queue->last_in_elapsed = elapsed;
1126     queue->bytes_in = 0;
1127   }
1128
1129   if (queue->byte_in_rate > 0.0) {
1130     queue->cur_level.rate_time =
1131         queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
1132   }
1133   GST_DEBUG_OBJECT (queue, "rates: in %f, time %" GST_TIME_FORMAT,
1134       queue->byte_in_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
1135 }
1136
1137 static void
1138 update_out_rates (GstQueue2 * queue)
1139 {
1140   gdouble elapsed, period;
1141   gdouble byte_out_rate;
1142
1143   if (!queue->out_timer_started) {
1144     queue->out_timer_started = TRUE;
1145     g_timer_start (queue->out_timer);
1146     return;
1147   }
1148
1149   elapsed = g_timer_elapsed (queue->out_timer, NULL);
1150
1151   /* recalc after each interval. */
1152   if (queue->last_out_elapsed + RATE_INTERVAL < elapsed) {
1153     period = elapsed - queue->last_out_elapsed;
1154
1155     GST_DEBUG_OBJECT (queue,
1156         "rates: period %f, out %" G_GUINT64_FORMAT, period, queue->bytes_out);
1157
1158     byte_out_rate = queue->bytes_out / period;
1159
1160     if (queue->byte_out_rate == 0.0)
1161       queue->byte_out_rate = byte_out_rate;
1162     else
1163       queue->byte_out_rate = AVG_OUT (queue->byte_out_rate, byte_out_rate);
1164
1165     /* reset the values to calculate rate over the next interval */
1166     queue->last_out_elapsed = elapsed;
1167     queue->bytes_out = 0;
1168   }
1169   if (queue->byte_in_rate > 0.0) {
1170     queue->cur_level.rate_time =
1171         queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
1172   }
1173   GST_DEBUG_OBJECT (queue, "rates: out %f, time %" GST_TIME_FORMAT,
1174       queue->byte_out_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
1175 }
1176
1177 static void
1178 update_cur_pos (GstQueue2 * queue, GstQueue2Range * range, guint64 pos)
1179 {
1180   guint64 reading_pos, max_reading_pos;
1181
1182   reading_pos = pos;
1183   max_reading_pos = range->max_reading_pos;
1184
1185   max_reading_pos = MAX (max_reading_pos, reading_pos);
1186
1187   GST_DEBUG_OBJECT (queue,
1188       "updating max_reading_pos from %" G_GUINT64_FORMAT " to %"
1189       G_GUINT64_FORMAT, range->max_reading_pos, max_reading_pos);
1190   range->max_reading_pos = max_reading_pos;
1191
1192   update_cur_level (queue, range);
1193 }
1194
1195 static gboolean
1196 perform_seek_to_offset (GstQueue2 * queue, guint64 offset)
1197 {
1198   GstEvent *event;
1199   gboolean res;
1200
1201   /* until we receive the FLUSH_STOP from this seek, we skip data */
1202   queue->seeking = TRUE;
1203   GST_QUEUE2_MUTEX_UNLOCK (queue);
1204
1205   debug_ranges (queue);
1206
1207   GST_DEBUG_OBJECT (queue, "Seeking to %" G_GUINT64_FORMAT, offset);
1208
1209   event =
1210       gst_event_new_seek (1.0, GST_FORMAT_BYTES,
1211       GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
1212       GST_SEEK_TYPE_NONE, -1);
1213
1214   res = gst_pad_push_event (queue->sinkpad, event);
1215   GST_QUEUE2_MUTEX_LOCK (queue);
1216
1217   if (res) {
1218     /* Between us sending the seek event and re-acquiring the lock, the source
1219      * thread might already have pushed data and moved along the range's
1220      * writing_pos beyond the seek offset. In that case we don't want to set
1221      * the writing position back to the requested seek position, as it would
1222      * cause data to be written to the wrong offset in the file or ring buffer.
1223      * We still do the add_range call to switch the current range to the
1224      * requested range, or create one if one doesn't exist yet. */
1225     queue->current = add_range (queue, offset, FALSE);
1226   }
1227
1228   return res;
1229 }
1230
1231 /* get the threshold for when we decide to seek rather than wait */
1232 static guint64
1233 get_seek_threshold (GstQueue2 * queue)
1234 {
1235   guint64 threshold;
1236
1237   /* FIXME, find a good threshold based on the incoming rate. */
1238   threshold = 1024 * 512;
1239
1240   if (QUEUE_IS_USING_RING_BUFFER (queue)) {
1241     threshold = MIN (threshold,
1242         QUEUE_MAX_BYTES (queue) - queue->cur_level.bytes);
1243   }
1244   return threshold;
1245 }
1246
1247 /* see if there is enough data in the file to read a full buffer */
1248 static gboolean
1249 gst_queue2_have_data (GstQueue2 * queue, guint64 offset, guint length)
1250 {
1251   GstQueue2Range *range;
1252
1253   GST_DEBUG_OBJECT (queue, "looking for offset %" G_GUINT64_FORMAT ", len %u",
1254       offset, length);
1255
1256   if ((range = find_range (queue, offset))) {
1257     if (queue->current != range) {
1258       GST_DEBUG_OBJECT (queue, "switching ranges, do seek to range position");
1259       perform_seek_to_offset (queue, range->writing_pos);
1260     }
1261
1262     GST_INFO_OBJECT (queue, "cur_level.bytes %u (max %" G_GUINT64_FORMAT ")",
1263         queue->cur_level.bytes, QUEUE_MAX_BYTES (queue));
1264
1265     /* we have a range for offset */
1266     GST_DEBUG_OBJECT (queue,
1267         "we have a range %p, offset %" G_GUINT64_FORMAT ", writing_pos %"
1268         G_GUINT64_FORMAT, range, range->offset, range->writing_pos);
1269
1270     if (!QUEUE_IS_USING_RING_BUFFER (queue) && queue->is_eos)
1271       return TRUE;
1272
1273     if (offset + length <= range->writing_pos)
1274       return TRUE;
1275     else
1276       GST_DEBUG_OBJECT (queue,
1277           "Need more data (%" G_GUINT64_FORMAT " bytes more)",
1278           (offset + length) - range->writing_pos);
1279
1280   } else {
1281     GST_INFO_OBJECT (queue, "not found in any range off %" G_GUINT64_FORMAT
1282         " len %u", offset, length);
1283     /* we don't have the range, see how far away we are */
1284     if (!queue->is_eos && queue->current) {
1285       guint64 threshold = get_seek_threshold (queue);
1286
1287       if (offset >= queue->current->offset && offset <=
1288           queue->current->writing_pos + threshold) {
1289         GST_INFO_OBJECT (queue,
1290             "requested data is within range, wait for data");
1291         return FALSE;
1292       }
1293     }
1294
1295     /* too far away, do a seek */
1296     perform_seek_to_offset (queue, offset);
1297   }
1298
1299   return FALSE;
1300 }
1301
1302 #ifdef HAVE_FSEEKO
1303 #define FSEEK_FILE(file,offset)  (fseeko (file, (off_t) offset, SEEK_SET) != 0)
1304 #elif defined (G_OS_UNIX) || defined (G_OS_WIN32)
1305 #define FSEEK_FILE(file,offset)  (lseek (fileno (file), (off_t) offset, SEEK_SET) == (off_t) -1)
1306 #else
1307 #define FSEEK_FILE(file,offset)  (fseek (file, offset, SEEK_SET) != 0)
1308 #endif
1309
1310 static GstFlowReturn
1311 gst_queue2_read_data_at_offset (GstQueue2 * queue, guint64 offset, guint length,
1312     guint8 * dst, gint64 * read_return)
1313 {
1314   guint8 *ring_buffer;
1315   size_t res;
1316
1317   ring_buffer = queue->ring_buffer;
1318
1319   if (QUEUE_IS_USING_TEMP_FILE (queue) && FSEEK_FILE (queue->temp_file, offset))
1320     goto seek_failed;
1321
1322   /* this should not block */
1323   GST_LOG_OBJECT (queue, "Reading %d bytes from offset %" G_GUINT64_FORMAT,
1324       length, offset);
1325   if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1326     res = fread (dst, 1, length, queue->temp_file);
1327   } else {
1328     memcpy (dst, ring_buffer + offset, length);
1329     res = length;
1330   }
1331
1332   GST_LOG_OBJECT (queue, "read %" G_GSIZE_FORMAT " bytes", res);
1333
1334   if (G_UNLIKELY (res < length)) {
1335     if (!QUEUE_IS_USING_TEMP_FILE (queue))
1336       goto could_not_read;
1337     /* check for errors or EOF */
1338     if (ferror (queue->temp_file))
1339       goto could_not_read;
1340     if (feof (queue->temp_file) && length > 0)
1341       goto eos;
1342   }
1343
1344   *read_return = res;
1345
1346   return GST_FLOW_OK;
1347
1348 seek_failed:
1349   {
1350     GST_ELEMENT_ERROR (queue, RESOURCE, SEEK, (NULL), GST_ERROR_SYSTEM);
1351     return GST_FLOW_ERROR;
1352   }
1353 could_not_read:
1354   {
1355     GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
1356     return GST_FLOW_ERROR;
1357   }
1358 eos:
1359   {
1360     GST_DEBUG ("non-regular file hits EOS");
1361     return GST_FLOW_EOS;
1362   }
1363 }
1364
1365 static GstFlowReturn
1366 gst_queue2_create_read (GstQueue2 * queue, guint64 offset, guint length,
1367     GstBuffer ** buffer)
1368 {
1369   GstBuffer *buf;
1370   GstMapInfo info;
1371   guint8 *data;
1372   guint64 file_offset;
1373   guint block_length, remaining, read_length;
1374   guint64 rb_size;
1375   guint64 max_size;
1376   guint64 rpos;
1377   GstFlowReturn ret = GST_FLOW_OK;
1378
1379   /* allocate the output buffer of the requested size */
1380   if (*buffer == NULL)
1381     buf = gst_buffer_new_allocate (NULL, length, NULL);
1382   else
1383     buf = *buffer;
1384
1385   if (!gst_buffer_map (buf, &info, GST_MAP_WRITE))
1386     goto buffer_write_fail;
1387   data = info.data;
1388
1389   GST_DEBUG_OBJECT (queue, "Reading %u bytes from %" G_GUINT64_FORMAT, length,
1390       offset);
1391
1392   rpos = offset;
1393   rb_size = queue->ring_buffer_max_size;
1394   max_size = QUEUE_MAX_BYTES (queue);
1395
1396   remaining = length;
1397   while (remaining > 0) {
1398     /* configure how much/whether to read */
1399     if (!gst_queue2_have_data (queue, rpos, remaining)) {
1400       read_length = 0;
1401
1402       if (QUEUE_IS_USING_RING_BUFFER (queue)) {
1403         guint64 level;
1404
1405         /* calculate how far away the offset is */
1406         if (queue->current->writing_pos > rpos)
1407           level = queue->current->writing_pos - rpos;
1408         else
1409           level = 0;
1410
1411         GST_DEBUG_OBJECT (queue,
1412             "reading %" G_GUINT64_FORMAT ", writing %" G_GUINT64_FORMAT
1413             ", level %" G_GUINT64_FORMAT ", max %" G_GUINT64_FORMAT,
1414             rpos, queue->current->writing_pos, level, max_size);
1415
1416         if (level >= max_size) {
1417           /* we don't have the data but if we have a ring buffer that is full, we
1418            * need to read */
1419           GST_DEBUG_OBJECT (queue,
1420               "ring buffer full, reading QUEUE_MAX_BYTES %"
1421               G_GUINT64_FORMAT " bytes", max_size);
1422           read_length = max_size;
1423         } else if (queue->is_eos) {
1424           /* won't get any more data so read any data we have */
1425           if (level) {
1426             GST_DEBUG_OBJECT (queue,
1427                 "EOS hit but read %" G_GUINT64_FORMAT " bytes that we have",
1428                 level);
1429             read_length = level;
1430             remaining = level;
1431             length = level;
1432           } else
1433             goto hit_eos;
1434         }
1435       }
1436
1437       if (read_length == 0) {
1438         if (QUEUE_IS_USING_RING_BUFFER (queue)) {
1439           GST_DEBUG_OBJECT (queue,
1440               "update current position [%" G_GUINT64_FORMAT "-%"
1441               G_GUINT64_FORMAT "]", rpos, queue->current->max_reading_pos);
1442           update_cur_pos (queue, queue->current, rpos);
1443           GST_QUEUE2_SIGNAL_DEL (queue);
1444         }
1445
1446         if (queue->use_buffering)
1447           update_buffering (queue);
1448
1449         GST_DEBUG_OBJECT (queue, "waiting for add");
1450         GST_QUEUE2_WAIT_ADD_CHECK (queue, queue->srcresult, out_flushing);
1451         continue;
1452       }
1453     } else {
1454       /* we have the requested data so read it */
1455       read_length = remaining;
1456     }
1457
1458     /* set range reading_pos to actual reading position for this read */
1459     queue->current->reading_pos = rpos;
1460
1461     /* configure how much and from where to read */
1462     if (QUEUE_IS_USING_RING_BUFFER (queue)) {
1463       file_offset =
1464           (queue->current->rb_offset + (rpos -
1465               queue->current->offset)) % rb_size;
1466       if (file_offset + read_length > rb_size) {
1467         block_length = rb_size - file_offset;
1468       } else {
1469         block_length = read_length;
1470       }
1471     } else {
1472       file_offset = rpos;
1473       block_length = read_length;
1474     }
1475
1476     /* while we still have data to read, we loop */
1477     while (read_length > 0) {
1478       gint64 read_return;
1479
1480       ret =
1481           gst_queue2_read_data_at_offset (queue, file_offset, block_length,
1482           data, &read_return);
1483       if (ret != GST_FLOW_OK)
1484         goto read_error;
1485
1486       file_offset += read_return;
1487       if (QUEUE_IS_USING_RING_BUFFER (queue))
1488         file_offset %= rb_size;
1489
1490       data += read_return;
1491       read_length -= read_return;
1492       block_length = read_length;
1493       remaining -= read_return;
1494
1495       rpos = (queue->current->reading_pos += read_return);
1496       update_cur_pos (queue, queue->current, queue->current->reading_pos);
1497     }
1498     GST_QUEUE2_SIGNAL_DEL (queue);
1499     GST_DEBUG_OBJECT (queue, "%u bytes left to read", remaining);
1500   }
1501
1502   gst_buffer_unmap (buf, &info);
1503   gst_buffer_resize (buf, 0, length);
1504
1505   GST_BUFFER_OFFSET (buf) = offset;
1506   GST_BUFFER_OFFSET_END (buf) = offset + length;
1507
1508   *buffer = buf;
1509
1510   return ret;
1511
1512   /* ERRORS */
1513 hit_eos:
1514   {
1515     GST_DEBUG_OBJECT (queue, "EOS hit and we don't have any requested data");
1516     gst_buffer_unmap (buf, &info);
1517     if (*buffer == NULL)
1518       gst_buffer_unref (buf);
1519     return GST_FLOW_EOS;
1520   }
1521 out_flushing:
1522   {
1523     GST_DEBUG_OBJECT (queue, "we are flushing");
1524     gst_buffer_unmap (buf, &info);
1525     if (*buffer == NULL)
1526       gst_buffer_unref (buf);
1527     return GST_FLOW_FLUSHING;
1528   }
1529 read_error:
1530   {
1531     GST_DEBUG_OBJECT (queue, "we have a read error");
1532     gst_buffer_unmap (buf, &info);
1533     if (*buffer == NULL)
1534       gst_buffer_unref (buf);
1535     return ret;
1536   }
1537 buffer_write_fail:
1538   {
1539     GST_ELEMENT_ERROR (queue, RESOURCE, WRITE, (NULL),
1540         ("Can't write to buffer"));
1541     if (*buffer == NULL)
1542       gst_buffer_unref (buf);
1543     return GST_FLOW_ERROR;
1544   }
1545 }
1546
1547 /* should be called with QUEUE_LOCK */
1548 static GstMiniObject *
1549 gst_queue2_read_item_from_file (GstQueue2 * queue)
1550 {
1551   GstMiniObject *item;
1552
1553   if (queue->stream_start_event != NULL) {
1554     item = GST_MINI_OBJECT_CAST (queue->stream_start_event);
1555     queue->stream_start_event = NULL;
1556   } else if (queue->starting_segment != NULL) {
1557     item = GST_MINI_OBJECT_CAST (queue->starting_segment);
1558     queue->starting_segment = NULL;
1559   } else {
1560     GstFlowReturn ret;
1561     GstBuffer *buffer = NULL;
1562     guint64 reading_pos;
1563
1564     reading_pos = queue->current->reading_pos;
1565
1566     ret =
1567         gst_queue2_create_read (queue, reading_pos, DEFAULT_BUFFER_SIZE,
1568         &buffer);
1569
1570     switch (ret) {
1571       case GST_FLOW_OK:
1572         item = GST_MINI_OBJECT_CAST (buffer);
1573         break;
1574       case GST_FLOW_EOS:
1575         item = GST_MINI_OBJECT_CAST (gst_event_new_eos ());
1576         break;
1577       default:
1578         item = NULL;
1579         break;
1580     }
1581   }
1582   return item;
1583 }
1584
1585 /* must be called with MUTEX_LOCK. Will briefly release the lock when notifying
1586  * the temp filename. */
1587 static gboolean
1588 gst_queue2_open_temp_location_file (GstQueue2 * queue)
1589 {
1590   gint fd = -1;
1591   gchar *name = NULL;
1592
1593   if (queue->temp_file)
1594     goto already_opened;
1595
1596   GST_DEBUG_OBJECT (queue, "opening temp file %s", queue->temp_template);
1597
1598   /* If temp_template was set, allocate a filename and open that file */
1599
1600   /* nothing to do */
1601   if (queue->temp_template == NULL)
1602     goto no_directory;
1603
1604   /* make copy of the template, we don't want to change this */
1605   name = g_strdup (queue->temp_template);
1606
1607 #ifdef __BIONIC__
1608   fd = g_mkstemp_full (name, O_RDWR | O_LARGEFILE, S_IRUSR | S_IWUSR);
1609 #else
1610   fd = g_mkstemp (name);
1611 #endif
1612
1613   if (fd == -1)
1614     goto mkstemp_failed;
1615
1616   /* open the file for update/writing */
1617   queue->temp_file = fdopen (fd, "wb+");
1618   /* error creating file */
1619   if (queue->temp_file == NULL)
1620     goto open_failed;
1621
1622   g_free (queue->temp_location);
1623   queue->temp_location = name;
1624
1625   GST_QUEUE2_MUTEX_UNLOCK (queue);
1626
1627   /* we can't emit the notify with the lock */
1628   g_object_notify (G_OBJECT (queue), "temp-location");
1629
1630   GST_QUEUE2_MUTEX_LOCK (queue);
1631
1632   GST_DEBUG_OBJECT (queue, "opened temp file %s", queue->temp_template);
1633
1634   return TRUE;
1635
1636   /* ERRORS */
1637 already_opened:
1638   {
1639     GST_DEBUG_OBJECT (queue, "temp file was already open");
1640     return TRUE;
1641   }
1642 no_directory:
1643   {
1644     GST_ELEMENT_ERROR (queue, RESOURCE, NOT_FOUND,
1645         (_("No Temp directory specified.")), (NULL));
1646     return FALSE;
1647   }
1648 mkstemp_failed:
1649   {
1650     GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1651         (_("Could not create temp file \"%s\"."), queue->temp_template),
1652         GST_ERROR_SYSTEM);
1653     g_free (name);
1654     return FALSE;
1655   }
1656 open_failed:
1657   {
1658     GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1659         (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
1660     g_free (name);
1661     if (fd != -1)
1662       close (fd);
1663     return FALSE;
1664   }
1665 }
1666
1667 static void
1668 gst_queue2_close_temp_location_file (GstQueue2 * queue)
1669 {
1670   /* nothing to do */
1671   if (queue->temp_file == NULL)
1672     return;
1673
1674   GST_DEBUG_OBJECT (queue, "closing temp file");
1675
1676   fflush (queue->temp_file);
1677   fclose (queue->temp_file);
1678
1679   if (queue->temp_remove) {
1680     if (remove (queue->temp_location) < 0) {
1681       GST_WARNING_OBJECT (queue, "Failed to remove temporary file %s: %s",
1682           queue->temp_location, g_strerror (errno));
1683     }
1684   }
1685
1686   queue->temp_file = NULL;
1687   clean_ranges (queue);
1688 }
1689
1690 static void
1691 gst_queue2_flush_temp_file (GstQueue2 * queue)
1692 {
1693   if (queue->temp_file == NULL)
1694     return;
1695
1696   GST_DEBUG_OBJECT (queue, "flushing temp file");
1697
1698   queue->temp_file = g_freopen (queue->temp_location, "wb+", queue->temp_file);
1699 }
1700
1701 static void
1702 gst_queue2_locked_flush (GstQueue2 * queue, gboolean full, gboolean clear_temp)
1703 {
1704   if (!QUEUE_IS_USING_QUEUE (queue)) {
1705     if (QUEUE_IS_USING_TEMP_FILE (queue) && clear_temp)
1706       gst_queue2_flush_temp_file (queue);
1707     init_ranges (queue);
1708   } else {
1709     while (!g_queue_is_empty (&queue->queue)) {
1710       GstQueue2Item *qitem = g_queue_pop_head (&queue->queue);
1711
1712       if (!full && qitem->type == GST_QUEUE2_ITEM_TYPE_EVENT
1713           && GST_EVENT_IS_STICKY (qitem->item)
1714           && GST_EVENT_TYPE (qitem->item) != GST_EVENT_SEGMENT
1715           && GST_EVENT_TYPE (qitem->item) != GST_EVENT_EOS) {
1716         gst_pad_store_sticky_event (queue->srcpad,
1717             GST_EVENT_CAST (qitem->item));
1718       }
1719
1720       /* Then lose another reference because we are supposed to destroy that
1721          data when flushing */
1722       if (qitem->type != GST_QUEUE2_ITEM_TYPE_QUERY)
1723         gst_mini_object_unref (qitem->item);
1724       g_slice_free (GstQueue2Item, qitem);
1725     }
1726   }
1727   queue->last_query = FALSE;
1728   g_cond_signal (&queue->query_handled);
1729   GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
1730   gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
1731   gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
1732   queue->sinktime = queue->srctime = GST_CLOCK_TIME_NONE;
1733   queue->sink_tainted = queue->src_tainted = TRUE;
1734   if (queue->starting_segment != NULL)
1735     gst_event_unref (queue->starting_segment);
1736   queue->starting_segment = NULL;
1737   queue->segment_event_received = FALSE;
1738   gst_event_replace (&queue->stream_start_event, NULL);
1739
1740   /* we deleted a lot of something */
1741   GST_QUEUE2_SIGNAL_DEL (queue);
1742 }
1743
1744 static gboolean
1745 gst_queue2_wait_free_space (GstQueue2 * queue)
1746 {
1747   /* We make space available if we're "full" according to whatever
1748    * the user defined as "full". */
1749   if (gst_queue2_is_filled (queue)) {
1750     gboolean started;
1751
1752     /* pause the timer while we wait. The fact that we are waiting does not mean
1753      * the byterate on the input pad is lower */
1754     if ((started = queue->in_timer_started))
1755       g_timer_stop (queue->in_timer);
1756
1757     GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
1758         "queue is full, waiting for free space");
1759     do {
1760       /* Wait for space to be available, we could be unlocked because of a flush. */
1761       GST_QUEUE2_WAIT_DEL_CHECK (queue, queue->sinkresult, out_flushing);
1762     }
1763     while (gst_queue2_is_filled (queue));
1764
1765     /* and continue if we were running before */
1766     if (started)
1767       g_timer_continue (queue->in_timer);
1768   }
1769   return TRUE;
1770
1771   /* ERRORS */
1772 out_flushing:
1773   {
1774     GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "queue is flushing");
1775     return FALSE;
1776   }
1777 }
1778
1779 static gboolean
1780 gst_queue2_create_write (GstQueue2 * queue, GstBuffer * buffer)
1781 {
1782   GstMapInfo info;
1783   guint8 *data, *ring_buffer;
1784   guint size, rb_size;
1785   guint64 writing_pos, new_writing_pos;
1786   GstQueue2Range *range, *prev, *next;
1787   gboolean do_seek = FALSE;
1788
1789   if (QUEUE_IS_USING_RING_BUFFER (queue))
1790     writing_pos = queue->current->rb_writing_pos;
1791   else
1792     writing_pos = queue->current->writing_pos;
1793   ring_buffer = queue->ring_buffer;
1794   rb_size = queue->ring_buffer_max_size;
1795
1796   if (!gst_buffer_map (buffer, &info, GST_MAP_READ))
1797     goto buffer_read_error;
1798
1799   size = info.size;
1800   data = info.data;
1801
1802   GST_DEBUG_OBJECT (queue, "Writing %u bytes to %" G_GUINT64_FORMAT, size,
1803       writing_pos);
1804
1805   /* sanity check */
1806   if (GST_BUFFER_OFFSET_IS_VALID (buffer) &&
1807       GST_BUFFER_OFFSET (buffer) != queue->current->writing_pos) {
1808     GST_WARNING_OBJECT (queue, "buffer offset does not match current writing "
1809         "position! %" G_GINT64_FORMAT " != %" G_GINT64_FORMAT,
1810         GST_BUFFER_OFFSET (buffer), queue->current->writing_pos);
1811   }
1812
1813   while (size > 0) {
1814     guint to_write;
1815
1816     if (QUEUE_IS_USING_RING_BUFFER (queue)) {
1817       gint64 space;
1818
1819       /* calculate the space in the ring buffer not used by data from
1820        * the current range */
1821       while (QUEUE_MAX_BYTES (queue) <= queue->cur_level.bytes) {
1822         /* wait until there is some free space */
1823         GST_QUEUE2_WAIT_DEL_CHECK (queue, queue->sinkresult, out_flushing);
1824       }
1825       /* get the amount of space we have */
1826       space = QUEUE_MAX_BYTES (queue) - queue->cur_level.bytes;
1827
1828       /* calculate if we need to split or if we can write the entire
1829        * buffer now */
1830       to_write = MIN (size, space);
1831
1832       /* the writing position in the ring buffer after writing (part
1833        * or all of) the buffer */
1834       new_writing_pos = (writing_pos + to_write) % rb_size;
1835
1836       prev = NULL;
1837       range = queue->ranges;
1838
1839       /* if we need to overwrite data in the ring buffer, we need to
1840        * update the ranges
1841        *
1842        * warning: this code is complicated and includes some
1843        * simplifications - pen, paper and diagrams for the cases
1844        * recommended! */
1845       while (range) {
1846         guint64 range_data_start, range_data_end;
1847         GstQueue2Range *range_to_destroy = NULL;
1848
1849         range_data_start = range->rb_offset;
1850         range_data_end = range->rb_writing_pos;
1851
1852         /* handle the special case where the range has no data in it */
1853         if (range->writing_pos == range->offset) {
1854           if (range != queue->current) {
1855             GST_DEBUG_OBJECT (queue,
1856                 "Removing range: offset %" G_GUINT64_FORMAT ", wpos %"
1857                 G_GUINT64_FORMAT, range->offset, range->writing_pos);
1858             /* remove range */
1859             range_to_destroy = range;
1860             if (prev)
1861               prev->next = range->next;
1862           }
1863           goto next_range;
1864         }
1865
1866         if (range_data_end > range_data_start) {
1867           if (writing_pos >= range_data_end && new_writing_pos >= writing_pos)
1868             goto next_range;
1869
1870           if (new_writing_pos > range_data_start) {
1871             if (new_writing_pos >= range_data_end) {
1872               GST_DEBUG_OBJECT (queue,
1873                   "Removing range: offset %" G_GUINT64_FORMAT ", wpos %"
1874                   G_GUINT64_FORMAT, range->offset, range->writing_pos);
1875               /* remove range */
1876               range_to_destroy = range;
1877               if (prev)
1878                 prev->next = range->next;
1879             } else {
1880               GST_DEBUG_OBJECT (queue,
1881                   "advancing offsets from %" G_GUINT64_FORMAT " (%"
1882                   G_GUINT64_FORMAT ") to %" G_GUINT64_FORMAT " (%"
1883                   G_GUINT64_FORMAT ")", range->offset, range->rb_offset,
1884                   range->offset + new_writing_pos - range_data_start,
1885                   new_writing_pos);
1886               range->offset += (new_writing_pos - range_data_start);
1887               range->rb_offset = new_writing_pos;
1888             }
1889           }
1890         } else {
1891           guint64 new_wpos_virt = writing_pos + to_write;
1892
1893           if (new_wpos_virt <= range_data_start)
1894             goto next_range;
1895
1896           if (new_wpos_virt > rb_size && new_writing_pos >= range_data_end) {
1897             GST_DEBUG_OBJECT (queue,
1898                 "Removing range: offset %" G_GUINT64_FORMAT ", wpos %"
1899                 G_GUINT64_FORMAT, range->offset, range->writing_pos);
1900             /* remove range */
1901             range_to_destroy = range;
1902             if (prev)
1903               prev->next = range->next;
1904           } else {
1905             GST_DEBUG_OBJECT (queue,
1906                 "advancing offsets from %" G_GUINT64_FORMAT " (%"
1907                 G_GUINT64_FORMAT ") to %" G_GUINT64_FORMAT " (%"
1908                 G_GUINT64_FORMAT ")", range->offset, range->rb_offset,
1909                 range->offset + new_writing_pos - range_data_start,
1910                 new_writing_pos);
1911             range->offset += (new_wpos_virt - range_data_start);
1912             range->rb_offset = new_writing_pos;
1913           }
1914         }
1915
1916       next_range:
1917         if (!range_to_destroy)
1918           prev = range;
1919
1920         range = range->next;
1921         if (range_to_destroy) {
1922           if (range_to_destroy == queue->ranges)
1923             queue->ranges = range;
1924           g_slice_free (GstQueue2Range, range_to_destroy);
1925           range_to_destroy = NULL;
1926         }
1927       }
1928     } else {
1929       to_write = size;
1930       new_writing_pos = writing_pos + to_write;
1931     }
1932
1933     if (QUEUE_IS_USING_TEMP_FILE (queue)
1934         && FSEEK_FILE (queue->temp_file, writing_pos))
1935       goto seek_failed;
1936
1937     if (new_writing_pos > writing_pos) {
1938       GST_INFO_OBJECT (queue,
1939           "writing %u bytes to range [%" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT
1940           "] (rb wpos %" G_GUINT64_FORMAT ")", to_write, queue->current->offset,
1941           queue->current->writing_pos, queue->current->rb_writing_pos);
1942       /* either not using ring buffer or no wrapping, just write */
1943       if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1944         if (fwrite (data, to_write, 1, queue->temp_file) != 1)
1945           goto handle_error;
1946       } else {
1947         memcpy (ring_buffer + writing_pos, data, to_write);
1948       }
1949
1950       if (!QUEUE_IS_USING_RING_BUFFER (queue)) {
1951         /* try to merge with next range */
1952         while ((next = queue->current->next)) {
1953           GST_INFO_OBJECT (queue,
1954               "checking merge with next range %" G_GUINT64_FORMAT " < %"
1955               G_GUINT64_FORMAT, new_writing_pos, next->offset);
1956           if (new_writing_pos < next->offset)
1957             break;
1958
1959           GST_DEBUG_OBJECT (queue, "merging ranges %" G_GUINT64_FORMAT,
1960               next->writing_pos);
1961
1962           /* remove the group */
1963           queue->current->next = next->next;
1964
1965           /* We use the threshold to decide if we want to do a seek or simply
1966            * read the data again. If there is not so much data in the range we
1967            * prefer to avoid to seek and read it again. */
1968           if (next->writing_pos > new_writing_pos + get_seek_threshold (queue)) {
1969             /* the new range had more data than the threshold, it's worth keeping
1970              * it and doing a seek. */
1971             new_writing_pos = next->writing_pos;
1972             do_seek = TRUE;
1973           }
1974           g_slice_free (GstQueue2Range, next);
1975         }
1976         goto update_and_signal;
1977       }
1978     } else {
1979       /* wrapping */
1980       guint block_one, block_two;
1981
1982       block_one = rb_size - writing_pos;
1983       block_two = to_write - block_one;
1984
1985       if (block_one > 0) {
1986         GST_INFO_OBJECT (queue, "writing %u bytes", block_one);
1987         /* write data to end of ring buffer */
1988         if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1989           if (fwrite (data, block_one, 1, queue->temp_file) != 1)
1990             goto handle_error;
1991         } else {
1992           memcpy (ring_buffer + writing_pos, data, block_one);
1993         }
1994       }
1995
1996       if (QUEUE_IS_USING_TEMP_FILE (queue) && FSEEK_FILE (queue->temp_file, 0))
1997         goto seek_failed;
1998
1999       if (block_two > 0) {
2000         GST_INFO_OBJECT (queue, "writing %u bytes", block_two);
2001         if (QUEUE_IS_USING_TEMP_FILE (queue)) {
2002           if (fwrite (data + block_one, block_two, 1, queue->temp_file) != 1)
2003             goto handle_error;
2004         } else {
2005           memcpy (ring_buffer, data + block_one, block_two);
2006         }
2007       }
2008     }
2009
2010   update_and_signal:
2011     /* update the writing positions */
2012     size -= to_write;
2013     GST_INFO_OBJECT (queue,
2014         "wrote %u bytes to %" G_GUINT64_FORMAT " (%u bytes remaining to write)",
2015         to_write, writing_pos, size);
2016
2017     if (QUEUE_IS_USING_RING_BUFFER (queue)) {
2018       data += to_write;
2019       queue->current->writing_pos += to_write;
2020       queue->current->rb_writing_pos = writing_pos = new_writing_pos;
2021     } else {
2022       queue->current->writing_pos = writing_pos = new_writing_pos;
2023     }
2024     if (do_seek)
2025       perform_seek_to_offset (queue, new_writing_pos);
2026
2027     update_cur_level (queue, queue->current);
2028
2029     /* update the buffering status */
2030     if (queue->use_buffering)
2031       update_buffering (queue);
2032
2033     GST_INFO_OBJECT (queue, "cur_level.bytes %u (max %" G_GUINT64_FORMAT ")",
2034         queue->cur_level.bytes, QUEUE_MAX_BYTES (queue));
2035
2036     GST_QUEUE2_SIGNAL_ADD (queue);
2037   }
2038
2039   gst_buffer_unmap (buffer, &info);
2040
2041   return TRUE;
2042
2043   /* ERRORS */
2044 out_flushing:
2045   {
2046     GST_DEBUG_OBJECT (queue, "we are flushing");
2047     gst_buffer_unmap (buffer, &info);
2048     /* FIXME - GST_FLOW_EOS ? */
2049     return FALSE;
2050   }
2051 seek_failed:
2052   {
2053     GST_ELEMENT_ERROR (queue, RESOURCE, SEEK, (NULL), GST_ERROR_SYSTEM);
2054     gst_buffer_unmap (buffer, &info);
2055     return FALSE;
2056   }
2057 handle_error:
2058   {
2059     switch (errno) {
2060       case ENOSPC:{
2061         GST_ELEMENT_ERROR (queue, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
2062         break;
2063       }
2064       default:{
2065         GST_ELEMENT_ERROR (queue, RESOURCE, WRITE,
2066             (_("Error while writing to download file.")),
2067             ("%s", g_strerror (errno)));
2068       }
2069     }
2070     gst_buffer_unmap (buffer, &info);
2071     return FALSE;
2072   }
2073 buffer_read_error:
2074   {
2075     GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL),
2076         ("Can't read from buffer"));
2077     return FALSE;
2078   }
2079 }
2080
2081 static gboolean
2082 buffer_list_create_write (GstBuffer ** buf, guint idx, gpointer q)
2083 {
2084   GstQueue2 *queue = q;
2085
2086   GST_TRACE_OBJECT (queue,
2087       "writing buffer %u of size %" G_GSIZE_FORMAT " bytes", idx,
2088       gst_buffer_get_size (*buf));
2089
2090   if (!gst_queue2_create_write (queue, *buf)) {
2091     GST_INFO_OBJECT (queue, "create_write() returned FALSE, bailing out");
2092     return FALSE;
2093   }
2094   return TRUE;
2095 }
2096
2097 static gboolean
2098 buffer_list_calc_size (GstBuffer ** buf, guint idx, gpointer data)
2099 {
2100   guint *p_size = data;
2101   gsize buf_size;
2102
2103   buf_size = gst_buffer_get_size (*buf);
2104   GST_TRACE ("buffer %u in has size %" G_GSIZE_FORMAT, idx, buf_size);
2105   *p_size += buf_size;
2106   return TRUE;
2107 }
2108
2109 /* enqueue an item an update the level stats */
2110 static void
2111 gst_queue2_locked_enqueue (GstQueue2 * queue, gpointer item,
2112     GstQueue2ItemType item_type)
2113 {
2114   if (item_type == GST_QUEUE2_ITEM_TYPE_BUFFER) {
2115     GstBuffer *buffer;
2116     guint size;
2117
2118     buffer = GST_BUFFER_CAST (item);
2119     size = gst_buffer_get_size (buffer);
2120
2121     /* add buffer to the statistics */
2122     if (QUEUE_IS_USING_QUEUE (queue)) {
2123       queue->cur_level.buffers++;
2124       queue->cur_level.bytes += size;
2125     }
2126     queue->bytes_in += size;
2127
2128     /* apply new buffer to segment stats */
2129     apply_buffer (queue, buffer, &queue->sink_segment, size, TRUE);
2130     /* update the byterate stats */
2131     update_in_rates (queue);
2132
2133     if (!QUEUE_IS_USING_QUEUE (queue)) {
2134       /* FIXME - check return value? */
2135       gst_queue2_create_write (queue, buffer);
2136     }
2137   } else if (item_type == GST_QUEUE2_ITEM_TYPE_BUFFER_LIST) {
2138     GstBufferList *buffer_list;
2139     guint size = 0;
2140
2141     buffer_list = GST_BUFFER_LIST_CAST (item);
2142
2143     gst_buffer_list_foreach (buffer_list, buffer_list_calc_size, &size);
2144     GST_LOG_OBJECT (queue, "total size of buffer list: %u bytes", size);
2145
2146     /* add buffer to the statistics */
2147     if (QUEUE_IS_USING_QUEUE (queue)) {
2148       queue->cur_level.buffers += gst_buffer_list_length (buffer_list);
2149       queue->cur_level.bytes += size;
2150     }
2151     queue->bytes_in += size;
2152
2153     /* apply new buffer to segment stats */
2154     apply_buffer_list (queue, buffer_list, &queue->sink_segment, TRUE);
2155
2156     /* update the byterate stats */
2157     update_in_rates (queue);
2158
2159     if (!QUEUE_IS_USING_QUEUE (queue)) {
2160       gst_buffer_list_foreach (buffer_list, buffer_list_create_write, queue);
2161     }
2162   } else if (item_type == GST_QUEUE2_ITEM_TYPE_EVENT) {
2163     GstEvent *event;
2164
2165     event = GST_EVENT_CAST (item);
2166
2167     switch (GST_EVENT_TYPE (event)) {
2168       case GST_EVENT_EOS:
2169         /* Zero the thresholds, this makes sure the queue is completely
2170          * filled and we can read all data from the queue. */
2171         GST_DEBUG_OBJECT (queue, "we have EOS");
2172         queue->is_eos = TRUE;
2173         break;
2174       case GST_EVENT_SEGMENT:
2175         apply_segment (queue, event, &queue->sink_segment, TRUE);
2176         /* This is our first new segment, we hold it
2177          * as we can't save it on the temp file */
2178         if (!QUEUE_IS_USING_QUEUE (queue)) {
2179           if (queue->segment_event_received)
2180             goto unexpected_event;
2181
2182           queue->segment_event_received = TRUE;
2183           if (queue->starting_segment != NULL)
2184             gst_event_unref (queue->starting_segment);
2185           queue->starting_segment = event;
2186           item = NULL;
2187         }
2188         /* a new segment allows us to accept more buffers if we got EOS
2189          * from downstream */
2190         queue->unexpected = FALSE;
2191         break;
2192       case GST_EVENT_GAP:
2193         apply_gap (queue, event, &queue->sink_segment, TRUE);
2194         break;
2195       case GST_EVENT_STREAM_START:
2196         if (!QUEUE_IS_USING_QUEUE (queue)) {
2197           gst_event_replace (&queue->stream_start_event, event);
2198           gst_event_unref (event);
2199           item = NULL;
2200         }
2201         break;
2202       case GST_EVENT_CAPS:{
2203         GstCaps *caps;
2204
2205         gst_event_parse_caps (event, &caps);
2206         GST_INFO ("got caps: %" GST_PTR_FORMAT, caps);
2207
2208         if (!QUEUE_IS_USING_QUEUE (queue)) {
2209           GST_LOG ("Dropping caps event, not using queue");
2210           gst_event_unref (event);
2211           item = NULL;
2212         }
2213         break;
2214       }
2215       default:
2216         if (!QUEUE_IS_USING_QUEUE (queue))
2217           goto unexpected_event;
2218         break;
2219     }
2220   } else if (GST_IS_QUERY (item)) {
2221     /* Can't happen as we check that in the caller */
2222     if (!QUEUE_IS_USING_QUEUE (queue))
2223       g_assert_not_reached ();
2224   } else {
2225     g_warning ("Unexpected item %p added in queue %s (refcounting problem?)",
2226         item, GST_OBJECT_NAME (queue));
2227     /* we can't really unref since we don't know what it is */
2228     item = NULL;
2229   }
2230
2231   if (item) {
2232     /* update the buffering status */
2233     if (queue->use_buffering)
2234       update_buffering (queue);
2235
2236     if (QUEUE_IS_USING_QUEUE (queue)) {
2237       GstQueue2Item *qitem = g_slice_new (GstQueue2Item);
2238       qitem->type = item_type;
2239       qitem->item = item;
2240       g_queue_push_tail (&queue->queue, qitem);
2241     } else {
2242       gst_mini_object_unref (GST_MINI_OBJECT_CAST (item));
2243     }
2244
2245     GST_QUEUE2_SIGNAL_ADD (queue);
2246   }
2247
2248   return;
2249
2250   /* ERRORS */
2251 unexpected_event:
2252   {
2253     gboolean is_custom = GST_EVENT_TYPE (item) < GST_EVENT_CUSTOM_UPSTREAM;
2254
2255     GST_WARNING_OBJECT (queue, "%s%s event can't be added to temp file: "
2256         "%" GST_PTR_FORMAT, is_custom ? "Unexpected " : "",
2257         GST_EVENT_TYPE_NAME (item), GST_EVENT_CAST (item));
2258     gst_event_unref (GST_EVENT_CAST (item));
2259     return;
2260   }
2261 }
2262
2263 /* dequeue an item from the queue and update level stats */
2264 static GstMiniObject *
2265 gst_queue2_locked_dequeue (GstQueue2 * queue, GstQueue2ItemType * item_type)
2266 {
2267   GstMiniObject *item;
2268
2269   if (!QUEUE_IS_USING_QUEUE (queue)) {
2270     item = gst_queue2_read_item_from_file (queue);
2271   } else {
2272     GstQueue2Item *qitem = g_queue_pop_head (&queue->queue);
2273
2274     if (qitem == NULL)
2275       goto no_item;
2276
2277     item = qitem->item;
2278     g_slice_free (GstQueue2Item, qitem);
2279   }
2280
2281   if (item == NULL)
2282     goto no_item;
2283
2284   if (GST_IS_BUFFER (item)) {
2285     GstBuffer *buffer;
2286     guint size;
2287
2288     buffer = GST_BUFFER_CAST (item);
2289     size = gst_buffer_get_size (buffer);
2290     *item_type = GST_QUEUE2_ITEM_TYPE_BUFFER;
2291
2292     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2293         "retrieved buffer %p from queue", buffer);
2294
2295     if (QUEUE_IS_USING_QUEUE (queue)) {
2296       queue->cur_level.buffers--;
2297       queue->cur_level.bytes -= size;
2298     }
2299     queue->bytes_out += size;
2300
2301     apply_buffer (queue, buffer, &queue->src_segment, size, FALSE);
2302     /* update the byterate stats */
2303     update_out_rates (queue);
2304     /* update the buffering */
2305     if (queue->use_buffering)
2306       update_buffering (queue);
2307
2308   } else if (GST_IS_EVENT (item)) {
2309     GstEvent *event = GST_EVENT_CAST (item);
2310
2311     *item_type = GST_QUEUE2_ITEM_TYPE_EVENT;
2312
2313     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2314         "retrieved event %p from queue", event);
2315
2316     switch (GST_EVENT_TYPE (event)) {
2317       case GST_EVENT_EOS:
2318         /* queue is empty now that we dequeued the EOS */
2319         GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
2320         break;
2321       case GST_EVENT_SEGMENT:
2322         apply_segment (queue, event, &queue->src_segment, FALSE);
2323         break;
2324       case GST_EVENT_GAP:
2325         apply_gap (queue, event, &queue->src_segment, FALSE);
2326         break;
2327       default:
2328         break;
2329     }
2330   } else if (GST_IS_BUFFER_LIST (item)) {
2331     GstBufferList *buffer_list;
2332     guint size = 0;
2333
2334     buffer_list = GST_BUFFER_LIST_CAST (item);
2335     gst_buffer_list_foreach (buffer_list, buffer_list_calc_size, &size);
2336     *item_type = GST_QUEUE2_ITEM_TYPE_BUFFER_LIST;
2337
2338     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2339         "retrieved buffer list %p from queue", buffer_list);
2340
2341     if (QUEUE_IS_USING_QUEUE (queue)) {
2342       queue->cur_level.buffers -= gst_buffer_list_length (buffer_list);
2343       queue->cur_level.bytes -= size;
2344     }
2345     queue->bytes_out += size;
2346
2347     apply_buffer_list (queue, buffer_list, &queue->src_segment, FALSE);
2348     /* update the byterate stats */
2349     update_out_rates (queue);
2350     /* update the buffering */
2351     if (queue->use_buffering)
2352       update_buffering (queue);
2353   } else if (GST_IS_QUERY (item)) {
2354     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2355         "retrieved query %p from queue", item);
2356     *item_type = GST_QUEUE2_ITEM_TYPE_QUERY;
2357   } else {
2358     g_warning
2359         ("Unexpected item %p dequeued from queue %s (refcounting problem?)",
2360         item, GST_OBJECT_NAME (queue));
2361     item = NULL;
2362     *item_type = GST_QUEUE2_ITEM_TYPE_UNKNOWN;
2363   }
2364   GST_QUEUE2_SIGNAL_DEL (queue);
2365
2366   return item;
2367
2368   /* ERRORS */
2369 no_item:
2370   {
2371     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "the queue is empty");
2372     return NULL;
2373   }
2374 }
2375
2376 static gboolean
2377 gst_queue2_handle_sink_event (GstPad * pad, GstObject * parent,
2378     GstEvent * event)
2379 {
2380   gboolean ret = TRUE;
2381   GstQueue2 *queue;
2382
2383   queue = GST_QUEUE2 (parent);
2384
2385   switch (GST_EVENT_TYPE (event)) {
2386     case GST_EVENT_FLUSH_START:
2387     {
2388       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush start event");
2389       if (GST_PAD_MODE (queue->srcpad) == GST_PAD_MODE_PUSH) {
2390         /* forward event */
2391         ret = gst_pad_push_event (queue->srcpad, event);
2392
2393         /* now unblock the chain function */
2394         GST_QUEUE2_MUTEX_LOCK (queue);
2395         queue->srcresult = GST_FLOW_FLUSHING;
2396         queue->sinkresult = GST_FLOW_FLUSHING;
2397         /* unblock the loop and chain functions */
2398         GST_QUEUE2_SIGNAL_ADD (queue);
2399         GST_QUEUE2_SIGNAL_DEL (queue);
2400         queue->last_query = FALSE;
2401         g_cond_signal (&queue->query_handled);
2402         GST_QUEUE2_MUTEX_UNLOCK (queue);
2403
2404         /* make sure it pauses, this should happen since we sent
2405          * flush_start downstream. */
2406         gst_pad_pause_task (queue->srcpad);
2407         GST_CAT_LOG_OBJECT (queue_dataflow, queue, "loop stopped");
2408       } else {
2409         GST_QUEUE2_MUTEX_LOCK (queue);
2410         /* flush the sink pad */
2411         queue->sinkresult = GST_FLOW_FLUSHING;
2412         GST_QUEUE2_SIGNAL_DEL (queue);
2413         queue->last_query = FALSE;
2414         g_cond_signal (&queue->query_handled);
2415         GST_QUEUE2_MUTEX_UNLOCK (queue);
2416
2417         gst_event_unref (event);
2418       }
2419       break;
2420     }
2421     case GST_EVENT_FLUSH_STOP:
2422     {
2423       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush stop event");
2424
2425       if (GST_PAD_MODE (queue->srcpad) == GST_PAD_MODE_PUSH) {
2426         /* forward event */
2427         ret = gst_pad_push_event (queue->srcpad, event);
2428
2429         GST_QUEUE2_MUTEX_LOCK (queue);
2430         gst_queue2_locked_flush (queue, FALSE, TRUE);
2431         queue->srcresult = GST_FLOW_OK;
2432         queue->sinkresult = GST_FLOW_OK;
2433         queue->is_eos = FALSE;
2434         queue->unexpected = FALSE;
2435         queue->seeking = FALSE;
2436         queue->src_tags_bitrate = queue->sink_tags_bitrate = 0;
2437         /* reset rate counters */
2438         reset_rate_timer (queue);
2439         gst_pad_start_task (queue->srcpad, (GstTaskFunction) gst_queue2_loop,
2440             queue->srcpad, NULL);
2441         GST_QUEUE2_MUTEX_UNLOCK (queue);
2442       } else {
2443         GST_QUEUE2_MUTEX_LOCK (queue);
2444         queue->segment_event_received = FALSE;
2445         queue->is_eos = FALSE;
2446         queue->unexpected = FALSE;
2447         queue->sinkresult = GST_FLOW_OK;
2448         queue->seeking = FALSE;
2449         queue->src_tags_bitrate = queue->sink_tags_bitrate = 0;
2450         GST_QUEUE2_MUTEX_UNLOCK (queue);
2451
2452         gst_event_unref (event);
2453       }
2454       break;
2455     }
2456     case GST_EVENT_TAG:{
2457       if (queue->use_tags_bitrate) {
2458         GstTagList *tags;
2459         guint bitrate;
2460
2461         gst_event_parse_tag (event, &tags);
2462         if (gst_tag_list_get_uint (tags, GST_TAG_BITRATE, &bitrate) ||
2463             gst_tag_list_get_uint (tags, GST_TAG_NOMINAL_BITRATE, &bitrate)) {
2464           GST_QUEUE2_MUTEX_LOCK (queue);
2465           queue->sink_tags_bitrate = bitrate;
2466           GST_QUEUE2_MUTEX_UNLOCK (queue);
2467           GST_LOG_OBJECT (queue, "Sink pad bitrate from tags now %u", bitrate);
2468         }
2469       }
2470       /* Fall-through */
2471     }
2472     default:
2473       if (GST_EVENT_IS_SERIALIZED (event)) {
2474         /* serialized events go in the queue */
2475         GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->sinkresult, out_flushing);
2476         if (queue->srcresult != GST_FLOW_OK) {
2477           /* Errors in sticky event pushing are no problem and ignored here
2478            * as they will cause more meaningful errors during data flow.
2479            * For EOS events, that are not followed by data flow, we still
2480            * return FALSE here though and report an error.
2481            */
2482           if (!GST_EVENT_IS_STICKY (event)) {
2483             goto out_flow_error;
2484           } else if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
2485             if (queue->srcresult == GST_FLOW_NOT_LINKED
2486                 || queue->srcresult < GST_FLOW_EOS) {
2487               GST_ELEMENT_ERROR (queue, STREAM, FAILED,
2488                   (_("Internal data flow error.")),
2489                   ("streaming task paused, reason %s (%d)",
2490                       gst_flow_get_name (queue->srcresult), queue->srcresult));
2491             }
2492             goto out_flow_error;
2493           }
2494         }
2495         /* refuse more events on EOS */
2496         if (queue->is_eos)
2497           goto out_eos;
2498         gst_queue2_locked_enqueue (queue, event, GST_QUEUE2_ITEM_TYPE_EVENT);
2499         GST_QUEUE2_MUTEX_UNLOCK (queue);
2500         gst_queue2_post_buffering (queue);
2501       } else {
2502         /* non-serialized events are passed upstream. */
2503         ret = gst_pad_push_event (queue->srcpad, event);
2504       }
2505       break;
2506   }
2507   return ret;
2508
2509   /* ERRORS */
2510 out_flushing:
2511   {
2512     GST_DEBUG_OBJECT (queue, "refusing event, we are flushing");
2513     GST_QUEUE2_MUTEX_UNLOCK (queue);
2514     gst_event_unref (event);
2515     return FALSE;
2516   }
2517 out_eos:
2518   {
2519     GST_DEBUG_OBJECT (queue, "refusing event, we are EOS");
2520     GST_QUEUE2_MUTEX_UNLOCK (queue);
2521     gst_event_unref (event);
2522     return FALSE;
2523   }
2524 out_flow_error:
2525   {
2526     GST_LOG_OBJECT (queue,
2527         "refusing event, we have a downstream flow error: %s",
2528         gst_flow_get_name (queue->srcresult));
2529     GST_QUEUE2_MUTEX_UNLOCK (queue);
2530     gst_event_unref (event);
2531     return FALSE;
2532   }
2533 }
2534
2535 static gboolean
2536 gst_queue2_handle_sink_query (GstPad * pad, GstObject * parent,
2537     GstQuery * query)
2538 {
2539   GstQueue2 *queue;
2540   gboolean res;
2541
2542   queue = GST_QUEUE2 (parent);
2543
2544   switch (GST_QUERY_TYPE (query)) {
2545     default:
2546       if (GST_QUERY_IS_SERIALIZED (query)) {
2547         GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received query %p", query);
2548         /* serialized events go in the queue. We need to be certain that we
2549          * don't cause deadlocks waiting for the query return value. We check if
2550          * the queue is empty (nothing is blocking downstream and the query can
2551          * be pushed for sure) or we are not buffering. If we are buffering,
2552          * the pipeline waits to unblock downstream until our queue fills up
2553          * completely, which can not happen if we block on the query..
2554          * Therefore we only potentially block when we are not buffering. */
2555         GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->sinkresult, out_flushing);
2556         if (QUEUE_IS_USING_QUEUE (queue) && (gst_queue2_is_empty (queue)
2557                 || !queue->use_buffering)) {
2558           if (!g_atomic_int_get (&queue->downstream_may_block)) {
2559             gst_queue2_locked_enqueue (queue, query,
2560                 GST_QUEUE2_ITEM_TYPE_QUERY);
2561
2562             STATUS (queue, queue->sinkpad, "wait for QUERY");
2563             g_cond_wait (&queue->query_handled, &queue->qlock);
2564             if (queue->sinkresult != GST_FLOW_OK)
2565               goto out_flushing;
2566             res = queue->last_query;
2567           } else {
2568             GST_DEBUG_OBJECT (queue, "refusing query, downstream might block");
2569             res = FALSE;
2570           }
2571         } else {
2572           GST_DEBUG_OBJECT (queue,
2573               "refusing query, we are not using the queue");
2574           res = FALSE;
2575         }
2576         GST_QUEUE2_MUTEX_UNLOCK (queue);
2577         gst_queue2_post_buffering (queue);
2578       } else {
2579         res = gst_pad_query_default (pad, parent, query);
2580       }
2581       break;
2582   }
2583   return res;
2584
2585   /* ERRORS */
2586 out_flushing:
2587   {
2588     GST_DEBUG_OBJECT (queue, "refusing query, we are flushing");
2589     GST_QUEUE2_MUTEX_UNLOCK (queue);
2590     return FALSE;
2591   }
2592 }
2593
2594 static gboolean
2595 gst_queue2_is_empty (GstQueue2 * queue)
2596 {
2597   /* never empty on EOS */
2598   if (queue->is_eos)
2599     return FALSE;
2600
2601   if (!QUEUE_IS_USING_QUEUE (queue) && queue->current) {
2602     return queue->current->writing_pos <= queue->current->max_reading_pos;
2603   } else {
2604     if (queue->queue.length == 0)
2605       return TRUE;
2606   }
2607
2608   return FALSE;
2609 }
2610
2611 static gboolean
2612 gst_queue2_is_filled (GstQueue2 * queue)
2613 {
2614   gboolean res;
2615
2616   /* always filled on EOS */
2617   if (queue->is_eos)
2618     return TRUE;
2619
2620 #define CHECK_FILLED(format,alt_max) ((queue->max_level.format) > 0 && \
2621     (queue->cur_level.format) >= ((alt_max) ? \
2622       MIN ((queue->max_level.format), (alt_max)) : (queue->max_level.format)))
2623
2624   /* if using a ring buffer we're filled if all ring buffer space is used
2625    * _by the current range_ */
2626   if (QUEUE_IS_USING_RING_BUFFER (queue)) {
2627     guint64 rb_size = queue->ring_buffer_max_size;
2628     GST_DEBUG_OBJECT (queue,
2629         "max bytes %u, rb size %" G_GUINT64_FORMAT ", cur bytes %u",
2630         queue->max_level.bytes, rb_size, queue->cur_level.bytes);
2631     return CHECK_FILLED (bytes, rb_size);
2632   }
2633
2634   /* if using file, we're never filled if we don't have EOS */
2635   if (QUEUE_IS_USING_TEMP_FILE (queue))
2636     return FALSE;
2637
2638   /* we are never filled when we have no buffers at all */
2639   if (queue->cur_level.buffers == 0)
2640     return FALSE;
2641
2642   /* we are filled if one of the current levels exceeds the max */
2643   res = CHECK_FILLED (buffers, 0) || CHECK_FILLED (bytes, 0)
2644       || CHECK_FILLED (time, 0);
2645
2646   /* if we need to, use the rate estimate to check against the max time we are
2647    * allowed to queue */
2648   if (queue->use_rate_estimate)
2649     res |= CHECK_FILLED (rate_time, 0);
2650
2651 #undef CHECK_FILLED
2652   return res;
2653 }
2654
2655 static GstFlowReturn
2656 gst_queue2_chain_buffer_or_buffer_list (GstQueue2 * queue,
2657     GstMiniObject * item, GstQueue2ItemType item_type)
2658 {
2659   /* we have to lock the queue since we span threads */
2660   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->sinkresult, out_flushing);
2661   /* when we received EOS, we refuse more data */
2662   if (queue->is_eos)
2663     goto out_eos;
2664   /* when we received unexpected from downstream, refuse more buffers */
2665   if (queue->unexpected)
2666     goto out_unexpected;
2667
2668   /* while we didn't receive the newsegment, we're seeking and we skip data */
2669   if (queue->seeking)
2670     goto out_seeking;
2671
2672   if (!gst_queue2_wait_free_space (queue))
2673     goto out_flushing;
2674
2675   /* put buffer in queue now */
2676   gst_queue2_locked_enqueue (queue, item, item_type);
2677   GST_QUEUE2_MUTEX_UNLOCK (queue);
2678   gst_queue2_post_buffering (queue);
2679
2680   return GST_FLOW_OK;
2681
2682   /* special conditions */
2683 out_flushing:
2684   {
2685     GstFlowReturn ret = queue->sinkresult;
2686
2687     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2688         "exit because task paused, reason: %s", gst_flow_get_name (ret));
2689     GST_QUEUE2_MUTEX_UNLOCK (queue);
2690     gst_mini_object_unref (item);
2691
2692     return ret;
2693   }
2694 out_eos:
2695   {
2696     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we received EOS");
2697     GST_QUEUE2_MUTEX_UNLOCK (queue);
2698     gst_mini_object_unref (item);
2699
2700     return GST_FLOW_EOS;
2701   }
2702 out_seeking:
2703   {
2704     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are seeking");
2705     GST_QUEUE2_MUTEX_UNLOCK (queue);
2706     gst_mini_object_unref (item);
2707
2708     return GST_FLOW_OK;
2709   }
2710 out_unexpected:
2711   {
2712     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we received EOS");
2713     GST_QUEUE2_MUTEX_UNLOCK (queue);
2714     gst_mini_object_unref (item);
2715
2716     return GST_FLOW_EOS;
2717   }
2718 }
2719
2720 static GstFlowReturn
2721 gst_queue2_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
2722 {
2723   GstQueue2 *queue;
2724
2725   queue = GST_QUEUE2 (parent);
2726
2727   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received buffer %p of "
2728       "size %" G_GSIZE_FORMAT ", time %" GST_TIME_FORMAT ", duration %"
2729       GST_TIME_FORMAT, buffer, gst_buffer_get_size (buffer),
2730       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
2731       GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
2732
2733   return gst_queue2_chain_buffer_or_buffer_list (queue,
2734       GST_MINI_OBJECT_CAST (buffer), GST_QUEUE2_ITEM_TYPE_BUFFER);
2735 }
2736
2737 static GstFlowReturn
2738 gst_queue2_chain_list (GstPad * pad, GstObject * parent,
2739     GstBufferList * buffer_list)
2740 {
2741   GstQueue2 *queue;
2742
2743   queue = GST_QUEUE2 (parent);
2744
2745   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2746       "received buffer list %p", buffer_list);
2747
2748   return gst_queue2_chain_buffer_or_buffer_list (queue,
2749       GST_MINI_OBJECT_CAST (buffer_list), GST_QUEUE2_ITEM_TYPE_BUFFER_LIST);
2750 }
2751
2752 static GstMiniObject *
2753 gst_queue2_dequeue_on_eos (GstQueue2 * queue, GstQueue2ItemType * item_type)
2754 {
2755   GstMiniObject *data;
2756
2757   GST_CAT_LOG_OBJECT (queue_dataflow, queue, "got EOS from downstream");
2758
2759   /* stop pushing buffers, we dequeue all items until we see an item that we
2760    * can push again, which is EOS or SEGMENT. If there is nothing in the
2761    * queue we can push, we set a flag to make the sinkpad refuse more
2762    * buffers with an EOS return value until we receive something
2763    * pushable again or we get flushed. */
2764   while ((data = gst_queue2_locked_dequeue (queue, item_type))) {
2765     if (*item_type == GST_QUEUE2_ITEM_TYPE_BUFFER) {
2766       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2767           "dropping EOS buffer %p", data);
2768       gst_buffer_unref (GST_BUFFER_CAST (data));
2769     } else if (*item_type == GST_QUEUE2_ITEM_TYPE_EVENT) {
2770       GstEvent *event = GST_EVENT_CAST (data);
2771       GstEventType type = GST_EVENT_TYPE (event);
2772
2773       if (type == GST_EVENT_EOS || type == GST_EVENT_SEGMENT) {
2774         /* we found a pushable item in the queue, push it out */
2775         GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2776             "pushing pushable event %s after EOS", GST_EVENT_TYPE_NAME (event));
2777         return data;
2778       }
2779       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2780           "dropping EOS event %p", event);
2781       gst_event_unref (event);
2782     } else if (*item_type == GST_QUEUE2_ITEM_TYPE_BUFFER_LIST) {
2783       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2784           "dropping EOS buffer list %p", data);
2785       gst_buffer_list_unref (GST_BUFFER_LIST_CAST (data));
2786     } else if (*item_type == GST_QUEUE2_ITEM_TYPE_QUERY) {
2787       queue->last_query = FALSE;
2788       g_cond_signal (&queue->query_handled);
2789       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "dropping EOS query %p", data);
2790     }
2791   }
2792   /* no more items in the queue. Set the unexpected flag so that upstream
2793    * make us refuse any more buffers on the sinkpad. Since we will still
2794    * accept EOS and SEGMENT we return _FLOW_OK to the caller so that the
2795    * task function does not shut down. */
2796   queue->unexpected = TRUE;
2797   return NULL;
2798 }
2799
2800 /* dequeue an item from the queue an push it downstream. This functions returns
2801  * the result of the push. */
2802 static GstFlowReturn
2803 gst_queue2_push_one (GstQueue2 * queue)
2804 {
2805   GstFlowReturn result = queue->srcresult;
2806   GstMiniObject *data;
2807   GstQueue2ItemType item_type;
2808
2809   data = gst_queue2_locked_dequeue (queue, &item_type);
2810   if (data == NULL)
2811     goto no_item;
2812
2813 next:
2814   STATUS (queue, queue->srcpad, "We have something dequeud");
2815   g_atomic_int_set (&queue->downstream_may_block,
2816       item_type == GST_QUEUE2_ITEM_TYPE_BUFFER ||
2817       item_type == GST_QUEUE2_ITEM_TYPE_BUFFER_LIST);
2818   GST_QUEUE2_MUTEX_UNLOCK (queue);
2819   gst_queue2_post_buffering (queue);
2820
2821   if (item_type == GST_QUEUE2_ITEM_TYPE_BUFFER) {
2822     GstBuffer *buffer;
2823
2824     buffer = GST_BUFFER_CAST (data);
2825
2826     result = gst_pad_push (queue->srcpad, buffer);
2827     g_atomic_int_set (&queue->downstream_may_block, 0);
2828
2829     /* need to check for srcresult here as well */
2830     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2831     if (result == GST_FLOW_EOS) {
2832       data = gst_queue2_dequeue_on_eos (queue, &item_type);
2833       if (data != NULL)
2834         goto next;
2835       /* Since we will still accept EOS and SEGMENT we return _FLOW_OK
2836        * to the caller so that the task function does not shut down */
2837       result = GST_FLOW_OK;
2838     }
2839   } else if (item_type == GST_QUEUE2_ITEM_TYPE_EVENT) {
2840     GstEvent *event = GST_EVENT_CAST (data);
2841     GstEventType type = GST_EVENT_TYPE (event);
2842
2843     if (type == GST_EVENT_TAG) {
2844       if (queue->use_tags_bitrate) {
2845         GstTagList *tags;
2846         guint bitrate;
2847
2848         gst_event_parse_tag (event, &tags);
2849         if (gst_tag_list_get_uint (tags, GST_TAG_BITRATE, &bitrate) ||
2850             gst_tag_list_get_uint (tags, GST_TAG_NOMINAL_BITRATE, &bitrate)) {
2851           GST_QUEUE2_MUTEX_LOCK (queue);
2852           queue->src_tags_bitrate = bitrate;
2853           GST_QUEUE2_MUTEX_UNLOCK (queue);
2854           GST_LOG_OBJECT (queue, "src pad bitrate from tags now %u", bitrate);
2855         }
2856       }
2857     }
2858
2859     gst_pad_push_event (queue->srcpad, event);
2860
2861     /* if we're EOS, return EOS so that the task pauses. */
2862     if (type == GST_EVENT_EOS) {
2863       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2864           "pushed EOS event %p, return EOS", event);
2865       result = GST_FLOW_EOS;
2866     }
2867
2868     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2869   } else if (item_type == GST_QUEUE2_ITEM_TYPE_BUFFER_LIST) {
2870     GstBufferList *buffer_list;
2871
2872     buffer_list = GST_BUFFER_LIST_CAST (data);
2873
2874     result = gst_pad_push_list (queue->srcpad, buffer_list);
2875     g_atomic_int_set (&queue->downstream_may_block, 0);
2876
2877     /* need to check for srcresult here as well */
2878     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2879     if (result == GST_FLOW_EOS) {
2880       data = gst_queue2_dequeue_on_eos (queue, &item_type);
2881       if (data != NULL)
2882         goto next;
2883       /* Since we will still accept EOS and SEGMENT we return _FLOW_OK
2884        * to the caller so that the task function does not shut down */
2885       result = GST_FLOW_OK;
2886     }
2887   } else if (item_type == GST_QUEUE2_ITEM_TYPE_QUERY) {
2888     GstQuery *query = GST_QUERY_CAST (data);
2889
2890     GST_LOG_OBJECT (queue->srcpad, "Peering query %p", query);
2891     queue->last_query = gst_pad_peer_query (queue->srcpad, query);
2892     GST_LOG_OBJECT (queue->srcpad, "Peered query");
2893     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2894         "did query %p, return %d", query, queue->last_query);
2895     g_cond_signal (&queue->query_handled);
2896     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2897     result = GST_FLOW_OK;
2898   }
2899   return result;
2900
2901   /* ERRORS */
2902 no_item:
2903   {
2904     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2905         "exit because we have no item in the queue");
2906     return GST_FLOW_ERROR;
2907   }
2908 out_flushing:
2909   {
2910     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are flushing");
2911     return GST_FLOW_FLUSHING;
2912   }
2913 }
2914
2915 /* called repeatedly with @pad as the source pad. This function should push out
2916  * data to the peer element. */
2917 static void
2918 gst_queue2_loop (GstPad * pad)
2919 {
2920   GstQueue2 *queue;
2921   GstFlowReturn ret;
2922
2923   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
2924
2925   /* have to lock for thread-safety */
2926   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
2927
2928   if (gst_queue2_is_empty (queue)) {
2929     gboolean started;
2930
2931     /* pause the timer while we wait. The fact that we are waiting does not mean
2932      * the byterate on the output pad is lower */
2933     if ((started = queue->out_timer_started))
2934       g_timer_stop (queue->out_timer);
2935
2936     GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
2937         "queue is empty, waiting for new data");
2938     do {
2939       /* Wait for data to be available, we could be unlocked because of a flush. */
2940       GST_QUEUE2_WAIT_ADD_CHECK (queue, queue->srcresult, out_flushing);
2941     }
2942     while (gst_queue2_is_empty (queue));
2943
2944     /* and continue if we were running before */
2945     if (started)
2946       g_timer_continue (queue->out_timer);
2947   }
2948   ret = gst_queue2_push_one (queue);
2949   queue->srcresult = ret;
2950   queue->sinkresult = ret;
2951   if (ret != GST_FLOW_OK)
2952     goto out_flushing;
2953
2954   GST_QUEUE2_MUTEX_UNLOCK (queue);
2955   gst_queue2_post_buffering (queue);
2956
2957   return;
2958
2959   /* ERRORS */
2960 out_flushing:
2961   {
2962     gboolean eos = queue->is_eos;
2963     GstFlowReturn ret = queue->srcresult;
2964
2965     gst_pad_pause_task (queue->srcpad);
2966     if (ret == GST_FLOW_FLUSHING) {
2967       gst_queue2_locked_flush (queue, FALSE, FALSE);
2968     } else {
2969       GST_QUEUE2_SIGNAL_DEL (queue);
2970       queue->last_query = FALSE;
2971       g_cond_signal (&queue->query_handled);
2972     }
2973     GST_QUEUE2_MUTEX_UNLOCK (queue);
2974     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
2975         "pause task, reason:  %s", gst_flow_get_name (queue->srcresult));
2976     /* let app know about us giving up if upstream is not expected to do so */
2977     /* EOS is already taken care of elsewhere */
2978     if (eos && (ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS)) {
2979       GST_ELEMENT_ERROR (queue, STREAM, FAILED,
2980           (_("Internal data flow error.")),
2981           ("streaming task paused, reason %s (%d)",
2982               gst_flow_get_name (ret), ret));
2983       gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
2984     }
2985     return;
2986   }
2987 }
2988
2989 static gboolean
2990 gst_queue2_handle_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
2991 {
2992   gboolean res = TRUE;
2993   GstQueue2 *queue = GST_QUEUE2 (parent);
2994
2995 #ifndef GST_DISABLE_GST_DEBUG
2996   GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "got event %p (%s)",
2997       event, GST_EVENT_TYPE_NAME (event));
2998 #endif
2999
3000   switch (GST_EVENT_TYPE (event)) {
3001     case GST_EVENT_FLUSH_START:
3002       if (QUEUE_IS_USING_QUEUE (queue)) {
3003         /* just forward upstream */
3004         res = gst_pad_push_event (queue->sinkpad, event);
3005       } else {
3006         /* now unblock the getrange function */
3007         GST_QUEUE2_MUTEX_LOCK (queue);
3008         GST_DEBUG_OBJECT (queue, "flushing");
3009         queue->srcresult = GST_FLOW_FLUSHING;
3010         GST_QUEUE2_SIGNAL_ADD (queue);
3011         GST_QUEUE2_MUTEX_UNLOCK (queue);
3012
3013         /* when using a temp file, we eat the event */
3014         res = TRUE;
3015         gst_event_unref (event);
3016       }
3017       break;
3018     case GST_EVENT_FLUSH_STOP:
3019       if (QUEUE_IS_USING_QUEUE (queue)) {
3020         /* just forward upstream */
3021         res = gst_pad_push_event (queue->sinkpad, event);
3022       } else {
3023         /* now unblock the getrange function */
3024         GST_QUEUE2_MUTEX_LOCK (queue);
3025         queue->srcresult = GST_FLOW_OK;
3026         GST_QUEUE2_MUTEX_UNLOCK (queue);
3027
3028         /* when using a temp file, we eat the event */
3029         res = TRUE;
3030         gst_event_unref (event);
3031       }
3032       break;
3033     case GST_EVENT_RECONFIGURE:
3034       GST_QUEUE2_MUTEX_LOCK (queue);
3035       /* assume downstream is linked now and try to push again */
3036       if (queue->srcresult == GST_FLOW_NOT_LINKED) {
3037         queue->srcresult = GST_FLOW_OK;
3038         queue->sinkresult = GST_FLOW_OK;
3039         if (GST_PAD_MODE (pad) == GST_PAD_MODE_PUSH) {
3040           gst_pad_start_task (pad, (GstTaskFunction) gst_queue2_loop, pad,
3041               NULL);
3042         }
3043       }
3044       GST_QUEUE2_MUTEX_UNLOCK (queue);
3045
3046       res = gst_pad_push_event (queue->sinkpad, event);
3047       break;
3048     default:
3049       res = gst_pad_push_event (queue->sinkpad, event);
3050       break;
3051   }
3052
3053   return res;
3054 }
3055
3056 static gboolean
3057 gst_queue2_handle_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
3058 {
3059   GstQueue2 *queue;
3060
3061   queue = GST_QUEUE2 (parent);
3062
3063   switch (GST_QUERY_TYPE (query)) {
3064     case GST_QUERY_POSITION:
3065     {
3066       gint64 peer_pos;
3067       GstFormat format;
3068
3069       if (!gst_pad_peer_query (queue->sinkpad, query))
3070         goto peer_failed;
3071
3072       /* get peer position */
3073       gst_query_parse_position (query, &format, &peer_pos);
3074
3075       /* FIXME: this code assumes that there's no discont in the queue */
3076       switch (format) {
3077         case GST_FORMAT_BYTES:
3078           peer_pos -= queue->cur_level.bytes;
3079           break;
3080         case GST_FORMAT_TIME:
3081           peer_pos -= queue->cur_level.time;
3082           break;
3083         default:
3084           GST_WARNING_OBJECT (queue, "dropping query in %s format, don't "
3085               "know how to adjust value", gst_format_get_name (format));
3086           return FALSE;
3087       }
3088       /* set updated position */
3089       gst_query_set_position (query, format, peer_pos);
3090       break;
3091     }
3092     case GST_QUERY_DURATION:
3093     {
3094       GST_DEBUG_OBJECT (queue, "doing peer query");
3095
3096       if (!gst_pad_peer_query (queue->sinkpad, query))
3097         goto peer_failed;
3098
3099       GST_DEBUG_OBJECT (queue, "peer query success");
3100       break;
3101     }
3102     case GST_QUERY_BUFFERING:
3103     {
3104       gint percent;
3105       gboolean is_buffering;
3106       GstBufferingMode mode;
3107       gint avg_in, avg_out;
3108       gint64 buffering_left;
3109
3110       GST_DEBUG_OBJECT (queue, "query buffering");
3111
3112       get_buffering_percent (queue, &is_buffering, &percent);
3113       gst_query_set_buffering_percent (query, is_buffering, percent);
3114
3115       get_buffering_stats (queue, percent, &mode, &avg_in, &avg_out,
3116           &buffering_left);
3117       gst_query_set_buffering_stats (query, mode, avg_in, avg_out,
3118           buffering_left);
3119
3120       if (!QUEUE_IS_USING_QUEUE (queue)) {
3121         /* add ranges for download and ringbuffer buffering */
3122         GstFormat format;
3123         gint64 start, stop, range_start, range_stop;
3124         guint64 writing_pos;
3125         gint64 estimated_total;
3126         gint64 duration;
3127         gboolean peer_res, is_eos;
3128         GstQueue2Range *queued_ranges;
3129
3130         /* we need a current download region */
3131         if (queue->current == NULL)
3132           return FALSE;
3133
3134         writing_pos = queue->current->writing_pos;
3135         is_eos = queue->is_eos;
3136
3137         if (is_eos) {
3138           /* we're EOS, we know the duration in bytes now */
3139           peer_res = TRUE;
3140           duration = writing_pos;
3141         } else {
3142           /* get duration of upstream in bytes */
3143           peer_res = gst_pad_peer_query_duration (queue->sinkpad,
3144               GST_FORMAT_BYTES, &duration);
3145         }
3146
3147         GST_DEBUG_OBJECT (queue, "percent %d, duration %" G_GINT64_FORMAT
3148             ", writing %" G_GINT64_FORMAT, percent, duration, writing_pos);
3149
3150         /* calculate remaining and total download time */
3151         if (peer_res && avg_in > 0.0)
3152           estimated_total = ((duration - writing_pos) * 1000) / avg_in;
3153         else
3154           estimated_total = -1;
3155
3156         GST_DEBUG_OBJECT (queue, "estimated-total %" G_GINT64_FORMAT,
3157             estimated_total);
3158
3159         gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
3160
3161         switch (format) {
3162           case GST_FORMAT_PERCENT:
3163             /* we need duration */
3164             if (!peer_res)
3165               goto peer_failed;
3166
3167             start = 0;
3168             /* get our available data relative to the duration */
3169             if (duration != -1)
3170               stop =
3171                   gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX, writing_pos,
3172                   duration);
3173             else
3174               stop = -1;
3175             break;
3176           case GST_FORMAT_BYTES:
3177             start = 0;
3178             stop = writing_pos;
3179             break;
3180           default:
3181             start = -1;
3182             stop = -1;
3183             break;
3184         }
3185
3186         /* fill out the buffered ranges */
3187         for (queued_ranges = queue->ranges; queued_ranges;
3188             queued_ranges = queued_ranges->next) {
3189           switch (format) {
3190             case GST_FORMAT_PERCENT:
3191               if (duration == -1) {
3192                 range_start = 0;
3193                 range_stop = 0;
3194                 break;
3195               }
3196               range_start =
3197                   gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
3198                   queued_ranges->offset, duration);
3199               range_stop =
3200                   gst_util_uint64_scale (GST_FORMAT_PERCENT_MAX,
3201                   queued_ranges->writing_pos, duration);
3202               break;
3203             case GST_FORMAT_BYTES:
3204               range_start = queued_ranges->offset;
3205               range_stop = queued_ranges->writing_pos;
3206               break;
3207             default:
3208               range_start = -1;
3209               range_stop = -1;
3210               break;
3211           }
3212           if (range_start == range_stop)
3213             continue;
3214           GST_DEBUG_OBJECT (queue,
3215               "range starting at %" G_GINT64_FORMAT " and finishing at %"
3216               G_GINT64_FORMAT, range_start, range_stop);
3217           gst_query_add_buffering_range (query, range_start, range_stop);
3218         }
3219
3220         gst_query_set_buffering_range (query, format, start, stop,
3221             estimated_total);
3222       }
3223       break;
3224     }
3225     case GST_QUERY_SCHEDULING:
3226     {
3227       gboolean pull_mode;
3228       GstSchedulingFlags flags = 0;
3229
3230       if (!gst_pad_peer_query (queue->sinkpad, query))
3231         goto peer_failed;
3232
3233       gst_query_parse_scheduling (query, &flags, NULL, NULL, NULL);
3234
3235       /* we can operate in pull mode when we are using a tempfile */
3236       pull_mode = !QUEUE_IS_USING_QUEUE (queue);
3237
3238       if (pull_mode)
3239         flags |= GST_SCHEDULING_FLAG_SEEKABLE;
3240       gst_query_set_scheduling (query, flags, 0, -1, 0);
3241       if (pull_mode)
3242         gst_query_add_scheduling_mode (query, GST_PAD_MODE_PULL);
3243       gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
3244       break;
3245     }
3246     default:
3247       /* peer handled other queries */
3248       if (!gst_pad_query_default (pad, parent, query))
3249         goto peer_failed;
3250       break;
3251   }
3252
3253   return TRUE;
3254
3255   /* ERRORS */
3256 peer_failed:
3257   {
3258     GST_DEBUG_OBJECT (queue, "failed peer query");
3259     return FALSE;
3260   }
3261 }
3262
3263 static gboolean
3264 gst_queue2_handle_query (GstElement * element, GstQuery * query)
3265 {
3266   GstQueue2 *queue = GST_QUEUE2 (element);
3267
3268   /* simply forward to the srcpad query function */
3269   return gst_queue2_handle_src_query (queue->srcpad, GST_OBJECT_CAST (element),
3270       query);
3271 }
3272
3273 static void
3274 gst_queue2_update_upstream_size (GstQueue2 * queue)
3275 {
3276   gint64 upstream_size = -1;
3277
3278   if (gst_pad_peer_query_duration (queue->sinkpad, GST_FORMAT_BYTES,
3279           &upstream_size)) {
3280     GST_INFO_OBJECT (queue, "upstream size: %" G_GINT64_FORMAT, upstream_size);
3281
3282     /* upstream_size can be negative but queue->upstream_size is unsigned.
3283      * Prevent setting negative values to it (the query can return -1) */
3284     if (upstream_size >= 0)
3285       queue->upstream_size = upstream_size;
3286     else
3287       queue->upstream_size = 0;
3288   }
3289 }
3290
3291 static GstFlowReturn
3292 gst_queue2_get_range (GstPad * pad, GstObject * parent, guint64 offset,
3293     guint length, GstBuffer ** buffer)
3294 {
3295   GstQueue2 *queue;
3296   GstFlowReturn ret;
3297
3298   queue = GST_QUEUE2_CAST (parent);
3299
3300   length = (length == -1) ? DEFAULT_BUFFER_SIZE : length;
3301   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, queue->srcresult, out_flushing);
3302   offset = (offset == -1) ? queue->current->reading_pos : offset;
3303
3304   GST_DEBUG_OBJECT (queue,
3305       "Getting range: offset %" G_GUINT64_FORMAT ", length %u", offset, length);
3306
3307   /* catch any reads beyond the size of the file here to make sure queue2
3308    * doesn't send seek events beyond the size of the file upstream, since
3309    * that would confuse elements such as souphttpsrc and/or http servers.
3310    * Demuxers often just loop until EOS at the end of the file to figure out
3311    * when they've read all the end-headers or index chunks. */
3312   if (G_UNLIKELY (offset >= queue->upstream_size)) {
3313     gst_queue2_update_upstream_size (queue);
3314     if (queue->upstream_size > 0 && offset >= queue->upstream_size)
3315       goto out_unexpected;
3316   }
3317
3318   if (G_UNLIKELY (offset + length > queue->upstream_size)) {
3319     gst_queue2_update_upstream_size (queue);
3320     if (queue->upstream_size > 0 && offset + length >= queue->upstream_size) {
3321       length = queue->upstream_size - offset;
3322       GST_DEBUG_OBJECT (queue, "adjusting length downto %d", length);
3323     }
3324   }
3325
3326   /* FIXME - function will block when the range is not yet available */
3327   ret = gst_queue2_create_read (queue, offset, length, buffer);
3328   GST_QUEUE2_MUTEX_UNLOCK (queue);
3329   gst_queue2_post_buffering (queue);
3330
3331   return ret;
3332
3333   /* ERRORS */
3334 out_flushing:
3335   {
3336     ret = queue->srcresult;
3337
3338     GST_DEBUG_OBJECT (queue, "we are flushing");
3339     GST_QUEUE2_MUTEX_UNLOCK (queue);
3340     return ret;
3341   }
3342 out_unexpected:
3343   {
3344     GST_DEBUG_OBJECT (queue, "read beyond end of file");
3345     GST_QUEUE2_MUTEX_UNLOCK (queue);
3346     return GST_FLOW_EOS;
3347   }
3348 }
3349
3350 /* sink currently only operates in push mode */
3351 static gboolean
3352 gst_queue2_sink_activate_mode (GstPad * pad, GstObject * parent,
3353     GstPadMode mode, gboolean active)
3354 {
3355   gboolean result;
3356   GstQueue2 *queue;
3357
3358   queue = GST_QUEUE2 (parent);
3359
3360   switch (mode) {
3361     case GST_PAD_MODE_PUSH:
3362       if (active) {
3363         GST_QUEUE2_MUTEX_LOCK (queue);
3364         GST_DEBUG_OBJECT (queue, "activating push mode");
3365         queue->srcresult = GST_FLOW_OK;
3366         queue->sinkresult = GST_FLOW_OK;
3367         queue->is_eos = FALSE;
3368         queue->unexpected = FALSE;
3369         reset_rate_timer (queue);
3370         GST_QUEUE2_MUTEX_UNLOCK (queue);
3371       } else {
3372         /* unblock chain function */
3373         GST_QUEUE2_MUTEX_LOCK (queue);
3374         GST_DEBUG_OBJECT (queue, "deactivating push mode");
3375         queue->srcresult = GST_FLOW_FLUSHING;
3376         queue->sinkresult = GST_FLOW_FLUSHING;
3377         GST_QUEUE2_SIGNAL_DEL (queue);
3378         /* Unblock query handler */
3379         queue->last_query = FALSE;
3380         g_cond_signal (&queue->query_handled);
3381         GST_QUEUE2_MUTEX_UNLOCK (queue);
3382
3383         /* wait until it is unblocked and clean up */
3384         GST_PAD_STREAM_LOCK (pad);
3385         GST_QUEUE2_MUTEX_LOCK (queue);
3386         gst_queue2_locked_flush (queue, TRUE, FALSE);
3387         GST_QUEUE2_MUTEX_UNLOCK (queue);
3388         GST_PAD_STREAM_UNLOCK (pad);
3389       }
3390       result = TRUE;
3391       break;
3392     default:
3393       result = FALSE;
3394       break;
3395   }
3396   return result;
3397 }
3398
3399 /* src operating in push mode, we start a task on the source pad that pushes out
3400  * buffers from the queue */
3401 static gboolean
3402 gst_queue2_src_activate_push (GstPad * pad, GstObject * parent, gboolean active)
3403 {
3404   gboolean result = FALSE;
3405   GstQueue2 *queue;
3406
3407   queue = GST_QUEUE2 (parent);
3408
3409   if (active) {
3410     GST_QUEUE2_MUTEX_LOCK (queue);
3411     GST_DEBUG_OBJECT (queue, "activating push mode");
3412     queue->srcresult = GST_FLOW_OK;
3413     queue->sinkresult = GST_FLOW_OK;
3414     queue->is_eos = FALSE;
3415     queue->unexpected = FALSE;
3416     result =
3417         gst_pad_start_task (pad, (GstTaskFunction) gst_queue2_loop, pad, NULL);
3418     GST_QUEUE2_MUTEX_UNLOCK (queue);
3419   } else {
3420     /* unblock loop function */
3421     GST_QUEUE2_MUTEX_LOCK (queue);
3422     GST_DEBUG_OBJECT (queue, "deactivating push mode");
3423     queue->srcresult = GST_FLOW_FLUSHING;
3424     queue->sinkresult = GST_FLOW_FLUSHING;
3425     /* the item add signal will unblock */
3426     GST_QUEUE2_SIGNAL_ADD (queue);
3427     GST_QUEUE2_MUTEX_UNLOCK (queue);
3428
3429     /* step 2, make sure streaming finishes */
3430     result = gst_pad_stop_task (pad);
3431   }
3432
3433   return result;
3434 }
3435
3436 /* pull mode, downstream will call our getrange function */
3437 static gboolean
3438 gst_queue2_src_activate_pull (GstPad * pad, GstObject * parent, gboolean active)
3439 {
3440   gboolean result;
3441   GstQueue2 *queue;
3442
3443   queue = GST_QUEUE2 (parent);
3444
3445   if (active) {
3446     GST_QUEUE2_MUTEX_LOCK (queue);
3447     if (!QUEUE_IS_USING_QUEUE (queue)) {
3448       if (QUEUE_IS_USING_TEMP_FILE (queue)) {
3449         /* open the temp file now */
3450         result = gst_queue2_open_temp_location_file (queue);
3451       } else if (!queue->ring_buffer) {
3452         queue->ring_buffer = g_malloc (queue->ring_buffer_max_size);
3453         result = ! !queue->ring_buffer;
3454       } else {
3455         result = TRUE;
3456       }
3457
3458       GST_DEBUG_OBJECT (queue, "activating pull mode");
3459       init_ranges (queue);
3460       queue->srcresult = GST_FLOW_OK;
3461       queue->sinkresult = GST_FLOW_OK;
3462       queue->is_eos = FALSE;
3463       queue->unexpected = FALSE;
3464       queue->upstream_size = 0;
3465     } else {
3466       GST_DEBUG_OBJECT (queue, "no temp file, cannot activate pull mode");
3467       /* this is not allowed, we cannot operate in pull mode without a temp
3468        * file. */
3469       queue->srcresult = GST_FLOW_FLUSHING;
3470       queue->sinkresult = GST_FLOW_FLUSHING;
3471       result = FALSE;
3472     }
3473     GST_QUEUE2_MUTEX_UNLOCK (queue);
3474   } else {
3475     GST_QUEUE2_MUTEX_LOCK (queue);
3476     GST_DEBUG_OBJECT (queue, "deactivating pull mode");
3477     queue->srcresult = GST_FLOW_FLUSHING;
3478     queue->sinkresult = GST_FLOW_FLUSHING;
3479     /* this will unlock getrange */
3480     GST_QUEUE2_SIGNAL_ADD (queue);
3481     result = TRUE;
3482     GST_QUEUE2_MUTEX_UNLOCK (queue);
3483   }
3484
3485   return result;
3486 }
3487
3488 static gboolean
3489 gst_queue2_src_activate_mode (GstPad * pad, GstObject * parent, GstPadMode mode,
3490     gboolean active)
3491 {
3492   gboolean res;
3493
3494   switch (mode) {
3495     case GST_PAD_MODE_PULL:
3496       res = gst_queue2_src_activate_pull (pad, parent, active);
3497       break;
3498     case GST_PAD_MODE_PUSH:
3499       res = gst_queue2_src_activate_push (pad, parent, active);
3500       break;
3501     default:
3502       GST_LOG_OBJECT (pad, "unknown activation mode %d", mode);
3503       res = FALSE;
3504       break;
3505   }
3506   return res;
3507 }
3508
3509 static GstStateChangeReturn
3510 gst_queue2_change_state (GstElement * element, GstStateChange transition)
3511 {
3512   GstQueue2 *queue;
3513   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
3514
3515   queue = GST_QUEUE2 (element);
3516
3517   switch (transition) {
3518     case GST_STATE_CHANGE_NULL_TO_READY:
3519       break;
3520     case GST_STATE_CHANGE_READY_TO_PAUSED:
3521       GST_QUEUE2_MUTEX_LOCK (queue);
3522       if (!QUEUE_IS_USING_QUEUE (queue)) {
3523         if (QUEUE_IS_USING_TEMP_FILE (queue)) {
3524           if (!gst_queue2_open_temp_location_file (queue))
3525             ret = GST_STATE_CHANGE_FAILURE;
3526         } else {
3527           if (queue->ring_buffer) {
3528             g_free (queue->ring_buffer);
3529             queue->ring_buffer = NULL;
3530           }
3531           if (!(queue->ring_buffer = g_malloc (queue->ring_buffer_max_size)))
3532             ret = GST_STATE_CHANGE_FAILURE;
3533         }
3534         init_ranges (queue);
3535       }
3536       queue->segment_event_received = FALSE;
3537       queue->starting_segment = NULL;
3538       gst_event_replace (&queue->stream_start_event, NULL);
3539       GST_QUEUE2_MUTEX_UNLOCK (queue);
3540       break;
3541     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
3542       break;
3543     default:
3544       break;
3545   }
3546
3547   if (ret == GST_STATE_CHANGE_FAILURE)
3548     return ret;
3549
3550   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
3551
3552   if (ret == GST_STATE_CHANGE_FAILURE)
3553     return ret;
3554
3555   switch (transition) {
3556     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
3557       break;
3558     case GST_STATE_CHANGE_PAUSED_TO_READY:
3559       GST_QUEUE2_MUTEX_LOCK (queue);
3560       if (!QUEUE_IS_USING_QUEUE (queue)) {
3561         if (QUEUE_IS_USING_TEMP_FILE (queue)) {
3562           gst_queue2_close_temp_location_file (queue);
3563         } else if (queue->ring_buffer) {
3564           g_free (queue->ring_buffer);
3565           queue->ring_buffer = NULL;
3566         }
3567         clean_ranges (queue);
3568       }
3569       if (queue->starting_segment != NULL) {
3570         gst_event_unref (queue->starting_segment);
3571         queue->starting_segment = NULL;
3572       }
3573       gst_event_replace (&queue->stream_start_event, NULL);
3574       GST_QUEUE2_MUTEX_UNLOCK (queue);
3575       break;
3576     case GST_STATE_CHANGE_READY_TO_NULL:
3577       break;
3578     default:
3579       break;
3580   }
3581
3582   return ret;
3583 }
3584
3585 /* changing the capacity of the queue must wake up
3586  * the _chain function, it might have more room now
3587  * to store the buffer/event in the queue */
3588 #define QUEUE_CAPACITY_CHANGE(q) \
3589   GST_QUEUE2_SIGNAL_DEL (queue); \
3590   if (queue->use_buffering)      \
3591     update_buffering (queue);
3592
3593 /* Changing the minimum required fill level must
3594  * wake up the _loop function as it might now
3595  * be able to preceed.
3596  */
3597 #define QUEUE_THRESHOLD_CHANGE(q)\
3598   GST_QUEUE2_SIGNAL_ADD (queue);
3599
3600 static void
3601 gst_queue2_set_temp_template (GstQueue2 * queue, const gchar * template)
3602 {
3603   GstState state;
3604
3605   /* the element must be stopped in order to do this */
3606   GST_OBJECT_LOCK (queue);
3607   state = GST_STATE (queue);
3608   if (state != GST_STATE_READY && state != GST_STATE_NULL)
3609     goto wrong_state;
3610   GST_OBJECT_UNLOCK (queue);
3611
3612   /* set new location */
3613   g_free (queue->temp_template);
3614   queue->temp_template = g_strdup (template);
3615
3616   return;
3617
3618 /* ERROR */
3619 wrong_state:
3620   {
3621     GST_WARNING_OBJECT (queue, "setting temp-template property in wrong state");
3622     GST_OBJECT_UNLOCK (queue);
3623   }
3624 }
3625
3626 static void
3627 gst_queue2_set_property (GObject * object,
3628     guint prop_id, const GValue * value, GParamSpec * pspec)
3629 {
3630   GstQueue2 *queue = GST_QUEUE2 (object);
3631
3632   /* someone could change levels here, and since this
3633    * affects the get/put funcs, we need to lock for safety. */
3634   GST_QUEUE2_MUTEX_LOCK (queue);
3635
3636   switch (prop_id) {
3637     case PROP_MAX_SIZE_BYTES:
3638       queue->max_level.bytes = g_value_get_uint (value);
3639       QUEUE_CAPACITY_CHANGE (queue);
3640       break;
3641     case PROP_MAX_SIZE_BUFFERS:
3642       queue->max_level.buffers = g_value_get_uint (value);
3643       QUEUE_CAPACITY_CHANGE (queue);
3644       break;
3645     case PROP_MAX_SIZE_TIME:
3646       queue->max_level.time = g_value_get_uint64 (value);
3647       /* set rate_time to the same value. We use an extra field in the level
3648        * structure so that we can easily access and compare it */
3649       queue->max_level.rate_time = queue->max_level.time;
3650       QUEUE_CAPACITY_CHANGE (queue);
3651       break;
3652     case PROP_USE_BUFFERING:
3653       queue->use_buffering = g_value_get_boolean (value);
3654       if (!queue->use_buffering && queue->is_buffering) {
3655         GST_DEBUG_OBJECT (queue, "Disabled buffering while buffering, "
3656             "posting 100%% message");
3657         SET_PERCENT (queue, 100);
3658         queue->is_buffering = FALSE;
3659       }
3660
3661       if (queue->use_buffering) {
3662         queue->is_buffering = TRUE;
3663         update_buffering (queue);
3664       }
3665       break;
3666     case PROP_USE_TAGS_BITRATE:
3667       queue->use_tags_bitrate = g_value_get_boolean (value);
3668       break;
3669     case PROP_USE_RATE_ESTIMATE:
3670       queue->use_rate_estimate = g_value_get_boolean (value);
3671       break;
3672     case PROP_LOW_PERCENT:
3673       queue->low_percent = g_value_get_int (value);
3674       break;
3675     case PROP_HIGH_PERCENT:
3676       queue->high_percent = g_value_get_int (value);
3677       break;
3678     case PROP_TEMP_TEMPLATE:
3679       gst_queue2_set_temp_template (queue, g_value_get_string (value));
3680       break;
3681     case PROP_TEMP_REMOVE:
3682       queue->temp_remove = g_value_get_boolean (value);
3683       break;
3684     case PROP_RING_BUFFER_MAX_SIZE:
3685       queue->ring_buffer_max_size = g_value_get_uint64 (value);
3686       break;
3687     default:
3688       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3689       break;
3690   }
3691
3692   GST_QUEUE2_MUTEX_UNLOCK (queue);
3693   gst_queue2_post_buffering (queue);
3694 }
3695
3696 static void
3697 gst_queue2_get_property (GObject * object,
3698     guint prop_id, GValue * value, GParamSpec * pspec)
3699 {
3700   GstQueue2 *queue = GST_QUEUE2 (object);
3701
3702   GST_QUEUE2_MUTEX_LOCK (queue);
3703
3704   switch (prop_id) {
3705     case PROP_CUR_LEVEL_BYTES:
3706       g_value_set_uint (value, queue->cur_level.bytes);
3707       break;
3708     case PROP_CUR_LEVEL_BUFFERS:
3709       g_value_set_uint (value, queue->cur_level.buffers);
3710       break;
3711     case PROP_CUR_LEVEL_TIME:
3712       g_value_set_uint64 (value, queue->cur_level.time);
3713       break;
3714     case PROP_MAX_SIZE_BYTES:
3715       g_value_set_uint (value, queue->max_level.bytes);
3716       break;
3717     case PROP_MAX_SIZE_BUFFERS:
3718       g_value_set_uint (value, queue->max_level.buffers);
3719       break;
3720     case PROP_MAX_SIZE_TIME:
3721       g_value_set_uint64 (value, queue->max_level.time);
3722       break;
3723     case PROP_USE_BUFFERING:
3724       g_value_set_boolean (value, queue->use_buffering);
3725       break;
3726     case PROP_USE_TAGS_BITRATE:
3727       g_value_set_boolean (value, queue->use_tags_bitrate);
3728       break;
3729     case PROP_USE_RATE_ESTIMATE:
3730       g_value_set_boolean (value, queue->use_rate_estimate);
3731       break;
3732     case PROP_LOW_PERCENT:
3733       g_value_set_int (value, queue->low_percent);
3734       break;
3735     case PROP_HIGH_PERCENT:
3736       g_value_set_int (value, queue->high_percent);
3737       break;
3738     case PROP_TEMP_TEMPLATE:
3739       g_value_set_string (value, queue->temp_template);
3740       break;
3741     case PROP_TEMP_LOCATION:
3742       g_value_set_string (value, queue->temp_location);
3743       break;
3744     case PROP_TEMP_REMOVE:
3745       g_value_set_boolean (value, queue->temp_remove);
3746       break;
3747     case PROP_RING_BUFFER_MAX_SIZE:
3748       g_value_set_uint64 (value, queue->ring_buffer_max_size);
3749       break;
3750     case PROP_AVG_IN_RATE:
3751     {
3752       gdouble in_rate = queue->byte_in_rate;
3753
3754       /* During the first RATE_INTERVAL, byte_in_rate will not have been
3755        * calculated, so calculate it here. */
3756       if (in_rate == 0.0 && queue->bytes_in
3757           && queue->last_update_in_rates_elapsed > 0.0)
3758         in_rate = queue->bytes_in / queue->last_update_in_rates_elapsed;
3759
3760       g_value_set_int64 (value, (gint64) in_rate);
3761       break;
3762     }
3763     default:
3764       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3765       break;
3766   }
3767
3768   GST_QUEUE2_MUTEX_UNLOCK (queue);
3769 }