Source Code for Module ZenUtils.Driver

  1  ########################################################################### 
  2  # 
  3  # This program is part of Zenoss Core, an open source monitoring platform. 
  4  # Copyright (C) 2007, Zenoss Inc. 
  5  # 
  6  # This program is free software; you can redistribute it and/or modify it 
  7  # under the terms of the GNU General Public License version 2 as published by 
  8  # the Free Software Foundation. 
  9  # 
 10  # For complete information please visit: http://www.zenoss.com/oss/ 
 11  # 
 12  ########################################################################### 
 13   
 14  __doc__='''Driver.py 
 15   
 16  Run generators that produce Deferreds. 
 17   
 18  twisted.flow has something like this, except I cannot understand it. 
 19   
 20  $Id$ 
 21  ''' 
 22   
 23  __version__ = "$Revision$"[11:-2] 
 24   
 25   
 26  from twisted.internet import defer, reactor 
 27  from twisted.python import failure 
 28   
 29 -class ShortCircuit: 
 30 -    def __init__(self, value): 
 31          self.value = value 
 32   
 33 -class Driver: 
 34      "Walk an iterable that returns a deferred." 
 35       
 36 -    def __init__(self): 
 37          self.defer = defer.Deferred() 
 38          self.result = None 
 39   
 40 -    def drive(self, iterable): 
 41          "Call the iterable and set up callbacks to finish" 
 42          self.iter = iterable 
 43          self._next() 
 44          return self.defer 
 45   
 46 -    def _next(self): 
 47          "Move on to the next iterable value" 
 48          try: 
 49              self.iter.next().addBoth(self._finish) 
 50          except StopIteration: 
 51              self.defer.callback(self.result) 
 52          except ShortCircuit, ex: 
 53              self.result = ex.value 
 54              self.defer.callback(self.result) 
 55          except Exception, ex: 
 56              self.defer.errback(failure.Failure(ex)) 
 57   
 58 -    def result(self): 
 59          "Provide the result of the iterable as a value or exception" 
 60          ex = self.result 
 61          if isinstance(self.result, failure.Failure): 
 62              ex = self.result.value 
 63          if isinstance(ex, Exception): 
 64              raise ex 
 65          return self.result 
 66      next = result                       # historical name for result 
 67   
 68 -    def finish(self, result): 
 69          raise ShortCircuit(result) 
 70   
 71 -    def _finish(self, result): 
 72          "Store the result of the last deferred for use in next()" 
 73          self.result = result 
 74          # prevent endless recursion 
 75          reactor.callLater(0, self._next) 
 76   
 77 -def drive(callable): 
 78      '''Typical use of Driver class: 
 79   
 80      def walk(driver): 
 81          yield thing1() 
 82          a = driver.next() 
 83          print "Thing 1 is",  
 84          yield thing2() 
 85          b = driver.next() 
 86          print "Thing 2 is",  
 87          driver.finish(a + b) 
 88   
 89      drive(walk) 
 90   
 91      ''' 
 92      d = Driver() 
 93      return d.drive(callable(d)) 
 94   
 95   
 96 -def driveLater(secs, callable): 
 97      "Drive the callable at a later time" 
 98      d = defer.Deferred() 
 99      def driveAgain(): 
100          drive(callable).chainDeferred(d) 
101      reactor.callLater(secs, driveAgain) 
102      return d 
103   
104   
105 -def test(): 
106      lst = [] 
107      def loop(d): 
108          for i in range(10): 
109              yield defer.succeed(i) 
110              lst.append(d.next()) 
111      def final(v): 
112          assert lst[-1] == v 
113          assert lst == range(10) 
114          def unloop(d): 
115              yield defer.fail(ZeroDivisionError('hahaha')) 
116              d.next() 
117          def checkError(err): 
118              assert isinstance(err.value, ZeroDivisionError) 
119              reactor.stop() 
120          drive(unloop).addErrback(checkError) 
121      drive(loop).addCallback(final) 
122           
123   
124  if __name__ == '__main__': 
125      test() 
126      reactor.run() 
127