How to wait for an object to change.
Inside the handler, async
I want to wait for the task state to change. For now, I just check the state in an infinite loop and wait. Here's an example, a function wait_until_done
:
import asyncio
class LongTask:
state = 'PENDING'
my_task = LongTask()
def done():
my_task.state = 'DONE'
async def wait_until_done():
while True:
if my_task.state == 'PENDING':
await asyncio.sleep(2)
else:
break
print("Finally, the task is done")
def main(loop, *args, **kwargs):
asyncio.ensure_future(wait_until_done())
loop.call_later(delay=5, callback=done)
loop = asyncio.get_event_loop()
main(loop)
loop.run_forever()
Is there a better way to do this?
source to share
Just to avoid confusion: I guess you are not talking about asyncio.Task , but some kind of state variable instead ?? //>
In this case, you are Future and synchronization primitives , which allows you to wait for some thing to change asynchronously.
If you need to switch between two states, asyncio.Event is probably what you want. Here's a small example:
import asyncio
my_task = asyncio.Event()
def done():
my_task.set()
async def wait_until_done():
await my_task.wait() # await until event would be .set()
print("Finally, the task is done")
async def main():
loop.call_later(delay=5, callback=done)
await wait_until_done()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
Upd:
A more complex example that supports the interface LongTask
:
import asyncio
class LongTask:
_event = asyncio.Event()
@property
def state(self):
return 'PENDING' if not type(self)._event.is_set() else 'DONE'
@state.setter
def state(self, val):
if val == 'PENDING':
type(self)._event.clear()
elif val == 'DONE':
type(self)._event.set()
else:
raise ValueError('Bad state value.')
async def is_done(self):
return (await type(self)._event.wait())
my_task = LongTask()
def done():
my_task.state = 'DONE'
async def wait_until_done():
await my_task.is_done()
print("Finally, the task is done")
async def main():
loop.call_later(delay=5, callback=done)
await wait_until_done()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
source to share
Isn't this a sample Observer design pattern? I think the Observer pattern can solve this problem. In this design pattern, we don't have to iterate endlessly to detect a change, but the task itself will tell you if it has made any changes. A simple implementation could be:
class ObservableTask:
def __init__(self):
self.subscribers = set()
self.state = 'PENDING'
def subscribe(self, who):
self.subscribers.add(who)
def unsubscribe(self, who):
self.subscribers.discard(who)
def dispatch(self, message):
for subscriber in self.subscribers:
subscriber.update(message)
def random_change(self):
for count in range(1, 10):
if count % 5 == 0:
print('Inside task:\tDivisible by 5')
self.state = 'DONE'
self.dispatch('state: DONE')
class Observer:
def __init__(self):
pass
def update(self, message):
print('Task is changed!\t' + message)
# Test:
task = ObservableTask()
observer = Observer()
task.subscribe(observer)
task.random_change()
Output:
Inside task: Divisible by 5
Task is changed! state: DONE
source to share