-
Notifications
You must be signed in to change notification settings - Fork 52
/
engines.py
74 lines (56 loc) · 1.47 KB
/
engines.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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import asyncio
from functools import partial
import attr
from tornado.ioloop import IOLoop
from arsenic.engines.aiohttp import Aiohttp
from arsenic.engines.tornado import Tornado
@attr.s
class LoopContext:
loop = attr.ib()
run = attr.ib()
stop = attr.ib()
def start_aiohttp_loop():
policy = asyncio.get_event_loop_policy()
old_loop = policy.get_event_loop()
loop = policy.new_event_loop()
policy.set_event_loop(loop)
loop = asyncio.get_event_loop_policy().new_event_loop()
def run(func, *args, **kwargs):
return loop.run_until_complete(asyncio.wait_for(func(*args, **kwargs), 15))
def stop():
policy.set_event_loop(old_loop)
return LoopContext(
loop,
run,
stop
)
def start_tornado_loop():
loop = IOLoop(make_current=True)
def run(func, *args, **kwargs):
return loop.run_sync(partial(func, *args, **kwargs), timeout=15)
def stop():
loop.clear_current()
if not IOLoop.initialized() or loop is not IOLoop.instance():
loop.close(all_fds=True)
return LoopContext(
loop,
run,
stop
)
@attr.s
class EngineContext:
start_loop = attr.ib()
engine = attr.ib()
name = attr.ib()
ENGINE_CONTEXTS = [
EngineContext(
start_loop=start_aiohttp_loop,
engine=Aiohttp,
name='aiohttp'
),
EngineContext(
start_loop=start_tornado_loop,
engine=Tornado,
name='tornado'
)
]