rtpsession.c

mediastreamer2是开源的网络传输媒体流的库

	}
	/* if we are in blocking mode, then suspend the process until the scheduler it's time to send  the
	 * next packet */
	/* if the timestamp of the packet queued is older than current time, then you we must
	 * not block */
	if (session->flags & RTP_SESSION_SCHEDULED)
	{
		packet_time =
			rtp_session_ts_to_time (session,
				     send_ts -
				     session->rtp.snd_ts_offset) +
					session->rtp.snd_time_offset;
		/*ortp_message("rtp_session_send_with_ts: packet_time=%i time=%i",packet_time,sched->time_);*/
		wait_point_lock(&session->snd.wp);
		if (TIME_IS_STRICTLY_NEWER_THAN (packet_time, sched->time_))
		{
			wait_point_wakeup_at(&session->snd.wp,packet_time,(session->flags & RTP_SESSION_BLOCKING_MODE)!=0);	
			session_set_clr(&sched->w_sessions,session);	/* the session has written */
		}
		else session_set_set(&sched->w_sessions,session);	/*to indicate select to return immediately */
		wait_point_unlock(&session->snd.wp);
	}
	
	if(mp==NULL) {/*for people who just want to be blocked but
		 do not want to send anything.*/
		session->rtp.snd_last_ts = packet_ts;
		return 0;
	}

	rtp=(rtp_header_t*)mp->b_rptr;
	
	packsize = msgdsize(mp) ;
	
	rtp->timestamp=packet_ts;
	if (session->snd.telephone_events_pt==rtp->paytype)
	{
		rtp->seq_number = session->rtp.snd_seq;
		session->rtp.snd_seq++;
	}
	else
		session->rtp.snd_seq=rtp->seq_number+1;
	session->rtp.snd_last_ts = packet_ts;


	ortp_global_stats.sent += packsize;
	stream->sent_payload_bytes+=packsize-RTP_FIXED_HEADER_SIZE;
	stream->stats.sent += packsize;
	ortp_global_stats.packet_sent++;
	stream->stats.packet_sent++;

	error = rtp_session_rtp_send (session, mp);
	/*send RTCP packet if needed */
	rtp_session_rtcp_process_send(session);
	/* receives rtcp packet if session is send-only*/
	/*otherwise it is done in rtp_session_recvm_with_ts */
	if (session->mode==RTP_SESSION_SENDONLY) rtp_session_rtcp_recv(session);
	return error;
}

/**
 *	Send the rtp datagram @mp to the destination set by rtp_session_set_remote_addr() 
 *	with timestamp @timestamp. For audio data, the timestamp is the number
 *	of the first sample resulting of the data transmitted. See rfc1889 for details.
 *  The packet (@mp) is freed once it is sended.
 *
 *@param session a rtp session.
 *@param mp a rtp packet presented as a mblk_t.
 *@param timestamp the timestamp of the data to be sent.
 * @return the number of bytes sent over the network.
**/

int rtp_session_sendm_with_ts(RtpSession *session, mblk_t *packet, uint32_t timestamp){
	return __rtp_session_sendm_with_ts(session,packet,timestamp,timestamp);
}




/**
 *	Send a rtp datagram to the destination set by rtp_session_set_remote_addr() containing
 *	the data from @buffer with timestamp @userts. This is a high level function that uses
 *	rtp_session_create_packet() and rtp_session_sendm_with_ts() to send the data.
 *
 *@param session a rtp session.
 *@param buffer a buffer containing the data to be sent in a rtp packet.
 *@param len the length of the data buffer, in bytes.
 *@param userts	the timestamp of the data to be sent. Refer to the rfc to know what it is.
 *
 *@param return the number of bytes sent over the network.
**/
int
rtp_session_send_with_ts (RtpSession * session, const uint8_t * buffer, int len,
			  uint32_t userts)
{
	mblk_t *m;
	int err;
#ifdef USE_SENDMSG
	m=rtp_session_create_packet_with_data(session,(uint8_t*)buffer,len,NULL);
#else
	m = rtp_session_create_packet(session,RTP_FIXED_HEADER_SIZE,(uint8_t*)buffer,len);
#endif
	err=rtp_session_sendm_with_ts(session,m,userts);
	return err;
}



extern void rtcp_parse(RtpSession *session, mblk_t *mp);



static void payload_type_changed_notify(RtpSession *session, int paytype){
	session->rcv.pt = paytype;
	rtp_signal_table_emit (&session->on_payload_type_changed);	
}


/**
 *	Try to get a rtp packet presented as a mblk_t structure from the rtp session.
 *	The @user_ts parameter is relative to the first timestamp of the incoming stream. In other
 *	words, the application does not have to know the first timestamp of the stream, it can
 *	simply call for the first time this function with @user_ts=0, and then incrementing it
 *	as it want. The RtpSession takes care of synchronisation between the stream timestamp
 *	and the user timestamp given here.
 *
 *	This function returns the entire packet (with header).
 *
 *	The behaviour of this function has changed since version 0.15.0. Previously the payload data could be 
 *	accessed using  mblk_t::b_cont::b_rptr field of the returned mblk_t.
 *	This is no more the case.
 *	The convenient way of accessing the payload data is to use rtp_get_payload() :
 *	@code
 *	unsigned char *payload;
 *	int payload_size;
 *	payload_size=rtp_get_payload(mp,&payload);
 *	@endcode
 *	OR simply skip the header this way, the data is then comprised between mp->b_rptr and mp->b_wptr:
 *	@code
 *	rtp_get_payload(mp,&mp->b_rptr);
 *	@endcode
 *
 *
 * @param session a rtp session.
 * @param user_ts a timestamp.
 *
 * @return a rtp packet presented as a mblk_t.
**/

mblk_t *
rtp_session_recvm_with_ts (RtpSession * session, uint32_t user_ts)
{
	mblk_t *mp = NULL;
	rtp_header_t *rtp;
	uint32_t ts;
	uint32_t packet_time;
	RtpScheduler *sched=session->sched;
	RtpStream *stream=&session->rtp;
	int rejected=0;
	bool_t read_socket=TRUE;

	/* if we are scheduled, remember the scheduler time at which the application has
	 * asked for its first timestamp */

	if (session->flags & RTP_SESSION_RECV_NOT_STARTED)
	{
		session->rtp.rcv_query_ts_offset = user_ts;
		/* Set initial last_rcv_time to first recv time. */
		if ((session->flags & RTP_SESSION_SEND_NOT_STARTED)
		|| session->mode == RTP_SESSION_RECVONLY){
			gettimeofday(&session->last_recv_time, NULL);
		}
		if (session->flags & RTP_SESSION_SCHEDULED)
		{
			session->rtp.rcv_time_offset = sched->time_;
			//ortp_message("setting snd_time_offset=%i",session->rtp.snd_time_offset);
		}
		rtp_session_unset_flag (session,RTP_SESSION_RECV_NOT_STARTED);
	}else{
		/*prevent reading from the sockets when two 
		consecutives calls for a same timestamp*/
		if (user_ts==session->rtp.rcv_last_app_ts)
			read_socket=FALSE;
	}
	session->rtp.rcv_last_app_ts = user_ts;
	if (read_socket){
		rtp_session_rtp_recv (session, user_ts);
		rtp_session_rtcp_recv(session);
	}
	/* check for telephone event first */
	mp=getq(&session->rtp.tev_rq);
	if (mp!=NULL){
		int msgsize=msgdsize(mp);
		ortp_global_stats.recv += msgsize;
		stream->stats.recv += msgsize;
		rtp_signal_table_emit2(&session->on_telephone_event_packet,(long)mp);
		rtp_session_check_telephone_events(session,mp);
		freemsg(mp);
		mp=NULL;
	}
	
	/* then now try to return a media packet, if possible */
	/* first condition: if the session is starting, don't return anything
	 * until the queue size reaches jitt_comp */
	
	if (session->flags & RTP_SESSION_RECV_SYNC)
	{
		queue_t *q = &session->rtp.rq;
		if (qempty(q))
		{
			ortp_debug ("Queue is empty.");
			goto end;
		}
		rtp = (rtp_header_t *) qfirst(q)->b_rptr;
		session->rtp.rcv_ts_offset = rtp->timestamp;
		session->rtp.rcv_last_ret_ts = user_ts;	/* just to have an init value */
		session->rcv.ssrc = rtp->ssrc;
		/* delete the recv synchronisation flag */
		rtp_session_unset_flag (session, RTP_SESSION_RECV_SYNC);
	}

	/*calculate the stream timestamp from the user timestamp */
	ts = jitter_control_get_compensated_timestamp(&session->rtp.jittctl,user_ts);
	if (session->rtp.jittctl.enabled==TRUE){
		if (session->permissive)
			mp = rtp_getq_permissive(&session->rtp.rq, ts,&rejected);
		else{
			mp = rtp_getq(&session->rtp.rq, ts,&rejected);
		}
	}else mp=getq(&session->rtp.rq);/*no jitter buffer at all*/
	
	stream->stats.outoftime+=rejected;
	ortp_global_stats.outoftime+=rejected;

	goto end;

      end:
	if (mp != NULL)
	{
		int msgsize = msgdsize (mp);	/* evaluate how much bytes (including header) is received by app */
		uint32_t packet_ts;
		ortp_global_stats.recv += msgsize;
		stream->stats.recv += msgsize;
		rtp = (rtp_header_t *) mp->b_rptr;
		packet_ts=rtp->timestamp;
		ortp_debug("Returning mp with ts=%i", packet_ts);
		/* check for payload type changes */
		if (session->rcv.pt != rtp->paytype)
		{
			payload_type_changed_notify(session, rtp->paytype);
		}
		/* update the packet's timestamp so that it corrected by the 
		adaptive jitter buffer mechanism */
		if (session->rtp.jittctl.adaptive){
			uint32_t changed_ts;
			/* only update correction offset between packets of different
			timestamps*/
			if (packet_ts!=session->rtp.rcv_last_ts)
				jitter_control_update_corrective_slide(&session->rtp.jittctl);
			changed_ts=packet_ts+session->rtp.jittctl.corrective_slide;
			rtp->timestamp=changed_ts;
			/*ortp_debug("Returned packet has timestamp %u, with clock slide compensated it is %u",packet_ts,rtp->timestamp);*/
		}
		session->rtp.rcv_last_ts = packet_ts;
		if (!(session->flags & RTP_SESSION_FIRST_PACKET_DELIVERED)){
			rtp_session_set_flag(session,RTP_SESSION_FIRST_PACKET_DELIVERED);
		}
	}
	else
	{
		ortp_debug ("No mp for timestamp queried");
		stream->stats.unavaillable++;
		ortp_global_stats.unavaillable++;
	}
	rtp_session_rtcp_process_recv(session);
	
	if (session->flags & RTP_SESSION_SCHEDULED)
	{
		/* 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->rcv.wp);
		if (TIME_IS_STRICTLY_NEWER_THAN (packet_time, sched->time_))
		{
			wait_point_wakeup_at(&session->rcv.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->rcv.wp);
	}
	return mp;
}


/**
 *	NOTE: use of this function is discouraged when sending payloads other than
 *	pcm/pcmu/pcma/adpcm types.
 *	rtp_session_recvm_with_ts() does better job.
 *
 *	Tries to read the bytes of the incoming rtp stream related to timestamp ts. In case 
 *	where the user supplied buffer @buffer is not large enough to get all the data 
 *	related to timestamp ts, 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 @ts 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.
 *@param session a rtp session.
 *@param buffer a user supplied buffer to write the data.
 *@param len the length in bytes of the user supplied buffer.
 *@param ts the timestamp wanted.
 *@param have_more the address of an integer to indicate if more data is availlable for the given timestamp.
 *
**/
int rtp_session_recv_with_ts (RtpSession * session, uint8_t * buffer,
			       int len, uint32_t ts, int * have_more){
	mblk_t *mp=NULL;
	int plen,blen=0;
	*have_more=0;
	while(1){
		if (session->pending){
			mp=session->pending;
			session->pending=NULL;
		}else {
			mp=rtp_session_recvm_with_ts(session,ts);
			if (mp!=NULL) rtp_get_payload(mp,&mp->b_rptr);
		}
		if (mp){
			plen=mp->b_wptr-mp->b_rptr;
			if (plen<=len){
				memcpy(buffer,mp->b_rptr,plen);
				buffer+=plen;
				blen+=plen;
				len-=plen;
				freemsg(mp);
				mp=NULL;
			}else{
				memcpy(buffer,mp->b_rptr,len);
				mp->b_rptr+=len;
				buffer+=len;
				blen+=len;
				len=0;
				session->pending=mp;
				*have_more=1;
				break;
			}
		}else break;
	}
	return blen;
}
/**
 *	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().
 *
 * @param session a rtp session.
 * @return 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->snd.profile,session->snd.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;