/*
 *
 *
 */

#include "rpsdist.h"


long   SEQUENCE  = 0;
char * DB_NAME;
trace_t *default_trace;
char * ENCODING = "gzip"; /* for new and redist. transactions */


void alarmHandler(int sig){
   peer_t * curr_obj;
   curr_obj = peer_list.first;
   
   write(STDOUT_FILENO, "* SENDING HEARTBEAT *\n", 22);
   
   while(curr_obj){
     make_heartbeat(curr_obj);
     curr_obj = curr_obj->next;
   }

   signal(SIGALRM, alarmHandler);
   alarm(10);

}

void handle_connection(peer_t * obj){
  static fd_set     read_fds;
  int ret, i;
  char tmp[10000];
  /*struct itimerval itimer;
  struct timeval timer;
  struct sigaction act;
  
  act.sa_handler = alarmHandler;
  act.sa_flags = SA_RESTART;  
  timer.tv_sec = 10;
  timer.tv_usec = 0;
  itimer.it_interval = itimer.it_value = timer;
  sigaction(SIGALRM, &act, NULL);
  setitimer(ITIMER_REAL, &itimer, NULL);*/
 
  while (1) {
    FD_ZERO(&read_fds);
    FD_SET(obj->fd, &read_fds);
    ret = select (obj->fd+1, &read_fds, NULL, NULL, NULL);
    if(ret < 0){
      /*printf("Select err in handle connection\n");
      exit(1);*/
      pthread_exit(NULL);
    }
    if (ret == 1){ /* if we can read */
      /* do it */
      i = parse_transaction(obj->fd);
      if(i < 0){
	printf("** Lost a connection to %s **\n", obj->host);
	delete_peer_obj(&peer_list, obj);
	break;
      }
    }
    else{
      printf("Breaking..exiting handle_trans()\n");
      break;
    }
  }
}

peer_t * spawn_new(int fd, char * host, int port){
   pthread_t *slave_t;
   pthread_attr_t attr;
   peer_t * temp_obj;
   pthread_attr_init(&attr);
   pthread_attr_setscope(&attr,PTHREAD_SCOPE_SYSTEM);

   slave_t = (pthread_t*)malloc(sizeof(pthread_t));
   temp_obj = create_peer_obj(fd, host, port);
   pthread_create(slave_t,&attr,(void *)handle_connection,(peer_t*)temp_obj);
   
   return temp_obj;
}


int init(char * argv[]){
  struct sockaddr_in addr, client_addr;
  int addrlen = sizeof(client_addr);
  int lsd, newsocket, ret, i;
  static struct timeval	tv;
  static fd_set		read_fds;
  static fd_set         write_fds;
  static char		tmp[1000];
  char * msg;
  pthread_t *slave_t;
  pthread_attr_t attr;
  peer_t * temp_obj;
  lsd = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
  
  pthread_attr_init(&attr);
  pthread_attr_setscope(&attr,PTHREAD_SCOPE_SYSTEM);


  tmp[0] = '\0';
  if(lsd == -1){
    printf("Unable to open socket\n");
    exit(1);
  }
  bzero(&addr, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = htonl(INADDR_ANY);
  addr.sin_port = htons( atoi( argv[1]) );
  
  if( bind( lsd, (struct sockaddr *) &addr, sizeof( addr ) ) == -1 ){
    printf( "Bind error\n");
    exit(1);
  }
  printf("Server started on port %d with socket %d\n", atoi(argv[1]), lsd);
  
  if( listen(lsd,10) < 0){
    printf("Listen error\n");
    exit(1);
  }
  
  signal(SIGALRM, alarmHandler);
  alarm(10);
  
  while(1)
    {
      newsocket = accept (lsd, (struct sockaddr*)&client_addr, &addrlen);  
      if (newsocket != -1){
	printf("** got a connection from %s**\n", inet_ntoa(client_addr.sin_addr));
	/*
	slave_t = (pthread_t*)malloc(sizeof(pthread_t));
	temp_obj = create_peer_obj(newsocket, inet_ntoa(client_addr.sin_addr), client_addr.sin_port);
	pthread_create(slave_t,&attr,(void *)handle_connection,(peer_t*)temp_obj);*/
	
	add_peer_obj(&peer_list, spawn_new(newsocket, inet_ntoa(client_addr.sin_addr), client_addr.sin_port));
      }
      else if(errno != EINTR){
	printf("accept error %d\n", newsocket);
	exit(1);
      }
	
    }
}

int main(int argc, char  * argv[]){
  char  cmd[10000];
  char * d;
  int i = 0;
  int fd;
  int ends[2];

  peer_t * temp;

  init_regexes();

  if(argc != 3)
    printf("Usage: %s <port> <db_name>\n", argv[0]);
  else{
    DB_NAME = argv[2];
    init(argv);
  }
  
  clear_regexes();
  
}

