gstbasesrc.c

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

    gst_segment_set_last_stop (&src->segment, src->segment.format, position);
  }

  if (G_UNLIKELY (src->priv->discont)) {
    buf = gst_buffer_make_metadata_writable (buf);
    GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
    src->priv->discont = FALSE;
  }

  ret = gst_pad_push (pad, buf);
  if (G_UNLIKELY (ret != GST_FLOW_OK)) {
    GST_INFO_OBJECT (src, "pausing after gst_pad_push() = %s",
        gst_flow_get_name (ret));
    goto pause;
  }

  if (eos) {
    GST_INFO_OBJECT (src, "pausing after EOS");
    ret = GST_FLOW_UNEXPECTED;
    goto pause;
  }

done:
  gst_object_unref (src);
  return;

  /* special cases */
pause:
  {
    const gchar *reason = gst_flow_get_name (ret);

    GST_DEBUG_OBJECT (src, "pausing task, reason %s", reason);
    src->data.ABI.running = FALSE;
    gst_pad_pause_task (pad);
    if (GST_FLOW_IS_FATAL (ret) || ret == GST_FLOW_NOT_LINKED) {
      if (ret == GST_FLOW_UNEXPECTED) {
        /* perform EOS logic */
        if (src->segment.flags & GST_SEEK_FLAG_SEGMENT) {
          gst_element_post_message (GST_ELEMENT_CAST (src),
              gst_message_new_segment_done (GST_OBJECT_CAST (src),
                  src->segment.format, src->segment.last_stop));
        } else {
          gst_pad_push_event (pad, gst_event_new_eos ());
          src->priv->last_sent_eos = TRUE;
        }
      } else {
        /* for fatal errors we post an error message, post the error
         * first so the app knows about the error first. */
        GST_ELEMENT_ERROR (src, STREAM, FAILED,
            (_("Internal data flow error.")),
            ("streaming task paused, reason %s (%d)", reason, ret));
        gst_pad_push_event (pad, gst_event_new_eos ());
        src->priv->last_sent_eos = TRUE;
      }
    }
    goto done;
  }
null_buffer:
  {
    GST_ELEMENT_ERROR (src, STREAM, FAILED,
        (_("Internal data flow error.")), ("element returned NULL buffer"));
    /* we finished the segment on error */
    src->data.ABI.running = FALSE;
    gst_pad_pause_task (pad);
    gst_pad_push_event (pad, gst_event_new_eos ());
    src->priv->last_sent_eos = TRUE;
    goto done;
  }
}

/* this will always be called between start() and stop(). So you can rely on
 * resources allocated by start() and freed from stop(). This needs to be added
 * to the docs at some point. */
static gboolean
gst_base_src_unlock (GstBaseSrc * basesrc)
{
  GstBaseSrcClass *bclass;
  gboolean result = TRUE;

  GST_DEBUG ("unlock");
  /* unblock whatever the subclass is doing */
  bclass = GST_BASE_SRC_GET_CLASS (basesrc);
  if (bclass->unlock)
    result = bclass->unlock (basesrc);

  GST_DEBUG ("unschedule clock");
  /* and unblock the clock as well, if any */
  GST_OBJECT_LOCK (basesrc);
  if (basesrc->clock_id) {
    gst_clock_id_unschedule (basesrc->clock_id);
  }
  GST_OBJECT_UNLOCK (basesrc);

  GST_DEBUG ("unlock done");

  return result;
}

/* this will always be called between start() and stop(). So you can rely on
 * resources allocated by start() and freed from stop(). This needs to be added
 * to the docs at some point. */
static gboolean
gst_base_src_unlock_stop (GstBaseSrc * basesrc)
{
  GstBaseSrcClass *bclass;
  gboolean result = TRUE;

  GST_DEBUG_OBJECT (basesrc, "unlock stop");

  /* Finish a previous unblock request, allowing subclasses to flush command
   * queues or whatever they need to do */
  bclass = GST_BASE_SRC_GET_CLASS (basesrc);
  if (bclass->unlock_stop)
    result = bclass->unlock_stop (basesrc);

  GST_DEBUG_OBJECT (basesrc, "unlock stop done");

  return result;
}

/* default negotiation code. 
 *
 * Take intersection between src and sink pads, take first
 * caps and fixate. 
 */
static gboolean
gst_base_src_default_negotiate (GstBaseSrc * basesrc)
{
  GstCaps *thiscaps;
  GstCaps *caps = NULL;
  GstCaps *peercaps = NULL;
  gboolean result = FALSE;

  /* first see what is possible on our source pad */
  thiscaps = gst_pad_get_caps (GST_BASE_SRC_PAD (basesrc));
  GST_DEBUG_OBJECT (basesrc, "caps of src: %" GST_PTR_FORMAT, thiscaps);
  /* nothing or anything is allowed, we're done */
  if (thiscaps == NULL || gst_caps_is_any (thiscaps))
    goto no_nego_needed;

  /* get the peer caps */
  peercaps = gst_pad_peer_get_caps (GST_BASE_SRC_PAD (basesrc));
  GST_DEBUG_OBJECT (basesrc, "caps of peer: %" GST_PTR_FORMAT, peercaps);
  if (peercaps) {
    GstCaps *icaps;

    /* get intersection */
    icaps = gst_caps_intersect (thiscaps, peercaps);
    GST_DEBUG_OBJECT (basesrc, "intersect: %" GST_PTR_FORMAT, icaps);
    gst_caps_unref (thiscaps);
    gst_caps_unref (peercaps);
    if (icaps) {
      /* take first (and best, since they are sorted) possibility */
      caps = gst_caps_copy_nth (icaps, 0);
      gst_caps_unref (icaps);
    }
  } else {
    /* no peer, work with our own caps then */
    caps = thiscaps;
  }
  if (caps) {
    caps = gst_caps_make_writable (caps);
    gst_caps_truncate (caps);

    /* now fixate */
    if (!gst_caps_is_empty (caps)) {
      gst_pad_fixate_caps (GST_BASE_SRC_PAD (basesrc), caps);
      GST_DEBUG_OBJECT (basesrc, "fixated to: %" GST_PTR_FORMAT, caps);

      if (gst_caps_is_any (caps)) {
        /* hmm, still anything, so element can do anything and
         * nego is not needed */
        result = TRUE;
      } else if (gst_caps_is_fixed (caps)) {
        /* yay, fixed caps, use those then */
        gst_pad_set_caps (GST_BASE_SRC_PAD (basesrc), caps);
        result = TRUE;
      }
    }
    gst_caps_unref (caps);
  }
  return result;

no_nego_needed:
  {
    GST_DEBUG_OBJECT (basesrc, "no negotiation needed");
    if (thiscaps)
      gst_caps_unref (thiscaps);
    return TRUE;
  }
}

static gboolean
gst_base_src_negotiate (GstBaseSrc * basesrc)
{
  GstBaseSrcClass *bclass;
  gboolean result = TRUE;

  bclass = GST_BASE_SRC_GET_CLASS (basesrc);

  if (bclass->negotiate)
    result = bclass->negotiate (basesrc);

  return result;
}

static gboolean
gst_base_src_start (GstBaseSrc * basesrc)
{
  GstBaseSrcClass *bclass;
  gboolean result;
  guint64 size;

  if (GST_OBJECT_FLAG_IS_SET (basesrc, GST_BASE_SRC_STARTED))
    return TRUE;

  GST_DEBUG_OBJECT (basesrc, "starting source");

  basesrc->num_buffers_left = basesrc->num_buffers;

  gst_segment_init (&basesrc->segment, basesrc->segment.format);
  basesrc->data.ABI.running = FALSE;

  bclass = GST_BASE_SRC_GET_CLASS (basesrc);
  if (bclass->start)
    result = bclass->start (basesrc);
  else
    result = TRUE;

  if (!result)
    goto could_not_start;

  GST_OBJECT_FLAG_SET (basesrc, GST_BASE_SRC_STARTED);

  /* figure out the size */
  if (basesrc->segment.format == GST_FORMAT_BYTES) {
    if (bclass->get_size) {
      if (!(result = bclass->get_size (basesrc, &size)))
        size = -1;
    } else {
      result = FALSE;
      size = -1;
    }
    /* only update the size when operating in bytes, subclass is supposed
     * to set duration in the start method for other formats */
    gst_segment_set_duration (&basesrc->segment, GST_FORMAT_BYTES, size);
  } else {
    size = -1;
  }

  GST_DEBUG_OBJECT (basesrc,
      "format: %d, have size: %d, size: %" G_GUINT64_FORMAT ", duration: %"
      G_GINT64_FORMAT, basesrc->segment.format, result, size,
      basesrc->segment.duration);

  /* check if we can seek */
  if (bclass->is_seekable)
    basesrc->seekable = bclass->is_seekable (basesrc);
  else
    basesrc->seekable = FALSE;

  GST_DEBUG_OBJECT (basesrc, "is seekable: %d", basesrc->seekable);

  /* update for random access flag */
  basesrc->random_access = basesrc->seekable &&
      basesrc->segment.format == GST_FORMAT_BYTES;

  GST_DEBUG_OBJECT (basesrc, "is random_access: %d", basesrc->random_access);

  /* run typefind if we are random_access and the typefinding is enabled. */
  if (basesrc->random_access && basesrc->data.ABI.typefind && size != -1) {
    GstCaps *caps;

    caps = gst_type_find_helper (basesrc->srcpad, size);
    gst_pad_set_caps (basesrc->srcpad, caps);
    gst_caps_unref (caps);
  } else {
    /* use class or default negotiate function */
    if (!gst_base_src_negotiate (basesrc))
      goto could_not_negotiate;
  }

  return TRUE;

  /* ERROR */
could_not_start:
  {
    GST_DEBUG_OBJECT (basesrc, "could not start");
    /* subclass is supposed to post a message. We don't have to call _stop. */
    return FALSE;
  }
could_not_negotiate:
  {
    GST_DEBUG_OBJECT (basesrc, "could not negotiate, stopping");
    GST_ELEMENT_ERROR (basesrc, STREAM, FORMAT,
        ("Could not negotiate format"), ("Check your filtered caps, if any"));
    /* we must call stop */
    gst_base_src_stop (basesrc);
    return FALSE;
  }
}

static gboolean
gst_base_src_stop (GstBaseSrc * basesrc)
{
  GstBaseSrcClass *bclass;
  gboolean result = TRUE;

  if (!GST_OBJECT_FLAG_IS_SET (basesrc, GST_BASE_SRC_STARTED))
    return TRUE;

  GST_DEBUG_OBJECT (basesrc, "stopping source");

  bclass = GST_BASE_SRC_GET_CLASS (basesrc);
  if (bclass->stop)
    result = bclass->stop (basesrc);

  if (result)
    GST_OBJECT_FLAG_UNSET (basesrc, GST_BASE_SRC_STARTED);

  return result;
}

static gboolean
gst_base_src_deactivate (GstBaseSrc * basesrc, GstPad * pad)
{
  gboolean result;

  GST_LIVE_LOCK (basesrc);
  basesrc->live_running = TRUE;
  GST_LIVE_SIGNAL (basesrc);
  GST_LIVE_UNLOCK (basesrc);

  /* step 1, unblock clock sync (if any) */
  result = gst_base_src_unlock (basesrc);

  /* step 2, make sure streaming finishes */
  result &= gst_pad_stop_task (pad);

  /* step 3, clear the unblock condition */
  result &= gst_base_src_unlock_stop (basesrc);

  return result;
}

static gboolean
gst_base_src_activate_push (GstPad * pad, gboolean active)
{
  GstBaseSrc *basesrc;
  GstEvent *event;

  basesrc = GST_BASE_SRC (GST_OBJECT_PARENT (pad));

  /* prepare subclass first */
  if (active) {
    GST_DEBUG_OBJECT (basesrc, "Activating in push mode");

    if (G_UNLIKELY (!basesrc->can_activate_push))
      goto no_push_activation;

    if (G_UNLIKELY (!gst_base_src_start (basesrc)))
      goto error_start;

    basesrc->priv->last_sent_eos = FALSE;

    /* do initial seek, which will start the task */
    GST_OBJECT_LOCK (basesrc);
    event = basesrc->data.ABI.pending_seek;
    basesrc->data.ABI.pending_seek = NULL;
    GST_OBJECT_UNLOCK (basesrc);

    /* no need to unlock anything, the task is certainly
     * not running here. The perform seek code will start the task when
     * finished. */
    if (G_UNLIKELY (!gst_base_src_perform_seek (basesrc, event, FALSE)))
      goto seek_failed;

    if (event)
      gst_event_unref (event);
  } else {
    GST_DEBUG_OBJECT (basesrc, "Deactivating in push mode");
    /* call the unlock function and stop the task */
    if (G_UNLIKELY (!gst_base_src_deactivate (basesrc, pad)))
      goto deactivate_failed;

    /* now we can stop the source */
    if (G_UNLIKELY (!gst_base_src_stop (basesrc)))
      goto error_stop;
  }
  return TRUE;

  /* ERRORS */
no_push_activation:
  {
    GST_WARNING_OBJECT (basesrc, "Subclass disabled push-mode activation");
    return FALSE;
  }
error_start:
  {
    GST_WARNING_OBJECT (basesrc, "Failed to start in push mode");
    return FALSE;
  }
seek_failed:
  {
    GST_ERROR_OBJECT (basesrc, "Failed to perform initial seek");
    gst_base_src_stop (basesrc);
    if (event)
      gst_event_unref (event);
    return FALSE;
  }
deactivate_failed:
  {
    GST_ERROR_OBJECT (basesrc, "Failed to deactivate in push mode");
    return FALSE;
  }
error_stop:
  {
    GST_DEBUG_OBJECT (basesrc, "Failed to stop in push mode");
    return FALSE;
  }
}

static gboolean
gst_base_src_activate_pull (GstPad * pad, gboolean active)
{
  GstBaseSrc *basesrc;

  basesrc = GST_BASE_SRC (GST_OBJECT_PARENT (pad));

  /* prepare subclass first */
  if (active) {
    GST_DEBUG_OBJECT (basesrc, "Activating in pull mode");
    if (G_UNLIKELY (!gst_base_src_start (basesrc)))
      goto error_start;

    /* if not random_access, we cannot operate in pull mode for now */
    if (G_UNLIKELY (!gst_base_src_check_get_range (basesrc)))
      g