rtpsession.c

来自「linphone源码-1.3.5.tar.gz,linphone源码-1.3.5」· C语言 代码 · 共 1,957 行 · 第 1/4 页

	{
		/* if we are in blocking mode, then suspend the calling process until timestamp
		 * wanted expires */
		/* but we must not block the process if the timestamp wanted by the application is older
		 * than current time */
		packet_time =
			rtp_session_ts_to_time (session,
				     user_ts -
				     session->rtp.rcv_query_ts_offset) +
			session->rtp.rcv_time_offset;
		ortp_debug ("rtp_session_recvm_with_ts: packet_time=%i, time=%i",packet_time, sched->time_);
		wait_point_lock(&session->recv_wp);
		if (TIME_IS_STRICTLY_NEWER_THAN (packet_time, sched->time_))
		{
			wait_point_wakeup_at(&session->recv_wp,packet_time, (session->flags & RTP_SESSION_BLOCKING_MODE)!=0);
			session_set_clr(&sched->r_sessions,session);
		}
		else session_set_set(&sched->r_sessions,session);	/*to unblock _select() immediately */
		wait_point_unlock(&session->recv_wp);
	}
	return mp;
}


int msg_to_buf (mblk_t * mp, char *buffer, int len)
{
	int rlen = len;
	mblk_t *m, *mprev;
	int mlen;
	m = mp->b_cont;
	mprev = mp;
	while (m != NULL)
	{
		mlen = (int) (m->b_wptr - m->b_rptr);
		if (mlen <= rlen)
		{
			mblk_t *consumed = m;
			memcpy (buffer, m->b_rptr, mlen);
			/* go to next mblk_t */
			mprev->b_cont = m->b_cont;
			m = m->b_cont;
			consumed->b_cont = NULL;
			freeb (consumed);
			buffer += mlen;
			rlen -= mlen;
		}
		else
		{		/*if mlen>rlen */
			memcpy (buffer, m->b_rptr, rlen);
			m->b_rptr += rlen;
			return len;
		}
	}
	return len - rlen;
}

/**
 *rtp_session_recv_with_ts:
 *@session: a rtp session.
 *@buffer:	a user supplied buffer to write the data.
 *@len:		the length in bytes of the user supplied buffer.
 *@time:	the timestamp wanted.
 *@have_more: the address of an integer to indicate if more data is availlable for the given timestamp.
 *
 *	Tries to read the bytes of the incoming rtp stream related to timestamp @time. In case 
 *	where the user supplied buffer @buffer is not large enough to get all the data 
 *	related to timestamp @time, then *( @have_more) is set to 1 to indicate that the application
 *	should recall the function with the same timestamp to get more data.
 *	
 *  When the rtp session is scheduled (see rtp_session_set_scheduling_mode() ), and the 
 *	blocking mode is on (see rtp_session_set_blocking_mode() ), then the calling thread
 *	is suspended until the timestamp given as argument expires, whatever a received packet 
 *	fits the query or not.
 *
 *	Important note: it is clear that the application cannot know the timestamp of the first
 *	packet of the incoming stream, because it can be random. The @time timestamp given to the
 *	function is used relatively to first timestamp of the stream. In simple words, 0 is a good
 *	value to start calling this function.
 *
 *	This function internally calls rtp_session_recvm_with_ts() to get a rtp packet. The content
 *	of this packet is then copied into the user supplied buffer in an intelligent manner:
 *	the function takes care of the size of the supplied buffer and the timestamp given in  
 *	argument. Using this function it is possible to read continous audio data (e.g. pcma,pcmu...)
 *	with for example a standart buffer of size of 160 with timestamp incrementing by 160 while the incoming
 *	stream has a different packet size.
 *
 *Returns: if a packet was availlable with the corresponding timestamp supplied in argument 
 *	then the number of bytes written in the user supplied buffer is returned. If no packets
 *	are availlable, either because the sender has not started to send the stream, or either
 *	because silence packet are not transmitted, or either because the packet was lost during
 *	network transport, then the function returns zero.
**/
int rtp_session_recv_with_ts (RtpSession * session, char * buffer,
			       int len, uint32_t time, int * have_more)
{
	mblk_t *mp;
	int rlen = len;
	int wlen, mlen;
	uint32_t ts_int = 0;	/*the length of the data returned in the user supplied buffer, in TIMESTAMP UNIT */
	PayloadType *payload;
	RtpStream *stream=&session->rtp;

	*have_more = 0;

	mp = rtp_session_recvm_with_ts (session, time);
	payload =rtp_profile_get_payload (session->profile,
					 session->recv_pt);
	if (payload==NULL){
		ortp_warning("rtp_session_recv_with_ts: unable to recv an unsupported payload.");
		if (mp!=NULL) freemsg(mp);
		return -1;
	}
	if (!(session->flags & RTP_SESSION_RECV_SYNC))
	{
		//ortp_debug("time=%i   rcv_last_ret_ts=%i",time,session->rtp.rcv_last_ret_ts);
		if (RTP_TIMESTAMP_IS_STRICTLY_NEWER_THAN
		    (time, session->rtp.rcv_last_ret_ts))
		{
			/* the user has missed some data previously, so we are going to give him now. */
			/* we must tell him to call the function once again with the same timestamp
			 * by setting *have_more=1 */
			*have_more = 1;
		}
		if (payload->type == PAYLOAD_AUDIO_CONTINUOUS)
		{
			ts_int = (len * payload->bits_per_sample) >> 3;
			session->rtp.rcv_last_ret_ts += ts_int;
			//ortp_debug("ts_int=%i",ts_int);
		}
		else
			ts_int = 0;
	}
	else return 0;

	/* try to fill the user buffer */
	while (1)
	{

		if (mp != NULL)
		{
			mlen = msgdsize (mp->b_cont);
			wlen = msg_to_buf (mp, buffer, rlen);
			buffer += wlen;
			rlen -= wlen;
			ortp_debug("mlen=%i wlen=%i rlen=%i", mlen, wlen,
				   rlen);
			/* do we fill all the buffer ? */
			if (rlen > 0)
			{
				/* we did not fill all the buffer */
				freemsg (mp);
				/* if we have continuous audio, try to get other packets to fill the buffer,
				 * ie continue the loop */
				//ortp_debug("User buffer not filled entirely");
				if (ts_int > 0)
				{
					time = session->rtp.rcv_last_ret_ts;
					ortp_debug("Need more: will ask for %i.",
						 time);
				}
				else
					return len - rlen;
			}
			else if (mlen > wlen)
			{
				int unread =
					mlen - wlen + (int) (mp->b_wptr -
						       mp->b_rptr);
				/* not enough space in the user supplied buffer */
				/* we re-enqueue the msg with its updated read pointers for next time */
				ortp_debug ("Re-enqueuing packet.");
				rtp_session_lock (session);
				rtp_putq (&session->rtp.rq, mp);
				rtp_session_unlock (session);
				/* quite ugly: I change the stats ... */
				ortp_global_stats.recv -= unread;
				stream->stats.recv -= unread;
				return len;
			}
			else
			{
				/* the entire packet was written to the user buffer */
				freemsg (mp);
				return len;
			}
		}
		else
		{
			/* fill with a zero pattern (silence) */
			if (payload->pattern_length != 0)
			{
				int i = 0, j = 0;
				while (i < rlen)
				{
					buffer[i] = payload->zero_pattern[j];
					i++;
					j++;
					if (j <= payload->pattern_length)
						j = 0;
				}
				return len;
			}
			*have_more = 0;
			return 0;
		}
		mp = rtp_session_recvm_with_ts (session, time);
		payload = rtp_profile_get_payload (session->profile,
						 session->recv_pt);
		if (payload==NULL){
			ortp_warning("rtp_session_recv_with_ts: unable to recv an unsupported payload.");
			if (mp!=NULL) freemsg(mp);
			return -1;
		}
	}
	return -1;
}
/**
 *rtp_session_get_current_send_ts:
 *@session: a rtp session.
 *
 *	When the rtp session is scheduled and has started to send packets, this function
 *	computes the timestamp that matches to the present time. Using this function can be 
 *	usefull when sending discontinuous streams. Some time can be elapsed between the end
 *	of a stream burst and the begin of a new stream burst, and the application may be not
 *	not aware of this elapsed time. In order to get a valid (current) timestamp to pass to 
 *	#rtp_session_send_with_ts() or #rtp_session_sendm_with_ts(), the application may
 *	use rtp_session_get_current_send_ts().
 *
 *Returns: the current send timestamp for the rtp session.
**/
uint32_t rtp_session_get_current_send_ts(RtpSession *session)
{
	uint32_t userts;
	uint32_t session_time;
	RtpScheduler *sched=session->sched;
	PayloadType *payload;
	payload=rtp_profile_get_payload(session->profile,session->send_pt);
	return_val_if_fail(payload!=NULL, 0);
	if ( (session->flags & RTP_SESSION_SCHEDULED)==0 ){
		ortp_warning("can't guess current timestamp because session is not scheduled.");
		return 0;
	}
	session_time=sched->time_-session->rtp.snd_time_offset;
	userts=  (uint32_t)( ( (double)(session_time) * (double) payload->clock_rate )/ 1000.0)
				+ session->rtp.snd_ts_offset;
	return userts;
}

/**
 *rtp_session_get_current_recv_ts:
 *@session: a rtp session.
 *
 * Same thing as rtp_session_get_current_send_ts() except that it's for an incoming stream.
 * Works only on scheduled mode.
 *
 * Returns: the theoritical that would have to be receive now.
 *
**/
uint32_t rtp_session_get_current_recv_ts(RtpSession *session){
	uint32_t userts;
	uint32_t session_time;
	RtpScheduler *sched=ortp_get_scheduler();
	PayloadType *payload;
	payload=rtp_profile_get_payload(session->profile,session->recv_pt);
	return_val_if_fail(payload!=NULL, 0);
	if ( (session->flags & RTP_SESSION_SCHEDULED)==0 ){
		ortp_warning("can't guess current timestamp because session is not scheduled.");
		return 0;
	}
	session_time=sched->time_-session->rtp.rcv_time_offset;
	userts=  (uint32_t)( ( (double)(session_time) * (double) payload->clock_rate )/ 1000.0)
				+ session->rtp.rcv_ts_offset;
	return userts;
}

/**
 *rtp_session_set_time_jump_limit:
 *@session: the rtp session
 *@ts_step: a time interval in miliseconds
 *
 * oRTP has the possibility to inform the application through a callback registered 
 * with rtp_session_signal_connect about crazy incoming RTP stream that jumps from 
 * a timestamp N to N+<some crazy value>. This lets the opportunity for the application
 * to reset the session in order to resynchronize, or any other action like stopping the call
 * and reporting an error.
**/
void rtp_session_set_time_jump_limit(RtpSession *session, int milisecs){
	uint32_t ts;
	session->rtp.time_jump=milisecs;
	ts=rtp_session_time_to_ts(session,milisecs);
	if (ts==0) session->rtp.ts_jump=1<<31;	/* do not detect ts jump */
	else session->rtp.ts_jump=ts;
}

void rtp_session_release_sockets(RtpSession *session){
	if (session->rtp.socket>=0) close_socket (session->rtp.socket);
	if (session->rtcp.socket>=0) close_socket (session->rtcp.socket);
	session->rtp.socket=-1;
	session->rtcp.socket=-1;
	/* don't discard remote addresses, then can be preserved for next use.
	session->rtp.rem_addrlen=0;
	session->rtcp.rem_addrlen=0;
	*/
}

void rtp_session_uninit (RtpSession * session)
{
	/* first of all remove the session from the scheduler */
	if (session->flags & RTP_SESSION_SCHEDULED)
	{
		rtp_scheduler_remove_session (session->sched,session);
	}
	/*flush all queues */
	flushq (&session->rtp.rq, FLUSHALL);

	/* close sockets */
	rtp_session_release_sockets(session);

	wait_point_uninit(&session->send_wp);
	wait_point_uninit(&session->recv_wp);
	ortp_mutex_destroy(&session->lock);
	if (session->current_tev!=NULL) freemsg(session->current_tev);
	if (session->rtp.cached_mp!=NULL) freemsg(session->rtp.cached_mp);
	if (session->rtcp.cached_mp!=NULL) freemsg(session->rtcp.cached_mp);
	if (session->sd!=NULL) freemsg(session->sd);

	session->signal_tables = o_list_free(session->signal_tables);
}

/**
 *rtp_session_reset:
 *@session: a rtp session.
 *
 *	Reset the session: local and remote addresses are kept unchanged but the internal
 *	queue for ordering and buffering packets is flushed, the session is ready to be
 *	re-synchronised to another incoming stream.
 *
**/
void rtp_session_reset (RtpSession * session)
{

	if (session->flags & RTP_SESSION_SCHEDULED) rtp_session_lock (session);
	
	flushq (&session->rtp.rq, FLUSHALL);
	rtp_session_set_flag (session, RTP_SESSION_RECV_SYNC);
	rtp_session_set_flag (session, RTP_SESSION_SEND_SYNC);
	rtp_session_set_flag (session, RTP_SESSION_RECV_NOT_STARTED);
	rtp_session_set_flag (session, RTP_SESSION_SEND_NOT_STARTED);
	//session->ssrc=0;
	session->rtp.snd_time_offset = 0;
	session->rtp.snd_ts_offset = 0;
	session->rtp.snd_rand_offset = 0;
	session->rtp.snd_last_ts = 0;
	session->rtp.rcv_time_offset = 0;
	session->rtp.rcv_ts_offset = 0;
	session->rtp.rcv_query_ts_offset = 0;
	session->rtp.rcv_diff_ts = 0;
	session->rtp.rcv_ts = 0;
	session->rtp.rcv_last_ts = 0;
	session->rtp.rcv_last_app_ts = 0;
	session->rtp.hwrcv_extseq = 0;
	session->rtp.hwrcv_since_last_SR=0;
	session->rtp.snd_seq = 0;
	rtp_stats_reset(&session->rtp.stats);
	jitter_control_init(&session->rtp.jittctl,-1,NULL);
	
	if (session->flags & RTP_SESSION_SCHEDULED) rtp_session_unlock (session);

}

/**
 *rtp_session_set_data:
 *	@session : a rtp session
 *	@data : an opaque pointer to be stored in the session
 *
 *	Stores some application specific data into the session, so that it is easy to retrieve it
 *	from the signal callbacks using rtp_session_get_data().
**/

void rtp_session_set_data(RtpSession *session, void *data){
	session->user_data=data;
}

/**
 *rtp_session_get_data:
 *	@session : a rtp session
 *
 *	Returns: the void pointer previously set using rtp_session_set_data()
**/
void *rtp_session_get_data(const RtpSession *session){
	return session->user_data;
}

/**
 *rtp_session_destroy:
 *@session: a rtp session.
 *
 *	Destroys a rtp session.
 *
**/
void rtp_session_destroy (RtpSession * session)
{
	rtp_session_uninit (session);
	ortp_free (session);
}

uint32_t rtp_session_time_to_ts(RtpSession *session, int time){
	PayloadType *payload;
	payload =
		rtp_profile_get_payload (session->profile,
					 session->send_pt);
	if (payload == NULL)
	{
		ortp_warning
			("rtp_session_ts_to_t: use of unsupported payload type %d.", session->send_pt);
		return 0;
	}
	/* the return value is in milisecond */
	return (uint32_t) (payload->clock_rate*(double) (time/1000.0f));
}

/* function used by the scheduler only:*/
uint32_t rtp_session_ts_to_time (RtpSession * session, uint32_t timestamp)
{
	PayloadType *payload;
	payload =
		rtp_profile_get_payload (session->profile,
					 session->send_pt);
	if (payload == NULL)
	{
		ortp_warning
			("rtp_session_ts_to_t: use of unsupported payload type %d.", session->send_pt);
		return 0;
	}
	/* the return value is in milisecond */
	return (uint32_t) (1000.0 *
			  ((double) timestamp /
			   (double) payload->clock_rate));
}


/* time is the number of miliseconds elapsed since the start of the scheduler */
void rtp_session_process (RtpSession * session, uint32_t time, RtpScheduler *sched)
{
	wait_point_lock(&session->send_wp);
	if (wait_point_check(&session->send_wp,time)){
		session_set_set(&sched->w_sessions,session);
		wait_point_wakeup(&session->send_wp);
	}
	wait_point_unlock(&session->send_wp);
	
	wait_point_lock(&session->recv_wp);
	if (wait_point_check(&session->recv_wp,time)){
		session_set_set(&sched->r_sessions,session);
		wait_point_wakeup(&session->recv_wp);
	}
	wait_point_unlock(&session->recv_wp);
}


void rtp_session_make_time_distorsion(RtpSession *session, int milisec)
{
	session->rtp.snd_time_offset+=milisec;
}


/* packet api */

void rtp_add_csrc(mblk_t *mp, uint32_t csrc)
{
	rtp_header_t *hdr=(rtp_header_t*)mp->b_rptr;
	hdr->csrc[hdr->cc]=csrc;
	hdr->cc++;
}

void
rtp_session_set_symmetric_rtp (RtpSession * session, int yesno)
{
	if (session==NULL) return;
	session->symmetric_rtp = 0;
	if (yesno)
	{
	  session->symmetric_rtp = 1;
	}
	return;
}