[Twisted-Python] DelayedCall enhancements
Bob Ippolito
bob at redivi.com
Mon Dec 23 13:52:01 MST 2002
Here's a patch that adds two particularly useful methods to the
IDelayedCall interface and implementation. It's quite possible to do
it currently, but it's a kludge.
Essentially what I added were methods to delay a DelayedCall by a
configurable amount of time, and to reset a DelayedCall to happen a
configurable amount of time from now.
I also added a test case for each, and they pass. I can check it in if
someone gives me the OK.
-bob
Index: twisted/internet/interfaces.py
===================================================================
RCS file: /cvs/Twisted/twisted/internet/interfaces.py,v
retrieving revision 1.57
diff -r1.57 interfaces.py
317a318,319
> It also may be rescheduled by calling its C{delay()}
> or C{reset()} methods.
347a350,370
> def delay(self, secondsLater):
> """Delay the scheduled call.
> @param secondsLater: how many seconds from its current firing
time to delay
>
> @raises twisted.internet.error.AlreadyCalled: if the call has
already
> happened.
> @raises twisted.internet.error.AlreadyCancelled: if the call
has already
> been cancelled.
> """
>
> def reset(self, secondsFromNow):
> """Reset the scheduled call's timer.
> @param secondsLater: how many seconds from now it should
fire, equivalent
> to C{self.cancel()} and then doing
another
> C{reactor.callLater(secondsLater, ...)}
>
> @raises twisted.internet.error.AlreadyCalled: if the call has
already
> happened.
> @raises twisted.internet.error.AlreadyCancelled: if the call
has already
> been cancelled.
> """
Index: twisted/internet/base.py
===================================================================
RCS file: /cvs/Twisted/twisted/internet/base.py,v
retrieving revision 1.29
diff -r1.29 base.py
48c48
< def __init__(self, time, func, args, kw, cancel):
---
> def __init__(self, time, func, args, kw, cancel, reset):
49a50
> self.resetter = reset
61a63,79
> def reset(self, secondsFromNow):
> if self.cancelled:
> raise error.AlreadyCancelled
> elif self.called:
> raise error.AlreadyCalled
> else:
> self.time = time() + secondsFromNow
> self.resetter(self)
>
> def delay(self, secondsLater):
> if self.cancelled:
> raise error.AlreadyCancelled
> elif self.called:
> raise error.AlreadyCalled
> else:
> self.time += secondsLater
> self.resetter(self)
266c284,290
< tple = DelayedCall(time() + seconds, f, args, kw,
self._pendingTimedCalls.remove)
---
> tple = DelayedCall(time() + seconds, f, args, kw,
self._pendingTimedCalls.remove, self._resetCallLater)
> insort(self._pendingTimedCalls, tple)
> return tple
>
> def _resetCallLater(self, tple):
> assert tple in self._pendingTimedCalls
> self._pendingTimedCalls.remove(tple)
Index: twisted/test/test_internet.py
===================================================================
RCS file: /cvs/Twisted/twisted/test/test_internet.py,v
retrieving revision 1.13
diff -r1.13 test_internet.py
102a103,128
> def _resetcallback(self):
> self._resetcallbackTime = time.time()
>
> def _delaycallback(self):
> self._delaycallbackTime = time.time()
>
> def testCallLaterDelayAndReset(self):
> self._resetcallbackTime = None
> self._delaycallbackTime = None
> ireset = reactor.callLater(0.5, self._resetcallback)
> idelay = reactor.callLater(0.5, self._delaycallback)
> start = time.time()
> # chug a little before delaying
> while time.time() - start < 0.2:
> reactor.iterate(0.01)
> ireset.reset(0.3)
> idelay.delay(0.3)
> # both should be called sometime during this
> while time.time() - start < 0.9:
> reactor.iterate(0.01)
> self.assert_(0 < self._resetcallbackTime - start - 0.5 < 0.2)
> self.assert_(0 < self._delaycallbackTime - start - 0.8 < 0.2)
>
> del self._resetcallbackTime
> del self._delaycallbackTime
>
More information about the Twisted-Python
mailing list