from socket import * # portable socket interface plus constants from math import * from BitTornado.bencode import bencode, bdecode import thread import SocketServer # get socket server, handler objects import time, struct, sys try: sum except: sum = lambda numberlist:reduce(lambda x,y:x+y,numberlist,0.0) class myserver: def __init__(self): self.data = 'the server is started' def printmyself(self): print self.data def find_and_bind(self,handle,minport, maxport, randomizer = False): e = 'maxport less than minport - no ports to check' myHost = '' # server machine, '' means local host if maxport-minport < 50 or not randomizer: portrange = range(minport, maxport+1) if randomizer: shuffle(portrange) portrange = portrange[:20] # check a maximum of 20 ports else: portrange = [] while len(portrange) < 20: listen_port = randrange(minport, maxport+1) if not listen_port in portrange: portrange.append(listen_port) for listen_port in portrange: try: myaddr = (myHost, listen_port) self.serverobj = SocketServer.ThreadingTCPServer(myaddr, handle) return listen_port except: print "the port %d is occupied" % listen_port def serve_forever(self): self.serverobj.serve_forever() class myconnect: def start(self,serverHost,serverPort): self.sockobj = None self.sockobj = socket(AF_INET, SOCK_STREAM) # make a TCP/IP socket object if self.sockobj: try: self.sockobj.connect((serverHost, serverPort)) # connect to server machine and port except: print "can not connect to host %s, port %d" %(serverHost,serverPort) else: print "error create the sock object!!!" def send(self,message): self.rawdata = message self.sockobj.sendall(bencode(message)) def send_size(self,message): self.rawdata = message data = bencode(message) data = struct.pack('>i', len(data))+ data # print 'data to be send off', data, type(data) self.sockobj.sendall(data) #print '---data length ---', len(data), type(data) #self.sockobj.sendall(data) def recv_size(self): #data length is packed into 4 bytes total_len=0;total_data=[];size=sys.maxint size_data=sock_data='';recv_size=8192 loop = 0 while (total_len4: size_data+=sock_data size=struct.unpack('>i', size_data[:4])[0] recv_size=size if recv_size>524288:recv_size=524288 total_data.append(size_data[4:]) else: size_data+=sock_data loop = loop + 1 else: total_data.append(sock_data) total_len=sum([len(i) for i in total_data ]) try: self.bendata = bdecode(''.join(total_data)) except: self.bendata ='bdecode error' # print '_/_/_/ num of type received: _/_/_/', total_len, size return self.bendata def recv(self,msglength): self.bendata = self.sockobj.recv(msglength) #print 'encoded data:',self.bendata self.bendata = bdecode(self.bendata) return self.bendata def recv_timeout(self,msglength,timeout=2): self.sockobj.setblocking(0) self.bendata = '' msglength = 1024 total_data=[];data='';begin=time.time() while 1: # if you got some data, then break after wait sec # if total_data and time.time()-begin > timeout: # break #if you got no data at all, wait a little longer # elif time.time() - begin > timeout*2: # break try: data= self.sockobj.recv(msglength) if data: total_data.append(data) begin= time.time() else: # time.sleep(0.1) break except: break self.bendata = None print 'bendata', total_data self.bendata = ''.join(total_data) try: self.bendata = bdecode(self.bendata) except: self.bendata ='bdecode error' return self.bendata def close(self,): self.sockobj.close() def sendMSG_simple(self,host,port,cmd='',msg=''): count = 0 while 1: data = {'CMD':cmd,'DAT':msg} self.start(host,port) #self.send(data) #data = self.recv_timeout(8192) #data = self.recv(8192*4) #using send_size method self.send_size(data) data = self.recv_size() try: pass except: print 'connection failed!!!' #print 'count:',count,'data:',data if data != 'bdecode error': self.close() command = data['CMD'] message = data['DAT'] break if count >=5: self.close() command = 'ACK' message = 'ERR' break time.sleep(0.2) count = count + 1 return command, message def sendMSG(self,host,port,cmd='',msg=''): data = {'CMD':cmd,'DAT':msg} self.start(host,port) self.send_size(data) data = self.recv_size() try: pass except: print 'connection failed!!!' #print 'count:',count,'data:',data if data != 'bdecode error': self.close() command = data['CMD'] message = data['DAT'] return command, message class myconnection(myconnect,myserver): def __init__(self): # need a super peer to register self.superPeerHost = 'p2ptv.et.tudelft.nl' self.superPeerPort = 20000 def register(self,superPeerHost,superPeerPort,peerinfo): ownport = peerinfo['ownport'] peerMsg ={'peer_id': peerinfo['peer_id'],'port':ownport} #print 'I am registering' [cmd,msg] = self.sendMSG(superPeerHost,superPeerPort,'REG',peerMsg) #print 'finishing registering' if msg['DAT'] != 'ERR': print 'register sucessed! peer:', peerinfo['peer_id'] else: print 'Client received wrong ack :',peerinfo['peer_id'] # server def serverthread(self,handle,peer): peerinfo = peer.info portbase = peerinfo['serverportbase'] instanceid = peerinfo['instanceid'] nodeid = peerinfo['nodeid'] peerid = peerinfo['peer_id'] minport = int (portbase + instanceid*40 + 20*(fmod(nodeid,2))) minport = int (portbase + peerid*10) #print 'minport', minport minport = int (portbase + 20*(instanceid*2 + fmod(nodeid,2)) ) maxport = minport + 20 ownport = self.find_and_bind(handle,minport,maxport) if ownport == None: ownport = 'unknown' ownip = gethostbyaddr(gethostname())[2][0] # this is das2 special peer.info['ownip'] = ownip peer.info['ownport'] = ownport #self.register(self.superPeerHost,self.superPeerPort,peerinfo) #comment it since registeration has to be done laterly self.serverobj.serve_forever() def start_server(self,handle,peer): connection = "The server started" thread.start_new(self.serverthread, (handle,peer)) # monitor server def monitorserverthread(self,handle,listen_port): #try: myHost='p2ptv.et.tudelft.nl' myaddr =(myHost, listen_port) #print myaddr monitorserver = SocketServer.ThreadingTCPServer(myaddr, handle) #except: #print "can not listen port %d in the monitor demon" % listen_port monitorserver.serve_forever() def start_monitor_server(self,handle,port): thread.start_new(self.monitorserverthread, (handle,port,))