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  *
   7  * gstqueue2.c:
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  22  * Boston, MA 02111-1307, USA.
  23  */
  24
  25 /**
  26  * SECTION:element-queue2
  27  *
  28  * Data is queued until one of the limits specified by the
  29  * #GstQueue2:max-size-buffers, #GstQueue2:max-size-bytes and/or
  30  * #GstQueue2:max-size-time properties has been reached. Any attempt to push
  31  * more buffers into the queue will block the pushing thread until more space
  32  * becomes available.
  33  *
  34  * The queue will create a new thread on the source pad to decouple the
  35  * processing on sink and source pad.
  36  *
  37  * You can query how many buffers are queued by reading the
  38  * #GstQueue2:current-level-buffers property.
  39  *
  40  * The default queue size limits are 100 buffers, 2MB of data, or
  41  * two seconds worth of data, whichever is reached first.
  42  *
  43  * If you set temp-tmpl to a value such as /tmp/gstreamer-XXXXXX, the element
  44  * will allocate a random free filename and buffer data in the file.
  45  * By using this, it will buffer the entire stream data on the file independently
  46  * of the queue size limits, they will only be used for buffering statistics.
  47  *
  48  * Since 0.10.24, setting the temp-location property with a filename is deprecated
  49  * because it's impossible to securely open a temporary file in this way. The
  50  * property will still be used to notify the application of the allocated
  51  * filename, though.
  52  *
  53  * Last reviewed on 2009-07-10 (0.10.24)
  54  */
  55
  56 #ifdef HAVE_CONFIG_H
  57 #include "config.h"
  58 #endif
  59
  60 #include "gstqueue2.h"
  61
  62 #include <glib/gstdio.h>
  63
  64 #include "gst/gst-i18n-lib.h"
  65
  66 #ifdef G_OS_WIN32
  67 #include <io.h>                 /* lseek, open, close, read */
  68 #undef lseek
  69 #define lseek _lseeki64
  70 #undef off_t
  71 #define off_t guint64
  72 #else
  73 #include <unistd.h>
  74 #endif
  75
  76 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  77     GST_PAD_SINK,
  78     GST_PAD_ALWAYS,
  79     GST_STATIC_CAPS_ANY);
  80
  81 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  82     GST_PAD_SRC,
  83     GST_PAD_ALWAYS,
  84     GST_STATIC_CAPS_ANY);
  85
  86 GST_DEBUG_CATEGORY_STATIC (queue_debug);
  87 #define GST_CAT_DEFAULT (queue_debug)
  88 GST_DEBUG_CATEGORY_STATIC (queue_dataflow);
  89
  90 enum
  91 {
  92   LAST_SIGNAL
  93 };
  94
  95 /* default property values */
  96 #define DEFAULT_MAX_SIZE_BUFFERS   100  /* 100 buffers */
  97 #define DEFAULT_MAX_SIZE_BYTES     (2 * 1024 * 1024)    /* 2 MB */
  98 #define DEFAULT_MAX_SIZE_TIME      2 * GST_SECOND       /* 2 seconds */
  99 #define DEFAULT_USE_BUFFERING      FALSE
 100 #define DEFAULT_USE_RATE_ESTIMATE  TRUE
 101 #define DEFAULT_LOW_PERCENT        10
 102 #define DEFAULT_HIGH_PERCENT       99
 103 #define DEFAULT_TEMP_REMOVE        TRUE
 104
 105 /* other defines */
 106 #define DEFAULT_BUFFER_SIZE 4096
 107 #define QUEUE_IS_USING_TEMP_FILE(queue) ((queue)->temp_location_set || (queue)->temp_template != NULL)
 108
 109 enum
 110 {
 111   PROP_0,
 112   PROP_CUR_LEVEL_BUFFERS,
 113   PROP_CUR_LEVEL_BYTES,
 114   PROP_CUR_LEVEL_TIME,
 115   PROP_MAX_SIZE_BUFFERS,
 116   PROP_MAX_SIZE_BYTES,
 117   PROP_MAX_SIZE_TIME,
 118   PROP_USE_BUFFERING,
 119   PROP_USE_RATE_ESTIMATE,
 120   PROP_LOW_PERCENT,
 121   PROP_HIGH_PERCENT,
 122   PROP_TEMP_TEMPLATE,
 123   PROP_TEMP_LOCATION,
 124   PROP_TEMP_REMOVE,
 125   PROP_LAST
 126 };
 127
 128 #define GST_QUEUE2_CLEAR_LEVEL(l) G_STMT_START {         \
 129   l.buffers = 0;                                        \
 130   l.bytes = 0;                                          \
 131   l.time = 0;                                           \
 132   l.rate_time = 0;                                      \
 133 } G_STMT_END
 134
 135 #define STATUS(queue, pad, msg) \
 136   GST_CAT_LOG_OBJECT (queue_dataflow, queue, \
 137                       "(%s:%s) " msg ": %u of %u buffers, %u of %u " \
 138                       "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \
 139                       " ns, %"G_GUINT64_FORMAT" items", \
 140                       GST_DEBUG_PAD_NAME (pad), \
 141                       queue->cur_level.buffers, \
 142                       queue->max_level.buffers, \
 143                       queue->cur_level.bytes, \
 144                       queue->max_level.bytes, \
 145                       queue->cur_level.time, \
 146                       queue->max_level.time, \
 147                       (guint64) (QUEUE_IS_USING_TEMP_FILE(queue) ? \
 148                         queue->current->writing_pos - queue->current->max_reading_pos : \
 149                         queue->queue->length))
 150
 151 #define GST_QUEUE2_MUTEX_LOCK(q) G_STMT_START {                          \
 152   g_mutex_lock (q->qlock);                                              \
 153 } G_STMT_END
 154
 155 #define GST_QUEUE2_MUTEX_LOCK_CHECK(q,label) G_STMT_START {              \
 156   GST_QUEUE2_MUTEX_LOCK (q);                                             \
 157   if (q->srcresult != GST_FLOW_OK)                                      \
 158     goto label;                                                         \
 159 } G_STMT_END
 160
 161 #define GST_QUEUE2_MUTEX_UNLOCK(q) G_STMT_START {                        \
 162   g_mutex_unlock (q->qlock);                                            \
 163 } G_STMT_END
 164
 165 #define GST_QUEUE2_WAIT_DEL_CHECK(q, label) G_STMT_START {               \
 166   STATUS (queue, q->sinkpad, "wait for DEL");                           \
 167   q->waiting_del = TRUE;                                                \
 168   g_cond_wait (q->item_del, queue->qlock);                              \
 169   q->waiting_del = FALSE;                                               \
 170   if (q->srcresult != GST_FLOW_OK) {                                    \
 171     STATUS (queue, q->srcpad, "received DEL wakeup");                   \
 172     goto label;                                                         \
 173   }                                                                     \
 174   STATUS (queue, q->sinkpad, "received DEL");                           \
 175 } G_STMT_END
 176
 177 #define GST_QUEUE2_WAIT_ADD_CHECK(q, label) G_STMT_START {               \
 178   STATUS (queue, q->srcpad, "wait for ADD");                            \
 179   q->waiting_add = TRUE;                                                \
 180   g_cond_wait (q->item_add, q->qlock);                                  \
 181   q->waiting_add = FALSE;                                               \
 182   if (q->srcresult != GST_FLOW_OK) {                                    \
 183     STATUS (queue, q->srcpad, "received ADD wakeup");                   \
 184     goto label;                                                         \
 185   }                                                                     \
 186   STATUS (queue, q->srcpad, "received ADD");                            \
 187 } G_STMT_END
 188
 189 #define GST_QUEUE2_SIGNAL_DEL(q) G_STMT_START {                          \
 190   if (q->waiting_del) {                                                 \
 191     STATUS (q, q->srcpad, "signal DEL");                                \
 192     g_cond_signal (q->item_del);                                        \
 193   }                                                                     \
 194 } G_STMT_END
 195
 196 #define GST_QUEUE2_SIGNAL_ADD(q) G_STMT_START {                          \
 197   if (q->waiting_add) {                                                 \
 198     STATUS (q, q->sinkpad, "signal ADD");                               \
 199     g_cond_signal (q->item_add);                                        \
 200   }                                                                     \
 201 } G_STMT_END
 202
 203 #define _do_init(bla) \
 204     GST_DEBUG_CATEGORY_INIT (queue_debug, "queue2", 0, "queue element"); \
 205     GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue2_dataflow", 0, \
 206         "dataflow inside the queue element");
 207
 208 GST_BOILERPLATE_FULL (GstQueue2, gst_queue2, GstElement, GST_TYPE_ELEMENT,
 209     _do_init);
 210
 211 static void gst_queue2_finalize (GObject * object);
 212
 213 static void gst_queue2_set_property (GObject * object,
 214     guint prop_id, const GValue * value, GParamSpec * pspec);
 215 static void gst_queue2_get_property (GObject * object,
 216     guint prop_id, GValue * value, GParamSpec * pspec);
 217
 218 static GstFlowReturn gst_queue2_chain (GstPad * pad, GstBuffer * buffer);
 219 static GstFlowReturn gst_queue2_bufferalloc (GstPad * pad, guint64 offset,
 220     guint size, GstCaps * caps, GstBuffer ** buf);
 221 static GstFlowReturn gst_queue2_push_one (GstQueue2 * queue);
 222 static void gst_queue2_loop (GstPad * pad);
 223
 224 static gboolean gst_queue2_handle_sink_event (GstPad * pad, GstEvent * event);
 225
 226 static gboolean gst_queue2_handle_src_event (GstPad * pad, GstEvent * event);
 227 static gboolean gst_queue2_handle_src_query (GstPad * pad, GstQuery * query);
 228
 229 static GstCaps *gst_queue2_getcaps (GstPad * pad);
 230 static gboolean gst_queue2_acceptcaps (GstPad * pad, GstCaps * caps);
 231
 232 static GstFlowReturn gst_queue2_get_range (GstPad * pad, guint64 offset,
 233     guint length, GstBuffer ** buffer);
 234 static gboolean gst_queue2_src_checkgetrange_function (GstPad * pad);
 235
 236 static gboolean gst_queue2_src_activate_pull (GstPad * pad, gboolean active);
 237 static gboolean gst_queue2_src_activate_push (GstPad * pad, gboolean active);
 238 static gboolean gst_queue2_sink_activate_push (GstPad * pad, gboolean active);
 239 static GstStateChangeReturn gst_queue2_change_state (GstElement * element,
 240     GstStateChange transition);
 241
 242 static gboolean gst_queue2_is_empty (GstQueue2 * queue);
 243 static gboolean gst_queue2_is_filled (GstQueue2 * queue);
 244
 245 /* static guint gst_queue2_signals[LAST_SIGNAL] = { 0 }; */
 246
 247 static void
 248 gst_queue2_base_init (gpointer g_class)
 249 {
 250   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 251
 252   gst_element_class_add_pad_template (gstelement_class,
 253       gst_static_pad_template_get (&srctemplate));
 254   gst_element_class_add_pad_template (gstelement_class,
 255       gst_static_pad_template_get (&sinktemplate));
 256
 257   gst_element_class_set_details_simple (gstelement_class, "Queue 2",
 258       "Generic",
 259       "Simple data queue",
 260       "Erik Walthinsen <omega@cse.ogi.edu>, "
 261       "Wim Taymans <wim.taymans@gmail.com>");
 262 }
 263
 264 static void
 265 gst_queue2_class_init (GstQueue2Class * klass)
 266 {
 267   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 268   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 269
 270   parent_class = g_type_class_peek_parent (klass);
 271
 272   gobject_class->set_property = gst_queue2_set_property;
 273   gobject_class->get_property = gst_queue2_get_property;
 274
 275   /* properties */
 276   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,
 277       g_param_spec_uint ("current-level-bytes", "Current level (kB)",
 278           "Current amount of data in the queue (bytes)",
 279           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 280   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BUFFERS,
 281       g_param_spec_uint ("current-level-buffers", "Current level (buffers)",
 282           "Current number of buffers in the queue",
 283           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 284   g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,
 285       g_param_spec_uint64 ("current-level-time", "Current level (ns)",
 286           "Current amount of data in the queue (in ns)",
 287           0, G_MAXUINT64, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 288
 289   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,
 290       g_param_spec_uint ("max-size-bytes", "Max. size (kB)",
 291           "Max. amount of data in the queue (bytes, 0=disable)",
 292           0, G_MAXUINT, DEFAULT_MAX_SIZE_BYTES,
 293           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 294   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BUFFERS,
 295       g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",
 296           "Max. number of buffers in the queue (0=disable)", 0, G_MAXUINT,
 297           DEFAULT_MAX_SIZE_BUFFERS,
 298           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 299   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 300       g_param_spec_uint64 ("max-size-time", "Max. size (ns)",
 301           "Max. amount of data in the queue (in ns, 0=disable)", 0, G_MAXUINT64,
 302           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 303
 304   g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,
 305       g_param_spec_boolean ("use-buffering", "Use buffering",
 306           "Emit GST_MESSAGE_BUFFERING based on low-/high-percent thresholds",
 307           DEFAULT_USE_BUFFERING, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 308   g_object_class_install_property (gobject_class, PROP_USE_RATE_ESTIMATE,
 309       g_param_spec_boolean ("use-rate-estimate", "Use Rate Estimate",
 310           "Estimate the bitrate of the stream to calculate time level",
 311           DEFAULT_USE_RATE_ESTIMATE,
 312           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 313   g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
 314       g_param_spec_int ("low-percent", "Low percent",
 315           "Low threshold for buffering to start", 0, 100, DEFAULT_LOW_PERCENT,
 316           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 317   g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
 318       g_param_spec_int ("high-percent", "High percent",
 319           "High threshold for buffering to finish", 0, 100,
 320           DEFAULT_HIGH_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 321
 322   g_object_class_install_property (gobject_class, PROP_TEMP_TEMPLATE,
 323       g_param_spec_string ("temp-template", "Temporary File Template",
 324           "File template to store temporary files in, should contain directory "
 325           "and XXXXXX. (NULL == disabled)",
 326           NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 327
 328   g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,
 329       g_param_spec_string ("temp-location", "Temporary File Location",
 330           "Location to store temporary files in (Deprecated: Only read this "
 331           "property, use temp-template to configure the name template)",
 332           NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 333
 334   /**
 335    * GstQueue2:temp-remove
 336    *
 337    * When temp-template is set, remove the temporary file when going to READY.
 338    *
 339    * Since: 0.10.26
 340    */
 341   g_object_class_install_property (gobject_class, PROP_TEMP_REMOVE,
 342       g_param_spec_boolean ("temp-remove", "Remove the Temporary File",
 343           "Remove the temp-location after use",
 344           DEFAULT_TEMP_REMOVE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 345
 346   /* set several parent class virtual functions */
 347   gobject_class->finalize = gst_queue2_finalize;
 348
 349   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_queue2_change_state);
 350 }
 351
 352 static void
 353 gst_queue2_init (GstQueue2 * queue, GstQueue2Class * g_class)
 354 {
 355   queue->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 356
 357   gst_pad_set_chain_function (queue->sinkpad,
 358       GST_DEBUG_FUNCPTR (gst_queue2_chain));
 359   gst_pad_set_activatepush_function (queue->sinkpad,
 360       GST_DEBUG_FUNCPTR (gst_queue2_sink_activate_push));
 361   gst_pad_set_event_function (queue->sinkpad,
 362       GST_DEBUG_FUNCPTR (gst_queue2_handle_sink_event));
 363   gst_pad_set_getcaps_function (queue->sinkpad,
 364       GST_DEBUG_FUNCPTR (gst_queue2_getcaps));
 365   gst_pad_set_acceptcaps_function (queue->sinkpad,
 366       GST_DEBUG_FUNCPTR (gst_queue2_acceptcaps));
 367   gst_pad_set_bufferalloc_function (queue->sinkpad,
 368       GST_DEBUG_FUNCPTR (gst_queue2_bufferalloc));
 369   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 370
 371   queue->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
 372
 373   gst_pad_set_activatepull_function (queue->srcpad,
 374       GST_DEBUG_FUNCPTR (gst_queue2_src_activate_pull));
 375   gst_pad_set_activatepush_function (queue->srcpad,
 376       GST_DEBUG_FUNCPTR (gst_queue2_src_activate_push));
 377   gst_pad_set_getrange_function (queue->srcpad,
 378       GST_DEBUG_FUNCPTR (gst_queue2_get_range));
 379   gst_pad_set_checkgetrange_function (queue->srcpad,
 380       GST_DEBUG_FUNCPTR (gst_queue2_src_checkgetrange_function));
 381   gst_pad_set_getcaps_function (queue->srcpad,
 382       GST_DEBUG_FUNCPTR (gst_queue2_getcaps));
 383   gst_pad_set_acceptcaps_function (queue->srcpad,
 384       GST_DEBUG_FUNCPTR (gst_queue2_acceptcaps));
 385   gst_pad_set_event_function (queue->srcpad,
 386       GST_DEBUG_FUNCPTR (gst_queue2_handle_src_event));
 387   gst_pad_set_query_function (queue->srcpad,
 388       GST_DEBUG_FUNCPTR (gst_queue2_handle_src_query));
 389   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 390
 391   /* levels */
 392   GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
 393   queue->max_level.buffers = DEFAULT_MAX_SIZE_BUFFERS;
 394   queue->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;
 395   queue->max_level.time = DEFAULT_MAX_SIZE_TIME;
 396   queue->max_level.rate_time = DEFAULT_MAX_SIZE_TIME;
 397   queue->use_buffering = DEFAULT_USE_BUFFERING;
 398   queue->use_rate_estimate = DEFAULT_USE_RATE_ESTIMATE;
 399   queue->low_percent = DEFAULT_LOW_PERCENT;
 400   queue->high_percent = DEFAULT_HIGH_PERCENT;
 401
 402   gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
 403   gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
 404
 405   queue->srcresult = GST_FLOW_WRONG_STATE;
 406   queue->is_eos = FALSE;
 407   queue->in_timer = g_timer_new ();
 408   queue->out_timer = g_timer_new ();
 409
 410   queue->qlock = g_mutex_new ();
 411   queue->waiting_add = FALSE;
 412   queue->item_add = g_cond_new ();
 413   queue->waiting_del = FALSE;
 414   queue->item_del = g_cond_new ();
 415   queue->queue = g_queue_new ();
 416
 417   /* tempfile related */
 418   queue->temp_template = NULL;
 419   queue->temp_location = NULL;
 420   queue->temp_location_set = FALSE;
 421   queue->temp_remove = DEFAULT_TEMP_REMOVE;
 422
 423   GST_DEBUG_OBJECT (queue,
 424       "initialized queue's not_empty & not_full conditions");
 425 }
 426
 427 /* called only once, as opposed to dispose */
 428 static void
 429 gst_queue2_finalize (GObject * object)
 430 {
 431   GstQueue2 *queue = GST_QUEUE2 (object);
 432
 433   GST_DEBUG_OBJECT (queue, "finalizing queue");
 434
 435   while (!g_queue_is_empty (queue->queue)) {
 436     GstMiniObject *data = g_queue_pop_head (queue->queue);
 437
 438     gst_mini_object_unref (data);
 439   }
 440
 441   g_queue_free (queue->queue);
 442   g_mutex_free (queue->qlock);
 443   g_cond_free (queue->item_add);
 444   g_cond_free (queue->item_del);
 445   g_timer_destroy (queue->in_timer);
 446   g_timer_destroy (queue->out_timer);
 447
 448   /* temp_file path cleanup  */
 449   g_free (queue->temp_template);
 450   g_free (queue->temp_location);
 451
 452   G_OBJECT_CLASS (parent_class)->finalize (object);
 453 }
 454
 455 static void
 456 debug_ranges (GstQueue2 * queue)
 457 {
 458   GstQueue2Range *walk;
 459
 460   for (walk = queue->ranges; walk; walk = walk->next) {
 461     GST_DEBUG_OBJECT (queue, "range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT,
 462         walk->offset, walk->writing_pos);
 463   }
 464 }
 465
 466 /* clear all the downloaded ranges */
 467 static void
 468 clean_ranges (GstQueue2 * queue)
 469 {
 470   GST_DEBUG_OBJECT (queue, "clean queue ranges");
 471
 472   g_slice_free_chain (GstQueue2Range, queue->ranges, next);
 473   queue->ranges = NULL;
 474   queue->current = NULL;
 475 }
 476
 477 /* find a range that contains @offset or NULL when nothing does */
 478 static GstQueue2Range *
 479 find_range (GstQueue2 * queue, guint64 offset, guint64 length)
 480 {
 481   GstQueue2Range *range = NULL;
 482   GstQueue2Range *walk;
 483
 484   /* first do a quick check for the current range */
 485   for (walk = queue->ranges; walk; walk = walk->next) {
 486     if (offset >= walk->offset && offset <= walk->writing_pos) {
 487       /* we can reuse an existing range */
 488       range = walk;
 489       break;
 490     }
 491   }
 492   return range;
 493 }
 494
 495 /* make a new range for @offset or reuse an existing range */
 496 static GstQueue2Range *
 497 add_range (GstQueue2 * queue, guint64 offset)
 498 {
 499   GstQueue2Range *range, *prev, *next;
 500
 501   GST_DEBUG_OBJECT (queue, "find range for %" G_GUINT64_FORMAT, offset);
 502
 503   if ((range = find_range (queue, offset, 0))) {
 504     GST_DEBUG_OBJECT (queue,
 505         "reusing range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, range->offset,
 506         range->writing_pos);
 507     range->writing_pos = offset;
 508   } else {
 509     GST_DEBUG_OBJECT (queue,
 510         "new range %" G_GUINT64_FORMAT "-%" G_GUINT64_FORMAT, offset, offset);
 511
 512     range = g_slice_new0 (GstQueue2Range);
 513     range->offset = offset;
 514     range->writing_pos = offset;
 515     range->reading_pos = offset;
 516     range->max_reading_pos = offset;
 517
 518     /* insert sorted */
 519     prev = NULL;
 520     next = queue->ranges;
 521     while (next) {
 522       if (next->offset > offset) {
 523         /* insert before next */
 524         GST_DEBUG_OBJECT (queue,
 525             "insert before range %p, offset %" G_GUINT64_FORMAT, next,
 526             next->offset);
 527         break;
 528       }
 529       /* try next */
 530       prev = next;
 531       next = next->next;
 532     }
 533     range->next = next;
 534     if (prev)
 535       prev->next = range;
 536     else
 537       queue->ranges = range;
 538   }
 539   debug_ranges (queue);
 540
 541   return range;
 542 }
 543
 544
 545 /* clear and init the download ranges for offset 0 */
 546 static void
 547 init_ranges (GstQueue2 * queue)
 548 {
 549   GST_DEBUG_OBJECT (queue, "init queue ranges");
 550
 551   /* get rid of all the current ranges */
 552   clean_ranges (queue);
 553   /* make a range for offset 0 */
 554   queue->current = add_range (queue, 0);
 555 }
 556
 557 static gboolean
 558 gst_queue2_acceptcaps (GstPad * pad, GstCaps * caps)
 559 {
 560   GstQueue2 *queue;
 561   GstPad *otherpad;
 562   gboolean result;
 563
 564   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
 565
 566   otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
 567   result = gst_pad_peer_accept_caps (otherpad, caps);
 568
 569   return result;
 570 }
 571
 572 static GstCaps *
 573 gst_queue2_getcaps (GstPad * pad)
 574 {
 575   GstQueue2 *queue;
 576   GstPad *otherpad;
 577   GstCaps *result;
 578
 579   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
 580
 581   otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);
 582   result = gst_pad_peer_get_caps (otherpad);
 583   if (result == NULL)
 584     result = gst_caps_new_any ();
 585
 586   return result;
 587 }
 588
 589 static GstFlowReturn
 590 gst_queue2_bufferalloc (GstPad * pad, guint64 offset, guint size,
 591     GstCaps * caps, GstBuffer ** buf)
 592 {
 593   GstQueue2 *queue;
 594   GstFlowReturn result;
 595
 596   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
 597
 598   /* Forward to src pad, without setting caps on the src pad */
 599   result = gst_pad_alloc_buffer (queue->srcpad, offset, size, caps, buf);
 600
 601   return result;
 602 }
 603
 604 /* calculate the diff between running time on the sink and src of the queue.
 605  * This is the total amount of time in the queue. */
 606 static void
 607 update_time_level (GstQueue2 * queue)
 608 {
 609   gint64 sink_time, src_time;
 610
 611   sink_time =
 612       gst_segment_to_running_time (&queue->sink_segment, GST_FORMAT_TIME,
 613       queue->sink_segment.last_stop);
 614
 615   src_time = gst_segment_to_running_time (&queue->src_segment, GST_FORMAT_TIME,
 616       queue->src_segment.last_stop);
 617
 618   GST_DEBUG_OBJECT (queue, "sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT,
 619       GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
 620
 621   if (sink_time >= src_time)
 622     queue->cur_level.time = sink_time - src_time;
 623   else
 624     queue->cur_level.time = 0;
 625 }
 626
 627 /* take a NEWSEGMENT event and apply the values to segment, updating the time
 628  * level of queue. */
 629 static void
 630 apply_segment (GstQueue2 * queue, GstEvent * event, GstSegment * segment)
 631 {
 632   gboolean update;
 633   GstFormat format;
 634   gdouble rate, arate;
 635   gint64 start, stop, time;
 636
 637   gst_event_parse_new_segment_full (event, &update, &rate, &arate,
 638       &format, &start, &stop, &time);
 639
 640   GST_DEBUG_OBJECT (queue,
 641       "received NEWSEGMENT update %d, rate %lf, applied rate %lf, "
 642       "format %d, "
 643       "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
 644       G_GINT64_FORMAT, update, rate, arate, format, start, stop, time);
 645
 646   if (format == GST_FORMAT_BYTES) {
 647   }
 648
 649   /* now configure the values, we use these to track timestamps on the
 650    * sinkpad. */
 651   if (format != GST_FORMAT_TIME) {
 652     /* non-time format, pretent the current time segment is closed with a
 653      * 0 start and unknown stop time. */
 654     update = FALSE;
 655     format = GST_FORMAT_TIME;
 656     start = 0;
 657     stop = -1;
 658     time = 0;
 659   }
 660   gst_segment_set_newsegment_full (segment, update,
 661       rate, arate, format, start, stop, time);
 662
 663   GST_DEBUG_OBJECT (queue,
 664       "configured NEWSEGMENT %" GST_SEGMENT_FORMAT, segment);
 665
 666   /* segment can update the time level of the queue */
 667   update_time_level (queue);
 668 }
 669
 670 /* take a buffer and update segment, updating the time level of the queue. */
 671 static void
 672 apply_buffer (GstQueue2 * queue, GstBuffer * buffer, GstSegment * segment)
 673 {
 674   GstClockTime duration, timestamp;
 675
 676   timestamp = GST_BUFFER_TIMESTAMP (buffer);
 677   duration = GST_BUFFER_DURATION (buffer);
 678
 679   /* if no timestamp is set, assume it's continuous with the previous
 680    * time */
 681   if (timestamp == GST_CLOCK_TIME_NONE)
 682     timestamp = segment->last_stop;
 683
 684   /* add duration */
 685   if (duration != GST_CLOCK_TIME_NONE)
 686     timestamp += duration;
 687
 688   GST_DEBUG_OBJECT (queue, "last_stop updated to %" GST_TIME_FORMAT,
 689       GST_TIME_ARGS (timestamp));
 690
 691   gst_segment_set_last_stop (segment, GST_FORMAT_TIME, timestamp);
 692
 693   /* calc diff with other end */
 694   update_time_level (queue);
 695 }
 696
 697 static void
 698 update_buffering (GstQueue2 * queue)
 699 {
 700   gint64 percent;
 701   gboolean post = FALSE;
 702
 703   if (!queue->use_buffering || queue->high_percent <= 0)
 704     return;
 705
 706 #define GET_PERCENT(format) ((queue->max_level.format) > 0 ? \
 707                 (queue->cur_level.format) * 100 / (queue->max_level.format) : 0)
 708
 709   if (queue->is_eos) {
 710     /* on EOS we are always 100% full, we set the var here so that it we can
 711      * reuse the logic below to stop buffering */
 712     percent = 100;
 713   } else {
 714     /* figure out the percent we are filled, we take the max of all formats. */
 715     percent = GET_PERCENT (bytes);
 716     percent = MAX (percent, GET_PERCENT (time));
 717     percent = MAX (percent, GET_PERCENT (buffers));
 718
 719     /* also apply the rate estimate when we need to */
 720     if (queue->use_rate_estimate)
 721       percent = MAX (percent, GET_PERCENT (rate_time));
 722   }
 723
 724   if (queue->is_buffering) {
 725     post = TRUE;
 726     /* if we were buffering see if we reached the high watermark */
 727     if (percent >= queue->high_percent)
 728       queue->is_buffering = FALSE;
 729   } else {
 730     /* we were not buffering, check if we need to start buffering if we drop
 731      * below the low threshold */
 732     if (percent < queue->low_percent) {
 733       queue->is_buffering = TRUE;
 734       queue->buffering_iteration++;
 735       post = TRUE;
 736     }
 737   }
 738   if (post) {
 739     GstMessage *message;
 740     GstBufferingMode mode;
 741     gint64 buffering_left = -1;
 742
 743     /* scale to high percent so that it becomes the 100% mark */
 744     percent = percent * 100 / queue->high_percent;
 745     /* clip */
 746     if (percent > 100)
 747       percent = 100;
 748
 749     if (QUEUE_IS_USING_TEMP_FILE (queue)) {
 750       GstFormat fmt = GST_FORMAT_BYTES;
 751       gint64 duration;
 752
 753       mode = GST_BUFFERING_DOWNLOAD;
 754       if (queue->byte_in_rate > 0) {
 755         if (gst_pad_query_peer_duration (queue->sinkpad, &fmt, &duration))
 756           buffering_left =
 757               (gdouble) ((duration -
 758                   queue->current->writing_pos) * 1000) / queue->byte_in_rate;
 759       } else {
 760         buffering_left = G_MAXINT64;
 761       }
 762     } else {
 763       mode = GST_BUFFERING_STREAM;
 764     }
 765
 766     GST_DEBUG_OBJECT (queue, "buffering %d percent", (gint) percent);
 767     message = gst_message_new_buffering (GST_OBJECT_CAST (queue),
 768         (gint) percent);
 769     gst_message_set_buffering_stats (message, mode,
 770         queue->byte_in_rate, queue->byte_out_rate, buffering_left);
 771
 772     gst_element_post_message (GST_ELEMENT_CAST (queue), message);
 773
 774   } else {
 775     GST_DEBUG_OBJECT (queue, "filled %d percent", (gint) percent);
 776   }
 777
 778 #undef GET_PERCENT
 779 }
 780
 781 static void
 782 reset_rate_timer (GstQueue2 * queue)
 783 {
 784   queue->bytes_in = 0;
 785   queue->bytes_out = 0;
 786   queue->byte_in_rate = 0.0;
 787   queue->byte_out_rate = 0.0;
 788   queue->last_in_elapsed = 0.0;
 789   queue->last_out_elapsed = 0.0;
 790   queue->in_timer_started = FALSE;
 791   queue->out_timer_started = FALSE;
 792 }
 793
 794 /* the interval in seconds to recalculate the rate */
 795 #define RATE_INTERVAL    0.2
 796 /* Tuning for rate estimation. We use a large window for the input rate because
 797  * it should be stable when connected to a network. The output rate is less
 798  * stable (the elements preroll, queues behind a demuxer fill, ...) and should
 799  * therefore adapt more quickly. */
 800 #define AVG_IN(avg,val)  ((avg) * 15.0 + (val)) / 16.0
 801 #define AVG_OUT(avg,val) ((avg) * 3.0 + (val)) / 4.0
 802
 803 static void
 804 update_in_rates (GstQueue2 * queue)
 805 {
 806   gdouble elapsed, period;
 807   gdouble byte_in_rate;
 808
 809   if (!queue->in_timer_started) {
 810     queue->in_timer_started = TRUE;
 811     g_timer_start (queue->in_timer);
 812     return;
 813   }
 814
 815   elapsed = g_timer_elapsed (queue->in_timer, NULL);
 816
 817   /* recalc after each interval. */
 818   if (queue->last_in_elapsed + RATE_INTERVAL < elapsed) {
 819     period = elapsed - queue->last_in_elapsed;
 820
 821     GST_DEBUG_OBJECT (queue,
 822         "rates: period %f, in %" G_GUINT64_FORMAT, period, queue->bytes_in);
 823
 824     byte_in_rate = queue->bytes_in / period;
 825
 826     if (queue->byte_in_rate == 0.0)
 827       queue->byte_in_rate = byte_in_rate;
 828     else
 829       queue->byte_in_rate = AVG_IN (queue->byte_in_rate, byte_in_rate);
 830
 831     /* reset the values to calculate rate over the next interval */
 832     queue->last_in_elapsed = elapsed;
 833     queue->bytes_in = 0;
 834   }
 835
 836   if (queue->byte_in_rate > 0.0) {
 837     queue->cur_level.rate_time =
 838         queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
 839   }
 840   GST_DEBUG_OBJECT (queue, "rates: in %f, time %" GST_TIME_FORMAT,
 841       queue->byte_in_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
 842 }
 843
 844 static void
 845 update_out_rates (GstQueue2 * queue)
 846 {
 847   gdouble elapsed, period;
 848   gdouble byte_out_rate;
 849
 850   if (!queue->out_timer_started) {
 851     queue->out_timer_started = TRUE;
 852     g_timer_start (queue->out_timer);
 853     return;
 854   }
 855
 856   elapsed = g_timer_elapsed (queue->out_timer, NULL);
 857
 858   /* recalc after each interval. */
 859   if (queue->last_out_elapsed + RATE_INTERVAL < elapsed) {
 860     period = elapsed - queue->last_out_elapsed;
 861
 862     GST_DEBUG_OBJECT (queue,
 863         "rates: period %f, out %" G_GUINT64_FORMAT, period, queue->bytes_out);
 864
 865     byte_out_rate = queue->bytes_out / period;
 866
 867     if (queue->byte_out_rate == 0.0)
 868       queue->byte_out_rate = byte_out_rate;
 869     else
 870       queue->byte_out_rate = AVG_OUT (queue->byte_out_rate, byte_out_rate);
 871
 872     /* reset the values to calculate rate over the next interval */
 873     queue->last_out_elapsed = elapsed;
 874     queue->bytes_out = 0;
 875   }
 876   if (queue->byte_in_rate > 0.0) {
 877     queue->cur_level.rate_time =
 878         queue->cur_level.bytes / queue->byte_in_rate * GST_SECOND;
 879   }
 880   GST_DEBUG_OBJECT (queue, "rates: out %f, time %" GST_TIME_FORMAT,
 881       queue->byte_out_rate, GST_TIME_ARGS (queue->cur_level.rate_time));
 882 }
 883
 884 static void
 885 gst_queue2_write_buffer_to_file (GstQueue2 * queue, GstBuffer * buffer)
 886 {
 887   guint size;
 888   guint8 *data;
 889   int ret;
 890   guint64 writing_pos, max_reading_pos;
 891   GstQueue2Range *next;
 892
 893   writing_pos = queue->current->writing_pos;
 894   max_reading_pos = queue->current->max_reading_pos;
 895
 896 #ifdef HAVE_FSEEKO
 897   fseeko (queue->temp_file, (off_t) writing_pos, SEEK_SET);
 898 #elif defined (G_OS_UNIX) || defined (G_OS_WIN32)
 899   lseek (fileno (queue->temp_file), (off_t) writing_pos, SEEK_SET);
 900 #else
 901   fseek (queue->temp_file, writing_pos, SEEK_SET);
 902 #endif
 903
 904   data = GST_BUFFER_DATA (buffer);
 905   size = GST_BUFFER_SIZE (buffer);
 906
 907   ret = fwrite (data, 1, size, queue->temp_file);
 908   if (ret < size) {
 909     /* FIXME do something useful here */
 910     GST_ERROR_OBJECT (queue, "fwrite returned error");
 911   }
 912   writing_pos += size;
 913
 914   if (writing_pos > max_reading_pos)
 915     queue->cur_level.bytes = writing_pos - max_reading_pos;
 916   else
 917     queue->cur_level.bytes = 0;
 918
 919   /* try to merge with next range */
 920   while ((next = queue->current->next)) {
 921     GST_DEBUG_OBJECT (queue,
 922         "cheking %" G_GUINT64_FORMAT " < %" G_GUINT64_FORMAT, writing_pos,
 923         next->offset);
 924     if (writing_pos < next->offset)
 925       break;
 926
 927     GST_DEBUG_OBJECT (queue, "merging ranges %" G_GUINT64_FORMAT,
 928         next->writing_pos);
 929     /* we ran over the offset of the next group */
 930     queue->current->writing_pos = writing_pos = next->writing_pos;
 931
 932     /* remove the group */
 933     queue->current->next = next->next;
 934     g_slice_free (GstQueue2Range, next);
 935
 936     debug_ranges (queue);
 937   }
 938   queue->current->writing_pos = writing_pos;
 939 }
 940
 941 static gboolean
 942 perform_seek_to_offset (GstQueue2 * queue, guint64 offset)
 943 {
 944   GstEvent *event;
 945   gboolean res;
 946
 947   GST_DEBUG_OBJECT (queue, "Seeking to %" G_GUINT64_FORMAT, offset);
 948
 949   event =
 950       gst_event_new_seek (1.0, GST_FORMAT_BYTES,
 951       GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE, GST_SEEK_TYPE_SET, offset,
 952       GST_SEEK_TYPE_NONE, -1);
 953
 954   GST_QUEUE2_MUTEX_UNLOCK (queue);
 955   res = gst_pad_push_event (queue->sinkpad, event);
 956   GST_QUEUE2_MUTEX_LOCK (queue);
 957
 958   if (res) {
 959     queue->current = add_range (queue, offset);
 960   }
 961   return res;
 962 }
 963
 964 /* see if there is enough data in the file to read a full buffer */
 965 static gboolean
 966 gst_queue2_have_data (GstQueue2 * queue, guint64 offset, guint length)
 967 {
 968   GstQueue2Range *range;
 969
 970   GST_DEBUG_OBJECT (queue, "offset %" G_GUINT64_FORMAT ", len %u", offset,
 971       length);
 972
 973   if ((range = find_range (queue, offset, length))) {
 974     if (queue->current != range) {
 975       GST_DEBUG_OBJECT (queue, "switching ranges");
 976       perform_seek_to_offset (queue, range->writing_pos);
 977     }
 978     /* we have a range for offset */
 979     GST_DEBUG_OBJECT (queue,
 980         "we have a range %p, offset %" G_GUINT64_FORMAT ", writing_pos %"
 981         G_GUINT64_FORMAT, range, range->offset, range->writing_pos);
 982
 983     if (queue->is_eos)
 984       return TRUE;
 985
 986     if (offset + length < range->writing_pos)
 987       return TRUE;
 988
 989   } else {
 990     /* we don't have the range, see how far away we are */
 991     perform_seek_to_offset (queue, offset);
 992   }
 993
 994   return FALSE;
 995 }
 996
 997 static GstFlowReturn
 998 gst_queue2_create_read (GstQueue2 * queue, guint64 offset, guint length,
 999     GstBuffer ** buffer)
1000 {
1001   size_t res;
1002   GstBuffer *buf;
1003   guint64 reading_pos, max_reading_pos, writing_pos;
1004
1005   /* check if we have enough data at @offset. If there is not enough data, we
1006    * block and wait. */
1007   while (!gst_queue2_have_data (queue, offset, length)) {
1008     GST_QUEUE2_WAIT_ADD_CHECK (queue, out_flushing);
1009   }
1010
1011 #ifdef HAVE_FSEEKO
1012   if (fseeko (queue->temp_file, (off_t) offset, SEEK_SET) != 0)
1013     goto seek_failed;
1014 #elif defined (G_OS_UNIX) || defined (G_OS_WIN32)
1015   if (lseek (fileno (queue->temp_file), (off_t) offset,
1016           SEEK_SET) == (off_t) - 1)
1017     goto seek_failed;
1018 #else
1019   if (fseek (queue->temp_file, (long) offset, SEEK_SET) != 0)
1020     goto seek_failed;
1021 #endif
1022
1023   buf = gst_buffer_new_and_alloc (length);
1024
1025   /* this should not block */
1026   GST_LOG_OBJECT (queue, "Reading %d bytes", length);
1027   res = fread (GST_BUFFER_DATA (buf), 1, length, queue->temp_file);
1028   GST_LOG_OBJECT (queue, "read %" G_GSIZE_FORMAT " bytes", res);
1029
1030   if (G_UNLIKELY (res == 0)) {
1031     /* check for errors or EOF */
1032     if (ferror (queue->temp_file))
1033       goto could_not_read;
1034     if (feof (queue->temp_file) && length > 0)
1035       goto eos;
1036   }
1037
1038   length = res;
1039
1040   GST_BUFFER_SIZE (buf) = length;
1041   GST_BUFFER_OFFSET (buf) = offset;
1042   GST_BUFFER_OFFSET_END (buf) = offset + length;
1043
1044   *buffer = buf;
1045
1046   reading_pos = queue->current->reading_pos;
1047   writing_pos = queue->current->writing_pos;
1048   max_reading_pos = queue->current->max_reading_pos;
1049
1050   reading_pos = offset + length;
1051   max_reading_pos = MAX (max_reading_pos, reading_pos);
1052
1053   if (writing_pos > max_reading_pos)
1054     queue->cur_level.bytes = writing_pos - max_reading_pos;
1055   else
1056     queue->cur_level.bytes = 0;
1057
1058   queue->current->reading_pos = reading_pos;
1059   queue->current->max_reading_pos = max_reading_pos;
1060
1061   return GST_FLOW_OK;
1062
1063   /* ERRORS */
1064 out_flushing:
1065   {
1066     GST_DEBUG_OBJECT (queue, "we are flushing");
1067     return GST_FLOW_WRONG_STATE;
1068   }
1069 seek_failed:
1070   {
1071     GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
1072     return GST_FLOW_ERROR;
1073   }
1074 could_not_read:
1075   {
1076     GST_ELEMENT_ERROR (queue, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
1077     gst_buffer_unref (buf);
1078     return GST_FLOW_ERROR;
1079   }
1080 eos:
1081   {
1082     GST_DEBUG ("non-regular file hits EOS");
1083     gst_buffer_unref (buf);
1084     return GST_FLOW_UNEXPECTED;
1085   }
1086 }
1087
1088 /* should be called with QUEUE_LOCK */
1089 static GstMiniObject *
1090 gst_queue2_read_item_from_file (GstQueue2 * queue)
1091 {
1092   GstMiniObject *item;
1093
1094   if (queue->starting_segment != NULL) {
1095     item = GST_MINI_OBJECT_CAST (queue->starting_segment);
1096     queue->starting_segment = NULL;
1097   } else {
1098     GstFlowReturn ret;
1099     GstBuffer *buffer;
1100     guint64 reading_pos;
1101
1102     reading_pos = queue->current->reading_pos;
1103
1104     ret =
1105         gst_queue2_create_read (queue, reading_pos, DEFAULT_BUFFER_SIZE,
1106         &buffer);
1107     switch (ret) {
1108       case GST_FLOW_OK:
1109         item = GST_MINI_OBJECT_CAST (buffer);
1110         break;
1111       case GST_FLOW_UNEXPECTED:
1112         item = GST_MINI_OBJECT_CAST (gst_event_new_eos ());
1113         break;
1114       default:
1115         item = NULL;
1116         break;
1117     }
1118   }
1119   return item;
1120 }
1121
1122 static gboolean
1123 gst_queue2_open_temp_location_file (GstQueue2 * queue)
1124 {
1125   gint fd = -1;
1126   gchar *name = NULL;
1127
1128   if (queue->temp_file)
1129     goto already_opened;
1130
1131   GST_DEBUG_OBJECT (queue, "opening temp file %s", queue->temp_template);
1132
1133   /* we have two cases:
1134    * - temp_location was set to something !NULL (Deprecated). in this case we
1135    *   open the specified filename.
1136    * - temp_template was set, allocate a filename and open that filename
1137    */
1138   if (!queue->temp_location_set) {
1139     /* nothing to do */
1140     if (queue->temp_template == NULL)
1141       goto no_directory;
1142
1143     /* make copy of the template, we don't want to change this */
1144     name = g_strdup (queue->temp_template);
1145     fd = g_mkstemp (name);
1146     if (fd == -1)
1147       goto mkstemp_failed;
1148
1149     /* open the file for update/writing */
1150     queue->temp_file = fdopen (fd, "wb+");
1151     /* error creating file */
1152     if (queue->temp_file == NULL)
1153       goto open_failed;
1154
1155     g_free (queue->temp_location);
1156     queue->temp_location = name;
1157
1158     g_object_notify (G_OBJECT (queue), "temp-location");
1159   } else {
1160     /* open the file for update/writing, this is deprecated but we still need to
1161      * support it for API/ABI compatibility */
1162     queue->temp_file = g_fopen (queue->temp_location, "wb+");
1163     /* error creating file */
1164     if (queue->temp_file == NULL)
1165       goto open_failed;
1166   }
1167   GST_DEBUG_OBJECT (queue, "opened temp file %s", queue->temp_template);
1168
1169   init_ranges (queue);
1170
1171   return TRUE;
1172
1173   /* ERRORS */
1174 already_opened:
1175   {
1176     GST_DEBUG_OBJECT (queue, "temp file was already open");
1177     return TRUE;
1178   }
1179 no_directory:
1180   {
1181     GST_ELEMENT_ERROR (queue, RESOURCE, NOT_FOUND,
1182         (_("No Temp directory specified.")), (NULL));
1183     return FALSE;
1184   }
1185 mkstemp_failed:
1186   {
1187     GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1188         (_("Could not create temp file \"%s\"."), queue->temp_template),
1189         GST_ERROR_SYSTEM);
1190     g_free (name);
1191     return FALSE;
1192   }
1193 open_failed:
1194   {
1195     GST_ELEMENT_ERROR (queue, RESOURCE, OPEN_READ,
1196         (_("Could not open file \"%s\" for reading."), name), GST_ERROR_SYSTEM);
1197     g_free (name);
1198     if (fd != -1)
1199       close (fd);
1200     return FALSE;
1201   }
1202 }
1203
1204 static void
1205 gst_queue2_close_temp_location_file (GstQueue2 * queue)
1206 {
1207   /* nothing to do */
1208   if (queue->temp_file == NULL)
1209     return;
1210
1211   GST_DEBUG_OBJECT (queue, "closing temp file");
1212
1213   fflush (queue->temp_file);
1214   fclose (queue->temp_file);
1215
1216   if (queue->temp_remove)
1217     remove (queue->temp_location);
1218
1219   queue->temp_file = NULL;
1220   clean_ranges (queue);
1221 }
1222
1223 static void
1224 gst_queue2_flush_temp_file (GstQueue2 * queue)
1225 {
1226   if (queue->temp_file == NULL)
1227     return;
1228
1229   GST_DEBUG_OBJECT (queue, "flushing temp file");
1230
1231   queue->temp_file = g_freopen (queue->temp_location, "wb+", queue->temp_file);
1232
1233   init_ranges (queue);
1234 }
1235
1236 static void
1237 gst_queue2_locked_flush (GstQueue2 * queue)
1238 {
1239   if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1240     gst_queue2_flush_temp_file (queue);
1241   } else {
1242     while (!g_queue_is_empty (queue->queue)) {
1243       GstMiniObject *data = g_queue_pop_head (queue->queue);
1244
1245       /* Then lose another reference because we are supposed to destroy that
1246          data when flushing */
1247       gst_mini_object_unref (data);
1248     }
1249   }
1250   GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
1251   gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
1252   gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
1253   if (queue->starting_segment != NULL)
1254     gst_event_unref (queue->starting_segment);
1255   queue->starting_segment = NULL;
1256   queue->segment_event_received = FALSE;
1257
1258   /* we deleted a lot of something */
1259   GST_QUEUE2_SIGNAL_DEL (queue);
1260 }
1261
1262 /* enqueue an item an update the level stats */
1263 static void
1264 gst_queue2_locked_enqueue (GstQueue2 * queue, gpointer item)
1265 {
1266   if (GST_IS_BUFFER (item)) {
1267     GstBuffer *buffer;
1268     guint size;
1269
1270     buffer = GST_BUFFER_CAST (item);
1271     size = GST_BUFFER_SIZE (buffer);
1272
1273     /* add buffer to the statistics */
1274     queue->cur_level.buffers++;
1275     queue->cur_level.bytes += size;
1276     queue->bytes_in += size;
1277
1278     /* apply new buffer to segment stats */
1279     apply_buffer (queue, buffer, &queue->sink_segment);
1280     /* update the byterate stats */
1281     update_in_rates (queue);
1282
1283     if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1284       gst_queue2_write_buffer_to_file (queue, buffer);
1285     }
1286
1287   } else if (GST_IS_EVENT (item)) {
1288     GstEvent *event;
1289
1290     event = GST_EVENT_CAST (item);
1291
1292     switch (GST_EVENT_TYPE (event)) {
1293       case GST_EVENT_EOS:
1294         /* Zero the thresholds, this makes sure the queue is completely
1295          * filled and we can read all data from the queue. */
1296         GST_DEBUG_OBJECT (queue, "we have EOS");
1297         queue->is_eos = TRUE;
1298         break;
1299       case GST_EVENT_NEWSEGMENT:
1300         apply_segment (queue, event, &queue->sink_segment);
1301         /* This is our first new segment, we hold it
1302          * as we can't save it on the temp file */
1303         if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1304           if (queue->segment_event_received)
1305             goto unexpected_event;
1306
1307           queue->segment_event_received = TRUE;
1308           if (queue->starting_segment != NULL)
1309             gst_event_unref (queue->starting_segment);
1310           queue->starting_segment = event;
1311           item = NULL;
1312         }
1313         /* a new segment allows us to accept more buffers if we got UNEXPECTED
1314          * from downstream */
1315         queue->unexpected = FALSE;
1316         break;
1317       default:
1318         if (QUEUE_IS_USING_TEMP_FILE (queue))
1319           goto unexpected_event;
1320         break;
1321     }
1322   } else {
1323     g_warning ("Unexpected item %p added in queue %s (refcounting problem?)",
1324         item, GST_OBJECT_NAME (queue));
1325     /* we can't really unref since we don't know what it is */
1326     item = NULL;
1327   }
1328
1329   if (item) {
1330     /* update the buffering status */
1331     update_buffering (queue);
1332
1333     if (!QUEUE_IS_USING_TEMP_FILE (queue))
1334       g_queue_push_tail (queue->queue, item);
1335     else
1336       gst_mini_object_unref (GST_MINI_OBJECT_CAST (item));
1337
1338     GST_QUEUE2_SIGNAL_ADD (queue);
1339   }
1340
1341   return;
1342
1343   /* ERRORS */
1344 unexpected_event:
1345   {
1346     g_warning
1347         ("Unexpected event of kind %s can't be added in temp file of queue %s ",
1348         gst_event_type_get_name (GST_EVENT_TYPE (item)),
1349         GST_OBJECT_NAME (queue));
1350     gst_event_unref (GST_EVENT_CAST (item));
1351     return;
1352   }
1353 }
1354
1355 /* dequeue an item from the queue and update level stats */
1356 static GstMiniObject *
1357 gst_queue2_locked_dequeue (GstQueue2 * queue)
1358 {
1359   GstMiniObject *item;
1360
1361   if (QUEUE_IS_USING_TEMP_FILE (queue))
1362     item = gst_queue2_read_item_from_file (queue);
1363   else
1364     item = g_queue_pop_head (queue->queue);
1365
1366   if (item == NULL)
1367     goto no_item;
1368
1369   if (GST_IS_BUFFER (item)) {
1370     GstBuffer *buffer;
1371     guint size;
1372
1373     buffer = GST_BUFFER_CAST (item);
1374     size = GST_BUFFER_SIZE (buffer);
1375
1376     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1377         "retrieved buffer %p from queue", buffer);
1378
1379     queue->cur_level.buffers--;
1380     queue->cur_level.bytes -= size;
1381     queue->bytes_out += size;
1382     apply_buffer (queue, buffer, &queue->src_segment);
1383     /* update the byterate stats */
1384     update_out_rates (queue);
1385     /* update the buffering */
1386     update_buffering (queue);
1387
1388   } else if (GST_IS_EVENT (item)) {
1389     GstEvent *event = GST_EVENT_CAST (item);
1390
1391     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1392         "retrieved event %p from queue", event);
1393
1394     switch (GST_EVENT_TYPE (event)) {
1395       case GST_EVENT_EOS:
1396         /* queue is empty now that we dequeued the EOS */
1397         GST_QUEUE2_CLEAR_LEVEL (queue->cur_level);
1398         break;
1399       case GST_EVENT_NEWSEGMENT:
1400         apply_segment (queue, event, &queue->src_segment);
1401         break;
1402       default:
1403         break;
1404     }
1405   } else {
1406     g_warning
1407         ("Unexpected item %p dequeued from queue %s (refcounting problem?)",
1408         item, GST_OBJECT_NAME (queue));
1409     item = NULL;
1410   }
1411   GST_QUEUE2_SIGNAL_DEL (queue);
1412
1413   return item;
1414
1415   /* ERRORS */
1416 no_item:
1417   {
1418     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "the queue is empty");
1419     return NULL;
1420   }
1421 }
1422
1423 static gboolean
1424 gst_queue2_handle_sink_event (GstPad * pad, GstEvent * event)
1425 {
1426   GstQueue2 *queue;
1427
1428   queue = GST_QUEUE2 (GST_OBJECT_PARENT (pad));
1429
1430   switch (GST_EVENT_TYPE (event)) {
1431     case GST_EVENT_FLUSH_START:
1432     {
1433       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush start event");
1434       if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1435         /* forward event */
1436         gst_pad_push_event (queue->srcpad, event);
1437
1438         /* now unblock the chain function */
1439         GST_QUEUE2_MUTEX_LOCK (queue);
1440         queue->srcresult = GST_FLOW_WRONG_STATE;
1441         /* unblock the loop and chain functions */
1442         g_cond_signal (queue->item_add);
1443         g_cond_signal (queue->item_del);
1444         GST_QUEUE2_MUTEX_UNLOCK (queue);
1445
1446         /* make sure it pauses, this should happen since we sent
1447          * flush_start downstream. */
1448         gst_pad_pause_task (queue->srcpad);
1449         GST_CAT_LOG_OBJECT (queue_dataflow, queue, "loop stopped");
1450       }
1451       goto done;
1452     }
1453     case GST_EVENT_FLUSH_STOP:
1454     {
1455       GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received flush stop event");
1456
1457       if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1458         /* forward event */
1459         gst_pad_push_event (queue->srcpad, event);
1460
1461         GST_QUEUE2_MUTEX_LOCK (queue);
1462         gst_queue2_locked_flush (queue);
1463         queue->srcresult = GST_FLOW_OK;
1464         queue->is_eos = FALSE;
1465         queue->unexpected = FALSE;
1466         /* reset rate counters */
1467         reset_rate_timer (queue);
1468         gst_pad_start_task (queue->srcpad, (GstTaskFunction) gst_queue2_loop,
1469             queue->srcpad);
1470         GST_QUEUE2_MUTEX_UNLOCK (queue);
1471       } else {
1472         GST_QUEUE2_MUTEX_LOCK (queue);
1473         queue->segment_event_received = FALSE;
1474         queue->is_eos = FALSE;
1475         queue->unexpected = FALSE;
1476         GST_QUEUE2_MUTEX_UNLOCK (queue);
1477       }
1478       goto done;
1479     }
1480     default:
1481       if (GST_EVENT_IS_SERIALIZED (event)) {
1482         /* serialized events go in the queue */
1483         GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1484         /* refuse more events on EOS */
1485         if (queue->is_eos)
1486           goto out_eos;
1487         gst_queue2_locked_enqueue (queue, event);
1488         GST_QUEUE2_MUTEX_UNLOCK (queue);
1489       } else {
1490         /* non-serialized events are passed upstream. */
1491         gst_pad_push_event (queue->srcpad, event);
1492       }
1493       break;
1494   }
1495 done:
1496   return TRUE;
1497
1498   /* ERRORS */
1499 out_flushing:
1500   {
1501     GST_DEBUG_OBJECT (queue, "refusing event, we are flushing");
1502     GST_QUEUE2_MUTEX_UNLOCK (queue);
1503     gst_event_unref (event);
1504     return FALSE;
1505   }
1506 out_eos:
1507   {
1508     GST_DEBUG_OBJECT (queue, "refusing event, we are EOS");
1509     GST_QUEUE2_MUTEX_UNLOCK (queue);
1510     gst_event_unref (event);
1511     return FALSE;
1512   }
1513 }
1514
1515 static gboolean
1516 gst_queue2_is_empty (GstQueue2 * queue)
1517 {
1518   /* never empty on EOS */
1519   if (queue->is_eos)
1520     return FALSE;
1521
1522   if (QUEUE_IS_USING_TEMP_FILE (queue)) {
1523     return queue->current->writing_pos == queue->current->max_reading_pos;
1524   } else {
1525     if (queue->queue->length == 0)
1526       return TRUE;
1527   }
1528
1529   return FALSE;
1530 }
1531
1532 static gboolean
1533 gst_queue2_is_filled (GstQueue2 * queue)
1534 {
1535   gboolean res;
1536
1537   /* always filled on EOS */
1538   if (queue->is_eos)
1539     return TRUE;
1540
1541   /* if using file, we're never filled if we don't have EOS */
1542   if (QUEUE_IS_USING_TEMP_FILE (queue))
1543     return FALSE;
1544
1545   /* we are never filled when we have no buffers at all */
1546   if (queue->cur_level.buffers == 0)
1547     return FALSE;
1548
1549 #define CHECK_FILLED(format) ((queue->max_level.format) > 0 && \
1550                 (queue->cur_level.format) >= (queue->max_level.format))
1551
1552   /* we are filled if one of the current levels exceeds the max */
1553   res = CHECK_FILLED (buffers) || CHECK_FILLED (bytes) || CHECK_FILLED (time);
1554
1555   /* if we need to, use the rate estimate to check against the max time we are
1556    * allowed to queue */
1557   if (queue->use_rate_estimate)
1558     res |= CHECK_FILLED (rate_time);
1559
1560 #undef CHECK_FILLED
1561   return res;
1562 }
1563
1564 static GstFlowReturn
1565 gst_queue2_chain (GstPad * pad, GstBuffer * buffer)
1566 {
1567   GstQueue2 *queue;
1568
1569   queue = GST_QUEUE2 (GST_OBJECT_PARENT (pad));
1570
1571   GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1572       "received buffer %p of size %d, time %" GST_TIME_FORMAT ", duration %"
1573       GST_TIME_FORMAT, buffer, GST_BUFFER_SIZE (buffer),
1574       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
1575       GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)));
1576
1577   /* we have to lock the queue since we span threads */
1578   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1579   /* when we received EOS, we refuse more data */
1580   if (queue->is_eos)
1581     goto out_eos;
1582   /* when we received unexpected from downstream, refuse more buffers */
1583   if (queue->unexpected)
1584     goto out_unexpected;
1585
1586   /* We make space available if we're "full" according to whatever
1587    * the user defined as "full". */
1588   if (gst_queue2_is_filled (queue)) {
1589     gboolean started;
1590
1591     /* pause the timer while we wait. The fact that we are waiting does not mean
1592      * the byterate on the input pad is lower */
1593     if ((started = queue->in_timer_started))
1594       g_timer_stop (queue->in_timer);
1595
1596     GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
1597         "queue is full, waiting for free space");
1598     do {
1599       /* Wait for space to be available, we could be unlocked because of a flush. */
1600       GST_QUEUE2_WAIT_DEL_CHECK (queue, out_flushing);
1601     }
1602     while (gst_queue2_is_filled (queue));
1603
1604     /* and continue if we were running before */
1605     if (started)
1606       g_timer_continue (queue->in_timer);
1607   }
1608
1609   /* put buffer in queue now */
1610   gst_queue2_locked_enqueue (queue, buffer);
1611   GST_QUEUE2_MUTEX_UNLOCK (queue);
1612
1613   return GST_FLOW_OK;
1614
1615   /* special conditions */
1616 out_flushing:
1617   {
1618     GstFlowReturn ret = queue->srcresult;
1619
1620     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1621         "exit because task paused, reason: %s", gst_flow_get_name (ret));
1622     GST_QUEUE2_MUTEX_UNLOCK (queue);
1623     gst_buffer_unref (buffer);
1624
1625     return ret;
1626   }
1627 out_eos:
1628   {
1629     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we received EOS");
1630     GST_QUEUE2_MUTEX_UNLOCK (queue);
1631     gst_buffer_unref (buffer);
1632
1633     return GST_FLOW_UNEXPECTED;
1634   }
1635 out_unexpected:
1636   {
1637     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1638         "exit because we received UNEXPECTED");
1639     GST_QUEUE2_MUTEX_UNLOCK (queue);
1640     gst_buffer_unref (buffer);
1641
1642     return GST_FLOW_UNEXPECTED;
1643   }
1644 }
1645
1646 /* dequeue an item from the queue an push it downstream. This functions returns
1647  * the result of the push. */
1648 static GstFlowReturn
1649 gst_queue2_push_one (GstQueue2 * queue)
1650 {
1651   GstFlowReturn result = GST_FLOW_OK;
1652   GstMiniObject *data;
1653
1654   data = gst_queue2_locked_dequeue (queue);
1655   if (data == NULL)
1656     goto no_item;
1657
1658 next:
1659   if (GST_IS_BUFFER (data)) {
1660     GstBuffer *buffer;
1661     GstCaps *caps;
1662
1663     buffer = GST_BUFFER_CAST (data);
1664     caps = GST_BUFFER_CAPS (buffer);
1665
1666     GST_QUEUE2_MUTEX_UNLOCK (queue);
1667
1668     /* set caps before pushing the buffer so that core does not try to do
1669      * something fancy to check if this is possible. */
1670     if (caps && caps != GST_PAD_CAPS (queue->srcpad))
1671       gst_pad_set_caps (queue->srcpad, caps);
1672
1673     result = gst_pad_push (queue->srcpad, buffer);
1674
1675     /* need to check for srcresult here as well */
1676     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1677     if (result == GST_FLOW_UNEXPECTED) {
1678       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1679           "got UNEXPECTED from downstream");
1680       /* stop pushing buffers, we dequeue all items until we see an item that we
1681        * can push again, which is EOS or NEWSEGMENT. If there is nothing in the
1682        * queue we can push, we set a flag to make the sinkpad refuse more
1683        * buffers with an UNEXPECTED return value until we receive something
1684        * pushable again or we get flushed. */
1685       while ((data = gst_queue2_locked_dequeue (queue))) {
1686         if (GST_IS_BUFFER (data)) {
1687           GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1688               "dropping UNEXPECTED buffer %p", data);
1689           gst_buffer_unref (GST_BUFFER_CAST (data));
1690         } else if (GST_IS_EVENT (data)) {
1691           GstEvent *event = GST_EVENT_CAST (data);
1692           GstEventType type = GST_EVENT_TYPE (event);
1693
1694           if (type == GST_EVENT_EOS || type == GST_EVENT_NEWSEGMENT) {
1695             /* we found a pushable item in the queue, push it out */
1696             GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1697                 "pushing pushable event %s after UNEXPECTED",
1698                 GST_EVENT_TYPE_NAME (event));
1699             goto next;
1700           }
1701           GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1702               "dropping UNEXPECTED event %p", event);
1703           gst_event_unref (event);
1704         }
1705       }
1706       /* no more items in the queue. Set the unexpected flag so that upstream
1707        * make us refuse any more buffers on the sinkpad. Since we will still
1708        * accept EOS and NEWSEGMENT we return _FLOW_OK to the caller so that the
1709        * task function does not shut down. */
1710       queue->unexpected = TRUE;
1711       result = GST_FLOW_OK;
1712     }
1713   } else if (GST_IS_EVENT (data)) {
1714     GstEvent *event = GST_EVENT_CAST (data);
1715     GstEventType type = GST_EVENT_TYPE (event);
1716
1717     GST_QUEUE2_MUTEX_UNLOCK (queue);
1718
1719     gst_pad_push_event (queue->srcpad, event);
1720
1721     GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1722     /* if we're EOS, return UNEXPECTED so that the task pauses. */
1723     if (type == GST_EVENT_EOS) {
1724       GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1725           "pushed EOS event %p, return UNEXPECTED", event);
1726       result = GST_FLOW_UNEXPECTED;
1727     }
1728   }
1729   return result;
1730
1731   /* ERRORS */
1732 no_item:
1733   {
1734     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1735         "exit because we have no item in the queue");
1736     return GST_FLOW_ERROR;
1737   }
1738 out_flushing:
1739   {
1740     GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are flushing");
1741     return GST_FLOW_WRONG_STATE;
1742   }
1743 }
1744
1745 /* called repeadedly with @pad as the source pad. This function should push out
1746  * data to the peer element. */
1747 static void
1748 gst_queue2_loop (GstPad * pad)
1749 {
1750   GstQueue2 *queue;
1751   GstFlowReturn ret;
1752
1753   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1754
1755   /* have to lock for thread-safety */
1756   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1757
1758   if (gst_queue2_is_empty (queue)) {
1759     gboolean started;
1760
1761     /* pause the timer while we wait. The fact that we are waiting does not mean
1762      * the byterate on the output pad is lower */
1763     if ((started = queue->out_timer_started))
1764       g_timer_stop (queue->out_timer);
1765
1766     GST_CAT_DEBUG_OBJECT (queue_dataflow, queue,
1767         "queue is empty, waiting for new data");
1768     do {
1769       /* Wait for data to be available, we could be unlocked because of a flush. */
1770       GST_QUEUE2_WAIT_ADD_CHECK (queue, out_flushing);
1771     }
1772     while (gst_queue2_is_empty (queue));
1773
1774     /* and continue if we were running before */
1775     if (started)
1776       g_timer_continue (queue->out_timer);
1777   }
1778   ret = gst_queue2_push_one (queue);
1779   queue->srcresult = ret;
1780   if (ret != GST_FLOW_OK)
1781     goto out_flushing;
1782
1783   GST_QUEUE2_MUTEX_UNLOCK (queue);
1784
1785   return;
1786
1787   /* ERRORS */
1788 out_flushing:
1789   {
1790     gboolean eos = queue->is_eos;
1791     GstFlowReturn ret = queue->srcresult;
1792
1793     gst_pad_pause_task (queue->srcpad);
1794     GST_CAT_LOG_OBJECT (queue_dataflow, queue,
1795         "pause task, reason:  %s", gst_flow_get_name (queue->srcresult));
1796     GST_QUEUE2_MUTEX_UNLOCK (queue);
1797     /* let app know about us giving up if upstream is not expected to do so */
1798     /* UNEXPECTED is already taken care of elsewhere */
1799     if (eos && (GST_FLOW_IS_FATAL (ret) || ret == GST_FLOW_NOT_LINKED) &&
1800         (ret != GST_FLOW_UNEXPECTED)) {
1801       GST_ELEMENT_ERROR (queue, STREAM, FAILED,
1802           (_("Internal data flow error.")),
1803           ("streaming task paused, reason %s (%d)",
1804               gst_flow_get_name (ret), ret));
1805       gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
1806     }
1807     return;
1808   }
1809 }
1810
1811 static gboolean
1812 gst_queue2_handle_src_event (GstPad * pad, GstEvent * event)
1813 {
1814   gboolean res = TRUE;
1815   GstQueue2 *queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1816
1817 #ifndef GST_DISABLE_GST_DEBUG
1818   GST_CAT_DEBUG_OBJECT (queue_dataflow, queue, "got event %p (%s)",
1819       event, GST_EVENT_TYPE_NAME (event));
1820 #endif
1821
1822   if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1823     /* just forward upstream */
1824     res = gst_pad_push_event (queue->sinkpad, event);
1825   } else {
1826     /* when using a temp file, we eat the event */
1827     res = TRUE;
1828     gst_event_unref (event);
1829   }
1830
1831   return res;
1832 }
1833
1834 static gboolean
1835 gst_queue2_peer_query (GstQueue2 * queue, GstPad * pad, GstQuery * query)
1836 {
1837   gboolean ret = FALSE;
1838   GstPad *peer;
1839
1840   if ((peer = gst_pad_get_peer (pad))) {
1841     ret = gst_pad_query (peer, query);
1842     gst_object_unref (peer);
1843   }
1844   return ret;
1845 }
1846
1847 static gboolean
1848 gst_queue2_handle_src_query (GstPad * pad, GstQuery * query)
1849 {
1850   GstQueue2 *queue;
1851
1852   queue = GST_QUEUE2 (GST_PAD_PARENT (pad));
1853
1854   switch (GST_QUERY_TYPE (query)) {
1855     case GST_QUERY_POSITION:
1856     {
1857       gint64 peer_pos;
1858       GstFormat format;
1859
1860       if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1861         goto peer_failed;
1862
1863       /* get peer position */
1864       gst_query_parse_position (query, &format, &peer_pos);
1865
1866       /* FIXME: this code assumes that there's no discont in the queue */
1867       switch (format) {
1868         case GST_FORMAT_BYTES:
1869           peer_pos -= queue->cur_level.bytes;
1870           break;
1871         case GST_FORMAT_TIME:
1872           peer_pos -= queue->cur_level.time;
1873           break;
1874         default:
1875           GST_WARNING_OBJECT (queue, "dropping query in %s format, don't "
1876               "know how to adjust value", gst_format_get_name (format));
1877           return FALSE;
1878       }
1879       /* set updated position */
1880       gst_query_set_position (query, format, peer_pos);
1881       break;
1882     }
1883     case GST_QUERY_DURATION:
1884     {
1885       GST_DEBUG_OBJECT (queue, "doing peer query");
1886
1887       if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1888         goto peer_failed;
1889
1890       GST_DEBUG_OBJECT (queue, "peer query success");
1891       break;
1892     }
1893     case GST_QUERY_BUFFERING:
1894     {
1895       GstFormat format;
1896
1897       GST_DEBUG_OBJECT (queue, "query buffering");
1898
1899       if (!QUEUE_IS_USING_TEMP_FILE (queue)) {
1900         /* no temp file, just forward to the peer */
1901         if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1902           goto peer_failed;
1903         GST_DEBUG_OBJECT (queue, "buffering forwarded to peer");
1904       } else {
1905         gint64 start, stop;
1906         guint64 writing_pos;
1907
1908         gst_query_parse_buffering_range (query, &format, NULL, NULL, NULL);
1909
1910         switch (format) {
1911           case GST_FORMAT_PERCENT:
1912           {
1913             gint64 duration;
1914             GstFormat peer_fmt;
1915
1916             peer_fmt = GST_FORMAT_BYTES;
1917
1918             if (!gst_pad_query_peer_duration (queue->sinkpad, &peer_fmt,
1919                     &duration))
1920               goto peer_failed;
1921
1922             writing_pos = queue->current->writing_pos;
1923
1924             GST_DEBUG_OBJECT (queue, "duration %" G_GINT64_FORMAT ", writing %"
1925                 G_GINT64_FORMAT, duration, writing_pos);
1926
1927             start = 0;
1928             /* get our available data relative to the duration */
1929             if (duration != -1)
1930               stop = GST_FORMAT_PERCENT_MAX * writing_pos / duration;
1931             else
1932               stop = -1;
1933             break;
1934           }
1935           case GST_FORMAT_BYTES:
1936             start = 0;
1937             stop = queue->current->writing_pos;
1938             break;
1939           default:
1940             start = -1;
1941             stop = -1;
1942             break;
1943         }
1944         gst_query_set_buffering_percent (query, queue->is_buffering, 100);
1945         gst_query_set_buffering_range (query, format, start, stop, -1);
1946       }
1947       break;
1948     }
1949     default:
1950       /* peer handled other queries */
1951       if (!gst_queue2_peer_query (queue, queue->sinkpad, query))
1952         goto peer_failed;
1953       break;
1954   }
1955
1956   return TRUE;
1957
1958   /* ERRORS */
1959 peer_failed:
1960   {
1961     GST_DEBUG_OBJECT (queue, "failed peer query");
1962     return FALSE;
1963   }
1964 }
1965
1966 static GstFlowReturn
1967 gst_queue2_get_range (GstPad * pad, guint64 offset, guint length,
1968     GstBuffer ** buffer)
1969 {
1970   GstQueue2 *queue;
1971   GstFlowReturn ret;
1972
1973   queue = GST_QUEUE2_CAST (gst_pad_get_parent (pad));
1974
1975   GST_QUEUE2_MUTEX_LOCK_CHECK (queue, out_flushing);
1976   length = (length == -1) ? DEFAULT_BUFFER_SIZE : length;
1977   offset = (offset == -1) ? queue->current->reading_pos : offset;
1978
1979   /* function will block when the range is not yet available */
1980   ret = gst_queue2_create_read (queue, offset, length, buffer);
1981   GST_QUEUE2_MUTEX_UNLOCK (queue);
1982
1983   gst_object_unref (queue);
1984
1985   return ret;
1986
1987   /* ERRORS */
1988 out_flushing:
1989   {
1990     GST_DEBUG_OBJECT (queue, "we are flushing");
1991     GST_QUEUE2_MUTEX_UNLOCK (queue);
1992     return GST_FLOW_WRONG_STATE;
1993   }
1994 }
1995
1996 static gboolean
1997 gst_queue2_src_checkgetrange_function (GstPad * pad)
1998 {
1999   GstQueue2 *queue;
2000   gboolean ret;
2001
2002   queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2003
2004   /* we can operate in pull mode when we are using a tempfile */
2005   ret = QUEUE_IS_USING_TEMP_FILE (queue);
2006
2007   gst_object_unref (GST_OBJECT (queue));
2008
2009   return ret;
2010 }
2011
2012 /* sink currently only operates in push mode */
2013 static gboolean
2014 gst_queue2_sink_activate_push (GstPad * pad, gboolean active)
2015 {
2016   gboolean result = TRUE;
2017   GstQueue2 *queue;
2018
2019   queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2020
2021   if (active) {
2022     GST_QUEUE2_MUTEX_LOCK (queue);
2023     GST_DEBUG_OBJECT (queue, "activating push mode");
2024     queue->srcresult = GST_FLOW_OK;
2025     queue->is_eos = FALSE;
2026     queue->unexpected = FALSE;
2027     reset_rate_timer (queue);
2028     GST_QUEUE2_MUTEX_UNLOCK (queue);
2029   } else {
2030     /* unblock chain function */
2031     GST_QUEUE2_MUTEX_LOCK (queue);
2032     GST_DEBUG_OBJECT (queue, "deactivating push mode");
2033     queue->srcresult = GST_FLOW_WRONG_STATE;
2034     gst_queue2_locked_flush (queue);
2035     GST_QUEUE2_MUTEX_UNLOCK (queue);
2036   }
2037
2038   gst_object_unref (queue);
2039
2040   return result;
2041 }
2042
2043 /* src operating in push mode, we start a task on the source pad that pushes out
2044  * buffers from the queue */
2045 static gboolean
2046 gst_queue2_src_activate_push (GstPad * pad, gboolean active)
2047 {
2048   gboolean result = FALSE;
2049   GstQueue2 *queue;
2050
2051   queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2052
2053   if (active) {
2054     GST_QUEUE2_MUTEX_LOCK (queue);
2055     GST_DEBUG_OBJECT (queue, "activating push mode");
2056     queue->srcresult = GST_FLOW_OK;
2057     queue->is_eos = FALSE;
2058     queue->unexpected = FALSE;
2059     result = gst_pad_start_task (pad, (GstTaskFunction) gst_queue2_loop, pad);
2060     GST_QUEUE2_MUTEX_UNLOCK (queue);
2061   } else {
2062     /* unblock loop function */
2063     GST_QUEUE2_MUTEX_LOCK (queue);
2064     GST_DEBUG_OBJECT (queue, "deactivating push mode");
2065     queue->srcresult = GST_FLOW_WRONG_STATE;
2066     /* the item add signal will unblock */
2067     g_cond_signal (queue->item_add);
2068     GST_QUEUE2_MUTEX_UNLOCK (queue);
2069
2070     /* step 2, make sure streaming finishes */
2071     result = gst_pad_stop_task (pad);
2072   }
2073
2074   gst_object_unref (queue);
2075
2076   return result;
2077 }
2078
2079 /* pull mode, downstream will call our getrange function */
2080 static gboolean
2081 gst_queue2_src_activate_pull (GstPad * pad, gboolean active)
2082 {
2083   gboolean result;
2084   GstQueue2 *queue;
2085
2086   queue = GST_QUEUE2 (gst_pad_get_parent (pad));
2087
2088   if (active) {
2089     if (QUEUE_IS_USING_TEMP_FILE (queue)) {
2090       /* open the temp file now */
2091       result = gst_queue2_open_temp_location_file (queue);
2092
2093       GST_QUEUE2_MUTEX_LOCK (queue);
2094       GST_DEBUG_OBJECT (queue, "activating pull mode");
2095       queue->srcresult = GST_FLOW_OK;
2096       queue->is_eos = FALSE;
2097       queue->unexpected = FALSE;
2098       GST_QUEUE2_MUTEX_UNLOCK (queue);
2099     } else {
2100       GST_QUEUE2_MUTEX_LOCK (queue);
2101       GST_DEBUG_OBJECT (queue, "no temp file, cannot activate pull mode");
2102       /* this is not allowed, we cannot operate in pull mode without a temp
2103        * file. */
2104       queue->srcresult = GST_FLOW_WRONG_STATE;
2105       result = FALSE;
2106       GST_QUEUE2_MUTEX_UNLOCK (queue);
2107     }
2108   } else {
2109     GST_QUEUE2_MUTEX_LOCK (queue);
2110     GST_DEBUG_OBJECT (queue, "deactivating pull mode");
2111     queue->srcresult = GST_FLOW_WRONG_STATE;
2112     /* this will unlock getrange */
2113     g_cond_signal (queue->item_add);
2114     result = TRUE;
2115     GST_QUEUE2_MUTEX_UNLOCK (queue);
2116   }
2117   gst_object_unref (queue);
2118
2119   return result;
2120 }
2121
2122 static GstStateChangeReturn
2123 gst_queue2_change_state (GstElement * element, GstStateChange transition)
2124 {
2125   GstQueue2 *queue;
2126   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
2127
2128   queue = GST_QUEUE2 (element);
2129
2130   switch (transition) {
2131     case GST_STATE_CHANGE_NULL_TO_READY:
2132       break;
2133     case GST_STATE_CHANGE_READY_TO_PAUSED:
2134       if (QUEUE_IS_USING_TEMP_FILE (queue)) {
2135         if (!gst_queue2_open_temp_location_file (queue))
2136           ret = GST_STATE_CHANGE_FAILURE;
2137       }
2138       queue->segment_event_received = FALSE;
2139       queue->starting_segment = NULL;
2140       break;
2141     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
2142       break;
2143     default:
2144       break;
2145   }
2146
2147   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2148
2149   switch (transition) {
2150     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2151       break;
2152     case GST_STATE_CHANGE_PAUSED_TO_READY:
2153       if (QUEUE_IS_USING_TEMP_FILE (queue))
2154         gst_queue2_close_temp_location_file (queue);
2155       if (queue->starting_segment != NULL) {
2156         gst_event_unref (queue->starting_segment);
2157         queue->starting_segment = NULL;
2158       }
2159       break;
2160     case GST_STATE_CHANGE_READY_TO_NULL:
2161       break;
2162     default:
2163       break;
2164   }
2165
2166   return ret;
2167 }
2168
2169 /* changing the capacity of the queue must wake up
2170  * the _chain function, it might have more room now
2171  * to store the buffer/event in the queue */
2172 #define QUEUE_CAPACITY_CHANGE(q)\
2173   g_cond_signal (queue->item_del);
2174
2175 /* Changing the minimum required fill level must
2176  * wake up the _loop function as it might now
2177  * be able to preceed.
2178  */
2179 #define QUEUE_THRESHOLD_CHANGE(q)\
2180   g_cond_signal (queue->item_add);
2181
2182 static void
2183 gst_queue2_set_temp_template (GstQueue2 * queue, const gchar * template)
2184 {
2185   GstState state;
2186
2187   /* the element must be stopped in order to do this */
2188   GST_OBJECT_LOCK (queue);
2189   state = GST_STATE (queue);
2190   if (state != GST_STATE_READY && state != GST_STATE_NULL)
2191     goto wrong_state;
2192   GST_OBJECT_UNLOCK (queue);
2193
2194   /* set new location */
2195   g_free (queue->temp_template);
2196   queue->temp_template = g_strdup (template);
2197
2198   return;
2199
2200 /* ERROR */
2201 wrong_state:
2202   {
2203     GST_WARNING_OBJECT (queue, "setting temp-template property in wrong state");
2204     GST_OBJECT_UNLOCK (queue);
2205   }
2206 }
2207
2208 static void
2209 gst_queue2_set_property (GObject * object,
2210     guint prop_id, const GValue * value, GParamSpec * pspec)
2211 {
2212   GstQueue2 *queue = GST_QUEUE2 (object);
2213
2214   /* someone could change levels here, and since this
2215    * affects the get/put funcs, we need to lock for safety. */
2216   GST_QUEUE2_MUTEX_LOCK (queue);
2217
2218   switch (prop_id) {
2219     case PROP_MAX_SIZE_BYTES:
2220       queue->max_level.bytes = g_value_get_uint (value);
2221       QUEUE_CAPACITY_CHANGE (queue);
2222       break;
2223     case PROP_MAX_SIZE_BUFFERS:
2224       queue->max_level.buffers = g_value_get_uint (value);
2225       QUEUE_CAPACITY_CHANGE (queue);
2226       break;
2227     case PROP_MAX_SIZE_TIME:
2228       queue->max_level.time = g_value_get_uint64 (value);
2229       /* set rate_time to the same value. We use an extra field in the level
2230        * structure so that we can easily access and compare it */
2231       queue->max_level.rate_time = queue->max_level.time;
2232       QUEUE_CAPACITY_CHANGE (queue);
2233       break;
2234     case PROP_USE_BUFFERING:
2235       queue->use_buffering = g_value_get_boolean (value);
2236       break;
2237     case PROP_USE_RATE_ESTIMATE:
2238       queue->use_rate_estimate = g_value_get_boolean (value);
2239       break;
2240     case PROP_LOW_PERCENT:
2241       queue->low_percent = g_value_get_int (value);
2242       break;
2243     case PROP_HIGH_PERCENT:
2244       queue->high_percent = g_value_get_int (value);
2245       break;
2246     case PROP_TEMP_TEMPLATE:
2247       gst_queue2_set_temp_template (queue, g_value_get_string (value));
2248       break;
2249     case PROP_TEMP_LOCATION:
2250       g_free (queue->temp_location);
2251       queue->temp_location = g_value_dup_string (value);
2252       /* you can set the property back to NULL to make it use the temp-tmpl
2253        * property. */
2254       queue->temp_location_set = queue->temp_location != NULL;
2255       break;
2256     case PROP_TEMP_REMOVE:
2257       queue->temp_remove = g_value_get_boolean (value);
2258       break;
2259     default:
2260       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2261       break;
2262   }
2263
2264   GST_QUEUE2_MUTEX_UNLOCK (queue);
2265 }
2266
2267 static void
2268 gst_queue2_get_property (GObject * object,
2269     guint prop_id, GValue * value, GParamSpec * pspec)
2270 {
2271   GstQueue2 *queue = GST_QUEUE2 (object);
2272
2273   GST_QUEUE2_MUTEX_LOCK (queue);
2274
2275   switch (prop_id) {
2276     case PROP_CUR_LEVEL_BYTES:
2277       g_value_set_uint (value, queue->cur_level.bytes);
2278       break;
2279     case PROP_CUR_LEVEL_BUFFERS:
2280       g_value_set_uint (value, queue->cur_level.buffers);
2281       break;
2282     case PROP_CUR_LEVEL_TIME:
2283       g_value_set_uint64 (value, queue->cur_level.time);
2284       break;
2285     case PROP_MAX_SIZE_BYTES:
2286       g_value_set_uint (value, queue->max_level.bytes);
2287       break;
2288     case PROP_MAX_SIZE_BUFFERS:
2289       g_value_set_uint (value, queue->max_level.buffers);
2290       break;
2291     case PROP_MAX_SIZE_TIME:
2292       g_value_set_uint64 (value, queue->max_level.time);
2293       break;
2294     case PROP_USE_BUFFERING:
2295       g_value_set_boolean (value, queue->use_buffering);
2296       break;
2297     case PROP_USE_RATE_ESTIMATE:
2298       g_value_set_boolean (value, queue->use_rate_estimate);
2299       break;
2300     case PROP_LOW_PERCENT:
2301       g_value_set_int (value, queue->low_percent);
2302       break;
2303     case PROP_HIGH_PERCENT:
2304       g_value_set_int (value, queue->high_percent);
2305       break;
2306     case PROP_TEMP_TEMPLATE:
2307       g_value_set_string (value, queue->temp_template);
2308       break;
2309     case PROP_TEMP_LOCATION:
2310       g_value_set_string (value, queue->temp_location);
2311       break;
2312     case PROP_TEMP_REMOVE:
2313       g_value_set_boolean (value, queue->temp_remove);
2314       break;
2315     default:
2316       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2317       break;
2318   }
2319
2320   GST_QUEUE2_MUTEX_UNLOCK (queue);
2321 }