static gboolean
_set_duration (GESTimelineElement * element, GstClockTime duration)
{
  GESTrackElement *object = GES_TRACK_ELEMENT (element);
  GESTrackElementPrivate *priv = object->priv;

  if (GST_CLOCK_TIME_IS_VALID (_MAXDURATION (element)) &&
      duration > _INPOINT (object) + _MAXDURATION (element))
    duration = _MAXDURATION (element) - _INPOINT (object);

  if (priv->gnlobject != NULL) {
    if (G_UNLIKELY (duration == _DURATION (object)))
      return FALSE;

    g_object_set (priv->gnlobject, "duration", duration, NULL);
  } else
    priv->pending_duration = duration;

  _update_control_bindings (element, ges_timeline_element_get_inpoint (element),
      duration);

  return TRUE;
}

void AudioTestSource_i::_new_gst_buffer(GstElement *sink, AudioTestSource_i* comp) {
	static GstBuffer *buffer;
	static std::vector<short> packet;

    /* Retrieve the buffer */
    g_signal_emit_by_name (sink, "pull-buffer", &buffer);
    if (buffer) {
    	BULKIO::PrecisionUTCTime T;

	    /* The only thing we do in this example is print a * to indicate a received buffer */
    	if (GST_CLOCK_TIME_IS_VALID(buffer->timestamp)) {
    		T = _from_gst_timestamp(buffer->timestamp);
    	} else {
    		T = _now();
    	}

    	packet.resize(buffer->size / 2); // TODO the division should come from reading buffer->caps
    	memcpy(&packet[0], buffer->data, buffer->size);

    	comp->audio_out->pushPacket(packet, T, false, comp->stream_id);
	    gst_buffer_unref (buffer);
    }
}

/* this function does the actual processing
 */
static GstFlowReturn
gst_audio_panorama_transform (GstBaseTransform * base, GstBuffer * inbuf,
    GstBuffer * outbuf)
{
  GstAudioPanorama *filter = GST_AUDIO_PANORAMA (base);
  GstClockTime timestamp, stream_time;
  GstMapInfo inmap, outmap;

  timestamp = GST_BUFFER_TIMESTAMP (inbuf);
  stream_time =
      gst_segment_to_stream_time (&base->segment, GST_FORMAT_TIME, timestamp);

  GST_DEBUG_OBJECT (filter, "sync to %" GST_TIME_FORMAT,
      GST_TIME_ARGS (timestamp));

  if (GST_CLOCK_TIME_IS_VALID (stream_time))
    gst_object_sync_values (GST_OBJECT (filter), stream_time);

  gst_buffer_map (inbuf, &inmap, GST_MAP_READ);
  gst_buffer_map (outbuf, &outmap, GST_MAP_WRITE);

  if (G_UNLIKELY (GST_BUFFER_FLAG_IS_SET (inbuf, GST_BUFFER_FLAG_GAP))) {
    GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_GAP);
    memset (outmap.data, 0, outmap.size);
  } else {
    /* output always stereo, input mono or stereo,
     * and info describes input format */
    guint num_samples = outmap.size / (2 * GST_AUDIO_INFO_BPS (&filter->info));

    filter->process (filter, inmap.data, outmap.data, num_samples);
  }

  gst_buffer_unmap (inbuf, &inmap);
  gst_buffer_unmap (outbuf, &outmap);

  return GST_FLOW_OK;
}

static GstFlowReturn
gst_mpegv_parse_parse_frame (GstBaseParse * parse, GstBaseParseFrame * frame)
{
  GstMpegvParse *mpvparse = GST_MPEGVIDEO_PARSE (parse);
  GstBuffer *buffer = frame->buffer;

  if (G_UNLIKELY (mpvparse->pichdr.pic_type == GST_MPEG_VIDEO_PICTURE_TYPE_I))
    GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
  else
    GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);

  /* maybe only sequence in this buffer, though not recommended,
   * so mark it as such and force 0 duration */
  if (G_UNLIKELY (mpvparse->pic_offset < 0)) {
    GST_DEBUG_OBJECT (mpvparse, "frame holds no picture data");
    frame->flags |= GST_BASE_PARSE_FRAME_FLAG_NO_FRAME;
    GST_BUFFER_DURATION (buffer) = 0;
  }

  if (mpvparse->pic_offset > 4) {
    gst_base_parse_set_ts_at_offset (parse, mpvparse->pic_offset - 4);
  }

  if (mpvparse->frame_repeat_count
      && GST_CLOCK_TIME_IS_VALID (GST_BUFFER_DURATION (buffer))) {
    GST_BUFFER_DURATION (buffer) =
        (1 + mpvparse->frame_repeat_count) * GST_BUFFER_DURATION (buffer) / 2;
  }

  if (G_UNLIKELY (mpvparse->drop && !mpvparse->config)) {
    GST_DEBUG_OBJECT (mpvparse, "dropping frame as no config yet");
    return GST_BASE_PARSE_FLOW_DROPPED;
  }

  gst_mpegv_parse_update_src_caps (mpvparse);
  return GST_FLOW_OK;
}

/* Actual processing. */
static GstFlowReturn
gst_burn_transform_frame (GstVideoFilter * vfilter,
    GstVideoFrame * in_frame, GstVideoFrame * out_frame)
{
  GstBurn *filter = GST_BURN (vfilter);
  gint video_size, adjustment;
  guint32 *src, *dest;
  GstClockTime timestamp;
  gint64 stream_time;

  src = GST_VIDEO_FRAME_PLANE_DATA (in_frame, 0);
  dest = GST_VIDEO_FRAME_PLANE_DATA (out_frame, 0);

  video_size = GST_VIDEO_FRAME_WIDTH (in_frame) *
      GST_VIDEO_FRAME_HEIGHT (in_frame);

  /* GstController: update the properties */
  timestamp = GST_BUFFER_TIMESTAMP (in_frame->buffer);
  stream_time =
      gst_segment_to_stream_time (&GST_BASE_TRANSFORM (filter)->segment,
      GST_FORMAT_TIME, timestamp);

  GST_DEBUG_OBJECT (filter, "sync to %" GST_TIME_FORMAT,
      GST_TIME_ARGS (timestamp));

  if (GST_CLOCK_TIME_IS_VALID (stream_time))
    gst_object_sync_values (GST_OBJECT (filter), stream_time);

  GST_OBJECT_LOCK (filter);
  adjustment = filter->adjustment;
  GST_OBJECT_UNLOCK (filter);

  /*** Now the image processing work.... ***/
  gaudi_orc_burn (dest, src, adjustment, video_size);

  return GST_FLOW_OK;
}

static GstFlowReturn
gst_rtp_mpv_pay_handle_buffer (GstBaseRTPPayload * basepayload,
    GstBuffer * buffer)
{
  GstRTPMPVPay *rtpmpvpay;
  guint avail, packet_len;
  GstClockTime timestamp, duration;
  GstFlowReturn ret;

  rtpmpvpay = GST_RTP_MPV_PAY (basepayload);

  timestamp = GST_BUFFER_TIMESTAMP (buffer);
  duration = GST_BUFFER_DURATION (buffer);

  gst_adapter_push (rtpmpvpay->adapter, buffer);
  avail = gst_adapter_available (rtpmpvpay->adapter);

  /* Initialize new RTP payload */
  if (avail == 0) {
    rtpmpvpay->first_ts = timestamp;
    rtpmpvpay->duration = duration;
  }

  /* get packet length of previous data and this new data,
   * payload length includes a 4 byte MPEG video-specific header */
  packet_len = gst_rtp_buffer_calc_packet_len (4 + avail, 0, 0);

  if (gst_basertppayload_is_filled (basepayload,
          packet_len, rtpmpvpay->duration + duration)) {
    ret = gst_rtp_mpv_pay_flush (rtpmpvpay, timestamp, duration);
  } else {
    if (GST_CLOCK_TIME_IS_VALID (duration))
      rtpmpvpay->duration += duration;
    ret = GST_FLOW_OK;
  }
  return ret;
}

/**
 * Utility function to handle transferring Gstreamer timestamp to OMX
 * timestamp.  This function handles discontinuities and timestamp
 * renormalization.
 *
 * @omx_buffer the destination OMX buffer for the timestamp
 * @buffer     the source Gstreamer buffer for the timestamp
 * @normalize  should this buffer be the one that we renormalize on
 *   (iff normalization is required)?  (ie. with TI OMX, you should
 *   only re-normalize on a video buffer)
 */
gboolean
gst_goo_timestamp_gst2omx (
		OMX_BUFFERHEADERTYPE* omx_buffer,
		GstBuffer* buffer,
		gboolean normalize)
{
	GstClockTime timestamp = GST_BUFFER_TIMESTAMP (buffer);

	if (GST_GOO_UTIL_IS_DISCONT (buffer))
	{
		needs_normalization = TRUE;
		GST_DEBUG ("needs_normalization");
	}

	if (needs_normalization && normalize)
	{
		GST_INFO ("Setting OMX_BUFFER_STARTTIME..");
		omx_buffer->nFlags |= OMX_BUFFERFLAG_STARTTIME;
		omx_normalize_timestamp = GST2OMX_TIMESTAMP ((gint64)timestamp);
		needs_normalization = FALSE;
		GST_DEBUG ("omx_normalize_timestamp=%lld", omx_normalize_timestamp);
	}

	/* transfer timestamp to openmax */
	if (GST_CLOCK_TIME_IS_VALID (timestamp))
	{
		omx_buffer->nTimeStamp = GST2OMX_TIMESTAMP ((gint64)timestamp) - omx_normalize_timestamp;
		GST_INFO ("OMX timestamp = %lld (%lld - %lld)", omx_buffer->nTimeStamp, GST2OMX_TIMESTAMP ((gint64)timestamp), omx_normalize_timestamp);
		return TRUE;
	}
	else
	{
		GST_WARNING ("Invalid timestamp!");
		return FALSE;
	}
}

/**
 * gst_timed_value_control_source_unset:
 * @self: the #GstTimedValueControlSource object
 * @timestamp: the time the control-change should be removed from
 *
 * Used to remove the value of given controller-handled property at a certain
 * time.
 *
 * Returns: FALSE if the value couldn't be unset (i.e. not found, TRUE otherwise.
 */
gboolean
gst_timed_value_control_source_unset (GstTimedValueControlSource * self,
    GstClockTime timestamp)
{
  GSequenceIter *iter;
  gboolean res = FALSE;
  GstControlPoint *cp = NULL;

  g_return_val_if_fail (GST_IS_TIMED_VALUE_CONTROL_SOURCE (self), FALSE);
  g_return_val_if_fail (GST_CLOCK_TIME_IS_VALID (timestamp), FALSE);

  g_mutex_lock (&self->lock);
  /* check if a control point for the timestamp exists */
  if (G_LIKELY (self->values) && (iter =
          g_sequence_lookup (self->values, &timestamp,
              (GCompareDataFunc) gst_control_point_find, NULL))) {

    /* Iter contains the iter right after timestamp, i.e.
     * we need to get the previous one and check the timestamp
     */
    cp = g_slice_dup (GstControlPoint, g_sequence_get (iter));
    g_sequence_remove (iter);
    self->nvalues--;
    self->valid_cache = FALSE;
    res = TRUE;
  }
  g_mutex_unlock (&self->lock);

  if (cp) {
    g_signal_emit (self,
        gst_timed_value_control_source_signals[VALUE_REMOVED_SIGNAL], 0, cp);
    g_slice_free (GstControlPoint, cp);
  }

  return res;
}

static GValue *
gst_direct_control_binding_get_value (GstControlBinding * _self,
    GstClockTime timestamp)
{
  GstDirectControlBinding *self = GST_DIRECT_CONTROL_BINDING (_self);
  GValue *dst_val = NULL;
  gdouble src_val;

  g_return_val_if_fail (GST_IS_DIRECT_CONTROL_BINDING (self), NULL);
  g_return_val_if_fail (GST_CLOCK_TIME_IS_VALID (timestamp), NULL);
  g_return_val_if_fail (GST_CONTROL_BINDING_PSPEC (self), FALSE);

  /* get current value via control source */
  if (gst_control_source_get_value (self->cs, timestamp, &src_val)) {
    dst_val = g_new0 (GValue, 1);
    g_value_init (dst_val, G_PARAM_SPEC_VALUE_TYPE (_self->pspec));
    self->convert_g_value (self, src_val, dst_val);
  } else {
    GST_LOG ("no control value for property %s at ts %" GST_TIME_FORMAT,
        _self->name, GST_TIME_ARGS (timestamp));
  }

  return dst_val;
}

static GstFlowReturn
gst_frame_positionner_transform_ip (GstBaseTransform * trans, GstBuffer * buf)
{
  GstFramePositionnerMeta *meta;
  GstFramePositionner *framepositionner = GST_FRAME_POSITIONNER (trans);
  GstClockTime timestamp = GST_BUFFER_TIMESTAMP (buf);

  if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
    gst_object_sync_values (GST_OBJECT (trans), timestamp);
  }

  meta =
      (GstFramePositionnerMeta *) gst_buffer_add_meta (buf,
      gst_frame_positionner_get_info (), NULL);

  GST_OBJECT_LOCK (framepositionner);
  meta->alpha = framepositionner->alpha;
  meta->posx = framepositionner->posx;
  meta->posy = framepositionner->posy;
  meta->zorder = framepositionner->zorder;
  GST_OBJECT_UNLOCK (framepositionner);

  return GST_FLOW_OK;
}

/**
 * ges_layer_add_asset:
 * @layer: a #GESLayer
 * @asset: The asset to add to
 * @start: The start value to set on the new #GESClip
 * @inpoint: The inpoint value to set on the new #GESClip
 * @duration: The duration value to set on the new #GESClip
 * @track_types: The #GESTrackType to set on the the new #GESClip
 *
 * Creates Clip from asset, adds it to layer and
 * returns a reference to it.
 *
 * Returns: (transfer none): Created #GESClip
 */
GESClip *
ges_layer_add_asset (GESLayer * layer,
    GESAsset * asset, GstClockTime start, GstClockTime inpoint,
    GstClockTime duration, GESTrackType track_types)
{
  GESClip *clip;

  g_return_val_if_fail (GES_IS_LAYER (layer), NULL);
  g_return_val_if_fail (GES_IS_ASSET (asset), NULL);
  g_return_val_if_fail (g_type_is_a (ges_asset_get_extractable_type
          (asset), GES_TYPE_CLIP), NULL);

  GST_DEBUG_OBJECT (layer, "Adding asset %s with: start: %" GST_TIME_FORMAT
      " inpoint: %" GST_TIME_FORMAT " duration: %" GST_TIME_FORMAT
      " track types: %d (%s)", ges_asset_get_id (asset), GST_TIME_ARGS (start),
      GST_TIME_ARGS (inpoint), GST_TIME_ARGS (duration), track_types,
      ges_track_type_name (track_types));

  clip = GES_CLIP (ges_asset_extract (asset, NULL));
  _set_start0 (GES_TIMELINE_ELEMENT (clip), start);
  _set_inpoint0 (GES_TIMELINE_ELEMENT (clip), inpoint);
  if (track_types != GES_TRACK_TYPE_UNKNOWN)
    ges_clip_set_supported_formats (clip, track_types);

  if (GST_CLOCK_TIME_IS_VALID (duration)) {
    _set_duration0 (GES_TIMELINE_ELEMENT (clip), duration);
  }

  if (!ges_layer_add_clip (layer, clip)) {
    gst_object_unref (clip);

    return NULL;
  }

  return clip;
}

/* GstBaseTransform vmethod implementations */
static GstFlowReturn
gst_audio_fx_base_iir_filter_transform_ip (GstBaseTransform * base,
    GstBuffer * buf)
{
  GstAudioFXBaseIIRFilter *filter = GST_AUDIO_FX_BASE_IIR_FILTER (base);
  guint num_samples;
  GstClockTime timestamp, stream_time;
  GstMapInfo map;

  timestamp = GST_BUFFER_TIMESTAMP (buf);
  stream_time =
      gst_segment_to_stream_time (&base->segment, GST_FORMAT_TIME, timestamp);

  GST_DEBUG_OBJECT (filter, "sync to %" GST_TIME_FORMAT,
      GST_TIME_ARGS (timestamp));

  if (GST_CLOCK_TIME_IS_VALID (stream_time))
    gst_object_sync_values (GST_OBJECT (filter), stream_time);

  gst_buffer_map (buf, &map, GST_MAP_READWRITE);
  num_samples = map.size / GST_AUDIO_FILTER_BPS (filter);

  g_mutex_lock (&filter->lock);
  if (filter->a == NULL || filter->b == NULL) {
    g_warn_if_fail (filter->a != NULL && filter->b != NULL);
    gst_buffer_unmap (buf, &map);
    g_mutex_unlock (&filter->lock);
    return GST_FLOW_ERROR;
  }
  filter->process (filter, map.data, num_samples);
  g_mutex_unlock (&filter->lock);

  gst_buffer_unmap (buf, &map);

  return GST_FLOW_OK;
}

static gboolean
gst_v4l2_video_dec_decide_allocation (GstVideoDecoder * decoder,
    GstQuery * query)
{
  GstV4l2VideoDec *self = GST_V4L2_VIDEO_DEC (decoder);
  GstClockTime latency;
  gboolean ret = FALSE;

  if (gst_v4l2_object_decide_allocation (self->v4l2capture, query))
    ret = GST_VIDEO_DECODER_CLASS (parent_class)->decide_allocation (decoder,
        query);

  if (GST_CLOCK_TIME_IS_VALID (self->v4l2capture->duration)) {
    latency = self->v4l2capture->min_buffers * self->v4l2capture->duration;
    GST_DEBUG_OBJECT (self, "Setting latency: %" GST_TIME_FORMAT " (%"
        G_GUINT32_FORMAT " * %" G_GUINT64_FORMAT, GST_TIME_ARGS (latency),
        self->v4l2capture->min_buffers, self->v4l2capture->duration);
    gst_video_decoder_set_latency (decoder, latency, latency);
  } else {
    GST_WARNING_OBJECT (self, "Duration invalid, not setting latency");
  }

  return ret;
}

static inline gboolean
ges_track_object_set_duration_internal (GESTrackObject * object,
    guint64 duration)
{
  GESTrackObjectPrivate *priv = object->priv;

  GST_DEBUG ("object:%p, duration:%" GST_TIME_FORMAT,
      object, GST_TIME_ARGS (duration));

  if (GST_CLOCK_TIME_IS_VALID (priv->maxduration) &&
      duration > object->inpoint + priv->maxduration)
    duration = priv->maxduration - object->inpoint;

  if (priv->gnlobject != NULL) {
    if (G_UNLIKELY (duration == object->duration))
      return FALSE;

    g_object_set (priv->gnlobject, "duration", duration,
        "media-duration", duration, NULL);
  } else
    priv->pending_duration = duration;

  return TRUE;
}

//.........这里部分代码省略.........
    if (gst_adapter_available (mimdec->adapter) < payload_size + 24)
      return GST_FLOW_OK;

    /* We have a whole packet and have read the header, lets flush it out */
    gst_adapter_flush (mimdec->adapter, 24);

    frame_body = gst_adapter_map (mimdec->adapter, payload_size);

    if (mimdec->buffer_size < 0) {
      /* Check if its a keyframe, otherwise skip it */
      if (GUINT32_FROM_LE (*((guint32 *) (frame_body + 12))) != 0) {
        gst_adapter_unmap (mimdec->adapter);
        gst_adapter_flush (mimdec->adapter, payload_size);
        return GST_FLOW_OK;
      }

      if (!mimic_decoder_init (mimdec->dec, frame_body)) {
        gst_adapter_unmap (mimdec->adapter);
        gst_adapter_flush (mimdec->adapter, payload_size);
        GST_ELEMENT_ERROR (mimdec, LIBRARY, INIT, (NULL),
            ("mimic_decoder_init error"));
        return GST_FLOW_ERROR;
      }

      if (!mimic_get_property (mimdec->dec, "buffer_size",
              &mimdec->buffer_size)) {
        gst_adapter_unmap (mimdec->adapter);
        gst_adapter_flush (mimdec->adapter, payload_size);
        GST_ELEMENT_ERROR (mimdec, LIBRARY, INIT, (NULL),
            ("mimic_get_property('buffer_size') error"));
        return GST_FLOW_ERROR;
      }

      mimic_get_property (mimdec->dec, "width", &width);
      mimic_get_property (mimdec->dec, "height", &height);
      GST_DEBUG_OBJECT (mimdec,
          "Initialised decoder with %d x %d payload size %d buffer_size %d",
          width, height, payload_size, mimdec->buffer_size);
      caps = gst_caps_new_simple ("video/x-raw",
          "format", G_TYPE_STRING, "RGB",
          "framerate", GST_TYPE_FRACTION, 0, 1,
          "width", G_TYPE_INT, width, "height", G_TYPE_INT, height, NULL);
      gst_pad_set_caps (mimdec->srcpad, caps);
      gst_caps_unref (caps);
    }


    if (mimdec->need_segment) {
      GstSegment segment;

      gst_segment_init (&segment, GST_FORMAT_TIME);

      if (GST_CLOCK_TIME_IS_VALID (in_time))
        segment.start = in_time;
      else
        segment.start = current_ts * GST_MSECOND;
      event = gst_event_new_segment (&segment);
    }
    mimdec->need_segment = FALSE;

    if (event)
      result = gst_pad_push_event (mimdec->srcpad, event);
    event = NULL;

    if (!result) {
      GST_WARNING_OBJECT (mimdec, "gst_pad_push_event failed");
      return GST_FLOW_ERROR;
    }


    out_buf = gst_buffer_new_allocate (NULL, mimdec->buffer_size, NULL);
    gst_buffer_map (out_buf, &map, GST_MAP_READWRITE);

    if (!mimic_decode_frame (mimdec->dec, frame_body, map.data)) {
      GST_WARNING_OBJECT (mimdec, "mimic_decode_frame error\n");

      gst_adapter_flush (mimdec->adapter, payload_size);

      gst_buffer_unmap (out_buf, &map);
      gst_buffer_unref (out_buf);
      GST_ELEMENT_ERROR (mimdec, STREAM, DECODE, (NULL),
          ("mimic_decode_frame error"));
      return GST_FLOW_ERROR;
    }
    gst_buffer_unmap (out_buf, &map);
    gst_adapter_flush (mimdec->adapter, payload_size);

    if (GST_CLOCK_TIME_IS_VALID (in_time))
      GST_BUFFER_TIMESTAMP (out_buf) = in_time;
    else
      GST_BUFFER_TIMESTAMP (out_buf) = current_ts * GST_MSECOND;

    res = gst_pad_push (mimdec->srcpad, out_buf);

    if (res != GST_FLOW_OK)
      break;
  }

  return res;
}

/**
 * @brief Chain function, this function does the actual processing.
 */
static GstFlowReturn
gst_tensor_aggregator_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
{
  GstTensorAggregator *self;
  GstFlowReturn ret = GST_FLOW_OK;
  GstAdapter *adapter;
  gsize avail, buf_size, frame_size, out_size;
  guint frames_in, frames_out, frames_flush;
  GstClockTime duration;

  self = GST_TENSOR_AGGREGATOR (parent);
  g_assert (self->tensor_configured);

  buf_size = gst_buffer_get_size (buf);
  g_return_val_if_fail (buf_size > 0, GST_FLOW_ERROR);

  frames_in = self->frames_in;
  frames_out = self->frames_out;
  frames_flush = self->frames_flush;
  frame_size = buf_size / frames_in;

  if (frames_in == frames_out) {
    /** push the incoming buffer (do concat if needed) */
    return gst_tensor_aggregator_push (self, buf, frame_size);
  }

  adapter = self->adapter;
  g_assert (adapter != NULL);

  duration = GST_BUFFER_DURATION (buf);
  if (GST_CLOCK_TIME_IS_VALID (duration)) {
    /** supposed same duration for incoming buffer */
    duration = gst_util_uint64_scale_int (duration, frames_out, frames_in);
  }

  gst_adapter_push (adapter, buf);

  out_size = frame_size * frames_out;
  g_assert (out_size > 0);

  while ((avail = gst_adapter_available (adapter)) >= out_size &&
      ret == GST_FLOW_OK) {
    GstBuffer *outbuf;
    GstClockTime pts, dts;
    guint64 pts_dist, dts_dist;
    gsize flush;

    pts = gst_adapter_prev_pts (adapter, &pts_dist);
    dts = gst_adapter_prev_dts (adapter, &dts_dist);

    /**
     * Update timestamp.
     * If frames-in is larger then frames-out, the same timestamp (pts and dts) would be returned.
     */
    if (frames_in > 1) {
      gint fn, fd;

      fn = self->in_config.rate_n;
      fd = self->in_config.rate_d;

      if (fn > 0 && fd > 0) {
        if (GST_CLOCK_TIME_IS_VALID (pts)) {
          pts +=
              gst_util_uint64_scale_int (pts_dist * fd, GST_SECOND,
              fn * frame_size);
        }

        if (GST_CLOCK_TIME_IS_VALID (dts)) {
          dts +=
              gst_util_uint64_scale_int (dts_dist * fd, GST_SECOND,
              fn * frame_size);
        }
      }
    }

    outbuf = gst_adapter_get_buffer (adapter, out_size);
    outbuf = gst_buffer_make_writable (outbuf);

    /** set timestamp */
    GST_BUFFER_PTS (outbuf) = pts;
    GST_BUFFER_DTS (outbuf) = dts;
    GST_BUFFER_DURATION (outbuf) = duration;

    ret = gst_tensor_aggregator_push (self, outbuf, frame_size);

    /** flush data */
    if (frames_flush > 0) {
      flush = frame_size * frames_flush;

      if (flush > avail) {
        /**
         * @todo flush data
         * Invalid state, tried to flush large size.
         * We have to determine how to handle this case. (flush the out-size or all available bytes)
         * Now all available bytes in adapter will be flushed.
         */
        flush = avail;
//.........这里部分代码省略.........

static void
gst_hls_demux_loop (GstHLSDemux * demux)
{
  GstBuffer *buf;
  GstFlowReturn ret;

  /* Loop for the source pad task. The task is started when we have
   * received the main playlist from the source element. It tries first to
   * cache the first fragments and then it waits until it has more data in the
   * queue. This task is woken up when we push a new fragment to the queue or
   * when we reached the end of the playlist  */

  if (G_UNLIKELY (demux->need_cache)) {
    if (!gst_hls_demux_cache_fragments (demux))
      goto cache_error;

    /* we can start now the updates thread */
    gst_hls_demux_start_update (demux);
    GST_INFO_OBJECT (demux, "First fragments cached successfully");
  }

  if (g_queue_is_empty (demux->queue)) {
    if (demux->end_of_playlist)
      goto end_of_playlist;

    goto empty_queue;
  }

  buf = g_queue_pop_head (demux->queue);

  /* Figure out if we need to create/switch pads */
  if (G_UNLIKELY (!demux->srcpad
          || GST_BUFFER_CAPS (buf) != GST_PAD_CAPS (demux->srcpad)
          || demux->need_segment)) {
    switch_pads (demux, GST_BUFFER_CAPS (buf));
    demux->need_segment = TRUE;
  }
  if (demux->need_segment) {
    /* And send a newsegment */
    GST_DEBUG_OBJECT (demux, "Sending new-segment. Segment start:%"
        GST_TIME_FORMAT, GST_TIME_ARGS (demux->position));
    gst_pad_push_event (demux->srcpad,
        gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, demux->position,
            GST_CLOCK_TIME_NONE, demux->position));
    demux->need_segment = FALSE;
  }

  if (GST_CLOCK_TIME_IS_VALID (GST_BUFFER_DURATION (buf)))
    demux->position += GST_BUFFER_DURATION (buf);

  ret = gst_pad_push (demux->srcpad, buf);
  if (ret != GST_FLOW_OK)
    goto error;

  return;

end_of_playlist:
  {
    GST_DEBUG_OBJECT (demux, "Reached end of playlist, sending EOS");
    gst_pad_push_event (demux->srcpad, gst_event_new_eos ());
    gst_hls_demux_stop (demux);
    return;
  }

cache_error:
  {
    gst_task_pause (demux->task);
    if (!demux->cancelled) {
      GST_ELEMENT_ERROR (demux, RESOURCE, NOT_FOUND,
          ("Could not cache the first fragments"), (NULL));
      gst_hls_demux_stop (demux);
    }
    return;
  }

error:
  {
    /* FIXME: handle error */
    GST_DEBUG_OBJECT (demux, "error, stopping task");
    gst_hls_demux_stop (demux);
    return;
  }

empty_queue:
  {
    gst_task_pause (demux->task);
    return;
  }
}

static gboolean
gst_hls_demux_src_query (GstPad * pad, GstQuery * query)
{
  GstHLSDemux *hlsdemux;
  gboolean ret = FALSE;

  if (query == NULL)
    return FALSE;

  hlsdemux = GST_HLS_DEMUX (gst_pad_get_element_private (pad));

  switch (query->type) {
    case GST_QUERY_DURATION:{
      GstClockTime duration = -1;
      GstFormat fmt;

      gst_query_parse_duration (query, &fmt, NULL);
      if (fmt == GST_FORMAT_TIME) {
        duration = gst_m3u8_client_get_duration (hlsdemux->client);
        if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0) {
          gst_query_set_duration (query, GST_FORMAT_TIME, duration);
          ret = TRUE;
        }
      }
      GST_INFO_OBJECT (hlsdemux, "GST_QUERY_DURATION returns %s with duration %"
          GST_TIME_FORMAT, ret ? "TRUE" : "FALSE", GST_TIME_ARGS (duration));
      break;
    }
    case GST_QUERY_URI:
      if (hlsdemux->client) {
        /* FIXME: Do we answer with the variant playlist, with the current
         * playlist or the the uri of the least downlowaded fragment? */
        gst_query_set_uri (query, hlsdemux->client->current->uri);
        ret = TRUE;
      }
      break;
    case GST_QUERY_SEEKING:{
      GstFormat fmt;
      gint64 stop = -1;

      gst_query_parse_seeking (query, &fmt, NULL, NULL, NULL);
      GST_INFO_OBJECT (hlsdemux, "Received GST_QUERY_SEEKING with format %d",
          fmt);
      if (fmt == GST_FORMAT_TIME) {
        GstClockTime duration;

        duration = gst_m3u8_client_get_duration (hlsdemux->client);
        if (GST_CLOCK_TIME_IS_VALID (duration) && duration > 0)
          stop = duration;

        gst_query_set_seeking (query, fmt,
            !gst_m3u8_client_is_live (hlsdemux->client), 0, stop);
        ret = TRUE;
        GST_INFO_OBJECT (hlsdemux, "GST_QUERY_SEEKING returning with stop : %"
            GST_TIME_FORMAT, GST_TIME_ARGS (stop));
      }
      break;
    }
    default:
      /* Don't fordward queries upstream because of the special nature of this
       * "demuxer", which relies on the upstream element only to be fed with the
       * first playlist */
      break;
  }

  return ret;
}

GstFlowReturn
gst_ks_video_device_read_frame (GstKsVideoDevice * self, guint8 * buf,
    gulong buf_size, gulong * bytes_read, GstClockTime * presentation_time,
    gulong * error_code, gchar ** error_str)
{
  GstKsVideoDevicePrivate *priv = GST_KS_VIDEO_DEVICE_GET_PRIVATE (self);
  guint req_idx;
  DWORD wait_ret;
  BOOL success;
  DWORD bytes_returned;

  g_assert (priv->cur_media_type != NULL);

  /* First time we're called, submit the requests. */
  if (G_UNLIKELY (!priv->requests_submitted)) {
    priv->requests_submitted = TRUE;

    for (req_idx = 0; req_idx < priv->num_requests; req_idx++) {
      ReadRequest *req = &g_array_index (priv->requests, ReadRequest, req_idx);

      if (!gst_ks_video_device_request_frame (self, req, error_code, error_str))
        goto error_request_failed;
    }
  }

  do {
    /* Wait for either a request to complete, a cancel or a timeout */
    wait_ret = WaitForMultipleObjects (priv->request_events->len,
        (HANDLE *) priv->request_events->data, FALSE, READ_TIMEOUT);
    if (wait_ret == WAIT_TIMEOUT)
      goto error_timeout;
    else if (wait_ret == WAIT_FAILED)
      goto error_wait;

    /* Stopped? */
    if (WaitForSingleObject (priv->cancel_event, 0) == WAIT_OBJECT_0)
      goto error_cancel;

    *bytes_read = 0;

    /* Find the last ReadRequest that finished and get the result, immediately
     * re-issuing each request that has completed. */
    for (req_idx = wait_ret - WAIT_OBJECT_0;
        req_idx < priv->num_requests; req_idx++) {
      ReadRequest *req = &g_array_index (priv->requests, ReadRequest, req_idx);

      /*
       * Completed? WaitForMultipleObjects() returns the lowest index if
       * multiple objects are in the signaled state, and we know that requests
       * are processed one by one so there's no point in looking further once
       * we've found the first that's non-signaled.
       */
      if (WaitForSingleObject (req->overlapped.hEvent, 0) != WAIT_OBJECT_0)
        break;

      success = GetOverlappedResult (priv->pin_handle, &req->overlapped,
          &bytes_returned, TRUE);

      ResetEvent (req->overlapped.hEvent);

      if (success) {
        KSSTREAM_HEADER *hdr = &req->params.header;
        KS_FRAME_INFO *frame_info = &req->params.frame_info;
        GstClockTime timestamp = GST_CLOCK_TIME_NONE;
        GstClockTime duration = GST_CLOCK_TIME_NONE;

        if (hdr->OptionsFlags & KSSTREAM_HEADER_OPTIONSF_TIMEVALID)
          timestamp = hdr->PresentationTime.Time * 100;

        if (hdr->OptionsFlags & KSSTREAM_HEADER_OPTIONSF_DURATIONVALID)
          duration = hdr->Duration * 100;

        /* Assume it's a good frame */
        *bytes_read = hdr->DataUsed;

        if (G_LIKELY (presentation_time != NULL))
          *presentation_time = timestamp;

        if (G_UNLIKELY (GST_DEBUG_IS_ENABLED ())) {
          gchar *options_flags_str =
              ks_options_flags_to_string (hdr->OptionsFlags);

          GST_DEBUG ("PictureNumber=%" G_GUINT64_FORMAT ", DropCount=%"
              G_GUINT64_FORMAT ", PresentationTime=%" GST_TIME_FORMAT
              ", Duration=%" GST_TIME_FORMAT ", OptionsFlags=%s: %d bytes",
              frame_info->PictureNumber, frame_info->DropCount,
              GST_TIME_ARGS (timestamp), GST_TIME_ARGS (duration),
              options_flags_str, hdr->DataUsed);

          g_free (options_flags_str);
        }

        /* Protect against old frames. This should never happen, see previous
         * comment on last_timestamp. */
        if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (timestamp))) {
          if (G_UNLIKELY (GST_CLOCK_TIME_IS_VALID (priv->last_timestamp) &&
                  timestamp < priv->last_timestamp)) {
            GST_WARNING ("got an old frame (last_timestamp=%" GST_TIME_FORMAT
                ", timestamp=%" GST_TIME_FORMAT ")",
                GST_TIME_ARGS (priv->last_timestamp),
//.........这里部分代码省略.........

static GstFlowReturn
gst_shape_wipe_video_sink_chain (GstPad * pad, GstObject * parent,
    GstBuffer * buffer)
{
  GstShapeWipe *self = GST_SHAPE_WIPE (parent);
  GstFlowReturn ret = GST_FLOW_OK;
  GstBuffer *mask = NULL, *outbuf = NULL;
  GstClockTime timestamp;
  gboolean new_outbuf = FALSE;
  GstVideoFrame inframe, outframe, maskframe;

  if (G_UNLIKELY (GST_VIDEO_INFO_FORMAT (&self->vinfo) ==
          GST_VIDEO_FORMAT_UNKNOWN))
    goto not_negotiated;

  timestamp = GST_BUFFER_TIMESTAMP (buffer);
  timestamp =
      gst_segment_to_stream_time (&self->segment, GST_FORMAT_TIME, timestamp);

  if (GST_CLOCK_TIME_IS_VALID (timestamp))
    gst_object_sync_values (GST_OBJECT (self), timestamp);

  GST_LOG_OBJECT (self,
      "Blending buffer with timestamp %" GST_TIME_FORMAT " at position %f",
      GST_TIME_ARGS (timestamp), self->mask_position);

  g_mutex_lock (&self->mask_mutex);
  if (self->shutdown)
    goto shutdown;

  if (!self->mask)
    g_cond_wait (&self->mask_cond, &self->mask_mutex);

  if (self->mask == NULL || self->shutdown) {
    goto shutdown;
  } else {
    mask = gst_buffer_ref (self->mask);
  }
  g_mutex_unlock (&self->mask_mutex);

  if (!gst_shape_wipe_do_qos (self, GST_BUFFER_TIMESTAMP (buffer)))
    goto qos;

  /* Try to blend inplace, if it's not possible
   * get a new buffer from downstream. */
  if (!gst_buffer_is_writable (buffer)) {
    outbuf = gst_buffer_new_allocate (NULL, gst_buffer_get_size (buffer), NULL);
    gst_buffer_copy_into (outbuf, buffer, GST_BUFFER_COPY_METADATA, 0, -1);
    new_outbuf = TRUE;
  } else {
    outbuf = buffer;
  }

  gst_video_frame_map (&inframe, &self->vinfo, buffer,
      new_outbuf ? GST_MAP_READ : GST_MAP_READWRITE);
  gst_video_frame_map (&outframe, &self->vinfo, outbuf,
      new_outbuf ? GST_MAP_WRITE : GST_MAP_READWRITE);

  gst_video_frame_map (&maskframe, &self->minfo, mask, GST_MAP_READ);

  switch (GST_VIDEO_INFO_FORMAT (&self->vinfo)) {
    case GST_VIDEO_FORMAT_AYUV:
    case GST_VIDEO_FORMAT_ARGB:
    case GST_VIDEO_FORMAT_ABGR:
      if (self->mask_bpp == 16)
        gst_shape_wipe_blend_argb_16 (self, &inframe, &maskframe, &outframe);
      else
        gst_shape_wipe_blend_argb_8 (self, &inframe, &maskframe, &outframe);
      break;
    case GST_VIDEO_FORMAT_BGRA:
    case GST_VIDEO_FORMAT_RGBA:
      if (self->mask_bpp == 16)
        gst_shape_wipe_blend_bgra_16 (self, &inframe, &maskframe, &outframe);
      else
        gst_shape_wipe_blend_bgra_8 (self, &inframe, &maskframe, &outframe);
      break;
    default:
      g_assert_not_reached ();
      break;
  }

  gst_video_frame_unmap (&outframe);
  gst_video_frame_unmap (&inframe);

  gst_video_frame_unmap (&maskframe);

  gst_buffer_unref (mask);
  if (new_outbuf)
    gst_buffer_unref (buffer);

  ret = gst_pad_push (self->srcpad, outbuf);
  if (G_UNLIKELY (ret != GST_FLOW_OK))
    goto push_failed;

  return ret;

  /* Errors */
not_negotiated:
  {
    GST_ERROR_OBJECT (self, "No valid caps yet");
//.........这里部分代码省略.........