gstbasesink.c

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

          break;
      }
    }
  }

done:
  gst_base_sink_perform_qos (basesink, late);

  GST_DEBUG_OBJECT (basesink, "object unref after render %p", obj);
  gst_mini_object_unref (obj);

  return ret;

  /* ERRORS */
sync_failed:
  {
    GST_DEBUG_OBJECT (basesink, "do_sync returned %s", gst_flow_get_name (ret));
    goto done;
  }
dropped:
  {
    priv->dropped++;
    GST_DEBUG_OBJECT (basesink, "buffer late, dropping");
    goto done;
  }
}

/* with STREAM_LOCK, PREROLL_LOCK
 *
 * Perform preroll on the given object. For buffers this means 
 * calling the preroll subclass method. 
 * If that succeeds, the state will be commited.
 *
 * function does not take ownership of obj.
 */
static GstFlowReturn
gst_base_sink_preroll_object (GstBaseSink * basesink, GstPad * pad,
    GstMiniObject * obj)
{
  GstFlowReturn ret;

  GST_DEBUG_OBJECT (basesink, "do preroll %p", obj);

  /* if it's a buffer, we need to call the preroll method */
  if (G_LIKELY (GST_IS_BUFFER (obj))) {
    GstBaseSinkClass *bclass;
    GstBuffer *buf = GST_BUFFER_CAST (obj);

    GST_DEBUG_OBJECT (basesink, "preroll buffer %" GST_TIME_FORMAT,
        GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));

    bclass = GST_BASE_SINK_GET_CLASS (basesink);
    if (bclass->preroll)
      if ((ret = bclass->preroll (basesink, buf)) != GST_FLOW_OK)
        goto preroll_failed;
  }

  /* commit state */
  if (G_LIKELY (basesink->playing_async)) {
    if (G_UNLIKELY (!gst_base_sink_commit_state (basesink)))
      goto stopping;
  }

  return GST_FLOW_OK;

  /* ERRORS */
preroll_failed:
  {
    GST_DEBUG_OBJECT (basesink, "preroll failed, abort state");
    gst_element_abort_state (GST_ELEMENT_CAST (basesink));
    return ret;
  }
stopping:
  {
    GST_DEBUG_OBJECT (basesink, "stopping while commiting state");
    return GST_FLOW_WRONG_STATE;
  }
}

/* with STREAM_LOCK, PREROLL_LOCK 
 *
 * Queue an object for rendering.
 * The first prerollable object queued will complete the preroll. If the
 * preroll queue if filled, we render all the objects in the queue.
 *
 * This function takes ownership of the object.
 */
static GstFlowReturn
gst_base_sink_queue_object_unlocked (GstBaseSink * basesink, GstPad * pad,
    GstMiniObject * obj, gboolean prerollable)
{
  GstFlowReturn ret = GST_FLOW_OK;
  gint length;
  GQueue *q;

  if (G_UNLIKELY (basesink->priv->received_eos))
    goto was_eos;

  if (G_UNLIKELY (basesink->need_preroll)) {
    if (G_LIKELY (prerollable))
      basesink->preroll_queued++;

    length = basesink->preroll_queued;

    GST_DEBUG_OBJECT (basesink, "now %d prerolled items", length);

    /* first prerollable item needs to finish the preroll */
    if (length == 1) {
      ret = gst_base_sink_preroll_object (basesink, pad, obj);
      if (G_UNLIKELY (ret != GST_FLOW_OK))
        goto preroll_failed;
    }
    /* need to recheck if we need preroll, commmit state during preroll 
     * could have made us not need more preroll. */
    if (G_UNLIKELY (basesink->need_preroll)) {
      /* see if we can render now. */
      if (G_UNLIKELY (length <= basesink->preroll_queue_max_len))
        goto more_preroll;
    }
  }

  /* we can start rendering (or blocking) the queued object
   * if any. */
  q = basesink->preroll_queue;
  while (G_UNLIKELY (!g_queue_is_empty (q))) {
    GstMiniObject *o;

    o = g_queue_pop_head (q);
    GST_DEBUG_OBJECT (basesink, "rendering queued object %p", o);

    /* do something with the return value */
    ret = gst_base_sink_render_object (basesink, pad, o);
    if (ret != GST_FLOW_OK)
      goto dequeue_failed;
  }

  /* now render the object */
  ret = gst_base_sink_render_object (basesink, pad, obj);
  basesink->preroll_queued = 0;

  return ret;

  /* special cases */
was_eos:
  {
    GST_DEBUG_OBJECT (basesink,
        "we are EOS, dropping object, return UNEXPECTED");
    gst_mini_object_unref (obj);
    return GST_FLOW_UNEXPECTED;
  }
preroll_failed:
  {
    GST_DEBUG_OBJECT (basesink, "preroll failed, reason %s",
        gst_flow_get_name (ret));
    gst_mini_object_unref (obj);
    return ret;
  }
more_preroll:
  {
    /* add object to the queue and return */
    GST_DEBUG_OBJECT (basesink, "need more preroll data %d <= %d",
        length, basesink->preroll_queue_max_len);
    g_queue_push_tail (basesink->preroll_queue, obj);
    return GST_FLOW_OK;
  }
dequeue_failed:
  {
    GST_DEBUG_OBJECT (basesink, "rendering queued objects failed, reason %s",
        gst_flow_get_name (ret));
    gst_mini_object_unref (obj);
    return ret;
  }
}

/* with STREAM_LOCK
 *
 * This function grabs the PREROLL_LOCK and adds the object to
 * the queue.
 *
 * This function takes ownership of obj.
 */
static GstFlowReturn
gst_base_sink_queue_object (GstBaseSink * basesink, GstPad * pad,
    GstMiniObject * obj, gboolean prerollable)
{
  GstFlowReturn ret;

  GST_PAD_PREROLL_LOCK (pad);
  if (G_UNLIKELY (basesink->flushing))
    goto flushing;

  ret = gst_base_sink_queue_object_unlocked (basesink, pad, obj, prerollable);
  GST_PAD_PREROLL_UNLOCK (pad);

  return ret;

  /* ERRORS */
flushing:
  {
    GST_DEBUG_OBJECT (basesink, "sink is flushing");
    GST_PAD_PREROLL_UNLOCK (pad);
    gst_mini_object_unref (obj);
    return GST_FLOW_WRONG_STATE;
  }
}

static gboolean
gst_base_sink_event (GstPad * pad, GstEvent * event)
{
  GstBaseSink *basesink;
  gboolean result = TRUE;
  GstBaseSinkClass *bclass;

  basesink = GST_BASE_SINK (gst_pad_get_parent (pad));

  bclass = GST_BASE_SINK_GET_CLASS (basesink);

  GST_DEBUG_OBJECT (basesink, "event %p (%s)", event,
      GST_EVENT_TYPE_NAME (event));

  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_EOS:
    {
      GstFlowReturn ret;

      /* EOS is a prerollable object */
      ret =
          gst_base_sink_queue_object (basesink, pad,
          GST_MINI_OBJECT_CAST (event), TRUE);

      if (G_UNLIKELY (ret != GST_FLOW_OK))
        result = FALSE;
      else
        /* we are now EOS, and refuse more buffers */
        basesink->priv->received_eos = TRUE;
      break;
    }
    case GST_EVENT_NEWSEGMENT:
    {
      GstFlowReturn ret;

      GST_DEBUG_OBJECT (basesink, "newsegment %p", event);

      if (G_UNLIKELY (basesink->priv->received_eos)) {
        /* we can't accept anything when we are EOS */
        result = FALSE;
        gst_event_unref (event);
      } else {
        /* the new segment is a non prerollable item and does not block anything,
         * we need to configure the current clipping segment and insert the event 
         * in the queue to serialize it with the buffers for rendering. */
        gst_base_sink_configure_segment (basesink, pad, event,
            basesink->abidata.ABI.clip_segment);

        ret =
            gst_base_sink_queue_object (basesink, pad,
            GST_MINI_OBJECT_CAST (event), FALSE);
        if (G_UNLIKELY (ret != GST_FLOW_OK))
          result = FALSE;
        else
          basesink->have_newsegment = TRUE;
      }
      break;
    }
    case GST_EVENT_FLUSH_START:
      if (bclass->event)
        bclass->event (basesink, event);

      GST_DEBUG_OBJECT (basesink, "flush-start %p", event);

      /* make sure we are not blocked on the clock also clear any pending
       * eos state. */
      gst_base_sink_set_flushing (basesink, pad, TRUE);

      /* we grab the stream lock but that is not needed since setting the
       * sink to flushing would make sure no state commit is being done
       * anymore */
      GST_PAD_STREAM_LOCK (pad);
      gst_base_sink_reset_qos (basesink);
      /* and we need to commit our state again on the next
       * prerolled buffer */
      basesink->playing_async = TRUE;
      gst_element_lost_state (GST_ELEMENT_CAST (basesink));
      GST_PAD_STREAM_UNLOCK (pad);

      gst_event_unref (event);
      break;
    case GST_EVENT_FLUSH_STOP:
      if (bclass->event)
        bclass->event (basesink, event);

      GST_DEBUG_OBJECT (basesink, "flush-stop %p", event);

      /* unset flushing so we can accept new data */
      gst_base_sink_set_flushing (basesink, pad, FALSE);

      /* we need new segment info after the flush. */
      gst_segment_init (&basesink->segment, GST_FORMAT_UNDEFINED);
      gst_segment_init (basesink->abidata.ABI.clip_segment,
          GST_FORMAT_UNDEFINED);
      basesink->have_newsegment = FALSE;
      /* we flush out the EOS event as well now */
      basesink->priv->received_eos = FALSE;

      gst_event_unref (event);
      break;
    default:
      /* other events are sent to queue or subclass depending on if they
       * are serialized. */
      if (GST_EVENT_IS_SERIALIZED (event)) {
        gst_base_sink_queue_object (basesink, pad,
            GST_MINI_OBJECT_CAST (event), FALSE);
      } else {
        if (bclass->event)
          bclass->event (basesink, event);
        gst_event_unref (event);
      }
      break;
  }
  gst_object_unref (basesink);

  return result;
}

/* default implementation to calculate the start and end
 * timestamps on a buffer, subclasses can override
 */
static void
gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
    GstClockTime * start, GstClockTime * end)
{
  GstClockTime timestamp, duration;

  timestamp = GST_BUFFER_TIMESTAMP (buffer);
  if (GST_CLOCK_TIME_IS_VALID (timestamp)) {

    /* get duration to calculate end time */
    duration = GST_BUFFER_DURATION (buffer);
    if (GST_CLOCK_TIME_IS_VALID (duration)) {
      *end = timestamp + duration;
    }
    *start = timestamp;
  }
}

/* must be called with PREROLL_LOCK */
static gboolean
gst_base_sink_is_prerolled (GstBaseSink * basesink)
{
  gboolean res;

  res = basesink->have_preroll || basesink->eos;
  GST_DEBUG_OBJECT (basesink, "have_preroll: %d, EOS: %d => prerolled: %d",
      basesink->have_preroll, basesink->eos, res);
  return res;
}

/* with STREAM_LOCK, PREROLL_LOCK 
 *
 * Takes a buffer and compare the timestamps with the last segment.
 * If the buffer falls outside of the segment boundaries, drop it.
 * Else queue the buffer for preroll and rendering.
 *
 * This function takes ownership of the buffer.
 */
static GstFlowReturn
gst_base_sink_chain_unlocked (GstBaseSink * basesink, GstPad * pad,
    GstBuffer * buf)
{
  GstFlowReturn result;
  GstClockTime start = GST_CLOCK_TIME_NONE, end = GST_CLOCK_TIME_NONE;
  GstSegment *clip_segment;

  if (G_UNLIKELY (basesink->flushing))
    goto flushing;

  /* for code clarity */
  clip_segment = basesink->abidata.ABI.clip_segment;

  if (G_UNLIKELY (!basesink->have_newsegment)) {
    gboolean sync;

    sync = gst_base_sink_get_sync (basesink);
    if (sync) {
      GST_ELEMENT_WARNING (basesink, STREAM, FAILED,
          (_("Internal data flow problem.")),
          ("Received buffer without a new-segment. Assuming timestamps start from 0."));
    }

    basesink->have_newsegment = TRUE;
    /* this means this sink will assume timestamps start from 0 */
    clip_segment->start = 0;
    clip_segment->stop = -1;
    basesink->segment.start = 0;
    basesink->segment.stop = -1;
  }

  /* check if the buffer needs to be dropped */
  /* we don't use the subclassed method as it may not return
   * valid values for our purpose here */
  gst_base_sink_get_times (basesink, buf, &start, &end);

  GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
      ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));

  /* a dropped buffer does not participate in anything */
  if (GST_CLOCK_TIME_IS_VALID (start) &&
      (clip_segment->format == GST_FORMAT_TIME)) {
    if (G_UNLIKELY (!gst_segment_clip (clip_segment,
                GST_FORMAT_TIME, (gint64) start, (gint64) end, NULL, NULL)))
      goto out_of_segment;
  }

  /* now we can process the buffer in the queue, this function takes ownership
   * of the buffer */
  result = gst_base_sink_queue_object_unlocked (basesink, pad,
      GST_MINI_OBJECT_CAST (buf), TRUE);

  return result;

  /* ERRORS */
flushing:
  {
    GST_DEBUG_OBJECT (basesink, "sink is flushing");
    gst_buffer_unref (buf);
    return GST_FLOW_WRONG_STATE;
  }
out_of_segment:
  {
    GST_DEBUG_OBJECT (basesink, "dropping buffer, out of clipping segment");