player_media_rtp.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. 2003.  All Rights Reserved.
 * 
 * Contributor(s): 
 *              Bill May        [e-mail]
 */

#include "mpeg4ip.h"
#include "player_session.h"
#include "player_media.h"
#include "player_sdp.h"
#include "player_util.h"
#include <rtp/memory.h>
#include "rtp_bytestream.h"
#include "our_config_file.h"
#include "rtp_plugin.h"
#include "media_utils.h"
#include "rfc3119_bytestream.h"
#include "mpeg3_rtp_bytestream.h"
#include "rtp_bytestream_plugin.h"
#include "liblibsrtp.h"

#include "codec_plugin_private.h"
//#define DROP_PAKS 1
static void c_recv_callback (struct rtp *session, rtp_event *e)
{
  CPlayerMedia *m = (CPlayerMedia *)rtp_get_recv_userdata(session);
  m->recv_callback(session, e);
}

static int c_rtcp_send_packet (void *ud, uint8_t *buffer, uint32_t buflen)
{
  return ((CPlayerMedia *)ud)->rtcp_send_packet(buffer, buflen);
}


int CPlayerMedia::rtp_receive_packet (unsigned char interleaved, 
				      struct rtp_packet *pak)
{
  int ret;
  if ((interleaved & 1) == 0) {
    ret = rtp_process_recv_data(m_rtp_session, 0, pak);
    if (ret < 0) {
      xfree(pak);
    }
  } else {

    rtp_process_ctrl(m_rtp_session, pak->packet_start, pak->packet_length);
    xfree(pak);
    ret = 0;
  }
  return ret;
}

/*
 * rtp_peridic - called from receive task; generates RTCP when needed; also
 * checks for end of stream - if so, it kicks the decoder task, to finish off
 */
void CPlayerMedia::rtp_periodic (void)
{
  if (m_rtp_session_from_outside == false) {
    rtp_send_ctrl(m_rtp_session, 
		  m_rtp_byte_stream != NULL ? 
		  m_rtp_byte_stream->get_last_rtp_timestamp() : 0, 
		  NULL);
    rtp_update(m_rtp_session);
  }
  if (m_rtp_byte_stream != NULL) {
    m_rtp_byte_stream->rtp_periodic();
    if (m_rtp_byte_stream->eof()) {
      bytestream_primed();
    }
  }
}

/*
 * rtp_check_payload - this is called after receiving packets, but before a
 * m_rtp_byte_stream is selected.
 */
void CPlayerMedia::rtp_check_payload (void)
{
  // note wmay - 3/16/2005 - need to push in text here.
  // we'll only every allow 1 type to be read, so we'll be able to
  // set up the interface directly.
  bool try_bytestream_buffering = false;

  if (get_sync_type() == TIMED_TEXT_SYNC) {
    determine_payload_type_from_rtp();
    // set the buffer time to 0, so we can pass buffering immediately
    m_rtp_byte_stream->set_rtp_buffer_time(0);
    try_bytestream_buffering = true;
  } else {
    if (m_head != NULL) {
      /*
       * Make sure that the payload type is the same
       */
      if (m_head->rtp_pak_pt == m_tail->rtp_pak_pt) {
	// we either want only 1 possible protocol, or at least
	// 10 packets of the same consecutively.  10 is arbitrary.
	if (m_media_info->fmt_list->next == NULL || 
	    m_rtp_queue_len > 10) { 
	  if (determine_payload_type_from_rtp() == FALSE) {
	    clear_rtp_packets(); 
	  } else {
	    // call this function again so we begin the buffering check
	    // right away - better than a go-to.
	    try_bytestream_buffering = true;
	  }
	}
      } else {
	clear_rtp_packets();
      }
    }
  }
  // if we've set the rtp bytestream, immediately try and see if they are done
  // buffering;  if so, start the decode task moving.
  if (try_bytestream_buffering) {
    if (m_rtp_byte_stream->check_buffering() != 0) {
      m_rtp_buffering = 1;
      start_decoding();
    }
  }
}

void CPlayerMedia::rtp_start (void)
{
  if (m_rtp_ssrc_set == TRUE) {
    rtp_set_my_ssrc(m_rtp_session, m_rtp_ssrc);
  } else {
    // For now - we'll set up not to wait for RTCP validation 
    // before indicating if rtp library should accept.
    rtp_set_option(m_rtp_session, RTP_OPT_WEAK_VALIDATION, FALSE);
    rtp_set_option(m_rtp_session, RTP_OPT_PROMISC, TRUE);
  }
  if (m_rtp_byte_stream != NULL) {
    //m_rtp_byte_stream->reset(); - gets called when pausing
    //    m_rtp_byte_stream->flush_rtp_packets();
  }
  m_rtp_buffering = 0;
}

void CPlayerMedia::rtp_end(void)
{
  if (m_rtp_session != NULL && m_rtp_session_from_outside == false) {
    rtp_send_bye(m_rtp_session);
    rtp_done(m_rtp_session);
  } else {
    rtp_set_rtp_callback(m_rtp_session, NULL, NULL);
  }

  m_rtp_session = NULL;
  if (m_srtp_session != NULL) {
    destroy_srtp(m_srtp_session);
    m_srtp_session = NULL;
  }
}

int CPlayerMedia::rtcp_send_packet (uint8_t *buffer, uint32_t buflen)
{
  if (config.get_config_value(CONFIG_SEND_RTCP_IN_RTP_OVER_RTSP) != 0) {
    return rtsp_thread_send_rtcp(m_parent->get_rtsp_client(),
				 m_rtp_media_number_in_session,
				 buffer, 
				 buflen);
  }
  return buflen;
}
/****************************************************************************
 * RTP receive routines
 ****************************************************************************/
int CPlayerMedia::recv_thread (void)
{
  struct timeval timeout;
  int retcode;
  CMsg *newmsg;
  int recv_thread_stop = 0;

  m_rtp_buffering = 0;
  if (m_ports != NULL) {
    /*
     * We need to free up the ports that we got before RTP tries to set 
     * them up, so we don't have any re-use conflicts.  There is a small
     * window here that they might get used...
     */
    delete m_ports; // free up the port numbers
    m_ports = NULL;
  }

#ifdef _WIN32
  WORD wVersionRequested;
  WSADATA wsaData;
  int ret;
 
  wVersionRequested = MAKEWORD( 2, 0 );
  
  ret = WSAStartup( wVersionRequested, &wsaData );
  if ( ret != 0 ) {
    abort();
  }
#endif
  rtp_init(false);
  if (m_rtp_session != NULL) {
    rtp_start();
  }

  
  while (recv_thread_stop == 0) {
    if ((newmsg = m_rtp_msg_queue.get_message()) != NULL) {
      //player_debug_message("recv thread message %d", newmsg->get_value());
      switch (newmsg->get_value()) {
      case MSG_STOP_THREAD:
	recv_thread_stop = 1;
	break;
      case MSG_PAUSE_SESSION:
	// Indicate that we need to restart the session.
	// But keep going...
	rtp_start();
	break;
      }
      delete newmsg;
      newmsg = NULL;
    }
    if (recv_thread_stop == 1) {
      continue;
    }
    if (m_rtp_session == NULL) {
      SDL_Delay(50); 
    } else {
      if (m_rtp_session_from_outside) {
	SDL_Delay(500);
      } else {
	timeout.tv_sec = 0;
	timeout.tv_usec = 500000;
	retcode = rtp_recv(m_rtp_session, &timeout, 0);
      }
      //      player_debug_message("rtp_recv return %d", retcode);
      // Run rtp periodic after each packet received or idle time.
      if (m_paused == false || m_stream_ondemand != 0)
	rtp_periodic();
    }
    
  }
  /*
   * When we're done, send a bye, close up rtp, and go home
   */
  rtp_end();
  return (0);
}

#ifdef DROP_PAKS
static int dropcount = 0;
#endif

/*
 * CPlayerMedia::recv_callback - callback from RTP with valid data
 */
void CPlayerMedia::recv_callback (struct rtp *session, rtp_event *e)
{
  if (e == NULL) return;
  /*
   * If we're paused, just dump the packet.  Multicast case
   */
  if (m_paused) {
    if (e->type == RX_RTP) {
      media_message(LOG_DEBUG, "%s dropping pak", get_name());
      xfree(e->data);
      return;
    }
  }
#if DROP_PAKS
    if (e->type == RX_RTP && dropcount >= 50) {
      xfree((rtp_packet *)e->data);
      dropcount = 0;
      return;
    } else { 
      dropcount++;
    }
#endif
  if (m_rtp_byte_stream != NULL) {
    if ((m_rtp_byte_stream->recv_callback(session, e) > 0) &&
	(e->type == RX_RTP)) {
      // indicates they are done buffering.
      if (m_rtp_buffering == 0) {
	m_rtp_buffering = 1;
	start_decoding();
      } else {
	// we're not buffering, but the decode thread might be waiting; if so, 
	// tweak it if we have a complete frame.
	if (m_decode_thread_waiting) {
	  if (m_rtp_byte_stream->check_rtp_frame_complete_for_payload_type()) {
	    bytestream_primed();
	  }
	}
      }
    }
    return;
  }
  // we only get here if there is not a valid rtp bytestream yet.
  switch (e->type) {
  case RX_RTP:
    /* regular rtp packet - add it to the queue */
    rtp_packet *rpak;
      
    rpak = (rtp_packet *)e->data;
    if (rpak->rtp_data_len == 0) {
      xfree(rpak);
    } else {
      rpak->pd.rtp_pd_timestamp = get_time_of_day();
      rpak->pd.rtp_pd_have_timestamp = true;
      add_rtp_packet_to_queue(rpak, &m_head, &m_tail, 
			      get_name());
      m_rtp_queue_len++;
      rtp_check_payload();
    }
    break;
  case RX_SR:
    rtcp_sr *srpak;
    srpak = (rtcp_sr *)e->data;

    m_rtcp_ntp_frac = srpak->ntp_frac;
    m_rtcp_ntp_sec = srpak->ntp_sec;
    m_rtcp_rtp_ts = srpak->rtp_ts;
    m_rtcp_received = 1;
    break;
  case RX_APP:
    free(e->data);
    break;
  default:
  case RX_RR:
  case RX_SDES:
  case RX_BYE:
  case SOURCE_CREATED:
  case SOURCE_DELETED:
  case RX_RR_EMPTY:
  case RX_RTCP_START:
  case RX_RTCP_FINISH:
  case RR_TIMEOUT:
#if 0
    media_message(LOG_DEBUG, "Thread %u - Callback from rtp with %d %p", 
		  SDL_ThreadID(),e->type, e->data);
#endif
    break;
  }
}

/*
 * determine_payload_type_from_rtp - determine with protocol we're dealing with
 * in the rtp session.  Set various calculations for the sync task, as well...
 */
int CPlayerMedia::determine_payload_type_from_rtp(void)