/* GStreamer

 * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>

 *                    2003 Colin Walters <cwalters@gnome.org>

 *                    2000,2005,2007 Wim Taymans <wim.taymans@gmail.com>

 *                    2007 Thiago Sousa Santos <thiagoss@lcc.ufcg.edu.br>

 *

 * gstqueue2.c:

 *

 * This library is free software; you can redistribute it and/or

 * modify it under the terms of the GNU Library General Public

 * License as published by the Free Software Foundation; either

 * version 2 of the License, or (at your option) any later version.

 *

 * This library is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

 * Library General Public License for more details.

 *

 * You should have received a copy of the GNU Library General Public

 * License along with this library; if not, write to the

 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,

 * Boston, MA 02111-1307, USA.

 */

/**

 * SECTION:element-queue2

 * @short_description: Asynchronous data queue.

 *

 * Data is queued until one of the limits specified by the

 * more buffers into the queue will block the pushing thread until more space

 * becomes available.

 *

 * The queue will create a new thread on the source pad to decouple the

 * processing on sink and source pad.

 *

 * You can query how many buffers are queued by reading the

 *

 * The default queue size limits are 100 buffers, 2MB of data, or

 * two seconds worth of data, whichever is reached first.

 *

 *

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include <glib/gstdio.h>

#include <gst/gst.h>

#include <gst/gst-i18n-plugin.h>

#endif

static const GstElementDetails gst_queue_details = GST_ELEMENT_DETAILS ("Queue",

    "Generic",

    "Simple data queue",

    "Erik Walthinsen <omega@cse.ogi.edu>, "

    "Wim Taymans <wim.taymans@gmail.com>");

static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",

    GST_PAD_SINK,

    GST_PAD_ALWAYS,

    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",

    GST_PAD_SRC,

    GST_PAD_ALWAYS,

    GST_STATIC_CAPS_ANY);

GST_DEBUG_CATEGORY_STATIC (queue_debug);

#define GST_CAT_DEFAULT (queue_debug)

GST_DEBUG_CATEGORY_STATIC (queue_dataflow);

enum

{

  LAST_SIGNAL

};

/* default property values */

#define DEFAULT_MAX_SIZE_BUFFERS   100  /* 100 buffers */

#define DEFAULT_MAX_SIZE_BYTES     (2 * 1024 * 1024)    /* 2 MB */

#define DEFAULT_MAX_SIZE_TIME      2 * GST_SECOND       /* 2 seconds */

#define DEFAULT_USE_BUFFERING      FALSE

#define DEFAULT_USE_RATE_ESTIMATE  TRUE

#define DEFAULT_LOW_PERCENT        10

#define DEFAULT_HIGH_PERCENT       99

/* other defines */

#define DEFAULT_BUFFER_SIZE 4096

enum

{

  PROP_0,

  PROP_CUR_LEVEL_BUFFERS,

  PROP_CUR_LEVEL_BYTES,

  PROP_CUR_LEVEL_TIME,

  PROP_MAX_SIZE_BUFFERS,

  PROP_MAX_SIZE_BYTES,

  PROP_MAX_SIZE_TIME,

  PROP_USE_BUFFERING,

  PROP_USE_RATE_ESTIMATE,

  PROP_LOW_PERCENT,

  PROP_HIGH_PERCENT,

  PROP_TEMP_LOCATION

};

#define GST_TYPE_QUEUE \

  (gst_queue_get_type())

#define GST_QUEUE(obj) \

  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_QUEUE,GstQueue))

#define GST_QUEUE_CLASS(klass) \

  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_QUEUE,GstQueueClass))

#define GST_IS_QUEUE(obj) \

  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_QUEUE))

#define GST_IS_QUEUE_CLASS(klass) \

  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_QUEUE))

#define GST_QUEUE_CAST(obj) \

  ((GstQueue *)(obj))

typedef struct _GstQueue GstQueue;

typedef struct _GstQueueSize GstQueueSize;

typedef struct _GstQueueClass GstQueueClass;

/* used to keep track of sizes (current and max) */

struct _GstQueueSize

{

  guint buffers;

  guint bytes;

  guint64 time;

  guint64 rate_time;

};

#define GST_QUEUE_CLEAR_LEVEL(l) G_STMT_START {         \

  l.buffers = 0;                                        \

  l.bytes = 0;                                          \

  l.time = 0;                                           \

  l.rate_time = 0;                                      \

} G_STMT_END

struct _GstQueue

{

  GstElement element;

  /*< private > */

  GstPad *sinkpad;

  GstPad *srcpad;

  /* segments to keep track of timestamps */

  GstSegment sink_segment;

  GstSegment src_segment;

  /* flowreturn when srcpad is paused */

  GstFlowReturn srcresult;

  gboolean is_eos;

  gboolean unexpected;

  /* the queue of data we're keeping our hands on */

  GQueue *queue;

  GstQueueSize cur_level;       /* currently in the queue */

  GstQueueSize max_level;       /* max. amount of data allowed in the queue */

  gboolean use_buffering;

  gboolean use_rate_estimate;

  GstClockTime buffering_interval;

  gint low_percent;             /* low/high watermarks for buffering */

  gint high_percent;

  /* current buffering state */

  gboolean is_buffering;

  guint buffering_iteration;

  /* for measuring input/output rates */

  GTimer *in_timer;

  gboolean in_timer_started;

  gdouble last_in_elapsed;

  guint64 bytes_in;

  gdouble byte_in_rate;

  GTimer *out_timer;

  gboolean out_timer_started;

  gdouble last_out_elapsed;

  guint64 bytes_out;

  gdouble byte_out_rate;

  GMutex *qlock;                /* lock for queue (vs object lock) */

  gboolean waiting_add;

  GCond *item_add;              /* signals buffers now available for reading */

  gboolean waiting_del;

  GCond *item_del;              /* signals space now available for writing */

  /* temp location stuff */

  gchar *temp_location;

  FILE *temp_file;

  guint64 writing_pos;

  guint64 reading_pos;

  /* we need this to send the first new segment event of the stream

   * because we can't save it on the file */

  gboolean segment_event_received;

  GstEvent *starting_segment;

};

struct _GstQueueClass

{

  GstElementClass parent_class;

};

#define STATUS(queue, pad, msg) \

  GST_CAT_LOG_OBJECT (queue_dataflow, queue, \

                      "(%s:%s) " msg ": %u of %u buffers, %u of %u " \

                      "bytes, %" G_GUINT64_FORMAT " of %" G_GUINT64_FORMAT \

                      " ns, %"G_GUINT64_FORMAT" items", \

                      GST_DEBUG_PAD_NAME (pad), \

                      queue->cur_level.buffers, \

                      queue->max_level.buffers, \

                      queue->cur_level.bytes, \

                      queue->max_level.bytes, \

                      queue->cur_level.time, \

                      queue->max_level.time, \

                      (guint64) (QUEUE_IS_USING_TEMP_FILE(queue) ? \

                        queue->queue->length))

#define GST_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \

  g_mutex_lock (q->qlock);                                              \

} G_STMT_END

#define GST_QUEUE_MUTEX_LOCK_CHECK(q,label) G_STMT_START {              \

  GST_QUEUE_MUTEX_LOCK (q);                                             \

  if (q->srcresult != GST_FLOW_OK)                                      \

    goto label;                                                         \

} G_STMT_END

#define GST_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                        \

  g_mutex_unlock (q->qlock);                                            \

} G_STMT_END

#define GST_QUEUE_WAIT_DEL_CHECK(q, label) G_STMT_START {               \

  STATUS (queue, q->sinkpad, "wait for DEL");                           \

  q->waiting_del = TRUE;                                                \

  g_cond_wait (q->item_del, queue->qlock);                              \

  q->waiting_del = FALSE;                                               \

  if (q->srcresult != GST_FLOW_OK) {                                    \

    STATUS (queue, q->srcpad, "received DEL wakeup");                   \

    goto label;                                                         \

  }                                                                     \

  STATUS (queue, q->sinkpad, "received DEL");                           \

} G_STMT_END

#define GST_QUEUE_WAIT_ADD_CHECK(q, label) G_STMT_START {               \

  STATUS (queue, q->srcpad, "wait for ADD");                            \

  q->waiting_add = TRUE;                                                \

  g_cond_wait (q->item_add, q->qlock);                                  \

  q->waiting_add = FALSE;                                               \

  if (q->srcresult != GST_FLOW_OK) {                                    \

    STATUS (queue, q->srcpad, "received ADD wakeup");                   \

    goto label;                                                         \

  }                                                                     \

  STATUS (queue, q->srcpad, "received ADD");                            \

} G_STMT_END

#define GST_QUEUE_SIGNAL_DEL(q) G_STMT_START {                          \

  if (q->waiting_del) {                                                 \

    STATUS (q, q->srcpad, "signal DEL");                                \

    g_cond_signal (q->item_del);                                        \

  }                                                                     \

} G_STMT_END

#define GST_QUEUE_SIGNAL_ADD(q) G_STMT_START {                          \

  if (q->waiting_add) {                                                 \

    STATUS (q, q->sinkpad, "signal ADD");                               \

    g_cond_signal (q->item_add);                                        \

  }                                                                     \

} G_STMT_END

#define _do_init(bla) \

/* can't use boilerplate as we need to register with Queue2 to avoid conflicts

 * with queue in core elements */

static void gst_queue_class_init (GstQueueClass * klass);

static void gst_queue_init (GstQueue * queue, GstQueueClass * g_class);

static GstElementClass *parent_class;

static GType

gst_queue_get_type (void)

{

  static GType gst_queue_type = 0;

  if (!gst_queue_type) {

    static const GTypeInfo gst_queue_info = {

      sizeof (GstQueueClass),

      NULL,

      NULL,

      (GClassInitFunc) gst_queue_class_init,

      NULL,

      NULL,

      sizeof (GstQueue),

      0,

      (GInstanceInitFunc) gst_queue_init,

      NULL

    };

    gst_queue_type =

        g_type_register_static (GST_TYPE_ELEMENT, "GstQueue2",

        &gst_queue_info, 0);

  }

  return gst_queue_type;

}

static void gst_queue_finalize (GObject * object);

static void gst_queue_set_property (GObject * object,

    guint prop_id, const GValue * value, GParamSpec * pspec);

static void gst_queue_get_property (GObject * object,

    guint prop_id, GValue * value, GParamSpec * pspec);

static GstFlowReturn gst_queue_chain (GstPad * pad, GstBuffer * buffer);

static GstFlowReturn gst_queue_bufferalloc (GstPad * pad, guint64 offset,

    guint size, GstCaps * caps, GstBuffer ** buf);

static GstFlowReturn gst_queue_push_one (GstQueue * queue);

static void gst_queue_loop (GstPad * pad);

static gboolean gst_queue_handle_sink_event (GstPad * pad, GstEvent * event);

static gboolean gst_queue_handle_src_event (GstPad * pad, GstEvent * event);

static gboolean gst_queue_handle_src_query (GstPad * pad, GstQuery * query);

static GstCaps *gst_queue_getcaps (GstPad * pad);

static GstFlowReturn gst_queue_get_range (GstPad * pad, guint64 offset,

    guint length, GstBuffer ** buffer);

static gboolean gst_queue_src_checkgetrange_function (GstPad * pad);

static gboolean gst_queue_src_activate_pull (GstPad * pad, gboolean active);

static gboolean gst_queue_src_activate_push (GstPad * pad, gboolean active);

static gboolean gst_queue_sink_activate_push (GstPad * pad, gboolean active);

static GstStateChangeReturn gst_queue_change_state (GstElement * element,

    GstStateChange transition);

static gboolean gst_queue_is_empty (GstQueue * queue);

static gboolean gst_queue_is_filled (GstQueue * queue);

/* static guint gst_queue_signals[LAST_SIGNAL] = { 0 }; */

static void

gst_queue_class_init (GstQueueClass * klass)

{

  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);

  parent_class = g_type_class_peek_parent (klass);

  gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_queue_set_property);

  gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_queue_get_property);

  /* properties */

  g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,

      g_param_spec_uint ("current-level-bytes", "Current level (kB)",

          "Current amount of data in the queue (bytes)",

  g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BUFFERS,

      g_param_spec_uint ("current-level-buffers", "Current level (buffers)",

          "Current number of buffers in the queue",

  g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,

      g_param_spec_uint64 ("current-level-time", "Current level (ns)",

          "Current amount of data in the queue (in ns)",

  g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BYTES,

      g_param_spec_uint ("max-size-bytes", "Max. size (kB)",

          "Max. amount of data in the queue (bytes, 0=disable)",

  g_object_class_install_property (gobject_class, PROP_MAX_SIZE_BUFFERS,

      g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",

  g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,

      g_param_spec_uint64 ("max-size-time", "Max. size (ns)",

  g_object_class_install_property (gobject_class, PROP_USE_BUFFERING,

      g_param_spec_boolean ("use-buffering", "Use buffering",

          "Emit GST_MESSAGE_BUFFERING based on low-/high-percent thresholds",

  g_object_class_install_property (gobject_class, PROP_USE_RATE_ESTIMATE,

      g_param_spec_boolean ("use-rate-estimate", "Use Rate Estimate",

          "Estimate the bitrate of the stream to calculate time level",

  g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,

      g_param_spec_int ("low-percent", "Low percent",

  g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,

      g_param_spec_int ("high-percent", "High percent",

  g_object_class_install_property (gobject_class, PROP_TEMP_LOCATION,

      g_param_spec_string ("temp-location", "Temporary File Location",

  gst_element_class_add_pad_template (gstelement_class,

      gst_static_pad_template_get (&srctemplate));

  gst_element_class_add_pad_template (gstelement_class,

      gst_static_pad_template_get (&sinktemplate));

  gst_element_class_set_details (gstelement_class, &gst_queue_details);

  /* set several parent class virtual functions */

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_queue_finalize);

  gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_queue_change_state);

}

static void

gst_queue_init (GstQueue * queue, GstQueueClass * g_class)

{

  queue->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");

  gst_pad_set_chain_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_chain));

  gst_pad_set_activatepush_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_sink_activate_push));

  gst_pad_set_event_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_handle_sink_event));

  gst_pad_set_getcaps_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_getcaps));

  gst_pad_set_bufferalloc_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_bufferalloc));

  gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);

  queue->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");

  gst_pad_set_activatepull_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_src_activate_pull));

  gst_pad_set_activatepush_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_src_activate_push));

  gst_pad_set_getrange_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_get_range));

  gst_pad_set_checkgetrange_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_src_checkgetrange_function));

  gst_pad_set_getcaps_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_getcaps));

  gst_pad_set_event_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_handle_src_event));

  gst_pad_set_query_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_handle_src_query));

  gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);

  /* levels */

  GST_QUEUE_CLEAR_LEVEL (queue->cur_level);

  queue->max_level.buffers = DEFAULT_MAX_SIZE_BUFFERS;

  queue->max_level.bytes = DEFAULT_MAX_SIZE_BYTES;

  queue->max_level.time = DEFAULT_MAX_SIZE_TIME;

  queue->max_level.rate_time = DEFAULT_MAX_SIZE_TIME;

  queue->use_buffering = DEFAULT_USE_BUFFERING;

  queue->use_rate_estimate = DEFAULT_USE_RATE_ESTIMATE;

  queue->low_percent = DEFAULT_LOW_PERCENT;

  queue->high_percent = DEFAULT_HIGH_PERCENT;

  gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);

  gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);

  queue->srcresult = GST_FLOW_WRONG_STATE;

  queue->is_eos = FALSE;

  queue->in_timer = g_timer_new ();

  queue->out_timer = g_timer_new ();

  queue->qlock = g_mutex_new ();

  queue->waiting_add = FALSE;

  queue->item_add = g_cond_new ();

  queue->waiting_del = FALSE;

  queue->item_del = g_cond_new ();

  queue->queue = g_queue_new ();

  /* tempfile related */

  queue->temp_location = NULL;

  GST_DEBUG_OBJECT (queue,