from __future__ import annotations import typing as T from collections import deque from random import randint Result = T.Any Burger = Result Coroutine = T.Callable[[], 'Future'] class Future: def __init__(self) -> None: self.done = False self.coroutine = None self.result = None def set_coroutine(self, coroutine: Coroutine) -> None: self.coroutine = coroutine def set_result(self, result: Result) -> None: self.done = True self.result = result def __iter__(self) -> Future: return self def __next__(self) -> Result: if not self.done: raise StopIteration return self.result class EventLoop: def __init__(self) -> None: self.tasks: T.Deque[Coroutine] = deque() def add_coroutine(self, coroutine: Coroutine) -> None: self.tasks.append(coroutine) def run_coroutine(self, task: T.Callable) -> None: future = task() future.set_coroutine(task) try: next(future) if not future.done: future.set_coroutine(task) self.add_coroutine(task) except StopIteration: return def run_forever(self) -> None: while self.tasks: self.run_coroutine(self.tasks.popleft()) def cook(on_done: T.Callable[[Burger], None]) -> None: burger: str = f"Burger #{randint(1, 10)}" print(f"{burger} is cooked!") on_done(burger) def cashier(burger: Burger, on_done: T.Callable[[Burger], None]) -> None: print("Burger is ready for pick up!") on_done(burger) def order_burger() -> Future: order = Future() def on_cook_done(burger: Burger) -> None: cashier(burger, on_cashier_done) def on_cashier_done(burger: Burger) -> None: print(f"{burger}? That's me! Mmmmmm!") order.set_result(burger) cook(on_cook_done) return order if __name__ == "__main__": event_loop = EventLoop() event_loop.add_coroutine(order_burger) event_loop.run_forever()
$ python3 future_burger.py
Burger #10 is cooked!
Burger is ready for pick up!
Burger #10? That’s me! Mmmmmm!