Thread: SF.net SVN: fclient: [5] trunk/fclient/fclient_lib/fcp/fcp20.py

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 454-5900

Revision: 5
          http://fclient.svn.sourceforge.net/fclient/?rev=5&view=rev
Author:   jurner
Date:     2007-10-16 03:19:24 -0700 (Tue, 16 Oct 2007)

Log Message:
-----------
continued working on FcpClient

Modified Paths:
--------------
    trunk/fclient/fclient_lib/fcp/fcp20.py

Modified: trunk/fclient/fclient_lib/fcp/fcp20.py
===================================================================

--- trunk/fclient/fclient_lib/fcp/fcp20.py	2007-10-15 17:42:24 UTC (rev 4)
+++ trunk/fclient/fclient_lib/fcp/fcp20.py	2007-10-16 10:19:24 UTC (rev 5)
@@ -1,9 +1,15 @@
- 
+
+import atexit
+import logging 
 import os
 import socket
+import subprocess
+import sys
 import time
 import thread
 import uuid
+
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
 #**************************************************************
 # consts
 #**************************************************************
@@ -13,7 +19,6 @@
 	DefaultFcpPort = int(os.environ.get('FCP_PORT', '').strip())
 except: pass
 SocketTimeout = 0.1
-KeyTypes = ('SSK@', 'KSK@', 'CHK@', 'USK@', 'SVK@')
 
 class JobIdentifiers:
 	# fixed job identifiers
@@ -200,6 +205,13 @@
 		fp.close()
 	return read
 
+def saveRemoveFile(fpath):
+	if fpath is not None:
+		if os.path.isfile(fpath):
+			os.remove(fpath)
+			return True
+	return False
+		
 
 def saveWriteFile(fpath, data):
 	"""Writes data to a file i the savest manner possible
@@ -221,6 +233,22 @@
 			fp.close()
 	return written
 
+def startFreenet(cmdline):
+	"""Starts freenet
+	@param cmdline: commandline to start freenet (like '/freenet/run.sh start' or 'c:\freenet\start.bat')
+	@return: (string) whatever freenet returns
+	"""
+	#TODO: on windows it may be necessary to hide the comand window
+	p = subprocess.Popen(
+				args=cmdline,
+				shell=True,
+				stdout=subprocess.PIPE,
+				stderr=subprocess.PIPE,
+				)
+	stdout, stderr = p.communicate()
+	return stdout
+
+
 def fcpBool(pythonBool):
 	"""Converts a python bool to a fcp bool
 	@param pythonBool: (bool)
@@ -237,6 +265,95 @@
 	"""
 	return fcpBool == 'true'
 
+
+def uriToList(uri):
+	"""Splits a freenet uri into its components
+	@param uri: (str) uri to split
+	@return: (list) or components
+	@note: additional dexoration like 'freenet:' or 'http://blah' are ignored
+	
+	>>> uriToList('SSK@foo')
+	['SSK@foo']
+	
+	>>> uriToList('SSK@foo/bar/baz')
+	['SSK@foo', 'bar', 'baz']
+	
+	>>> uriToList('freenet:SSK@foo')
+	['SSK@foo']
+	
+	>>> uriToList('http://foo/SSK@foo')
+	['SSK@foo']
+	
+	>>> uriToList('http://foo')
+	[]
+	
+	"""
+	if uri.startswith('freenet:'):
+		uri = uri[len('freenet:'): ]
+	
+	components = []
+	head = uri
+	while head:
+		head, tail = posixpath.split(head)
+		components.append(tail)
+		
+	components.reverse()	
+	
+	while components:
+		if components[0][:4] in FcpKeys.KeysAll:
+			break
+		else:
+			components.pop(0)
+			
+	return components
+
+def uriKeyType(uri):
+	pass
+
+
+def isValidUri(uri):
+	"""Checks if an uri is a valid freenet uri
+	@param uri: (str) uri to check
+	@reuturn: (bool)
+	"""
+	return bool(UriToList(uri))
+
+
+def splitUri(uri):
+	"""Splits an uri into uri and filename
+	@param uri: uri to split
+	@return: tuple(uri, filename)
+	
+	>>> splitUri('SSK@foo/bar/baz')
+	('SSK@foo', 'bar/baz')
+	
+	>>> splitUri('SSK@foo')
+	('SSK@foo', '')
+	
+	>>> splitUri('NoUri')
+	('NoUri', '')
+	
+	"""
+	L = uriToList(uri)
+	if not L:
+		return (uri, '')
+	names = L[1:]
+	if not names:
+		name = ''
+	else:
+		name = '/'.join(names)
+	return (L[0], name)
+
+
+def fileNameFromUri(uri):
+	"""Returns the filename part of an uri
+	@return: (str) filename. If no filename is found the uri is returned unchanged
+	"""
+	tmp_uri, name = splitUri(uri)
+	if name:
+		return name
+	return uri
+	
 #**********************************************************************
 # classes
 #**********************************************************************
@@ -303,7 +420,6 @@
 #**************************************************************************
 # jobs
 #**************************************************************************
-#TODO: do somrthing that this class does not lock the queue
 class JobBase(object):
 	"""Base class for jobs"""
 	
@@ -311,18 +427,28 @@
 	
 	def __init__(self, fcpClient, identifier, message):
 		
-		self.fcpClient = fcpClient
-		self.fcpIdentifier = identifier
-		self.fcpMessage = message
-		self.fcpResult = None
-		self.fcpTime = 0
+		self.fcpClient = fcpClient					# FcpClient() instance the job belongs to
+		self.fcpError = None						# last error (either this is set or dcpResult)
+		self.fcpIdentifier = identifier				#
+		self.fcpMessage = message				# message send to node
+		self.fcpResult = None						# job result
+		self.fcpTime = 0								# start time (will hld duration whern the job is complte)
 		
+	def displayName(self):
+		return 'JobBase'
+	
 	def start(self):
 		self.fcpTime = time.time()
 		self.fcpClient.sendMessageEx(self.fcpMessage)
 		
+	def error(self, msg):
+		self.fcpTime = time.time() - self.fcpTime
+		self.fcpError = msg
+		self.fcpResult = None
+	
 	def stop(self, result):
 		self.fcpTime = time.time() - self.fcpTime
+		self.fcpError = None
 		self.fcpResult = result
 
 
@@ -338,8 +464,10 @@
 				)
 		JobBase.__init__(self, fcpClient, JobIdentifiers.ClientHello, message)
 
+	def displayName(self):
+		return 'NodeHello'
+	
 
-
 class JobListPeers(JobBase):
 	
 	_fcp_auto_remove_ = True
@@ -353,15 +481,21 @@
 		JobBase.__init__(self, fcpClient, JobIdentifiers.ListPeers, message)
 		
 	
+	def displayName(self):
+		return 'ListPeers'
+	
 	def handlePeer(self,msg):
 		pass
 
 
 
-class JobFileInfo(JobBase):
+class JobGetFileInfo(JobBase):
+	""""""
 	
 	_fcp_auto_remove_ = False
 	
+	
+	# idea is to provoke a GetFailed message and take mimetype and size from it
 	def __init__(self, fcpClient, uri, **params):
 		"""
 		@param fcpClient: FcpClient() instance
@@ -378,7 +512,7 @@
 				Messages.ClientGet, 
 				Identifier=identifier, 
 				URI=uri,
-				MaxSize='ase0',
+				MaxSize='0',
 				ReturnType='none',
 				Verbosity='1',
 				**params
@@ -386,36 +520,37 @@
 		JobBase.__init__(self, fcpClient, identifier, message)
 		
 			
+	def displayName(self):
+		return 'GetFileInfo'
+	
 	def handleProgress(self, msg):
 		pass
 
 
-	def stop(self, msg):
-		JobBase.stop(self, msg)
-		error = result = None
+	def error(self, msg):
+		JobBase.error(self, msg)
 		if msg.name == Messages.GetFailed:
 			if msg['Code'] == FetchErrors.TooBig:
-				result = (
+				self.fcpError = None
+				self.fcpResult = (
 								msg.get('ExpectedMetadata.ContentType', ''), 
 								msg.get('ExpectedDataLength', '')
 							)
-			else:
-				error, result = msg['Code'], msg
-							
-		elif msg.name == Messages.DataFound:
-			result = (
+		#else:
+		#	raise ValueError('Unhandled message: %s' % msg.name)
+	
+	
+	def stop(self, msg):
+		JobBase.stop(self, msg)
+		if msg.name == Messages.DataFound:
+			self.fcpResult = (
 								msg.get('Metadata.ContentType', ''), 
 								msg.get('DataLength', '')
 							)
-		
-		elif msg.name == Messages.ProtocolError:
-			error, result = msg['Code'], msg
-			
 		else:
 			raise ValueError('Unhandled message: %s' % msg.name)
-				
-		self.fcpResult = error, result
 		
+
 		
 #TODO: handle case where directories are registered multiple times
 class JobTestDDA(JobBase):
@@ -432,7 +567,9 @@
 		JobBase.__init__(self, fcpClient, directory, message)
 		self.fcpTmpFile = None
 		
-
+	def displayName(self):
+		return 'TestDDA'
+	
 	def handleTestDDAReply(self, msg):
 		fpathWrite = msg.params.get('WriteFilename', None)
 		fpathRead = msg.params.get('ReadFilename', None)
@@ -456,19 +593,58 @@
 				ReadContent=readContent,
 				)	
 	
+	def error(self, msg):
+		JobBase.error(self, msg)
+		saveRemoveFile(self.fcpTmpFile)
+		self.fcpTmpFile = None
+	
+	
 	def stop(self, msg):
 		JobBase.stop(self, msg)
-		if self.fcpTmpFile is not None:
-			if os.path.isfile(self.fcpTmpFile):
-				os.remove(self.fcpTmpFile)
+		saveRemoveFile(self.fcpTmpFile)
+		self.fcpTmpFile = None
 
 #**************************************************************************
 # fcp client
 #**************************************************************************
+class LogMessages:
+	Connecting = 'Connecting to node...'
+	Connected = 'Connected to node'
+	ConnectionRetry = 'Connecting to node failed... retrying'
+	ConnectingFailed = 'Connecting to node failed'
+	
+	ClientClose = 'Closing client'	
+	
+	MessageSend = 'Message send'
+	MessageReceived = 'Message received'
+	
+	JobStart = 'Starting job: '
+	JobStop  = 'Stopping job: '
+	JobsCompleted = 'All jobs completed'
+	
+	KeyboardInterrupt = 'Keyboard interrupt'
+	SocketDead = 'Socket is dead'
+	
+	
+#TODO: no idea what happens on reconnect if socket died. What about running jobs?
+#TODO: name as specified in NodeHello seems to be usable to keep jobs alive. Have to test this.
 class FcpClient(object):
 	
-	def __init__(self):
-		
+	def __init__(self, 
+			name='', 
+			errorHandler=None, 
+			verbosity=logging.WARNING, 
+			logMessages=LogMessages
+			):
+		"""
+		@param name: name of the client instance or '' (for debugging)
+		@param errorHandler: will be called if the socket conncetion to the node is dead
+		with two params: FcpSocketError + details. When the handler is called the client
+		is already closed.
+		@param verbosity: verbosity level for debugging
+		@param logMessages: LogMessages class containing messages
+		"""
+				
 		self._isConnected = False
 		self._jobs = {
 			'all': {},
@@ -476,24 +652,34 @@
 			'running': [],
 			'complete': [],
 			}
+		self._errrorHandler = errorHandler
+		self._log = logging.getLogger('FcpClient20:%s' % name)
+		self._logMessages = logMessages
 		self._lock = thread.allocate_lock()
 		self._socket = None
-				
 		
+		self.setVerbosity(verbosity)		
+		atexit.register(self.close)
 		
 	def close(self):
+		self._log.info(self._logMessages.ClientClose)
 		if self._socket is not None:
 			self._socket.close()
 			self._socket = None
 	
 	
+	#TODO: an iterator would be nice for Guis, to enshure they stay responsitive in the call
 	def connect(self, host=DefaultFcpHost, port=DefaultFcpPort, repeat=20, timeout=0.5):
+		
+		self._log.info(self._logMessages.Connecting)
+		
 		# poll untill freenet responds
 		time_elapsed = 0
 		while time_elapsed <= repeat:
 						
 			# try to Connect socket
-			self.close()
+			if self._socket is not None:
+				self.close()
 			self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 			self._socket.settimeout(SocketTimeout)
 			try:
@@ -501,19 +687,19 @@
 			except Exception, d:
 				pass
 			else:
-				#self._isConnected = True
+				self._log.info(self._logMessages.Connected)
 				return True
 				
+			self._log.info(self._logMessages.ConnectionRetry)
+			
 			# continue polling
 			time_elapsed += timeout	
 			time.sleep(timeout)
 		
+		self._log.info(self._logMessages.ConnectingFailed)
 		return False
 	
 	
-	#def __nonzero__(self):
-	#	return self._isConnected
-	
 	def addJob(self, job):
 		self._lock.acquire(True)			
 		try:
@@ -523,9 +709,12 @@
 			self._jobs['running'].append(job)
 		finally:
 			self._lock.release()
+		
+		self._log.info(self._logMessages.JobStart + job.displayName())
 		job.start()
 		
-	def finishJob(self, identifier, msg):
+	
+	def stopJob(self, identifier, msg, error=False):
 		self._lock.acquire(True)
 		try:
 			job = self._jobs['all'].get(identifier, None)
@@ -540,7 +729,11 @@
 			
 		if job is None:
 			raise ValueError('No such job: %r' % identifier)
-		job.stop(msg)
+		self._log.info(self._logMessages.JobStop + job.displayName())
+		if error:
+			job.error(msg)
+		else:
+			job.stop(msg)
 				
 	
 	def notifyJob(self, identifier, handler, msg):
@@ -558,27 +751,25 @@
 		
 		# TODO: 
 		#    x. push pending jobs
-		#    x. on error stop this thingy
-		
-		n = 0
-		while True:
-			if not self._lock.acquire(False):
-				continue
+		try:
+			while True:
+				if not self._lock.acquire(False):
+					continue
+							
+				try:
+					if not self._jobs['pending'] and not self._jobs['running']:
+						self._log.info(self._logMessages.JobsCompleted)
+						break
+				finally:
+					self._lock.release()
+				
+				self.next()
+		except KeyboardInterrupt:
+			self._log(self._logMessages.KeyboardInterrupt)
+			self.close()
 						
-			try:
-				if not self._jobs['pending'] and not self._jobs['running']:
-					break
-			finally:
-				self._lock.release()
-			
-			msg = self.readMessage()
-			self.handleMessage(msg)
-			
-			
-			n += 1
-			if n > 50: break
 		
-	
+	#TODO: some info when all jobs are completed
 	def next(self):
 		msg = self.readMessage()
 		self.handleMessage(msg)
@@ -586,33 +777,32 @@
 	
 	def handleMessage(self, msg):
 			
-		print msg.pprint()
-		
+		self._log.debug(self._logMessages.MessageReceived + '\n' + msg.pprint())
+				
 		if msg.name == Messages.NodeHello:
 			#connectionIdentifier = msg['ConnectionIdentifier']
-			self.finishJob(JobIdentifiers.ClientHello, msg)
+			self.stopJob(JobIdentifiers.ClientHello, msg)
 				
 		elif msg.name == Messages.ProtocolError:
 			code = msg['Code']
-						
 			if code == ProtocolErrors.NoLateClientHellos:
-				self.finishJob(JobIdentifiers.ClientHello, msg)
+				self.stopJob(JobIdentifiers.ClientHello, msg, error=True)
 			
 			else:
 				identifier = msg.get('Identifier', None)
 				if identifier is None:
 					pass # raise ???
 				else:
-					self.finishJob(identifier, msg)	
+					self.stopJob(identifier, msg, error=True)	
 				
 		elif msg.name == Messages.Peer:
 			self.notifyJob(JobIdentifiers.ListPeers, 'handlePeer', msg)
 		
 		elif msg.name == Messages.EndListPeers:
-			self.finishJob(IdentifierListPeers, msg)
+			self.stopJob(IdentifierListPeers, msg)
 			
 		elif msg.name == Messages.GetFailed:
-			self.finishJob(msg['Identifier'], msg)
+			self.stopJob(msg['Identifier'], msg, error=True)
 					
 		elif msg.name == Messages.SimpleProgress:
 			self.notifyJob(msg['Identifier'], 'handleProgress', msg)
@@ -621,7 +811,7 @@
 			self.notifyJob(msg['Directory'], 'handleTestDDAReply', msg)
 		
 		elif msg.name == Messages.TestDDAComplete:
-			self.finishJob(msg['Directory'], msg)
+			self.stopJob(msg['Directory'], msg)
 			
 		elif msg.name == Messages.IdentifierCollision:
 			pass
@@ -642,6 +832,10 @@
 				msg = MessageSocketTimeout()
 				break
 			except Exception, d:
+				self._log.info(self._logMessages.SocketDead)
+				self.close()
+				if self._errorHandler is not None:
+					self._errorHandler(FcpSocketError, d)
 				raise FcpSocketError(d)			#!!
 						
 			if p == '\r':		# ignore
@@ -662,7 +856,12 @@
 				n = int(msg.params['DataLength'])
 				try:
 					msg.data = self._socket.recv(n)
+					if not msg.data: raise ValueError('Socket is dead')
 				except Exception, d:
+					self._log.info(self._logMessages.SocketDead)
+					self.close()
+					if self._errorHandler is not None:
+						self._errorHandler(FcpSocketError, d)
 					raise FcpSocketError(d)			#!!
 							
 			else:
@@ -674,8 +873,13 @@
 										
 		return msg
 		
-		
 	
+	def setLogMessages(self, logMessages):
+		self._logMessages = logMessages
+	
+	def setVerbosity(self, verbosity):
+		self._log.setLevel(verbosity)	
+	
 	def sendMessage(self, name, data=None, **params):
 		"""Sends a message to freenet
 		@param name: name of the message to send
@@ -691,25 +895,50 @@
 		@param msg: (Message) message to send
 		@return: Message
 		"""
-		#self.log.info('SendMessage\n' + msg.pprint())
 		rawMsg = msg.toString()
+		self._log.debug(self._logMessages.MessageSend + '\n' + msg.pprint())
 		try:
 			self._socket.sendall(rawMsg)
 		except Exception, d:
+			self._log.info(self._logMessages.SocketDead)
+			self.close()
+			if self._errorHandler is not None:
+				self._errorHandler(FcpSocketError, d)
 			raise FcpSocketError(d)
-			#TODO: allow for an error handler to handle
 		return msg
 	
 #*****************************************************************************
 #
 #*****************************************************************************
 if __name__ == '__main__':
-	c = FcpClient()
+	c = FcpClient(name='test', verbosity=logging.DEBUG)
 	if c.connect():
 		job1 = JobNodeHello(c)
 		c.addJob(job1)
 
 		c.run()
 		print '---------------------------'
-		print job1.fcpResult.pprint()
+		print job1.fcpError
+		print job1.fcpResult
+		print job1.fcpTime
+		print '---------------------------'
+		
+		
+	job2 = JobTestDDA(c, os.path.dirname(__file__))
+	c.addJob(job2)
+	c.run()
+	print '---------------------------'
+	print job1.fcpError
+	print job2.fcpResult
+	print job2.fcpTime
+	print '---------------------------'
+	
+	job2 = JobGetFileInfo(c, 'CHK@sdNenKGj5mupxaSwo44jcW8dsX7vYTLww~BsRPtur0k,ZNRm9reMjtKEl9e-xFByKXbW6q4f6OQyfg~l9GRSAes,AAIC--8/snow_002%20%281%29.jpg')
+	c.addJob(job2)
+	c.run()
+	print '---------------------------'
+	print job1.fcpError
+	print job2.fcpResult
+	print job2.fcpTime
+	print '---------------------------'	
 


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.




Thread: SF.net SVN: fclient: [5] trunk/fclient/fclient_lib/fcp/fcp20.py

fclient-commit