Package Products :: Package ZenUtils :: Module Driver |
[hide private]
[frames] | no frames]

# Source Code for Module Products.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