rtpproxy_relay.c

简单的基于SIP的会话边界控制器

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/time.h>

#include <sys/socket.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <signal.h>

#ifdef HAVE_PTHREAD_SETSCHEDPARAM
   #include <sched.h>
#endif

#include <osipparser2/osip_parser.h>

#include "nsrsbc.h"
#include "cmdline.h"
#include "rtpproxy.h"

/* configuration storage */
extern struct gengetopt_args_info args_info;

/*
 * table to remember all active rtp proxy streams
 */
rtp_proxytable_t rtp_proxytable[RTPPROXY_SIZE];

/*
 * Mutex for thread synchronization (locking when accessing common 
 * data structures -> rtp_proxytable[]).
 *
 * use a 'fast' mutex for synchronizing - as these are portable... 
 */
static pthread_mutex_t rtp_proxytable_mutex = PTHREAD_MUTEX_INITIALIZER;

/* thread id of RTP proxy */
static pthread_t rtpproxy_tid=0;

/* master fd_set */
static fd_set master_fdset;
static int    master_fd_max;

/* forward declarations */
static void *rtpproxy_main(void *i);
static int rtp_recreate_fdset(void);
void rtpproxy_kill( void );
static void sighdl_alm(int sig) {/* just wake up from select() */};


/*
 * initialize and create rtp_relay proxy thread
 *
 * RETURNS
 *	STS_SUCCESS on success
 */
int rtp_relay_init( void ) {
   int sts;
   int arg=0;
   struct sigaction sigact;
   
   /* cancel RTP thread at exit */
   atexit(rtpproxy_kill);  

   /* clean proxy table */
   memset (rtp_proxytable, 0, sizeof(rtp_proxytable));

   /* initialize fd set for RTP proxy thread */
   FD_ZERO(&master_fdset); /* start with an empty fdset */
   master_fd_max=-1;

   /* install signal handler for SIGALRM - used to wake up
      the rtpproxy thread from select() hibernation */
   sigact.sa_handler = sighdl_alm;
   sigemptyset(&sigact.sa_mask);
   sigact.sa_flags=0;
   sigaction(SIGALRM, &sigact, NULL);

   sts=pthread_create(&rtpproxy_tid, NULL, rtpproxy_main, (void *)&arg);

   return STS_SUCCESS;
}


/*
 * main() of rtpproxy
 */
static void *rtpproxy_main(void *arg) {
   struct timeval tv;
   fd_set fdset;
   int fd_max;
   time_t t, last_t=0;
   int i, sts;
   int num_fd;
   osip_call_id_t callid;
   static char rtp_buff[RTP_BUFFER_SIZE];
   int count;
   struct sockaddr_in from;
   socklen_t fromlen;

   memcpy(&fdset, &master_fdset, sizeof(fdset));
   fd_max=master_fd_max;

   /* loop forever... */
   for (;;) {

      tv.tv_sec = 5;
      tv.tv_usec = 0;

      num_fd=select(fd_max+1, &fdset, NULL, NULL, &tv);
      pthread_testcancel();
      if ((num_fd<0) && (errno==EINTR)) {
         /*
          * wakeup due to a change in the proxy table:
          * lock mutex, copy master FD set and unlock
          */
         pthread_mutex_lock(&rtp_proxytable_mutex);
         memcpy(&fdset, &master_fdset, sizeof(fdset));
         fd_max=master_fd_max;
         pthread_mutex_unlock(&rtp_proxytable_mutex);
         continue;
      }

      time(&t);

      /*
       * LOCK the MUTEX
       */
      pthread_mutex_lock(&rtp_proxytable_mutex);

      /* check for data available and send to destination */
      for (i=0;(i<RTPPROXY_SIZE) && (num_fd>0);i++) {
         if ( (rtp_proxytable[i].rtp_rx_sock != 0) && 
            FD_ISSET(rtp_proxytable[i].rtp_rx_sock, &fdset) ) {
            /* have some data to send */
            num_fd--;

	    /* read from sock rtp_proxytable[i].sock*/
	     fromlen = sizeof(struct sockaddr_in);
            count=recvfrom(rtp_proxytable[i].rtp_rx_sock, &rtp_buff, sizeof(rtp_buff)-1 , 0, (struct sockaddr*)&from, &fromlen);
            rtp_buff[RTP_BUFFER_SIZE] = '\0';
	     printf("RTP Packet from %s:%d\n", inet_ntoa(from.sin_addr), ntohs(from.sin_port));
            if (count < 0) {
               printf("rtp sock read error!\n");
               if (errno == EAGAIN)
                  continue;
            }

            /*
             * forwarding an RTP packet only makes sense if we really
             * have got some data in it (count > 0)
             */
            if (count > 0) {
               /* write nat_ipaddr and nat_port */
               if( rtp_proxytable[i].direction == DIR_OUTGOING) {
                     memcpy(&rtp_proxytable[i].nat_ipaddr, &from.sin_addr, sizeof(struct in_addr));
              	rtp_proxytable[i].nat_port = ntohs(from.sin_port);
                 	//printf("RTPTable: %s:%d\n", inet_ntoa(rtp_proxytable[i].nat_ipaddr), rtp_proxytable[i].nat_port);
               }
               /* find the corresponding TX socket */
               if (rtp_proxytable[i].rtp_tx_sock == 0) {
                  int j;
                  int rtp_direction = rtp_proxytable[i].direction;
                  int media_stream_no = rtp_proxytable[i].media_stream_no;

                  callid.number = rtp_proxytable[i].callid_number;
                  callid.host = rtp_proxytable[i].callid_host;

                  for (j=0;(j<RTPPROXY_SIZE);j++) {
                     char *client_id = rtp_proxytable[i].client_id;
                     osip_call_id_t cid;
                     cid.number = rtp_proxytable[j].callid_number;
                     cid.host = rtp_proxytable[j].callid_host;

                     /* match on:
                      * - same call ID
                      * - same media stream
                      * - opposite direction
                      * - different client ID
                      */
                     if ( (rtp_proxytable[j].rtp_rx_sock != 0) &&
                          (compare_callid(&callid, &cid) == STS_SUCCESS) &&
                          (media_stream_no == rtp_proxytable[j].media_stream_no) &&
                          (rtp_direction != rtp_proxytable[j].direction) &&
                          (strcmp(rtp_proxytable[j].client_id, client_id) != 0) ) {
                        rtp_proxytable[i].rtp_tx_sock = rtp_proxytable[j].rtp_rx_sock;
                        rtp_proxytable[i].opposite_entry=j+1;    
                        printf("media_stream_no:%d----%d\n",media_stream_no,rtp_proxytable[j].media_stream_no);
                        printf("rtp_direction:%d----%d\n",rtp_direction,rtp_proxytable[j].direction);
                        printf("client_id:%s----%s\n",client_id,rtp_proxytable[j].client_id);
                        break;
                     }
                  }
               } /* rtp_tx_sock == 0 */
		 printf("rtp_tx_sock %d\n", rtp_proxytable[i].rtp_tx_sock);       
               if (rtp_proxytable[i].rtp_tx_sock != 0) {
                  /* write to dest via socket rtp_tx_sock */
                  struct sockaddr_in dst_addr;
                  dst_addr.sin_family = AF_INET;
                  if(rtp_proxytable[i].direction == DIR_INCOMING) {      
                  	int j = rtp_proxytable[i].opposite_entry-1;
                  	memcpy(&dst_addr.sin_addr.s_addr,
                         &rtp_proxytable[j].nat_ipaddr, 
                         sizeof(struct in_addr));
                  	dst_addr.sin_port= htons(rtp_proxytable[j].nat_port);
                  	//printf("RTP Packet %s:%d\n", inet_ntoa(rtp_proxytable[j].nat_ipaddr), rtp_proxytable[j].nat_port);
                  }
                  if(rtp_proxytable[i].direction == DIR_OUTGOING) {
			 memcpy(&dst_addr.sin_addr.s_addr,
                         &rtp_proxytable[i].remote_ipaddr, 
                         sizeof(struct in_addr));
                  	 dst_addr.sin_port= htons(rtp_proxytable[i].remote_port);
               	// printf("RTP Packet %s:%d\n", inet_ntoa(rtp_proxytable[i].remote_ipaddr), rtp_proxytable[i].remote_port);
                  } 
                  printf("RTP Packet send to %s:%d\n", inet_ntoa(dst_addr.sin_addr.s_addr), ntohs(dst_addr.sin_port));	
                  sts = sendto(rtp_proxytable[i].rtp_tx_sock, rtp_buff,
                               count, 0, (const struct sockaddr *)&dst_addr,
                               (socklen_t)sizeof(dst_addr));
                  if (sts == -1) {
                     if (errno != ECONNREFUSED) {
                        printf("sendto() [%s:%i size=%i] call failed: %s\n",
                        inet_ntoa(rtp_proxytable[i].remote_ipaddr),
                        rtp_proxytable[i].remote_port, count, strerror(errno));

                    /* if sendto() fails with bad filedescriptor,
                     * this means that the opposite stream has been
                     * canceled or timed out.
                     * we should then cancel this stream as well.*/

                    //printf("stopping opposite stream\n");
                    /* don't lock the mutex, as we own the lock */
                    //callid.number=rtp_proxytable[i].callid_number;
                    //callid.host=rtp_proxytable[i].callid_host;
                    //rtp_relay_stop_fwd(&callid, rtp_proxytable[i].direction, -1, 1);
                    }
                 }
              }
            } /* count > 0 */

            /* update timestamp of last usage for both (RX and TX) entries.
             * This allows silence (no data) on one stream without breaking
             * the connection after the RTP timeout */
            rtp_proxytable[i].timestamp=t;
            if (rtp_proxytable[i].opposite_entry > 0) {
               rtp_proxytable[rtp_proxytable[i].opposite_entry-1].timestamp=t;
            }
         }
      } /* for i */

      /*
       * age and clean rtp_proxytable (check every 10 seconds)
       */
      if (t > (last_t+10) ) {
         last_t = t;
	  for (i=0;i<RTPPROXY_SIZE; i++) {
            if ( (rtp_proxytable[i].rtp_rx_sock != 0) &&
		 ((rtp_proxytable[i].timestamp+args_info.rtp_timeout_arg)<t)) {
               /* this one has expired, clean it up */
               callid.number=rtp_proxytable[i].callid_number;
               callid.host=rtp_proxytable[i].callid_host;
               /* Don't lock the mutex, as we own the lock already here */
               /* Only stop the stream we caught is timeout and not everything. */
               rtp_relay_stop_fwd(&callid, rtp_proxytable[i].direction,
                                  rtp_proxytable[i].media_stream_no, 1);
	    }
	 }
      }

      /* copy master FD set */
      memcpy(&fdset, &master_fdset, sizeof(fdset));
      fd_max=master_fd_max;

      /*
       * UNLOCK the MUTEX
       */
      pthread_mutex_unlock(&rtp_proxytable_mutex);
   } /* for(;;) */

   return NULL;
}


/*
 * start an rtp stream on the proxy
 *
 * RETURNS
 *	STS_SUCCESS on success
 *	STS_FAILURE on error
 */
int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id,
                         int rtp_direction, int media_stream_no, 
                         struct in_addr local_ipaddr,int *local_port, 
                         struct in_addr remote_ipaddr,int remote_port) {
	static int prev_used_port = 0;
  	int num_ports;
   	int i2, i, j;
   	int sock, port;
  	int freeidx;
   	int sts=STS_SUCCESS;
   	int tos;