rtpsession.c

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

	userts=  (uint32_t)( ( (double)(session_time) * (double) payload->clock_rate )/ 1000.0)
				+ session->rtp.snd_ts_offset;
	return userts;
}

/**
 * Same thing as rtp_session_get_current_send_ts() except that it's for an incoming stream.
 * Works only on scheduled mode.
 *
 * @param session a rtp session.
 * @return 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->rcv.profile,session->rcv.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;
}

/**
 * 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.
 * @param session the rtp session
 * @param ts_step a time interval in miliseconds
 *
**/
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;
}

/**
 * Closes the rtp and rtcp sockets.
**/
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;
	if (session->rtp.tr!=NULL)
	  ortp_free(session->rtp.tr);
	if (session->rtcp.tr!=NULL)
	  ortp_free(session->rtcp.tr);
	session->rtp.tr = 0;
	session->rtcp.tr = 0;

	/* don't discard remote addresses, then can be preserved for next use.
	session->rtp.rem_addrlen=0;
	session->rtcp.rem_addrlen=0;
	*/
}

ortp_socket_t rtp_session_get_rtp_socket(const RtpSession *session){
	return rtp_session_using_transport(session, rtp) ? (session->rtp.tr->t_getsocket)(session->rtp.tr) : session->rtp.socket;
}

ortp_socket_t rtp_session_get_rtcp_socket(const RtpSession *session){
	return rtp_session_using_transport(session, rtcp) ? (session->rtcp.tr->t_getsocket)(session->rtcp.tr) : session->rtcp.socket;
}

/**
 * Register an event queue.
 * An application can use an event queue to get informed about various RTP events.
**/
void rtp_session_register_event_queue(RtpSession *session, OrtpEvQueue *q){
	session->eventqs=o_list_append(session->eventqs,q);
}

void rtp_session_unregister_event_queue(RtpSession *session, OrtpEvQueue *q){
	session->eventqs=o_list_remove(session->eventqs,q);
}

void rtp_session_dispatch_event(RtpSession *session, OrtpEvent *ev){
	OList *it;
	int i;
	for(i=0,it=session->eventqs;it!=NULL;it=it->next,++i){
		ortp_ev_queue_put((OrtpEvQueue*)it->data,ortp_event_dup(ev));
	}	
	ortp_event_destroy(ev);
}


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);
	flushq(&session->rtp.tev_rq, FLUSHALL);

	if (session->eventqs!=NULL) o_list_free(session->eventqs);
	/* close sockets */
	rtp_session_release_sockets(session);

	wait_point_uninit(&session->snd.wp);
	wait_point_uninit(&session->rcv.wp);
	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);
	msgb_allocator_uninit(&session->allocator);
}

/**
 * Resynchronize to the incoming RTP streams.
 * This can be useful to handle discoutinuous timestamps.
 * For example, call this function from the timestamp_jump signal handler.
 * @param session the rtp session
**/
void rtp_session_resync(RtpSession *session){
	flushq (&session->rtp.rq, FLUSHALL);
	rtp_session_set_flag(session, RTP_SESSION_RECV_SYNC);
	rtp_session_unset_flag(session,RTP_SESSION_FIRST_PACKET_DELIVERED);
	jitter_control_init(&session->rtp.jittctl,-1,NULL);
}

/**
 * Reset the session: local and remote addresses are kept. It resets timestamp, sequence 
 * number, and calls rtp_session_resync().
 *
 * @param session a rtp session.
**/
void rtp_session_reset (RtpSession * session)
{
	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_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;
	session->rtp.sent_payload_bytes=0;
	rtp_session_clear_send_error_code(session);
	rtp_session_clear_recv_error_code(session);
	rtp_stats_reset(&session->rtp.stats);
	rtp_session_resync(session);
	session->ssrc_set=FALSE;
}

/**
 * Retrieve the session's statistics.
**/
const rtp_stats_t * rtp_session_get_stats(const RtpSession *session){
	return &session->rtp.stats;
}

void rtp_session_reset_stats(RtpSession *session){
	memset(&session->rtp.stats,0,sizeof(rtp_stats_t));
}

/**
 * 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().
 * @param session a rtp session
 * @param data an opaque pointer to be stored in the session
**/

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

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

/**
 * Enable or disable the "rtp symmetric" hack which consists of the following:
 * after the first packet is received, the source address of the packet 
 * is set to be the destination address for all next packets.
 * This is useful to pass-through firewalls.
 * @param session a rtp session
 * @param yesno a boolean to enable or disable the feature
 *
**/
void
rtp_session_set_symmetric_rtp (RtpSession * session, bool_t yesno)
{
	session->symmetric_rtp =yesno;
}

/**
 *	If yesno is TRUE, thus a connect() syscall is done on the socket to 
 *	the destination address set by rtp_session_set_remote_addr(), or
 *	if the session does symmetric rtp (see rtp_session_set_symmetric_rtp())
 *	a the connect() is done to the source address of the first packet received.
 *	Connecting a socket has effect of rejecting all incoming packets that 
 *	don't come from the address specified in connect().
 *	It also makes ICMP errors (such as connection refused) available to the
 *	application.
 *	@param session a rtp session
 *	@param yesno a boolean to enable or disable the feature
 *
**/
void rtp_session_set_connected_mode(RtpSession *session, bool_t yesno){
	session->use_connect=yesno;
}

static float compute_bw(struct timeval *orig, unsigned int bytes){
	struct timeval current;
	float bw;
	float time;
	if (bytes==0) return 0;
	gettimeofday(&current,NULL);
	time=(float)(current.tv_sec - orig->tv_sec) +
		((float)(current.tv_usec - orig->tv_usec)*1e-6);
	bw=((float)bytes)*8/(time+0.001); 
	/*+0.0001 avoids a division by zero without changing the results significatively*/
	return bw;
}

float rtp_session_compute_recv_bandwidth(RtpSession *session){
	float bw;
	bw=compute_bw(&session->rtp.recv_bw_start,session->rtp.recv_bytes);
	session->rtp.recv_bytes=0;
	return bw;
}

float rtp_session_compute_send_bandwidth(RtpSession *session){
	float bw;
	bw=compute_bw(&session->rtp.send_bw_start,session->rtp.sent_bytes);
	session->rtp.sent_bytes=0;
	return bw;
}

int rtp_session_get_last_send_error_code(RtpSession *session){
	return session->rtp.send_errno;
}

void rtp_session_clear_send_error_code(RtpSession *session){
	session->rtp.send_errno=0;
}

int rtp_session_get_last_recv_error_code(RtpSession *session){
	return session->rtp.recv_errno;
}

void rtp_session_clear_recv_error_code(RtpSession *session){
	session->rtp.send_errno=0;
}

/**
 * Destroys a rtp session.
 * All memory allocated for the RtpSession is freed.
 *
 * @param session a rtp session.
**/
void rtp_session_destroy (RtpSession * session)
{
	rtp_session_uninit (session);
	ortp_free (session);
}

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++;
}

/**
 * Get a pointer to the beginning of the payload data of the RTP packet.
 * @param packet a RTP packet represented as a mblk_t
 * @param start a pointer to the beginning of the payload data, pointing inside the packet.
 * @return the length of the payload data.
**/
int rtp_get_payload(mblk_t *packet, unsigned char **start){
	unsigned char *tmp;
	int header_len=RTP_FIXED_HEADER_SIZE+(rtp_get_cc(packet)*4);
	tmp=packet->b_rptr+header_len;
	if (tmp>packet->b_wptr){
		if (packet->b_cont!=NULL){
			tmp=packet->b_cont->b_rptr+(header_len- (packet->b_wptr-packet->b_rptr));
			if (tmp<=packet->b_cont->b_wptr){
				*start=tmp;
				return packet->b_cont->b_wptr-tmp;
			}
		}
		ortp_warning("Invalid RTP packet");
		return -1;
	}
	*start=tmp;
	return packet->b_wptr-tmp;
}


/**
 *  Gets last time a valid RTP or RTCP packet was received.
 * @param session RtpSession to get last receive time from.
 * @param tv Pointer to struct timeval to fill.
 *
**/
void
rtp_session_get_last_recv_time(RtpSession *session, struct timeval *tv)
{
#ifdef PERF
	ortp_error("rtp_session_get_last_recv_time() feature disabled.");
#else
    	*tv = session->last_recv_time;
#endif
}



uint32_t rtp_session_time_to_ts(RtpSession *session, int millisecs){
	PayloadType *payload;
	payload =
		rtp_profile_get_payload (session->snd.profile,
					 session->snd.pt);
	if (payload == NULL)
	{
		ortp_warning
			("rtp_session_ts_to_t: use of unsupported payload type %d.", session->snd.pt);
		return 0;
	}
	/* the return value is in milisecond */
	return (uint32_t) (payload->clock_rate*(double) (millisecs/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->snd.profile,
					 session->snd.pt);
	if (payload == NULL)
	{
		ortp_warning
			("rtp_session_ts_to_t: use of unsupported payload type %d.", session->snd.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->snd.wp);
	if (wait_point_check(&session->snd.wp,time)){
		session_set_set(&sched->w_sessions,session);
		wait_point_wakeup(&session->snd.wp);
	}
	wait_point_unlock(&session->snd.wp);
	
	wait_point_lock(&session->rcv.wp);
	if (wait_point_check(&session->rcv.wp,time)){
		session_set_set(&sched->r_sessions,session);
		wait_point_wakeup(&session->rcv.wp);
	}
	wait_point_unlock(&session->rcv.wp);
}