-
Notifications
You must be signed in to change notification settings - Fork 0
/
eventloop.py
54 lines (40 loc) · 1.4 KB
/
eventloop.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# Written by s4y @ https://github.com/s4y/runloop.py -- license unknown
try:
from queue import Queue
except ImportError:
from Queue import Queue
from collections import namedtuple
Invocation = namedtuple('Invocation', ('fn', 'args', 'kwargs'))
class RunLoop:
def __init__(self):
self.queue = Queue()
self.running = False
def add(self, fn, args=(), kwargs={}):
self.queue.put(Invocation(fn, args, kwargs))
def every(self, fn, interval, args=(), kwargs={}):
import threading
invocation = Invocation(fn, args, kwargs)
def do_repeat():
event = threading.Event()
while True:
event.wait(interval)
self.queue.put(invocation)
thread = threading.Thread(target=do_repeat)
thread.daemon = True
thread.start()
def run(self):
self.running = True
while self.running:
inv = self.queue.get()
inv.fn(*inv.args, **inv.kwargs)
def stop(self):
self.running = False
def nop(): # no operation (like assembler instruction) does nothing
pass
self.add(nop) # to make queue.get() call in run method quit and let while-loop stop
def onLoop(self, fn):
from functools import wraps
@wraps(fn)
def wrapper(*args, **kwargs):
self.add(fn, args, kwargs)
return wrapper