Fix 13245:
sched.scheduler class constructor's timefunc and delayfunct parameters are now optional.
scheduler.enter and scheduler.enterabs methods gained a new kwargs parameter.
Patch contributed by Matt Mulsow.
diff --git a/Lib/sched.py b/Lib/sched.py
index a119892..6c01e69 100644
--- a/Lib/sched.py
+++ b/Lib/sched.py
@@ -28,12 +28,13 @@
# XXX instead of having to define a module or class just to hold
# XXX the global state of your particular time and delay functions.
+import time
import heapq
from collections import namedtuple
__all__ = ["scheduler"]
-class Event(namedtuple('Event', 'time, priority, action, argument')):
+class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')):
def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority)
def __ne__(s, o): return (s.time, s.priority) != (o.time, o.priority)
def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority)
@@ -42,32 +43,33 @@
def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority)
class scheduler:
- def __init__(self, timefunc, delayfunc):
+
+ def __init__(self, timefunc=time.time, delayfunc=time.sleep):
"""Initialize a new instance, passing the time and delay
functions"""
self._queue = []
self.timefunc = timefunc
self.delayfunc = delayfunc
- def enterabs(self, time, priority, action, argument):
+ def enterabs(self, time, priority, action, argument=[], kwargs={}):
"""Enter a new event in the queue at an absolute time.
Returns an ID for the event which can be used to remove it,
if necessary.
"""
- event = Event(time, priority, action, argument)
+ event = Event(time, priority, action, argument, kwargs)
heapq.heappush(self._queue, event)
return event # The ID
- def enter(self, delay, priority, action, argument):
+ def enter(self, delay, priority, action, argument=[], kwargs={}):
"""A variant that specifies the time as a relative time.
This is actually the more commonly used interface.
"""
time = self.timefunc() + delay
- return self.enterabs(time, priority, action, argument)
+ return self.enterabs(time, priority, action, argument, kwargs)
def cancel(self, event):
"""Remove an event from the queue.
@@ -111,7 +113,7 @@
timefunc = self.timefunc
pop = heapq.heappop
while q:
- time, priority, action, argument = checked_event = q[0]
+ time, priority, action, argument, kwargs = checked_event = q[0]
now = timefunc()
if now < time:
delayfunc(time - now)
@@ -120,7 +122,7 @@
# Verify that the event was not removed or altered
# by another thread after we last looked at q[0].
if event is checked_event:
- action(*argument)
+ action(*argument, **kwargs)
delayfunc(0) # Let other threads run
else:
heapq.heappush(q, event)
diff --git a/Lib/test/test_sched.py b/Lib/test/test_sched.py
index 29fd277..1af43cb 100644
--- a/Lib/test/test_sched.py
+++ b/Lib/test/test_sched.py
@@ -72,6 +72,18 @@
scheduler.run()
self.assertEqual(scheduler._queue, [])
+ def test_args_kwargs(self):
+ flag = []
+
+ def fun(*a, **b):
+ flag.append(None)
+ self.assertEqual(a, (1,2,3))
+ self.assertEqual(b, {"foo":1})
+
+ scheduler = sched.scheduler(time.time, time.sleep)
+ z = scheduler.enterabs(0.01, 1, fun, argument=(1,2,3), kwargs={"foo":1})
+ scheduler.run()
+ self.assertEqual(flag, [None])
def test_main():
support.run_unittest(TestCase)