myvtm/Framework/libtoolkit/rtpsession.c

#include "precompile.h"
#include "toolkit.h"
#include "rtpsession.h"
#include "memutil.h"
#include <winpr/winsock.h>
#ifndef _WIN32
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif //NOT _WIN32

#define MAX_RTCP_BUF 1500

#define TAG TOOLKIT_TAG("rtpsession")

struct rtp_session_t 
{
	int rtp_fd;
	int rtcp_fd;
	struct sockaddr_in local_rtp_addr;
	struct sockaddr_in local_rtcp_addr;
	struct sockaddr_in remote_rtp_addr;
	struct sockaddr_in remote_rtcp_addr;
	unsigned flags;
	rtp_state *rtpstat;
};

static __inline int bind_udp_socket_port(int sock, 
										 unsigned long local_ip, 
										 unsigned short port)
{
	struct sockaddr_in addr = {0};
	addr.sin_family = AF_INET;
	addr.sin_port = port;
	addr.sin_addr.s_addr = local_ip;
	return _bind(sock, (struct sockaddr*)&addr, sizeof(addr));
}

static __inline int bind_rtp_socket_pair(int rtp_sock, 
								int rtcp_sock, 
								unsigned long local_ip, 
								unsigned short rtp_port, 
								unsigned short rtcp_port)
{
	if (bind_udp_socket_port(rtp_sock, local_ip, rtp_port) == 0 &&
		bind_udp_socket_port(rtcp_sock, local_ip, rtcp_port) == 0) {
		return 0;
	}
	
	return TOOLKIT_UNKNOWN;
}

static __inline int is_addr_equal(const struct sockaddr_in *addr1, const struct sockaddr_in *addr2)
{
	return addr1->sin_addr.s_addr == addr2->sin_addr.s_addr && 
		addr1->sin_port == addr2->sin_port;
}

#ifndef SIO_UDP_CONNRESET
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR,12)
#endif

/*TODO: remove it and use sockutil one*/
static __inline int enable_udp_connreset(int sock, BOOL bOn)
{
#if defined(_MSC_VER)
    DWORD dwBytesReturned = 0;
    BOOL bNewBehavior = bOn;
    return WSAIoctl(sock, SIO_UDP_CONNRESET, &bNewBehavior, sizeof(bNewBehavior),
                    NULL, 0, &dwBytesReturned, NULL, NULL);
#else
	return 0;
#endif //_MSC_VER
}

static int udp_block_send(int sock, 
						  const struct sockaddr*addr, 
						  int addrlen, 
						  const char *buf, 
						  int n)
{

	int ret;
	WSABUF wsabuf[1];
	DWORD BytesSent = 0;

	wsabuf[0].buf = (char*)buf;
	wsabuf[0].len = n;

	ret = WSASendTo(sock, &wsabuf[0], 1, &BytesSent, 0, addr, addrlen, NULL, NULL);
	if (ret == 0) {
		return 0;
	} else {
		if (WSAGetLastError() == WSAEWOULDBLOCK) {
			fd_set fs;
			FD_ZERO(&fs);
			FD_SET(sock, &fs);
			
#if defined(_MSC_VER)
			return select(1, NULL, &fs, NULL, NULL);
#else
			return _select(1 + sock, NULL, &fs, NULL, NULL);
#endif //_MSC_VER
		}
	}
	return -1;
}

static int udp_block_send2(int sock, 
						   const struct sockaddr*addr, 
						   int addrlen, 
						   const char *buf0, 
						   int n0, 
						   const char *buf1, 
						   int n1)
{
	int ret;
	WSABUF wsabuf[2];
	DWORD BytesSent;

	wsabuf[0].buf = (char*)(buf0);
	wsabuf[0].len = n0;
	wsabuf[1].buf = (char*)(buf1);
	wsabuf[1].len = n1;

	ret = WSASendTo(sock, &wsabuf[0], 2, &BytesSent, 0, addr, addrlen, NULL, NULL);
	if (ret == 0) {
		return 0;
	} else {
		if (WSAGetLastError() == WSAEWOULDBLOCK) {
			fd_set fs;
			FD_ZERO(&fs);
			FD_SET(sock, &fs);
#if defined(_MSC_VER)
			return select(1, NULL, &fs, NULL, NULL);
#else
			return _select(sock + 1, NULL, &fs, NULL, NULL);
#endif //_MSC_VER
		}
	}

	return -1;
}

static int udp_poll_recv(int sock, struct sockaddr*addr, int *addrlen, char *buf, int n)
{
	int ret;
	WSABUF wsabuf[1];
	DWORD dwBytesRecv = 0;
	DWORD dwFlags = 0;

	wsabuf[0].buf = buf;
	wsabuf[0].len = n;

	ret = WSARecvFrom(sock, &wsabuf[0], 1, &dwBytesRecv, &dwFlags, addr, addrlen, NULL, NULL);

	return ret == 0 ? (int)dwBytesRecv : -1;
}

static int udp_poll_recv2(int sock, 
						  struct sockaddr*addr, 
						  int *addrlen, 
						  char *buf0, 
						  int n0, 
						  char *buf1, 
						  int n1)
{
	int ret;
	WSABUF wsabuf[2];
	DWORD dwBytesRecv = 0;
	DWORD dwFlags = 0;

	wsabuf[0].buf = buf0;
	wsabuf[0].len = n0;
	wsabuf[1].buf = buf1;
	wsabuf[1].len = n1;

	ret = WSARecvFrom(sock, &wsabuf[0], 2, &dwBytesRecv, &dwFlags, addr, addrlen, NULL, NULL);

	return ret == 0 ? (int)dwBytesRecv : -1;
}

static int rtp_session_send_rtcp_fb_tmmbn(rtp_session_t *sess, unsigned int ssrc) 
{
	if (sess && !(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		char tmp[MAX_RTCP_BUF];
		int n;
		n = rtp_state_rtcp_make_rtcp_fb_tmmbn(sess->rtpstat, tmp, sizeof(tmp), ssrc);
		if (n > 0) {
			udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
		  		sizeof(struct sockaddr_in), tmp, n);
		}
		return 0;
	}

	return TOOLKIT_EINVAL;
}

//notice，this function must be called after rtp_state_on_recv_rtcp
//because rtp_state_on_recv_rtcp will update tmmbr_info.received before we send tmmbn.
static void handle_rtcp_feedback_packet(rtp_session_t *sess, const char *buf, unsigned *flags) {
	rtcp_fb_header_t *fbh;
	rtcp_common_t *ch = (rtcp_common_t *)&buf[0];
	//we send tmmbn
	switch ((rtcp_rtpfb_type_t)ch->count) {
		case RTCP_RTPFB_TMMBR:
			fbh = (rtcp_fb_header_t *)(buf + sizeof(rtcp_common_t));
		  	rtp_session_send_rtcp_fb_tmmbn(sess, ntohl(fbh->packet_sender_ssrc));
		    *flags |= RTP_SESSION_HAS_RTPFB_TMMBR;
		  	break;
	  default:
		  break;
    }
}

static void handle_rtcp_sr_packet(rtp_session_t *sess, const char *buf) {
	//RTP_SESSION_FLAG_RECVONLY, we send RR
	if (!(sess->flags & RTP_SESSION_FLAG_SENDONLY)) {
		char tmp[MAX_RTCP_BUF];
		int length = rtp_state_rtcp_make_rr(sess->rtpstat, tmp, sizeof(tmp));
		if (length > 0) {
			udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
				sizeof(struct sockaddr_in), tmp, length);
		}
	}
}

static void handle_rtcp_packet(rtp_session_t *sess, const char *buf, int len) {

	int n  = 0;
	while( n < len) {
		const rtcp_common_t *common = (rtcp_common_t *)((const char*)buf + n);
		if (common->version != 2)
			return ;
		n += sizeof(rtcp_common_t);
		
		switch (common->pt) {
		case RTCP_SR:
			handle_rtcp_sr_packet(sess, (char *)common);
			break;
		default:
			break;
		}
		n += ntohs(common->length) * 4;
	}
}


static int recv_rtcp_raw(rtp_session_t *sess, char *buf, int n)
{
	int rc;
	struct sockaddr_in addr;
	int addrlen = sizeof(addr);
	if (!sess || !buf || n <= 0)
		return TOOLKIT_EINVAL;

	/* try recv rtcp packet */
	memset(&addr, 0, sizeof(struct sockaddr_in));
	rc = udp_poll_recv(sess->rtcp_fd, (struct sockaddr*)&addr, &addrlen, buf, n);
	if (rc > 0) {
		if (is_addr_equal(&sess->remote_rtcp_addr, &addr)) {
			handle_rtcp_packet(sess, buf, n);
			return rc;
		}
	}

	return -1;
}

static int recv_rtcp(rtp_session_t *sess, unsigned *flags)
{
	int ret;
	char tmp[MAX_RTCP_BUF];

	if (flags)
		*flags = 0;
	ret = recv_rtcp_raw(sess, tmp, sizeof(tmp));
	if (ret > 0 && rtp_state_on_recv_rtcp(sess->rtpstat, tmp, ret) == 0) {
		unsigned f = 0;
		int left = ret;
		while (left > 0) {
			rtcp_common_t *common = (rtcp_common_t*)&tmp[ret - left];
			switch (common->pt) {
			case RTCP_SDES:
				f |= RTP_SESSION_HAS_SDES;
				break;
			case RTCP_SR:
				f |= RTP_SESSION_HAS_SR;
				break;
			case RTCP_RR:
				f |= RTP_SESSION_HAS_RR;
				break;
			case RTCP_BYE:
				f |= RTP_SESSION_HAS_BYE;
				break;
			case RTCP_APP:
				f |= RTP_SESSION_HAS_APP;
				break;
			case RTCP_FIR:
				f |= RTP_SESSION_HAS_FIR;
				break;
			case RTCP_NACK:
				f |= RTP_SESSION_HAS_NACK;
				break;
			case RTCP_RTPFB:
				handle_rtcp_feedback_packet(sess, (char *)common, &f);
				break;
			default:
				break;
			}
			left -= (ntohs(common->length)+1)*4;
		}

		if (flags)
			*flags = f;

		return 0;
	}

	return -1;
}

TOOLKIT_API int rtp_session_create(const char *ip,
					   unsigned short start_port, 
					   unsigned short num_port, 
					   rtp_session_t **pp_s)
{
	return rtp_session_create2(inet_addr(ip), start_port, num_port, pp_s);
}

TOOLKIT_API int rtp_session_create2(unsigned long local_ip, 
	unsigned short start_port, 
	unsigned short num_port, 
	rtp_session_t **pp_s)
{
	rtp_session_t *sess;
	unsigned short rtp_port, end_port;
	int rtp_fd = INVALID_SOCKET;
	int rtcp_fd = INVALID_SOCKET;

	if ((start_port & 1) || num_port < 2) {
		return TOOLKIT_EINVAL;
	}

	end_port = start_port + num_port;

	for (rtp_port = start_port; rtp_port < end_port; rtp_port+=2) 
	{
		rtp_fd = _socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
		if (rtp_fd == INVALID_SOCKET) {
            return toolkit_translate_sys_error(errno);
		}

		rtcp_fd = _socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
		if (rtcp_fd == INVALID_SOCKET) {
			const int err = errno;
			closesocket(rtp_fd);
			return toolkit_translate_sys_error(err);
		}

		/** 联系下面逻辑，这里一定要执行成功，否则将直接报错*/
		if (bind_rtp_socket_pair(rtp_fd, rtcp_fd, local_ip, htons(rtp_port), htons(rtp_port+1)) == 0) {
			u_long mode = 1;
			_ioctlsocket(rtp_fd, FIONBIO, &mode);
			_ioctlsocket(rtcp_fd, FIONBIO, &mode);
			enable_udp_connreset(rtp_fd, FALSE);
			enable_udp_connreset(rtcp_fd, FALSE);
			break;
		}

		if (rtp_fd != INVALID_SOCKET) {
			closesocket(rtp_fd);
		}
		if (rtcp_fd != INVALID_SOCKET) {
			closesocket(rtcp_fd);
		}
	}

	if (rtp_port == end_port) {
		return TOOLKIT_EINVAL;
	}

    sess = malloc(sizeof(struct rtp_session_t));
    sess->local_rtp_addr.sin_family = AF_INET;
    sess->local_rtp_addr.sin_port = _htons(rtp_port);
    sess->local_rtp_addr.sin_addr.s_addr = local_ip;
    sess->local_rtcp_addr.sin_family = AF_INET;
    sess->local_rtcp_addr.sin_port = _htons(rtp_port + 1);
    sess->local_rtcp_addr.sin_addr.s_addr = local_ip;
    sess->rtp_fd = rtp_fd;
    sess->rtcp_fd = rtcp_fd;
    sess->rtpstat = rtp_state_create(-1);

    *pp_s = sess;
    return 0;
}

TOOLKIT_API int rtp_session_destroy(rtp_session_t *sess)
{
	if (!sess)
		return TOOLKIT_EINVAL;
	if (sess->rtpstat)
		rtp_state_destroy(sess->rtpstat);
	if (sess->rtp_fd != INVALID_SOCKET)
		closesocket(sess->rtp_fd);
	if (sess->rtcp_fd != INVALID_SOCKET)
		closesocket(sess->rtcp_fd);
	free(sess);
	return 0;
}

TOOLKIT_API int rtp_session_reset(rtp_session_t *sess, 
					  unsigned flags,  // 0:none 1:tx 2:rx 3:both
				      /* remote address , network byte order, NULL for wait for peer to connect in, in NAT traverse case */
					  const char *remote_ip, 
					  unsigned short remote_rtp_port, 
					  unsigned short remote_rtcp_port) // if rtcp port == 0, no rtcp
{
	return rtp_session_reset2(sess, flags, inet_addr(remote_ip), remote_rtp_port, remote_rtcp_port);
}

TOOLKIT_API int rtp_session_reset2(rtp_session_t *sess, 
	unsigned flags, 
	unsigned long iremote_ip, /* remote address , network byte order */
	unsigned short remote_rtp_port, 
	unsigned short remote_rtcp_port) 
{
	int rc = 1;
    char tmp[MAX_RTCP_BUF];
	if (!sess || !sess->rtpstat) {
		return TOOLKIT_EINVAL;
	}
		
	if (!iremote_ip) {
		remote_rtp_port = 0;
		remote_rtcp_port = 0;
	}
	sess->remote_rtp_addr.sin_family = AF_INET;
	sess->remote_rtp_addr.sin_port = _htons(remote_rtp_port);
	sess->remote_rtp_addr.sin_addr.s_addr = iremote_ip;
	sess->remote_rtcp_addr.sin_family = AF_INET;
	sess->remote_rtcp_addr.sin_port = _htons(remote_rtcp_port);
	sess->remote_rtcp_addr.sin_addr.s_addr = iremote_ip;
	sess->flags = flags;
	rtp_state_reset(sess->rtpstat, GetTickCount(), 0, 0);
	rtp_state_set_rtcp_sdes_string(sess->rtpstat, RTCP_SDES_CNAME, "vvvrtp_statevvv");

	/* clear all kernel buffer */
    do {
        struct sockaddr_in addr;
        int addrlen = sizeof(addr);
		memset(&addr, 0, sizeof(struct sockaddr_in));
        rc = udp_poll_recv(sess->rtp_fd, (struct sockaddr*)&addr, &addrlen, tmp, sizeof(tmp));
    } while (rc > 0);

    do {
        struct sockaddr_in addr;
        int addrlen = sizeof(addr);
		memset(&addr, 0, sizeof(struct sockaddr_in));
        rc = udp_poll_recv(sess->rtcp_fd, (struct sockaddr*)&addr, &addrlen, tmp, sizeof(tmp));
    } while (rc > 0);

	return 0;
}

TOOLKIT_API int rtp_session_send(rtp_session_t *sess, 
					 unsigned *rtcp_rx_flags,
					 unsigned pt,
					 unsigned mark, 
					 unsigned delta_ts, 
					 const char *buf, 
					 int n)
{
	return rtp_session_send_hook(sess, rtcp_rx_flags, pt, mark, delta_ts, buf, n, 0, 0);
}

TOOLKIT_API int rtp_session_send_hook(rtp_session_t *sess, 
	unsigned *rtcp_rx_flags,
	unsigned pt,
	unsigned mark, 
	unsigned delta_ts, 
	const char *buf, 
	int n, 
	void (*on_tx_hook)(const char *buf, int size, void *arg), void *hook_arg)
{
	rtp_hdr hdr;
	int ret;
	if (!sess || !buf || n <= 0) {
        return TOOLKIT_EINVAL;
	}
	
	rtp_state_fill_rtp(sess->rtpstat, &hdr, pt, mark, delta_ts);
	return rtp_session_send_hook2(sess, rtcp_rx_flags, &hdr, buf, n, on_tx_hook, hook_arg);
}


TOOLKIT_API int rtp_session_send_hook2(rtp_session_t *sess, 
	unsigned *rtcp_rx_flags,
	const rtp_hdr *hdr,
	const char *buf, 
	int n, 
	void (*on_tx_hook)(const char *buf, int size, void *arg), void *hook_arg)
{
	int ret;
	if (!sess || !buf || n <= 0) {
		return TOOLKIT_EINVAL;
	}

	if (rtcp_rx_flags)
		*rtcp_rx_flags = 0;
	if (!(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		//only in sendonly we process recv_rtcp in rtp_session_send_hook function.
		if (!(sess->flags & RTP_SESSION_FLAG_RECVONLY)) {
			if (sess->remote_rtcp_addr.sin_port != 0)
				recv_rtcp(sess, rtcp_rx_flags);
		}
	} else {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
	}

	{
		static int ss_seq = 0;
		if (++ss_seq == 60) {
			ret = 0;
		} else {
			WLog_DBG(TAG, "to invoke udp_block_send2...");
			ret = udp_block_send2(sess->rtp_fd, (struct sockaddr*)&sess->remote_rtp_addr, 
				sizeof(struct sockaddr_in), (char*)hdr, sizeof(*hdr), buf, n);
			WLog_DBG(TAG, "invoke udp_block_send2 returned %d", ret);
		}
	}
	if (on_tx_hook) {
		int size = sizeof(*hdr) + n;
		char *tbuf = (char*)malloc(size);
		memcpy(tbuf, hdr, sizeof(*hdr));
		memcpy(tbuf+sizeof(*hdr), buf, n);
		(*on_tx_hook)(tbuf, size, hook_arg);
		free(tbuf);
	}
	if (ret != 0) {
		WLog_DBG(TAG, "return %d before rtp_state_on_send_rtp");
		return ret;
	}

	rtp_state_on_send_rtp(sess->rtpstat, sizeof(*hdr)+n);
	if (!(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		if (rtp_state_need_send_rtcp(sess->rtpstat, TRUE) && sess->remote_rtcp_addr.sin_port != 0) {
			char tmp[MAX_RTCP_BUF];
			u__int64_t mxtbr;
			ret = rtp_state_rtcp_make_sr(sess->rtpstat, tmp, sizeof(tmp));
			udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
				sizeof(struct sockaddr_in), tmp, ret);
			//retry send tmmbr
			if (rtp_state_get_tmmbr_wait_send_maxbitrate(sess->rtpstat, &mxtbr) == 0){
				rtp_session_send_rtcp_fb_tmmbr(sess, mxtbr);
			} else {
				WLog_DBG(TAG, "%s:: %d", __FUNCTION__, __LINE__);
			}
		} else {
			WLog_DBG(TAG, "%s:: %d", __FUNCTION__, __LINE__);
		}
	} else {
		WLog_DBG(TAG, "session flag %d not has RTP_SESSION_FLAG_NO_RTCP", sess->flags);
	}
	return 0;
}


TOOLKIT_API int rtp_session_send_raw(rtp_session_t *sess, 
						 unsigned *rtcp_rx_flags,
						 const char *buf, 
						 int n)
{
	int ret;
	if (!sess || !buf || n <= 0) {
		return TOOLKIT_EINVAL;
	}

	if (rtcp_rx_flags)
		*rtcp_rx_flags = 0;
	if (!(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		if (!(sess->flags & RTP_SESSION_FLAG_RECVONLY)) {
			if (sess->remote_rtcp_addr.sin_port != 0)
				recv_rtcp(sess, rtcp_rx_flags);
		}
	}
	ret = udp_block_send(sess->rtp_fd, (struct sockaddr*)&sess->remote_rtp_addr, 
		sizeof(struct sockaddr_in), buf, n);
	if (ret != 0) {
		WLog_DBG(TAG, "%s:: %d", __FUNCTION__, __LINE__);
		return ret;
	}

	rtp_state_on_send_rtp(sess->rtpstat, n);
	if (!(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		if (rtp_state_need_send_rtcp(sess->rtpstat, TRUE) && sess->remote_rtcp_addr.sin_port != 0) {
			char tmp[MAX_RTCP_BUF];
			u__int64_t mxtbr;
			ret = rtp_state_rtcp_make_sr(sess->rtpstat, tmp, sizeof(tmp));
			udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
				sizeof(struct sockaddr_in), tmp, ret);
			//retry send tmmbr
			if (rtp_state_get_tmmbr_wait_send_maxbitrate(sess->rtpstat, &mxtbr) == 0){
				rtp_session_send_rtcp_fb_tmmbr(sess, mxtbr);
			} else {
				WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
			}
		} else {
			WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
		}
	} else {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
	}
	return 0;
}

TOOLKIT_API int rtp_session_send_rtcp_h261_fir(rtp_session_t *sess)
{
	if (sess && !(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		char tmp[MAX_RTCP_BUF];
		int n;
		n = rtp_state_rtcp_make_h261_fir(sess->rtpstat, tmp, sizeof(tmp));
		udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
			sizeof(struct sockaddr_in), tmp, n);
	} else {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
		return -1;
	}
	return 0;
}


TOOLKIT_API int rtp_session_send_rtcp_fb_tmmbr(rtp_session_t *sess, u__int64_t mxtbr) {
	if (sess && !(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		char tmp[MAX_RTCP_BUF];
		int n;
		n = rtp_state_rtcp_make_rtcp_fb_tmmbr(sess->rtpstat, tmp, sizeof(tmp), mxtbr, IP_UDP_OVERHEAD);
		udp_block_send(sess->rtcp_fd, (struct sockaddr*)&sess->remote_rtcp_addr, 
			sizeof(struct sockaddr_in), tmp, n);
	} else {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
		return -1;
	}
	return 0;
}

TOOLKIT_API int rtp_session_recv_hook(rtp_session_t *sess, 
	unsigned *pt,
	unsigned *mark, 
	unsigned *ts, 
	unsigned short *seq,
	char *buf, 
	int n,
	void (*on_rx_hook)(const char *buf, int size, void *arg), void *hook_arg)
{
	int result;
	rtp_hdr hdr;
	if (!sess || !buf || n <= 0) {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
		return -1;
	}
	
	result = rtp_session_recv_hook2(sess, &hdr, buf, n, on_rx_hook, hook_arg);
	if (result > 0) {
		if (pt)
			*pt = hdr.pt;
		if (mark)
			*mark = hdr.m;
		if (ts)
			*ts = ntohl(hdr.ts);
		if (seq)
			*seq = ntohs(hdr.seq);
	} else {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
	}
	return result;
}


TOOLKIT_API int rtp_session_recv_hook2(rtp_session_t *sess, 
	rtp_hdr *hdr,
	char *buf, 
	int n,
	void (*on_rx_hook)(const char *buf, int size, void *arg), void *hook_arg)
{
	int ret;
	int max_cnt;
	int i;
	struct sockaddr_in addr = {0};
	int addrlen = sizeof(addr);
	if (!sess || !buf || n <= 0) {
		return TOOLKIT_EINVAL;
	}
		
	max_cnt = 5;
	for (i = 0 ; i < max_cnt; ++i) {
		memset(&addr, 0, sizeof(struct sockaddr_in));
		addrlen = sizeof(addr);
		ret = udp_poll_recv2(sess->rtp_fd, (struct sockaddr*)&addr, &addrlen, (char*)hdr, sizeof(*hdr), buf, n);
		if (ret > 0) {
			if (on_rx_hook) {
				if (ret <= sizeof(*hdr)) {
					(*on_rx_hook)((char*)hdr, ret, hook_arg);
				} else {
					char *tbuf = (char*)malloc(ret);
					memcpy(tbuf, (char*)hdr, sizeof(*hdr));
					memcpy(tbuf+sizeof(*hdr), buf, ret - sizeof(*hdr)); 
					(*on_rx_hook)(tbuf, ret, hook_arg);
					free(tbuf);
				}
			}
			if (is_addr_equal(&sess->remote_rtp_addr, &addr)) {
				if (rtp_state_on_recv_rtp(sess->rtpstat, hdr, ret)==0) {
					break;
				}
			}
		} else {
			WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
			return -1;
		}
	}
	if (i >= 5) {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
        return -1;
	}

	return ret - sizeof(*hdr);
}

TOOLKIT_API int rtp_session_recv(rtp_session_t *sess, 
					 unsigned *pt,
					 unsigned *mark, 
					 unsigned *ts, 
					 unsigned short *seq,
					 char *buf, 
					 int n)
{
	return rtp_session_recv_hook(sess, pt, mark, ts, seq, buf, n, 0, 0);
}

TOOLKIT_API int rtp_session_recv_raw(rtp_session_t *sess, char *buf, int n)
{
	int ret;
	int max_cnt;
	int i;
	struct sockaddr_in addr;
	int addrlen = sizeof(addr);

    if (!sess || !buf || n <= 0) {
		return TOOLKIT_EINVAL;
    }

	max_cnt = 5;
	for (i = 0 ; i < max_cnt; ++i) {
		memset(&addr, 0, sizeof(struct sockaddr_in));
		addrlen = sizeof(addr);
		ret = udp_poll_recv(sess->rtp_fd, (struct sockaddr*)&addr, &addrlen, buf, n);
		if (ret > 0) {
			if (is_addr_equal(&sess->remote_rtp_addr, &addr)) {
				rtp_hdr *p_hdr = (rtp_hdr*)&buf[0];
				if (rtp_state_on_recv_rtp(sess->rtpstat, p_hdr, ret)==0) {
					break;
				}
			}
		}
	}
	if (i >= 5) {
		WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
        return -1;
	}
	return ret;
}

TOOLKIT_API int rtp_session_recv_rtcp(rtp_session_t *sess, unsigned *rtcp_rx_flags)
{
	if (!sess || !rtcp_rx_flags) {
		return TOOLKIT_EINVAL;
	}
	*rtcp_rx_flags = 0;
	if (!(sess->flags & RTP_SESSION_FLAG_NO_RTCP)) {
		if (sess->remote_rtcp_addr.sin_port != 0) {
			recv_rtcp(sess, rtcp_rx_flags);
		}
	} else {
        WLog_DBG(TAG, "%s::%d", __FUNCTION__, __LINE__);
	}
	return 0;
}

TOOLKIT_API int rtp_session_advance_timestamp(rtp_session_t *sess, unsigned int delta_ts)
{
    if (!sess) {
		return TOOLKIT_EINVAL;
    }
	return rtp_state_advance_timestamp(sess->rtpstat, delta_ts);
}

TOOLKIT_API int rtp_session_get_rtcp_stat(rtp_session_t *sess, rtcp_statistics *stat)
{
    if (!sess || !stat) {
		return TOOLKIT_EINVAL;
    }
	return rtp_state_get_stat(sess->rtpstat, stat);
}

TOOLKIT_API rtp_state *rtp_session_get_rtp_state(rtp_session_t *sess)
{
    if (!sess) {
		return NULL;
    }
	return sess->rtpstat;
}

TOOLKIT_API int rtp_session_get_local_ip(rtp_session_t *sess, unsigned long *p_local_ip)
{
    if (!sess || !p_local_ip) {
		return TOOLKIT_EINVAL;
    }
	*p_local_ip = sess->local_rtp_addr.sin_addr.s_addr;
	return 0;
}

TOOLKIT_API int rtp_session_get_local_rtp_port(rtp_session_t *sess, unsigned short *p_rtp)
{
    if (!sess || !p_rtp) {
		return TOOLKIT_EINVAL;
    }
	*p_rtp = ntohs(sess->local_rtp_addr.sin_port);
	return 0;
}

TOOLKIT_API int rtp_session_get_local_rtcp_port(rtp_session_t *sess, unsigned short *p_rtcp)
{
    if (!sess || !p_rtcp) {
		return TOOLKIT_EINVAL;
    }
	*p_rtcp = ntohs(sess->local_rtcp_addr.sin_port);
	return 0;
}

TOOLKIT_API int rtp_session_get_remote_ip(rtp_session_t *sess, unsigned long *p_remote_ip)
{
    if (!sess || !p_remote_ip) {
		return TOOLKIT_EINVAL;
    }
	*p_remote_ip = sess->remote_rtp_addr.sin_addr.s_addr;
	return 0;
}

TOOLKIT_API int rtp_session_get_remote_rtp_port(rtp_session_t *sess, unsigned short *p_rtp)
{
    if (!sess || !p_rtp) {
		return TOOLKIT_EINVAL;
    }
	*p_rtp = ntohs(sess->remote_rtp_addr.sin_port);
	return 0;
}

TOOLKIT_API int rtp_session_get_remote_rtcp_port(rtp_session_t *sess, unsigned short *p_rtcp)
{
    if (!sess || !p_rtcp) {
		return TOOLKIT_EINVAL;
    }
	*p_rtcp = ntohs(sess->remote_rtcp_addr.sin_port);
	return 0;
}

TOOLKIT_API int rtp_session_get_raw_fd(rtp_session_t *sess, int *rtp_fd, int *rtcp_fd)
{
    if (!sess) {
        return TOOLKIT_EINVAL;
    }
	if (rtp_fd)
		*rtp_fd = sess->rtp_fd;
	if (rtcp_fd)
		*rtcp_fd = sess->rtcp_fd;
	return 0;
}