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