path: root/kopete/protocols/jabber/jingle/libjingle/talk/third_party/ortp/rtpsession.c

/*
  The oRTP library is an RTP (Realtime Transport Protocol - rfc1889) stack.
  Copyright (C) 2001  Simon MORLAT simon.morlat@linphone.org

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/


#include <ortp.h>
#include <telephonyevents.h>
#include "rtpmod.h"
#include "jitterctl.h"
#include "scheduler.h"
#include "port_fct.h"
#include "utils.h"

#include <fcntl.h>
#include <errno.h>
#include <stdlib.h>
#include <stdio.h>

#ifndef _WIN32
# include <sys/types.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <arpa/inet.h>
# ifdef INET6
#  include <netdb.h>
# endif
#else
# include <winsock2.h>
# include "errno-win32.h"
#endif


#if defined(HAVE_POLL_H)
#include <poll.h>
#elif defined(HAVE_SYS_POLL_H)
#include <sys/poll.h>
#endif
#ifdef HAVE_SYS_UIO_H
#include <sys/uio.h>
#define USE_SENDMSG 1
#endif



void wait_point_init(WaitPoint *wp){
	wp->lock=g_mutex_new();
	wp->cond=g_cond_new();
	wp->time=0;
	wp->wakeup=FALSE;
}
void wait_point_uninit(WaitPoint *wp){
	g_cond_free(wp->cond);
	g_mutex_free(wp->lock);
}

#define wait_point_lock(wp) g_mutex_lock((wp)->lock)
#define wait_point_unlock(wp) g_mutex_unlock((wp)->lock)

void wait_point_wakeup_at(WaitPoint *wp, guint32 t, gboolean dosleep){
	wp->time=t;
	wp->wakeup=TRUE;
	if (dosleep) g_cond_wait(wp->cond,wp->lock);
}


gboolean wait_point_check(WaitPoint *wp, guint32 t){
	gboolean ok=FALSE;
	
	if (wp->wakeup){
		if (TIME_IS_NEWER_THAN(t,wp->time)){
			wp->wakeup=FALSE;
			ok=TRUE;
			
		}
	}
	return ok;
}
#define wait_point_wakeup(wp) g_cond_signal((wp)->cond);

extern void rtp_parse(RtpSession *session, mblk_t *mp, guint32 local_str_ts);


static guint32 guint32_random(){
	return random();
}

void
rtp_session_init (RtpSession * session, gint mode)
{
	memset (session, 0, sizeof (RtpSession));
	session->lock = g_mutex_new ();
	session->rtp.max_rq_size = RTP_MAX_RTQ_SIZE;
	session->mode = mode;
	if ((mode == RTP_SESSION_RECVONLY) || (mode == RTP_SESSION_SENDRECV))
	{
		rtp_session_set_flag (session, RTP_SESSION_RECV_SYNC);
		rtp_session_set_flag (session, RTP_SESSION_RECV_NOT_STARTED);
		
	}
	if ((mode == RTP_SESSION_SENDONLY) || (mode == RTP_SESSION_SENDRECV))
	{
		rtp_session_set_flag (session, RTP_SESSION_SEND_NOT_STARTED);
		rtp_session_set_flag (session, RTP_SESSION_SEND_SYNC);
		session->send_ssrc=guint32_random();
		/* set default source description */
		rtp_session_set_source_description(session,"unknown@unknown",NULL,NULL,
				NULL,NULL,"oRTP-" ORTP_VERSION,"This is free sofware (LGPL) !");
	}
	session->telephone_events_pt=-1;	/* not defined a priori */
	rtp_session_set_profile (session, &av_profile); /*the default profile to work with */
	session->payload_type=0;/* default to something */
	qinit(&session->rtp.rq);
	qinit(&session->rtp.tev_rq);
	qinit(&session->contributing_sources);
	/* init signal tables */
	rtp_signal_table_init (&session->on_ssrc_changed, session,"ssrc_changed");
	rtp_signal_table_init (&session->on_payload_type_changed, session,"payload_type_changed");
	rtp_signal_table_init (&session->on_telephone_event, session,"telephone-event");
	rtp_signal_table_init (&session->on_telephone_event_packet, session,"telephone-event_packet");
	rtp_signal_table_init (&session->on_timestamp_jump,session,"timestamp_jump");
	rtp_signal_table_init (&session->on_network_error,session,"network_error");
	wait_point_init(&session->send_wp);
	wait_point_init(&session->recv_wp);
	rtp_session_set_jitter_compensation(session,RTP_DEFAULT_JITTER_TIME);
	rtp_session_enable_adaptive_jitter_compensation(session,FALSE);
	rtp_session_set_time_jump_limit(session,5000);
	session->max_buf_size = UDP_MAX_SIZE;
}

/**
 *rtp_session_new:
 *@mode: One of the #RtpSessionMode flags.
 *
 *	Creates a new rtp session.
 *  If the session is able to send data (RTP_SESSION_SENDONLY or RTP_SESSION_SENDRECV), then a
 *	random SSRC number is choosed for the outgoing stream.
 *
 *Returns: the newly created rtp session.
**/

RtpSession *
rtp_session_new (gint mode)
{
	RtpSession *session;
	session = g_malloc (sizeof (RtpSession));
	rtp_session_init (session, mode);
	return session;
}

/**
 *rtp_session_set_scheduling_mode:
 *@session: a rtp session.
 *@yesno:	a boolean to indicate the scheduling mode.
 *
 *	Sets the scheduling mode of the rtp session. If @yesno is TRUE, the rtp session is in
 *	the scheduled mode, that means that you can use session_set_select() to block until it's time
 *	to receive or send on this session according to the timestamp passed to the respective functions.
 *  You can also use blocking mode (see rtp_session_set_blocking_mode() ), to simply block within
 *	the receive and send functions.
 *	If @yesno is FALSE, the ortp scheduler will not manage those sessions, meaning that blocking mode 
 *  and the use of session_set_select() for this session are disabled.
 *
**/

void
rtp_session_set_scheduling_mode (RtpSession * session, gint yesno)
{
	if (yesno)
	{
		RtpScheduler *sched;
		sched = ortp_get_scheduler ();
		if (sched != NULL)
		{
			rtp_session_set_flag (session, RTP_SESSION_SCHEDULED);
			session->sched = sched;
			rtp_scheduler_add_session (sched, session);
		}
		else
			g_warning
				("rtp_session_set_scheduling_mode: Cannot use scheduled mode because the "
				 "scheduler is not started. Use ortp_scheduler_init() before.");
	}
	else
		rtp_session_unset_flag (session, RTP_SESSION_SCHEDULED);
}


/**
 *rtp_session_set_blocking_mode:
 *@session: a rtp session
 *@yesno: a boolean
 *
 *	Using this function implies that you previously enabled scheduled mode on the session
 *  (see rtp_session_set_scheduling_mode() ).
 *	rtp_session_set_blocking_mode() defines the behaviour of the rtp_session_recv_with_ts() and 
 *	rtp_session_send_with_ts() functions. If @yesno is TRUE, rtp_session_recv_with_ts()
 *	will block until it is time for the packet to be received, according to the timestamp
 *	passed to the function. After this time, the function returns.
 *	For rtp_session_send_with_ts(), it will block until it is time for the packet to be sent.
 *	If @yesno is FALSE, then the two functions will return immediately.
 *
**/
void
rtp_session_set_blocking_mode (RtpSession * session, gint yesno)
{
	if (yesno)
		rtp_session_set_flag (session, RTP_SESSION_BLOCKING_MODE);
	else
		rtp_session_unset_flag (session, RTP_SESSION_BLOCKING_MODE);
}

/**
 *rtp_session_set_profile:
 *@session: a rtp session
 *@profile: a rtp profile
 *
 *	Set the RTP profile to be used for the session. By default, all session are created by
 *	rtp_session_new() are initialized with the AV profile, as defined in RFC 1890. The application
 *	can set any other profile instead using that function.
 *
 *
**/

void
rtp_session_set_profile (RtpSession * session, RtpProfile * profile)
{
	session->profile = profile;
	rtp_session_telephone_events_supported(session);
}


/**
 *rtp_session_signal_connect:
 *@session: 	a rtp session
 *@signal:		the name of a signal
 *@cb:			a #RtpCallback
 *@user_data:	a pointer to any data to be passed when invoking the callback.
 *
 *	This function provides the way for an application to be informed of various events that
 *	may occur during a rtp session. @signal is a string identifying the event, and @cb is 
 *	a user supplied function in charge of processing it. The application can register
 *	several callbacks for the same signal, in the limit of #RTP_CALLBACK_TABLE_MAX_ENTRIES.
 *	Here are name and meaning of supported signals types:
 *
 *	"ssrc_changed" : the SSRC of the incoming stream has changed.
 *
 *	"payload_type_changed" : the payload type of the incoming stream has changed.
 *
 *	"telephone-event_packet" : a telephone-event rtp packet (RFC2833) is received.
 *
 *	"telephone-event" : a telephone event has occured. This is a high-level shortcut for "telephone-event_packet".
 *
 *	"network_error" : a network error happened on a socket. Arguments of the callback functions are
 *						a const char * explaining the error, an int errno error code and the user_data as usual.
 *
 *	"timestamp_jump" : we have received a packet with timestamp in far future compared to last timestamp received.
 *						The farness of far future is set by rtp_sesssion_set_time_jump_limit()
 *
 *	Returns: 0 on success, -EOPNOTSUPP if the signal does not exists, -1 if no more callbacks
 *	can be assigned to the signal type.
**/
int
rtp_session_signal_connect (RtpSession * session, const char *signal,
			    RtpCallback cb, gpointer user_data)
{
	OList *elem;
	for (elem=session->signal_tables;elem!=NULL;elem=o_list_next(elem)){
		RtpSignalTable *s=(RtpSignalTable*) elem->data;
		if (strcmp(signal,s->signal_name)==0){
			return rtp_signal_table_add(s,cb,user_data);
		}
	}
	g_warning ("rtp_session_signal_connect: inexistant signal %s",signal);
	return -1;
}


/**
 *rtp_session_signal_disconnect_by_callback:
 *@session: a rtp session
 *@signal:	a signal name
 *@cb:		a callback function.
 *
 *	Removes callback function @cb to the list of callbacks for signal @signal.
 *
 *Returns: 0 on success, -ENOENT if the callbacks was not found.
**/

int
rtp_session_signal_disconnect_by_callback (RtpSession * session, const gchar *signal,
					   RtpCallback cb)
{
	OList *elem;
	for (elem=session->signal_tables;elem!=NULL;elem=o_list_next(elem)){
		RtpSignalTable *s=(RtpSignalTable*) elem->data;
		if (strcmp(signal,s->signal_name)==0){
			return rtp_signal_table_remove_by_callback(s,cb);
		}
	}
	g_warning ("rtp_session_signal_connect: inexistant signal %s",signal);
	return -1;
}

/**
 *rtp_session_set_local_addr:
 *@session:		a rtp session freshly created.
 *@addr:		a local IP address in the xxx.xxx.xxx.xxx form.
 *@port:		a local port.
 *
 *	Specify the local addr to be use to listen for rtp packets or to send rtp packet from.
 *	In case where the rtp session is send-only, then it is not required to call this function:
 *	when calling rtp_session_set_remote_addr(), if no local address has been set, then the 
 *	default INADRR_ANY (0.0.0.0) IP address with a random port will be used. Calling 
 *	rtp_sesession_set_local_addr() is mandatory when the session is recv-only or duplex.
 *
 *	Returns: 0 on success.
**/

gint
rtp_session_set_local_addr (RtpSession * session, const gchar * addr, gint port)
{
	gint err;
	gint optval = 1;
#ifdef INET6
	char num[8];
	struct addrinfo hints, *res0, *res;
#endif
	
	if (session->rtp.socket>0) {
		/* dont try to rebind, close socket before */
		close_socket(session->rtp.socket);
		close_socket(session->rtcp.socket);
		session->rtp.socket=0;
		session->rtcp.socket=0;
	}
	
#ifdef INET6
	
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = PF_UNSPEC;
	hints.ai_socktype = SOCK_DGRAM;
	snprintf(num, sizeof(num), "%d",port);
	err = getaddrinfo(addr,num, &hints, &res0);
	if (err!=0) {
		g_warning ("Error: %s", gai_strerror(err));
		return err;
	}
	
	for (res = res0; res; res = res->ai_next) {
		session->rtp.socket = socket(res->ai_family, res->ai_socktype, 0);
		if (session->rtp.socket < 0)
		  continue;
                
		err = setsockopt (session->rtp.socket, SOL_SOCKET, SO_REUSEADDR,
				  (void*)&optval, sizeof (optval));
		if (err < 0)
		{
			g_warning ("Fail to set rtp address reusable: %s.", getSocketError());
		}

		session->rtp.socktype=res->ai_family;
		memcpy(&session->rtp.loc_addr, res->ai_addr, res->ai_addrlen);
		err = bind (session->rtp.socket, res->ai_addr, res->ai_addrlen);
		if (err != 0)
		  {
		    g_warning ("Fail to bind rtp socket to port %i: %s.", port, getSocketError());
		    close_socket (session->rtp.socket);
		    continue;
		  }
#ifndef __hpux
		switch (res->ai_family)
		  {
		    case AF_INET:
		      if (IN_MULTICAST(ntohl(((struct sockaddr_in *) res->ai_addr)->sin_addr.s_addr)))
			{
		          struct ip_mreq mreq;
			  mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *) res->ai_addr)->sin_addr.s_addr;
			  mreq.imr_interface.s_addr = INADDR_ANY;
			  err = setsockopt(session->rtp.socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
			  if (err < 0)
			    {
			      g_warning ("Fail to join address group: %s.", getSocketError());
			      close_socket (session->rtp.socket);
			      continue;
			    }
			}
		      break;
		    case AF_INET6:
		      if (IN6_IS_ADDR_MULTICAST(&(((struct sockaddr_in6 *) res->ai_addr)->sin6_addr)))
			{
			  struct ipv6_mreq mreq;
			  mreq.ipv6mr_multiaddr = ((struct sockaddr_in6 *) res->ai_addr)->sin6_addr;
			  mreq.ipv6mr_interface = 0;
			  err = setsockopt(session->rtp.socket, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq, sizeof(mreq));
			  if (err < 0)
			    {
			      g_warning ("Fail to join address group: %s.", getSocketError());
			      close_socket (session->rtp.socket);
			      continue;
			    }
			}
		      break;
		  }
#endif
		break;
	}
	freeaddrinfo(res0);
	if (session->rtp.socket < 0){
		if (session->mode==RTP_SESSION_RECVONLY) g_warning("Could not create rtp socket with address %s: %s",addr,getSocketError());
		return -1;
	}

	memset(&hints, 0, sizeof(hints));
	hints.ai_family = PF_UNSPEC;
	hints.ai_socktype = SOCK_DGRAM;
	snprintf(num, sizeof(num), "%d", (port + 1));

	err = getaddrinfo(addr, num, &hints, &res0);
	if (err!=0) {
		g_warning ("Error: %s", gai_strerror(err));
		return err;
	}
	
	for (res = res0; res; res = res->ai_next) {
		session->rtcp.socket = socket(res->ai_family, res->ai_socktype, 0);

		if (session->rtcp.socket < 0)
		  continue;

		err = setsockopt (session->rtcp.socket, SOL_SOCKET, SO_REUSEADDR,
				  (void*)&optval, sizeof (optval));
		if (err < 0)
		{
			g_warning ("Fail to set rtcp address reusable: %s.",getSocketError());
		}
		session->rtcp.socktype=res->ai_family;
		memcpy( &session->rtcp.loc_addr, res->ai_addr, res->ai_addrlen);
		err = bind (session->rtcp.socket, res->ai_addr, res->ai_addrlen);
		if (err != 0)
		  {
		    g_warning ("Fail to bind rtp socket to port %i: %s.", port, getSocketError());
		    close_socket (session->rtp.socket);
		    close_socket (session->rtcp.socket);
		    continue;
		  }
#ifndef __hpux
		switch (res->ai_family)
		  {
		    case AF_INET:
		      if (IN_MULTICAST(ntohl(((struct sockaddr_in *) res->ai_addr)->sin_addr.s_addr)))
			{
		          struct ip_mreq mreq;
			  mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *) res->ai_addr)->sin_addr.s_addr;
			  mreq.imr_interface.s_addr = INADDR_ANY;
			  err = setsockopt(session->rtcp.socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
			  if (err < 0)
			    {
			      g_warning ("Fail to join address group: %s.", getSocketError());
			      close_socket (session->rtp.socket);
			      close_socket (session->rtcp.socket);
			      continue;
			    }
			}
		      break;
		    case AF_INET6:
		      if (IN6_IS_ADDR_MULTICAST(&(((struct sockaddr_in6 *) res->ai_addr)->sin6_addr)))
			{
			  struct ipv6_mreq mreq;
			  mreq.ipv6mr_multiaddr = ((struct sockaddr_in6 *) res->ai_addr)->sin6_addr;
			  mreq.ipv6mr_interface = 0;
			  err = setsockopt(session->rtcp.socket, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq, sizeof(mreq));
			  if (err < 0)
			    {
			      g_warning ("Fail to join address group: %s.", getSocketError());
			      close_socket (session->rtp.socket);
			      close_socket (session->rtcp.socket);
			      continue;
			    }
			}
		      break;
		  }
#endif
		
		break;
	}
	freeaddrinfo(res0);
	if (session->rtp.socket < 0){
		g_warning("Could not create rtcp socket with address %s: %s",addr,getSocketError());
		return -1;
	}
#else
	session->rtp.loc_addr.sin_family = AF_INET;

	err = inet_aton (addr, &session->rtp.loc_addr.sin_addr);

	if (err < 0)
	{
		g_warning ("Error in socket address:%s.", getSocketError());
		return err;
	}
	session->rtp.loc_addr.sin_port = htons (port);

	session->rtp.socket = socket (PF_INET, SOCK_DGRAM, 0);
	g_return_val_if_fail (session->rtp.socket > 0, -1);
	
	err = setsockopt (session->rtp.socket, SOL_SOCKET, SO_REUSEADDR,
			  (void*)&optval, sizeof (optval));
	if (err < 0)
	{
		g_warning ("Fail to set rtp address reusable: %s.",getSocketError());
	}

	err = bind (session->rtp.socket,
		    (struct sockaddr *) &session->rtp.loc_addr,
		    sizeof (struct sockaddr_in));

	if (err != 0)
	{
		g_warning ("Fail to bind rtp socket to port %i: %s.", port, getSocketError());
		close_socket (session->rtp.socket);
		return -1;
	}
	memcpy (&session->rtcp.loc_addr, &session->rtp.loc_addr,
		sizeof (struct sockaddr_in));
	session->rtcp.loc_addr.sin_port = htons (port + 1);
	session->rtcp.socket = socket (PF_INET, SOCK_DGRAM, 0);
	g_return_val_if_fail (session->rtcp.socket > 0, -1);

	err = setsockopt (session->rtcp.socket, SOL_SOCKET, SO_REUSEADDR,
			  (void*)&optval, sizeof (optval));
	if (err < 0)
	{
		g_warning ("Fail to set rtcp address reusable: %s.",getSocketError());
	}

	err = bind (session->rtcp.socket,
		    (struct sockaddr *) &session->rtcp.loc_addr,
		    sizeof (struct sockaddr_in));
	if (err != 0)
	{
		g_warning ("Fail to bind rtcp socket to port %i: %s.", port + 1, getSocketError());
		close_socket (session->rtp.socket);
		close_socket (session->rtcp.socket);
		return -1;
	}
#ifndef __hpux
	if (IN_MULTICAST(ntohl(session->rtp.loc_addr.sin_addr.s_addr)))
	  {
	    struct ip_mreq mreq;
	    mreq.imr_multiaddr.s_addr = session->rtp.loc_addr.sin_addr.s_addr;
	    mreq.imr_interface.s_addr = INADDR_ANY;
	    err = setsockopt(session->rtp.socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
	    if (err == 0)
	      err = setsockopt(session->rtcp.socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
	    if (err < 0)
	      {
		g_warning ("Fail to join address group: %s.", getSocketError());
		close_socket (session->rtp.socket);
		close_socket (session->rtcp.socket);
		return -1;
	      }
	  }
#endif
#endif
	/* set RTP socket options */
	set_non_blocking_socket (session->rtp.socket);
	/* set RTCP socket options */
	set_non_blocking_socket (session->rtcp.socket);
	return 0;
}


/**
 *rtp_session_set_remote_addr:
 *@session:		a rtp session freshly created.
 *@addr:		a local IP address in the xxx.xxx.xxx.xxx form.
 *@port:		a local port.
 *
 *	Sets the remote address of the rtp session, ie the destination address where rtp packet
 *	are sent. If the session is recv-only or duplex, it also sets the origin of incoming RTP 
 *	packets. Rtp packets that don't come from addr:port are discarded.
 *
 *	Returns: 0 on success.
**/

gint
rtp_session_set_remote_addr (RtpSession * session, const gchar * addr, gint port)
{
	gint err;
#ifdef INET6
	struct addrinfo hints, *res0, *res;
	char num[8];
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = PF_UNSPEC;
	hints.ai_socktype = SOCK_DGRAM;
	snprintf(num, sizeof(num), "%d", port);
	err = getaddrinfo(addr, num, &hints, &res0);
	if (err) {
		g_warning ("Error in socket address: %s", gai_strerror(err));
		return err;
	}
#endif

	if (session->rtp.socket == 0)
	{
		int retry;
		/* the session has not its socket bound, do it */
		g_message ("Setting random local addresses.");
		for (retry=0;retry<10;retry++)
		{
			int localport;
			do
			{
				localport = (rand () + 5000) & 0xfffe;
			}
			while ((localport < 5000) || (localport > 0xffff));
#ifdef INET6
			/* bind to an address type that matches the destination address */
			if (res0->ai_addr->sa_family==AF_INET6)
				err = rtp_session_set_local_addr (session, "::", localport);
			else err=rtp_session_set_local_addr (session, "0.0.0.0", localport);
#else
			err = rtp_session_set_local_addr (session, "0.0.0.0", localport);
#endif

			if (err == 0)
				break;
		}
		if (retry == 10){
			g_warning("rtp_session_set_remote_addr: Could not find a random local address for socket !");
			return -1;
		}
	}


#ifdef INET6
	err=1;
	for (res = res0; res; res = res->ai_next) {
		/* set a destination address that has the same type as the local address */
		if (res->ai_family==session->rtp.socktype ) {
			memcpy( &session->rtp.rem_addr, res->ai_addr, res->ai_addrlen);
			session->rtp.addrlen=res->ai_addrlen;
		  	err=0;
		  	break;
		}
	}
	freeaddrinfo(res0);
	if (err) {
		g_warning("Could not set destination for RTP socket to %s:%i.",addr,port);
		return -1;
	}
	
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = PF_UNSPEC;
	hints.ai_socktype = SOCK_DGRAM;
	snprintf(num, sizeof(num), "%d", (port + 1));
	err = getaddrinfo(addr, num, &hints, &res0);
	if (err) {
		g_warning ("Error: %s", gai_strerror(err));
		return err;
	}
	err=1;
	for (res = res0; res; res = res->ai_next) {
		/* set a destination address that has the same type as the local address */
		if (res->ai_family==session->rtp.socktype ) {
		  	err=0;
		  	memcpy( &session->rtcp.rem_addr, res->ai_addr, res->ai_addrlen);
			session->rtcp.addrlen=res->ai_addrlen;
		  	break;
		}
	}
	freeaddrinfo(res0);
	if (err) {
		g_warning("Could not set destination for RCTP socket to %s:%i.",addr,port+1);
		return -1;
	}
#else
	session->rtp.addrlen=sizeof(session->rtp.rem_addr);
	session->rtp.rem_addr.sin_family = AF_INET;

	err = inet_aton (addr, &session->rtp.rem_addr.sin_addr);
	if (err < 0)
	{
		g_warning ("Error in socket address:%s.", getSocketError());
		return err;
	}
	session->rtp.rem_addr.sin_port = htons (port);

	memcpy (&session->rtcp.rem_addr, &session->rtp.rem_addr,
		sizeof (struct sockaddr_in));
	session->rtcp.rem_addr.sin_port = htons (port + 1);
	session->rtcp.addrlen=sizeof(session->rtp.rem_addr);
#endif
#ifndef NOCONNECT
	if (session->mode == RTP_SESSION_SENDONLY)
	{
		err = connect (session->rtp.socket,
			       (struct sockaddr *) &session->rtp.rem_addr,
#ifdef INET6
			       session->rtp.addrlen);
#else
			       sizeof (struct sockaddr_in));
#endif
		if (err != 0)
		{
			g_message ("Can't connect rtp socket: %s.",getSocketError());
			return err;
		}
		err = connect (session->rtcp.socket,
			       (struct sockaddr *) &session->rtcp.rem_addr,
#ifdef INET6
			       session->rtcp.addrlen);
#else
			       sizeof (struct sockaddr_in));
#endif
		if (err != 0)
		{
			g_message ("Can't connect rtp socket: %s.",getSocketError());
			return err;
		}
	}
#endif
	return 0;
}

void rtp_session_set_sockets(RtpSession *session, gint rtpfd, gint rtcpfd)
{
	if (rtpfd>0) set_non_blocking_socket(rtpfd);
	if (rtcpfd>0) set_non_blocking_socket(rtcpfd);
	session->rtp.socket=rtpfd;
	session->rtcp.socket=rtcpfd;
	session->flags|=RTP_SESSION_USING_EXT_SOCKETS;
}

/**
 *rtp_session_flush_sockets:
 *@session: a rtp session
 *
 * Flushes the sockets for all pending incoming packets.
 * This can be usefull if you did not listen to the stream for a while
 * and wishes to start to receive again. During the time no receive is made
 * packets get bufferised into the internal kernel socket structure.
 *
**/
void rtp_session_flush_sockets(RtpSession *session){
	char trash[4096];
#ifdef INET6
	struct sockaddr_storage from;
#else
	struct sockaddr from;
#endif
	socklen_t fromlen=sizeof(from);
	if (session->rtp.socket>0){
		while (recvfrom(session->rtp.socket,(void*)trash,sizeof(trash),0,(struct sockaddr *)&from,&fromlen)>0){};
	}
	if (session->rtcp.socket>0){
		while (recvfrom(session->rtcp.socket,(void*)trash,sizeof(trash),0,(struct sockaddr*)&from,&fromlen)>0){};
	}
}

/**
 *rtp_session_set_seq_number:
 *@session:		a rtp session freshly created.
 *@addr:			a 16 bit unsigned number.
 *
 * sets the initial sequence number of a sending session.
 *
**/
void rtp_session_set_seq_number(RtpSession *session, guint16 seq){
	session->rtp.snd_seq=seq;
}


guint16 rtp_session_get_seq_number(RtpSession *session){
	return session->rtp.snd_seq;
}


#ifdef USE_SENDMSG 
#define MAX_IOV 10
static gint rtp_sendmsg(int sock,mblk_t *m, struct sockaddr *rem_addr, int addr_len){
	int error;
	struct msghdr msg;
	struct iovec iov[MAX_IOV];
	int iovlen;
	for(iovlen=0; iovlen<MAX_IOV && m!=NULL; m=m->b_cont,iovlen++){
		iov[iovlen].iov_base=m->b_rptr;
		iov[iovlen].iov_len=m->b_wptr-m->b_rptr;
	}
	msg.msg_name=(void*)rem_addr;
	msg.msg_namelen=addr_len;
	msg.msg_iov=&iov[0];
	msg.msg_iovlen=iovlen;
	msg.msg_control=NULL;
	msg.msg_controllen=0;
	msg.msg_flags=0;
	
	error=sendmsg(sock,&msg,0);
	return error;
}
#endif	

static gint
ortp_rtp_send (RtpSession * session, mblk_t * m)
{
	gint error;
	int i;
	rtp_header_t *hdr;

	hdr = (rtp_header_t *) m->b_rptr;
	/* perform host to network conversions */
	hdr->ssrc = htonl (hdr->ssrc);
	hdr->timestamp = htonl (hdr->timestamp);
	hdr->seq_number = htons (hdr->seq_number);
	for (i = 0; i < hdr->cc; i++)
		hdr->csrc[i] = htonl (hdr->csrc[i]);
	
#ifdef USE_SENDMSG
	if (session->flags & RTP_SESSION_USING_EXT_SOCKETS){
		error=rtp_sendmsg(session->rtp.socket,m,(struct sockaddr *)NULL,0);
	}else {
		error=rtp_sendmsg(session->rtp.socket,m,(struct sockaddr *) &session->rtp.rem_addr,
		session->rtp.addrlen);
	}
#else
	if (m->b_cont!=NULL){
		mblk_t *newm=msgpullup(m,-1);
		freemsg(m);
		m=newm;
	}
	if (session->flags & RTP_SESSION_USING_EXT_SOCKETS){
		error=send(session->rtp.socket, m->b_rptr, (m->b_wptr - m->b_rptr),0);
	}else error = sendto (session->rtp.socket, m->b_rptr,
		(m->b_wptr - m->b_rptr), 0,
		(struct sockaddr *) &session->rtp.rem_addr,
		session->rtp.addrlen);
#endif
	if (error < 0){
		if (session->on_network_error.count>0){
			rtp_signal_table_emit3(&session->on_network_error,(gpointer)"Error sending RTP packet",INT_TO_POINTER(getSocketErrorCode()));
		}else g_warning ("Error sending rtp packet: %s ; socket=%i", getSocketError(), session->rtp.socket);
	}
	freemsg (m);
	return error;
}

gint
ortp_rtcp_send (RtpSession * session, mblk_t * m)
{
	gint error=0;
	gboolean using_ext_socket=(session->flags & RTP_SESSION_USING_EXT_SOCKETS)!=0;
	if ( (using_ext_socket && session->rtcp.socket>0 ) || session->rtcp.addrlen>0){
		
#ifndef USE_SENDMSG
		if (m->b_cont!=NULL){
			mblk_t *newm=msgpullup(m,-1);
			freemsg(m);
			m=newm;
		}
#endif
		if (using_ext_socket &&  session->rtcp.socket>0 ){
#ifdef USE_SENDMSG
			error=rtp_sendmsg(session->rtcp.socket,m,(struct sockaddr *)NULL,0);
#else
			error=send(session->rtcp.socket, m->b_rptr, (m->b_wptr - m->b_rptr),0);
#endif
		}else {
#ifdef USE_SENDMSG
			error=rtp_sendmsg(session->rtcp.socket,m,(struct sockaddr *) &session->rtcp.rem_addr,
			session->rtcp.addrlen);
#else
			error = sendto (session->rtcp.socket, m->b_rptr,
			(m->b_wptr - m->b_rptr), 0,
			(struct sockaddr *) &session->rtcp.rem_addr,
			session->rtcp.addrlen);
#endif
		}
	
		if (error < 0){
			if (session->on_network_error.count>0){
				rtp_signal_table_emit3(&session->on_network_error,(gpointer)"Error sending RTCP packet",INT_TO_POINTER(getSocketErrorCode()));
			}else g_warning ("Error sending rtcp packet: %s ; socket=%i", getSocketError(), session->rtcp.socket);
		}
	}else g_warning("Cannot send rtcp report because I don't know the remote address.");
	freemsg (m);
	return error;
}


/**
 *rtp_session_set_ssrc:
 *@session: a rtp session.
 *@ssrc: an unsigned 32bit integer representing the synchronisation source identifier (SSRC).
 *
 *	Sets the SSRC for the outgoing stream.
 *  If not done, a random ssrc is used.
 *
**/
void
rtp_session_set_ssrc (RtpSession * session, guint32 ssrc)
{
	session->send_ssrc = ssrc;
}

/* this function initialize all session parameter's that depend on the payload type */
static void payload_type_changed(RtpSession *session, PayloadType *pt){
	jitter_control_set_payload(&session->rtp.jittctl,pt);
	session->rtp.rtcp_report_snt_interval=RTCP_DEFAULT_REPORT_INTERVAL*pt->clock_rate;
	rtp_session_set_time_jump_limit(session,session->rtp.time_jump);
}

/**
 *rtp_session_set_payload_type:
 *@session: a rtp session
 *@paytype: the payload type
 *
 *	Sets the payload type of the rtp session. It decides of the payload types written in the
 *	of the rtp header for the outgoing stream, if the session is SENDRECV or SENDONLY.
 *	For the incoming stream, it sets the waited payload type. If that value does not match
 *	at any time this waited value, then the application can be informed by registering
 *	for the "payload_type_changed" signal, so that it can make the necessary changes
 *	on the downstream decoder that deals with the payload of the packets.
 *
 *Returns: 0 on success, -1 if the payload is not defined.
**/

int
rtp_session_set_payload_type (RtpSession * session, int paytype)
{
	PayloadType *pt;
	session->payload_type = paytype;
	pt=rtp_profile_get_payload(session->profile,paytype);
	if (pt!=NULL){
		payload_type_changed(session,pt);
	}
	return 0;
}

int rtp_session_get_payload_type(RtpSession *session){
	return session->payload_type;
}


/**
 *rtp_session_set_payload_type_with_string:
 *@session: a rtp session
 *@paytype: the payload type
 *
 *	Sets the payload type of the rtp session. It decides of the payload types written in the
 *	of the rtp header for the outgoing stream, if the session is SENDRECV or SENDONLY.
 * 	Unlike #rtp_session_set_payload_type(), it takes as argument a string referencing the
 *	payload type (mime type).
 *	For the incoming stream, it sets the waited payload type. If that value does not match
 *	at any time this waited value, then the application can be informed by registering
 *	for the "payload_type_changed" signal, so that it can make the necessary changes
 *	on the downstream decoder that deals with the payload of the packets.
 *
 *Returns: 0 on success, -1 if the payload is not defined.
**/

int
rtp_session_set_payload_type_with_string (RtpSession * session, const char * mime)
{
	int pt;
	pt=rtp_profile_get_payload_number_from_mime(session->profile,mime);
	if (pt<0) {
		g_warning("%s is not a know mime string within the rtpsession's profile.",mime);
		return -1;
	}
	rtp_session_set_payload_type(session,pt);
	return 0;
}


/**
 *rtp_session_create_packet:
 *@session:		a rtp session.
 *@header_size:	the rtp header size. For standart size (without extensions), it is #RTP_FIXED_HEADER_SIZE
 *@payload		:data to be copied into the rtp packet.
 *@payload_size	: size of data carried by the rtp packet.
 *
 *	Allocates a new rtp packet. In the header, ssrc and payload_type according to the session's
 *	context. Timestamp and seq number are not set, there will be set when the packet is going to be
 *	sent with rtp_session_sendm_with_ts().
 *
 *Returns: a rtp packet in a mblk_t (message block) structure.
**/
mblk_t * rtp_session_create_packet(RtpSession *session,gint header_size, const char *payload, gint payload_size)
{
	mblk_t *mp;
	gint msglen=header_size+payload_size;
	rtp_header_t *rtp;
	
	mp=allocb(msglen,BPRI_MED);
	rtp=(rtp_header_t*)mp->b_rptr;
	rtp->version = 2;
	rtp->padbit = 0;
	rtp->extbit = 0;
	rtp->markbit= 0;
	rtp->cc = 0;
	rtp->paytype = session->payload_type;
	rtp->ssrc = session->send_ssrc;
	rtp->timestamp = 0;	/* set later, when packet is sended */
	rtp->seq_number = 0; /*set later, when packet is sended */
	/*copy the payload */
	mp->b_wptr+=header_size;
	memcpy(mp->b_wptr,payload,payload_size);
	mp->b_wptr+=payload_size;
	return mp;
}

/**
 *rtp_session_create_packet_with_data:
 *@session:		a rtp session.
 *@payload		: the data to be sent with this packet
 *@payload_size	: size of data
 *@freefn		: a function that will be called when the payload buffer is no more needed.
 *
 *	Creates a new rtp packet using the given payload buffer (no copy). The header will be allocated separetely.
 *  In the header, ssrc and payload_type according to the session's
 *	context. Timestamp and seq number are not set, there will be set when the packet is going to be
 *	sent with rtp_session_sendm_with_ts().
 *	oRTP will send this packet using libc's sendmsg() (if this function is availlable!) so that there will be no
 *	packet concatenation involving copies to be done in user-space.
 *  @freefn can be NULL, in that case payload will be kept untouched.
 *
 *Returns: a rtp packet in a mblk_t (message block) structure.
**/

mblk_t * rtp_session_create_packet_with_data(RtpSession *session, char *payload, gint payload_size, void (*freefn)(void*))
{
	mblk_t *mp,*mpayload;
	gint header_size=RTP_FIXED_HEADER_SIZE; /* revisit when support for csrc is done */
	rtp_header_t *rtp;
	
	mp=allocb(header_size,BPRI_MED);
	rtp=(rtp_header_t*)mp->b_rptr;
	rtp->version = 2;
	rtp->padbit = 0;
	rtp->extbit = 0;
	rtp->markbit= 0;
	rtp->cc = 0;
	rtp->paytype = session->payload_type;
	rtp->ssrc = session->send_ssrc;
	rtp->timestamp = 0;	/* set later, when packet is sended */
	rtp->seq_number = 0; /*set later, when packet is sended */
	mp->b_wptr+=header_size;
	/* create a mblk_t around the user supplied payload buffer */
	mpayload=allocb_with_buf(payload,payload_size,BPRI_MED,freefn);
	mpayload->b_wptr+=payload_size;
	/* link it with the header */
	mp->b_cont=mpayload;
	return mp;
}


/**
 *rtp_session_create_packet_in_place:
 *@session:		a rtp session.
 *@buffer:	a buffer that contains first just enough place to write a RTP header, then the data to send.
 *@size		: the size of the buffer
 *@freefn : a function that will be called once the buffer is no more needed (the data has been sent).
 *
 *	Creates a new rtp packet using the buffer given in arguments (no copy). 
 *  In the header, ssrc and payload_type according to the session's
 *	context. Timestamp and seq number are not set, there will be set when the packet is going to be
 *	sent with rtp_session_sendm_with_ts().
 *  @freefn can be NULL, in that case payload will be kept untouched.
 *
 *Returns: a rtp packet in a mblk_t (message block) structure.
**/
mblk_t * rtp_session_create_packet_in_place(RtpSession *session,char *buffer, gint size, void (*freefn)(void*) )
{
	mblk_t *mp;
	rtp_header_t *rtp;
	
	mp=allocb_with_buf(buffer,size,BPRI_MED,freefn);

	rtp=(rtp_header_t*)mp->b_rptr;
	rtp->version = 2;
	rtp->padbit = 0;
	rtp->extbit = 0;
	rtp->markbit= 0;
	rtp->cc = 0;
	rtp->paytype = session->payload_type;
	rtp->ssrc = session->send_ssrc;
	rtp->timestamp = 0;	/* set later, when packet is sended */
	rtp->seq_number = 0; /*set later, when packet is sended */
	return mp;
}


/**
 *rtp_session_sendm_with_ts:
 *@session	: a rtp session.
 *@mp		:	a rtp packet presented as a mblk_t.
 *@timestamp:	the timestamp of the data to be sent. Refer to the rfc to know what it is.
 *
 *	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.
 *
 *Returns: the number of bytes sent over the network.
**/
gint
rtp_session_sendm_with_ts (RtpSession * session, mblk_t *mp, guint32 timestamp)
{
	rtp_header_t *rtp;
	guint32 packet_time;
	gint error = 0;
	gint payloadsize;
	RtpScheduler *sched=session->sched;
	RtpStream *stream=&session->rtp;

	if (session->flags & RTP_SESSION_SEND_NOT_STARTED)
	{
		session->rtp.snd_ts_offset = timestamp;
		if (session->flags & RTP_SESSION_SCHEDULED)
		{
			session->rtp.snd_time_offset = sched->time_;
		}
		rtp_session_unset_flag (session,RTP_SESSION_SEND_NOT_STARTED);
	}
	/* 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,
				     timestamp -
				     session->rtp.snd_ts_offset) +
					session->rtp.snd_time_offset;
		/*g_message("rtp_session_send_with_ts: packet_time=%i time=%i",packet_time,sched->time_);*/
		wait_point_lock(&session->send_wp);
		if (TIME_IS_STRICTLY_NEWER_THAN (packet_time, sched->time_))
		{
			wait_point_wakeup_at(&session->send_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->send_wp);
	}
	

	rtp=(rtp_header_t*)mp->b_rptr;
	
	payloadsize = msgdsize(mp) - RTP_FIXED_HEADER_SIZE - (rtp->cc*sizeof(guint32));
	rtp_session_lock (session);
	
	/* set a seq number */
	rtp->seq_number=session->rtp.snd_seq;
	rtp->timestamp=timestamp;
	session->rtp.snd_seq++;
	session->rtp.snd_last_ts = timestamp;


	ortp_global_stats.sent += payloadsize;
	stream->stats.sent += payloadsize;
	ortp_global_stats.packet_sent++;
	stream->stats.packet_sent++;

	error = ortp_rtp_send (session, mp);
	rtp_session_rtcp_process(session);
	rtp_session_unlock (session);
	
	return error;
}


/**
 *rtp_session_send_with_ts:
 *@session: a rtp session.
 *@buffer:	a buffer containing the data to be sent in a rtp packet.
 *@len:		the length of the data buffer, in bytes.
 *@userts:	the timestamp of the data to be sent. Refer to the rfc to know what it is.
 *
 *	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.
 *
 *
 *Returns: the number of bytes sent over the network.
**/
gint
rtp_session_send_with_ts (RtpSession * session, const gchar * buffer, gint len,
			  guint32 userts)
{
	mblk_t *m;
	int err;
#ifdef USE_SENDMSG
	m=rtp_session_create_packet_with_data(session,(gchar*)buffer,len,NULL);
#else
	m = rtp_session_create_packet(session,RTP_FIXED_HEADER_SIZE,(gchar*)buffer,len);
#endif
	err=rtp_session_sendm_with_ts(session,m,userts);
	return err;
}


static gint
rtp_recv (RtpSession * session, guint32 user_ts)
{
	gint error;
	struct sockaddr remaddr;
	socklen_t addrlen = sizeof (remaddr);
	char *p;
	mblk_t *mp;
	RtpStream *stream=&session->rtp;

	if (session->rtp.socket<1) return -1;  /*session has no sockets for the moment*/
	

	while (1)
	{
		if (session->rtp.cached_mp==NULL)
			 session->rtp.cached_mp = allocb (session->max_buf_size, 0);
		mp=session->rtp.cached_mp;
		if (session->flags & RTP_SESSION_USING_EXT_SOCKETS){
			error=recv(session->rtp.socket,mp->b_wptr,session->max_buf_size,0);
		}else error = recvfrom (session->rtp.socket, mp->b_wptr,
				  session->max_buf_size, 0,
				  (struct sockaddr *) &remaddr,
				  &addrlen);
		if (error > 0)
		{
			if (error<RTP_FIXED_HEADER_SIZE){
				g_warning("Packet too small to be a rtp packet (%i)!",error);
				stream->stats.bad++;
				ortp_global_stats.bad++;
				/* don't free, it will be reused next time */
			}else{
				/* resize the memory allocated to fit the udp message */

				p = g_realloc (mp->b_wptr, error);
				if (p != mp->b_wptr)
					ortp_debug("The recv area has moved during reallocation.");
				mp->b_datap->db_base = mp->b_rptr =
					mp->b_wptr = p;
				mp->b_wptr += error;
				mp->b_datap->db_lim = mp->b_wptr;
				/* then parse the message and put on queue */
				rtp_parse (session, mp, user_ts + session->rtp.hwrcv_diff_ts);
				session->rtp.cached_mp=NULL;
			}
		}
		else
		{
			if (error == 0)
			{
				g_warning
					("rtp_recv: strange... recv() returned zero.");
			}
			else if (errno!=EWOULDBLOCK && errno!=EAGAIN)
			{
				if (session->on_network_error.count>0){
					rtp_signal_table_emit3(&session->on_network_error,(gpointer)"Error receiving RTP packet",INT_TO_POINTER(getSocketErrorCode()));
				}else g_warning("Error receiving RTP packet: %s.",getSocketError());
			}
			/* don't free the cached_mp, it will be reused next time */
			return -1;	/* avoids an infinite loop ! */
		}
	}
	return error;
}

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

static gint
rtcp_recv (RtpSession * session)
{
	gint error;
	struct sockaddr remaddr;
	socklen_t addrlen=0;
	char *p;
	mblk_t *mp;
	

	if (session->rtcp.socket<1) return -1;  /*session has no rtcp sockets for the moment*/
	

	while (1)
	{
		if (session->rtcp.cached_mp==NULL)
			 session->rtcp.cached_mp = allocb (RTCP_MAX_RECV_BUFSIZE, 0);
		
		mp=session->rtcp.cached_mp;
		if (session->flags & RTP_SESSION_USING_EXT_SOCKETS){
			error=recv(session->rtcp.socket,mp->b_wptr,RTCP_MAX_RECV_BUFSIZE,0);
		}else {
			addrlen=sizeof (remaddr);
			error=recvfrom (session->rtcp.socket, mp->b_wptr,
				  RTCP_MAX_RECV_BUFSIZE, 0,
				  (struct sockaddr *) &remaddr,
				  &addrlen);
		}
		if (error > 0)
		{
			/* resize the memory allocated to fit the udp message */
	
			p = g_realloc (mp->b_wptr, error);
			if (p != mp->b_wptr)
				ortp_debug("The recv area has moved during reallocation.");
			mp->b_datap->db_base = mp->b_rptr =
				mp->b_wptr = p;
			mp->b_wptr += error;
			mp->b_datap->db_lim = mp->b_wptr;
			/* then parse the message */
			rtcp_parse (session, mp);
			freemsg(mp);
			session->rtcp.cached_mp=NULL;
			if (addrlen>0){
				/* store the sender rtcp address to send him receiver reports */
				memcpy(&session->rtcp.rem_addr,&remaddr,addrlen);
			}
		}
		else
		{
			if (error == 0)
			{
				g_warning
					("rtcp_recv: strange... recv() returned zero.");
			}
			else if (errno!=EWOULDBLOCK && errno!=EAGAIN)
			{
				if (session->on_network_error.count>0){
					rtp_signal_table_emit3(&session->on_network_error,(gpointer)"Error receiving RTCP packet",INT_TO_POINTER(getSocketErrorCode()));
				}else g_warning("Error receiving RTCP packet: %s.",getSocketError());
			}
			/* don't free the cached_mp, it will be reused next time */
			return -1;	/* avoids an infinite loop ! */
		}
	}
	return error;
}


static void payload_type_changed_incoming(RtpSession *session, int paytype){
	/* check if we support this payload type */
	PayloadType *pt=rtp_profile_get_payload(session->profile,paytype);
	if (pt!=0){
		g_message ("rtp_parse: payload type changed to %i(%s) !",
				 paytype,pt->mime_type);
		session->payload_type = paytype;
		payload_type_changed(session,pt);
		rtp_signal_table_emit (&session->on_payload_type_changed);	
	}else{
		g_warning("Receiving packet with unknown payload type %i.",paytype);
	}
}


/**
 *rtp_session_recvm_with_ts:
 *@session: a rtp session.
 *@user_ts:	a timestamp.
 *
 *	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.
 *
 *Returns: a rtp packet presented as a mblk_t.
**/

mblk_t *
rtp_session_recvm_with_ts (RtpSession * session, guint32 user_ts)
{
	mblk_t *mp = NULL;
	rtp_header_t *rtp;
	guint32 ts;
	guint32 packet_time;
	RtpScheduler *sched=session->sched;
	RtpStream *stream=&session->rtp;
	gint rejected=0;

	/* 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;
		if (session->flags & RTP_SESSION_SCHEDULED)
		{
			session->rtp.rcv_time_offset = sched->time_;
			//g_message("setting snd_time_offset=%i",session->rtp.snd_time_offset);
		}
		rtp_session_unset_flag (session,RTP_SESSION_RECV_NOT_STARTED);
	}
	session->rtp.rcv_last_app_ts = user_ts;
	rtp_recv (session, user_ts);
	rtcp_recv(session);
	/* check for telephone event first */
	/* first lock the session */
	rtp_session_lock (session);
	mp=getq(&session->rtp.tev_rq);
	if (mp!=NULL){
		rtp_signal_table_emit2(&session->on_telephone_event_packet,(gpointer)mp);
		if (session->on_telephone_event.count>0){
			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)
	{
		rtp_header_t *oldest, *newest;
		queue_t *q = &session->rtp.rq;
		if (qempty(q))
		{
			ortp_debug ("Queue is empty.");
			goto end;
		}
		oldest = (rtp_header_t *) qfirst(q)->b_rptr;
		newest = (rtp_header_t *) qlast(q)->b_rptr;
		if ((guint32) (newest->timestamp - oldest->timestamp) <
		    session->rtp.jittctl.jitt_comp_ts)
		{
			ortp_debug("Not enough packet bufferised.");
			goto end;
		}
		/* enough packet bufferised */
		mp = getq (&session->rtp.rq);
		rtp = (rtp_header_t *) mp->b_rptr;
		session->rtp.rcv_ts_offset = rtp->timestamp;
		/* remember the timestamp offset between the stream timestamp (random)
		 * and the user timestamp, that very often starts at zero */
		session->rtp.rcv_diff_ts = rtp->timestamp - user_ts;
		/* remember the difference between the last received on the socket timestamp and the user timestamp */
		session->rtp.hwrcv_diff_ts=session->rtp.rcv_diff_ts + session->rtp.jittctl.jitt_comp_ts;
		session->rtp.rcv_last_ret_ts = user_ts;	/* just to have an init value */
		session->rtp.rcv_last_ts = rtp->timestamp;
		session->recv_ssrc = rtp->ssrc;
		/* delete the recv synchronisation flag */
		rtp_session_unset_flag (session, RTP_SESSION_RECV_SYNC);
		ortp_debug("Returning FIRST packet with ts=%i, hwrcv_diff_ts=%i, rcv_diff_ts=%i", rtp->timestamp,
					session->rtp.hwrcv_diff_ts,session->rtp.rcv_diff_ts);

		goto end;
	}
	/* else this the normal case */
	/*calculate the stream timestamp from the user timestamp */
	ts = user_ts + session->rtp.rcv_diff_ts;
	session->rtp.rcv_last_ts = ts;
	mp = rtp_getq (&session->rtp.rq, ts,&rejected);
	
	stream->stats.skipped+=rejected;
	ortp_global_stats.skipped+=rejected;

	/* perhaps we can now make some checks to see if a resynchronization is needed */
	/* TODO */
	goto end;

      end:
	if (mp != NULL)
	{
		int msgsize = msgdsize (mp);	/* evaluate how much bytes (including header) is received by app */
		guint32 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->payload_type != rtp->paytype)
		{
			payload_type_changed_incoming(session, rtp->paytype);
		}
		/* patch the packet so that it has a timestamp compensated by the 
		adaptive jitter buffer mechanism */
		if (session->rtp.jittctl.adaptive){
			rtp->timestamp-=session->rtp.jittctl.corrective_slide;
			/*printf("Returned packet has timestamp %u, with clock slide compensated it is %u\n",packet_ts,rtp->timestamp);*/
		}
	}
	else
	{
		ortp_debug ("No mp for timestamp queried");
		stream->stats.unavaillable++;
		ortp_global_stats.unavaillable++;
	}
	rtp_session_rtcp_process(session);
	rtp_session_unlock (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->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;
}


gint msg_to_buf (mblk_t * mp, char *buffer, gint len)
{
	gint rlen = len;
	mblk_t *m, *mprev;
	gint mlen;
	m = mp->b_cont;
	mprev = mp;
	while (m != NULL)
	{
		mlen = 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.
**/
gint rtp_session_recv_with_ts (RtpSession * session, gchar * buffer,
			       gint len, guint32 time, gint * have_more)
{
	mblk_t *mp;
	gint rlen = len;
	gint wlen, mlen;
	guint32 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->payload_type);
	if (payload==NULL){
		g_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 + (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->payload_type);
		if (payload==NULL){
			g_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.
**/
guint32 rtp_session_get_current_send_ts(RtpSession *session)
{
	guint32 userts;
	guint32 session_time;
	RtpScheduler *sched=session->sched;
	PayloadType *payload;
	g_return_val_if_fail (session->payload_type<128, 0);
	payload=rtp_profile_get_payload(session->profile,session->payload_type);
	g_return_val_if_fail(payload!=NULL, 0);
	if ( (session->flags & RTP_SESSION_SCHEDULED)==0 ){
		g_warning("can't guess current timestamp because session is not scheduled.");
		return 0;
	}
	session_time=sched->time_-session->rtp.snd_time_offset;
	userts=  (guint32)( ( (gdouble)(session_time) * (gdouble) 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.
 *
**/
guint32 rtp_session_get_current_recv_ts(RtpSession *session){
	guint32 userts;
	guint32 session_time;
	RtpScheduler *sched=ortp_get_scheduler();
	PayloadType *payload;
	g_return_val_if_fail (session->payload_type<128, 0);
	payload=rtp_profile_get_payload(session->profile,session->payload_type);
	g_return_val_if_fail(payload!=NULL, 0);
	if ( (session->flags & RTP_SESSION_SCHEDULED)==0 ){
		g_warning("can't guess current timestamp because session is not scheduled.");
		return 0;
	}
	session_time=sched->time_-session->rtp.rcv_time_offset;
	userts=  (guint32)( ( (gdouble)(session_time) * (gdouble) 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, gint milisecs){
	guint32 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_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 */
	close_socket (session->rtp.socket);
	close_socket (session->rtcp.socket);

	wait_point_uninit(&session->send_wp);
	wait_point_uninit(&session->recv_wp);
	g_mutex_free (session->lock);
	session->lock=NULL;
	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);
}

/**
 *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.one = 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_destroy:
 *@session: a rtp session.
 *
 *	Destroys a rtp session.
 *
**/
void rtp_session_destroy (RtpSession * session)
{
	rtp_session_uninit (session);
	g_free (session);
}

guint32 rtp_session_time_to_ts(RtpSession *session, gint time){
	PayloadType *payload;
	g_return_val_if_fail (session->payload_type < 127, 0);
	payload =
		rtp_profile_get_payload (session->profile,
					 session->payload_type);
	if (payload == NULL)
	{
		g_warning
			("rtp_session_ts_to_t: use of unsupported payload type.");
		return 0;
	}
	/* the return value is in milisecond */
	return (double)payload->clock_rate*(double)time/1000.0;
}

/* function used by the scheduler only:*/
guint32 rtp_session_ts_to_time (RtpSession * session, guint32 timestamp)
{
	PayloadType *payload;
	g_return_val_if_fail (session->payload_type < 127, 0);
	payload =
		rtp_profile_get_payload (session->profile,
					 session->payload_type);
	if (payload == NULL)
	{
		g_warning
			("rtp_session_ts_to_t: use of unsupported payload type.");
		return 0;
	}
	/* the return value is in milisecond */
	return (guint32) (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, guint32 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, gint milisec)
{
	session->rtp.snd_time_offset+=milisec;
}


/* packet api */

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