www.pudn.com > streamrtp.rar > rtprecv.c

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

#include 
#include 
#include 
#include 
#include 
#include 

/*------------------------------------------------------------------------
 * rtprecv - receive and process an incoming RTP packet
 *------------------------------------------------------------------------
 */
int
rtprecv(struct session *psn)
{

	struct rtpln		*pln;
	struct sockaddr_in	from;
	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);

	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.sin_addr.s_addr == RTP_INADDRUNINITIALIZED) {
		psn->sn_rtcpto.sin_addr = from.sin_addr;
	}

	/*
	 * Get pointer to RTP header.
	 */
	prtp = &pln->rln_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;
		}
	}
	else {
		if (RTP_INACTIVE(pstm->stm_inactive)) {
			rtpreleasestream(psn, pstm);
			bufpoolfreebuf(pln);
			return ERROR;
		}
	}
  
	if (pstm->stm_ip.sin_addr.s_addr == RTP_INADDRUNINITIALIZED)
		pstm->stm_ip = from;

	/*
	 * 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;
}