gstbasesink.c

gnash 在pc和嵌入式下开发需要的源码

/**
 * gst_base_sink_query_latency:
 * @sink: the sink
 * @live: if the sink is live
 * @upstream_live: if an upstream element is live
 * @min_latency: the min latency of the upstream elements
 * @max_latency: the max latency of the upstream elements
 *
 * Query the sink for the latency parameters. The latency will be queried from
 * the upstream elements. @live will be TRUE if @sink is configured to
 * synchronize against the clock. @upstream_live will be TRUE if an upstream
 * element is live. 
 *
 * If both @live and @upstream_live are TRUE, the sink will want to compensate
 * for the latency introduced by the upstream elements by setting the
 * @min_latency to a strictly possitive value.
 *
 * This function is mostly used by subclasses. 
 *
 * Returns: TRUE if the query succeeded.
 *
 * Since: 0.10.12
 */
gboolean
gst_base_sink_query_latency (GstBaseSink * sink, gboolean * live,
    gboolean * upstream_live, GstClockTime * min_latency,
    GstClockTime * max_latency)
{
  gboolean l, us_live, res, have_latency;
  GstClockTime min, max;
  GstQuery *query;
  GstClockTime us_min, us_max;

  /* we are live when we sync to the clock */
  GST_OBJECT_LOCK (sink);
  l = sink->sync;
  have_latency = sink->priv->have_latency;
  GST_OBJECT_UNLOCK (sink);

  /* assume no latency */
  min = 0;
  max = -1;
  us_live = FALSE;

  /* we are live */
  if (l) {
    if (have_latency) {
      /* we are live and ready for a latency query */
      query = gst_query_new_latency ();

      /* ask the peer for the latency */
      if (!(res = gst_base_sink_peer_query (sink, query)))
        goto query_failed;

      /* get upstream min and max latency */
      gst_query_parse_latency (query, &us_live, &us_min, &us_max);
      gst_query_unref (query);

      if (us_live) {
        /* upstream live, use its latency, subclasses should use these
         * values to create the complete latency. */
        min = us_min;
        max = us_max;
      }
    } else {
      /* we are live but are not yet ready for a latency query */
      res = FALSE;
    }
  } else {
    /* not live, result is always TRUE */
    res = TRUE;
  }

  GST_DEBUG_OBJECT (sink, "latency query: live: %d, have_latency %d,"
      " upstream: %d, min %" GST_TIME_FORMAT ", max %" GST_TIME_FORMAT, l,
      have_latency, us_live, GST_TIME_ARGS (min), GST_TIME_ARGS (max));

  if (live)
    *live = l;
  if (upstream_live)
    *upstream_live = us_live;
  if (min_latency)
    *min_latency = min;
  if (max_latency)
    *max_latency = max;

  return res;

  /* ERRORS */
query_failed:
  {
    GST_DEBUG_OBJECT (sink, "latency query failed");
    gst_query_unref (query);
    return FALSE;
  }
}

static void
gst_base_sink_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstBaseSink *sink = GST_BASE_SINK (object);

  switch (prop_id) {
    case PROP_PREROLL_QUEUE_LEN:
      /* preroll lock necessary to serialize with finish_preroll */
      GST_PAD_PREROLL_LOCK (sink->sinkpad);
      sink->preroll_queue_max_len = g_value_get_uint (value);
      GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
      break;
    case PROP_SYNC:
      gst_base_sink_set_sync (sink, g_value_get_boolean (value));
      break;
    case PROP_MAX_LATENESS:
      gst_base_sink_set_max_lateness (sink, g_value_get_int64 (value));
      break;
    case PROP_QOS:
      gst_base_sink_set_qos_enabled (sink, g_value_get_boolean (value));
      break;
    case PROP_ASYNC:
      gst_base_sink_set_async_enabled (sink, g_value_get_boolean (value));
      break;
    case PROP_TS_OFFSET:
      gst_base_sink_set_ts_offset (sink, g_value_get_int64 (value));
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static void
gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
    GParamSpec * pspec)
{
  GstBaseSink *sink = GST_BASE_SINK (object);

  switch (prop_id) {
    case PROP_PREROLL_QUEUE_LEN:
      GST_PAD_PREROLL_LOCK (sink->sinkpad);
      g_value_set_uint (value, sink->preroll_queue_max_len);
      GST_PAD_PREROLL_UNLOCK (sink->sinkpad);
      break;
    case PROP_SYNC:
      g_value_set_boolean (value, gst_base_sink_get_sync (sink));
      break;
    case PROP_MAX_LATENESS:
      g_value_set_int64 (value, gst_base_sink_get_max_lateness (sink));
      break;
    case PROP_QOS:
      g_value_set_boolean (value, gst_base_sink_is_qos_enabled (sink));
      break;
    case PROP_ASYNC:
      g_value_set_boolean (value, gst_base_sink_is_async_enabled (sink));
      break;
    case PROP_TS_OFFSET:
      g_value_set_int64 (value, gst_base_sink_get_ts_offset (sink));
      break;
    case PROP_LAST_BUFFER:
      gst_value_take_buffer (value, gst_base_sink_get_last_buffer (sink));
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}


static GstCaps *
gst_base_sink_get_caps (GstBaseSink * sink)
{
  return NULL;
}

static gboolean
gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
{
  return TRUE;
}

static GstFlowReturn
gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
    GstCaps * caps, GstBuffer ** buf)
{
  *buf = NULL;
  return GST_FLOW_OK;
}

/* with PREROLL_LOCK, STREAM_LOCK */
static void
gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
{
  GstMiniObject *obj;

  GST_DEBUG_OBJECT (basesink, "flushing queue %p", basesink);
  while ((obj = g_queue_pop_head (basesink->preroll_queue))) {
    GST_DEBUG_OBJECT (basesink, "popped %p", obj);
    gst_mini_object_unref (obj);
  }
  /* we can't have EOS anymore now */
  basesink->eos = FALSE;
  basesink->priv->received_eos = FALSE;
  basesink->have_preroll = FALSE;
  basesink->eos_queued = FALSE;
  basesink->preroll_queued = 0;
  basesink->buffers_queued = 0;
  basesink->events_queued = 0;
  /* can't report latency anymore until we preroll again */
  if (basesink->priv->async_enabled) {
    GST_OBJECT_LOCK (basesink);
    basesink->priv->have_latency = FALSE;
    GST_OBJECT_UNLOCK (basesink);
  }
  /* and signal any waiters now */
  GST_PAD_PREROLL_SIGNAL (pad);
}

/* with STREAM_LOCK, configures given segment with the event information. */
static void
gst_base_sink_configure_segment (GstBaseSink * basesink, GstPad * pad,
    GstEvent * event, GstSegment * segment)
{
  gboolean update;
  gdouble rate, arate;
  GstFormat format;
  gint64 start;
  gint64 stop;
  gint64 time;

  /* the newsegment event is needed to bring the buffer timestamps to the
   * stream time and to drop samples outside of the playback segment. */
  gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format,
      &start, &stop, &time);

  /* The segment is protected with both the STREAM_LOCK and the OBJECT_LOCK.
   * We protect with the OBJECT_LOCK so that we can use the values to
   * safely answer a POSITION query. */
  GST_OBJECT_LOCK (basesink);
  gst_segment_set_newsegment_full (segment, update, rate, arate, format, start,
      stop, time);

  if (format == GST_FORMAT_TIME) {
    GST_DEBUG_OBJECT (basesink,
        "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
        "format GST_FORMAT_TIME, "
        "%" GST_TIME_FORMAT " -- %" GST_TIME_FORMAT
        ", time %" GST_TIME_FORMAT ", accum %" GST_TIME_FORMAT,
        update, rate, arate, GST_TIME_ARGS (segment->start),
        GST_TIME_ARGS (segment->stop), GST_TIME_ARGS (segment->time),
        GST_TIME_ARGS (segment->accum));
  } else {
    GST_DEBUG_OBJECT (basesink,
        "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
        "format %d, "
        "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
        G_GINT64_FORMAT ", accum %" G_GINT64_FORMAT, update, rate, arate,
        segment->format, segment->start, segment->stop, segment->time,
        segment->accum);
  }
  GST_OBJECT_UNLOCK (basesink);
}

/* with PREROLL_LOCK, STREAM_LOCK */
static gboolean
gst_base_sink_commit_state (GstBaseSink * basesink)
{
  /* commit state and proceed to next pending state */
  GstState current, next, pending, post_pending;
  gboolean post_paused = FALSE;
  gboolean post_async_done = FALSE;
  gboolean post_playing = FALSE;
  gboolean sync;

  /* we are certainly not playing async anymore now */
  basesink->playing_async = FALSE;

  GST_OBJECT_LOCK (basesink);
  current = GST_STATE (basesink);
  next = GST_STATE_NEXT (basesink);
  pending = GST_STATE_PENDING (basesink);
  post_pending = pending;
  sync = basesink->sync;

  switch (pending) {
    case GST_STATE_PLAYING:
    {
      GstBaseSinkClass *bclass;
      GstStateChangeReturn ret;

      bclass = GST_BASE_SINK_GET_CLASS (basesink);

      GST_DEBUG_OBJECT (basesink, "commiting state to PLAYING");

      basesink->need_preroll = FALSE;
      post_async_done = TRUE;
      basesink->priv->commited = TRUE;
      post_playing = TRUE;
      /* post PAUSED too when we were READY */
      if (current == GST_STATE_READY) {
        post_paused = TRUE;
      }

      /* make sure we notify the subclass of async playing */
      if (bclass->async_play) {
        ret = bclass->async_play (basesink);
        if (ret == GST_STATE_CHANGE_FAILURE)
          goto async_failed;
      }
      break;
    }
    case GST_STATE_PAUSED:
      GST_DEBUG_OBJECT (basesink, "commiting state to PAUSED");
      post_paused = TRUE;
      post_async_done = TRUE;
      basesink->priv->commited = TRUE;
      post_pending = GST_STATE_VOID_PENDING;
      break;
    case GST_STATE_READY:
    case GST_STATE_NULL:
      goto stopping;
    case GST_STATE_VOID_PENDING:
      goto nothing_pending;
    default:
      break;
  }

  /* we can report latency queries now */
  basesink->priv->have_latency = TRUE;

  GST_STATE (basesink) = pending;
  GST_STATE_NEXT (basesink) = GST_STATE_VOID_PENDING;
  GST_STATE_PENDING (basesink) = GST_STATE_VOID_PENDING;
  GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_SUCCESS;
  GST_OBJECT_UNLOCK (basesink);

  if (post_paused) {
    GST_DEBUG_OBJECT (basesink, "posting PAUSED state change message");
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
            current, next, post_pending));
  }
  if (post_async_done) {
    GST_DEBUG_OBJECT (basesink, "posting async-done message");
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_async_done (GST_OBJECT_CAST (basesink)));
  }
  if (post_playing) {
    GST_DEBUG_OBJECT (basesink, "posting PLAYING state change message");
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
            next, pending, GST_STATE_VOID_PENDING));
  }

  GST_STATE_BROADCAST (basesink);

  return TRUE;

nothing_pending:
  {
    /* Depending on the state, set our vars. We get in this situation when the
     * state change function got a change to update the state vars before the
     * streaming thread did. This is fine but we need to make sure that we
     * update the need_preroll var since it was TRUE when we got here and might
     * become FALSE if we got to PLAYING. */
    GST_DEBUG_OBJECT (basesink, "nothing to commit, now in %s",
        gst_element_state_get_name (current));
    switch (current) {
      case GST_STATE_PLAYING:
        basesink->need_preroll = FALSE;
        break;
      case GST_STATE_PAUSED:
        basesink->need_preroll = TRUE;
        break;
      default:
        basesink->need_preroll = FALSE;
        basesink->flushing = TRUE;
        break;
    }
    /* we can report latency queries now */
    basesink->priv->have_latency = TRUE;
    GST_OBJECT_UNLOCK (basesink);
    return TRUE;
  }
stopping:
  {
    /* app is going to READY */
    GST_DEBUG_OBJECT (basesink, "stopping");
    basesink->need_preroll = FALSE;
    basesink->flushing = TRUE;
    GST_OBJECT_UNLOCK (basesink);
    return FALSE;
  }
async_failed:
  {
    GST_DEBUG_OBJECT (basesink, "async commit failed");
    GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_FAILURE;
    GST_OBJECT_UNLOCK (basesink);
    return FALSE;
  }
}


/* with STREAM_LOCK, PREROLL_LOCK
 *
 * Returns TRUE if the object needs synchronisation and takes therefore
 * part in prerolling.
 *
 * rsstart/rsstop contain the start/stop in stream time.
 * rrstart/rrstop contain the start/stop in running time.
 */
static gboolean
gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
    GstClockTime * rsstart, GstClockTime * rsstop,
    GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync,
    GstSegment * segment)
{
  GstBaseSinkClass *bclass;
  GstBuffer *buffer;
  GstClockTime start, stop;     /* raw start/stop timestamps */
  gint64 cstart, cstop;         /* clipped raw timestamps */
  gint64 rstart, rstop;         /* clipped timestamps converted to running time */
  GstClockTime sstart, sstop;   /* clipped timestamps converted to stream time */
  GstFormat format;
  GstBaseSinkPrivate *priv;

  priv = basesink->priv;

  /* start with nothing */
  start = stop = sstart = sstop = rstart = rstop = -1;

  if (G_UNLIKELY (GST_IS_EVENT (obj))) {
    GstEvent *event = GST_EVENT_CAST (obj);

    switch (GST_EVENT_TYPE (event)) {
        /* EOS event needs syncing */
      case GST_EVENT_EOS: