#include "rpsdist.h"

/*
 *  Make a heartbeat message.  Used to make sure connection is still 
 *  alive but no submissions have some across it.  Time between beats
 *  is a maximum of 1 day.  
 */
int make_heartbeat(peer_t * obj){
  char buffer[200];
  fd_set write_fd;
  struct timeval tv;
  
  tv.tv_sec = 5; /* one second is plenty */
  tv.tv_usec = 0;
  FD_ZERO(&write_fd);
  FD_SET(obj->fd, &write_fd);
 
  sprintf(buffer, "\nheartbeat: %s\n", DB_NAME);
  sprintf(buffer+strlen(buffer), "sequence: %ld\n", SEQUENCE);

  pthread_mutex_lock(&obj->write);

  if( select(obj->fd+1, NULL, &write_fd, NULL, &tv) > 0){
    write(obj->fd, buffer, strlen(buffer));
    time_stamp(obj->fd);
  }
  else{
    printf("Write timeout: make_heartbeat fd(%d)\n", obj->fd);
    pthread_mutex_unlock(&obj->write);
    return -1;
  }

  pthread_mutex_unlock(&obj->write);
  
  return 1;

}

/****** NEW ******/
int encode_transaction(int fd, char * method){
  char buffer[4096];
  int length = 0;
 
  char* enc_f_name = strdup(tmpnam(NULL));
  int enc_file_d;
  struct stat file_nfo;
  

  /* encode to enc_file_d file */
  if(!strcasecmp(method, "gzip")){
    /*lseek(pln_file_d, 0, SEEK_SET);*/
    enc_file_d = open(enc_f_name,O_CREAT|O_RDWR, S_IRUSR|S_IWUSR);
    zip_stream(fd, enc_file_d);
    /*close(enc_file_d);*/
    close(fd);
  }
  else
    enc_file_d = fd;
 
  /* get file size */
  if( fstat(enc_file_d, &file_nfo) < 0){
    printf("Stat error\n");
    return -1;
  }
  sprintf(buffer, "transaction-begin: %d\ntransfer-method: %s\n\n", file_nfo.st_size, method);
  send_to_peers(buffer, strlen(buffer), enc_file_d);

  close(enc_file_d);
  if(!strcasecmp(method, "gzip")){
    if(remove(enc_f_name) < 0)
      printf("encode: Unable to remove enc_file %s\n", enc_f_name);
  }
  free(enc_f_name);
 
}


/* 
 *  add in transaction labels, sequence, etc,
 *  We have the incoming transaction on fd, setup a pipe
 */
int encapsulate_transaction(int fd){
  int fildes[2];
  char buffer[4096];
  struct stat file_nfo;
  
  if( !strcasecmp(ENCODING, "plain")){
    /* get file size */
    if( fstat(fd, &file_nfo) < 0){
      printf("Stat error\n");
      return -1;
    }
    sprintf(buffer, "transaction-begin: %d\ntransfer-method: %s\n\n", file_nfo.st_size, ENCODING);
    send_to_peers(buffer, strlen(buffer), fd);
  }
  else{
  if(pipe(fildes) < 0){
    printf("Pipe error\n");
    return -1;
  }
  
  sprintf(buffer, "transaction-label: %s\nsequence: %ld\n", DB_NAME, SEQUENCE);
  write(fildes[1], buffer, strlen(buffer));
  time_stamp(fildes[1]);
  sprintf(buffer, "integrity: %s\n\n", "no-auth");
  write(fildes[1], buffer, strlen(buffer));
  
  while(read(fd, buffer, 4096) > 0)
    write(fildes[1], buffer, strlen(buffer));
  
  close(fildes[1]);
  encode_transaction(fildes[0], ENCODING);  /* closes fildes[0] */
  }
}
  
/* 
 *  source_fd MUST be a seekable descriptor
 */
void send_to_peers(char * source_buf, int length, int source_fd){
  peer_t * curr_obj = peer_list.first;
  char buffer[4096];
  int t_length = 0;

  
  while(curr_obj){
    write(STDOUT_FILENO, "writing to a peer\n", 18);
    /* write the buffer */
    
    pthread_mutex_lock(&curr_obj->write);
    /*write(STDOUT_FILENO, "got its lock\n", 13);*/
    if(source_buf)
      if( write(curr_obj->fd, source_buf, length) < 0){
	printf("send to peers: error, socket closed, removing object\n");
	delete_peer_obj(&peer_list, curr_obj);
	curr_obj = curr_obj->next;
	continue;
      }

    /*printf("Wrote buffer\n");*/

    /* write the file */
    if(source_fd >= 0){
      lseek(source_fd, 0, SEEK_SET);
      /*printf("Reset file to 0\n");*/
      while( (t_length = read(source_fd, buffer, 4096)) > 0){
	if(write(curr_obj->fd, buffer, t_length) <= 0){
	  printf("send to peers: error writing output, socket closed\n");
	  delete_peer_obj(&peer_list, curr_obj);
	  break;
      }
	/*printf("Read %d\n", t_length);*/
    } 
  }
    
    pthread_mutex_unlock(&curr_obj->write);
    curr_obj = curr_obj->next;   
  }
 
}
  
int confirm_transaction(int fd, struct initial_transaction * tran){
  char buffer[2000];
 
  sprintf(buffer, "transaction-confirm: %s %d\n", tran->dbname, tran->ident);
  sprintf(buffer+strlen(buffer),"confirmed-operation: %s\n", "add");
  sprintf(buffer+strlen(buffer),"commit-status: %s\n", "commit");

  write(fd, buffer, strlen(buffer));

}
  
  
			       
