p2pool/util/variable.py

   1 import itertools
   2 import weakref
   3
   4 from twisted.internet import defer, reactor
   5 from twisted.python import failure, log
   6
   7 class Event(object):
   8     def __init__(self):
   9         self.observers = {}
  10         self.id_generator = itertools.count()
  11         self._once = None
  12         self.times = 0
  13
  14     def run_and_watch(self, func):
  15         func()
  16         return self.watch(func)
  17     def watch_weakref(self, obj, func):
  18         # func must not contain a reference to obj!
  19         watch_id = self.watch(lambda *args: func(obj_ref(), *args))
  20         obj_ref = weakref.ref(obj, lambda _: self.unwatch(watch_id))
  21     def watch(self, func):
  22         id = self.id_generator.next()
  23         self.observers[id] = func
  24         return id
  25     def unwatch(self, id):
  26         self.observers.pop(id)
  27
  28     @property
  29     def once(self):
  30         res = self._once
  31         if res is None:
  32             res = self._once = Event()
  33         return res
  34
  35     def happened(self, *event):
  36         self.times += 1
  37
  38         once, self._once = self._once, None
  39
  40         for id, func in sorted(self.observers.iteritems()):
  41             try:
  42                 func(*event)
  43             except:
  44                 log.err(None, "Error while processing Event callbacks:")
  45
  46         if once is not None:
  47             once.happened(*event)
  48
  49     def get_deferred(self, timeout=None):
  50         once = self.once
  51         df = defer.Deferred()
  52         id1 = once.watch(lambda *event: df.callback(event))
  53         if timeout is not None:
  54             def do_timeout():
  55                 df.errback(failure.Failure(defer.TimeoutError('in Event.get_deferred')))
  56                 once.unwatch(id1)
  57                 once.unwatch(x)
  58             delay = reactor.callLater(timeout, do_timeout)
  59             x = once.watch(lambda *event: delay.cancel())
  60         return df
  61
  62 class Variable(object):
  63     def __init__(self, value):
  64         self.value = value
  65         self.changed = Event()
  66         self.transitioned = Event()
  67
  68     def set(self, value):
  69         if value == self.value:
  70             return
  71
  72         oldvalue = self.value
  73         self.value = value
  74         self.changed.happened(value)
  75         self.transitioned.happened(oldvalue, value)
  76
  77     @defer.inlineCallbacks
  78     def get_when_satisfies(self, func):
  79         while True:
  80             if func(self.value):
  81                 defer.returnValue(self.value)
  82             yield self.changed.once.get_deferred()
  83
  84     def get_not_none(self):
  85         return self.get_when_satisfies(lambda val: val is not None)