flute.c

这个程序实现了FLUTE协议

/* $Author: peltotas $ $Date: 2006/03/16 13:45:54 $ $Revision: 1.19 $ */
/*
 *   MAD-FLUTELIB: Implementation of FLUTE protocol.
 *   Copyright (c) 2003-2006 TUT - Tampere University of Technology
 *   main authors/contacts: [e-mail] and [e-mail]
 *
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#include "flute_inc.h"

/**** Private functions ****/

int all_files_received(char *filetable[10], int file_nb);
void file_received(char *filetable[10], int file_nb, char *filepath);

/*
 * This function is flute sender's main function.
 *
 * Params:	arguments_t *a: Pointer to arguments structure where command line arguments are parsed,
 *			int *s_id: Pointer to session identifier,
 *			int *ch_id: Pointer to channel identifier.
 *
 * Return:	int: Different values (0, -1, -2, -4) depending on how function ends.
 *
 */
int flute_sender(arguments_t *a, int *s_id, int *ch_id) {

  unsigned short i;
  int j, k, l, m;
  int retval = 0;
  int retcode = 0;
  
  struct sockaddr_in ipv4;
  unsigned long addr_nb;
  int port_nb;
  unsigned short ipv6addr[8];
  int nb_ipv6_part;
  int dup_sep = 0;
  char tmp[5];
  
  char addrs[MAX_CHANNELS_IN_SESSION][MAX_LENGTH];	/* Mcast addresses on which to send */
  char ports[MAX_CHANNELS_IN_SESSION][MAX_LENGTH];	/* Local port numbers  */
  
  time_t systime;
  
#ifdef WIN32
  ULONGLONG curr_time;
#else
  unsigned long long curr_time;
#endif

  char *sdp_buf = NULL;
  FILE *sdp_fp;
  struct stat sdp_file_stats;
  int nbytes;

  if(strcmp(a->sdp_file, "") != 0) {

		if(stat(a->sdp_file, &sdp_file_stats) == -1) {
			printf("Error: %s is not valid file name\n", a->sdp_file);
			fflush(stdout);
			return -1;
		}
	
		/* Allocate memory for buf */
		if(!(sdp_buf = (char*)calloc((sdp_file_stats.st_size + 1), sizeof(char)))) {
			printf("Could not alloc memory for sdp buffer!\n");
			fflush(stdout);
			return -1;
		}

		if((sdp_fp = fopen(a->sdp_file, "rb")) == NULL) {
			printf("Error: unable to open sdp_file %s\n", a->sdp_file);
			fflush(stdout);
			free(sdp_buf);
			return -1;
		}

		nbytes = fread(sdp_buf, 1, sdp_file_stats.st_size, sdp_fp); 

		if(parse_sdp_file(a, addrs, ports, sdp_buf) == -1) {
			return -1;
		}
	}
	else {

		if(a->alc_a.addr_family == PF_INET) {
 
			addr_nb = ntohl(inet_addr(a->alc_a.addr));
			port_nb = atoi(a->alc_a.port);

			for(j = 0; j < a->alc_a.nb_channel; j++) {

				ipv4.sin_addr.s_addr = htonl(addr_nb);

				memset(addrs[j], 0, MAX_LENGTH);
				memset(ports[j], 0, MAX_LENGTH);

				sprintf(addrs[j], "%s", inet_ntoa(ipv4.sin_addr));
				sprintf(ports[j], "%i", port_nb);

				addr_nb++;
				port_nb++;
			}
		}
		else if(a->alc_a.addr_family == PF_INET6) {
			
			port_nb = atoi(a->alc_a.port);

			retcode = ushort_ipv6addr(a->alc_a.addr, &ipv6addr[0], &nb_ipv6_part);

			if(retcode == -1) {
				return -1;				
			}

			for(j = 0; j < a->alc_a.nb_channel; j++) {

				memset(addrs[j], 0, MAX_LENGTH);
				memset(ports[j], 0, MAX_LENGTH);
				dup_sep = 0;

				for(k = 0; k < nb_ipv6_part + 1; k++) {

					memset(tmp, 0, 5);

					if(ipv6addr[k] != 0) {
						sprintf(tmp, "%x", ipv6addr[k]);
						strcat(addrs[j], tmp);
					}
					else {
						if(dup_sep == 0) {
							dup_sep = 1;
						}
						else {
							printf("Invalid IPv6 address!\n");
							fflush(stdout);
							return -1;	
						}	
					}

					if(k != nb_ipv6_part) {
						strcat(addrs[j], ":");
					}
				}

				sprintf(ports[j], "%i", port_nb);

				port_nb++;

				if(j != a->alc_a.nb_channel - 1) {

					for(l = nb_ipv6_part;; l--) {

						if(l == 0) {
							printf("Only %i channel possible!\n", (j + 1));
							fflush(stdout);
							return -1;	
						}

						if(ipv6addr[l] + 1 > 0xFFFF) {
							continue;
						}
							
						if(ipv6addr[l] == 0) {

							if(nb_ipv6_part == 7) {
								ipv6addr[l]++;
								break;
							}

							for(m = nb_ipv6_part; m > l; m--) {
								ipv6addr[m + 1] = ipv6addr[m];	
							}

							ipv6addr[l + 1] = 1;

							nb_ipv6_part++;
							break;
						}
						else {
							ipv6addr[l]++;
							break;
						}
					}
				}
			}
		}
	}

	if(a->alc_a.stop_time != 0) {
		time(&systime);
		curr_time = systime + 2208988800U;

		if(a->alc_a.stop_time <= curr_time) {
			printf("Session end time reached\n");
			fflush(stdout);
			return -1;
		}
	}

	*s_id = open_alc_session(&a->alc_a);

	for(i = 0; (int)i < a->alc_a.nb_channel; i++) {

		if(a->alc_a.addr_type == 1) {
			*(ch_id + i) = add_alc_channel(*s_id, ports[i], addrs[0], a->alc_a.intface, a->alc_a.intface_name);
		}
		else {
			*(ch_id + i) = add_alc_channel(*s_id, ports[i], addrs[i], a->alc_a.intface, a->alc_a.intface_name);
		}

		if(*(ch_id + i) == -1) {

			while(i) {
				retcode = remove_alc_channel(*s_id, *(ch_id + i - 1));
				i--;
			}

		close_alc_session(*s_id);
			return -4;
		}	
	}

	/* Generate fdt file first */
	if(strcmp(a->fdt_file, "") == 0) {

	  memset(a->fdt_file, 0, MAX_LENGTH);
	  strcpy(a->fdt_file, "fdt_tsi");
	  
	  memset(tmp, 0, 5);
	  
#ifdef WIN32
	  sprintf(tmp, "%I64u", a->alc_a.tsi);
#else
	  sprintf(tmp, "%llu", a->alc_a.tsi);
#endif
	  
	  strcat(a->fdt_file, tmp);
	  strcat(a->fdt_file, ".xml");
  
	  retcode = generate_fdt(a->file_path, a->alc_a.base_dir, s_id, a->fdt_file, a->complete_fdt,
				 a->alc_a.verbosity);
	  
	  if(retcode < 0) {
	    return -1;
	  }
	}

	/***** FDT based send *****/

	retval = sender_in_fdt_based_mode(s_id, ch_id, a);

	return retval;
}

/*
 * This function is flute receiver's main function.
 *
 * Params:	arguments_t *a: Pointer to arguments structure where command line arguments are parsed,
 *			int *s_id: Pointer to session identifier,
 *			int *ch_id: Pointer to channel identifier,
 *			fdt_th_args_t *fdt_th_args: Pointer to structure conmtaining FDT and some other information.
 *
 * Return:	int: Different values (0, -1, -2, -3, -4, -5) depending on how function ends.
 *
 */

int flute_receiver(arguments_t *a, int *s_id, int *ch_id, fdt_th_args_t *fdt_th_args) {
	
  unsigned short i;
  int j, k, l, m;
  int retval = 0;
  int retcode = 0;
  
  char filetoken[MAX_PATH];
  
  struct sockaddr_in ipv4;
  unsigned long addr_nb;
  int port_nb;
  unsigned short ipv6addr[8];
  int nb_ipv6_part;
  int dup_sep = 0;
  char tmp[5];
  
  char addrs[MAX_CHANNELS_IN_SESSION][MAX_LENGTH];	/* Mcast addresses on which to send */
  char ports[MAX_CHANNELS_IN_SESSION][MAX_LENGTH];	/* Local port numbers  */

  time_t systime;
  bool is_printed = false;
  
#ifdef WIN32
  ULONGLONG curr_time;
#else
  unsigned long long curr_time;
#endif
  
#ifndef WIN32
  pthread_t fdt_thread_id; 
#endif
  
  char *sdp_buf = NULL;
  FILE *sdp_fp;
  struct stat	sdp_file_stats;
  int nbytes;

#ifdef WIN32
  if(mkdir(a->alc_a.base_dir) < 0) {					
#else		
  if(mkdir(a->alc_a.base_dir, S_IRWXU) < 0) {
#endif
    if(errno != EEXIST) {
      printf("mkdir failed: cannot create directory %s (errno=%i)\n", a->alc_a.base_dir, errno);
      fflush(stdout);
      return -1;
    }
  }

  if(strcmp(a->sdp_file, "") != 0) {

    if(stat(a->sdp_file, &sdp_file_stats) == -1) {
      printf("Error: %s is not valid file name\n", a->sdp_file);
      fflush(stdout);
      return -1;
    }
    
    /* Allocate memory for buf */
    if(!(sdp_buf = (char*)calloc((sdp_file_stats.st_size + 1), sizeof(char)))) {
      printf("Could not alloc memory for sdp buffer!\n");
      fflush(stdout);
      return -1;
    }
    
    if((sdp_fp = fopen(a->sdp_file, "rb")) == NULL) {
      printf("Error: unable to open sdp_file %s\n", a->sdp_file);
      fflush(stdout);
      free(sdp_buf);
      return -1;
    }
    
    nbytes = fread(sdp_buf, 1, sdp_file_stats.st_size, sdp_fp); 
    
    if(parse_sdp_file(a, addrs, ports, sdp_buf) == -1) {
      return -1;
    }
    
    if(a->alc_a.nb_channel == 0) {
      printf("Error: No acceptable channels found in SDP.");
      fflush(stdout);
      return -1;
    }
  }
  else {
    if(((a->alc_a.cc_id == Null) && (a->alc_a.nb_channel != 1))) {
      
      if(a->alc_a.addr_family == PF_INET) {
	
	addr_nb = ntohl(inet_addr(a->alc_a.addr));
	port_nb = atoi(a->alc_a.port);
	
	for(j = 0; j < a->alc_a.nb_channel; j++) {
	  
	  ipv4.sin_addr.s_addr = htonl(addr_nb);
	  
	  memset(addrs[j], 0, MAX_LENGTH);
	  memset(ports[j], 0, MAX_LENGTH);
	  
	  sprintf(addrs[j], "%s", inet_ntoa(ipv4.sin_addr));
	  sprintf(ports[j], "%i", port_nb);

	  addr_nb++;
	  port_nb++;
	}
      }
      else if(a->alc_a.addr_family == PF_INET6) {
	
	port_nb = atoi(a->alc_a.port);
	
	retcode = ushort_ipv6addr(a->alc_a.addr, &ipv6addr[0], &nb_ipv6_part);
	
	if(retcode == -1) {
	  return -1;				
	}
	
	for(j = 0; j < a->alc_a.nb_channel; j++) {
	  
	  memset(addrs[j], 0, MAX_LENGTH);
	  memset(ports[j], 0, MAX_LENGTH);
	  dup_sep = 0;
	  
	  for(k = 0; k < nb_ipv6_part + 1; k++) {
	    
	    memset(tmp, 0, 5);
	    
	    if(ipv6addr[k] != 0) {
	      sprintf(tmp, "%x", ipv6addr[k]);
	      strcat(addrs[j], tmp);
	    }
	    else {
	      if(dup_sep == 0) {
		dup_sep = 1;
	      }
	      else {
		printf("Invalid IPv6 address!\n");
		fflush(stdout);
		return -1;	
	      }	
	    }
	    
	    if(k != nb_ipv6_part) {
	      strcat(addrs[j], ":");
	    }
	  }
	  
	  sprintf(ports[j], "%i", port_nb);
	  
	  port_nb++;
	  
	  if(j != a->alc_a.nb_channel - 1) {
	    
	    for(l = nb_ipv6_part;; l--) {
	      
	      if(l == 0) {
		printf("Only %i channel possible!\n", (j + 1));
		fflush(stdout);
		return -1;	
	      }
	      
	      if(ipv6addr[l] + 1 > 0xFFFF) {
		continue;
	      }
	      
	      if(ipv6addr[l] == 0) {
		
		if(nb_ipv6_part == 7) {
		  ipv6addr[l]++;
		  break;
		}
		
		for(m = nb_ipv6_part; m > l; m--) {
		  ipv6addr[m + 1] = ipv6addr[m];	
		}
		
		ipv6addr[l + 1] = 1;
		
		nb_ipv6_part++;
		break;
	      }
	      else {
		ipv6addr[l]++;
		break;
	      }
	    }
	  }
	}
      }
    }
    else {
      memset(addrs[0], 0, MAX_LENGTH);
      memset(ports[0], 0, MAX_LENGTH);
      
      memcpy(addrs[0], a->alc_a.addr, strlen(a->alc_a.addr));
      memcpy(ports[0], a->alc_a.port, strlen(a->alc_a.port));
    }
  }
  
  if(a->alc_a.stop_time != 0) {
    time(&systime);
    curr_time = systime + 2208988800U;
    
    if(a->alc_a.stop_time <= curr_time) {
      printf("Session end time reached\n");
      fflush(stdout);
      return -1;
    }
  }
  
  *s_id = open_alc_session(&a->alc_a);
  
  if(a->alc_a.start_time != 0) {
    while(1) {
      
      time(&systime);
      curr_time = systime + 2208988800U;
      
      if((a->alc_a.start_time - 3) > curr_time) {
	
	if(!is_printed) {
	  printf("Waiting for session start time...\n");
	  fflush(stdout);
	  is_printed = true;
	}
#ifdef WIN32
	Sleep(1000);
#else
	sleep(1);
#endif
      }
      else {
	break;
      }
      
      if(get_session_state(*s_id) == SExiting) {
	return -5;
      }
    }
  }
  
  if(a->alc_a.cc_id == Null) {
    
    for(i = 0; (int)i < a->alc_a.nb_channel; i++) {
      
      if(a->alc_a.addr_type == 1) {
	*(ch_id + i) = add_alc_channel(*s_id, ports[i], addrs[0], a->alc_a.intface, a->alc_a.intface_name);
      }
      else {
	*(ch_id + i) = add_alc_channel(*s_id, ports[i], addrs[i], a->alc_a.intface, a->alc_a.intface_name);
      }
      
      if(*(ch_id + i) == -1) {
	
	while(i) {
	  retcode = remove_alc_channel(*s_id, *(ch_id + i - 1));
	  i--;
	}