Created
July 6, 2015 02:50
-
-
Save numberoverzero/1c6a8242e60873d2b801 to your computer and use it in GitHub Desktop.
Variable that can be `await`ed on to reach a certain value, without blocking an event loop
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
missing = object() | |
class Value: | |
''' | |
Simple class that enables `await value.wait_for(foo)` to wait until the | |
variable is set to the expected value, without blocking the event loop. | |
Loosely modeled after asyncio.Event and asyncio.Condition | |
Usage: | |
import asyncio | |
loop = asyncio.new_event_loop() | |
signal = Value(loop=loop, value=False) | |
async def iterate_values(values, fut): | |
for value in values: | |
signal.value = value | |
print("Set signal to {}".format(value)) | |
await asyncio.sleep(1, loop=loop) | |
fut.set_result(True) | |
async def print_when(value): | |
await signal.wait_for(value) | |
print("Signal reached desired value {}".format(value)) | |
complete = asyncio.Future(loop=loop) | |
loop.create_task(print_when("foo")) | |
loop.create_task(print_when(3)) | |
loop.create_task(iterate_values([1, 2, 3, "foo", 4], complete)) | |
loop.run_until_complete(complete) | |
''' | |
def __init__(self, loop, value=None, expected=missing): | |
self.loop = loop | |
self.watchers = set() | |
self._value = value | |
self._expected = expected | |
@property | |
def value(self): | |
return self._value | |
@value.setter | |
def value(self, v): | |
self._value = v | |
# FOrce all `wait_for` to re-check the value | |
for watcher in self.watchers: | |
watcher.set() | |
async def wait_for(self, value): | |
''' | |
Yields when the Value is set to the given value. | |
Doesn't block the event loop with a busy `while` loop. | |
''' | |
# Don't wait if we're already at the expected value | |
if value == self.value: | |
return | |
watcher = asyncio.Event(loop=self.loop) | |
self.watchers.add(watcher) | |
# We'll only re-check the condition when the value changes, | |
# yielding back to the poller when it's not equal. | |
while self.value != value: | |
await watcher.wait() | |
watcher.clear() | |
# Clean up the watcher now that we've reached the expected value | |
self.watchers.remove(watcher) | |
async def wait(self): | |
''' | |
Waits until the Value is set to the given `expected` value. | |
Raises if `expected` was not set. | |
''' | |
if self._expected is missing: | |
raise RuntimeError( | |
"Must define `expected` when intializing the Value") | |
await self.wait_for(self._expected) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment