synread.c

linux环境下用纯c写的RTP协议的通用开发库

/*-------------------------------------------------------------------------
 * synread.c - synread, synreadnextpacket, synreadwait
 *-------------------------------------------------------------------------
 */

#include <syn.h>
#include <sample.h>
#include <frame.h>
#include <packet.h>
#include <util.h>
#include <errno.h>

/*------------------------------------------------------------------------
 * synread - read a stream's data by calling read function appropriate 
 * to the stream's type (e.g., packet, sample, or frame). Up to buflen
 * bytes of data are copied to the location pointed to by buf.
 * On return ts contains the timestamp of the first data returned and
 * time contains the local system time corresponding to ts. Readparam
 * is expected to point to the type specific read parameters structure
 * appropriate to the stream type (i.e., packet, frame, or sample).
 *------------------------------------------------------------------------
 */
int
synread(struct synsession *pssn, ssrc_t ssrc, struct timespec *time, mediatime_t *ts, char *buf, int buflen, void *readparam)
{

	struct synstream	*psstm;
	int			rv;

	psstm = syngetstream(pssn, ssrc);
	if (psstm == NULL)
		return ERROR;

	/*
	 * Check that stream still exists at lower layer.
	 * I.e., we did not miss a timedout event.
	 */
	if (rtpctl(pssn->ssn_session, RTP_CTL_STREAM_EXISTS, NULL, 0, ssrc) < 0) {
		synstreamoff(pssn, ssrc);
		synreleasestream(pssn, psstm);
		return ERROR;
	}

	if (pthread_mutex_lock(&psstm->sstm_readmutex) != 0) {
		synreleasestream(pssn, psstm);
		return ERROR;
	}

	switch(psstm->sstm_type) {

	case SYN_STREAMTYPE_PACKET:
		rv = packread(pssn, psstm, time, ts, buf, buflen);
		break;

	case SYN_STREAMTYPE_SAMPLE:
		rv = sampread(pssn, psstm, time, ts, buf, buflen, &((struct sampreadparam *)readparam)->sr_shift, &((struct sampreadparam *)readparam)->sr_mark);
		break;
    
	case SYN_STREAMTYPE_FRAME:
		rv = frameread(pssn, psstm, time, ts, buf, buflen, ((struct framereadparam *)readparam)->fr_extraheaderbuf, &((struct framereadparam *)readparam)->fr_extraheaderbuflen, &((struct framereadparam *)readparam)->fr_mark);
		break;
    
	default:
		rv = ERROR;
		break;
	}

	pthread_mutex_unlock(&psstm->sstm_readmutex);
	synreleasestream(pssn, psstm);
	return rv;
}

/*------------------------------------------------------------------------
 * synreadnextpacket - get next packet from queue, blocking if approaite.
 * Synreadnextpacket is invoked by the {sample,frame,packet}read functions.
 *------------------------------------------------------------------------
 */
int
synreadnextpacket(struct synsession *pssn, struct synstream *psstm, struct rtpln **ppln)
{

	int rv;
	bool            block;

	/*
	 * Check whether or not reads on this stream 
	 * should be non-blocking.
	 */
	block = (psstm->sstm_readflags & SYN_READ_BLOCK) != 0;
	while(TRUE) {

		/*
		 * Check for zombie state.
		 */
		if (psstm->sstm_zombie == TRUE)
			return ERROR;
    
		/*
		 * Check if buffering is complete.
		 */     
		if (psstm->sstm_buffering == TRUE) {
			if (block == TRUE) {
				/*
				 * Wait for buffering to complete.
				 */
				rv = pthread_cond_wait(&psstm->sstm_readcond, &psstm->sstm_readmutex);
				if (rv != 0 && rv != ETIMEDOUT)
					return ERROR;
				else
					/*
					 * Continue to top of while(TRUE) loop
					 * and try again.
					 */
					continue;
			}
			else
				return SYN_BUFFERING;	
		}
    
		/*
		 * Extract head of the stream's RTP data queue.
		 */
		*ppln = rtpqextracthead(pssn->ssn_session, psstm->sstm_ssrc);
    
		/*
		 * Check for empty queue.
		 */
		if (*ppln == NULL) {
			if (block == TRUE) {
				/*
				 * Wait for data to arrive.
				 */
				if (pthread_cond_wait(&psstm->sstm_readcond, &psstm->sstm_readmutex) != 0)
					return ERROR;
				else
					continue;
			}
			else
				return SYN_NODATA;
		}  
		return OK;
	}
}

/*------------------------------------------------------------------------
 * synreadwait - check if data is too new to return to the application
 * and block or return error as appropriate.
 * Synreadwait is invoked by the {sample,frame,packet}read functions.
 *------------------------------------------------------------------------
 */
int
synreadwait(struct synsession *pssn, struct synstream *psstm, struct rtpln *pln, mediatime_t readfrom, struct timespec now, mediatime_t mnow, struct timespec *time)
{

	mediatime_t     delta;
	struct timespec tmptime;
	int             rv;
	bool            block;

	/*
	 * Check whether or not reads on this stream 
	 * should be non-blocking.
	 */
	block = (psstm->sstm_readflags & SYN_READ_BLOCK) != 0;

	/*
	 * Check if the data in the packet is too new 
	 * relative to the current time.
	 */
	if (tsgt(readfrom, mnow + psstm->sstm_leeway)) {

		/*
		 * If so, reinsert the packet into the RTP
		 * data queue.
		 */
		rtpqinsert(pssn->ssn_session, psstm->sstm_ssrc, pln);

		/*
		 * Determine the length of time to wait 
		 * before the packet's data can be sent 
		 * to the application layer.
		 * This time includes a ``leeway'' that 
		 * accounts for the fact that a timedwait
		 * waits *at least* the amount of time
		 * specified, and most likely more.
		 */
		delta = readfrom - mnow;
		tmptime = timeunflatten(delta, psstm->sstm_clkrt);
		tmptime = timesub(tmptime, psstm->sstm_leewayts);
		tmptime = timeadd(now, tmptime);
    
		if (block == TRUE) {
			/*
			 * Wait until appropriate time *or* a new packet
			 * arrives and is inserted at the head of the RTP
			 * data queue.
			 */
			rv = pthread_cond_timedwait(&psstm->sstm_readcond,
						    &psstm->sstm_readmutex,
						    &tmptime);
			if (rv != 0 && rv != ETIMEDOUT)
				return ERROR;
			/*
			 * The value SYN_CONTINUE tells the XXXread function
			 * that currnet data should now be at the head of
			 * the RTP data queue.
			 */
			return SYN_CONTINUE;
		}
		else {
			*time = tmptime;
			return SYN_NODATA;
		}
	}
	return OK;
}