rtp_bytestream.cpp

完整的RTP RTSP代码库

/*
 * The contents of this file are subject to the Mozilla Public
 * License Version 1.1 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of
 * the License at http://www.mozilla.org/MPL/
 * 
 * Software distributed under the License is distributed on an "AS
 * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * rights and limitations under the License.
 * 
 * The Original Code is MPEG4IP.
 * 
 * The Initial Developer of the Original Code is Cisco Systems Inc.
 * Portions created by Cisco Systems Inc. are
 * Copyright (C) Cisco Systems Inc. 2001.  All Rights Reserved.
 * 
 * Contributor(s): 
 *              Bill May        [e-mail]
 */

#include "mpeg4ip.h"
#include <rtp/rtp.h>
#include <rtp/memory.h>
#include <sdp/sdp.h> // for NTP_TO_UNIX_TIME
#include "rtp_bytestream.h"
#include "our_config_file.h"

//#define DEBUG_RTP_PAKS 1
//#define DEBUG_RTP_FRAMES 1
//#define DEBUG_RTP_BCAST 1
//#define DEBUG_RTP_WCLOCK 1
//#define DEBUG_RTP_TS 1
//#define DEBUG_SEQUENCE_DROPS 1
#ifdef _WIN32
DEFINE_MESSAGE_MACRO(rtp_message, "rtpbyst")
#else
#define rtp_message(loglevel, fmt...) message(loglevel, "rtpbyst", fmt)
#endif
/*
 * add_rtp_packet_to_queue() - adds rtp packet to doubly linked lists - 
 * this is used both by the bytestream, and by the player_media when trying
 * to determine which rtp payload type is being used.
 */
int add_rtp_packet_to_queue (rtp_packet *pak, 
			     rtp_packet **head,
			     rtp_packet **tail,
			     const char *name)
{
  rtp_packet *q;
  bool inserted = true;
  int16_t head_diff = 0, tail_diff = 0;
#ifdef DEBUG_RTP_PAKS
  rtp_message(LOG_DEBUG, "%s - m %u pt %u seq %u ts %x len %d", 
	      name,
	      pak->rtp_pak_m, pak->rtp_pak_pt, pak->rtp_pak_seq, 
	      pak->rtp_pak_ts, pak->rtp_data_len);
#endif

  if (*head == NULL) {
    // no packets on queue
    *head = *tail = pak;
    pak->rtp_next = pak;
    pak->rtp_prev = pak;
  } else {
    // take the differences between the head and tail sequence numbers
    tail_diff = pak->rtp_pak_seq - (*tail)->rtp_pak_seq;
    head_diff = pak->rtp_pak_seq - (*head)->rtp_pak_seq;

    if (head_diff == 0 || tail_diff == 0) {
      // duplicate of head - ignore it
      inserted = false;
      rtp_message(LOG_ERR, "%s - Duplicate of pak sequence #%u", 
		  name, pak->rtp_pak_seq);
    } else if (head_diff > 0 && tail_diff < 0) {
      // insert in middle
      q = (*head)->rtp_next;

      for (q = (*head)->rtp_next; 
	   q->rtp_pak_seq - pak->rtp_pak_seq <= 0;
	   q = q->rtp_next);

      if (q->rtp_pak_seq == pak->rtp_pak_seq) {
	// duplicate
	rtp_message(LOG_ERR, "%s - duplicate pak received %u", 
		    name, pak->rtp_pak_seq);
	inserted = false;
      } else {
	rtp_message(LOG_DEBUG, "%s - insert %u before %u",
		    name, pak->rtp_pak_seq, q->rtp_pak_seq);
	// insert in the middle
	q->rtp_prev->rtp_next = pak;
	pak->rtp_prev = q->rtp_prev;
	q->rtp_prev = pak;
	pak->rtp_next = q;
      }
    } else {
      // not in the middle.  Insert between the tail and head
      (*head)->rtp_prev = pak;
      (*tail)->rtp_next = pak;
      pak->rtp_next = *head;
      pak->rtp_prev = *tail;
      if (abs(head_diff) > abs(tail_diff)) {
	*tail = pak;
      } else if (head_diff < 0 && head_diff > -10) {
	// head is closer, so, insert at begin
	rtp_message(LOG_DEBUG, "%s inserting %u at head %u tail %u",
		    name,
		    pak->rtp_pak_seq,
		    (*head)->rtp_pak_seq,
		    (*tail)->rtp_pak_seq);
	*head = pak;
      } else {
	// insert at tail
#if 0
	if (head_diff > 1000 || head_diff < -1000 ||
	    tail_diff > 1000 || tail_diff < -1000) {
	  rtp_message(LOG_DEBUG, "%s inserting %u at tail - head %u tail %u",
		      name, 
		      pak->rtp_pak_seq,
		      (*head)->rtp_pak_seq,
		      (*tail)->rtp_pak_seq);
	}
#endif
	*tail = pak;
      }
    }
  }

  if (inserted == false) {
    rtp_message(LOG_ERR, "%s Couldn't insert pak", name);
    rtp_message(LOG_DEBUG, "pak seq %u", pak->rtp_pak_seq);
    if (*head != NULL) {
      rtp_message(LOG_DEBUG, "head seq %u, tail seq %u", 
		  (*head)->rtp_pak_seq,
		  (*tail)->rtp_pak_seq);
    }
    xfree(pak);
    return (0);
  } 
  
  q = *head;
  if (q->rtp_next == *head) return 1;
#ifdef DEBUG_SEQUENCE_DROPS
  bool dump_list = false;
  int16_t diff;
  do {
    diff = q->rtp_next->rtp_pak_seq - q->rtp_pak_seq;
    if (diff < 0 || diff > 2) {
      dump_list = true;
    } else 
      q = q->rtp_next;
  } while (dump_list == false && q != *tail);
  if (dump_list) {
    rtp_message(LOG_DEBUG, "%s possible queue error - inserted %u %d %d", name,
		pak->rtp_pak_seq, head_diff, tail_diff);
    rtp_message(LOG_DEBUG, "%s seq %u %u diff %d", name, 
		q->rtp_pak_seq, q->rtp_next->rtp_pak_seq, diff);
#if 0
    q = *head;
    do {
      head_diff = q->rtp_next->rtp_pak_seq - q->rtp_pak_seq;
      rtp_message(LOG_DEBUG, "%u diff next %d", q->rtp_pak_seq, head_diff);
      q = q->rtp_next;
    } while (q != *head);
#endif
    rtp_message(LOG_DEBUG, "%s - head %u tail %u", 
		name, (*head)->rtp_pak_seq, (*tail)->rtp_pak_seq);
  }
#endif
  return (1);
}

CRtpByteStreamBase::CRtpByteStreamBase(const char *name,
				       format_list_t *fmt,
				       unsigned int rtp_pt,
				       int ondemand,
				       uint64_t tps,
				       rtp_packet **head, 
				       rtp_packet **tail,
				       int rtp_seq_set,
				       uint16_t rtp_base_seq,
				       int rtp_ts_set,
				       uint32_t rtp_base_ts,
				       int rtcp_received,
				       uint32_t ntp_frac,
				       uint32_t ntp_sec,
				       uint32_t rtp_ts) :
  COurInByteStream(name)
{
  m_fmt = fmt;
  m_head = *head;
  *head = NULL;
  m_tail = *tail;
  *tail = NULL;

  if (rtp_ts_set) {
    set_rtp_base_ts(rtp_base_ts);
  } else {
    m_base_ts_set = false;
  }

  if (rtp_seq_set) {
    set_rtp_base_seq(rtp_base_seq);
  } else {
    m_rtp_base_seq_set = false;
  }

  m_have_first_pak_ts = false;
  m_rtp_pt = rtp_pt;
  uint64_t temp;
  temp = config.get_config_value(CONFIG_RTP_BUFFER_TIME_MSEC);
  if (temp > 0) {
    m_rtp_buffer_time = temp;
  } else {
    m_rtp_buffer_time = TO_U64(2000);
  }

  m_timescale = tps;

  reset();

  m_last_rtp_ts = 0;
  m_total =0;
  m_skip_on_advance_bytes = 0;
  m_stream_ondemand = ondemand;
  m_rtcp_received = false;
  m_rtp_packet_mutex = SDL_CreateMutex();
  m_buffering = 0;
  m_eof = 0;
  m_psptr = NULL;
  m_have_sync_info = false;
  if (rtcp_received) {
    calculate_wallclock_offset_from_rtcp(ntp_frac, ntp_sec, rtp_ts);
  }
}

CRtpByteStreamBase::~CRtpByteStreamBase (void)
{
  flush_rtp_packets();
  if (m_rtp_packet_mutex) {
    SDL_DestroyMutex(m_rtp_packet_mutex);
    m_rtp_packet_mutex = NULL;
  }
}

// set_sync - this is for audio only - it will send messages to any
// video rtp bytestream to perform the syncronizatio
void CRtpByteStreamBase::set_sync (CPlayerSession *psptr) 
{ 
  m_psptr = psptr; 
}

void CRtpByteStreamBase::init (void)
{
  m_wrap_offset = 0;
  m_offset_in_pak = m_skip_on_advance_bytes;
  m_eof = 0;
}
void CRtpByteStreamBase::set_wallclock_offset (uint64_t wclock, 
					       uint32_t rtp_ts) 
{
  int32_t rtp_ts_diff;
  int64_t wclock_diff;
  uint64_t wclock_calc;
  bool set = true;
  bool had_recvd_rtcp;
  if (m_rtcp_received == 1 /*&&
			     m_stream_ondemand == 0*/) {
    rtp_ts_diff = rtp_ts;
    rtp_ts_diff -= m_rtcp_rtp_ts;
    wclock_diff = (int64_t)rtp_ts_diff;
    wclock_diff *= TO_D64(1000);
    wclock_diff /= (int64_t)m_timescale;
    wclock_calc = m_rtcp_ts;
    wclock_calc += wclock_diff;
    set = false;
    if (wclock_calc != wclock) {
#ifdef DEBUG_RTP_WCLOCK
      rtp_message(LOG_DEBUG, 
		  "%s - set wallclock - wclock should be "U64" is "U64, 
		m_name, wclock_calc, wclock);
#endif
      // don't change wclock offset if it's > 100 msec - otherwise, 
      // it's annoying noise
      int64_t diff = wclock_calc - wclock;
      if (abs(diff) > 2 && abs(diff) < 100) {
	set = false;
	//	rtp_message(LOG_DEBUG, "not changing");
	// we'll allow a msec drift here or there to allow for rounding - 
	// we want this to change every so often
      }
    }
    
  }
  had_recvd_rtcp = m_rtcp_received;
  m_rtcp_received = true;
  SDL_LockMutex(m_rtp_packet_mutex);
  if (set) {
    m_rtcp_ts = wclock;
    m_rtcp_rtp_ts = rtp_ts;
  }
  if (m_have_first_pak_ts) {
    // we only want positives here
    int32_t diff;
    diff = rtp_ts - m_first_pak_rtp_ts;
    int32_t compare = 3600 * m_timescale;
#ifdef DEBUG_RTP_WCLOCK
    rtp_message(LOG_DEBUG, "%s - 1st rtp ts %u rtp %u %u", 
		m_name, m_first_pak_rtp_ts, rtp_ts, diff);
    rtp_message(LOG_DEBUG, "%s - 1st ts "U64, m_name, m_first_pak_ts);
#endif
    if (diff > compare) {
      // adjust once an hour, to keep errors low
      // we'll adjust the timestamp and rtp timestamp
      int64_t ts_diff;
      ts_diff = (int64_t)diff;
      ts_diff *= TO_U64(1000);
      ts_diff /= (int64_t)m_timescale;
      m_first_pak_ts += ts_diff;
      m_first_pak_rtp_ts += diff;
#ifdef DEBUG_RTP_WCLOCK
      rtp_message(LOG_DEBUG, "CHANGE %s - first pak ts is now "U64" rtp %u", 
		  m_name, m_first_pak_ts, m_first_pak_rtp_ts);
#endif
    }
    // We've received an RTCP - see if we need to syncronize
    // the video streams.
    if (m_psptr != NULL) {
      rtcp_sync_t sync;
      sync.first_pak_ts = m_first_pak_ts;
      sync.first_pak_rtp_ts = m_first_pak_rtp_ts;
      sync.rtcp_ts = m_rtcp_ts;
      sync.rtcp_rtp_ts = m_rtcp_rtp_ts;
      sync.timescale = m_timescale;
      m_psptr->synchronize_rtp_bytestreams(&sync);
    } else {
      // if this is our first rtcp, try to synchronize
      if (!had_recvd_rtcp) synchronize(NULL);
    }
  }

  SDL_UnlockMutex(m_rtp_packet_mutex);
}

/*
 * calculate_wallclock_offset_from_rtcp
 * Given certain information from an rtcp message, Calculate what the
 * wallclock time for rtp sequence number 0 would be.
 */
void
CRtpByteStreamBase::calculate_wallclock_offset_from_rtcp (uint32_t ntp_frac,
							  uint32_t ntp_sec,
							  uint32_t rtp_ts)
{
  uint64_t wclock;
  wclock = ntp_frac;
  wclock *= TO_U64(1000);
  wclock /= (TO_U64(1) << 32);
  uint64_t offset;
  offset = ntp_sec;
  offset -= NTP_TO_UNIX_TIME;
  offset *= TO_U64(1000);
  wclock += offset;
#ifdef DEBUG_RTP_WCLOCK
  rtp_message(LOG_DEBUG, "%s RTCP data - sec %u frac %u value "U64" ts %u", 
	      m_name, ntp_sec, ntp_frac, wclock, rtp_ts);
#endif
  set_wallclock_offset(wclock, rtp_ts);
}

/*
 * check_buffering is called to check and see if we should be buffering
 */
int CRtpByteStreamBase::check_buffering (void)
{
  if (m_buffering == 0) {
    uint32_t head_ts, tail_ts;
    if (m_head != NULL) {
      /*
       * Payload type the same.  Make sure we have at least 2 seconds of
       * good data
       */
      if (check_rtp_frame_complete_for_payload_type()) {
	head_ts = m_head->rtp_pak_ts;
	tail_ts = m_tail->rtp_pak_ts;
	if (head_ts > tail_ts &&
	    ((head_ts & (1 << 31)) == (tail_ts & (1 << 31)))) {
	  return 0;
	}
	uint64_t calc;
	calc = tail_ts;
	calc -= head_ts;
	calc *= TO_U64(1000);