p2pool/util/variable.py

   1 import itertools
   2
   3 from twisted.internet import defer, reactor
   4 from twisted.python import failure, log
   5
   6 class Event(object):
   7     def __init__(self):
   8         self.observers = {}
   9         self.id_generator = itertools.count()
  10         self._once = None
  11
  12     def watch(self, func):
  13         id = self.id_generator.next()
  14         self.observers[id] = func
  15         return id
  16     def unwatch(self, id):
  17         self.observers.pop(id)
  18
  19     @property
  20     def once(self):
  21         res = self._once
  22         if res is None:
  23             res = self._once = Event()
  24         return res
  25
  26     def happened(self, *event):
  27         once, self._once = self._once, None
  28
  29         for id, func in sorted(self.observers.iteritems()):
  30             try:
  31                 func(*event)
  32             except:
  33                 log.err(None, "Error while processing Event callbacks:")
  34
  35         if once is not None:
  36             once.happened(*event)
  37
  38     def get_deferred(self, timeout=None):
  39         once = self.once
  40         df = defer.Deferred()
  41         id1 = once.watch(lambda *event: df.callback(event))
  42         if timeout is not None:
  43             def do_timeout():
  44                 df.errback(failure.Failure(defer.TimeoutError('in Event.get_deferred')))
  45                 once.unwatch(id1)
  46                 once.unwatch(x)
  47             delay = reactor.callLater(timeout, do_timeout)
  48             x = once.watch(lambda *event: delay.cancel())
  49         return df
  50
  51 class Variable(object):
  52     def __init__(self, value):
  53         self.value = value
  54         self.changed = Event()
  55
  56     def set(self, value):
  57         if value == self.value:
  58             return
  59
  60         self.value = value
  61         self.changed.happened(value)
  62
  63     def get_not_none(self):
  64         if self.value is not None:
  65             return defer.succeed(self.value)
  66         else:
  67             df = defer.Deferred()
  68             self.changed.once.watch(df.callback)
  69             return df