43965d5289
We defer starting a new timer until we've completely emptied the queue, since we know we'll get to any new events added inside one of our callbacks. The old design in Foolscap (which copied the list, cleared the original, then fired everything in the copy) didn't look at these new events. OTOH, this pop(0)-until-empty approach makes it easier to get into an infinite loop (any callback which queues a new callback will get priority over anything else). But the code is simpler.
51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
# inspired-by/adapted-from Foolscap's eventual.py, which Glyph wrote for me
|
|
# years ago.
|
|
|
|
from twisted.internet.defer import Deferred
|
|
from twisted.internet.interfaces import IReactorTime
|
|
from twisted.python import log
|
|
|
|
class EventualQueue(object):
|
|
def __init__(self, clock):
|
|
# pass clock=reactor unless you're testing
|
|
self._clock = IReactorTime(clock)
|
|
self._calls = []
|
|
self._flush_d = None
|
|
self._timer = None
|
|
|
|
def eventually(self, f, *args, **kwargs):
|
|
self._calls.append( (f, args, kwargs) )
|
|
if not self._timer:
|
|
self._timer = self._clock.callLater(0, self._turn)
|
|
|
|
def fire_eventually(self, value=None):
|
|
d = Deferred()
|
|
self.eventually(d.callback, value)
|
|
return d
|
|
|
|
def _turn(self):
|
|
while self._calls:
|
|
(f, args, kwargs) = self._calls.pop(0)
|
|
try:
|
|
f(*args, **kwargs)
|
|
except:
|
|
log.err()
|
|
self._timer = None
|
|
d, self._flush_d = self._flush_d, None
|
|
if d:
|
|
d.callback(None)
|
|
|
|
def flush_sync(self):
|
|
# if you have control over the Clock, this will synchronously flush the
|
|
# queue
|
|
assert self._clock.advance, "needs clock=twisted.internet.task.Clock()"
|
|
while self._calls:
|
|
self._clock.advance(0)
|
|
|
|
def flush(self):
|
|
# this is for unit tests, not application code
|
|
assert not self._flush_d, "only one flush at a time"
|
|
self._flush_d = Deferred()
|
|
self.eventually(lambda: None)
|
|
return self._flush_d
|