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