/* GStreamer

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

 *                    2000 Wim Taymans <wtay@chello.be>

 *                    2003 Colin Walters <cwalters@gnome.org>

 *                    2005 Wim Taymans <wim@fluendo.com>

 *

 * gstqueue.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-queue

 *

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

 * #GstQueue:max-size-buffers, #GstQueue:max-size-bytes and/or

 * #GstQueue:max-size-time properties has been reached. Any attempt to push

 * 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

 * #GstQueue:current-level-buffers property. You can track changes

 * by connecting to the notify::current-level-buffers signal (which

 * like all signals will be emitted from the streaming thread). The same

 * applies to the #GstQueue:current-level-time and

 * #GstQueue:current-level-bytes properties.

 *

 * The default queue size limits are 200 buffers, 10MB of data, or

 * one second worth of data, whichever is reached first.

 *

 * As said earlier, the queue blocks by default when one of the specified

 * maximums (bytes, time, buffers) has been reached. You can set the

 * #GstQueue:leaky property to specify that instead of blocking it should

 * leak (drop) new or old buffers.

 *

 * The #GstQueue::underrun signal is emitted when the queue has less data than

 * the specified minimum thresholds require (by default: when the queue is

 * empty). The #GstQueue::overrun signal is emitted when the queue is filled

 * up. Both signals are emitted from the context of the streaming thread.

 */

#include "gst/gst_private.h"

#include <gst/gst.h>

#include "gstqueue.h"

#include "../../gst/gst-i18n-lib.h"

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);

#define STATUS(queue, pad, msg) \

  GST_CAT_LOG_OBJECT (queue_dataflow, queue, \

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

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

                      "-%" G_GUINT64_FORMAT " ns, %u items", \

                      GST_DEBUG_PAD_NAME (pad), \

                      queue->cur_level.buffers, \

                      queue->min_threshold.buffers, \

                      queue->max_size.buffers, \

                      queue->cur_level.bytes, \

                      queue->min_threshold.bytes, \

                      queue->max_size.bytes, \

                      queue->cur_level.time, \

                      queue->min_threshold.time, \

                      queue->max_size.time, \

                      queue->queue->length)

/* Queue signals and args */

enum

{

  SIGNAL_UNDERRUN,

  SIGNAL_RUNNING,

  SIGNAL_OVERRUN,

  SIGNAL_PUSHING,

  LAST_SIGNAL

};

enum

{

  ARG_0,

  /* FIXME: don't we have another way of doing this

   * "Gstreamer format" (frame/byte/time) queries? */

  ARG_CUR_LEVEL_BUFFERS,

  ARG_CUR_LEVEL_BYTES,

  ARG_CUR_LEVEL_TIME,

  ARG_MAX_SIZE_BUFFERS,

  ARG_MAX_SIZE_BYTES,

  ARG_MAX_SIZE_TIME,

  ARG_MIN_THRESHOLD_BUFFERS,

  ARG_MIN_THRESHOLD_BYTES,

  ARG_MIN_THRESHOLD_TIME,

  ARG_LEAKY

      /* FILL ME */

};

/* default property values */

#define DEFAULT_MAX_SIZE_BUFFERS  200   /* 200 buffers */

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

#define DEFAULT_MAX_SIZE_TIME     GST_SECOND    /* 1 second    */

#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 (q, q->sinkpad, "wait for DEL");                               \

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

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

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

    goto label;                                                         \

  }                                                                     \

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

} G_STMT_END

#define GST_QUEUE_WAIT_ADD_CHECK(q, label) G_STMT_START {               \

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

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

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

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

    goto label;                                                         \

  }                                                                     \

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

} G_STMT_END

#define GST_QUEUE_SIGNAL_DEL(q) G_STMT_START {                          \

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

  g_cond_signal (q->item_del);                                          \

} G_STMT_END

#define GST_QUEUE_SIGNAL_ADD(q) G_STMT_START {                          \

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

  g_cond_signal (q->item_add);                                          \

} G_STMT_END

#define _do_init(bla) \

    GST_DEBUG_CATEGORY_INIT (queue_debug, "queue", 0, "queue element"); \

    GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue_dataflow", 0, \

        "dataflow inside the queue element");

GST_BOILERPLATE_FULL (GstQueue, gst_queue, GstElement,

    GST_TYPE_ELEMENT, _do_init);

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 GstPadLinkReturn gst_queue_link_sink (GstPad * pad, GstPad * peer);

static GstPadLinkReturn gst_queue_link_src (GstPad * pad, GstPad * peer);

static void gst_queue_locked_flush (GstQueue * queue);

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

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

static gboolean gst_queue_is_empty (GstQueue * queue);

static gboolean gst_queue_is_filled (GstQueue * queue);

#define GST_TYPE_QUEUE_LEAKY (queue_leaky_get_type ())

static GType

queue_leaky_get_type (void)

{

  static GType queue_leaky_type = 0;

  static const GEnumValue queue_leaky[] = {

    {GST_QUEUE_NO_LEAK, "Not Leaky", "no"},

    {GST_QUEUE_LEAK_UPSTREAM, "Leaky on upstream (new buffers)", "upstream"},

    {GST_QUEUE_LEAK_DOWNSTREAM, "Leaky on downstream (old buffers)",

        "downstream"},

    {0, NULL, NULL},

  };

  if (!queue_leaky_type) {

    queue_leaky_type = g_enum_register_static ("GstQueueLeaky", queue_leaky);

  }

  return queue_leaky_type;

}

static guint gst_queue_signals[LAST_SIGNAL] = { 0 };

static void

gst_queue_base_init (gpointer g_class)

{

  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);

  gst_element_class_set_details_simple (gstelement_class,

      "Queue",

      "Generic", "Simple data queue", "Erik Walthinsen <omega@cse.ogi.edu>");

  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));

}

static void

gst_queue_class_init (GstQueueClass * klass)

{

  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

  gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_queue_set_property);

  gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_queue_get_property);

  /* signals */

  /**

   * GstQueue::underrun:

   * @queue: the queue instance

   *

   * Reports that the buffer became empty (underrun).

   * A buffer is empty if the total amount of data inside it (num-buffers, time,

   * size) is lower than the boundary values which can be set through the

   * GObject properties.

   */

  gst_queue_signals[SIGNAL_UNDERRUN] =

      g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstQueueClass, underrun), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  /**

   * GstQueue::running:

   * @queue: the queue instance

   *

   * Reports that enough (min-threshold) data is in the queue. Use this signal

   * together with the underrun signal to pause the pipeline on underrun and

   * wait for the queue to fill-up before resume playback.

   */

  gst_queue_signals[SIGNAL_RUNNING] =

      g_signal_new ("running", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstQueueClass, running), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  /**

   * GstQueue::overrun:

   * @queue: the queue instance

   *

   * Reports that the buffer became full (overrun).

   * A buffer is full if the total amount of data inside it (num-buffers, time,

   * size) is higher than the boundary values which can be set through the

   * GObject properties.

   */

  gst_queue_signals[SIGNAL_OVERRUN] =

      g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstQueueClass, overrun), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  /**

   * GstQueue::pushing:

   * @queue: the queue instance

   *

   * Reports when the queue has enough data to start pushing data again on the

   * source pad.

   */

  gst_queue_signals[SIGNAL_PUSHING] =

      g_signal_new ("pushing", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstQueueClass, pushing), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  /* properties */

  g_object_class_install_property (gobject_class, ARG_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, ARG_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, ARG_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, ARG_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, ARG_MAX_SIZE_BUFFERS,

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

  g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,

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

  g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_BYTES,

      g_param_spec_uint ("min-threshold-bytes", "Min. threshold (kB)",

          "Min. amount of data in the queue to allow reading (bytes, 0=disable)",

  g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_BUFFERS,

      g_param_spec_uint ("min-threshold-buffers", "Min. threshold (buffers)",

          "Min. number of buffers in the queue to allow reading (0=disable)",

  g_object_class_install_property (gobject_class, ARG_MIN_THRESHOLD_TIME,

      g_param_spec_uint64 ("min-threshold-time", "Min. threshold (ns)",

          "Min. amount of data in the queue to allow reading (in ns, 0=disable)",

  g_object_class_install_property (gobject_class, ARG_LEAKY,

      g_param_spec_enum ("leaky", "Leaky",

          "Where the queue leaks, if at all",

  /* set several parent class virtual functions */

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_queue_finalize);

}

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_link_function (queue->sinkpad,

      GST_DEBUG_FUNCPTR (gst_queue_link_sink));

  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_activatepush_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_src_activate_push));

  gst_pad_set_link_function (queue->srcpad,

      GST_DEBUG_FUNCPTR (gst_queue_link_src));

  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);

  GST_QUEUE_CLEAR_LEVEL (queue->cur_level);

  queue->max_size.buffers = DEFAULT_MAX_SIZE_BUFFERS;

  queue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;

  queue->max_size.time = DEFAULT_MAX_SIZE_TIME;

  GST_QUEUE_CLEAR_LEVEL (queue->min_threshold);

  GST_QUEUE_CLEAR_LEVEL (queue->orig_min_threshold);

  gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);

  gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);

  queue->head_needs_discont = queue->tail_needs_discont = FALSE;

  queue->leaky = GST_QUEUE_NO_LEAK;

  queue->srcresult = GST_FLOW_WRONG_STATE;

  queue->qlock = g_mutex_new ();

  queue->item_add = g_cond_new ();

  queue->item_del = g_cond_new ();

  queue->queue = g_queue_new ();

  GST_DEBUG_OBJECT (queue,

      "initialized queue's not_empty & not_full conditions");

}

/* called only once, as opposed to dispose */

static void

gst_queue_finalize (GObject * object)

{

  GstQueue *queue = GST_QUEUE (object);

  GST_DEBUG_OBJECT (queue, "finalizing queue");

  while (!g_queue_is_empty (queue->queue)) {

    GstMiniObject *data = g_queue_pop_head (queue->queue);

    gst_mini_object_unref (data);

  }

  g_queue_free (queue->queue);

  g_mutex_free (queue->qlock);

  g_cond_free (queue->item_add);

  g_cond_free (queue->item_del);

  G_OBJECT_CLASS (parent_class)->finalize (object);

}

static GstCaps *

gst_queue_getcaps (GstPad * pad)

{

  GstQueue *queue;

  GstPad *otherpad;

  GstCaps *result;

  queue = GST_QUEUE (GST_PAD_PARENT (pad));

  otherpad = (pad == queue->srcpad ? queue->sinkpad : queue->srcpad);

  result = gst_pad_peer_get_caps (otherpad);

  if (result == NULL)

    result = gst_caps_new_any ();

  return result;

}

static GstPadLinkReturn

gst_queue_link_sink (GstPad * pad, GstPad * peer)

{

  return GST_PAD_LINK_OK;

}

static GstPadLinkReturn

gst_queue_link_src (GstPad * pad, GstPad * peer)

{

  GstPadLinkReturn result = GST_PAD_LINK_OK;

  GstQueue *queue;

  queue = GST_QUEUE (gst_pad_get_parent (pad));

  GST_DEBUG_OBJECT (queue, "queue linking source pad");

  if (GST_PAD_LINKFUNC (peer)) {

    result = GST_PAD_LINKFUNC (peer) (peer, pad);

  }

  if (GST_PAD_LINK_SUCCESSFUL (result)) {

    GST_QUEUE_MUTEX_LOCK (queue);

    if (queue->srcresult == GST_FLOW_OK) {

      gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad);

      GST_DEBUG_OBJECT (queue, "starting task as pad is linked");

    } else {

      GST_DEBUG_OBJECT (queue, "not starting task reason %s",

          gst_flow_get_name (queue->srcresult));

    }

    GST_QUEUE_MUTEX_UNLOCK (queue);

  }

  gst_object_unref (queue);

  return result;

}

static GstFlowReturn

gst_queue_bufferalloc (GstPad * pad, guint64 offset, guint size, GstCaps * caps,

    GstBuffer ** buf)

{

  GstQueue *queue;

  GstFlowReturn result;

  queue = GST_QUEUE (GST_PAD_PARENT (pad));

  /* Forward to src pad, without setting caps on the src pad */

  result = gst_pad_alloc_buffer (queue->srcpad, offset, size, caps, buf);

  return result;

}

/* calculate the diff between running time on the sink and src of the queue.

 * This is the total amount of time in the queue. */

static void

update_time_level (GstQueue * queue)

{

  gint64 sink_time, src_time;

  sink_time =

      gst_segment_to_running_time (&queue->sink_segment, GST_FORMAT_TIME,

      queue->sink_segment.last_stop);

  src_time = gst_segment_to_running_time (&queue->src_segment, GST_FORMAT_TIME,

      queue->src_segment.last_stop);