hust_rtprecv.c

实现基于ip2022的MPEG2 TS的IP组播接收

/*-------------------------------------------------------------------------
 * rtprecv.c - rtprecv
 *-------------------------------------------------------------------------
 */

#include <ipOS.h>
#include <ipHAL.h>
#include <ipStack.h>
#include <ipEthernet.h>
#include "hust_rtp.h"
#include "hust_hash.h"
#include "hust_event.h"
#include "hust_rtplibcommon.h"
////#include <unistd.h>



/*------------------------------------------------------------------------
 * rtprecv - 接收一个rtp包，并将包放到所属源的stream中
 	     把stream中的rtp包解析成为连续的TS流的操作
 	     将是在stream中积累一定量的连续序列号的rtp包之后
 *------------------------------------------------------------------------
 */
int
rtprecv(struct session *psn)
{

	struct rtpln		*pln;
	////队列节点上的rtp包
	
	u32_t	from;
	///rtp包的源地址
	
	int			fromlen, len;
	///地址长度
	
	struct rtp		*prtp;
	struct stream		*pstm;
	bool			head;
	int			rv;

	pln = (struct rtpln *) bufpoolgetbuf(&psn->sn_bpool);
	///获得固定大小的缓冲区

	if (pln ==  NULL) {
		rtppostevent(psn, EVENT_NOBUFS, 0, NULL, 0);
		return ERROR;
	}
	////处理缓冲区用完的情况

	fromlen = sizeof(from);
///	len = recvfrom(psn->sn_rtpfd, (char *) &pln->rln_rtp, psn->sn_bpool.bp_size - sizeof(struct rtpln), 0, (struct sockaddr *) &from, &fromlen);
///用来接收远程主机指定的soket传来的数据，数据存放在pln->rln_trp里面

	if (len < 1) {
		bufpoolfreebuf(pln);
		return ERROR;
	}
	////接收失败的情况

	pln->rln_len = len;

	/*
	 * Extract source address as RTCP destination for unicast session 
	 * if the RTCP destination has not already been learned.
	 */
	if (psn->sn_rtcpto== RTP_INADDRUNINITIALIZED) {
		psn->sn_rtcpto = from;
	}
	////必要时，给rtcp赋目的地址

	/*
	 * Get pointer to RTP header.
	 */
	prtp = &pln->rln_rtp;  
	////prtp作为指向rtp头部的指针

	/*
	 * Convert header byte order and check version.
	 */
	////rtpntoh(prtp);
	////这里不考虑字节的位的顺序问题
	if (prtp->rtp_ver != RTP_CURRVERS) {
		bufpoolfreebuf(pln);
		return ERROR;
	}
	///检查版本

	/*
	 * Lookup stream by SSRC.
	 */
	pstm = rtpgetstream(psn, prtp->rtp_ssrc);

	/* 
	 * Create new stream if newly discovered source.
	 */
	if (pstm == NULL) {
		if ((pstm = rtpnewstream(psn, prtp->rtp_ssrc)) == NULL) {
			bufpoolfreebuf(pln);
			return ERROR;
		}
	}
	///如果是新的源，就要为这个源创建一个stream
	else {
		if (RTP_INACTIVE(pstm->stm_inactive)) {
			rtpreleasestream(psn, pstm);
			bufpoolfreebuf(pln);
			return ERROR;
		}
	}
  
	if (pstm->stm_ip== RTP_INADDRUNINITIALIZED)
		pstm->stm_ip = from;
////stream.stm_ip是数据源的地址
	/*
	 * Update stream's statistics and check if stream is still
	 * on probation.
	 */
	if (rtpupdate(psn, pstm, prtp) == ERROR) {
		rtpreleasestream(psn, pstm);
		bufpoolfreebuf(pln);
		return ERROR;
	}

	/*
	 * Enqueue the packet if stream is ``on''.
	 */
	if (pstm->stm_enqueue == TRUE) {
		/* 
		 * Tag packet with *extended* sequence number.
		 */
		pln->rln_seq = ((pstm->stm_roll) << 16) | prtp->rtp_seq;
		////pthread_mutex_lock(&pstm->stm_queue.rq_mutex);
		rv = _rtpqinsert(&pstm->stm_queue, pln, &head);
		////pthread_mutex_unlock(&pstm->stm_queue.rq_mutex);

		/*
		 * Post event if packet added at front of queue.
		 * Doing so will notify a thread waiting on an
		 * empty queue.
		 */
		if (rv == OK) {
			rtppostevent(psn, (head == TRUE ? EVENT_RTP_HEAD : EVENT_RTP), pstm->stm_ssrc, NULL, 0);
			rtpreleasestream(psn, pstm);
			return OK;
		}
		else {
			/* 
			 * Free packet and return error if 
			 * enqueue operation failed.
			 */
			rtpreleasestream(psn, pstm);
			bufpoolfreebuf(pln);
			return ERROR;
		}
	}
	rtpreleasestream(psn, pstm);

	/*
	 * Post an RTP-packet-received event.
	 */
	rtppostevent(psn, EVENT_RTP, prtp->rtp_ssrc, NULL, 0);
	bufpoolfreebuf(pln);
	return OK;
}