/* chain function
 * this function does the actual processing
 */
static GstFlowReturn gst_throttle_chain(GstPad * pad, GstBuffer * buf)
{
	GstThrottle * filter = GST_THROTTLE(GST_OBJECT_PARENT(pad));
	
	if (filter->printOnly)
	{
		GstCaps * caps = gst_buffer_get_caps(buf);
		gchar * capsStr = gst_caps_to_string(caps);
		gst_caps_unref(caps);
		GST_LOG_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %sof type %s",
			GST_TIME_ARGS(buf->timestamp),
			GST_BUFFER_IS_DISCONT(buf) ? "and discontinuity " : "",
			capsStr
		);
		g_free(capsStr);
		
		GstFlowReturn ret = gst_pad_push(filter->srcpad, buf);
		GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " processed with status %d", GST_TIME_ARGS(buf->timestamp), ret);
		return ret;
	}
	
	if (filter->clock == NULL)
	{
		return gst_pad_push(filter->srcpad, buf);
	}
	
	GstClockTime realTs = gst_clock_get_time(filter->clock);
	
	if (filter->haveStartTime)
	{
		const char * discont = GST_BUFFER_IS_DISCONT(buf) ? " with discotinuity" : "";
		
		GstClockTime expectedRealTs = filter->streamStartRealTime + buf->timestamp;
		gboolean early = realTs < expectedRealTs;
		if (early)
		{
			GstClockID * cid = gst_clock_new_single_shot_id(filter->clock, expectedRealTs);
			GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %s, waiting for %ld ms", GST_TIME_ARGS(buf->timestamp), discont, (expectedRealTs - realTs)/1000000);
			gst_clock_id_wait(cid, NULL);
			gst_clock_id_unref(cid);
		}
		else
		{
			GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %s, pad on time", GST_TIME_ARGS(buf->timestamp), discont);
		}
	}
	else
	{
		filter->streamStartRealTime = realTs - buf->timestamp;
		filter->haveStartTime = TRUE;
	}
	
	return gst_pad_push(filter->srcpad, buf);
}

static GstBuffer *
gst_rtp_mp4v_depay_process (GstRTPBaseDepayload * depayload, GstBuffer * buf)
{
  GstRtpMP4VDepay *rtpmp4vdepay;
  GstBuffer *pbuf, *outbuf = NULL;
  GstRTPBuffer rtp = { NULL };
  gboolean marker;

  rtpmp4vdepay = GST_RTP_MP4V_DEPAY (depayload);

  /* flush remaining data on discont */
  if (GST_BUFFER_IS_DISCONT (buf))
    gst_adapter_clear (rtpmp4vdepay->adapter);

  gst_rtp_buffer_map (buf, GST_MAP_READ, &rtp);
  pbuf = gst_rtp_buffer_get_payload_buffer (&rtp);
  marker = gst_rtp_buffer_get_marker (&rtp);
  gst_rtp_buffer_unmap (&rtp);

  gst_adapter_push (rtpmp4vdepay->adapter, pbuf);

  /* if this was the last packet of the VOP, create and push a buffer */
  if (marker) {
    guint avail;

    avail = gst_adapter_available (rtpmp4vdepay->adapter);
    outbuf = gst_adapter_take_buffer (rtpmp4vdepay->adapter, avail);

    GST_DEBUG ("gst_rtp_mp4v_depay_chain: pushing buffer of size %"
        G_GSIZE_FORMAT, gst_buffer_get_size (outbuf));
  }
  return outbuf;
}

static GstFlowReturn
gst_wavpack_parse_chain (GstPad * pad, GstBuffer * buf)
{
  GstWavpackParse *wvparse = GST_WAVPACK_PARSE (GST_PAD_PARENT (pad));
  GstFlowReturn ret = GST_FLOW_OK;
  WavpackHeader wph;
  const guint8 *tmp_buf;

  if (!wvparse->adapter) {
    wvparse->adapter = gst_adapter_new ();
  }

  if (GST_BUFFER_IS_DISCONT (buf)) {
    gst_adapter_clear (wvparse->adapter);
    wvparse->discont = TRUE;
  }

  gst_adapter_push (wvparse->adapter, buf);

  if (gst_adapter_available (wvparse->adapter) < sizeof (WavpackHeader))
    return ret;

  if (!gst_wavpack_parse_resync_adapter (wvparse->adapter))
    return ret;

  tmp_buf = gst_adapter_peek (wvparse->adapter, sizeof (WavpackHeader));
  gst_wavpack_read_header (&wph, (guint8 *) tmp_buf);

  while (gst_adapter_available (wvparse->adapter) >= wph.ckSize + 4 * 1 + 4) {
    GstBuffer *outbuf =
        gst_adapter_take_buffer (wvparse->adapter, wph.ckSize + 4 * 1 + 4);

    if (!outbuf)
      return GST_FLOW_ERROR;

    if (wvparse->srcpad == NULL) {
      if (!gst_wavpack_parse_create_src_pad (wvparse, outbuf, &wph)) {
        GST_ERROR_OBJECT (wvparse, "Failed to create src pad");
        ret = GST_FLOW_ERROR;
        break;
      }
    }

    ret = gst_wavpack_parse_push_buffer (wvparse, outbuf, &wph);

    if (ret != GST_FLOW_OK)
      break;

    if (gst_adapter_available (wvparse->adapter) >= sizeof (WavpackHeader)) {
      tmp_buf = gst_adapter_peek (wvparse->adapter, sizeof (WavpackHeader));

      if (!gst_wavpack_parse_resync_adapter (wvparse->adapter))
        break;

      gst_wavpack_read_header (&wph, (guint8 *) tmp_buf);
    }
  }

  return ret;
}

static GstFlowReturn
gst_fluid_dec_chain (GstPad * sinkpad, GstObject * parent, GstBuffer * buffer)
{
  GstFlowReturn res = GST_FLOW_OK;
  GstFluidDec *fluiddec;
  GstClockTime pts;

  fluiddec = GST_FLUID_DEC (parent);

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    fluiddec->discont = TRUE;
  }

  pts = GST_BUFFER_PTS (buffer);

  if (pts != GST_CLOCK_TIME_NONE) {
    guint64 sample =
        gst_util_uint64_scale_int (pts, FLUID_DEC_RATE, GST_SECOND);

    if (fluiddec->last_pts == GST_CLOCK_TIME_NONE) {
      fluiddec->last_pts = pts;
      fluiddec->last_sample = sample;
    } else if (fluiddec->last_pts < pts) {
      /* generate samples for the elapsed time */
      res = produce_samples (fluiddec, pts, sample);
    }
  }

  if (res == GST_FLOW_OK) {
    handle_buffer (fluiddec, buffer);
  }
  gst_buffer_unref (buffer);

  return res;
}

static GstFlowReturn
gst_rdt_depay_chain (GstPad * pad, GstBuffer * buf)
{
  GstRDTDepay *rdtdepay;
  GstFlowReturn ret;
  GstClockTime timestamp;
  gboolean more;
  GstRDTPacket packet;

  rdtdepay = GST_RDT_DEPAY (GST_PAD_PARENT (pad));

  if (GST_BUFFER_IS_DISCONT (buf)) {
    GST_LOG_OBJECT (rdtdepay, "received discont");
    rdtdepay->discont = TRUE;
  }

  if (rdtdepay->header) {
    GstBuffer *out;

    out = rdtdepay->header;
    rdtdepay->header = NULL;

    /* push header data first */
    ret = gst_rdt_depay_push (rdtdepay, out);
  }

  /* save timestamp */
  timestamp = GST_BUFFER_TIMESTAMP (buf);

  ret = GST_FLOW_OK;

  GST_LOG_OBJECT (rdtdepay, "received buffer timestamp %" GST_TIME_FORMAT,
      GST_TIME_ARGS (timestamp));

  /* data is in RDT format. */
  more = gst_rdt_buffer_get_first_packet (buf, &packet);
  while (more) {
    GstRDTType type;

    type = gst_rdt_packet_get_type (&packet);
    GST_DEBUG_OBJECT (rdtdepay, "Have packet of type %04x", type);

    if (GST_RDT_IS_DATA_TYPE (type)) {
      GST_DEBUG_OBJECT (rdtdepay, "We have a data packet");
      ret = gst_rdt_depay_handle_data (rdtdepay, timestamp, &packet);
    } else {
      switch (type) {
        default:
          GST_DEBUG_OBJECT (rdtdepay, "Ignoring packet");
          break;
      }
    }
    if (ret != GST_FLOW_OK)
      break;

    more = gst_rdt_packet_move_to_next (&packet);
  }

  return ret;
}

static void
lose_and_recover_test (GstHarness * h, guint16 lost_seq,
    gconstpointer recbuf, gsize recbuf_size)
{
  guint64 duration = 222222;
  guint64 timestamp = 111111;
  GstBuffer *bufout;

  push_lost_event (h, lost_seq, timestamp, duration, FALSE);

  bufout = gst_harness_pull (h);
  fail_unless_equals_int (gst_buffer_get_size (bufout), recbuf_size);
  fail_unless_equals_int (GST_BUFFER_PTS (bufout), timestamp);
  fail_unless (gst_buffer_memcmp (bufout, 0, recbuf, recbuf_size) == 0);
  fail_unless (!GST_BUFFER_FLAG_IS_SET (bufout, GST_RTP_BUFFER_FLAG_REDUNDANT));
  gst_buffer_unref (bufout);

  /* Pushing the next buffer with discont flag set */
  bufout = gst_buffer_new ();
  GST_BUFFER_FLAG_SET (bufout, GST_BUFFER_FLAG_DISCONT);
  bufout = gst_harness_push_and_pull (h, bufout);
  /* Checking the flag was unset */
  fail_unless (!GST_BUFFER_IS_DISCONT (bufout));
  gst_buffer_unref (bufout);
}

static GstBuffer *
gst_rtp_mp4v_depay_process (GstBaseRTPDepayload * depayload, GstBuffer * buf)
{
  GstRtpMP4VDepay *rtpmp4vdepay;
  GstBuffer *outbuf;

  rtpmp4vdepay = GST_RTP_MP4V_DEPAY (depayload);

  /* flush remaining data on discont */
  if (GST_BUFFER_IS_DISCONT (buf))
    gst_adapter_clear (rtpmp4vdepay->adapter);

  outbuf = gst_rtp_buffer_get_payload_buffer (buf);
  gst_adapter_push (rtpmp4vdepay->adapter, outbuf);

  /* if this was the last packet of the VOP, create and push a buffer */
  if (gst_rtp_buffer_get_marker (buf)) {
    guint avail;

    avail = gst_adapter_available (rtpmp4vdepay->adapter);

    outbuf = gst_adapter_take_buffer (rtpmp4vdepay->adapter, avail);

    GST_DEBUG ("gst_rtp_mp4v_depay_chain: pushing buffer of size %d",
        GST_BUFFER_SIZE (outbuf));

    return outbuf;
  }
  return NULL;
}

/* this tests that the output is a correct discontinuous stream
 * if the input is; ie input drops in time come out the same way */
static void
test_discont_stream_instance (int inrate, int outrate, int samples,
    int numbuffers)
{
  GstElement *audioresample;
  GstBuffer *inbuffer, *outbuffer;
  GstCaps *caps;
  GstClockTime ints;

  int i, j;
  gint16 *p;

  audioresample = setup_audioresample (2, inrate, outrate);
  caps = gst_pad_get_negotiated_caps (mysrcpad);
  fail_unless (gst_caps_is_fixed (caps));

  fail_unless (gst_element_set_state (audioresample,
          GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS,
      "could not set to playing");

  for (j = 1; j <= numbuffers; ++j) {

    inbuffer = gst_buffer_new_and_alloc (samples * 4);
    GST_BUFFER_DURATION (inbuffer) = samples * GST_SECOND / inrate;
    /* "drop" half the buffers */
    ints = GST_BUFFER_DURATION (inbuffer) * 2 * (j - 1);
    GST_BUFFER_TIMESTAMP (inbuffer) = ints;
    GST_BUFFER_OFFSET (inbuffer) = (j - 1) * 2 * samples;
    GST_BUFFER_OFFSET_END (inbuffer) = j * 2 * samples + samples;

    gst_buffer_set_caps (inbuffer, caps);

    p = (gint16 *) GST_BUFFER_DATA (inbuffer);

    /* create a 16 bit signed ramp */
    for (i = 0; i < samples; ++i) {
      *p = -32767 + i * (65535 / samples);
      ++p;
      *p = -32767 + i * (65535 / samples);
      ++p;
    }

    /* pushing gives away my reference ... */
    fail_unless (gst_pad_push (mysrcpad, inbuffer) == GST_FLOW_OK);

    /* check if the timestamp of the pushed buffer matches the incoming one */
    outbuffer = g_list_nth_data (buffers, g_list_length (buffers) - 1);
    fail_if (outbuffer == NULL);
    fail_unless_equals_uint64 (ints, GST_BUFFER_TIMESTAMP (outbuffer));
    if (j > 1) {
      fail_unless (GST_BUFFER_IS_DISCONT (outbuffer),
          "expected discont buffer");
    }
  }

  /* cleanup */
  gst_caps_unref (caps);
  cleanup_audioresample (audioresample);
}

HRESULT COutputPin::DeliverSample(GstBuffer *pBuffer)
{
    HRESULT hr = S_OK;
    IMediaSample *pSample = NULL;
    REFERENCE_TIME start = -1;
    REFERENCE_TIME stop = -1;

    hr = m_pAlloc->SetGstBuffer(pBuffer);
    if (FAILED(hr))
        return hr;

    hr = GetDeliveryBuffer(&pSample, NULL, NULL, 0);
    if (FAILED(hr))
        return hr;

    // Set media time
    pSample->SetMediaTime(NULL, NULL);

    // Set time
    if (GST_BUFFER_TIMESTAMP_IS_VALID(pBuffer))
    {
        start = GST_BUFFER_TIMESTAMP(pBuffer) / 100;

        if (GST_BUFFER_DURATION_IS_VALID(pBuffer))
        {
            stop = (GST_BUFFER_TIMESTAMP(pBuffer) + GST_BUFFER_DURATION(pBuffer)) / 100;
        }
        else
        {
            stop = start + 1;
        }

        if (stop <= start) // Sometimes it may happen
            stop = start + 1;

        pSample->SetTime(&start, &stop);
    }
    else
    {
        pSample->SetTime(NULL, NULL);
    }

    if (GST_BUFFER_IS_DISCONT(pBuffer))
        pSample->SetDiscontinuity(TRUE);

    hr = Deliver(pSample);
    pSample->Release();
    if (FAILED(hr))
        return hr;

    return S_OK;
}

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

  rtpmpvpay = GST_RTP_MPV_PAY (basepayload);

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

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    GST_DEBUG_OBJECT (rtpmpvpay, "DISCONT");
    gst_rtp_mpv_pay_reset (rtpmpvpay);
  }

  avail = gst_adapter_available (rtpmpvpay->adapter);

  if (duration == -1)
    duration = 0;

  if (rtpmpvpay->first_ts == GST_CLOCK_TIME_NONE || avail == 0)
    rtpmpvpay->first_ts = timestamp;

  if (avail == 0) {
    rtpmpvpay->duration = duration;
  } else {
    rtpmpvpay->duration += duration;
  }

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

  /* 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 (avail, 4, 0);
  GST_LOG_OBJECT (rtpmpvpay, "available %d, rtp packet length %d", avail,
      packet_len);

  if (gst_rtp_base_payload_is_filled (basepayload,
          packet_len, rtpmpvpay->duration)) {
    ret = gst_rtp_mpv_pay_flush (rtpmpvpay);
  } else {
    rtpmpvpay->first_ts = timestamp;
  }

  return ret;
}

static GstFlowReturn
gst_rtp_mpa_pay_handle_buffer (GstRTPBasePayload * basepayload,
    GstBuffer * buffer)
{
  GstRtpMPAPay *rtpmpapay;
  GstFlowReturn ret;
  guint size, avail;
  guint packet_len;
  GstClockTime duration, timestamp;

  rtpmpapay = GST_RTP_MPA_PAY (basepayload);

  size = gst_buffer_get_size (buffer);
  duration = GST_BUFFER_DURATION (buffer);
  timestamp = GST_BUFFER_PTS (buffer);

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    GST_DEBUG_OBJECT (rtpmpapay, "DISCONT");
    gst_rtp_mpa_pay_reset (rtpmpapay);
  }

  avail = gst_adapter_available (rtpmpapay->adapter);

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

  /* if this buffer is going to overflow the packet, flush what we
   * have. */
  if (gst_rtp_base_payload_is_filled (basepayload,
          packet_len, rtpmpapay->duration + duration)) {
    ret = gst_rtp_mpa_pay_flush (rtpmpapay);
    avail = 0;
  } else {
    ret = GST_FLOW_OK;
  }

  if (avail == 0) {
    GST_DEBUG_OBJECT (rtpmpapay,
        "first packet, save timestamp %" GST_TIME_FORMAT,
        GST_TIME_ARGS (timestamp));
    rtpmpapay->first_ts = timestamp;
    rtpmpapay->duration = 0;
  }

  gst_adapter_push (rtpmpapay->adapter, buffer);
  rtpmpapay->duration = duration;

  return ret;
}

static GstFlowReturn
gst_smart_encoder_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
{
  GstSmartEncoder *smart_encoder;
  GstFlowReturn res = GST_FLOW_OK;
  gboolean discont, keyframe;

  smart_encoder = GST_SMART_ENCODER (parent);

  discont = GST_BUFFER_IS_DISCONT (buf);
  keyframe = !GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);

  GST_DEBUG ("New buffer %s %s %" GST_TIME_FORMAT,
      discont ? "discont" : "",
      keyframe ? "keyframe" : "", GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));

  if (keyframe) {
    GST_DEBUG ("Got a keyframe");

    /* If there's a pending GOP, flush it out */
    if (smart_encoder->pending_gop) {
      /* Mark gop_stop */
      smart_encoder->gop_stop = GST_BUFFER_TIMESTAMP (buf);

      /* flush pending */
      res = gst_smart_encoder_push_pending_gop (smart_encoder);
      if (G_UNLIKELY (res != GST_FLOW_OK))
        goto beach;
    }

    /* Mark gop_start for new gop */
    smart_encoder->gop_start = GST_BUFFER_TIMESTAMP (buf);
  }

  /* Store buffer */
  smart_encoder->pending_gop = g_list_append (smart_encoder->pending_gop, buf);
  /* Update GOP stop position */
  if (GST_BUFFER_TIMESTAMP_IS_VALID (buf)) {
    smart_encoder->gop_stop = GST_BUFFER_TIMESTAMP (buf);
    if (GST_BUFFER_DURATION_IS_VALID (buf))
      smart_encoder->gop_stop += GST_BUFFER_DURATION (buf);
  }

  GST_DEBUG ("Buffer stored , Current GOP : %" GST_TIME_FORMAT " -- %"
      GST_TIME_FORMAT, GST_TIME_ARGS (smart_encoder->gop_start),
      GST_TIME_ARGS (smart_encoder->gop_stop));

beach:
  return res;
}

/* we expect buffers as exactly one complete AU
 */
static GstFlowReturn
gst_rtp_mp4g_pay_handle_buffer (GstRTPBasePayload * basepayload,
    GstBuffer * buffer)
{
  GstRtpMP4GPay *rtpmp4gpay;

  rtpmp4gpay = GST_RTP_MP4G_PAY (basepayload);

  rtpmp4gpay->first_timestamp = GST_BUFFER_TIMESTAMP (buffer);
  rtpmp4gpay->first_duration = GST_BUFFER_DURATION (buffer);
  rtpmp4gpay->discont = GST_BUFFER_IS_DISCONT (buffer);

  /* we always encode and flush a full AU */
  gst_adapter_push (rtpmp4gpay->adapter, buffer);

  return gst_rtp_mp4g_pay_flush (rtpmp4gpay);
}

static GstFlowReturn
opus_dec_chain (GstPad * pad, GstBuffer * buf)
{
  GstFlowReturn res;
  GstOpusDec *dec;

  dec = GST_OPUS_DEC (gst_pad_get_parent (pad));

  if (GST_BUFFER_IS_DISCONT (buf)) {
    dec->discont = TRUE;
  }

  res = opus_dec_chain_parse_data (dec, buf, GST_BUFFER_TIMESTAMP (buf),
      GST_BUFFER_DURATION (buf));

//done:
  dec->packetno++;

  gst_buffer_unref (buf);
  gst_object_unref (dec);

  return res;
}

static GstFlowReturn
gst_rdt_manager_chain_rdt (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
  GstFlowReturn res;
  GstRDTManager *rdtmanager;
  GstRDTManagerSession *session;
  GstClockTime timestamp;
  GstRDTPacket packet;
  guint32 ssrc;
  guint8 pt;
  gboolean more;

  rdtmanager = GST_RDT_MANAGER (parent);

  GST_DEBUG_OBJECT (rdtmanager, "got RDT packet");

  ssrc = 0;
  pt = 0;

  GST_DEBUG_OBJECT (rdtmanager, "SSRC %08x, PT %d", ssrc, pt);

  /* find session */
  session = gst_pad_get_element_private (pad);

  /* see if we have the pad */
  if (!session->active) {
    activate_session (rdtmanager, session, ssrc, pt);
    session->active = TRUE;
  }

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    GST_DEBUG_OBJECT (rdtmanager, "received discont");
    session->discont = TRUE;
  }

  res = GST_FLOW_OK;

  /* take the timestamp of the buffer. This is the time when the packet was
   * received and is used to calculate jitter and clock skew. We will adjust
   * this timestamp with the smoothed value after processing it in the
   * jitterbuffer. */
  timestamp = GST_BUFFER_TIMESTAMP (buffer);
  /* bring to running time */
  timestamp = gst_segment_to_running_time (&session->segment, GST_FORMAT_TIME,
      timestamp);

  more = gst_rdt_buffer_get_first_packet (buffer, &packet);
  while (more) {
    GstRDTType type;

    type = gst_rdt_packet_get_type (&packet);
    GST_DEBUG_OBJECT (rdtmanager, "Have packet of type %04x", type);

    if (GST_RDT_IS_DATA_TYPE (type)) {
      GST_DEBUG_OBJECT (rdtmanager, "We have a data packet");
      res = gst_rdt_manager_handle_data_packet (session, timestamp, &packet);
    } else {
      switch (type) {
        default:
          GST_DEBUG_OBJECT (rdtmanager, "Ignoring packet");
          break;
      }
    }
    if (res != GST_FLOW_OK)
      break;

    more = gst_rdt_packet_move_to_next (&packet);
  }

  gst_buffer_unref (buffer);

  return res;
}

static GstFlowReturn
theora_enc_chain (GstPad * pad, GstBuffer * buffer)
{
  GstTheoraEnc *enc;
  ogg_packet op;
  GstClockTime timestamp, duration, running_time;
  GstFlowReturn ret;

  enc = GST_THEORA_ENC (GST_PAD_PARENT (pad));

  /* we keep track of two timelines.
   * - The timestamps from the incomming buffers, which we copy to the outgoing
   *   encoded buffers as-is. We need to do this as we simply forward the
   *   newsegment events.
   * - The running_time of the buffers, which we use to construct the granulepos
   *   in the packets.
   */
  timestamp = GST_BUFFER_TIMESTAMP (buffer);
  duration = GST_BUFFER_DURATION (buffer);
  running_time =
      gst_segment_to_running_time (&enc->segment, GST_FORMAT_TIME, timestamp);

  /* make sure we copy the discont flag to the next outgoing buffer when it's
   * set on the incomming buffer */
  if (GST_BUFFER_IS_DISCONT (buffer)) {
    enc->next_discont = TRUE;
  }

  if (enc->packetno == 0) {
    /* no packets written yet, setup headers */
    GstCaps *caps;
    GstBuffer *buf1, *buf2, *buf3;

    enc->granulepos_offset = 0;
    enc->timestamp_offset = 0;

    GST_DEBUG_OBJECT (enc, "output headers");
    /* Theora streams begin with three headers; the initial header (with
       most of the codec setup parameters) which is mandated by the Ogg
       bitstream spec.  The second header holds any comment fields.  The
       third header holds the bitstream codebook.  We merely need to
       make the headers, then pass them to libtheora one at a time;
       libtheora handles the additional Ogg bitstream constraints */

    /* first packet will get its own page automatically */
    if (theora_encode_header (&enc->state, &op) != 0)
      goto encoder_disabled;

    ret =
        theora_buffer_from_packet (enc, &op, GST_CLOCK_TIME_NONE,
        GST_CLOCK_TIME_NONE, GST_CLOCK_TIME_NONE, &buf1);
    if (ret != GST_FLOW_OK) {
      goto header_buffer_alloc;
    }

    /* create the remaining theora headers */
    theora_comment_clear (&enc->comment);
    theora_comment_init (&enc->comment);

    if (theora_encode_comment (&enc->comment, &op) != 0)
      goto encoder_disabled;

    ret =
        theora_buffer_from_packet (enc, &op, GST_CLOCK_TIME_NONE,
        GST_CLOCK_TIME_NONE, GST_CLOCK_TIME_NONE, &buf2);
    /* Theora expects us to put this packet buffer into an ogg page,
     * in which case it becomes the ogg library's responsibility to
     * free it. Since we're copying and outputting a gst_buffer,
     * we need to free it ourselves. */
    if (op.packet)
      free (op.packet);

    if (ret != GST_FLOW_OK) {
      gst_buffer_unref (buf1);
      goto header_buffer_alloc;
    }

    if (theora_encode_tables (&enc->state, &op) != 0)
      goto encoder_disabled;

    ret =
        theora_buffer_from_packet (enc, &op, GST_CLOCK_TIME_NONE,
        GST_CLOCK_TIME_NONE, GST_CLOCK_TIME_NONE, &buf3);
    if (ret != GST_FLOW_OK) {
      gst_buffer_unref (buf1);
      gst_buffer_unref (buf2);
      goto header_buffer_alloc;
    }

    /* mark buffers and put on caps */
    caps = gst_pad_get_caps (enc->srcpad);
    caps = theora_set_header_on_caps (caps, buf1, buf2, buf3);
    GST_DEBUG ("here are the caps: %" GST_PTR_FORMAT, caps);
    gst_pad_set_caps (enc->srcpad, caps);

    gst_buffer_set_caps (buf1, caps);
    gst_buffer_set_caps (buf2, caps);
    gst_buffer_set_caps (buf3, caps);

    gst_caps_unref (caps);
//.........这里部分代码省略.........

static GstFlowReturn
gst_ogg_avi_parse_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
  GstFlowReturn result = GST_FLOW_OK;
  GstOggAviParse *ogg;
  guint size;
  gchar *oggbuf;
  gint ret = -1;

  ogg = GST_OGG_AVI_PARSE (parent);

  size = gst_buffer_get_size (buffer);

  GST_LOG_OBJECT (ogg, "Chain function received buffer of size %d", size);

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    ogg_sync_reset (&ogg->sync);
    ogg->discont = TRUE;
  }

  /* write data to sync layer */
  oggbuf = ogg_sync_buffer (&ogg->sync, size);
  gst_buffer_extract (buffer, 0, oggbuf, size);
  ogg_sync_wrote (&ogg->sync, size);
  gst_buffer_unref (buffer);

  /* try to get as many packets out of the stream as possible */
  do {
    ogg_page page;

    /* try to swap out a page */
    ret = ogg_sync_pageout (&ogg->sync, &page);
    if (ret == 0) {
      GST_DEBUG_OBJECT (ogg, "need more data");
      break;
    } else if (ret == -1) {
      GST_DEBUG_OBJECT (ogg, "discont in pages");
      ogg->discont = TRUE;
    } else {
      /* new unknown stream, init the ogg stream with the serial number of the
       * page. */
      if (ogg->serial == -1) {
        ogg->serial = ogg_page_serialno (&page);
        ogg_stream_init (&ogg->stream, ogg->serial);
      }

      /* submit page */
      if (ogg_stream_pagein (&ogg->stream, &page) != 0) {
        GST_WARNING_OBJECT (ogg, "ogg stream choked on page resetting stream");
        ogg_sync_reset (&ogg->sync);
        ogg->discont = TRUE;
        continue;
      }

      /* try to get as many packets as possible out of the page */
      do {
        ogg_packet packet;

        ret = ogg_stream_packetout (&ogg->stream, &packet);
        GST_LOG_OBJECT (ogg, "packetout gave %d", ret);
        switch (ret) {
          case 0:
            break;
          case -1:
            /* out of sync, We mark a DISCONT. */
            ogg->discont = TRUE;
            break;
          case 1:
            result = gst_ogg_avi_parse_push_packet (ogg, &packet);
            if (result != GST_FLOW_OK)
              goto done;
            break;
          default:
            GST_WARNING_OBJECT (ogg,
                "invalid return value %d for ogg_stream_packetout, resetting stream",
                ret);
            break;
        }
      }
      while (ret != 0);
    }
  }
  while (ret != 0);

done:
  return result;
}

static GstFlowReturn
gst_audio_fx_base_fir_filter_transform (GstBaseTransform * base,
    GstBuffer * inbuf, GstBuffer * outbuf)
{
  GstAudioFXBaseFIRFilter *self = GST_AUDIO_FX_BASE_FIR_FILTER (base);
  GstClockTime timestamp, expected_timestamp;
  gint channels = GST_AUDIO_FILTER_CHANNELS (self);
  gint rate = GST_AUDIO_FILTER_RATE (self);
  gint bps = GST_AUDIO_FILTER_BPS (self);
  GstMapInfo inmap, outmap;
  guint input_samples;
  guint output_samples;
  guint generated_samples;
  guint64 output_offset;
  gint64 diff = 0;
  GstClockTime stream_time;

  timestamp = GST_BUFFER_TIMESTAMP (outbuf);

  if (!GST_CLOCK_TIME_IS_VALID (timestamp)
      && !GST_CLOCK_TIME_IS_VALID (self->start_ts)) {
    GST_ERROR_OBJECT (self, "Invalid timestamp");
    return GST_FLOW_ERROR;
  }

  g_mutex_lock (&self->lock);
  stream_time =
      gst_segment_to_stream_time (&base->segment, GST_FORMAT_TIME, timestamp);

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

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

  g_return_val_if_fail (self->kernel != NULL, GST_FLOW_ERROR);
  g_return_val_if_fail (channels != 0, GST_FLOW_ERROR);

  if (GST_CLOCK_TIME_IS_VALID (self->start_ts))
    expected_timestamp =
        self->start_ts + gst_util_uint64_scale_int (self->nsamples_in,
        GST_SECOND, rate);
  else
    expected_timestamp = GST_CLOCK_TIME_NONE;

  /* Reset the residue if already existing on discont buffers */
  if (GST_BUFFER_IS_DISCONT (inbuf)
      || (GST_CLOCK_TIME_IS_VALID (expected_timestamp)
          && (ABS (GST_CLOCK_DIFF (timestamp,
                      expected_timestamp) > 5 * GST_MSECOND)))) {
    GST_DEBUG_OBJECT (self, "Discontinuity detected - flushing");
    if (GST_CLOCK_TIME_IS_VALID (expected_timestamp))
      gst_audio_fx_base_fir_filter_push_residue (self);
    self->buffer_fill = 0;
    g_free (self->buffer);
    self->buffer = NULL;
    self->start_ts = timestamp;
    self->start_off = GST_BUFFER_OFFSET (inbuf);
    self->nsamples_out = 0;
    self->nsamples_in = 0;
  } else if (!GST_CLOCK_TIME_IS_VALID (self->start_ts)) {
    self->start_ts = timestamp;
    self->start_off = GST_BUFFER_OFFSET (inbuf);
  }

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

  input_samples = (inmap.size / bps) / channels;
  output_samples = (outmap.size / bps) / channels;

  self->nsamples_in += input_samples;

  generated_samples =
      self->process (self, inmap.data, outmap.data, input_samples);

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

  g_assert (generated_samples <= output_samples);
  self->nsamples_out += generated_samples;
  if (generated_samples == 0)
    goto no_samples;

  /* Calculate the number of samples we can push out now without outputting
   * latency zeros in the beginning */
  diff = ((gint64) self->nsamples_out) - ((gint64) self->latency);
  if (diff < 0)
    goto no_samples;

  if (diff < generated_samples) {
    gint64 tmp = diff;
    diff = generated_samples - diff;
    generated_samples = tmp;
  } else {
    diff = 0;
  }

  gst_buffer_resize (outbuf, diff * bps * channels,
      generated_samples * bps * channels);
//.........这里部分代码省略.........

static GstFlowReturn
gst_base_rtp_depayload_chain (GstPad * pad, GstBuffer * in)
{
  GstBaseRTPDepayload *filter;
  GstBaseRTPDepayloadPrivate *priv;
  GstBaseRTPDepayloadClass *bclass;
  GstFlowReturn ret = GST_FLOW_OK;
  GstBuffer *out_buf;
  GstClockTime timestamp;
  guint16 seqnum;
  guint32 rtptime;
  gboolean reset_seq, discont;
  gint gap;

  filter = GST_BASE_RTP_DEPAYLOAD (GST_OBJECT_PARENT (pad));
  priv = filter->priv;

  /* we must have a setcaps first */
  if (G_UNLIKELY (!priv->negotiated))
    goto not_negotiated;

  /* we must validate, it's possible that this element is plugged right after a
   * network receiver and we don't want to operate on invalid data */
  if (G_UNLIKELY (!gst_rtp_buffer_validate (in)))
    goto invalid_buffer;

  priv->discont = GST_BUFFER_IS_DISCONT (in);

  timestamp = GST_BUFFER_TIMESTAMP (in);
  /* convert to running_time and save the timestamp, this is the timestamp
   * we put on outgoing buffers. */
  timestamp = gst_segment_to_running_time (&filter->segment, GST_FORMAT_TIME,
      timestamp);
  priv->timestamp = timestamp;
  priv->duration = GST_BUFFER_DURATION (in);

  seqnum = gst_rtp_buffer_get_seq (in);
  rtptime = gst_rtp_buffer_get_timestamp (in);
  reset_seq = TRUE;
  discont = FALSE;

  GST_LOG_OBJECT (filter, "discont %d, seqnum %u, rtptime %u, timestamp %"
      GST_TIME_FORMAT, priv->discont, seqnum, rtptime,
      GST_TIME_ARGS (timestamp));

  /* Check seqnum. This is a very simple check that makes sure that the seqnums
   * are striclty increasing, dropping anything that is out of the ordinary. We
   * can only do this when the next_seqnum is known. */
  if (G_LIKELY (priv->next_seqnum != -1)) {
    gap = gst_rtp_buffer_compare_seqnum (seqnum, priv->next_seqnum);

    /* if we have no gap, all is fine */
    if (G_UNLIKELY (gap != 0)) {
      GST_LOG_OBJECT (filter, "got packet %u, expected %u, gap %d", seqnum,
          priv->next_seqnum, gap);
      if (gap < 0) {
        /* seqnum > next_seqnum, we are missing some packets, this is always a
         * DISCONT. */
        GST_LOG_OBJECT (filter, "%d missing packets", gap);
        discont = TRUE;
      } else {
        /* seqnum < next_seqnum, we have seen this packet before or the sender
         * could be restarted. If the packet is not too old, we throw it away as
         * a duplicate, otherwise we mark discont and continue. 100 misordered
         * packets is a good threshold. See also RFC 4737. */
        if (gap < 100)
          goto dropping;

        GST_LOG_OBJECT (filter,
            "%d > 100, packet too old, sender likely restarted", gap);
        discont = TRUE;
      }
    }
  }
  priv->next_seqnum = (seqnum + 1) & 0xffff;

  if (G_UNLIKELY (discont && !priv->discont)) {
    GST_LOG_OBJECT (filter, "mark DISCONT on input buffer");
    /* we detected a seqnum discont but the buffer was not flagged with a discont,
     * set the discont flag so that the subclass can throw away old data. */
    priv->discont = TRUE;
    GST_BUFFER_FLAG_SET (in, GST_BUFFER_FLAG_DISCONT);
  }

  bclass = GST_BASE_RTP_DEPAYLOAD_GET_CLASS (filter);

  if (G_UNLIKELY (bclass->process == NULL))
    goto no_process;

  /* let's send it out to processing */
  out_buf = bclass->process (filter, in);
  if (out_buf) {
    /* we pass rtptime as backward compatibility, in reality, the incomming
     * buffer timestamp is always applied to the outgoing packet. */
    ret = gst_base_rtp_depayload_push_ts (filter, rtptime, out_buf);
  }
  gst_buffer_unref (in);

  return ret;

//.........这里部分代码省略.........

static GstFlowReturn
gst_y4m_dec_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
  GstY4mDec *y4mdec;
  int n_avail;
  GstFlowReturn flow_ret = GST_FLOW_OK;
#define MAX_HEADER_LENGTH 80
  char header[MAX_HEADER_LENGTH];
  int i;
  int len;

  y4mdec = GST_Y4M_DEC (parent);

  GST_DEBUG_OBJECT (y4mdec, "chain");

  if (GST_BUFFER_IS_DISCONT (buffer)) {
    GST_DEBUG ("got discont");
    gst_adapter_clear (y4mdec->adapter);
  }

  gst_adapter_push (y4mdec->adapter, buffer);
  n_avail = gst_adapter_available (y4mdec->adapter);

  if (!y4mdec->have_header) {
    gboolean ret;
    GstCaps *caps;

    if (n_avail < MAX_HEADER_LENGTH)
      return GST_FLOW_OK;

    gst_adapter_copy (y4mdec->adapter, (guint8 *) header, 0, MAX_HEADER_LENGTH);

    header[MAX_HEADER_LENGTH - 1] = 0;
    for (i = 0; i < MAX_HEADER_LENGTH; i++) {
      if (header[i] == 0x0a)
        header[i] = 0;
    }

    ret = gst_y4m_dec_parse_header (y4mdec, header);
    if (!ret) {
      GST_ELEMENT_ERROR (y4mdec, STREAM, DECODE,
          ("Failed to parse YUV4MPEG header"), (NULL));
      return GST_FLOW_ERROR;
    }

    y4mdec->header_size = strlen (header) + 1;
    gst_adapter_flush (y4mdec->adapter, y4mdec->header_size);

    caps = gst_video_info_to_caps (&y4mdec->info);
    ret = gst_pad_set_caps (y4mdec->srcpad, caps);
    gst_caps_unref (caps);
    if (!ret) {
      GST_DEBUG_OBJECT (y4mdec, "Couldn't set caps on src pad");
      return GST_FLOW_ERROR;
    }

    y4mdec->have_header = TRUE;
  }

  if (y4mdec->have_new_segment) {
    GstEvent *event;
    GstClockTime start = gst_y4m_dec_bytes_to_timestamp (y4mdec,
        y4mdec->segment.start);
    GstClockTime stop = gst_y4m_dec_bytes_to_timestamp (y4mdec,
        y4mdec->segment.stop);
    GstClockTime time = gst_y4m_dec_bytes_to_timestamp (y4mdec,
        y4mdec->segment.time);
    GstSegment seg;

    gst_segment_init (&seg, GST_FORMAT_TIME);
    seg.start = start;
    seg.stop = stop;
    seg.time = time;
    event = gst_event_new_segment (&seg);

    gst_pad_push_event (y4mdec->srcpad, event);
    //gst_event_unref (event);

    y4mdec->have_new_segment = FALSE;
    y4mdec->frame_index = gst_y4m_dec_bytes_to_frames (y4mdec,
        y4mdec->segment.time);
    GST_DEBUG ("new frame_index %d", y4mdec->frame_index);

  }

  while (1) {
    n_avail = gst_adapter_available (y4mdec->adapter);
    if (n_avail < MAX_HEADER_LENGTH)
      break;

    gst_adapter_copy (y4mdec->adapter, (guint8 *) header, 0, MAX_HEADER_LENGTH);
    header[MAX_HEADER_LENGTH - 1] = 0;
    for (i = 0; i < MAX_HEADER_LENGTH; i++) {
      if (header[i] == 0x0a)
        header[i] = 0;
    }
    if (memcmp (header, "FRAME", 5) != 0) {
      GST_ELEMENT_ERROR (y4mdec, STREAM, DECODE,
          ("Failed to parse YUV4MPEG frame"), (NULL));
      flow_ret = GST_FLOW_ERROR;
//.........这里部分代码省略.........