SF.net SVN: fclient: [10] trunk/fclient/fclient_lib/fcp/fcp2_0.py

[<jU...@us...>]

Revision: 10
          http://fclient.svn.sourceforge.net/fclient/?rev=10&view=rev
Author:   jUrner
Date:     2007-10-22 02:02:04 -0700 (Mon, 22 Oct 2007)

Log Message:
-----------
continued working on protocol implementation

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

Modified: trunk/fclient/fclient_lib/fcp/fcp2_0.py
===================================================================
--- trunk/fclient/fclient_lib/fcp/fcp2_0.py	2007-10-20 13:37:51 UTC (rev 9)
+++ trunk/fclient/fclient_lib/fcp/fcp2_0.py	2007-10-22 09:02:04 UTC (rev 10)
@@ -16,10 +16,10 @@
 #**************************************************************
 NameClient = 'Fcp20Client'
 DefaultFcpHost = os.environ.get('FCP_HOST', '127.0.0.1').strip()
-DefaultFcpPort = 9481
 try:
 	DefaultFcpPort = int(os.environ.get('FCP_PORT', '').strip())
-except: pass
+except ValueError:
+	DefaultFcpPort = 9481
 SocketTimeout = 0.1
 
 class FixedJobIdentifiers:
@@ -33,13 +33,13 @@
 class Priorities:
 	"""All priorities supported by the client"""
 	
-	Maximum = 0
-	Interactive = 1
-	SemiInteractive = 2
-	Updatable = 3
-	Bulk = 4
-	Prefetch = 5
-	Minimum = 6
+	Maximum = '0'
+	Interactive = '1'
+	SemiInteractive = '2'
+	Updatable = '3'
+	Bulk = '4'
+	Prefetch = '5'
+	Minimum = '6'
 	
 	PriorityMin = Minimum
 	PriorityDefault = Bulk
@@ -471,42 +471,40 @@
 class JobBase(object):
 	"""Base class for jobs"""
 	
-	_fcp_auto_remove_ = True
-	
+		
 	def __init__(self, fcpClient, identifier, message):
 		"""
 		@param fcpClient: FcpClient() instance
 		@param identifier: (str) identifier of the job
 		@param message: (Message) to send to the node whne the job ist started
-		@ivar fcpClient: FcpClient() instance
-		@ivar fcpError: holding the error message if an error was encountered while running 
-		the job, None otherwise
-		@ivar fcpIdentifier: identifier of the job
-		@ivar fcpMessage: initial message send to the node
-		@ivar fcpResult: if no error was encountered, holding the result of the job when complete		
-		@ivar fcpTimeStart:  time the job was started
-		@ivar fcpTimeStop:  time the job was stopped
+		@ivar jobClient: FcpClient() instance of the job
+		@ivar jobIdentifier: identifier of the job
+		@ivar jobMessage: message to be send to the node
+		@ivar jobResult: if no error was encountered, holding the result of the job when complete		
+		@ivar jobTimeStart:  time the job was started
+		@ivar jobTimeStop:  time the job was stopped
 		"""
-		self.fcpClient = fcpClient					# FcpClient() instance the job belongs to
-		self.fcpIdentifier = identifier				# job identifier
-		self.fcpMessage = message				# message send to node
-		self.fcpResult = None						# job result (bool error, Message msg)
-		self.fcpTimeStart = 0							# time the job was started
-		self.fcpTimeStop = 0							# time the job was stopped
+		self.jobClient = fcpClient
+		self.jobIdentifier = identifier
+		self.jobMessage = message
+		self.jobResult = None
+		self.jobTimeStart = 0
+		self.jobTimeStop = 0
+		
 			
 	def handleStart(self):
 		"""Starts the job"""
-		self.fcpResult = None
-		self.fcpTimeStart = time.time()
-		self.fcpClient.sendMessageEx(self.fcpMessage)
+		self.jobResult = None
+		self.jobTimeStart = time.time()
+		self.jobClient.sendMessageEx(self.jobMessage)
 			
 	def handleStop(self, flagError, msg):
 		"""Called on job completion to stop the job
 		@param flagError: True if an error was encountered, False otherwise
 		@param msg: (Message) to pass to the job
 		"""
-		self.fcpTimeStop = time.time()
-		self.fcpResult = (flagError, msg)
+		self.jobTimeStop = time.time()
+		self.jobResult = (flagError, msg)
 
 
 class JobClientHello(JobBase):
@@ -516,8 +514,7 @@
 	goes well, you will get a NodeHello in response.
 	"""
 	
-	_fcp_auto_remove_ = True
-	
+		
 	def __init__(self, fcpClient, name=None, expectedVersion='2.0'):
 		"""
 		@param name: (str) connection name or None, to use an arbitrary name
@@ -538,8 +535,7 @@
 	"""Lists all known peers of the node 
 	"""
 	
-	_fcp_auto_remove_ = True
-	
+		
 	def __init__(self, fcpClient, withMetaData=False, withVolantile=False):
 		message = Message(
 				Message.ListPeers, 
@@ -561,13 +557,10 @@
 	
 	On completion fcpResult will hold a tuple((str) content-type, (str) size) of the file.
 	Note, that both members may be '' (empty string) 
-	
 	"""
 	
-	_fcp_auto_remove_ = False
-	
-	
-	# idea is to provoke a GetFailed message and take mimetype and size from it
+		
+	# idea is to provoke a GetFailed message and take mimetype and size from 'GetFailed'
 	def __init__(self, fcpClient, uri, **params):
 		"""
 		@param fcpClient: FcpClient() instance
@@ -584,7 +577,9 @@
 				Message.ClientGet, 
 				Identifier=identifier, 
 				URI=uri,
-				MaxSize='0',
+				# suggested by Mathew Toseland to use about 32k for mimeType requests
+				# basic sizes of keys are: 1k for SSks and 32k for CHKs
+				MaxSize='32000',
 				ReturnType='none',
 				Verbosity='1',
 				**params
@@ -592,34 +587,77 @@
 		JobBase.__init__(self, fcpClient, identifier, message)
 		
 		
+	def getPrority(self):
+		return self.jobMessage.get('PriorityClass', Priorities.PriorityDefault)
+		
+	
+	def setPriority(self, priority):
+		if not priority in Priorities:
+			raise ValueError('Invalid priority: %r' % priority)
+		self.jobClient.sendMessage(
+				Message.ModifyPersistentRequest,
+				Identifier=self.jobIdentifier,
+				Global=fcpBool(False),
+				PriorityClass=priority,
+				)
+		# not shure if the response arrives in any case, so set it here
+		self.jobMessage['PriorityClass'] = priority
+	
+	
+	def stopRequest(self):
+		self.jobClient.sendMessage(
+			Message.RemovePersistentRequest,
+			Global=fcpBool(False),
+			Identifier=self.jobIdentifier,
+			)
+		
+	
+	def handlePriorityChanged(self):
+		self.jobMessage['PriorityClass'] = priority
+	
 	def handleProgress(self, msg):
 		"""Handles the next progress made. Overwrite to process.
 		"""
+		
+	def handleRequestRemoved(self, msg):
+		if self.jobClient.jobIsRunning(self.jobIdentifier):
+			self.jobClient.jobStop(self.jobIdentifier)
 	
+	
 	def handleStop(self, flagError, msg):
+		"""called when a job is finished
+		@note: jobResult may contain an 'IdentifierCollision' message. Make shure 
+		to handle this appropriately
+		"""
 		JobBase.handleStop(self,flagError, msg)
 		if msg.name == Message.DataFound:
-			self.fcpResult = (
+			self.jobResult = (
 								msg.get('Metadata.ContentType', ''), 
 								msg.get('DataLength', '')
 							)
 		elif msg.name == Message.GetFailed:
 			if msg['Code'] == FetchError.TooBig:
-				self.fcpResult = (False, msg)
-				self.fcpResult = (
+				self.jobResult = (False, msg)
+				self.jobResult = (
 								msg.get('ExpectedMetadata.ContentType', ''), 
 								msg.get('ExpectedDataLength', '')
 							)
-		
+		# not shure when removing the request is actually required
+		# so send it anyway
+		self.jobClient.sendMessage(
+			Message.RemovePersistentRequest,
+			Global=fcpBool(False),
+			Identifier=self.jobIdentifier,
+			)
+				
 
-		
+
 #TODO: handle case where directories are registered multiple times
 class JobTestDDA(JobBase):
 	"""Tests a directory for read / write accesss
 	"""
 		
-	_fcp_auto_remove_ = False
-	
+		
 	def __init__(self, fcpClient, directory, read=True, write=True):
 		if not os.path.isdir(directory):
 			raise ValueError('No such directory: %r' % directory)
@@ -631,7 +669,7 @@
 				WantWriteDirectory=fcpBool(write),
 				)
 		JobBase.__init__(self, fcpClient, directory, message)
-		self.fcpTmpFile = None
+		self.jobTmpFile = None
 		
 		
 	def handleTestDDAReply(self, msg):
@@ -644,14 +682,14 @@
 				if os.path.isfile(fpathWrite):
 					os.remove(fpathWrite)
 			else:
-				self.fcpTmpFile = fpathWrite
+				self.jobTmpFile = fpathWrite
 									
 		if fpathRead is not None:
 			readContent = saveReadFile(fpathRead)
 			if readContent is None:
 				readContent = ''
 			
-		self.fcpClient.sendMessage(
+		self.jobClient.sendMessage(
 				Message.TestDDAResponse,
 				Directory=msg['Directory'],
 				ReadContent=readContent,
@@ -660,8 +698,8 @@
 	
 	def handleStop(self, flagError, msg):
 		JobBase.handleStop(self, flagError, msg)
-		saveRemoveFile(self.fcpTmpFile)
-		self.fcpTmpFile = None
+		saveRemoveFile(self.jobTmpFile)
+		self.jobTmpFile = None
 
 #**************************************************************************
 # fcp client
@@ -682,6 +720,7 @@
 	JobStop  = 'Stopping job: '
 	JobsCompleted = 'All jobs completed'
 	
+	
 	KeyboardInterrupt = 'Keyboard interrupt'
 	SocketDead = 'Socket is dead'
 	
@@ -710,10 +749,10 @@
 				
 		self._isConnected = False
 		self._jobs = {
-			'all': {},
-			'pending': [],			# ???
-			'running': [],
-			'complete': [],		# ???
+			#TODO: check if JobList is still required
+			'JobMapping': {},
+			'JobList': [],
+			'RegisteredDirectories': [],
 			}
 		self._errorHandler = errorHandler
 		self._log = logging.getLogger(name)
@@ -734,7 +773,7 @@
 			self._socket = None
 	
 	
-	#TODO: an iterator would be nice for Guis, to enshure they stay responsitive in the call
+	#TODO: an iterator would be nice to enshure Guis stay responsitive in the call
 	def connect(self, host=DefaultFcpHost, port=DefaultFcpPort, repeat=20, timeout=0.5):
 		"""Establishes the connection to a freenet node
 		@param host: (str) host of th node
@@ -767,7 +806,7 @@
 				#           as expected the socket simply breaks. So take it over. 
 				job = JobClientHello(self)
 				self.jobAdd(job, synchron=True)
-				error, msg = job.fcpResult
+				error, msg = job.jobResult
 				assert not error, 'Error should have been caught by handleMessage()'
 				return msg
 				
@@ -782,9 +821,13 @@
 	
 
 	def handleMessage(self, msg):
-		"""Handles the next message from the freenet node
-		@param msg: Message() to handle
+		"""Handles a message from the freenet node
+		@param msg: (Message) to handle
+		@return: True if the message was handled, False otherwise
 		"""	
+		
+		
+		#if msg.name != 'USocketTimeOut':
 		self._log.debug(self._logMessages.MessageReceived + msg.pprint())
 				
 		if msg.name == Message.NodeHello:
@@ -803,30 +846,57 @@
 				#TODO: inform caller
 				raise ProtocolError(msg)
 			else:
-				self.jobStop(identifier, msg, flagError=True)	
+				return self.jobStop(identifier, msg, flagError=True)	
 				
-		elif msg.name == Message.Peer:
-			self.jobNotify(FixedJobIdentifiers.ListPeers, 'handlePeer', msg)
+		if msg.name == Message.DataFound:
+			return self.jobStop(msg['Identifier'], msg, flagError=True)
 		
 		elif msg.name == Message.EndListPeers:
-			self.jobStop(FixedJobIdentifiers.ListPeers, msg)
+			return self.jobStop(FixedJobIdentifiers.ListPeers, msg)
 			
 		elif msg.name == Message.GetFailed:
-			self.jobStop(msg['Identifier'], msg, flagError=True)
+			return self.jobStop(msg['Identifier'], msg, flagError=True)
 					
+		elif msg.name == Message.PersistentRequestRemoved:
+			return self.jobNotify(FixedJobIdentifiers.ListPeers, 'RequestRemoved', msg)
+		
 		elif msg.name == Message.SimpleProgress:
-			self.jobNotify(msg['Identifier'], 'handleProgress', msg)
+			return self.jobNotify(msg['Identifier'], 'handleProgress', msg)
 				
+		elif msg.name == Message.TestDDAComplete:
+			return self.jobStop(msg['Directory'], msg)
+				
 		elif msg.name == Message.TestDDAReply:
-			self.jobNotify(msg['Directory'], 'handleTestDDAReply', msg)
+			return self.jobNotify(msg['Directory'], 'handleTestDDAReply', msg)
 		
-		elif msg.name == Message.TestDDAComplete:
-			self.jobStop(msg['Directory'], msg)
+		elif msg.name == Message.IdentifierCollision:
+			# identifier is unique to us, but for some reason not unique to the node
+			#TODO: maybe for future implementations find a way to automise 
+			# reinsertion and a notification via a handleJobChanged() for the 
+			# caller to overwrite
+			return self.jobStop(msg['Identifier'], msg, flagError=True)
 			
-		elif msg.name == Message.IdentifierCollision:
-			pass
+		elif msg.name == Message.Peer:
+			return self.jobNotify(FixedJobIdentifiers.ListPeers, 'handlePeer', msg)
+			
+		return False
 	
 	
+	#########################################################
+	## jobs
+	#########################################################
+	def hasJobsRunning(self):
+		"""Checks if the client has running jobs
+		@return: (bool) True if so, False otherwise
+		"""
+		self._lock.acquire(True)			
+		try:
+			result = bool(self._jobs['JobList'])
+		finally:
+			self._lock.release()
+		return result
+	
+	
 	def jobAdd(self, job, synchron=False):
 		"""Adds a job to the client
 		@param job: (Job*) job to add
@@ -834,35 +904,62 @@
 		"""
 		self._lock.acquire(True)			
 		try:
-			if job.fcpIdentifier in self._jobs['all']:
-				raise ValueError('Duplicate job: %r' % job.identifier)
-			self._jobs['all'][job.fcpIdentifier] = job
-			self._jobs['running'].append(job)
+			if job.jobIdentifier in self._jobs['JobMapping']:
+				raise ValueError('Duplicate job: %r' % job.jobIdentifier)
+			self._jobs['JobMapping'][job.jobIdentifier] = job
+			self._jobs['JobList'].append(job)
 		finally:
 			self._lock.release()
 		
-		self._log.info(self._logMessages.JobStart + job.fcpMessage.name)
+		self._log.info(self._logMessages.JobStart + job.jobMessage.name)
 		job.handleStart()
 		if synchron:
-			while job.fcpResult is None:
+			while job.jobResult is None:
 				self.next()
 		
 	
+	def jobGet(self, identifier):
+		"""Returns a job given its identifier
+		@param identifier: identifier of the job
+		@return: (Job*) instance or None, if no corrosponding job was found
+		"""
+		self._lock.acquire(True)
+		try:
+			result = self._jobs['JobMapping'].get(identifier, None)
+		finally:
+			self._lock.release()
+		return result
 	
+	
+	def jobIsRunning(self, identifier):
+		"""Checks if a job is running
+		@param identifier: identifier of the job
+		@return: True if so, False otherwise
+		"""
+		self._lock.acquire(True)
+		try:
+			result = identifier in self._jobs['JobMapping']
+		finally:
+			self._lock.release()
+		return result
+		
+	
 	def jobNotify(self, identifier, handler, msg):
 		"""Notifies a job about an event while it is running
 		@param identifier: identifier of the job to notify
 		@param handler: (str) method of the job to call to handle the notification
 		@param msg: Message() to pass to the job
+		@return: True if a job was found to be notified, False otherwise
 		"""
 		self._lock.acquire(True)
 		try:
-			job = self._jobs['all'].get(identifier, None)
+			job = self._jobs['JobMapping'].get(identifier, None)
 		finally:
 			self._lock.release()
 		if job is None:
-			raise ValueError('No such job: %r' % identifier)
+			return False
 		getattr(job, handler)(msg)
+		return True
 	
 	
 	#TODO: quite unclear when to remove a job
@@ -871,25 +968,23 @@
 		@param identifier: identifier of the job to stop
 		@param msg: Message() to pass to the job as result
 		@param flagError: set to True to indicate unsuccessful completion of the job, True otherwisse
+		@return: True if a job was found to be stopped, False otherwise
 		"""
 		self._lock.acquire(True)
 		try:
-			job = self._jobs['all'].get(identifier, None)
+			job = self._jobs['JobMapping'].get(identifier, None)
 			if job is not None:
-				self._jobs['running'].remove(job)
-				if job._fcp_auto_remove_:
-					del self._jobs['all'][identifier]
-				else:
-					self._jobs['complete'].append(job)
+				self._jobs['JobList'].remove(job)
+				del self._jobs['JobMapping'][identifier]
 		finally:
 			self._lock.release()
-			
 		if job is None:
-			raise ValueError('No such job: %r' % identifier)
-		self._log.info(self._logMessages.JobStop + job.fcpMessage.name)
+			return False
+		self._log.info(self._logMessages.JobStop + job.jobMessage.name)
 		job.handleStop(flagError, msg)
+		return True
+		
 	
-	
 	#TODO: some info when all jobs are completed
 	def next(self):
 		"""Pumps the next message waiting
@@ -963,21 +1058,8 @@
 		"""Runs the client untill all jobs passed to it are completed
 		@note: use KeyboardInterrupt to stop prematurely
 		"""
-		
-		# TODO: 
-		#    x. push pending jobs
 		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()
-				
+			while self.hasJobsRunning():
 				self.next()
 		except KeyboardInterrupt:
 			self._log(self._logMessages.KeyboardInterrupt)
@@ -1026,6 +1108,16 @@
 		""""""
 		self._log.setLevel(verbosity)	
 
+		
+	########################################################
+	##
+	########################################################
+	def getFileInfo(self, job):
+		pass
+		
+	def getFile(self, job):
+		pass
+
 #*****************************************************************************
 #
 #*****************************************************************************
@@ -1039,8 +1131,7 @@
 
 			c.run()
 			print '---------------------------'
-			print job1.fcpError
-			print job1.fcpResult
+			print job1.jobResult
 			print '---------------------------'
 		# should raise
 		#foo()
@@ -1052,7 +1143,7 @@
 			c.jobAdd(job2)
 			c.run()
 			print '---------------------------'
-			print job2.fcpResult
+			print job2.jobResult
 			print '---------------------------'
 		#foo()
 	
@@ -1061,7 +1152,7 @@
 		c.jobAdd(job2)
 		c.run()
 		print '---------------------------'
-		print job2.fcpResult
+		print job2.jobResult
 		print '---------------------------'	
 	#foo()
 	
@@ -1071,6 +1162,6 @@
 		c.jobAdd(job2)
 		c.run()
 		print '---------------------------'
-		print job2.fcpResult
+		print job2.jobResult
 		print '---------------------------'	
 	#foo()


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