use non blocking sockets everywhere in FT

src/common/connection.py

@@ -407,16 +407,16 @@ class Connection:
 			sock5 = socks5.Socks5Receiver(host = streamhost['host'], \
 				port = int(streamhost['port']), initiator = streamhost['jid'], 
 				target = target, sid = sid, file_props = file_props)
-			ret = gajim.socks5queue.add_receiver(self.name, sock5)
-			if ret is None:
-				continue
 			iq = common.xmpp.Iq(to = streamhost['jid'], typ = 'result', frm = target)
 			iq.setAttr('id', id)
 			query = iq.setTag('query')
 			query.setNamespace(common.xmpp.NS_BYTESTREAM)
 			stream_tag = query.setTag('streamhost-used')
 			stream_tag.setAttr('jid', streamhost['jid'])
-			self.to_be_sent.append(iq)
+			ret = gajim.socks5queue.add_receiver(self.name, sock5, 
+				self.to_be_sent.append, iq)
+			if ret is None:
+				continue
 			raise common.xmpp.NodeProcessed
 		
 	def _bytestreamResultCB(self, con, iq_obj):

src/common/socks5.py

@@ -21,6 +21,7 @@
 
 import socket
 import select
+import os
 try:
 	import fcntl
 except:
@@ -28,6 +29,8 @@ except:
 import struct
 import sha
 
+MAX_BUFF_LEN = 65536
+
 class SocksQueue:
 	''' queue for all file requests objects '''
 	def __init__(self, complete_transfer_cb = None, \
@@ -64,19 +67,26 @@ class SocksQueue:
 			result = sender.send_file(file_props)
 			self.process_result(result, sender)
 	
-	def add_receiver(self, account, sock5_receiver):
+	def add_receiver(self, account, sock5_receiver, auth_cb, auth_param):
 		''' add new file request '''
 		self.readers[self.idx] = sock5_receiver
 		sock5_receiver.queue_idx = self.idx
 		sock5_receiver.queue = self
 		sock5_receiver.account = account
+		sock5_receiver.auth_cb = auth_cb
+		sock5_receiver.auth_param = auth_param
 		self.idx += 1
 		result = sock5_receiver.connect()
-		self.connected += 1
+		if result != None:
+			sock5_receiver._sock.setblocking(False)
+			result = sock5_receiver.main()
+			self.process_result(result, sock5_receiver)
+			self.connected += 1
+			return 1
 		# we don't need blocking sockets anymore
 		# this unblocks ui! 
-		sock5_receiver._sock.setblocking(False)
+		
-		return result
+		return None
 	
 	def add_file_props(self, account, file_props):
 		if file_props is None or \
@@ -107,33 +117,51 @@ class SocksQueue:
 		for idx in self.senders.keys():
 			sender = self.senders[idx]
 			if sender.connected:
-				if sender.state == 1:
+				
+				if sender.state < 5:
 					if sender.pending_data():
-						result = sender.get_data()
+						result = sender.main()
-						if result is not None:
+						if sender.state == 4:
-							sender.state = 2
+							self.result_sha(sender.sha_msg, idx)
-							self.result_sha(result, idx)
+						if result is None:
-						else:
+							continue
+						if result == -1:
 							sender.disconnect()
-				elif sender.state == 3:
+				elif sender.state == 7:
-					result = sender.write_next()
+					for i in range(10):
-					self.process_result(result, sender)
+						if sender.file_props['paused']:
-				elif sender.state == 4:
+							break
+						if not sender.connected:
+							self.process_result(-1, sender)
+							break
+						if sender.state == 8:
+							self.remove_sender(idx)
+							break
+						result = sender.write_next()
+						self.process_result(result, sender)
+				elif sender.state == 8:
 					self.remove_sender(idx)
 			else:
-				self.remove_sender(idx)
+				self.remove_sender(idx)
 		for idx in self.readers.keys():
 			receiver = self.readers[idx]
 			if receiver.connected:
 				if receiver.file_props['paused']:
 					continue
-				if receiver.pending_data():
+				if receiver.state == 5:
 					result = receiver.get_file_contents(timeout)
 					self.process_result(result, receiver)
+				else:
+					pd = receiver.pending_data()
+					if pd:
+						result = receiver.main(timeout)
+						self.process_result(result, receiver)
 			else:
 				self.remove_receiver(idx)
 		
 	def process_result(self, result, actor):
+		if result is None:
+			return
 		if result in [0, -1] and \
 			self.complete_transfer_cb is not None:
 			self.complete_transfer_cb(actor.account, 
@@ -141,7 +169,7 @@ class SocksQueue:
 		elif self.progress_transfer_cb is not None:
 			self.progress_transfer_cb(actor.account, 
 				actor.file_props)
-	
+	
 	def remove_receiver(self, idx):
 		if idx != -1:
 			if self.readers.has_key(idx):
@@ -165,57 +193,74 @@ class Socks5:
 		self.sid = sid
 		self._sock = None
 		self.account = None
+		self.state = 0 # not connected
 	
 	def connect(self):
 		self._sock=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-		self._sock.connect((self.host, self.port))
+		try:
-		self._sock.setblocking(True)
+			self._sock.connect((self.host, self.port))
-		self._send=self._sock.sendall
+			self._sock.setblocking(False)
-		self._recv=self._sock.recv
+			self._send=self._sock.send
+			self._recv=self._sock.recv
+		except Exception, e:
+			return None
+		self.buff = ''
 		self.connected = True
-		return self.send_connect()
+		self.state = 1 # connected
+		return 1
 	
 	def receive(self):
-		''' Reads all pending incoming data. 
+		''' Reads small pending incoming data. 
 			Calls owner's disconnected() method if appropriate.'''
-		try: 
+		
-			received = self._recv(64)
+		if self.pending_read():
-		except: 
 			received = ''
-	
+			while self.pending_read():
-		while self.pending_data():
+				try: 
-			try: 
+					add = self._recv(64)
-				add = self._recv(64)
+				except Exception, e: 
-			except: 
+					add=''
-				add=''
+				received +=add
-			received +=add
+				if not add: 
-			if not add: 
+					break
-				break
+			if len(received) == 0:
-		if len(received) == 0:
+				self.disconnect()
-			self.disconnect()
+		else:
+			return None
 		return received
+		
+
 
 	def send_raw(self,raw_data):
 		''' Writes raw outgoing data. Blocks until done.
 			If supplied data is unicode string, encodes it to utf-8 before send.'''
 		try:
-			self._send(raw_data)
+			lenn = self._send(raw_data)
-		except:
+		except Exception, e:
 			self.disconnect()
-		pass
+		return len(raw_data)
 			
 	def disconnect(self):
 		''' Closes the socket. '''
 		self._sock.close()
 		self.connected = False
 		
-	def pending_data(self,timeout=0):
+	def pending_read(self,timeout=0):
 		''' Returns true if there is a data ready to be read. '''
 		if self._sock is None:
 			return False
 		try:
 			return select.select([self._sock],[],[],timeout)[0]
-		except:
+		except Exception, e:
+			return False
+		
+	def pending_write(self,timeout=0):
+		''' Returns true if there is a data ready to be read. '''
+		if self._sock is None:
+			return False
+		try:
+			return select.select([],[self._sock],[],timeout)[0]
+		except Exception, e:
 			return False
 			
 	def pending_connection(self,timeout=0):
@@ -224,32 +269,9 @@ class Socks5:
 			return False
 		try:
 			return select.select([],[self._sock],[],timeout)[0]
-		except:
+		except Exception, e:
 			return False
 	
-	def send_connect(self):
-		''' begin negotiation. on success 'address' != 0 '''
-		self.send_raw(self._get_auth_buff())
-		buff = self.receive()
-		version, method = struct.unpack('!BB', buff[:2])
-		if version != 0x05 or method == 0xff:
-			self.disconnect()
-			return None
-		self.send_raw(self._get_request_buff(self._get_sha1_auth()))
-		buff = self.receive()
-		version, command, rsvd, address_type = struct.unpack('!BBBB', buff[:4])
-		addrlen, address, port = 0, 0, 0
-		if address_type == 0x03:
-			addrlen = ord(buff[4])
-			address = struct.unpack('!%ds' % addrlen, buff[5:addrlen + 5])
-			portlen = len(buff[addrlen + 5])
-			if portlen == 1: # Gaim bug :)
-				port, = struct.unpack('!B', buff[addrlen + 5])
-			else: 
-				port, = struct.unpack('!H', buff[addrlen + 5])
-		return (version, command, rsvd, address_type, addrlen, address, port)
-		
-		
 	def _get_auth_buff(self):
 		''' Message, that we support 1 one auth mechanism: 
 		the 'no auth' mechanism. '''
@@ -308,12 +330,15 @@ class Socks5:
 	
 	def _get_sha1_auth(self):
 		return sha.new("%s%s%s" % (self.sid, self.initiator, self.target)).hexdigest()
+
 class Socks5Sender(Socks5):
+	''' class for sending file to socket over socks5 '''
 	def __init__(self, sock_hash, parent, _sock, host = None, port = None):
 		self.queue_idx = sock_hash
 		self.queue = parent
 		Socks5.__init__(self, host, port, None, None, None)
 		self._sock = _sock
+		self._sock.setblocking(False)
 		self._recv = _sock.recv
 		self._send = _sock.send
 		self.connected = True
@@ -322,19 +347,18 @@ class Socks5Sender(Socks5):
 		self.remaining_buff = ''
 		
 	def write_next(self):
-		send_size = 65536
 		if self.remaining_buff != '':
 			buff = self.remaining_buff
 			self.remaining_buff = ''
 		else:
-			buff = self.fd.read(send_size)
+			buff = self.fd.read(MAX_BUFF_LEN)
 		if len(buff) > 0:
 			lenn = 0
 			try:
 				lenn = self._send(buff)
 			except Exception, e:
 				if e.args[0] != 11:
-					self.state = 4
+					self.state = 8
 					self.fd.close()
 					self.disconnect()
 					self.file_props['error'] = -1
@@ -342,7 +366,7 @@ class Socks5Sender(Socks5):
 			self.size += lenn
 			self.file_props['received-len'] = self.size
 			if self.size == int(self.file_props['size']):
-				self.state = 4
+				self.state = 8
 				self.file_props['error'] = 0
 				self.fd.close()
 				self.disconnect()
@@ -355,13 +379,16 @@ class Socks5Sender(Socks5):
 				self.pauses +=1
 			else:
 				self.pauses = 0
-			if self.pauses > 20:
+			if self.pauses > 24:
-				self.file_props['paused'] = True
+				self.file_props['stalled'] = True
 			else:
-				self.file_props['paused'] = False
+				self.file_props['stalled'] = False
+			self.state = 7
+			if lenn == 0 and self.file_props['stalled'] is False:
+				return None
 			return lenn
 		else:
-			self.state = 4
+			self.state = 8
 			self.disconnect()
 			return -1
 		
@@ -372,31 +399,28 @@ class Socks5Sender(Socks5):
 		file_props['started'] = True
 		file_props['completed'] = False
 		file_props['paused'] = False
+		file_props['stalled'] = False
 		file_props['received-len'] = 0
+		
 		self.pauses = 0
 		self.file_props = file_props
 		self.size = 0
-		self.state = 3
-		self._sock.setblocking(False)
-		
 		return self.write_next()
 		
-	def get_data(self):
+	def main(self):
 		if self.state == 1:
 			buff = self.receive()
 			if not self.connected:
 				return -1
 			mechs = self._parse_auth_buff(buff)
-			self._sock.setblocking(True)
+		elif self.state == 2:
 			self.send_raw(self._get_auth_response())
-			
+		elif self.state == 3:
 			buff = self.receive()
-			
+			(req_type, self.sha_msg, port) = self._parse_request_buff(buff)
-			(req_type, sha_msg, port) = self._parse_request_buff(buff)
+		elif self.state == 4:
-			self.send_raw(self._get_request_buff(sha_msg, 0x00))
+			self.send_raw(self._get_request_buff(self.sha_msg, 0x00))
-			self.state = 2
+		self.state += 1
-			self._sock.setblocking(False)
-			return sha_msg
 		return None
 			
 	def pending_data(self,timeout=0):
@@ -404,10 +428,10 @@ class Socks5Sender(Socks5):
 		if self._sock is None:
 			return False
 		try:
-			if self.state == 1:
+			if self.state in [1, 3]:
-				return select.select([self._sock],[],[],timeout)[0]
+				return self.pending_read()
-			elif self.state == 3:
+			elif self.state in [2, 4, 5]:
-				return select.select([],[self._sock],[],timeout)[0]
+				return True
 		except Exception, e:
 			return False
 		return False
@@ -417,7 +441,8 @@ class Socks5Sender(Socks5):
 		# close connection and remove us from the queue
 		self._sock.close()
 		self.connected = False
-		self.file_props['disconnect_cb'] = None
+		if self.file_props is not None:
+			self.file_props['disconnect_cb'] = None
 		if self.queue is not None:
 			self.queue.remove_sender(self.queue_idx)
 
@@ -446,8 +471,7 @@ class Socks5Listener:
 	def accept_conn(self):
 		self._serv.accept.__doc__
 		_sock  = self._serv.accept()
-		# block it untill authorization is sent
+		_sock[0].setblocking(False)
-		_sock[0].setblocking(True)
 		return _sock
 	
 	def pending_connection(self,timeout=0):
@@ -467,23 +491,78 @@ class Socks5Receiver(Socks5):
 		self.queue = None
 		self.file_props = file_props
 		self.file_props['started'] = True
-		self.connected = False
+		self.connected = False
+		self.pauses = 0
 		if file_props:
 			file_props['disconnect_cb'] = self.disconnect
 			file_props['error'] = 0
 			self.file_props['started'] = True
 			self.file_props['completed'] = False
 			self.file_props['paused'] = False
-			self.file_props['started'] = True
+			self.file_props['stalled'] = False
+			self.file_props['started'] = True
 		Socks5.__init__(self, host, port, initiator, target, sid)
 	
+	
+	def main(self, timeout = 0):
+		''' begin negotiation. on success 'address' != 0 '''
+		if self.state == 1:
+			self.send_raw(self._get_auth_buff())
+		elif self.state == 2:
+			buff = self.receive()
+			if buff is None or len(buff) != 2:
+				return None
+			version, method = struct.unpack('!BB', buff[:2])
+			if version != 0x05 or method == 0xff:
+				self.disconnect()
+		elif self.state == 3:
+			self.send_raw(self._get_request_buff(self._get_sha1_auth()))
+		elif self.state == 4:
+			buff = self.receive()
+			if buff == None:
+				return None
+			version, command, rsvd, address_type = struct.unpack('!BBBB', buff[:4])
+			addrlen, address, port = 0, 0, 0
+			if address_type == 0x03:
+				addrlen = ord(buff[4])
+				address = struct.unpack('!%ds' % addrlen, buff[5:addrlen + 5])
+				portlen = len(buff[addrlen + 5])
+				if portlen == 1: # Gaim bug :)
+					port, = struct.unpack('!B', buff[addrlen + 5])
+				else:
+					port, = struct.unpack('!H', buff[addrlen + 5])
+			self.state = 5
+			self.auth_cb(self.auth_param)
+			return None
+		if self.state < 5:
+			self.state += 1
+			return None
+		# we have set the connection, retrieve file
+		
+		return self.get_file_contents(timeout)
+	
+	
+	def pending_data(self, timeout=0):
+		''' Returns true if there is a data ready to be read. ''', self.state
+		if self._sock is None:
+			return False
+		try:
+			if self.state in [2, 4]:
+				return self.pending_read()
+			elif self.state in [1, 3, 5]:
+				return True
+		except Exception, e:
+			return False
+		return False
+		
 	def get_file_contents(self, timeout):
-		''' read file contents from socket and write them to file "'''
+		''' read file contents from socket and write them to file '''
 		if self.file_props is None or \
 			self.file_props.has_key('file-name') is False:
-			return 
+			self.file_props['error'] = -2
-			#TODO error
+			return None
-		while self.pending_data(timeout):
+		fd = None
+		while self.pending_read(timeout):
 			if self.file_props.has_key('fd'):
 				fd = self.file_props['fd']
 			else:
@@ -491,8 +570,8 @@ class Socks5Receiver(Socks5):
 				self.file_props['fd'] = fd
 				self.file_props['received-len'] = 0
 			try: 
-				buff = self._recv(65536)
+				buff = self._recv(MAX_BUFF_LEN)
-			except: 
+			except Exception, e:
 				buff = ''
 			self.file_props['received-len'] += len(buff)
 			fd.write(buff)
@@ -503,7 +582,7 @@ class Socks5Receiver(Socks5):
 				try:
 					# file is not complete, remove it
 					os.remove(self.file_props['file-name'])
-				except:
+				except Exception, e:
 					# unable to remove the incomplete file
 					pass
 				self.disconnect()
@@ -518,7 +597,20 @@ class Socks5Receiver(Socks5):
 				self.file_props['completed'] = True
 				return 0
 		# return number of read bytes. It can be used in progressbar
-		return self.file_props['received-len']
+		if fd == None:
+			self.pauses +=1
+		else:
+			self.pauses = 0
+		if self.pauses > 24:
+			self.file_props['stalled'] = True
+		else:
+			self.file_props['stalled'] = False
+		if fd == None and self.file_props['stalled'] is False:
+			return None
+		if self.file_props.has_key('received-len'):
+			if self.file_props['received-len'] != 0:
+				return self.file_props['received-len']
+		return None
 		
 	def disconnect(self):
 		''' Closes the socket. '''

src/gajim.py

@@ -733,6 +733,8 @@ class Interface:
 				file_props['received-len'])
 		else:
 			ft.set_status(file_props['type'], file_props['sid'], 'stop')
+		if file_props['stalled'] or file_props['paused']:
+			return
 		if gajim.config.get('notify_on_file_complete'):
 			if gajim.config.get('autopopupaway') or \
 				gajim.connections[account].connected in (2, 3): # we're online or chat