gstdataqueue.c

GStreamer是一个开源的多媒体框架库。利用它

      queue->cur_level.bytes, queue->cur_level.time, queue->checkdata);
}

/**
 * gst_data_queue_flush:
 * @queue: a #GstDataQueue.
 *
 * Flushes all the contents of the @queue. Any call to #gst_data_queue_pull and
 * #gst_data_queue_pop will be released.
 * MT safe.
 */

void
gst_data_queue_flush (GstDataQueue * queue)
{
  GST_DEBUG ("queue:%p", queue);
  GST_DATA_QUEUE_MUTEX_LOCK (queue);
  gst_data_queue_locked_flush (queue);
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
}

/**
 * gst_data_queue_is_empty:
 * @queue: a #GstDataQueue.
 *
 * Queries if there are any items in the @queue.
 * MT safe.
 *
 * Returns: #TRUE if @queue is empty.
 */

gboolean
gst_data_queue_is_empty (GstDataQueue * queue)
{
  gboolean res;

  GST_DATA_QUEUE_MUTEX_LOCK (queue);
  res = gst_data_queue_locked_is_empty (queue);
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);

  return res;
}

/**
 * gst_data_queue_is_full:
 * @queue: a #GstDataQueue.
 *
 * Queries if @queue is full. This check will be done using the
 * #GstDataQueueCheckFullCallback registered with @queue.
 * MT safe.
 *
 * Returns: #TRUE if @queue is full.
 */

gboolean
gst_data_queue_is_full (GstDataQueue * queue)
{
  gboolean res;

  GST_DATA_QUEUE_MUTEX_LOCK (queue);
  res = gst_data_queue_locked_is_full (queue);
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);

  return res;
}

/**
 * gst_data_queue_set_flushing:
 * @queue: a #GstDataQueue.
 * @flushing: a #gboolean stating if the queue will be flushing or not.
 *
 * Sets the queue to flushing state if @flushing is #TRUE. If set to flushing
 * state, any incoming data on the @queue will be discarded. Any call currently
 * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
 * away with a return value of #FALSE. While the @queue is in flushing state, 
 * all calls to those two functions will return #FALSE.
 * MT Safe.
 */

void
gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
{

  GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
  GST_DATA_QUEUE_MUTEX_LOCK (queue);
  queue->flushing = flushing;
  if (flushing) {
    /* release push/pop functions */
    g_cond_signal (queue->item_add);
    g_cond_signal (queue->item_del);
  }
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
}

/**
 * gst_data_queue_push:
 * @queue: a #GstDataQueue.
 * @item: a #GstDataQueueItem.
 *
 * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
 * on the @queue. If the @queue is full, the call will block until space is
 * available, OR the @queue is set to flushing state.
 * MT safe.
 *
 * Returns: #TRUE if the @item was successfully pushed on the @queue.
 */
gboolean
gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
{
  gboolean res = FALSE;

  GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, done);

  STATUS (queue, "before pushing");

  /* We ALWAYS need to check for queue fillness */
  while (gst_data_queue_locked_is_full (queue)) {
    GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
    g_signal_emit (G_OBJECT (queue), gst_data_queue_signals[SIGNAL_FULL], 0);
    GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, done);

    /* signal might have removed some items */
    while (gst_data_queue_locked_is_full (queue)) {
      g_cond_wait (queue->item_del, queue->qlock);
      if (queue->flushing)
        goto done;
    }
  }

  g_queue_push_tail (queue->queue, item);
  res = TRUE;

  if (item->visible)
    queue->cur_level.visible++;
  queue->cur_level.bytes += item->size;
  queue->cur_level.time += item->duration;

  STATUS (queue, "after pushing");

  g_cond_signal (queue->item_add);

done:
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);

  GST_DEBUG ("queue:%p, result:%d", queue, res);

  return res;
}

/**
 * gst_data_queue_pop:
 * @queue: a #GstDataQueue.
 * @item: pointer to store the returned #GstDataQueueItem.
 *
 * Retrieves the first @item available on the @queue. If the queue is currently
 * empty, the call will block until at least one item is available, OR the
 * @queue is set to the flushing state.
 * MT safe.
 *
 * Returns: #TRUE if an @item was successfully retrieved from the @queue.
 */

gboolean
gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
{
  gboolean res = FALSE;

  GST_DEBUG ("queue:%p", queue);

  g_return_val_if_fail (item != NULL, FALSE);

  GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, done);

  STATUS (queue, "before popping");

  while (gst_data_queue_locked_is_empty (queue)) {
    GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
    g_signal_emit (G_OBJECT (queue), gst_data_queue_signals[SIGNAL_EMPTY], 0);
    GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, done);

    while (gst_data_queue_locked_is_empty (queue)) {
      g_cond_wait (queue->item_add, queue->qlock);
      if (queue->flushing)
        goto done;
    }
  }

  /* Get the item from the GQueue */
  *item = g_queue_pop_head (queue->queue);
  res = TRUE;

  /* update current level counter */
  if ((*item)->visible)
    queue->cur_level.visible--;
  queue->cur_level.bytes -= (*item)->size;
  queue->cur_level.time -= (*item)->duration;

  g_cond_signal (queue->item_del);

done:
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);

  GST_DEBUG ("queue:%p , res:%d", queue, res);

  return res;
}

/**
 * gst_data_queue_drop_head:
 * @queue: The #GstDataQueue to drop an item from.
 * @type: The #GType of the item to drop.
 *
 * Pop and unref the head-most #GstMiniObject with the given #GType.
 *
 * Returns: TRUE if an element was removed.
 */

gboolean
gst_data_queue_drop_head (GstDataQueue * queue, GType type)
{
  gboolean res = FALSE;
  GList *item;
  GstDataQueueItem *leak = NULL;

  GST_DEBUG ("queue:%p", queue);

  GST_DATA_QUEUE_MUTEX_LOCK (queue);
  for (item = g_queue_peek_head_link (queue->queue); item; item = item->next) {
    GstDataQueueItem *tmp = (GstDataQueueItem *) item->data;

    if (G_TYPE_CHECK_INSTANCE_TYPE (tmp->object, type)) {
      leak = tmp;
      break;
    }
  }

  if (!leak)
    goto done;

  g_queue_delete_link (queue->queue, item);

  if (leak->visible)
    queue->cur_level.visible--;
  queue->cur_level.bytes -= leak->size;
  queue->cur_level.time -= leak->duration;

  leak->destroy (leak);

  res = TRUE;

done:
  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);

  GST_DEBUG ("queue:%p , res:%d", queue, res);

  return res;
}

static void
gst_data_queue_set_property (GObject * object,
    guint prop_id, const GValue * value, GParamSpec * pspec)
{
  switch (prop_id) {
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static void
gst_data_queue_get_property (GObject * object,
    guint prop_id, GValue * value, GParamSpec * pspec)
{
  GstDataQueue *queue = GST_DATA_QUEUE (object);

  GST_DATA_QUEUE_MUTEX_LOCK (queue);

  switch (prop_id) {
    case ARG_CUR_LEVEL_BYTES:
      g_value_set_uint (value, queue->cur_level.bytes);
      break;
    case ARG_CUR_LEVEL_VISIBLE:
      g_value_set_uint (value, queue->cur_level.visible);
      break;
    case ARG_CUR_LEVEL_TIME:
      g_value_set_uint64 (value, queue->cur_level.time);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }

  GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
}