gstbasesrc.c

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

    case PROP_BLOCKSIZE:
      src->blocksize = g_value_get_ulong (value);
      break;
    case PROP_NUM_BUFFERS:
      src->num_buffers = g_value_get_int (value);
      break;
    case PROP_TYPEFIND:
      src->data.ABI.typefind = g_value_get_boolean (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static void
gst_base_src_get_property (GObject * object, guint prop_id, GValue * value,
    GParamSpec * pspec)
{
  GstBaseSrc *src;

  src = GST_BASE_SRC (object);

  switch (prop_id) {
    case PROP_BLOCKSIZE:
      g_value_set_ulong (value, src->blocksize);
      break;
    case PROP_NUM_BUFFERS:
      g_value_set_int (value, src->num_buffers);
      break;
    case PROP_TYPEFIND:
      g_value_set_boolean (value, src->data.ABI.typefind);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

/* with STREAM_LOCK and LOCK */
static GstClockReturn
gst_base_src_wait (GstBaseSrc * basesrc, GstClockTime time)
{
  GstClockReturn ret;
  GstClockID id;
  GstClock *clock;

  /* get clock, if no clock, we don't sync */
  if ((clock = GST_ELEMENT_CLOCK (basesrc)) == NULL)
    return GST_CLOCK_OK;

  id = gst_clock_new_single_shot_id (clock, time);

  basesrc->clock_id = id;
  /* release the object lock while waiting */
  GST_OBJECT_UNLOCK (basesrc);

  ret = gst_clock_id_wait (id, NULL);

  GST_OBJECT_LOCK (basesrc);
  gst_clock_id_unref (id);
  basesrc->clock_id = NULL;

  return ret;
}

/* perform synchronisation on a buffer. 
 * with STREAM_LOCK.
 */
static GstClockReturn
gst_base_src_do_sync (GstBaseSrc * basesrc, GstBuffer * buffer)
{
  GstClockReturn result;
  GstClockTime start, end;
  GstBaseSrcClass *bclass;
  GstClockTime base_time;

  bclass = GST_BASE_SRC_GET_CLASS (basesrc);

  start = end = -1;
  if (bclass->get_times)
    bclass->get_times (basesrc, buffer, &start, &end);

  /* if we don't have a timestamp, we don't sync */
  if (!GST_CLOCK_TIME_IS_VALID (start))
    goto invalid_start;

  /* now do clocking */
  GST_OBJECT_LOCK (basesrc);
  base_time = GST_ELEMENT_CAST (basesrc)->base_time;

  GST_LOG_OBJECT (basesrc,
      "waiting for clock, base time %" GST_TIME_FORMAT
      ", stream_start %" GST_TIME_FORMAT,
      GST_TIME_ARGS (base_time), GST_TIME_ARGS (start));

  result = gst_base_src_wait (basesrc, start + base_time);
  GST_OBJECT_UNLOCK (basesrc);

  GST_LOG_OBJECT (basesrc, "clock entry done: %d", result);

  return result;

  /* special cases */
invalid_start:
  {
    GST_DEBUG_OBJECT (basesrc, "get_times returned invalid start");
    return GST_CLOCK_OK;
  }
}

static gboolean
gst_base_src_update_length (GstBaseSrc * src, guint64 offset, guint * length)
{
  guint64 size, maxsize;
  GstBaseSrcClass *bclass;

  bclass = GST_BASE_SRC_GET_CLASS (src);

  /* only operate if we are working with bytes */
  if (src->segment.format != GST_FORMAT_BYTES)
    return TRUE;

  /* get total file size */
  size = (guint64) src->segment.duration;

  /* the max amount of bytes to read is the total size or
   * up to the segment.stop if present. */
  if (src->segment.stop != -1)
    maxsize = MIN (size, src->segment.stop);
  else
    maxsize = size;

  GST_DEBUG_OBJECT (src,
      "reading offset %" G_GUINT64_FORMAT ", length %u, size %" G_GINT64_FORMAT
      ", segment.stop %" G_GINT64_FORMAT ", maxsize %" G_GINT64_FORMAT, offset,
      *length, size, src->segment.stop, maxsize);

  /* check size if we have one */
  if (maxsize != -1) {
    /* if we run past the end, check if the file became bigger and 
     * retry. */
    if (G_UNLIKELY (offset + *length >= maxsize)) {
      /* see if length of the file changed */
      if (bclass->get_size)
        if (!bclass->get_size (src, &size))
          size = -1;

      gst_segment_set_duration (&src->segment, GST_FORMAT_BYTES, size);

      /* make sure we don't exceed the configured segment stop
       * if it was set */
      if (src->segment.stop != -1)
        maxsize = MIN (size, src->segment.stop);
      else
        maxsize = size;

      /* if we are at or past the end, EOS */
      if (G_UNLIKELY (offset >= maxsize))
        goto unexpected_length;

      /* else we can clip to the end */
      if (G_UNLIKELY (offset + *length >= maxsize))
        *length = maxsize - offset;

    }
  }

  /* keep track of current position. segment is in bytes, we checked 
   * that above. */
  gst_segment_set_last_stop (&src->segment, GST_FORMAT_BYTES, offset);

  return TRUE;

  /* ERRORS */
unexpected_length:
  {
    return FALSE;
  }
}

static GstFlowReturn
gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length,
    GstBuffer ** buf)
{
  GstFlowReturn ret;
  GstBaseSrcClass *bclass;
  GstClockReturn status;

  bclass = GST_BASE_SRC_GET_CLASS (src);

  ret = gst_base_src_wait_playing (src);
  if (ret != GST_FLOW_OK)
    goto stopped;

  if (G_UNLIKELY (!GST_OBJECT_FLAG_IS_SET (src, GST_BASE_SRC_STARTED)))
    goto not_started;

  if (G_UNLIKELY (!bclass->create))
    goto no_function;

  if (G_UNLIKELY (!gst_base_src_update_length (src, offset, &length)))
    goto unexpected_length;

  /* normally we don't count buffers */
  if (G_UNLIKELY (src->num_buffers_left >= 0)) {
    if (src->num_buffers_left == 0)
      goto reached_num_buffers;
    else
      src->num_buffers_left--;
  }

  GST_DEBUG_OBJECT (src,
      "calling create offset %" G_GUINT64_FORMAT " length %u, time %"
      G_GINT64_FORMAT, offset, length, src->segment.time);

  ret = bclass->create (src, offset, length, buf);
  if (G_UNLIKELY (ret != GST_FLOW_OK))
    goto done;

  /* no timestamp set and we are at offset 0, we can timestamp with 0 */
  if (offset == 0 && src->segment.time == 0
      && GST_BUFFER_TIMESTAMP (*buf) == -1)
    GST_BUFFER_TIMESTAMP (*buf) = 0;

  /* now sync before pushing the buffer */
  status = gst_base_src_do_sync (src, *buf);
  switch (status) {
    case GST_CLOCK_EARLY:
      /* the buffer is too late. We currently don't drop the buffer. */
      GST_DEBUG_OBJECT (src, "buffer too late!, returning anyway");
      break;
    case GST_CLOCK_OK:
      /* buffer synchronised properly */
      GST_DEBUG_OBJECT (src, "buffer ok");
      break;
    case GST_CLOCK_UNSCHEDULED:
      /* this case is triggered when we were waiting for the clock and
       * it got unlocked because we did a state change. We return 
       * WRONG_STATE in this case to stop the dataflow also get rid of the
       * produced buffer. */
      GST_DEBUG_OBJECT (src,
          "clock was unscheduled (%d), returning WRONG_STATE", status);
      gst_buffer_unref (*buf);
      *buf = NULL;
      ret = GST_FLOW_WRONG_STATE;
      break;
    default:
      /* all other result values are unexpected and errors */
      GST_ELEMENT_ERROR (src, CORE, CLOCK,
          (_("Internal clock error.")),
          ("clock returned unexpected return value %d", status));
      gst_buffer_unref (*buf);
      *buf = NULL;
      ret = GST_FLOW_ERROR;
      break;
  }
done:
  return ret;

  /* ERROR */
stopped:
  {
    GST_DEBUG_OBJECT (src, "wait_playing returned %d", ret);
    return ret;
  }
not_started:
  {
    GST_DEBUG_OBJECT (src, "getrange but not started");
    return GST_FLOW_WRONG_STATE;
  }
no_function:
  {
    GST_DEBUG_OBJECT (src, "no create function");
    return GST_FLOW_ERROR;
  }
unexpected_length:
  {
    GST_DEBUG_OBJECT (src, "unexpected length %u (offset=%" G_GUINT64_FORMAT
        ", size=%" G_GINT64_FORMAT ")", length, offset, src->segment.duration);
    return GST_FLOW_UNEXPECTED;
  }
reached_num_buffers:
  {
    GST_DEBUG_OBJECT (src, "sent all buffers");
    return GST_FLOW_UNEXPECTED;
  }
}

static GstFlowReturn
gst_base_src_pad_get_range (GstPad * pad, guint64 offset, guint length,
    GstBuffer ** buf)
{
  GstBaseSrc *src;
  GstFlowReturn res;

  src = GST_BASE_SRC (gst_pad_get_parent (pad));

  res = gst_base_src_get_range (src, offset, length, buf);

  gst_object_unref (src);

  return res;
}

static gboolean
gst_base_src_default_check_get_range (GstBaseSrc * src)
{
  gboolean res;

  if (!GST_OBJECT_FLAG_IS_SET (src, GST_BASE_SRC_STARTED)) {
    GST_LOG_OBJECT (src, "doing start/stop to check get_range support");
    if (G_LIKELY (gst_base_src_start (src)))
      gst_base_src_stop (src);
  }

  /* we can operate in getrange mode if the native format is bytes
   * and we are seekable, this condition is set in the random_access
   * flag and is set in the _start() method. */
  res = src->random_access;

  return res;
}

static gboolean
gst_base_src_check_get_range (GstBaseSrc * src)
{
  GstBaseSrcClass *bclass;
  gboolean res;

  bclass = GST_BASE_SRC_GET_CLASS (src);

  if (bclass->check_get_range == NULL)
    goto no_function;

  res = bclass->check_get_range (src);
  GST_LOG_OBJECT (src, "%s() returned %d",
      GST_DEBUG_FUNCPTR_NAME (bclass->check_get_range), (gint) res);

  return res;

  /* ERRORS */
no_function:
  {
    GST_WARNING_OBJECT (src, "no check_get_range function set");
    return FALSE;
  }
}

static gboolean
gst_base_src_pad_check_get_range (GstPad * pad)
{
  GstBaseSrc *src;
  gboolean res;

  src = GST_BASE_SRC (gst_pad_get_parent (pad));

  res = gst_base_src_check_get_range (src);

  gst_object_unref (src);

  return res;
}

static void
gst_base_src_loop (GstPad * pad)
{
  GstBaseSrc *src;
  GstBuffer *buf = NULL;
  GstFlowReturn ret;
  gint64 position;
  gboolean eos;

  eos = FALSE;

  src = GST_BASE_SRC (gst_pad_get_parent (pad));

  src->priv->last_sent_eos = FALSE;

  /* if we operate in bytes, we can calculate an offset */
  if (src->segment.format == GST_FORMAT_BYTES)
    position = src->segment.last_stop;
  else
    position = -1;

  ret = gst_base_src_get_range (src, position, src->blocksize, &buf);
  if (G_UNLIKELY (ret != GST_FLOW_OK)) {
    GST_INFO_OBJECT (src, "pausing after gst_base_src_get_range() = %s",
        gst_flow_get_name (ret));
    goto pause;
  }
  /* this should not happen */
  if (G_UNLIKELY (buf == NULL))
    goto null_buffer;

  /* push events to close/start our segment before we push the buffer. */
  if (src->priv->close_segment) {
    gst_pad_push_event (pad, src->priv->close_segment);
    src->priv->close_segment = NULL;
  }
  if (src->priv->start_segment) {
    gst_pad_push_event (pad, src->priv->start_segment);
    src->priv->start_segment = NULL;
  }

  /* figure out the new position */
  switch (src->segment.format) {
    case GST_FORMAT_BYTES:
      position += GST_BUFFER_SIZE (buf);
      break;
    case GST_FORMAT_TIME:
    {
      GstClockTime start, duration;

      start = GST_BUFFER_TIMESTAMP (buf);
      duration = GST_BUFFER_DURATION (buf);

      if (GST_CLOCK_TIME_IS_VALID (start))
        position = start;
      else
        position = src->segment.last_stop;

      if (GST_CLOCK_TIME_IS_VALID (duration))
        position += duration;
      break;
    }
    case GST_FORMAT_DEFAULT:
      position = GST_BUFFER_OFFSET_END (buf);
      break;
    default:
      position = -1;
      break;
  }
  if (position != -1) {
    if (src->segment.stop != -1) {
      if (position >= src->segment.stop) {
        eos = TRUE;
        position = src->segment.stop;
      }
    }