plugins/elements/gstqueue2.c

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