Source Code for Module brisa.core.threaded_call

  1  # Licensed under the MIT license 
  2  # http://opensource.org/licenses/mit-license.php or see LICENSE file. 
  3  # Copyright 2007-2008 Brisa Team <brisa-develop@garage.maemo.org> 
  4   
  5  """ Runs a call asynchronously and forwards the result/error to specified 
  6  callbacks. 
  7  """ 
  8   
  9  import thread 
 10  import threading 
 11   
 12   
 13  from brisa.core import log 
 14  from brisa.utils.safe_sleep import safe_sleep 
 15   
 16   
 17 -def run_async_function(f, param_tuple=(), delay=0): 
 18      """ Calls a function passing a parameters tuple. Note that this 
 19      function returns nothing. If you want an asynchronous call with a 
 20      monitor object, see brisa.core.threaded_call.run_async_call and 
 21      brisa.core.threaded_call.ThreadedCall. 
 22   
 23      @param f: function to be called 
 24      @param param_tuple: tuple param for the function 
 25      @param delay: wait time before calling the function 
 26      """ 
 27      if delay > 0: 
 28          # If delay is valid, schedule a timer for that call 
 29          t = threading.Timer(delay, f, args=list(param_tuple)) 
 30          t.start() 
 31      else: 
 32          # Instant call 
 33          thread.start_new_thread(f, param_tuple) 
 34   
 35   
 36 -def run_async_call(function, success_callback=None, error_callback=None, 
 37                     success_callback_cargo=None, error_callback_cargo=None, 
 38                     delay=0, *args, **kwargs): 
 39      """ Interface for running an asynchronous call. 
 40   
 41      @param function: function to be called passing *args and **kwargs 
 42      @param success_callback: called in case of success, receives call result 
 43      @param error_callback: called in case of error, receives call result 
 44      @param success_callback_cargo: success callback additional parameters 
 45      @param error_callback_cargo: error callback additional parameters 
 46      @param delay: time to be wait before performing the call 
 47      @param args: arguments to the function 
 48      @param kwargs: arguments to the function 
 49   
 50      @type function: callable 
 51      @type success_callback: callable 
 52      @type error_callback: callable 
 53      @type delay: float 
 54   
 55      @return: object for monitoring the call 
 56      @rtype: ThreadedCall 
 57      """ 
 58      tcall = ThreadedCall(function, success_callback, error_callback, 
 59                                   success_callback_cargo, error_callback_cargo, 
 60                                   delay, *args, **kwargs) 
 61   
 62      # Perform the call and return the object 
 63      tcall.start() 
 64      return tcall 
 65   
 66   
 67 -class ThreadedCall(threading.Thread): 
 68      """ This class runs a call asynchronously and forwards the result/error 
 69      to specified callbacks. 
 70   
 71      One can instantiate this class directly or use the run_async_call function 
 72      located at package brisa.core.threaded_call. 
 73   
 74      @param function: function to be called passing *args and **kwargs 
 75      @param success_callback: called in case of success, receives call result 
 76      @param error_callback: called in case of error, receives call result 
 77      @param delay: time to be wait before performing the call 
 78      @param args: arguments to the function 
 79      @param kwargs: arguments to the function 
 80   
 81      @type function: callable 
 82      @type success_callback: callable 
 83      @type error_callback: callable 
 84      @type delay: float 
 85      """ 
 86   
 87 -    def __init__(self, function, success_callback=None, error_callback=None, 
 88                   success_callback_cargo=None, error_callback_cargo=None, 
 89                   delay=None, *args, **kwargs): 
 90          threading.Thread.__init__(self) 
 91          self.setDaemon(True) 
 92          self.function = function 
 93          self.args = args 
 94          self.kwargs = kwargs 
 95          self.success_callback = success_callback 
 96          self.success_callback_cargo = success_callback_cargo 
 97          self.error_callback = error_callback 
 98          self.error_callback_cargo = error_callback_cargo 
 99          self.delay = delay 
100          self.result = None 
101          self.cancelled = False 
102          self.completed_flag = False 
103          self.completed = threading.Semaphore() 
104   
105 -    def cleanup(self): 
106          """ Removes references to objects that may hurt garbage collection 
107          """ 
108          self.function = lambda: None 
109          self.args = () 
110          self.kwargs = {} 
111          self.success_callback = self.success_callback_cargo = None 
112          self.error_callback = self.error_callback_cargo = None 
113   
114 -    def is_cancelled(self): 
115          return self.cancelled 
116   
117 -    def run(self): 
118          """ Implementation of the call procedure. 
119          """ 
120          if self.delay: 
121              # This runs in a thread. We can sleep here instead using time 
122              safe_sleep(self.delay) 
123              log.debug('sleeping for %d' % self.delay) 
124   
125          if self.is_cancelled(): 
126              self.cleanup() 
127              return 
128   
129          try: 
130              log.debug('calling function') 
131              # Performing the call 
132              self.result = self.function(*self.args, **self.kwargs) 
133              log.debug('got result %s' % self.result) 
134              if self.success_callback: 
135                  log.debug('forwarding to success_callback') 
136                  self.success_callback(self.result, self.success_callback_cargo) 
137   
138              self.set_completed() 
139   
140          except Exception, e: 
141              log.debug('exception happened (%s), forwarding...' 
142                        % e.message) 
143              # Storing exception for handling 
144              self.result = e 
145   
146              if self.error_callback: 
147                  self.error_callback(self.error_callback_cargo, e) 
148   
149              self.set_completed() 
150   
151 -    def cancel(self): 
152          """ Cancel the call. 
153          """ 
154          self.cancelled = True 
155   
156 -    def stop(self): 
157          """ Prepares to stop. 
158          """ 
159          if not self.is_completed(): 
160              self.cancel() 
161          self.set_completed() 
162   
163 -    def is_completed(self): 
164          """ Returns whether the call has been completed or not. 
165          """ 
166          self.completed.acquire() 
167          res = self.completed_flag 
168          self.completed.release() 
169          return res 
170   
171 -    def set_completed(self): 
172          """ Sets the call as completed. This should not be called directly. 
173          Use stop() instead. 
174          """ 
175          self.completed.acquire() 
176          self.completed_flag = True 
177          self.completed.release() 
178          self.cleanup() 
179