rtsp_thread.c

网络MPEG4IP流媒体开发源代码

/*
 * 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. 2000, 2001.  All Rights Reserved.
 * 
 * Contributor(s): 
 *              Bill May        [e-mail]
 */
/*
 * rtsp_thread.c
 */
#include "rtsp_private.h"
#include <rtp/rtp.h>
#include <rtp/memory.h>

typedef enum rtsp_rtp_state_t {
  RTP_DATA_UNKNOWN = 0,
  RTP_DATA_START = 1,
  RTP_DATA_CONTINUE = 2,
  RTSP_HEADER_CHECK = 3,
  RTP_HEADER_CHECK = 4,
} rtsp_rtp_state_t;

#ifdef DEBUG_RTP_STATES
static const char *states[] = {
  "DATA UNKNOWN",
  "DATA START",
  "DATA CONTINUE",
  "RTSP HEADER CHECK",
  "RTP HEADER CHECK"
};
#endif

typedef struct rtp_state_t {
  rtp_packet *rtp_ptr;
  rtsp_rtp_state_t state;
  int try_periodic;
  int receiving_rtsp_response;
  unsigned short rtp_len, rtp_len_gotten;
  unsigned char header[3];
} rtp_state_t;

/*
 * rtsp_thread_start_cmd()
 * Handle the start command - create the socket
 */
static int rtsp_thread_start_cmd (rtsp_client_t *info)
{
  int ret;
#ifdef _WIN32
  WORD wVersionRequested;
  WSADATA wsaData;
 
  wVersionRequested = MAKEWORD( 2, 0 );
 
  ret = WSAStartup( wVersionRequested, &wsaData );
  if ( ret != 0 ) {
    /* Tell the user that we couldn't find a usable */
    /* WinSock DLL.*/
    return (ret);
  }
#endif
  ret = rtsp_create_socket(info);
  return (ret);
}

/*
 * rtsp_thread_send_and_get()
 * send a command and get the response
 */
static int rtsp_thread_send_and_get (rtsp_client_t *info)
{
  rtsp_msg_send_and_get_t msg;
  int ret;

  ret = rtsp_thread_ipc_receive(info, (char *)&msg, sizeof(msg));
  
  if (ret != sizeof(msg)) {
    rtsp_debug(LOG_DEBUG, "Send and get - recv %d", ret);
    return -1;
  }
  ret = rtsp_send(info, msg.buffer, msg.buflen);
  rtsp_debug(LOG_DEBUG, "Send and get - send %d", ret);
  return ret;
}

static int rtsp_msg_thread_perform_callback (rtsp_client_t *info)
{
  rtsp_msg_callback_t callback;
  int ret;
  uint32_t cbs;

  cbs = sizeof(callback);
  ret = rtsp_thread_ipc_receive(info, (char *)&callback, cbs);
  if (ret != cbs) {
    rtsp_debug(LOG_ERR, "Perform callback msg - recvd %d instead of %u",
	       ret, cbs); 
    return -1;
  }

  return ((callback.func)(callback.ud));
}

static int rtsp_msg_thread_set_rtp_callback (rtsp_client_t *info)
{
  rtsp_msg_rtp_callback_t callback;
  int interleave_num;
  int ret;
  uint32_t cbs = sizeof(callback);

  rtsp_debug(LOG_DEBUG, "In rtsp_msg_thread_set_rtp_callback");
  ret = rtsp_thread_ipc_receive(info, (char *)&callback, cbs);
  if (ret != cbs) {
    rtsp_debug(LOG_ERR, "Perform callback msg - recvd %d instead of %u",
	       ret, cbs);
    return -1;
  }

  interleave_num = callback.interleave;
  if (interleave_num >= MAX_RTP_THREAD_SESSIONS) {
    return -1;
  }
  if (info->m_callback[interleave_num].rtp_callback_set != 0) {
    return -2;
  }
  info->m_callback[interleave_num].rtp_callback_set = 1;
  info->m_callback[interleave_num].rtp_callback = callback.callback_func;
  info->m_callback[interleave_num].rtp_periodic = callback.periodic_func;
  info->m_callback[interleave_num].rtp_userdata = callback.ud;
  return 0;
}

int rtsp_thread_send_rtcp (rtsp_client_t *info,
			   int interleave,
			   uint8_t *buffer,
			   int buflen)
{
  int ret;
  uint8_t buff[4];

  buff[0] = '$';
  buff[1] = (interleave * 2) + 1;
  buff[2] = (buflen >> 8) & 0xff;
  buff[3] = (buflen & 0xff);
  ret = send(info->server_socket, buff, 4, 0);
  ret += send(info->server_socket, buffer, buflen, 0);
  return ret;
}

static void callback_for_rtp_packet (rtsp_client_t *info,
				     unsigned char interleave,
				     rtp_packet *rtp_ptr,
				     unsigned short rtp_len)
{
  unsigned char  which = interleave;
  interleave /= 2;
  if (info->m_callback[interleave].rtp_callback_set) {
    (info->m_callback[interleave].rtp_callback)(info->m_callback[interleave].rtp_userdata,
						which,
						rtp_ptr,
						rtp_len);
  } else {
    xfree(rtp_ptr);
  }
}

/****************************************************************************
 *
 * Thread RTP data receive state machine
 *
 ****************************************************************************/
static __inline void change_state (rtp_state_t *state, rtsp_rtp_state_t s)
{
  if (state->state != s) {
    state->state = s;
    //rtsp_debug(LOG_DEBUG, "changing state to %d %s", s,states[s]);
  }
}
static void move_end_of_buffer (rtsp_client_t *info, int bytes)
{
  memmove(info->m_resp_buffer,
	  &info->m_resp_buffer[info->m_offset_on],
	  bytes);
  info->m_offset_on = 0;
  info->m_buffer_len = bytes;
}

static int get_rtp_packet (rtsp_client_t *info, rtp_state_t *state)
{
  int ret;
  if (state->rtp_ptr == NULL) {
    //rtsp_debug(LOG_DEBUG, "PAK %d %d", state->header[0], state->rtp_len);
    state->rtp_ptr =
      (rtp_packet *)xmalloc(state->rtp_len + RTP_PACKET_HEADER_SIZE);
    state->rtp_len_gotten = 0;
  }
  ret = 
    rtsp_recv(info,
	      ((char *)state->rtp_ptr) + RTP_PACKET_HEADER_SIZE + state->rtp_len_gotten,
	      state->rtp_len - state->rtp_len_gotten);
  //rtsp_debug(LOG_DEBUG, "recv %d", ret);
  state->rtp_len_gotten += ret;

  if (state->rtp_len_gotten == state->rtp_len) {
    callback_for_rtp_packet(info,
			    state->header[0],
			    state->rtp_ptr,
			    state->rtp_len);
    state->rtp_ptr = NULL;
    state->rtp_len = 0;
    state->rtp_len_gotten = 0;
    state->try_periodic = 1;
    change_state(state, RTP_DATA_START);
    return 0;
  } 
  change_state(state, RTP_DATA_CONTINUE);
  return 1;
}

static const char *rtsp_cmp = "rtsp/1.0";
static int rtsp_get_resp (rtsp_client_t *info,
			  rtp_state_t *state)
{
  int ret;

  ret = rtsp_get_response(info);
  if (ret == RTSP_RESPONSE_RECV_ERROR) {
    change_state(state, RTP_DATA_UNKNOWN);
  } else {
    change_state(state, RTP_DATA_START);
    if (state->receiving_rtsp_response != 0) {
      /*
       * Are they waiting ?  If so, pop them the return value
       */
      rtsp_thread_ipc_respond(info, ret);
      state->receiving_rtsp_response = 0;
    }
  }
  return 1;
}
  
static int check_rtsp_resp (rtsp_client_t *info,
			    rtp_state_t *state)
{
  int len, blen, ret;

  len = strlen(rtsp_cmp);

  blen = rtsp_bytes_in_buffer(info);

  if (len > blen) {
    if (info->m_offset_on != 0) {
      move_end_of_buffer(info, blen);
    }
    ret = rtsp_receive_socket(info,
			      info->m_resp_buffer + blen,
			      len - blen,
			      0,
			      0);
    if (ret < 0) return 0;
    info->m_buffer_len += ret;
    blen = rtsp_bytes_in_buffer(info);
  }
  
  if (len <= blen) {
    if (strncasecmp(rtsp_cmp,
		    &info->m_resp_buffer[info->m_offset_on],
		    len) == 0) {
      //rtsp_debug(LOG_DEBUG, "Found resp");
      return (rtsp_get_resp(info, state));
    }

    info->m_offset_on++;
    change_state(state, RTP_DATA_UNKNOWN);
    return 0;
  }
  return -1;
}

static int check_rtp_header (rtsp_client_t *info,
			     rtp_state_t *state)
{
  int ix;
  int blen;

  blen = rtsp_bytes_in_buffer(info);

  if (blen < 3) {
    return -1;
  }

  state->header[0] = info->m_resp_buffer[info->m_offset_on];
  state->header[1] = info->m_resp_buffer[info->m_offset_on + 1];
  state->header[2] = info->m_resp_buffer[info->m_offset_on + 2];

  ix = state->header[0] / 2;
  if ((ix >= MAX_RTP_THREAD_SESSIONS) ||
      !(info->m_callback[ix].rtp_callback_set)) {
    // header failure
    info->m_offset_on++;
    change_state(state, RTP_DATA_UNKNOWN);
    return 0;
  }

  state->rtp_len = (state->header[1] << 8) | state->header[2];
  if (state->rtp_len < 1514) {
    info->m_offset_on += 3; // increment past the header
    return (get_rtp_packet(info, state));
  }
  // length is most likely incorrect
  info->m_offset_on++;