subprojects/gstreamer/plugins/elements/gstqueue2.c

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