~p2psp/p2psp/jpl850

#!/usr/bin/python


# See: http://wiki.python.org/moin/TcpCommunication 
# See: http://docs.python.org/2/library/socket.html 

import socket 
import struct
import time
from threading import Thread
import argparse






class Splitter():
	BUFFER_SIZE = 512
	SOURCE_HOST = "127.0.0.1"
	PUERTO_SOURCE = 8000
	PUERTO_ESCUCHA = 8001
	BLOCK_SIZE = 1024
	SOURCE = (SOURCE_HOST, PUERTO_SOURCE)
	CHANNEL = 'ejemplo.ogg'
	GET_MESSAGE = 'GET /' + CHANNEL + " HTTP/1.1\r\n\r\n"
	BLOCK_FORMAT = "H" + str(BLOCK_SIZE) + "s"
	PEER_FORMAT = "4sH"
	
	
	
	
	def __init__(self, source_hostname = SOURCE_HOST, source_port = PUERTO_SOURCE, block_size = BLOCK_SIZE , listening_port = PUERTO_ESCUCHA, channel = CHANNEL, buffer_size = BUFFER_SIZE):
		


		if source_hostname:
			Splitter.SOURCE_HOST = source_hostname
		if source_port:
			Splitter.PUERTO_SOURCE = int(source_port)
		if block_size:
			Splitter.BLOCK_SIZE = int(block_size)
		if listening_port:
			Splitter.PUERTO_ESCUCHA = int(listening_port)
		if channel:
			Splitter.CHANNEL = channel
		if buffer_size:
			Splitter.BUFFER_SIZE = int(buffer_size)
		
		
		self.lista_peers = []
		self.numero_bloques = 0
		self.gatherer = None
		self.sock_source = self.conectar_source()
		self.destino_bloque = [('0.0.0.0',0) for i in xrange(self.BUFFER_SIZE)]
		self.fallos_envio = {}
		self.peer_handler = peer_handler(self.lista_peers, self.fallos_envio)
		
		self.sock_gatherer, self.gatherer = self.peer_handler.esperar_gatherer()	
		self.peer_handler.start()
		
		
		print "Cerrando conexion TCP con el gatherer..."
		self.sock_gatherer.close()
		
		self.sock_cluster = self.crear_sock_cluster()
		
		self.cluster_listener = cluster_listener(self.lista_peers, self.sock_cluster, self.destino_bloque, self.fallos_envio)
		self.cluster_listener.start()
		
		
		

	
	@staticmethod
	def evitar_bloqueo_socket(sock):
		try:
			sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		except:
			pass


	def conectar_source(self):
		
		sock_source = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self.evitar_bloqueo_socket(sock_source)
		sock_source.connect(self.SOURCE)
		print "Conectado a la fuente ", sock_source.getsockname()	
		sock_source.sendall(self.GET_MESSAGE)
		return sock_source




	def recibir_bloque(self):
		intentos = 0
		block = self.sock_source.recv(self.BLOCK_SIZE)
		while len(block) < self.BLOCK_SIZE:
			intentos += 1
			if intentos > 3:
				time.sleep(1)
				self.sock_source.close()
				self.sock_source = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
				self.evitar_bloqueo_socket(self.sock_source)
				self.sock_source.connect(self.SOURCE)
				self.sock_source.sendall(self.GET_MESSAGE)
				
			block += self.sock_source.recv(self.BLOCK_SIZE - len(block))
				
		return block



	def crear_sock_cluster(self):
		
		sock_cluster = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
		self.evitar_bloqueo_socket(sock_cluster)
		sock_cluster.bind((self.SOURCE_HOST, self.PUERTO_ESCUCHA))
		return sock_cluster
	
	@staticmethod
	def empaquetar_bloque(bloque, numero_bloque):		
		message = struct.pack(Splitter.BLOCK_FORMAT, socket.htons(numero_bloque), bloque)
		return message
		

	def go(self):
		print "Transmitiendo a los peers/gatherer segun corresponda..."
		
		while True:
			
			block = self.recibir_bloque()
			
			message = Splitter.empaquetar_bloque(block, self.numero_bloques)
			if len(self.lista_peers) > 0:
				for peer in self.lista_peers:
					
					self.sock_cluster.sendto(message, peer)
					self.destino_bloque[self.numero_bloques % self.BUFFER_SIZE] = peer
			else:				
				self.sock_cluster.sendto(message, self.gatherer)
				self.destino_bloque[self.numero_bloques % self.BUFFER_SIZE] = self.gatherer
			self.numero_bloques += 1
			

class peer_handler(Thread):
	
	
	
	def __init__(self, lista_peers, fallo_env):
		Thread.__init__(self)
		self.gatherer = None
		self.lista_peers = lista_peers
		self.fallo_envio = fallo_env
		self.setDaemon(True)
		self.daemon = True
		
		
	def obtener_socket_peer(self):
		sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		Splitter.evitar_bloqueo_socket(sock)
		sock.bind((Splitter.SOURCE_HOST, Splitter.PUERTO_ESCUCHA))
		sock.listen(0)
		return sock
	
	def esperar_gatherer(self):
		sock_gatherer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		Splitter.evitar_bloqueo_socket(sock_gatherer)
		sock_gatherer.bind((Splitter.SOURCE_HOST, Splitter.PUERTO_ESCUCHA))
		sock_gatherer.listen(0)	
		print "Esperando conexion del gatherer en el puerto --> ", Splitter.PUERTO_ESCUCHA
		sock_gatherer, gatherer = sock_gatherer.accept()
		self.gatherer = gatherer
		print "Conectado al gatherer --> ", gatherer
		sock_gatherer.setblocking(0)
		return sock_gatherer, gatherer
	
	def empaquetar_peer(self, peer):
		message = struct.pack(
                    Splitter.PEER_FORMAT, socket.inet_aton(peer[0]),
                    socket.htons(peer[1]))
		return message

	def run(self):
		while True:
			print "Esperando conexion de algun peer..."
			sock_peer, peer = self.obtener_socket_peer().accept()
			print "Conectado correctamente al peer ", peer
			self.lista_peers.append(peer)
			message = struct.pack("H", socket.htons(len(self.lista_peers)))
			print "Enviando numero de peers conectados a los demas peers..."
			sock_peer.sendall(message)
			print "Enviando lista de peers conectados a los demas peers..."
			# PRIMERO ENVIAR EL GATHERER Y LUEGO LOS DEMAS
			print self.gatherer[0], "-->", self.gatherer[1]
			sock_peer.sendall(self.empaquetar_peer(self.gatherer))
			for peerList in self.lista_peers:
				sock_peer.sendall(self.empaquetar_peer(peerList))
				self.fallo_envio[peerList] = 0
			
			print "Hecho"
			sock_peer.close()
	
			
class cluster_listener(Thread):
	#Esta clase crea un thread para escuchar las despedidas voluntarias de peers del cluster
	
	
	def __init__(self, lista, socketCluster, destino, fallo_envio):
		Thread.__init__(self)
		self.lista_peers = lista
		self.sock_cluster = socketCluster
		self.setDaemon(True)
		self.daemon = True
		self.destino_bloque = destino
		self.fallos_envio = fallo_envio
	
	def run(self):
		message, peer = self.sock_cluster.recvfrom(struct.calcsize("H"))
		if len(message) == 0:			
			try:
				print "Mensaje de despedida recibido desde el peer --> ", peer
				self.lista_peers.remove(peer)
				print "Eliminado correctamente de la lista de peers"
			except:
				print "Error al eliminar el peer, seguro que esta en la lista?"
				pass
		else:
			#En este caso nos esta enviando el numero de un bloque perdido
			numero_bloque_perdido = struct.unpack("!H", message)[0]
			destino_perdido = self.destino_bloque[numero_bloque_perdido]
			print "Bloque numero ", numero_bloque_perdido, " con destino el nodo --> ", destino_perdido
			
			self.fallos_envio[destino_perdido] += 1
			
			if self.fallos_envio[destino_perdido] > 5:
				print "Debido a las perdidas de bloques, el siguiente peer sera eliminado de la lista --> ", destino_perdido
			
			






parser = argparse.ArgumentParser(description='Splitter --> Proyecto p2psp')


parser.add_argument('--listening_port')

parser.add_argument('--source_hostname')

parser.add_argument('--channel')

parser.add_argument('--source_port')

parser.add_argument('--block_size')

parser.add_argument('--buffer_size')



args = parser.parse_known_args()[0]







splitter = Splitter(args.source_hostname, args.source_port, args.block_size, args.listening_port, args.channel, args.buffer_size)
splitter.go()