forked from UWPCE-PythonCert/ProgrammingInPython
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathasync_timer.py
More file actions
54 lines (40 loc) · 1.31 KB
/
async_timer.py
File metadata and controls
54 lines (40 loc) · 1.31 KB
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
#!/usr/bin/env python
"""
Simple async example derived from python docs.
Will only work on Python 3.5 and above
"""
import asyncio
import time
import datetime
import random
# using "async" makes this a coroutine:
# its code can be run by the event loop
async def display_date(num):
end_time = time.time() + 10.0 # we want it to run for 10 seconds.
while True: # keep doing this until break
print("instance: {} Time: {}".format(num, datetime.datetime.now()))
if (time.time()) >= end_time:
print("instance: {} is all done".format(num))
break
# pause for a random amount of time
await asyncio.sleep(random.randint(0, 3))
def shutdown():
print("shutdown called")
# you can access the event loop this way:
loop = asyncio.get_event_loop()
loop.stop()
# You register "futures" on the loop this way:
asyncio.ensure_future(display_date(1))
asyncio.ensure_future(display_date(2))
loop = asyncio.get_event_loop()
# or add tasks to the loop like this:
loop.create_task(display_date(3))
loop.create_task(display_date(4))
for i in range(5, 20):
loop.create_task(display_date(i))
# this will shut the event loop down in 15 seconds
loop.call_later(15, shutdown)
print("about to run loop")
# this is a blocking call
loop.run_forever()
print("loop exited")